jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-28 04:51:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(relay): require fenced memory revisions

Browse Source 
Co-authored-by: Codex <codex@openai.com>
Signed-off-by: jm <jm@squareup.com>
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
jm
2026-10-14 19:58:53 +00:00
parent 0bc15a282e
commit 249fd718c7
9 changed files with 574 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

107
crates/buzz-core/src/ifc.rs
View File

@@ -47,6 +47,15 @@ pub struct IfcPolicyRef {
pub generation: i64,
}
/// Signed optimistic-concurrency precondition for a fenced owner-memory slot.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum IfcMemoryExpectedHead {
/// The coordinate must not have a live head.
Absent,
/// The coordinate's live head must be this exact event ID.
Event([u8; 32]),
}
/// Closed destination admitted by a version-one publication fence.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum IfcPublicationTarget {
@@ -58,6 +67,8 @@ pub enum IfcPublicationTarget {
owner_pubkey: String,
/// Opaque HMAC address used by the existing memory protocol.
d_tag: String,
/// Exact head or absence frozen into the signed publication.
expected_head: IfcMemoryExpectedHead,
},
/// Existing owner-encrypted ephemeral telemetry route.
OwnerTelemetry {
@@ -211,6 +222,7 @@ pub fn parse_publication_fence(event: &Event) -> Result<Option<IfcPublicationFen
let owner_pubkey = owner_memory.ok_or_else(|| invalid("missing ifc-owner-memory tag"))?;
let mut d_tag = None;
let mut p_tag = None;
let mut expected_head = None;
for tag in event.tags.iter() {
let parts = tag.as_slice();
let Some(name) = parts.first().map(String::as_str) else {
@@ -223,18 +235,24 @@ pub fn parse_publication_fence(event: &Event) -> Result<Option<IfcPublicationFen
"p" if p_tag.is_none() && parts.len() == 2 && lowercase_hex_32(&parts[1]) => {
p_tag = Some(parts[1].clone());
}
"expected-revision" if expected_head.is_none() && parts.len() == 2 => {
expected_head = Some(parse_memory_expected_head(&parts[1])?);
}
"ifc" | "ifc-source" | "ifc-owner-memory" => {}
_ => return Err(invalid("owner memory contains an unsupported tag")),
}
}
let d_tag = d_tag.ok_or_else(|| invalid("owner memory requires one opaque d tag"))?;
let p_tag = p_tag.ok_or_else(|| invalid("owner memory requires one owner p tag"))?;
let expected_head = expected_head
.ok_or_else(|| invalid("owner memory requires one expected-revision tag"))?;
if p_tag != owner_pubkey {
return Err(invalid("owner memory p tag does not match its fixed owner"));
}
IfcPublicationTarget::OwnerMemory {
owner_pubkey,
d_tag,
expected_head,
}
} else {
if destination.is_some() || owner_memory.is_some() {
@@ -305,6 +323,21 @@ fn lowercase_hex_32(value: &str) -> bool {
.all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
}
fn parse_memory_expected_head(value: &str) -> Result<IfcMemoryExpectedHead, IfcTagError> {
if value == "none" {
return Ok(IfcMemoryExpectedHead::Absent);
}
if !lowercase_hex_32(value) {
return Err(invalid("owner memory expected revision is not canonical"));
}
let bytes = hex::decode(value)
.map_err(|_| invalid("owner memory expected revision is not canonical"))?;
let event_id: [u8; 32] = bytes
.try_into()
.map_err(|_| invalid("owner memory expected revision is not canonical"))?;
Ok(IfcMemoryExpectedHead::Event(event_id))
}
fn canonical_uuid(value: &str, field: &str) -> Result<Uuid, IfcTagError> {
let parsed = Uuid::parse_str(value).map_err(|_| invalid(&format!("{field} is not a UUID")))?;
if parsed.to_string() != value {
@@ -403,6 +436,7 @@ mod tests {
],
&["ifc-source", "11111111-1111-4111-8111-111111111111", "17"],
&["ifc-owner-memory", owner],
&["expected-revision", "none"],
],
))
.expect("parse")
@@ -412,6 +446,40 @@ mod tests {
IfcPublicationTarget::OwnerMemory {
owner_pubkey: owner.to_owned(),
d_tag: d_tag.to_owned(),
expected_head: IfcMemoryExpectedHead::Absent,
}
);
}
#[test]
fn parses_owner_memory_expected_event_head() {
let owner = "2222222222222222222222222222222222222222222222222222222222222222";
let d_tag = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
let head = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
let parsed = parse_publication_fence(&event(
KIND_AGENT_ENGRAM,
&[
&["d", d_tag],
&["p", owner],
&[
"ifc",
"1",
"aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
"1792000000",
],
&["ifc-source", "11111111-1111-4111-8111-111111111111", "17"],
&["ifc-owner-memory", owner],
&["expected-revision", head],
],
))
.expect("parse")
.expect("fence");
assert_eq!(
parsed.target,
IfcPublicationTarget::OwnerMemory {
owner_pubkey: owner.to_owned(),
d_tag: d_tag.to_owned(),
expected_head: IfcMemoryExpectedHead::Event([0xbb; 32]),
}
);
}
@@ -496,6 +564,7 @@ mod tests {
"1792000000",
];
let source = ["ifc-source", "11111111-1111-4111-8111-111111111111", "17"];
let expected = ["expected-revision", "none"];
for tags in [
vec![&["d", d_tag][..], &["p", owner], &common, &source],
vec![
@@ -504,6 +573,7 @@ mod tests {
&common,
&source,
&["ifc-owner-memory", owner],
&expected,
],
vec![
&["d", d_tag],
@@ -512,6 +582,43 @@ mod tests {
&common,
&source,
&["ifc-owner-memory", owner],
&expected,
],
vec![
&["d", d_tag],
&["p", owner],
&common,
&source,
&["ifc-owner-memory", owner],
&[
"expected-revision",
"ABCDEFABCDEFABCDEFABCDEFABCDEFABCDEFABCDEFABCDEFABCDEFABCDEFABCD",
],
],
vec![
&["d", d_tag],
&["p", owner],
&common,
&source,
&["ifc-owner-memory", owner],
&["expected-revision", "abcd"],
],
vec![
&["d", d_tag],
&["p", owner],
&common,
&source,
&["ifc-owner-memory", owner],
&expected,
&expected,
],
vec![
&["d", d_tag],
&["p", owner],
&common,
&source,
&["ifc-owner-memory", owner],
&["expected-revision", "none", "extra"],
],
] {
assert!(parse_publication_fence(&event(KIND_AGENT_ENGRAM, &tags)).is_err());

228
crates/buzz-db/src/store/ifc.rs
View File

@@ -2,7 +2,7 @@
use std::collections::BTreeMap;
use buzz_core::ifc::{IfcPublicationFence, IfcPublicationTarget};
use buzz_core::ifc::{IfcMemoryExpectedHead, IfcPublicationFence, IfcPublicationTarget};
use buzz_core::{CommunityId, StoredEvent};
use chrono::{DateTime, Utc};
use nostr::Event;
@@ -436,6 +436,7 @@ impl Db {
IfcPublicationTarget::OwnerMemory {
owner_pubkey,
d_tag,
..
} => {
super::replaceable::acquire_parameterized_event_write_lock(&mut tx, event, d_tag)
.await?;
@@ -511,17 +512,27 @@ impl Db {
IfcPublishOutcome::Inserted(Box::new(stored))
}
}
IfcPublicationTarget::OwnerMemory { d_tag, .. } => {
IfcPublicationTarget::OwnerMemory {
d_tag,
expected_head,
..
} => {
if thread_meta.is_some() || canvas_precondition.is_some() {
return Err(IfcError::Invalid);
}
let precondition = match expected_head {
IfcMemoryExpectedHead::Absent => ParameterizedReplacePrecondition::ExpectNoHead,
IfcMemoryExpectedHead::Event(event_id) => {
ParameterizedReplacePrecondition::ExpectedRevision(event_id.as_slice())
}
};
let replacement = self
.replace_parameterized_event_in_transaction(
&mut tx,
event,
d_tag,
None,
ParameterizedReplacePrecondition::Unconditional,
precondition,
)
.await?;
match replacement.status {
@@ -762,14 +773,14 @@ mod postgres_tests {
community: CommunityId,
source: (Uuid, i64),
d_tag: &str,
timing: (u64, i64),
revision_and_timing: (&str, u64, i64),
content: &str,
) -> Event {
let owner_hex = owner.public_key().to_hex();
let source_channel = source.0.to_string();
let source_generation = source.1.to_string();
let community_text = community.to_string();
let expiry = timing.1.to_string();
let expiry = revision_and_timing.2.to_string();
EventBuilder::new(Kind::Custom(KIND_AGENT_ENGRAM as u16), content)
.tags([
Tag::parse(["d", d_tag]).expect("d tag"),
@@ -783,8 +794,10 @@ mod postgres_tests {
])
.expect("source tag"),
Tag::parse(["ifc-owner-memory", owner_hex.as_str()]).expect("owner-memory tag"),
Tag::parse(["expected-revision", revision_and_timing.0])
.expect("expected revision tag"),
])
.custom_created_at(Timestamp::from(timing.0))
.custom_created_at(Timestamp::from(revision_and_timing.1))
.sign_with_keys(keys)
.expect("sign owner memory")
}
@@ -1540,7 +1553,7 @@ mod postgres_tests {
community,
(source, 1),
d_tag,
(now, Utc::now().timestamp() + 120),
("none", now, Utc::now().timestamp() + 120),
"first ciphertext",
);
let first_fence = parse_publication_fence(&first)
@@ -1565,7 +1578,7 @@ mod postgres_tests {
community,
(source, 1),
d_tag,
(now + 1, Utc::now().timestamp() + 120),
(&first.id.to_hex(), now + 1, Utc::now().timestamp() + 120),
"second ciphertext",
);
let second_fence = parse_publication_fence(&second)
@@ -1590,7 +1603,7 @@ mod postgres_tests {
community,
(source, 1),
d_tag,
(now - 1, Utc::now().timestamp() + 120),
(&second.id.to_hex(), now - 1, Utc::now().timestamp() + 120),
"stale ciphertext",
);
let stale_fence = parse_publication_fence(&stale)
@@ -1615,6 +1628,38 @@ mod postgres_tests {
.await
.expect("stale receipt"));
let wrong_head_text = "42".repeat(32);
let wrong_head = fenced_memory_event(
&agent,
&owner,
community,
(source, 1),
d_tag,
(&wrong_head_text, now + 2, Utc::now().timestamp() + 120),
"wrong expected head",
);
let wrong_head_fence = parse_publication_fence(&wrong_head)
.expect("parse wrong head")
.expect("wrong head fence");
assert!(matches!(
db.publish_ifc_event(
community,
&wrong_head,
&wrong_head_fence,
&agent.public_key().to_hex(),
None,
None
)
.await,
Ok(IfcPublishOutcome::MemoryConflict(
ParameterizedReplaceStatus::RevisionMismatch
))
));
assert!(!db
.ifc_event_was_accepted(community, wrong_head.id.as_bytes())
.await
.expect("wrong-head receipt"));
sqlx::query("UPDATE channels SET visibility='open' WHERE community_id=$1 AND id=$2")
.bind(community.as_uuid())
.bind(source)
@@ -1640,7 +1685,7 @@ mod postgres_tests {
community,
(source, 2),
"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
(now + 2, Utc::now().timestamp() + 120),
("none", now + 2, Utc::now().timestamp() + 120),
"wrong owner ciphertext",
);
let wrong_fence = parse_publication_fence(&wrong)
@@ -1659,6 +1704,136 @@ mod postgres_tests {
Err(IfcError::AccessDenied)
));
let interleaved_d = "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd";
let frozen_absence = fenced_memory_event(
&agent,
&owner,
community,
(source, 2),
interleaved_d,
("none", now + 5, Utc::now().timestamp() + 120),
"frozen absence",
);
let frozen_absence_fence = parse_publication_fence(&frozen_absence)
.expect("parse frozen absence")
.expect("frozen absence fence");
let owner_hex = owner.public_key().to_hex();
let legacy = EventBuilder::new(Kind::Custom(KIND_AGENT_ENGRAM as u16), "legacy ciphertext")
.tags([
Tag::parse(["d", interleaved_d]).expect("legacy d tag"),
Tag::parse(["p", owner_hex.as_str()]).expect("legacy owner tag"),
])
.custom_created_at(Timestamp::from(now + 4))
.sign_with_keys(&agent)
.expect("sign legacy memory");
assert!(
db.replace_parameterized_event(community, &legacy, interleaved_d, None)
.await
.expect("insert interleaved legacy head")
.1
);
assert!(matches!(
db.publish_ifc_event(
community,
&frozen_absence,
&frozen_absence_fence,
&agent.public_key().to_hex(),
None,
None
)
.await,
Ok(IfcPublishOutcome::MemoryConflict(
ParameterizedReplaceStatus::RevisionMismatch
))
));
assert!(!db
.ifc_event_was_accepted(community, frozen_absence.id.as_bytes())
.await
.expect("frozen absence receipt"));
let tombstone = fenced_memory_event(
&agent,
&owner,
community,
(source, 2),
interleaved_d,
(&legacy.id.to_hex(), now + 6, Utc::now().timestamp() + 120),
"encrypted value null",
);
let tombstone_fence = parse_publication_fence(&tombstone)
.expect("parse tombstone")
.expect("tombstone fence");
assert!(matches!(
db.publish_ifc_event(
community,
&tombstone,
&tombstone_fence,
&agent.public_key().to_hex(),
None,
None
)
.await,
Ok(IfcPublishOutcome::Inserted(_))
));
let missing_tombstone_head = fenced_memory_event(
&agent,
&owner,
community,
(source, 2),
interleaved_d,
("none", now + 7, Utc::now().timestamp() + 120),
"must not replace tombstone",
);
let missing_tombstone_fence = parse_publication_fence(&missing_tombstone_head)
.expect("parse missing tombstone head")
.expect("missing tombstone head fence");
assert!(matches!(
db.publish_ifc_event(
community,
&missing_tombstone_head,
&missing_tombstone_fence,
&agent.public_key().to_hex(),
None,
None
)
.await,
Ok(IfcPublishOutcome::MemoryConflict(
ParameterizedReplaceStatus::RevisionMismatch
))
));
assert!(!db
.ifc_event_was_accepted(community, missing_tombstone_head.id.as_bytes())
.await
.expect("missing tombstone receipt"));
let after_tombstone = fenced_memory_event(
&agent,
&owner,
community,
(source, 2),
interleaved_d,
(
&tombstone.id.to_hex(),
now + 8,
Utc::now().timestamp() + 120,
),
"replacement after tombstone",
);
let after_tombstone_fence = parse_publication_fence(&after_tombstone)
.expect("parse replacement after tombstone")
.expect("replacement after tombstone fence");
assert!(matches!(
db.publish_ifc_event(
community,
&after_tombstone,
&after_tombstone_fence,
&agent.public_key().to_hex(),
None,
None
)
.await,
Ok(IfcPublishOutcome::Inserted(_))
));
let expires_at = Utc::now().timestamp() + 2;
let expiring = fenced_memory_event(
&agent,
@@ -1666,7 +1841,7 @@ mod postgres_tests {
community,
(source, 2),
"cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
(now + 3, expires_at),
("none", now + 9, expires_at),
"expiring ciphertext",
);
let expiring_fence = parse_publication_fence(&expiring)
@@ -1684,6 +1859,24 @@ mod postgres_tests {
.await,
Ok(IfcPublishOutcome::Inserted(_))
));
let expiring_d = "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc";
let later_legacy = EventBuilder::new(
Kind::Custom(KIND_AGENT_ENGRAM as u16),
"later legacy ciphertext",
)
.tags([
Tag::parse(["d", expiring_d]).expect("later legacy d tag"),
Tag::parse(["p", owner_hex.as_str()]).expect("later legacy owner tag"),
])
.custom_created_at(Timestamp::from(now + 10))
.sign_with_keys(&agent)
.expect("sign later legacy memory");
assert!(
db.replace_parameterized_event(community, &later_legacy, expiring_d, None)
.await
.expect("insert later legacy head")
.1
);
tokio::time::sleep(std::time::Duration::from_secs(3)).await;
assert!(matches!(
db.publish_ifc_event(
@@ -1698,6 +1891,19 @@ mod postgres_tests {
Ok(IfcPublishOutcome::HistoricDuplicate)
));
let expiring_head: Vec<u8> = sqlx::query_scalar(
"SELECT id FROM events WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \
AND d_tag=$4 AND deleted_at IS NULL",
)
.bind(community.as_uuid())
.bind(KIND_AGENT_ENGRAM as i32)
.bind(agent.public_key().as_bytes())
.bind(expiring_d)
.fetch_one(db.pool())
.await
.expect("read post-retry memory head");
assert_eq!(expiring_head, later_legacy.id.as_bytes().to_vec());
let active: Vec<(Vec<u8>,)> = sqlx::query_as(
"SELECT id FROM events WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \
AND d_tag=$4 AND deleted_at IS NULL",

53
crates/buzz-db/src/store/replaceable.rs
View File

@@ -33,6 +33,8 @@ pub enum ParameterizedReplacePrecondition<'a> {
Unconditional,
/// Require the live head to match this validated event ID.
ExpectedRevision(&'a [u8]),
/// Require the coordinate to have no live head.
ExpectNoHead,
/// Accept only an exact live-head replay and perform no mutation otherwise.
ExactReplayOnly,
}
@@ -226,6 +228,15 @@ async fn replace_parameterized_event_in_transaction_impl(
));
}
if precondition == ParameterizedReplacePrecondition::ExpectNoHead && existing.is_some() {
return Ok(ParameterizedReplaceResult::new(
event,
received_at,
channel_id,
ParameterizedReplaceStatus::RevisionMismatch,
));
}
if let ParameterizedReplacePrecondition::ExpectedRevision(expected_revision) = precondition {
let status = match existing.as_ref() {
None => Some(ParameterizedReplaceStatus::RevisionMissing),
@@ -1323,6 +1334,28 @@ mod postgres_tests {
);
tx.rollback().await.expect("roll back stale revision tx");
let mut tx = db
.begin_event_write_transaction(community)
.await
.expect("begin occupied-coordinate tx");
let occupied = db
.replace_parameterized_event_in_transaction(
&mut tx,
&new,
&d_tag,
None,
replaceable::ParameterizedReplacePrecondition::ExpectNoHead,
)
.await
.expect("evaluate occupied coordinate");
assert_eq!(
occupied.status,
replaceable::ParameterizedReplaceStatus::RevisionMismatch
);
tx.rollback()
.await
.expect("roll back occupied-coordinate tx");
let missing_d_tag = format!("missing-project-{}", Uuid::new_v4().simple());
let missing = EventBuilder::new(
Kind::Custom(buzz_core::kind::KIND_PROJECT as u16),
@@ -1355,6 +1388,26 @@ mod postgres_tests {
replaceable::ParameterizedReplaceStatus::RevisionMissing
);
tx.rollback().await.expect("roll back missing revision tx");
let mut tx = db
.begin_event_write_transaction(community)
.await
.expect("begin absent-coordinate tx");
let absent = db
.replace_parameterized_event_in_transaction(
&mut tx,
&missing,
&missing_d_tag,
None,
replaceable::ParameterizedReplacePrecondition::ExpectNoHead,
)
.await
.expect("insert absent coordinate");
assert_eq!(
absent.status,
replaceable::ParameterizedReplaceStatus::Inserted
);
tx.rollback().await.expect("roll back absent-coordinate tx");
}
#[tokio::test]

1
crates/buzz-relay/examples/generate_ifc_fixture.rs
View File

@@ -119,6 +119,7 @@ fn main() -> anyhow::Result<()> {
Tag::parse(["ifc", "1", community, "1792000000"])?,
Tag::parse(["ifc-source", source, "17"])?,
Tag::parse(["ifc-owner-memory", owner.as_str()])?,
Tag::parse(["expected-revision", "none"])?,
])
.custom_created_at(Timestamp::from(1_791_999_920u64))
.build(broker.public_key())

1
crates/buzz-relay/src/api/bridge.rs
View File

@@ -7780,6 +7780,7 @@ pub(crate) mod postgres_tests {
.expect("ifc tag"),
Tag::parse(["ifc-source", source_text.as_str(), "1"]).expect("source tag"),
Tag::parse(["ifc-owner-memory", owner_hex.as_str()]).expect("memory target"),
Tag::parse(["expected-revision", "none"]).expect("memory expected revision"),
]);
}
EventBuilder::new(Kind::Custom(KIND_AGENT_ENGRAM as u16), ciphertext)

1
crates/buzz-relay/src/handlers/event.rs
View File

@@ -2850,6 +2850,7 @@ mod tests {
.expect("ifc tag"),
Tag::parse(["ifc-source", source_text.as_str(), "1"]).expect("source tag"),
Tag::parse(["ifc-owner-memory", owner_hex.as_str()]).expect("memory target"),
Tag::parse(["expected-revision", "none"]).expect("memory expected revision"),
])
.sign_with_keys(agent)
.expect("sign memory")

207
crates/buzz-relay/src/handlers/ingest.rs
View File

@@ -7026,22 +7026,29 @@ mod postgres_tests {
)
.expect("encrypt memory")
};
let memory = |created_at: u64, value: &str, extra: &[Tag]| {
let mut tags = vec![
Tag::parse(["d", d.as_str()]).expect("d tag"),
Tag::parse(["p", owner_hex.as_str()]).expect("p tag"),
Tag::parse(["ifc", "1", community_text.as_str(), expiry.as_str()])
.expect("ifc tag"),
Tag::parse(["ifc-source", source_text.as_str(), "1"]).expect("source tag"),
Tag::parse(["ifc-owner-memory", owner_hex.as_str()]).expect("owner-memory tag"),
];
tags.extend_from_slice(extra);
EventBuilder::new(Kind::Custom(KIND_AGENT_ENGRAM as u16), encrypt(value))
.tags(tags)
.custom_created_at(Timestamp::from(created_at))
.sign_with_keys(&agent)
.expect("sign fenced memory")
};
let memory =
|created_at: u64, value: &str, expected_revision: Option<&str>, extra: &[Tag]| {
let mut tags = vec![
Tag::parse(["d", d.as_str()]).expect("d tag"),
Tag::parse(["p", owner_hex.as_str()]).expect("p tag"),
Tag::parse(["ifc", "1", community_text.as_str(), expiry.as_str()])
.expect("ifc tag"),
Tag::parse(["ifc-source", source_text.as_str(), "1"]).expect("source tag"),
Tag::parse(["ifc-owner-memory", owner_hex.as_str()]).expect("owner-memory tag"),
];
if let Some(expected_revision) = expected_revision {
tags.push(
Tag::parse(["expected-revision", expected_revision])
.expect("expected revision tag"),
);
}
tags.extend_from_slice(extra);
EventBuilder::new(Kind::Custom(KIND_AGENT_ENGRAM as u16), encrypt(value))
.tags(tags)
.custom_created_at(Timestamp::from(created_at))
.sign_with_keys(&agent)
.expect("sign fenced memory")
};
let tracer: Arc<dyn buzz_conformance::Tracer> = Arc::new(VecTracer::default());
let http = || IngestAuth::Http {
pubkey: agent.public_key(),
@@ -7055,21 +7062,109 @@ mod postgres_tests {
conn_id: Uuid::new_v4(),
};
let now = Utc::now().timestamp() as u64;
let first = memory(now, "first", &[]);
let first = memory(now, "first", Some("none"), &[]);
let conn_id = Uuid::new_v4();
let (event_tx, mut event_rx) = tokio::sync::mpsc::channel(4);
let (control_tx, _control_rx) = tokio::sync::mpsc::channel(4);
state.conn_manager.register(
conn_id,
event_tx,
control_tx,
tokio::sync::mpsc::channel(1).0,
None,
tokio_util::sync::CancellationToken::new(),
community,
Arc::new(std::sync::atomic::AtomicU8::new(0)),
Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new())),
3,
crate::state::CommunityConnectionControl::new(
tokio_util::sync::CancellationToken::new(),
),
);
state
.conn_manager
.set_authenticated_pubkey(conn_id, owner.public_key().to_bytes().to_vec());
state.sub_registry.register_scoped(
community,
conn_id,
"memory-cas-retry".to_owned(),
vec![nostr::Filter::new()
.kind(Kind::Custom(KIND_AGENT_ENGRAM as u16))
.id(first.id)],
None,
);
ingest_event_inner(&state, &tracer, &tenant, first.clone(), http())
.await
.expect("insert first memory");
let second = memory(now + 1, "second", &[]);
assert!(
tokio::time::timeout(std::time::Duration::from_secs(2), event_rx.recv())
.await
.expect("first fanout deadline")
.is_some(),
"first acceptance must fan out"
);
let first_id = first.id.to_hex();
let second = memory(now + 1, "second", Some(first_id.as_str()), &[]);
ingest_event_inner(&state, &tracer, &tenant, second.clone(), websocket())
.await
.expect("insert second memory");
let stale = memory(now - 1, "stale", &[]);
let second_id = second.id.to_hex();
let stale = memory(now - 1, "stale", Some(second_id.as_str()), &[]);
assert!(matches!(
ingest_event_inner(&state, &tracer, &tenant, stale.clone(), http()).await,
Err(IngestError::CanvasConflict(message))
if message == "conflict: stale owner-memory replacement"
));
let legacy = EventBuilder::new(Kind::Custom(KIND_AGENT_ENGRAM as u16), encrypt("legacy"))
.tags([
Tag::parse(["d", d.as_str()]).expect("legacy d tag"),
Tag::parse(["p", owner_hex.as_str()]).expect("legacy p tag"),
])
.custom_created_at(Timestamp::from(now + 2))
.sign_with_keys(&agent)
.expect("sign legacy memory");
ingest_event_inner(&state, &tracer, &tenant, legacy.clone(), websocket())
.await
.expect("insert ordinary unfenced replacement");
let raced = memory(now + 3, "raced", Some(second_id.as_str()), &[]);
assert!(matches!(
ingest_event_inner(&state, &tracer, &tenant, raced.clone(), http()).await,
Err(IngestError::CanvasConflict(message))
if message == "conflict: stale owner-memory replacement"
));
let legacy_id = legacy.id.to_hex();
let tombstone_ciphertext = nip44::encrypt(
agent.secret_key(),
&owner.public_key(),
format!(r#"{{"slug":"{slug}","value":null}}"#),
Version::V2,
)
.expect("encrypt memory tombstone");
let tombstone =
EventBuilder::new(Kind::Custom(KIND_AGENT_ENGRAM as u16), tombstone_ciphertext)
.tags(memory(now + 4, "tombstone template", Some(legacy_id.as_str()), &[]).tags)
.custom_created_at(Timestamp::from(now + 4))
.sign_with_keys(&agent)
.expect("sign memory tombstone");
ingest_event_inner(&state, &tracer, &tenant, tombstone.clone(), http())
.await
.expect("replace memory with encrypted tombstone");
let absent_after_tombstone = memory(now + 5, "must conflict", Some("none"), &[]);
assert!(matches!(
ingest_event_inner(
&state,
&tracer,
&tenant,
absent_after_tombstone.clone(),
websocket()
)
.await,
Err(IngestError::CanvasConflict(message))
if message == "conflict: stale owner-memory replacement"
));
state
.db
.update_channel(
@@ -7086,8 +7181,15 @@ mod postgres_tests {
.await
.expect("historic memory retry");
assert_eq!(retry.message, "duplicate:");
assert!(
tokio::time::timeout(std::time::Duration::from_millis(100), event_rx.recv())
.await
.is_err(),
"historic acceptance must not fan out again"
);
let stale_source = memory(now + 2, "stale source", &[]);
let tombstone_id = tombstone.id.to_hex();
let stale_source = memory(now + 6, "stale source", Some(tombstone_id.as_str()), &[]);
assert!(matches!(
ingest_event_inner(&state, &tracer, &tenant, stale_source.clone(), http()).await,
Err(IngestError::CanvasConflict(message)) if message == "ifc-policy-changed"
@@ -7095,14 +7197,62 @@ mod postgres_tests {
let destination =
Tag::parse(["ifc-destination", source_text.as_str(), "2"]).expect("destination tag");
let malformed = memory(now + 2, "malformed", &[destination]);
let malformed = memory(
now + 6,
"malformed",
Some(tombstone_id.as_str()),
&[destination],
);
assert!(matches!(
ingest_event_inner(&state, &tracer, &tenant, malformed.clone(), http()).await,
Err(IngestError::Rejected(message)) if message == "ifc-invalid"
));
let missing_precondition = memory(now + 7, "missing", None, &[]);
assert!(matches!(
ingest_event_inner(
&state,
&tracer,
&tenant,
missing_precondition.clone(),
websocket()
)
.await,
Err(IngestError::Rejected(message)) if message == "ifc-invalid"
));
let malformed_precondition = memory(now + 7, "bad expected", Some("abcd"), &[]);
assert!(matches!(
ingest_event_inner(
&state,
&tracer,
&tenant,
malformed_precondition.clone(),
http()
)
.await,
Err(IngestError::Rejected(message)) if message == "ifc-invalid"
));
let duplicate_expected = Tag::parse(["expected-revision", tombstone_id.as_str()])
.expect("duplicate expected tag");
let duplicate_precondition = memory(
now + 7,
"duplicate expected",
Some(tombstone_id.as_str()),
&[duplicate_expected],
);
assert!(matches!(
ingest_event_inner(
&state,
&tracer,
&tenant,
duplicate_precondition.clone(),
websocket()
)
.await,
Err(IngestError::Rejected(message)) if message == "ifc-invalid"
));
let bad_cipher = EventBuilder::new(Kind::Custom(KIND_AGENT_ENGRAM as u16), "not-nip44")
.tags(memory(now + 3, "template", &[]).tags)
.custom_created_at(Timestamp::from(now + 3))
.tags(memory(now + 7, "template", Some(tombstone_id.as_str()), &[]).tags)
.custom_created_at(Timestamp::from(now + 7))
.sign_with_keys(&agent)
.expect("sign malformed ciphertext");
assert!(matches!(
@@ -7117,15 +7267,22 @@ mod postgres_tests {
.bind(vec![
first.id.as_bytes().as_slice(),
second.id.as_bytes().as_slice(),
legacy.id.as_bytes().as_slice(),
tombstone.id.as_bytes().as_slice(),
stale.id.as_bytes().as_slice(),
raced.id.as_bytes().as_slice(),
absent_after_tombstone.id.as_bytes().as_slice(),
stale_source.id.as_bytes().as_slice(),
malformed.id.as_bytes().as_slice(),
missing_precondition.id.as_bytes().as_slice(),
malformed_precondition.id.as_bytes().as_slice(),
duplicate_precondition.id.as_bytes().as_slice(),
bad_cipher.id.as_bytes().as_slice(),
])
.fetch_one(&pool)
.await
.expect("count memory production rows");
assert_eq!(stored, 2);
assert_eq!(stored, 4);
let active: Vec<(Vec<u8>,)> = sqlx::query_as(
"SELECT id FROM events WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \
AND d_tag=$4 AND deleted_at IS NULL",
@@ -7137,7 +7294,7 @@ mod postgres_tests {
.fetch_all(&pool)
.await
.expect("read memory head");
assert_eq!(active, vec![(second.id.as_bytes().to_vec(),)]);
assert_eq!(active, vec![(tombstone.id.as_bytes().to_vec(),)]);
}
/// End-to-end CAS dispatch wiring: a tagged write inserts, a stale same-head

8
crates/buzz-relay/src/ifc.rs
View File

@@ -461,7 +461,7 @@ mod tests {
use super::*;
use base64::Engine;
use buzz_core::engram::{conversation_key, d_tag, Body};
use buzz_core::ifc::{parse_publication_fence, IfcPublicationTarget};
use buzz_core::ifc::{parse_publication_fence, IfcMemoryExpectedHead, IfcPublicationTarget};
use nostr::nips::nip44::v2;
use nostr::Keys;
@@ -562,7 +562,11 @@ mod tests {
.expect("memory d tag");
assert!(matches!(
memory_fence.target,
IfcPublicationTarget::OwnerMemory { ref d_tag, .. } if d_tag == expected_d
IfcPublicationTarget::OwnerMemory {
ref d_tag,
expected_head: IfcMemoryExpectedHead::Absent,
..
} if d_tag == expected_d
));
let broker = Keys::parse(
fixture["test_keys"]["broker_secret_hex"]

8
crates/buzz-relay/tests/fixtures/ifc-v1-golden.json vendored
View File

@@ -37,10 +37,10 @@
"event": {
"content": "AkREREREREREREREREREREREREREREREREREREREREREQpFvZ/ygpdh8oBcoxHeFvIPj5KMRE7gj18SwdfeX5Qsv8BDW9VIdrJoWkmnz0nw78QBsb7HFbjDX21PMHa8CL5ot5ULx1c/BOEU4tarpV0XACUUICh9rMtp5BFgXoJMTNbe5ZRRuUFpO1UP1lU9yYNIbuLE7J+jKQLGalyDNe5liOxmA0Zs4V54zLCOuf7PWQvBdeDTgrhWcu7IFe+uJe3pm6qBv7wu9e5/882VunBBPRQN/bI2m18Wn+w+aAmLqkewdsw6JzeahGt9lbBVVRO8mgFG4yIK/OAe7XP6czR9O/vj3Aoysuh5NI06W9SNjPjNyKTgZAyjS9NE9WmCLD4MiMZs4zeuaT3QF5eu1YwLyPGbXv3PVaqeR6eszCqJF1y6BZ52HqcLm0VKkjc0KFRJLcCqImqyDaBQGBbmfEpa7YfyOLKEmXFloSjxdTP6cBxeHMojOogDpSih1xha8+6Hy",
"created_at": 1791999920,
"id": "1386ef4d498595826ff186e25fa7324c7f41ec0d1af8e43d506cf8c5d7879c3d",
"id": "3f75f3ed90d00a402ab5d9ee0da9692860be6ce62add02734fa92c992c0baeac",
"kind": 30174,
"pubkey": "f9308a019258c31049344f85f89d5229b531c845836f99b08601f113bce036f9",
"sig": "de03b626272d1608821b846e2d2d43e53cc3845820fea23da4f34e64135d01e584d6f7bb7ae1504a6ca4e7a193cf0e92abb0fa4d8b754a2edb6a731ec9e18653",
"sig": "eb6cb03ea0ff80c1b291238aef0bb017732d273b8ff9f038098311fa77aa06d7057483cd8dab567eac4d8508154641d5414f096b1105749121acdd89545e8af5",
"tags": [
[
"d",
@@ -64,6 +64,10 @@
[
"ifc-owner-memory",
"c6047f9441ed7d6d3045406e95c07cd85c778e4b8cef3ca7abac09b95c709ee5"
],
[
"expected-revision",
"none"
]
]
},