jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-30 22:11:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

Make moderation discovery retries convergent

Browse Source 
Signed-off-by: kenny lopez <klopez4212@gmail.com>
This commit is contained in:
kenny lopez
2026-08-21 14:29:42 +01:00
parent 896378c902
commit b0a324357f
4 changed files with 167 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

104
crates/buzz-relay/src/handlers/moderation_notices.rs
View File

@@ -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=<dm_channel_id>` and a
// 3. Insert the relay-signed kind:9 notice with `h=<dm_channel_id>` 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<i32> = 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<i32> = 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(),

31
crates/buzz-relay/src/handlers/side_effects.rs
View File

@@ -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<Tag> = 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(())
}

15
mobile/lib/features/channels/channels_provider.dart
View File

@@ -40,6 +40,7 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
void Function()? _unsubscribeDmVisibility;
String? _dmVisibilityRelayBaseUrl;
String? _dmVisibilityPubkey;
String? _latestDmVisibilityEventId;
Future<void> _liveSubscriptionQueue = Future.value();
List<Channel> _desiredLiveChannels = const [];
Set<String> _desiredLiveChannelIds = const {};
@@ -83,6 +84,7 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
_memberSnapshotRelayBaseUrl = relayBaseUrl;
_memberSnapshotPubkey = pubkey;
_memberSnapshotsByChannelId = const {};
_latestDmVisibilityEventId = null;
}
final connected = Completer<void>();
final sessionState = ref.read(relaySessionProvider);
@@ -464,13 +466,17 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
) 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<List<Channel>> {
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<List<Channel>> {
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<List<Channel>> {
_unsubscribeDmVisibility = null;
_dmVisibilityRelayBaseUrl = null;
_dmVisibilityPubkey = null;
_latestDmVisibilityEventId = null;
_backstopTimer?.cancel();
_backstopTimer = null;
}

45
mobile/test/features/channels/channels_provider_test.dart
View File

@@ -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;