External/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-08-18 06:50:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix(relay): move fleet-wide NIP-43 membership sweep off the pre-bind startup path

Browse Source 
The relay awaited reconcile_nip43_membership_snapshots for EVERY
provisioned community before binding the listener. The sweep is
sequential, two queries per community plus a publication per repair
(~9.5ms/community measured locally). Cost is linear in community count,
so on a deployment with many communities the pre-bind sweep takes
minutes — far past a typical startup probe window. The probe SIGKILLs
the pod mid-sweep, the restart begins again at community number one,
and the pod crashloops forever: low RSS, port never bound, zero log
lines at RUST_LOG=error.

Fix, minimal by design:
- Pre-bind: reconcile ONLY the deployment community (one snapshot check,
  bounded), with explicit phase logs carrying elapsed_ms.
- The fleet-wide sweep moves entirely into the periodic post-bind task,
  now jittered (random fraction of the interval, same rationale as the
  usage-metrics poller) and leader-gated via a session advisory lock so
  N replicas do not run N concurrent fleet-wide sweeps. Lock is held
  only for the duration of a sweep; a replica dying mid-sweep releases
  it with its session.
- Sweep telemetry: start/progress(1000)/complete/fail logs with
  elapsed_ms, plus a buzz_nip43_membership_sweep_seconds histogram.

Pagination/resume and per-query deadlines are a deliberate follow-up PR.

Verified: cargo test -p buzz-relay — 837 passed, 1 pre-existing failure
(api::mesh_demo::tests::demo_join_forwarded_arm_round_trips_echo) which
fails identically at origin/main 5e0efb0bb with this diff stashed.

Co-authored-by: npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz>
Signed-off-by: npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz>
This commit is contained in:
npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d
2026-08-03 12:31:08 -04:00
parent 5e0efb0bb9
commit f0dfd21bbb
2 changed files with 128 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/buzz-relay/src/handlers/side_effects.rs
+50 -24
View File
@@ -2872,39 +2872,53 @@ async fn emit_initial_ref_state(
Ok(())
}
/// Reconcile one community's event-backed NIP-43 membership view.
///
/// `relay_members` is canonical. The snapshot is rebuilt only when it is
/// absent or its member/role set differs from the canonical rows, so this is
/// cheap (two queries) when nothing changed. Returns whether a repair
/// publication happened.
pub async fn reconcile_nip43_membership_snapshot(
tenant: &TenantContext,
state: &Arc<AppState>,
) -> anyhow::Result<bool> {
if !state
.db
.nip43_membership_snapshot_needs_reconciliation(
tenant.community(),
&state.relay_keypair.public_key(),
)
.await?
{
return Ok(false);
}
publish_nip43_membership_list(tenant, state).await?;
Ok(true)
}
/// Reconcile every community's event-backed NIP-43 membership view.
///
/// `relay_members` is canonical. A snapshot is rebuilt only when it is absent
/// or its member/role set differs from the canonical rows. This makes the sweep
/// safe to run at startup and periodically without producing an event stream
/// when nothing changed. A failure in one community is logged and counted but
/// does not prevent the remaining communities from being repaired.
/// A failure in one community is logged and counted but does not prevent the
/// remaining communities from being repaired.
///
/// This sweep is O(communities) — sequential, two queries each, plus a
/// publication per repair. On a large deployment it takes minutes, so it MUST
/// NOT run on the pre-bind startup path (a startup probe SIGKILLs the pod
/// mid-sweep and the restart begins again at community #1, forever). It runs
/// post-bind, jittered, and leader-gated — see the sweep task in `main.rs`.
pub async fn reconcile_nip43_membership_snapshots(state: &Arc<AppState>) -> anyhow::Result<usize> {
let started_at = std::time::Instant::now();
let communities = state.db.usage_community_hosts().await?;
let total = communities.len();
let mut scanned = 0usize;
let mut reconciled = 0usize;
for community in communities {
let community_id = buzz_core::CommunityId::from_uuid(community.id);
let host = community.host;
let result = async {
if !state
.db
.nip43_membership_snapshot_needs_reconciliation(
community_id,
&state.relay_keypair.public_key(),
)
.await?
{
return Ok::<bool, anyhow::Error>(false);
}
let tenant = TenantContext::resolved(community_id, host.clone());
publish_nip43_membership_list(&tenant, state).await?;
Ok::<bool, anyhow::Error>(true)
}
.await;
match result {
let tenant = TenantContext::resolved(community_id, host.clone());
match reconcile_nip43_membership_snapshot(&tenant, state).await {
Ok(true) => reconciled += 1,
Ok(false) => {}
Err(error) => {
@@ -2913,9 +2927,21 @@ pub async fn reconcile_nip43_membership_snapshots(state: &Arc<AppState>) -> anyh
warn!(%community_id, %host, %error, "NIP-43 membership reconciliation failed");
}
}
scanned += 1;
if scanned.is_multiple_of(1000) {
info!(
scanned,
total,
reconciled,
elapsed_ms = started_at.elapsed().as_millis() as u64,
"NIP-43 membership sweep progress"
);
}
}
metrics::counter!("buzz_nip43_membership_reconciliations_total").increment(reconciled as u64);
metrics::histogram!("buzz_nip43_membership_sweep_seconds")
.record(started_at.elapsed().as_secs_f64());
Ok(reconciled)
}
crates/buzz-relay/src/main.rs
+78 -14
View File
@@ -83,6 +83,13 @@ impl EmissionScope {
const USAGE_METRICS_LOCK_KEY: i64 = 0x4255_5A5A_4D45_5452;
/// Session advisory lock serializing the fleet-wide NIP-43 membership sweep.
///
/// The sweep is O(communities) — minutes at fleet scale — so only one replica
/// may run it at a time; the rest skip their tick. Same detached-session
/// leadership mechanism as the usage-metrics poller, different key.
const NIP43_SWEEP_LOCK_KEY: i64 = 0x4255_5A5A_4E50_3433;
#[tokio::main]
async fn main() -> anyhow::Result<()> {
// Install the ring CryptoProvider for rustls. Required before any rustls
@@ -527,16 +534,35 @@ async fn main() -> anyhow::Result<()> {
);
}
// NIP-43: reconcile the event-backed roster for every provisioned
// community before opening the listener. `relay_members` is canonical;
// this repairs pre-snapshot communities and any publication that failed
// after a membership transaction committed.
// NIP-43: reconcile the event-backed roster for the deployment's own
// community before opening the listener. The fleet-wide sweep across every
// provisioned community deliberately does NOT run here: it is
// O(communities) — minutes at fleet scale — and a startup probe that
// SIGKILLs the pod mid-sweep restarts it from community #1 forever
// (permanent crashloop). The fleet sweep runs post-bind, jittered, and
// leader-gated in the periodic task spawned below.
if config.require_relay_membership {
match buzz_relay::handlers::side_effects::reconcile_nip43_membership_snapshots(&state).await
{
Ok(count) => info!(count, "NIP-43 membership snapshots reconciled on startup"),
Err(error) => {
tracing::warn!(%error, "NIP-43 membership snapshot startup reconciliation failed")
// `deployment_community` is always Some here: startup fails fast above
// when membership is enforced and the community cannot be ensured.
if let Some(community) = deployment_community {
let host = buzz_relay::tenant::relay_url_authority(&config.relay_url);
let tenant = buzz_core::tenant::TenantContext::resolved(community, host);
let started_at = std::time::Instant::now();
info!(community = %community, "NIP-43 startup phase: reconciling deployment community snapshot");
match buzz_relay::handlers::side_effects::reconcile_nip43_membership_snapshot(
&tenant, &state,
)
.await
{
Ok(repaired) => info!(
community = %community,
repaired,
elapsed_ms = started_at.elapsed().as_millis() as u64,
"NIP-43 startup phase: deployment community snapshot reconciled"
),
Err(error) => {
tracing::warn!(%error, "NIP-43 deployment community startup reconciliation failed")
}
}
}
@@ -547,24 +573,62 @@ async fn main() -> anyhow::Result<()> {
.unwrap_or(60)
.max(1);
tokio::spawn(async move {
// Jitter the first tick by a random fraction of the interval so a
// rolling deploy of N pods doesn't contend for the sweep lock (and
// hammer `communities`) simultaneously at boot. True per-process
// randomness — PID-derived seeds are unsafe when every pod is PID 1.
let jitter_secs = rand::random::<u64>() % interval_secs;
tokio::time::sleep(std::time::Duration::from_secs(jitter_secs)).await;
let mut interval = tokio::time::interval(std::time::Duration::from_secs(interval_secs));
interval.tick().await;
// A fleet-scale sweep takes longer than the interval; skip ticks
// rather than scheduling a catch-up burst behind it.
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
loop {
interval.tick().await;
// Per-tick leadership: hold the sweep lock only while sweeping,
// so a replica that dies mid-sweep releases it with its session
// and any peer picks up on its next tick.
let leader = match reconcile_state
.db
.try_lock_usage_metrics(NIP43_SWEEP_LOCK_KEY)
.await
{
Ok(Some(guard)) => guard,
Ok(None) => continue,
Err(error) => {
tracing::warn!(%error, "NIP-43 sweep leadership check failed");
continue;
}
};
let started_at = std::time::Instant::now();
info!("NIP-43 membership sweep starting (leader)");
match buzz_relay::handlers::side_effects::reconcile_nip43_membership_snapshots(
&reconcile_state,
)
.await
{
Ok(count) if count > 0 => {
info!(count, "NIP-43 membership snapshots repaired")
Ok(count) => {
let elapsed_ms = started_at.elapsed().as_millis() as u64;
if count > 0 {
info!(
count,
elapsed_ms, "NIP-43 membership sweep complete: snapshots repaired"
)
} else {
info!(
elapsed_ms,
"NIP-43 membership sweep complete: nothing to repair"
)
}
}
Ok(_) => {}
Err(error) => tracing::warn!(
%error,
"periodic NIP-43 membership snapshot reconciliation failed"
elapsed_ms = started_at.elapsed().as_millis() as u64,
"NIP-43 membership sweep failed"
),
}
drop(leader);
}
});
}