fix(relay): refresh members after discovery contention

Publish the independently locked member snapshot before reporting exhausted metadata retries; competing producers may stop before their own roster refresh.

Co-authored-by: Honor <e0bf479eeebe47dde1e35705212a4df70616a7bc834d907ddfb18cf0405f8b1b@seiler-test.communities.public.builderlabstaging.xyz>
Signed-off-by: Brad Seiler <seiler@squareup.com>
This commit is contained in:
Brad Seiler
2026-11-04 15:23:24 -05:00
parent 638c65ccc6
commit 070b77b00e
2 changed files with 101 additions and 3 deletions

View File

@@ -95,3 +95,98 @@ async fn delayed_discovery_cannot_restore_stale_metadata() {
"delayed discovery restored auto_join after the accepted false edit"
);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn discovery_exhaustion_still_refreshes_members() {
let state = state().await;
let tenant = community(&state, "discovery-exhaustion").await;
let owner = Keys::generate();
let channel = create_channel(&state, &tenant, &owner, &[]).await;
let member = Keys::generate().public_key();
state
.db
.add_member(
tenant.community(),
channel,
&member.to_bytes(),
buzz_core::channel::MemberRole::Member,
None,
)
.await
.unwrap();
let lists_member = |tags: &[Vec<String>]| {
tags.iter().any(|tag| {
tag.first().map(String::as_str) == Some("p") && tag.get(1) == Some(&member.to_hex())
})
};
assert!(!lists_member(
&group_state_tags(&state, &tenant, channel, 39002).await
));
let arm_pause = || {
let (captured, did_capture) = oneshot::channel();
let (resume, resumed) = oneshot::channel();
pauses()
.lock()
.unwrap()
.insert(channel, (captured, resumed));
(did_capture, resume)
};
let mut pause = Some(arm_pause());
let producer = tokio::spawn({
let state = state.clone();
let tenant = tenant.clone();
async move { emit_group_discovery_events(&tenant, &state, channel).await }
});
for attempt in 0..8 {
let (captured, resume) = pause.take().unwrap();
tokio::time::timeout(std::time::Duration::from_secs(10), captured)
.await
.expect("producer captured on every attempt")
.unwrap();
let heads = state
.db
.query_events_for_event_write(&buzz_db::event::EventQuery {
kinds: Some(vec![39000]),
authors: Some(vec![state.relay_keypair.public_key().to_bytes().to_vec()]),
channel_id: Some(channel),
limit: Some(1),
..buzz_db::event::EventQuery::for_community(tenant.community())
})
.await
.unwrap();
let head = &heads[0].event;
// Publish only metadata, modelling a competing producer that disappears
// before its independent member-list refresh. Every expected head loses.
let competitor = EventBuilder::new(Kind::Custom(39000), "")
.tags(head.tags.iter().cloned())
.custom_created_at(nostr::Timestamp::from(head.created_at.as_secs() + 1))
.sign_with_keys(&state.relay_keypair)
.unwrap();
assert!(
state
.db
.replace_addressable_event(tenant.community(), &competitor, Some(channel))
.await
.unwrap()
.1
);
if attempt < 7 {
pause = Some(arm_pause());
}
resume.send(()).unwrap();
}
let result = tokio::time::timeout(std::time::Duration::from_secs(10), producer)
.await
.unwrap()
.unwrap();
assert!(result
.unwrap_err()
.to_string()
.contains("repeatedly changed"));
assert!(
lists_member(&group_state_tags(&state, &tenant, channel, 39002).await),
"exhausted metadata retries skipped the independent member-list refresh"
);
}

View File

@@ -1758,9 +1758,6 @@ pub async fn emit_group_discovery_events(
break;
}
}
if !published {
anyhow::bail!("group discovery publication repeatedly changed for {channel_id}");
}
let relay_pubkey_hex = hex::encode(state.relay_keypair.public_key().to_bytes());
// Re-capture membership behind the writer lock immediately before the
@@ -1775,6 +1772,12 @@ pub async fn emit_group_discovery_events(
member_snapshot.release().await?;
dispatch_group_members_event(tenant, state, stored_members, &relay_pubkey_hex).await;
// Metadata/admin contention must not skip the independent member snapshot.
// Winning producers can disappear before they publish their own 39002.
if !published {
anyhow::bail!("group discovery publication repeatedly changed for {channel_id}");
}
Ok(())
}