diff --git a/crates/buzz-core/src/ifc.rs b/crates/buzz-core/src/ifc.rs index bfd3408b93..b3000c1087 100644 --- a/crates/buzz-core/src/ifc.rs +++ b/crates/buzz-core/src/ifc.rs @@ -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 Result { 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 { + 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 { 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()); diff --git a/crates/buzz-db/src/store/ifc.rs b/crates/buzz-db/src/store/ifc.rs index bbaa371258..06ad704dc7 100644 --- a/crates/buzz-db/src/store/ifc.rs +++ b/crates/buzz-db/src/store/ifc.rs @@ -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 = 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,)> = 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", diff --git a/crates/buzz-db/src/store/replaceable.rs b/crates/buzz-db/src/store/replaceable.rs index 68d11f4b85..cb54e9df58 100644 --- a/crates/buzz-db/src/store/replaceable.rs +++ b/crates/buzz-db/src/store/replaceable.rs @@ -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] diff --git a/crates/buzz-relay/examples/generate_ifc_fixture.rs b/crates/buzz-relay/examples/generate_ifc_fixture.rs index 5e3a11454d..55e343cb90 100644 --- a/crates/buzz-relay/examples/generate_ifc_fixture.rs +++ b/crates/buzz-relay/examples/generate_ifc_fixture.rs @@ -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()) diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index eee0b82270..80fdac0c4d 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -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) diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index ae3fac6123..05ae3aa1bd 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -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") diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 99a7935308..694dc921b9 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -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 = 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,)> = 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 diff --git a/crates/buzz-relay/src/ifc.rs b/crates/buzz-relay/src/ifc.rs index 61f2e27fb8..d80e7786f1 100644 --- a/crates/buzz-relay/src/ifc.rs +++ b/crates/buzz-relay/src/ifc.rs @@ -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"] diff --git a/crates/buzz-relay/tests/fixtures/ifc-v1-golden.json b/crates/buzz-relay/tests/fixtures/ifc-v1-golden.json index 150e72b34d..551af1fa28 100644 --- a/crates/buzz-relay/tests/fixtures/ifc-v1-golden.json +++ b/crates/buzz-relay/tests/fixtures/ifc-v1-golden.json @@ -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" ] ] },