diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index 36ae4a1dc6b..eb9b16b6d6e 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -706,6 +706,32 @@ pub struct ReapedEphemeralChannel { pub channel_id: Uuid, } +/// The channel row, deleted or not, on a caller-owned transaction, or `None` +/// when it is missing. +pub(crate) async fn channel_in_transaction( + tx: &mut crate::AdmittedTx, + channel_id: Uuid, +) -> Result> { + let community_id = tx.community(); + let row = sqlx::query( + r#" + SELECT id, name, channel_type::text AS channel_type, visibility::text AS visibility, + description, canvas, + created_by, created_at, updated_at, archived_at, deleted_at, + nip29_group_id, topic_required, max_members, + topic, topic_set_by, topic_set_at, + purpose, purpose_set_by, purpose_set_at, + ttl_seconds, ttl_deadline, labels, join_role, unlisted, auto_join + FROM channels WHERE community_id = $1 AND id = $2 + "#, + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .fetch_optional(tx.conn()) + .await?; + row.map(row_to_channel_record).transpose() +} + pub(crate) fn row_to_channel_record(row: sqlx::postgres::PgRow) -> Result { let id: Uuid = row.try_get("id")?; let topic_required: bool = row.try_get("topic_required")?; diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index 2c139228a8e..7c1dc744af3 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -413,15 +413,26 @@ pub struct ChannelIdentity { pub created_by: Vec, } -/// An active member roster captured while holding the channel's membership -/// serialization lock on one writer connection. +/// A channel's group state (row and active roster) captured while holding the +/// replacement locks of its relay-signed group-state events (kinds +/// 39000–39002) and its membership serialization lock, on one writer +/// connection. +/// +/// Every channel writer commits before it publishes group state, so an +/// emission that captures and replaces under these locks either finishes +/// before a later emission starts or reads that writer's committed change. +/// An older capture therefore can never replace a newer one. pub struct LockedMemberSnapshot { /// Canonical active members captured behind the lock. pub members: Vec, - /// The channel's type, labels and creator, read on the same connection, so the - /// roster carries the same identity tags as the channel's other - /// group-state events. `None` when no channel row exists. + /// The channel's type, labels and creator, from the same row read as + /// `channel`, so the roster carries the same identity tags as the + /// channel's other group-state events. Present for a deleted channel; + /// `None` when no channel row exists. pub identity: Option, + /// The live channel row, read on the same connection. `None` when the + /// channel is missing or deleted. + pub channel: Option, channel_id: Uuid, relay_pubkey: Vec, tx: AdmittedTx, @@ -433,13 +444,16 @@ impl LockedMemberSnapshot { self.channel_id } - /// Return the newest relay-authored member snapshot timestamp for the - /// locked coordinate, using this guard's existing connection. - pub async fn latest_member_event_timestamp(&mut self) -> Result> { + /// Return the newest relay-authored group-state event timestamp of `kind` + /// (39000–39002) for the locked channel, using this guard's existing + /// connection. + pub async fn latest_group_state_timestamp(&mut self, kind: i32) -> Result> { + require_group_state_kind(kind)?; let value: Option> = sqlx::query_scalar( - "SELECT created_at FROM events WHERE community_id = $1 AND kind = 39002 AND pubkey = $2 AND channel_id = $3 AND deleted_at IS NULL ORDER BY created_at DESC, id ASC LIMIT 1", + "SELECT created_at FROM events WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND channel_id = $4 AND deleted_at IS NULL ORDER BY created_at DESC, id ASC LIMIT 1", ) .bind(self.tx.community().as_uuid()) + .bind(kind) .bind(self.relay_pubkey.as_slice()) .bind(self.channel_id) .fetch_optional(self.tx.conn()) @@ -447,10 +461,11 @@ impl LockedMemberSnapshot { Ok(value.map(|timestamp| timestamp.timestamp() as u64)) } - /// Replace the relay-authored member snapshot on this guard's existing - /// connection. The membership lock therefore spans capture and replacement - /// without a nested pool checkout. - pub async fn replace_member_event( + /// Replace a relay-authored group-state event (kind 39000, 39001 or + /// 39002) of the locked channel on this guard's existing connection. The + /// locks therefore span capture and replacement without a nested pool + /// checkout. + pub async fn replace_group_state_event( &mut self, event: &nostr::Event, ) -> Result<(buzz_core::StoredEvent, bool)> { @@ -460,15 +475,11 @@ impl LockedMemberSnapshot { || crate::event::extract_d_tag(event) != Some(channel_id.to_string()) { return Err(DbError::InvalidData( - "member snapshot replacement does not match its locked coordinate".into(), + "group-state replacement does not match its locked coordinate".into(), )); } let kind = buzz_core::kind::event_kind_i32(event); - if kind != 39002 { - return Err(DbError::InvalidData( - "member snapshot replacement requires kind 39002".into(), - )); - } + require_group_state_kind(kind)?; let pubkey = event.pubkey.to_bytes(); let created_at_secs = event.created_at.as_secs() as i64; let created_at = chrono::DateTime::from_timestamp(created_at_secs, 0) @@ -499,7 +510,9 @@ impl LockedMemberSnapshot { } } let tags = serde_json::to_value(&event.tags)?; - validate_channel_roster_snapshot(&mut self.tx, channel_id, &tags).await?; + if kind == 39002 { + validate_channel_roster_snapshot(&mut self.tx, channel_id, &tags).await?; + } sqlx::query("UPDATE events SET deleted_at = NOW() WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND channel_id = $4 AND deleted_at IS NULL") .bind(community_id.as_uuid()).bind(kind).bind(pubkey.as_slice()).bind(channel_id) .execute(self.tx.conn()).await?; @@ -512,7 +525,7 @@ impl LockedMemberSnapshot { .bind(crate::event::extract_d_tag(event)).execute(self.tx.conn()).await?; if inserted.rows_affected() == 0 { return Err(DbError::InvalidData( - "member snapshot event id already exists".into(), + "group-state event id already exists".into(), )); } crate::store::event_follow_up::after_admitted_insert( @@ -534,19 +547,34 @@ impl LockedMemberSnapshot { )) } - /// Commit the replacement and release the membership lock. + /// Commit the replacements and release the locks. pub async fn release(self) -> Result<()> { self.tx.commit().await?; Ok(()) } } -/// Capture all active members while holding the same per-channel lock used by -/// membership writers. +/// The relay-signed group-state kinds a [`LockedMemberSnapshot`] fences, in +/// the order it takes their replacement locks. +const GROUP_STATE_KINDS: [i32; 3] = [39000, 39001, 39002]; + +fn require_group_state_kind(kind: i32) -> Result<()> { + if GROUP_STATE_KINDS.contains(&kind) { + Ok(()) + } else { + Err(DbError::InvalidData(format!( + "group-state replacement requires kind 39000, 39001 or 39002, not {kind}" + ))) + } +} + +/// Capture the channel row and all active members while holding the +/// replacement locks of the channel's group-state events and the same +/// per-channel lock used by membership writers. /// /// The returned guard must remain alive through publication. This prevents a -/// rolling relay from publishing an older roster after a concurrent add or -/// remove has committed and published newer membership state. +/// relay from publishing older group state after a concurrent metadata edit, +/// add or remove has committed and published newer state. pub async fn lock_member_snapshot( pool: &PgPool, community_id: CommunityId, @@ -559,23 +587,27 @@ pub async fn lock_member_snapshot( crate::observability::WriterOperation::EventWrite, ) .await?; - // Match the canonical replacement writer's lock order. Old binaries take - // this key before INSERT; migration 0032 then takes the membership key in - // the INSERT trigger. Taking both in that order avoids mixed-version - // duplicate heads without introducing a lock-order inversion. - let replacement_lock = crate::replaceable::event_replacement_lock_key( - community_id, - 39002, - relay_pubkey, - Some(channel_id.as_bytes()), - ); - crate::observability::observe_advisory_lock( - crate::observability::LockType::Replacement, - sqlx::query("SELECT pg_advisory_xact_lock($1)") - .bind(replacement_lock) - .execute(tx.conn()), - ) - .await?; + // Match the canonical replacement writer's lock order: each coordinate's + // replacement key, then (migration 0032's INSERT trigger, for 39002) the + // membership key. The keys are the ones `replace_addressable_event` takes, + // so a relay that still writes each kind on its own transaction + // serializes with this one. No writer holds one of these keys while it + // waits for another, so taking all three in kind order cannot invert. + for kind in GROUP_STATE_KINDS { + let replacement_lock = crate::replaceable::event_replacement_lock_key( + community_id, + kind, + relay_pubkey, + Some(channel_id.as_bytes()), + ); + crate::observability::observe_advisory_lock( + crate::observability::LockType::Replacement, + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(replacement_lock) + .execute(tx.conn()), + ) + .await?; + } acquire_channel_membership_lock(tx.conn(), community_id, channel_id).await?; let rows = sqlx::query( r#" @@ -594,21 +626,16 @@ pub async fn lock_member_snapshot( .into_iter() .map(row_to_member_record) .collect::>>()?; - let identity = sqlx::query_as::<_, (String, Vec, Vec)>( - "SELECT channel_type::text, labels, created_by FROM channels WHERE community_id = $1 AND id = $2", - ) - .bind(community_id.as_uuid()) - .bind(channel_id) - .fetch_optional(tx.conn()) - .await? - .map(|(channel_type, labels, created_by)| ChannelIdentity { - channel_type, - labels, - created_by, - }); + // One read feeds both: the roster's identity tags (kept for a deleted + // channel) and the live row behind 39000/39001. Label edits do not take + // the membership lock, so two reads could straddle one. + let row = crate::channel::channel_in_transaction(&mut tx, channel_id).await?; + let identity = row.as_ref().map(ChannelRecord::identity); + let channel = row.filter(|channel| channel.deleted_at.is_none()); Ok(LockedMemberSnapshot { members, identity, + channel, channel_id, relay_pubkey: relay_pubkey.to_vec(), tx, @@ -4379,7 +4406,7 @@ mod postgres_tests { .sign_with_keys(&relay_keys) .expect("sign roster"); let (_, inserted) = snapshot - .replace_member_event(&event) + .replace_group_state_event(&event) .await .expect("replace roster"); assert!(inserted); @@ -4717,7 +4744,7 @@ mod postgres_tests { .await .expect("lock snapshot"); let error = snapshot - .replace_member_event(&stale) + .replace_group_state_event(&stale) .await .expect_err("app must reject stale roster without trigger"); assert!(matches!(error, DbError::Sqlx(sqlx::Error::Database(ref db)) @@ -4732,7 +4759,7 @@ mod postgres_tests { .expect("lock current snapshot"); assert!( snapshot - .replace_member_event(¤t) + .replace_group_state_event(¤t) .await .expect("current roster") .1 @@ -5081,9 +5108,10 @@ mod postgres_tests { assert_eq!(count, 0); } - /// A captured roster holds the same lock as membership writers until the - /// publisher explicitly releases it. This is the freshness fence used by - /// rolling-deploy reconciliation. + /// A captured snapshot holds the same lock as membership writers, and the + /// replacement keys of every group-state kind, until the publisher + /// explicitly releases it. This is the freshness fence used by group-state + /// publication and rolling-deploy reconciliation. #[tokio::test] #[ignore = "requires Postgres"] async fn locked_member_snapshot_blocks_post_capture_membership_mutation() { @@ -5121,6 +5149,16 @@ mod postgres_tests { .await .expect("capture locked roster"); assert_eq!(snapshot.members.len(), 1); + assert_eq!( + snapshot.channel.as_ref().map(|captured| captured.id), + Some(channel.id), + "the live channel row is captured behind the same locks" + ); + assert_eq!( + snapshot.identity, + snapshot.channel.as_ref().map(ChannelRecord::identity), + "the roster's identity comes from the captured row, not a second read" + ); let event = nostr::EventBuilder::new(nostr::Kind::Custom(39002), "") .tags(vec![ nostr::Tag::parse(["d", &channel.id.to_string()]).expect("d tag"), @@ -5137,13 +5175,13 @@ mod postgres_tests { .expect("sign roster for another channel"); assert!( matches!( - snapshot.replace_member_event(&other_channel).await, + snapshot.replace_group_state_event(&other_channel).await, Err(DbError::InvalidData(_)) ), "a roster whose d tag names another channel must not replace the locked one" ); let (_, inserted) = snapshot - .replace_member_event(&event) + .replace_group_state_event(&event) .await .expect("replace roster on held connection"); assert!(inserted); @@ -5163,6 +5201,24 @@ mod postgres_tests { !acquired, "membership mutation must wait until the captured roster is published" ); + // A relay that still replaces 39000 and 39001 on their own + // transactions takes these keys, so it waits for this publication too. + for kind in [39000, 39001] { + let acquired: bool = sqlx::query_scalar("SELECT pg_try_advisory_xact_lock($1)") + .bind(crate::replaceable::event_replacement_lock_key( + community, + kind, + &relay_keys.public_key().to_bytes(), + Some(channel.id.as_bytes()), + )) + .fetch_one(&mut *contender) + .await + .expect("try group-state replacement lock"); + assert!( + !acquired, + "a kind:{kind} replacement must wait until the captured state is published" + ); + } contender.rollback().await.expect("rollback contender"); snapshot.release().await.expect("release snapshot fence"); @@ -5185,6 +5241,45 @@ mod postgres_tests { ); } + /// A deleted channel has no live row for 39000/39001, but its roster keeps + /// the identity tags, so reconciliation still publishes a valid 39002. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn locked_member_snapshot_of_deleted_channel_keeps_identity() { + let pool = setup_pool().await; + let community_id = make_test_community(&pool).await; + let community = CommunityId::from_uuid(community_id); + let owner = random_pubkey(); + let channel = create_test_channel( + &pool, + community_id, + "snapshot-deleted-identity", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &owner, + None, + ) + .await + .expect("create channel"); + assert!( + crate::channel::soft_delete_channel(&pool, community, channel.id) + .await + .expect("soft delete channel") + ); + + let snapshot = lock_member_snapshot(&pool, community, channel.id, &[7u8; 32]) + .await + .expect("capture deleted channel"); + assert!( + snapshot.channel.is_none(), + "a deleted channel has no live row" + ); + assert_eq!(snapshot.identity, Some(channel.identity())); + assert!(snapshot.members.is_empty()); + snapshot.release().await.expect("release snapshot"); + } + /// Captured roster publication must hold the same shared community-deletion /// lock as serving event writes so deletion fencing can serialize against it. #[tokio::test] diff --git a/crates/buzz-db/src/store/event_follow_up_postgres_tests.rs b/crates/buzz-db/src/store/event_follow_up_postgres_tests.rs index e6beb22b514..6b4f23d0071 100644 --- a/crates/buzz-db/src/store/event_follow_up_postgres_tests.rs +++ b/crates/buzz-db/src/store/event_follow_up_postgres_tests.rs @@ -760,7 +760,7 @@ async fn assert_entry_points(db: &Db, community: CommunityId, channel: Uuid, key .expect("sign member snapshot"); assert!( snapshot - .replace_member_event(&roster) + .replace_group_state_event(&roster) .await .expect("replace member snapshot") .1 diff --git a/crates/buzz-db/src/store/replaceable.rs b/crates/buzz-db/src/store/replaceable.rs index 7c1789c3c49..4ab34e11ae4 100644 --- a/crates/buzz-db/src/store/replaceable.rs +++ b/crates/buzz-db/src/store/replaceable.rs @@ -863,7 +863,7 @@ mod postgres_tests { ); assert!( snapshot - .replace_member_event(&fresh_b) + .replace_group_state_event(&fresh_b) .await .expect("new writer publishes B") .1 diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index fcf4e6ddb16..92a2258cb73 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -886,7 +886,7 @@ const ADMITTED_TX_SIGNATURE_MARKERS: [&str; 2] = ["&mut AdmittedTx", "&mut crate const GUARDED_WRITE_FUNCTION_EXCEPTIONS: [&str; 1] = ["pub async fn verify_floor_guard_behavior("]; const GUARDED_TX_ADAPTER_METHOD_PINS: [(&str, &str); 1] = [( - "pub async fn replace_member_event(", + "pub async fn replace_group_state_event(", "pub async fn lock_member_snapshot(", )]; diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 8f2977bb04e..798bb17d5b4 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -1297,59 +1297,6 @@ async fn emit_typed_membership_notification( Ok(()) } -/// Sign, store (replacing previous), and fan-out a single addressable discovery event. -async fn emit_addressable_discovery_event( - tenant: &TenantContext, - state: &Arc, - channel_id: Uuid, - kind: u32, - tags: Vec, - relay_pubkey_hex: &str, -) -> anyhow::Result<()> { - // Ensure the new event's created_at is strictly greater than any existing event - // of the same (kind, pubkey, channel_id). Without this, rapid successive updates - // (e.g. set topic then set purpose in the same second) can produce events with - // identical created_at, causing the second to be rejected by stale-write protection - // (NIP-16 tiebreaker: lower event ID wins, which is random). - let now = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs(); - let min_ts = { - let existing = state - .db - .query_events_for_event_write(&buzz_db::event::EventQuery { - kinds: Some(vec![kind as i32]), - channel_id: Some(channel_id), - limit: Some(1), - ..buzz_db::event::EventQuery::for_community(tenant.community()) - }) - .await - .unwrap_or_default(); - existing - .first() - .map(|e| e.event.created_at.as_secs() + 1) - .unwrap_or(now) - }; - let ts = now.max(min_ts); - - let event = EventBuilder::new(Kind::Custom(kind as u16), "") - .tags(tags) - .custom_created_at(nostr::Timestamp::from(ts)) - .sign_with_keys(&state.relay_keypair) - .map_err(|e| anyhow::anyhow!("failed to sign kind:{kind}: {e}"))?; - - let (stored, was_inserted) = state - .db - .replace_addressable_event(tenant.community(), &event, Some(channel_id)) - .await?; - if was_inserted { - let kind_u32 = event_kind_u32(&stored.event); - dispatch_persistent_event(tenant, state, &stored, kind_u32, relay_pubkey_hex, None).await; - } - Ok(()) -} - /// Tags that identify the channel on every relay-signed group-state event: /// the channel type `t` first, then one `t` per label, then the creator `P`. fn group_state_identity_tags( @@ -1381,26 +1328,45 @@ fn group_members_tags( Ok(tags) } -async fn store_group_members_event( +/// Sign `builder` as the next head of its relay-signed group-state coordinate +/// and replace that head on the locked snapshot's transaction. `created_at` +/// is strictly past the current head's: a same-second replacement whose random +/// event id sorts higher would otherwise lose the NIP-16 tiebreak and be +/// dropped as stale. Returns the stored event when it became the head. +async fn store_group_state_event( state: &Arc, - member_snapshot: &mut buzz_db::channel::LockedMemberSnapshot, + snapshot: &mut buzz_db::channel::LockedMemberSnapshot, + kind: u32, + builder: EventBuilder, ) -> anyhow::Result> { - let group_id = member_snapshot.channel_id().to_string(); - let tags = group_members_tags( - &group_id, - member_snapshot.identity.as_ref(), - &member_snapshot.members, - )?; let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() .as_secs(); - let ts = member_snapshot - .latest_member_event_timestamp() + let ts = snapshot + .latest_group_state_timestamp(kind as i32) .await? .map(|timestamp| timestamp + 1) .unwrap_or(now) .max(now); + let event = builder + .custom_created_at(nostr::Timestamp::from(ts)) + .sign_with_keys(&state.relay_keypair) + .map_err(|error| anyhow::anyhow!("failed to sign kind:{kind}: {error}"))?; + let (stored, inserted) = snapshot.replace_group_state_event(&event).await?; + Ok(inserted.then_some(stored)) +} + +async fn store_group_members_event( + state: &Arc, + member_snapshot: &mut buzz_db::channel::LockedMemberSnapshot, +) -> anyhow::Result> { + let group_id = member_snapshot.channel_id().to_string(); + let tags = group_members_tags( + &group_id, + member_snapshot.identity.as_ref(), + &member_snapshot.members, + )?; // A relay-signed roster of a channel the relay is itself a member of (the // relay's moderation-DM key participates in the {relay, recipient} DM used // for moderation notices) MUST retain the relay's own `p` tag. nostr's @@ -1408,32 +1374,22 @@ async fn store_group_members_event( // would drop the relay from the snapshot and fail migration 0032's roster // fence against the canonical two-member DM. `allow_self_tagging` keeps the // snapshot faithful to `channel_members`. - let event = EventBuilder::new(Kind::Custom(KIND_NIP29_GROUP_MEMBERS as u16), "") + let builder = EventBuilder::new(Kind::Custom(KIND_NIP29_GROUP_MEMBERS as u16), "") .tags(tags) - .allow_self_tagging() - .custom_created_at(nostr::Timestamp::from(ts)) - .sign_with_keys(&state.relay_keypair) - .map_err(|error| anyhow::anyhow!("failed to sign member snapshot: {error}"))?; - let (stored, inserted) = member_snapshot.replace_member_event(&event).await?; - Ok(inserted.then_some(stored)) + .allow_self_tagging(); + store_group_state_event(state, member_snapshot, KIND_NIP29_GROUP_MEMBERS, builder).await } -async fn dispatch_group_members_event( +/// Fan out a group-state event that became its coordinate's head. +async fn dispatch_group_state_event( tenant: &TenantContext, state: &Arc, stored: Option, relay_pubkey_hex: &str, ) { if let Some(stored) = stored { - dispatch_persistent_event( - tenant, - state, - &stored, - KIND_NIP29_GROUP_MEMBERS, - relay_pubkey_hex, - None, - ) - .await; + let kind = event_kind_u32(&stored.event); + dispatch_persistent_event(tenant, state, &stored, kind, relay_pubkey_hex, None).await; } } @@ -1442,6 +1398,12 @@ async fn dispatch_group_members_event( /// Events are stored channel-scoped (`channel_id = Some(...)`) so that existing /// access control applies — private channel member lists are only visible to members. /// +/// The channel row and roster are captured behind the group-state locks +/// (see [`buzz_db::channel::LockedMemberSnapshot`]) and all three events +/// replace their heads on that one transaction. Every channel writer commits +/// before it calls this, so an emission that captured older state cannot +/// replace one that captured the writer's change. +/// /// NOTE: Channel-scoped storage means live global subscriptions (e.g. `{kinds:[39000]}`) /// won't receive these events via fan-out. Clients discover groups via historical REQ /// queries. Live push for open-channel discovery is a future enhancement. @@ -1450,67 +1412,50 @@ pub async fn emit_group_discovery_events( state: &Arc, channel_id: Uuid, ) -> anyhow::Result<()> { - let channel = state - .db - .get_channel_for_event_write(tenant.community(), channel_id) - .await?; - let members = state - .db - .get_members_for_event_write(tenant.community(), channel_id) - .await?; - - let relay_pubkey_hex = hex::encode(state.relay_keypair.public_key().to_bytes()); + let relay_pubkey = state.relay_keypair.public_key().to_bytes(); + let relay_pubkey_hex = hex::encode(relay_pubkey); let group_id = channel_id.to_string(); - { - let tags = buzz_db::channel::group_metadata_tags(&channel, &members) - .into_iter() - .map(Tag::parse) - .collect::, _>>()?; - emit_addressable_discovery_event( - tenant, - state, - channel_id, - KIND_NIP29_GROUP_METADATA, - tags, - &relay_pubkey_hex, - ) + let mut snapshot = state + .db + .lock_member_snapshot(tenant.community(), channel_id, &relay_pubkey) .await?; - } + let channel = snapshot + .channel + .clone() + .ok_or(buzz_db::DbError::ChannelNotFound(channel_id))?; + #[cfg(test)] + crate::nip_fi_test_hooks::after_group_state_capture(tenant.community()).await; + let metadata_tags = buzz_db::channel::group_metadata_tags(&channel, &snapshot.members) + .into_iter() + .map(Tag::parse) + .collect::, _>>()?; + let mut admin_tags: Vec = vec![Tag::parse(["d", &group_id])?]; + admin_tags.extend(group_state_identity_tags(&channel.identity())?); + for m in snapshot + .members + .iter() + .filter(|m| m.role == "owner" || m.role == "admin") { - let mut tags: Vec = vec![Tag::parse(["d", &group_id])?]; - tags.extend(group_state_identity_tags(&channel.identity())?); - for m in members - .iter() - .filter(|m| m.role == "owner" || m.role == "admin") - { - let pubkey_hex = hex::encode(&m.pubkey); - tags.push(Tag::parse(["p", &pubkey_hex, &m.role])?); - } - emit_addressable_discovery_event( - tenant, - state, - channel_id, - KIND_NIP29_GROUP_ADMINS, - tags, - &relay_pubkey_hex, - ) - .await?; + let pubkey_hex = hex::encode(&m.pubkey); + admin_tags.push(Tag::parse(["p", &pubkey_hex, &m.role])?); } - // Re-capture membership behind the writer lock immediately before the - // authoritative 39002 replacement. Metadata/admin snapshots retain their - // existing behavior; only membership publication needs this freshness fence. - let relay_pubkey = state.relay_keypair.public_key().to_bytes(); - let mut member_snapshot = state - .db - .lock_member_snapshot(tenant.community(), channel_id, &relay_pubkey) - .await?; - let stored_members = store_group_members_event(state, &mut member_snapshot).await?; - member_snapshot.release().await?; - dispatch_group_members_event(tenant, state, stored_members, &relay_pubkey_hex).await; + let mut stored = Vec::with_capacity(3); + for (kind, tags) in [ + (KIND_NIP29_GROUP_METADATA, metadata_tags), + (KIND_NIP29_GROUP_ADMINS, admin_tags), + ] { + let builder = EventBuilder::new(Kind::Custom(kind as u16), "").tags(tags); + stored.push(store_group_state_event(state, &mut snapshot, kind, builder).await?); + } + stored.push(store_group_members_event(state, &mut snapshot).await?); + snapshot.release().await?; + for event in stored { + dispatch_group_state_event(tenant, state, event, &relay_pubkey_hex).await; + } Ok(()) } @@ -4086,7 +4031,7 @@ pub async fn reconcile_large_channel_member_snapshots( let tenant = TenantContext::resolved(candidate.community_id, candidate.host.clone()); let stored_members = store_group_members_event(state, &mut member_snapshot).await?; member_snapshot.release().await?; - dispatch_group_members_event(&tenant, state, stored_members, &relay_pubkey_hex).await; + dispatch_group_state_event(&tenant, state, stored_members, &relay_pubkey_hex).await; Ok::(true) } .await; @@ -4371,7 +4316,7 @@ pub async fn publish_dm_visibility_snapshot( // Force created_at strictly past any prior snapshot for this viewer: a same-second // replacement whose random event id sorts higher is rejected by stale-write // protection, so a hide→re-open within one second could otherwise strand the stale - // snapshot. Same guard as emit_addressable_discovery_event. + // snapshot. Same guard as store_group_state_event. let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() @@ -4760,4 +4705,134 @@ mod tests { assert!(actor_is_channel_owner_or_admin(&members, &actor)); } + + mod postgres_tests { + use super::*; + + /// The `auto_join` tag of the channel's live kind:39000 head. + async fn metadata_head_has_auto_join( + state: &Arc, + community: buzz_core::tenant::CommunityId, + channel_id: Uuid, + ) -> bool { + let heads = state + .db + .query_events_for_event_write(&buzz_db::event::EventQuery { + kinds: Some(vec![KIND_NIP29_GROUP_METADATA as i32]), + channel_id: Some(channel_id), + limit: Some(1), + ..buzz_db::event::EventQuery::for_community(community) + }) + .await + .expect("read metadata head"); + let head = heads.first().expect("a metadata head"); + head.event + .tags + .iter() + .any(|tag| tag.as_slice() == ["auto_join"]) + } + + /// A membership emission that captured the channel before a kind:9002 + /// edit committed must not replace the edit's own emission: the edit's + /// emission waits for the held capture, then reads and publishes the + /// edited row last. Before the fence, the edit's emission published + /// first and the held one replaced it with the stale `auto_join` tag. + /// + /// Mutation: restore the unlocked channel read and per-kind replacement + /// of 39000 in `emit_group_discovery_events` (the hook between them) → + /// the edit's emission finishes while the capture is held → RED. + #[tokio::test] + #[ignore = "requires Postgres — runs in postgres-ci nextest lane"] + async fn held_group_state_capture_cannot_replace_a_newer_metadata_edit() { + use crate::nip_fi_test_hooks::group_state_capture_hook; + use buzz_db::channel::{ChannelType, ChannelUpdate, ChannelVisibility}; + use std::time::Duration; + + let pool = sqlx::PgPool::connect(&crate::test_support::database_url()) + .await + .expect("PostgreSQL must be available — set BUZZ_TEST_DATABASE_URL"); + let state = crate::state::tests::test_state_with_database_pool(pool).await; + let community_id = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community_id) + .bind(format!("group-state-{}.example", community_id.simple())) + .execute(state.db.pool()) + .await + .expect("insert community"); + let community = buzz_core::tenant::CommunityId::from_uuid(community_id); + let tenant = TenantContext::resolved(community, "test.local".to_string()); + let owner = nostr::Keys::generate(); + state + .db + .ensure_user_for_authorization(community, owner.public_key().as_bytes()) + .await + .expect("seed owner"); + let channel_id = state + .db + .create_channel( + community, + "general", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &owner.public_key().to_bytes(), + None, + ) + .await + .expect("create channel") + .id; + let set_auto_join = |value: bool| { + let state = Arc::clone(&state); + async move { + state + .db + .update_channel( + community, + channel_id, + ChannelUpdate { + auto_join: Some(value), + ..Default::default() + }, + ) + .await + .expect("update auto_join"); + } + }; + set_auto_join(true).await; + emit_group_discovery_events(&tenant, &state, channel_id) + .await + .expect("baseline emission"); + assert!(metadata_head_has_auto_join(&state, community, channel_id).await); + + let (captured, release) = group_state_capture_hook::arm(community); + let held = tokio::spawn({ + let (tenant, state) = (tenant.clone(), Arc::clone(&state)); + async move { emit_group_discovery_events(&tenant, &state, channel_id).await } + }); + tokio::time::timeout(Duration::from_secs(10), captured) + .await + .expect("the held emission reaches its capture") + .expect("capture hook dropped"); + + set_auto_join(false).await; + let mut edit = tokio::spawn({ + let (tenant, state) = (tenant.clone(), Arc::clone(&state)); + async move { emit_group_discovery_events(&tenant, &state, channel_id).await } + }); + assert!( + tokio::time::timeout(Duration::from_secs(2), &mut edit) + .await + .is_err(), + "the edit's emission must wait for the held capture" + ); + + release.notify_one(); + held.await.expect("held task").expect("held emission"); + edit.await.expect("edit task").expect("edit emission"); + assert!( + !metadata_head_has_auto_join(&state, community, channel_id).await, + "the edited row is the live kind:39000 head" + ); + } + } } diff --git a/crates/buzz-relay/src/nip_fi_test_hooks.rs b/crates/buzz-relay/src/nip_fi_test_hooks.rs index b37209b451d..fa4f32cf28f 100644 --- a/crates/buzz-relay/src/nip_fi_test_hooks.rs +++ b/crates/buzz-relay/src/nip_fi_test_hooks.rs @@ -211,6 +211,11 @@ make_hook!(auto_join_channel_hook, before_auto_join_channel); // `after_auto_join`: fires when an admission auto-join task has finished, so // a test can assert what it did (or did not) write. make_hook!(auto_join_done_hook, after_auto_join); +// `after_group_state_capture`: fires in `emit_group_discovery_events` once the +// channel row and roster are captured behind the group-state locks, before +// the events are built and replaced. Holding it lets a test commit a channel +// edit and start that edit's own emission while this one holds older state. +make_hook!(group_state_capture_hook, after_group_state_capture); // `before_http_ban_recheck`: fires in `buzz_v1` and Git settings after the // shared membership check passes, immediately before the fresh ban reread. // A ban or closed pool here makes only the reread refuse.