diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 543641e76..4a223a27b 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -90,6 +90,24 @@ const USAGE_METRICS_LOCK_KEY: i64 = 0x4255_5A5A_4D45_5452; /// leadership mechanism as the usage-metrics poller, different key. const NIP43_SWEEP_LOCK_KEY: i64 = 0x4255_5A5A_4E50_3433; +/// Run `task` only after the listener-bound signal fires. +/// +/// This is the structural guarantee that background work gated on it cannot +/// precede `serve()` binding the listeners — a jitter delay alone is +/// probabilistic (zero is a valid draw) and does not establish the ordering. +/// If the sender is dropped without firing (startup failed before bind), the +/// task never runs. +async fn after_listener_bound(bound: tokio::sync::oneshot::Receiver<()>, task: F) +where + F: FnOnce() -> Fut, + Fut: std::future::Future, +{ + if bound.await.is_err() { + return; + } + task().await; +} + #[tokio::main] async fn main() -> anyhow::Result<()> { // Install the ring CryptoProvider for rustls. Required before any rustls @@ -539,8 +557,11 @@ async fn main() -> anyhow::Result<()> { // 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 + // (permanent crashloop). The fleet sweep runs post-bind (structurally: + // gated on the listener-bound signal fired inside `serve`), jittered, and // leader-gated in the periodic task spawned below. + let (listener_bound_tx, listener_bound_rx) = tokio::sync::oneshot::channel::<()>(); + let mut listener_bound_rx = Some(listener_bound_rx); if config.require_relay_membership { // `deployment_community` is always Some here: startup fails fast above // when membership is enforced and the community cannot be ensured. @@ -572,11 +593,16 @@ async fn main() -> anyhow::Result<()> { .and_then(|value| value.parse::().ok()) .unwrap_or(60) .max(1); - tokio::spawn(async move { + let sweep_bound_rx = listener_bound_rx + .take() + .expect("listener_bound_rx is consumed exactly once, here"); + tokio::spawn(after_listener_bound(sweep_bound_rx, move || 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. + // Ordering vs bind is NOT the jitter's job: `after_listener_bound` + // already guarantees this task starts only after `serve()` binds. let jitter_secs = rand::random::() % interval_secs; tokio::time::sleep(std::time::Duration::from_secs(jitter_secs)).await; @@ -630,7 +656,7 @@ async fn main() -> anyhow::Result<()> { } drop(leader); } - }); + })); } // Emit kind:39000/39002 discovery events for channels that exist in the DB @@ -1137,7 +1163,7 @@ async fn main() -> anyhow::Result<()> { }); } - serve(router, health_router, Arc::clone(&state)).await?; + serve(router, health_router, Arc::clone(&state), listener_bound_tx).await?; state.community_revalidator_cancel.cancel(); // Signal the audit worker to stop accepting, flush buffered entries, and @@ -1157,6 +1183,59 @@ async fn main() -> anyhow::Result<()> { Ok(()) } +#[cfg(test)] +mod after_listener_bound_tests { + use super::after_listener_bound; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::Arc; + + /// The invariant PR review demanded be structural: the gated task must not + /// start before the bound signal, even with zero delay anywhere. Uses + /// `tokio::task::yield_now` generously so any "task runs immediately on + /// spawn" regression is caught deterministically, not probabilistically. + #[tokio::test] + async fn task_does_not_run_until_listener_bound_signal() { + let (tx, rx) = tokio::sync::oneshot::channel::<()>(); + let ran = Arc::new(AtomicBool::new(false)); + let ran_in_task = Arc::clone(&ran); + let handle = tokio::spawn(after_listener_bound(rx, move || async move { + ran_in_task.store(true, Ordering::SeqCst); + })); + + // Give the spawned task every chance to (incorrectly) run early. + for _ in 0..64 { + tokio::task::yield_now().await; + } + assert!( + !ran.load(Ordering::SeqCst), + "sweep task ran before the listener-bound signal" + ); + + tx.send(()).expect("receiver alive"); + handle.await.expect("gated task completes"); + assert!(ran.load(Ordering::SeqCst), "task must run after the signal"); + } + + /// Startup that fails before bind drops the sender; the gated task must + /// never run in that case (no sweep against a relay that never served). + #[tokio::test] + async fn task_never_runs_if_sender_dropped_before_bind() { + let (tx, rx) = tokio::sync::oneshot::channel::<()>(); + let ran = Arc::new(AtomicBool::new(false)); + let ran_in_task = Arc::clone(&ran); + let handle = tokio::spawn(after_listener_bound(rx, move || async move { + ran_in_task.store(true, Ordering::SeqCst); + })); + + drop(tx); + handle.await.expect("gated task exits cleanly"); + assert!( + !ran.load(Ordering::SeqCst), + "task must not run when startup fails before bind" + ); + } +} + #[cfg(test)] mod env_filter_tests { use super::log_env_filter; @@ -1257,6 +1336,7 @@ async fn serve( router: axum::Router, health_router: axum::Router, state: Arc, + listener_bound_tx: tokio::sync::oneshot::Sender<()>, ) -> anyhow::Result<()> { let config = &state.config; @@ -1300,6 +1380,11 @@ async fn serve( .await .map_err(|e| anyhow::anyhow!("Failed to bind {}: {e}", config.bind_addr))?; info!(addr = %config.bind_addr, "buzz-relay TCP listening"); + // Release background work gated on the listener being bound (NIP-43 fleet + // sweep). Fired only here — if startup fails before this point the sender + // drops and the gated tasks never run. A receiver dropped earlier (e.g. + // membership disabled) is fine; ignore the send result. + let _ = listener_bound_tx.send(()); #[cfg(unix)] if let Some(ref uds_path) = config.uds_path {