fix(relay): structurally gate NIP-43 fleet sweep on listener bind

Review found the sweep-after-bind ordering was probabilistic: the sweep
task is spawned before serve(), zero jitter is a valid draw, and
interval.tick() fires immediately — so the fleet sweep could still start
pre-bind. Replace the timing assumption with a structural guarantee:

- new after_listener_bound(oneshot::Receiver, task) helper; the gated
  task runs only after the signal fires and never runs if the sender is
  dropped (startup failed before bind)
- serve() takes a listener_bound_tx: oneshot::Sender<()> and fires it
  immediately after the main TCP listener binds
- the fleet sweep task is wrapped in after_listener_bound; jitter is now
  explicitly only lock-contention spreading, not ordering

Adds two deterministic tests: a yield-storm test proving the task cannot
run before the signal even with zero delay, and a dropped-sender test
proving the task never runs when startup fails pre-bind.

Co-authored-by: npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz>
Signed-off-by: npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz>

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<F, Fut>(bound: tokio::sync::oneshot::Receiver<()>, task: F)
+where
+    F: FnOnce() -> Fut,
+    Fut: std::future::Future<Output = ()>,
+{
+    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::<u64>().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::<u64>() % 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<AppState>,
+    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 {