diff --git a/crates/mobee-core/Cargo.toml b/crates/mobee-core/Cargo.toml index 74681a07..850937a2 100644 --- a/crates/mobee-core/Cargo.toml +++ b/crates/mobee-core/Cargo.toml @@ -22,7 +22,7 @@ git-delivery = ["gateway", "dep:git2", "dep:reqwest", "dep:base64"] # The `wallet` feature also brings up the persistent buyer daemon (crate::buyer): the # exclusive home lock (libc flock), the durable state DB (rusqlite, bundled sqlite so # there is no system dependency), and the async unix-socket surface (tokio net + io-util). -wallet = ["git-delivery", "gateway", "dep:cashu", "dep:cdk", "dep:cdk-sqlite", "dep:nostr-sdk", "dep:rusqlite", "dep:libc", "tokio/sync", "tokio/time", "tokio/net", "tokio/io-util"] +wallet = ["git-delivery", "gateway", "dep:cashu", "dep:cdk", "dep:cdk-sqlite", "dep:nostr-sdk", "dep:rusqlite", "dep:libc", "tokio/sync", "tokio/time", "tokio/net", "tokio/io-util", "tokio/signal"] test-support = [] [dependencies] diff --git a/crates/mobee-core/src/seller_node/buzz_wire_it.rs b/crates/mobee-core/src/seller_node/buzz_wire_it.rs new file mode 100644 index 00000000..0598cb44 --- /dev/null +++ b/crates/mobee-core/src/seller_node/buzz_wire_it.rs @@ -0,0 +1,327 @@ +//! LOCAL-RELAY teeth for the buzz persona's PRODUCTION WIRING — `SellerNodeRunner::boot`, the path +//! `mobee sell` actually runs. The persona's own behaviour (publish, clobber guard, heartbeat, +//! clean-shutdown clear) is fenced in [`super::buzz_relay_it`]; what is fenced HERE is that the +//! shipping daemon calls it at all, that a node without `[buzz]` stays silent on the buzz wire, and +//! that a buzz relay the node cannot reach never keeps it from booting. +//! +//! Both assertions ride the SAME in-process relay and the SAME observer, so the configured node is +//! the positive control for the unconfigured one: the silence asserted for a `[buzz]`-less node is +//! only evidence because the instrument demonstrably sees the configured node's traffic. + +use super::run::{ + BUZZ_START_TIMEOUT, BuzzStartOutcome, SellerNodeRunner, ShutdownSignals, start_buzz_bounded, +}; +use crate::home::{self, BuzzConfig}; + +use nostr_relay_builder::prelude::{ + LocalRelay, PolicyResult, QueryPolicy, RelayBuilder, WritePolicy, +}; +use nostr_sdk::prelude::{ + BoxedFuture, Client, Event, Filter, Keys, Kind, PublicKey, RelayPoolNotification, +}; +use std::net::SocketAddr; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use super::SellerNode; +use super::buzz::PRESENCE_KIND; + +static WIRE_SEQ: AtomicU64 = AtomicU64::new(0); + +fn unique_root(label: &str) -> std::path::PathBuf { + let n = WIRE_SEQ.fetch_add(1, Ordering::SeqCst); + std::env::temp_dir().join(format!( + "mobee-buzz-wire-{label}-{}-{n}", + std::process::id() + )) +} + +async fn start_relay() -> (LocalRelay, String) { + let relay = LocalRelay::new(RelayBuilder::default()); + relay.run().await.expect("relay run"); + let url = relay.url().await.to_string(); + (relay, url) +} + +async fn connect_client(relay_url: &str) -> Client { + let client = Client::new(Keys::generate()); + client.add_relay(relay_url).await.expect("add relay"); + client.connect().await; + client.wait_for_connection(Duration::from_secs(5)).await; + client +} + +/// A seller home pointed at `relay_url` for the marketplace leg. `buzz_relay` is the `[buzz]` +/// section's relay — `None` leaves the section absent, which is the inert contract under test. +fn seller_home( + root: &std::path::Path, + relay_url: &str, + buzz_relay: Option<&str>, +) -> home::MobeeHome { + let mut h = home::bootstrap(root).expect("bootstrap home"); + h.config.relay_url = relay_url.to_string(); + h.config.buzz = buzz_relay.map(|url| BuzzConfig { + relay_url: url.to_string(), + name: "Rocky".to_string(), + about: None, + rate_sats: Some(50), + capabilities: vec!["code".to_string()], + mint: None, + // Fast beat so a heartbeat lands inside the test window. + heartbeat_secs: 1, + }); + h +} + +/// Collect every kind-0 / presence event seen from each watched author, keyed by author. +fn spawn_buzz_traffic_collector(client: &Client) -> Arc>> { + let seen = Arc::new(Mutex::new(Vec::new())); + let sink = seen.clone(); + let mut notif = client.notifications(); + tokio::spawn(async move { + while let Ok(n) = notif.recv().await { + if let RelayPoolNotification::Event { event, .. } = n { + sink.lock() + .unwrap_or_else(|e| e.into_inner()) + .push((event.pubkey, event.kind.as_u16())); + } + } + }); + seen +} + +async fn wait_until bool>(timeout: Duration, mut cond: F) -> bool { + let deadline = tokio::time::Instant::now() + timeout; + loop { + if cond() { + return true; + } + if tokio::time::Instant::now() >= deadline { + return false; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } +} + +fn count_from(seen: &Arc>>, author: PublicKey, kind: u16) -> usize { + seen.lock() + .unwrap_or_else(|e| e.into_inner()) + .iter() + .filter(|(pubkey, seen_kind)| *pubkey == author && *seen_kind == kind) + .count() +} + +/// The wiring tooth (#199) AND the inert-compat tooth (invariant 1), as one differential: two nodes +/// boot against one relay, `[buzz]` present on one and absent on the other, and the observer watches +/// both keys. Configured ⇒ the persona reaches the relay from the production boot path; absent ⇒ not +/// one event on the buzz wire, measured by the instrument that just proved it can see them. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn boot_publishes_the_persona_only_when_buzz_is_configured() { + let (relay, relay_url) = start_relay().await; + + let wired_root = unique_root("wired"); + let inert_root = unique_root("inert"); + let wired_home = seller_home(&wired_root, &relay_url, Some(&relay_url)); + let inert_home = seller_home(&inert_root, &relay_url, None); + let wired_pk = PublicKey::parse(&home::public_key_hex(&wired_home).expect("wired pubkey")) + .expect("parse wired pubkey"); + let inert_pk = PublicKey::parse(&home::public_key_hex(&inert_home).expect("inert pubkey")) + .expect("parse inert pubkey"); + + // Watch BOTH keys from before either boots, so no early publish is missed. + let observer = connect_client(&relay_url).await; + observer + .subscribe( + Filter::new() + .authors([wired_pk, inert_pk]) + .kinds([Kind::Metadata, Kind::Custom(PRESENCE_KIND)]), + None, + ) + .await + .expect("subscribe buzz kinds"); + let seen = spawn_buzz_traffic_collector(&observer); + + // Both boots run concurrently — the NIP-42 wait is the same for each, so this costs one wait. + let (wired, inert) = tokio::join!( + SellerNodeRunner::boot(wired_home), + SellerNodeRunner::boot(inert_home), + ); + let _wired = wired.expect("boot with [buzz] configured"); + let _inert = inert.expect("boot with [buzz] absent"); + + // Positive control: the configured node's persona reached the relay THROUGH THE BOOT PATH. + let persona_landed = wait_until(Duration::from_secs(15), || { + count_from(&seen, wired_pk, Kind::Metadata.as_u16()) > 0 + && count_from(&seen, wired_pk, PRESENCE_KIND) > 0 + }) + .await; + let wired_kind0 = count_from(&seen, wired_pk, Kind::Metadata.as_u16()); + let wired_presence = count_from(&seen, wired_pk, PRESENCE_KIND); + eprintln!("wired node: kind0={wired_kind0} presence={wired_presence}"); + assert!( + persona_landed, + "booting with [buzz] configured must publish the persona kind-0 and a presence beat \ + (kind0={wired_kind0}, presence={wired_presence}) — the production boot path is the thing \ + under test" + ); + + // The inert claim, asserted only now that the instrument has demonstrably seen buzz traffic. + let inert_kind0 = count_from(&seen, inert_pk, Kind::Metadata.as_u16()); + let inert_presence = count_from(&seen, inert_pk, PRESENCE_KIND); + eprintln!("inert node: kind0={inert_kind0} presence={inert_presence}"); + assert_eq!( + (inert_kind0, inert_presence), + (0, 0), + "a node with no [buzz] section must put NOTHING on the buzz wire — got kind0={inert_kind0} \ + presence={inert_presence}" + ); + + observer.disconnect().await; + relay.shutdown(); + let _ = std::fs::remove_dir_all(&wired_root); + let _ = std::fs::remove_dir_all(&inert_root); +} + +/// Invariant 2 (the #149 brick class): a buzz relay the node cannot reach degrades to no persona and +/// the node boots anyway. A propagated buzz error — or an unbounded bring-up — would take the seller +/// down with the buzz relay, which is exactly what may never happen: the persona is discovery +/// context, the marketplace leg is the product. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_buzz_relay_that_is_down_never_keeps_the_node_from_booting() { + let (relay, relay_url) = start_relay().await; + + // A port that is bound and immediately released: nothing is listening, so every connect fails. + let closed_port = { + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind ephemeral"); + listener.local_addr().expect("addr").port() + }; + let dead_buzz = format!("ws://127.0.0.1:{closed_port}"); + + let root = unique_root("dead-buzz"); + let home = seller_home(&root, &relay_url, Some(&dead_buzz)); + + let started = std::time::Instant::now(); + let booted = SellerNodeRunner::boot(home).await; + let elapsed = started.elapsed(); + eprintln!( + "boot with a dead buzz relay: ok={} in {elapsed:?}", + booted.is_ok() + ); + + let runner = booted.expect("a dead buzz relay must NOT fail the boot"); + assert!( + !runner.seller_pubkey().is_empty(), + "the booted node must still be usable with no persona" + ); + // The bring-up is bounded, so the seller's readiness to claim cannot be held hostage by a sick + // buzz relay. CONNECT_WAIT for the marketplace leg is already spent by this point; the margin + // covers it. + assert!( + elapsed < BUZZ_START_TIMEOUT + Duration::from_secs(30), + "the buzz bring-up must be bounded — boot took {elapsed:?}" + ); + + relay.shutdown(); + let _ = std::fs::remove_dir_all(&root); +} + +/// A relay that completes the websocket handshake and then answers NOTHING — the REQ never gets its +/// EOSE, the EVENT never gets its OK. This is the dishonest failure: the socket looks alive. +#[derive(Debug)] +struct NeverAnswers; + +impl QueryPolicy for NeverAnswers { + fn admit_query<'a>( + &'a self, + _query: &'a Filter, + _addr: &'a SocketAddr, + ) -> BoxedFuture<'a, PolicyResult> { + Box::pin(std::future::pending()) + } +} + +impl WritePolicy for NeverAnswers { + fn admit_event<'a>( + &'a self, + _event: &'a Event, + _addr: &'a SocketAddr, + ) -> BoxedFuture<'a, PolicyResult> { + Box::pin(std::future::pending()) + } +} + +/// A relay that hangs mid-conversation degrades — AND the arm that saves the node is the bring-up's +/// INNER bounds, not the outer backstop. That second half is a tripwire on `BUZZ_START_TIMEOUT`'s +/// headroom, because the bound is only a backstop while this arithmetic holds: +/// +/// ```text +/// kind-0 fetch 8s buzz::KIND0_FETCH_TIMEOUT_SECS +/// OK-wait 10s nostr-relay-pool relay::constants::WAIT_FOR_OK_TIMEOUT +/// ──────────────────── +/// worst reproducible 18s < 25s BUZZ_START_TIMEOUT +/// ``` +/// +/// The assertion is on the RETURNED VARIANT, never on elapsed time — the call has returned, so there +/// is no window here to be wrong about and nothing that rots when a machine or a suite gets slower. +/// If a dependency bump ever pushes that sum past the bound, this flips to `TimedOut` and goes red: +/// the day the backstop starts bearing load is the day someone needs to know. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_relay_that_hangs_is_caught_by_the_inner_bounds_not_the_backstop() { + let relay = LocalRelay::new( + RelayBuilder::default() + .query_policy(NeverAnswers) + .write_policy(NeverAnswers), + ); + relay.run().await.expect("relay run"); + let hung_url = relay.url().await.to_string(); + + // Marketplace leg unused here — this drives the bring-up directly, so no boot/NIP-42 wait. + let root = unique_root("hung-buzz"); + let home = seller_home(&root, &hung_url, Some(&hung_url)); + let node = SellerNode::open(home).await.expect("open node"); + + let started = std::time::Instant::now(); + let outcome = start_buzz_bounded(&node).await; + let elapsed = started.elapsed(); + + let arm = match &outcome { + BuzzStartOutcome::Live(_) => "Live", + BuzzStartOutcome::Inert => "Inert", + BuzzStartOutcome::Failed(error) => { + eprintln!("inner-bound failure: {error}"); + "Failed" + } + BuzzStartOutcome::TimedOut => "TimedOut", + }; + eprintln!("hung relay: arm={arm} elapsed={elapsed:?} (bound {BUZZ_START_TIMEOUT:?})"); + + assert_eq!( + arm, "Failed", + "a hung relay must be refused by the bring-up's own bounds, leaving BUZZ_START_TIMEOUT a \ + backstop. arm={arm} means the arithmetic in this test's docs no longer holds — if it is \ + TimedOut, the inner bounds now sum past {BUZZ_START_TIMEOUT:?} and the backstop is bearing \ + load" + ); + + relay.shutdown(); + let _ = std::fs::remove_dir_all(&root); +} + +/// The other half of invariant 1: a signal receiver REPLACES the process default for that signal, so +/// the run loop installs one only for a daemon carrying a live persona. A node with no persona +/// installs nothing and keeps dying on SIGTERM exactly as it does today. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn shutdown_signals_are_installed_only_for_a_live_persona() { + assert!( + ShutdownSignals::install(false).is_none(), + "a buzz-inert daemon must install NO signal handler — installing one silently changes how \ + every seller daemon dies" + ); + #[cfg(unix)] + assert!( + ShutdownSignals::install(true).is_some(), + "a daemon with a live persona must install the handlers, or the clean-shutdown clear can \ + never fire" + ); +} diff --git a/crates/mobee-core/src/seller_node/mod.rs b/crates/mobee-core/src/seller_node/mod.rs index 3f220cf2..ade04cd7 100644 --- a/crates/mobee-core/src/seller_node/mod.rs +++ b/crates/mobee-core/src/seller_node/mod.rs @@ -234,6 +234,9 @@ impl SellerNode { #[cfg(test)] mod buzz_relay_it; +#[cfg(test)] +mod buzz_wire_it; + #[cfg(test)] mod tests { use super::*; diff --git a/crates/mobee-core/src/seller_node/run.rs b/crates/mobee-core/src/seller_node/run.rs index f1597f47..48e165ad 100644 --- a/crates/mobee-core/src/seller_node/run.rs +++ b/crates/mobee-core/src/seller_node/run.rs @@ -39,7 +39,7 @@ use crate::seller_git::{self, DeliveryAgentIdentity}; use super::outbox::drain_once; use super::publisher::RelayPublisher; -use super::{now_unix, NodeError, SellerNode}; +use super::{buzz, now_unix, NodeError, SellerNode}; /// How long (seconds) the outbox publisher keeps retrying a claim event before it expires. Matches /// the legacy claim TTL: a claim outlives a slow relay but never lingers indefinitely. @@ -1109,6 +1109,11 @@ fn probe_sentinel_present(workdir: &std::path::Path, sentinel: &str) -> bool { const CONNECT_WAIT: Duration = Duration::from_secs(20); /// Cadence of the outbox drain / housekeeping tick. const DRAIN_INTERVAL: Duration = Duration::from_secs(5); +/// Upper bound on the buzz persona bring-up at boot. The legs inside [`buzz::start`] are individually +/// bounded (connect, kind-0 fetch) but the publish is not, and the persona is discovery context that +/// the money path never reads — so one outer bound keeps a sick buzz relay from delaying the moment +/// this seller is ready to claim. +pub(super) const BUZZ_START_TIMEOUT: Duration = Duration::from_secs(25); /// A booted seller node with its live relay surface. pub struct SellerNodeRunner { @@ -1131,6 +1136,148 @@ pub struct SellerNodeRunner { /// Homogeneous execution-slot admission (reserve-at-claim). Behind an `Arc` so it is shared with /// the off-loop execution tasks; see [`SlotGate`]. slots: Arc, + /// The live buzz persona, held for the node's lifetime so presence stays up (see + /// [`start_buzz_or_degrade`]). `None` when `[buzz]` is absent — or when the bring-up degraded. + /// + /// Behind a `Mutex` because the clean-exit path must TAKE the handle: + /// [`buzz::BuzzHandle::shutdown`] consumes it to join the presence task, while the loop holds + /// `Arc` (execution runs off the loop), so there is no owned `self` to move out of and a + /// borrow will not do. The lock is never held across an await. + buzz: Mutex>, +} + +/// What the bounded bring-up did. The arms are named rather than collapsed into an `Option` because +/// "the relay refused us inside its own bounds" and "we outran our own backstop" are the same value +/// to the caller and completely different facts about [`BUZZ_START_TIMEOUT`] — the second one says +/// the backstop is bearing load, which is the thing worth noticing. +/// +/// `Live` is far larger than the other variants, and boxing it would buy nothing: exactly one of +/// these exists per process boot and it is destructured immediately, so the size difference costs a +/// few hundred stack bytes once — where an allocation would cost one every time. +#[allow(clippy::large_enum_variant)] +pub(super) enum BuzzStartOutcome { + /// A live persona; the handle is held for the node's lifetime. + Live(buzz::BuzzHandle), + /// `[buzz]` absent — inert by contract: no connection, no publish. + Inert, + /// The bring-up failed within its own bounds (relay refused, clobber guard, signer). + Failed(buzz::BuzzError), + /// The bring-up outran [`BUZZ_START_TIMEOUT`]. + TimedOut, +} + +/// Run the persona bring-up under [`BUZZ_START_TIMEOUT`] and report which arm ran. Silent — the boot +/// path logs (see [`start_buzz_or_degrade`]), so a test can assert the arm without parsing output. +pub(super) async fn start_buzz_bounded(node: &SellerNode) -> BuzzStartOutcome { + match tokio::time::timeout(BUZZ_START_TIMEOUT, node.start_buzz()).await { + Ok(Ok(Some(handle))) => BuzzStartOutcome::Live(handle), + Ok(Ok(None)) => BuzzStartOutcome::Inert, + Ok(Err(error)) => BuzzStartOutcome::Failed(error), + Err(_) => BuzzStartOutcome::TimedOut, + } +} + +/// Bring up the node's buzz persona at boot when `[buzz]` is configured, bounded by +/// [`BUZZ_START_TIMEOUT`]. +/// +/// The persona is discovery/identity only — nothing here feeds the pay gate — so NO buzz outcome may +/// stop this node from selling: an absent section is inert and silent, and a bring-up that fails or +/// outruns the bound degrades to no persona with a loud line. Only a live persona yields a handle. +async fn start_buzz_or_degrade(node: &SellerNode) -> Option { + match start_buzz_bounded(node).await { + BuzzStartOutcome::Live(handle) => { + eprintln!( + "seller node buzz persona live: pubkey={} kind0={}", + handle.pubkey_hex(), + handle.kind0_event_id + ); + Some(handle) + } + // Inert by contract: nothing opened, nothing published, and no line to log. + BuzzStartOutcome::Inert => None, + BuzzStartOutcome::Failed(error) => { + eprintln!( + "seller node BUZZ DEGRADE: persona bring-up failed; selling continues with no \ + persona: {error}" + ); + None + } + BuzzStartOutcome::TimedOut => { + eprintln!( + "seller node BUZZ DEGRADE: persona bring-up exceeded {}s; selling continues with no \ + persona", + BUZZ_START_TIMEOUT.as_secs() + ); + None + } + } +} + +/// The operator's stop request for a daemon carrying a live buzz persona: SIGTERM (what a supervisor +/// sends) or SIGINT (Ctrl-C in a terminal). +/// +/// INSTALLED ONLY WHEN A PERSONA IS LIVE. Registering a signal receiver REPLACES the process default +/// for that signal, so installing one unconditionally would change how every seller daemon dies. A +/// buzz-inert node installs nothing ([`ShutdownSignals::install`] answers `None`) and keeps exactly +/// today's behaviour: the signal terminates the process and there is no presence to clear. +pub(super) struct ShutdownSignals { + #[cfg(unix)] + terminate: tokio::signal::unix::Signal, + #[cfg(unix)] + interrupt: tokio::signal::unix::Signal, +} + +impl ShutdownSignals { + /// Install the receivers when a persona is live. `None` ⇒ nothing is registered — for a + /// buzz-inert node, for a platform without unix signals, and for the (logged) case where the + /// runtime refuses the registration, which degrades to TTL expiry rather than failing the boot. + pub(super) fn install(persona_live: bool) -> Option { + if !persona_live { + return None; + } + #[cfg(unix)] + { + use tokio::signal::unix::{SignalKind, signal}; + match ( + signal(SignalKind::terminate()), + signal(SignalKind::interrupt()), + ) { + (Ok(terminate), Ok(interrupt)) => Some(Self { + terminate, + interrupt, + }), + (Err(error), _) | (_, Err(error)) => { + eprintln!( + "seller node WARN: shutdown signal handlers unavailable ({error}); buzz \ + presence will clear on the relay's TTL instead of at exit" + ); + None + } + } + } + #[cfg(not(unix))] + None + } + + /// Resolve on the first stop request. + async fn requested(&mut self) { + #[cfg(unix)] + tokio::select! { + _ = self.terminate.recv() => {} + _ = self.interrupt.recv() => {} + } + #[cfg(not(unix))] + std::future::pending::<()>().await + } +} + +/// The run loop's stop-request future: the installed signals when a persona is live, and a future +/// that never resolves otherwise (belt to the loop branch's own guard). +async fn shutdown_requested(signals: Option<&mut ShutdownSignals>) { + match signals { + Some(signals) => signals.requested().await, + None => std::future::pending().await, + } } impl SellerNodeRunner { @@ -1234,6 +1381,10 @@ impl SellerNodeRunner { lapse_secs.unwrap_or(DEFAULT_CLAIM_AWARD_TIMEOUT_SECS) ); + // The persona comes up LAST, once the marketplace surface is authenticated and ready: buzz is + // discovery context, so it may never sit in front of the money path's connect. + let buzz = start_buzz_or_degrade(&node).await; + Ok(Self { node, client, @@ -1243,6 +1394,7 @@ impl SellerNodeRunner { boot_auth, agents, slots, + buzz: Mutex::new(buzz), }) } @@ -1251,6 +1403,15 @@ impl SellerNodeRunner { self.seller_pubkey.to_hex() } + /// Whether a live persona is held — the single reader of that fact, so the signal-handler + /// decision and the clean-exit clear cannot disagree about it. + /// + /// A poisoned lock reads as NO persona: that installs no signal handlers and leaves presence to + /// expire on the relay's TTL, which is the documented degrade path rather than a panic at boot. + fn persona_live(&self) -> bool { + self.buzz.lock().map(|slot| slot.is_some()).unwrap_or(false) + } + /// Subscribe (or re-subscribe) the offer REQ. `open_pool` false forces the targeted-only shape — /// used by the boot/recovery path when the seller has not opted into the open pool, and by the /// `CLOSED` degrade, which keeps targeted claiming alive after the relay refuses the grouped REQ. @@ -1490,8 +1651,18 @@ impl SellerNodeRunner { // never once succeeded went unnoticed (#171). The next answered probe names it. let mut stalled_since_recovery = false; let mut manual_recovery_succeeded = false; + // Registered only for a daemon carrying a live persona — see [`ShutdownSignals`]. + let mut shutdown = ShutdownSignals::install(self.persona_live()); loop { tokio::select! { + // A stop request exists only when a persona is live: end the loop so presence is + // cleared on the way out instead of lingering until the relay's TTL expires it. With + // no persona the branch is disabled and no handler was ever installed, so the signal + // terminates the process exactly as it does today. + _ = shutdown_requested(shutdown.as_mut()), if shutdown.is_some() => { + eprintln!("seller node: stop requested; clearing the buzz persona and ending the loop"); + break; + } _ = drain_tick.tick() => { self.sweep_lapsed_claims(); self.start_due_harness_probes(); @@ -1827,6 +1998,15 @@ impl SellerNodeRunner { } } } + // Clean exit: clear presence NOW rather than leaving the persona advertised as online until + // the relay's TTL expires it. A crash still falls back to that TTL — this is the clean path. + // Taken in one statement so the guard is released before the await below — the lock never + // spans a suspension point, and a second exit path finds `None` rather than a live handle. + let persona = self.buzz.lock().ok().and_then(|mut slot| slot.take()); + if let Some(buzz) = persona { + buzz.shutdown().await; + eprintln!("seller node buzz persona cleared (clean shutdown)"); + } Ok(()) }