From b0a324357ffab6be41fdf7dbcfddd9641beb81e3 Mon Sep 17 00:00:00 2001 From: kenny lopez Date: Fri, 21 Aug 2026 14:29:42 +0100 Subject: [PATCH] Make moderation discovery retries convergent Signed-off-by: kenny lopez --- .../src/handlers/moderation_notices.rs | 104 ++++++++++++++++-- .../buzz-relay/src/handlers/side_effects.rs | 31 +++--- .../features/channels/channels_provider.dart | 15 ++- .../channels/channels_provider_test.dart | 45 ++++++++ 4 files changed, 167 insertions(+), 28 deletions(-) diff --git a/crates/buzz-relay/src/handlers/moderation_notices.rs b/crates/buzz-relay/src/handlers/moderation_notices.rs index fe41e65c65..d5ab303ba8 100644 --- a/crates/buzz-relay/src/handlers/moderation_notices.rs +++ b/crates/buzz-relay/src/handlers/moderation_notices.rs @@ -118,6 +118,11 @@ pub async fn send_moderation_notice( ) .await? { + // A prior attempt may have committed the notice before a + // transient discovery failure. Duplicate delivery is still a + // content no-op, but retry the idempotent discovery generation so + // the channel eventually has a complete 39000/39001/39002 set. + emit_group_discovery_events(tenant, state, existing_dm.id).await?; return Ok(()); } } @@ -146,20 +151,12 @@ pub async fn send_moderation_notice( .increment(1); } - // 2. Ensure the relay's "{host} Moderation" kind:0 profile exists, and 3. - // the DM's kind:39000 discovery (with `hidden` / `t=dm` / `p`). Both are - // replaceable events, so we emit them on EVERY send rather than gating on - // first creation: if discovery failed on the first delivery (it is - // `?`-propagated), a `was_created`-gated retry would skip it forever and - // leave the thread permanently undiscoverable — a notice delivered into a - // channel no client can render. Notices are rare; unconditional re-emit is - // cheap and `replace_addressable_event` makes it idempotent. + // 2. Ensure the relay's "{host} Moderation" kind:0 profile exists. if let Err(e) = publish_moderation_profile(tenant, state, &relay_pubkey_hex).await { warn!(error = %e, "moderation profile publish failed (continuing)"); } - emit_group_discovery_events(tenant, state, dm_channel_id).await?; - // 4. Insert the relay-signed kind:9 notice with `h=` and a + // 3. Insert the relay-signed kind:9 notice with `h=` and a // `moderation_source` tag naming the source row id (idempotency + // client linking). let tags = vec![ @@ -197,6 +194,14 @@ pub async fn send_moderation_notice( let kind_u32 = event_kind_u32(&stored.event); dispatch_persistent_event(tenant, state, &stored, kind_u32, &relay_pubkey_hex, None).await; + // 4. Publish the idempotent discovery generation after the notice is + // durable. The roster is established first inside the helper, so a roster + // failure cannot leave metadata/admin heads without kind 39002. Propagate + // any remaining failure after the notice commit; an idempotent caller retry + // takes the duplicate path above and repairs discovery without reinserting + // or resurfacing the notice. + emit_group_discovery_events(tenant, state, dm_channel_id).await?; + Ok(()) } @@ -546,6 +551,20 @@ mod tests { hidden_after_failure, "failed notice insertion must not expose stale moderation content" ); + let discovery_after_failure: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM events \ + WHERE community_id = $1 AND channel_id = $2 \ + AND kind IN (39000, 39001, 39002) AND deleted_at IS NULL", + ) + .bind(tenant.community().as_uuid()) + .bind(preexisting_dm.id) + .fetch_one(&pool) + .await + .expect("count discovery after failed notice"); + assert_eq!( + discovery_after_failure, 0, + "a rejected notice must not publish a partial discovery generation" + ); sqlx::raw_sql( "DROP TRIGGER fail_moderation_notice_insert ON events; \ DROP FUNCTION fail_moderation_notice_insert();", @@ -554,9 +573,72 @@ mod tests { .await .expect("remove notice failure injection"); + sqlx::raw_sql( + "CREATE FUNCTION fail_moderation_metadata_insert() RETURNS trigger \ + LANGUAGE plpgsql AS $$ BEGIN \ + IF NEW.kind = 39000 THEN RAISE EXCEPTION 'injected discovery failure'; END IF; \ + RETURN NEW; \ + END $$; \ + CREATE TRIGGER fail_moderation_metadata_insert \ + BEFORE INSERT ON events FOR EACH ROW \ + EXECUTE FUNCTION fail_moderation_metadata_insert();", + ) + .execute(&pool) + .await + .expect("install discovery failure injection"); send_moderation_notice(&tenant, &state, recipient_bytes.as_slice(), notice.clone()) .await - .expect("deliver notice"); + .expect_err("post-commit discovery failure must request a retry"); + let partial_discovery_kinds: Vec = sqlx::query_scalar( + "SELECT kind FROM events \ + WHERE community_id = $1 AND channel_id = $2 \ + AND kind IN (39000, 39001, 39002) AND deleted_at IS NULL \ + ORDER BY kind", + ) + .bind(tenant.community().as_uuid()) + .bind(preexisting_dm.id) + .fetch_all(&pool) + .await + .expect("read roster-first discovery state"); + assert_eq!( + partial_discovery_kinds, + vec![39002], + "metadata failure must not strand 39000/39001 without a roster" + ); + let notice_count_after_discovery_failure: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM events \ + WHERE community_id = $1 AND channel_id = $2 AND kind = 9 \ + AND deleted_at IS NULL", + ) + .bind(tenant.community().as_uuid()) + .bind(preexisting_dm.id) + .fetch_one(&pool) + .await + .expect("count durable notice after discovery failure"); + assert_eq!(notice_count_after_discovery_failure, 1); + sqlx::raw_sql( + "DROP TRIGGER fail_moderation_metadata_insert ON events; \ + DROP FUNCTION fail_moderation_metadata_insert();", + ) + .execute(&pool) + .await + .expect("remove discovery failure injection"); + + send_moderation_notice(&tenant, &state, recipient_bytes.as_slice(), notice.clone()) + .await + .expect("duplicate retry repairs discovery"); + let discovery_kinds: Vec = sqlx::query_scalar( + "SELECT kind FROM events \ + WHERE community_id = $1 AND channel_id = $2 \ + AND kind IN (39000, 39001, 39002) AND deleted_at IS NULL \ + ORDER BY kind", + ) + .bind(tenant.community().as_uuid()) + .bind(preexisting_dm.id) + .fetch_all(&pool) + .await + .expect("read converged moderation discovery"); + assert_eq!(discovery_kinds, vec![39000, 39001, 39002]); let participant_hash = buzz_db::dm::compute_participant_hash(&[ recipient_bytes.as_slice(), diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 0924074fd2..ac7922726c 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -1112,11 +1112,25 @@ pub async fn emit_group_discovery_events( channel_id: Uuid, ) -> anyhow::Result<()> { let channel = state.db.get_channel(tenant.community(), channel_id).await?; - let members = state.db.get_members(tenant.community(), channel_id).await?; - let relay_pubkey_hex = hex::encode(state.relay_keypair.public_key().to_bytes()); let group_id = channel_id.to_string(); + // Establish the authoritative roster first, from one membership snapshot + // held behind the writer lock. Metadata/admin discovery is emitted only + // after that transaction commits, so a roster validation failure cannot + // strand a partial 39000/39001 generation without kind 39002. Keeping the + // captured members for all three events also prevents cross-query drift. + 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 members = member_snapshot.members.clone(); + let stored_members = + store_group_members_event(tenant, state, channel_id, &mut member_snapshot).await?; + member_snapshot.release().await?; + dispatch_group_members_event(tenant, state, stored_members, &relay_pubkey_hex).await; + { let mut tags: Vec = vec![Tag::parse(["d", &group_id])?]; tags.push(Tag::parse(["name", &channel.name])?); @@ -1200,19 +1214,6 @@ pub async fn emit_group_discovery_events( .await?; } - // 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(tenant, state, channel_id, &mut member_snapshot).await?; - member_snapshot.release().await?; - dispatch_group_members_event(tenant, state, stored_members, &relay_pubkey_hex).await; - Ok(()) } diff --git a/mobile/lib/features/channels/channels_provider.dart b/mobile/lib/features/channels/channels_provider.dart index 56609908e9..4316391459 100644 --- a/mobile/lib/features/channels/channels_provider.dart +++ b/mobile/lib/features/channels/channels_provider.dart @@ -40,6 +40,7 @@ class ChannelsNotifier extends AsyncNotifier> { void Function()? _unsubscribeDmVisibility; String? _dmVisibilityRelayBaseUrl; String? _dmVisibilityPubkey; + String? _latestDmVisibilityEventId; Future _liveSubscriptionQueue = Future.value(); List _desiredLiveChannels = const []; Set _desiredLiveChannelIds = const {}; @@ -83,6 +84,7 @@ class ChannelsNotifier extends AsyncNotifier> { _memberSnapshotRelayBaseUrl = relayBaseUrl; _memberSnapshotPubkey = pubkey; _memberSnapshotsByChannelId = const {}; + _latestDmVisibilityEventId = null; } final connected = Completer(); final sessionState = ref.read(relaySessionProvider); @@ -464,13 +466,17 @@ class ChannelsNotifier extends AsyncNotifier> { ) async { try { final events = await session.fetchHistory(NostrFilters.hiddenDms(myPk)); - if (events.isEmpty) return const {}; + if (events.isEmpty) { + _latestDmVisibilityEventId = null; + return const {}; + } NostrEvent latest = events.first; for (final event in events.skip(1)) { if (event.createdAt > latest.createdAt) { latest = event; } } + _latestDmVisibilityEventId = latest.id; return { for (final tag in latest.tags) if (tag.length >= 2 && tag[0] == 'h') tag[1], @@ -601,7 +607,9 @@ class ChannelsNotifier extends AsyncNotifier> { tags: { '#p': [myPk], }, - limit: 0, + // Replay the latest snapshot before EOSE so a publication between + // the history fetch and this subscription cannot be missed. + limit: 1, ), _handleDmVisibilityEvent, ); @@ -808,6 +816,8 @@ class ChannelsNotifier extends AsyncNotifier> { void _handleDmVisibilityEvent(NostrEvent event) { if (event.kind != EventKind.dmVisibility) return; + if (_latestDmVisibilityEventId == event.id) return; + _latestDmVisibilityEventId = event.id; unawaited(refresh()); } @@ -949,6 +959,7 @@ class ChannelsNotifier extends AsyncNotifier> { _unsubscribeDmVisibility = null; _dmVisibilityRelayBaseUrl = null; _dmVisibilityPubkey = null; + _latestDmVisibilityEventId = null; _backstopTimer?.cancel(); _backstopTimer = null; } diff --git a/mobile/test/features/channels/channels_provider_test.dart b/mobile/test/features/channels/channels_provider_test.dart index 9b270e4a5d..97e230a55a 100644 --- a/mobile/test/features/channels/channels_provider_test.dart +++ b/mobile/test/features/channels/channels_provider_test.dart @@ -466,6 +466,7 @@ void main() { ); expect(session.dmVisibilitySubscribeFilters, hasLength(1)); expect(session.dmVisibilitySubscribeFilters.single.tags['#p'], [myPk]); + expect(session.dmVisibilitySubscribeFilters.single.limit, 1); final visibleSnapshot = _hiddenDms(const [], pubkey: myPk); session.hiddenDmEvents = [visibleSnapshot]; @@ -485,6 +486,45 @@ void main() { expect(session.dmVisibilitySubscribeFilters, hasLength(1)); }); + test( + 'visibility subscription replays a snapshot published during setup', + () async { + final session = _FakeRelaySession( + memberships: [ + _membership(_channelA, myPk), + _membership(_channelB, myPk), + ], + metadata: [ + _meta(id: _channelA, name: 'Alice', channelType: 'dm'), + _meta(id: _channelB, name: 'Bob', channelType: 'dm'), + ], + hiddenDmEvents: [ + _hiddenDms([_channelA], pubkey: myPk), + ], + )..pauseNextSubscribe(); + final container = _buildContainer(session: session); + addTearDown(container.dispose); + + final initialLoad = container.read(channelsProvider.future); + await session.nextSubscribeStarted; + session.hiddenDmEvents = [_hiddenDms(const [], pubkey: myPk)]; + session.resumePausedSubscribe(); + + await initialLoad; + await _waitUntil( + () => + container + .read(channelsProvider) + .value + ?.map((channel) => channel.id) + .toSet() + .containsAll({_channelA, _channelB}) == + true, + ); + expect(session.activeChannels, {_channelA, _channelB}); + }, + ); + test( 'archived kind:39000 metadata sets Channel.isArchived (covers TTL auto-archive)', () async { @@ -926,6 +966,11 @@ class _FakeRelaySession extends RelaySessionNotifier { } final subscriptionKey = ++_nextSubscriptionKey; _subscriptions[subscriptionKey] = (filter, onEvent); + if (filter.kinds.contains(EventKind.dmVisibility) && filter.limit > 0) { + for (final event in hiddenDmEvents.take(filter.limit)) { + onEvent(event); + } + } return () { final subscription = _subscriptions.remove(subscriptionKey); if (subscription == null) return;