From f9d5aa635f6a7ff23eb6f668892e25b576b9b4b4 Mon Sep 17 00:00:00 2001 From: orveth Date: Sun, 26 Jul 2026 23:05:10 -0700 Subject: [PATCH 1/7] buyer: make it impossible for the signer to park the daemon silently MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both legs of a buyer signer round-trip were timer-less: `send` parks forever on a full queue the actor is not draining, and `rx.await` parks forever if the actor is alive but never answers. Neither arms a timer, so the runtime has nothing to wake for — a caller reached from a long-lived task's `select!` branch body is simply never polled again, at 0% CPU, with nothing logged anywhere. That is the diagnostic filter, too: a permanent park proves no timer was pending, which excludes every bounded await and leaves exactly these channel awaits. Both legs now route through one bounded `round_trip`, and the error names the call and the leg so an operator sees which round-trip stalled instead of a bare "actor gone". A bound cannot make a stuck actor answer; it converts an invisible permanent park into a named, recoverable failure at the exact site. The tooth runs on paused time, so the production bound elapses instantly in wall-clock, and the outer timeout makes a revert fail cleanly rather than hang the suite. It calls twice: the second proves the caller is left usable, which is the property a long-lived daemon task actually needs. Co-Authored-By: Claude Opus 5 (1M context) --- crates/mobee-core/src/buyer/signer.rs | 123 ++++++++++++++++++++++++-- 1 file changed, 116 insertions(+), 7 deletions(-) diff --git a/crates/mobee-core/src/buyer/signer.rs b/crates/mobee-core/src/buyer/signer.rs index 2538e70c..77829e91 100644 --- a/crates/mobee-core/src/buyer/signer.rs +++ b/crates/mobee-core/src/buyer/signer.rs @@ -26,13 +26,29 @@ pub struct SignerHandle { public_key_hex: String, } -/// The signer task exited. +/// How long a single signer round-trip may take before it is abandoned. +/// +/// Deliberately generous: everything the actor does is local cryptography measured in milliseconds, +/// so this is not a latency budget — it is a liveness bound. Its job is to guarantee the call +/// *returns*, not to police how fast. +const SIGNER_CALL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); + +/// A signer round-trip that did not complete: the actor exited, or it failed to answer within +/// [`SIGNER_CALL_TIMEOUT`]. Carries which call and which leg, so the operator log names the exact +/// site instead of a bare "actor gone". #[derive(Debug)] -pub struct SignerActorGone; +pub struct SignerActorGone { + call: &'static str, + cause: &'static str, +} impl std::fmt::Display for SignerActorGone { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(formatter, "signer actor is not running") + write!( + formatter, + "signer round-trip `{}` did not complete: {}", + self.call, self.cause + ) } } @@ -44,15 +60,57 @@ impl SignerHandle { &self.public_key_hex } + /// Send one command and await its reply, with BOTH legs bounded. + /// + /// Both legs must be bounded because both are timer-less, and a timer-less await is the one + /// thing that can park this daemon permanently and silently (#173). `send` parks forever if the + /// queue is full and the actor is not draining it; `rx.await` parks forever if the actor is + /// alive but never answers. Neither arms a timer, so the runtime has nothing to wake for — a + /// caller reached from a `select!` branch body takes the whole loop down with it, at 0% CPU, + /// with no error logged anywhere. + /// + /// A bound cannot make a stuck actor answer. What it does is convert an invisible permanent + /// park into a named, logged, recoverable failure at this exact call site. + async fn round_trip( + &self, + call: &'static str, + command: Command, + rx: oneshot::Receiver, + ) -> Result { + match tokio::time::timeout(SIGNER_CALL_TIMEOUT, self.tx.send(command)).await { + Ok(Ok(())) => {} + Ok(Err(_)) => { + return Err(SignerActorGone { + call, + cause: "actor exited", + }) + } + Err(_) => { + return Err(SignerActorGone { + call, + cause: "queue stayed full (actor not draining)", + }) + } + } + match tokio::time::timeout(SIGNER_CALL_TIMEOUT, rx).await { + Ok(Ok(value)) => Ok(value), + Ok(Err(_)) => Err(SignerActorGone { + call, + cause: "actor dropped the reply", + }), + Err(_) => Err(SignerActorGone { + call, + cause: "actor never answered", + }), + } + } + /// The buyer public key (hex), routed through the actor queue. Proves the /// serialized signer path end to end (later phases sign over this same slot). pub async fn public_key_via_actor(&self) -> Result { let (reply, rx) = oneshot::channel(); - self.tx - .send(Command::PublicKey { reply }) + self.round_trip("public_key", Command::PublicKey { reply }, rx) .await - .map_err(|_| SignerActorGone)?; - rx.await.map_err(|_| SignerActorGone) } } @@ -95,6 +153,57 @@ mod tests { std::env::temp_dir().join(format!("mobee-buyer-signer-{label}-{}-{id}", std::process::id())) } + // TOOTH (#173) — a signer round-trip is BOUNDED, so an actor that never answers cannot park the + // caller forever. The daemon reaches this handle from tasks that also own the trade loop, so a + // timer-less await here is not a slow call: it is a task that is never polled again, sitting at + // 0% CPU with nothing logged. The diagnostic that identifies the class: a permanent park proves + // no timer was pending, which excludes every BOUNDED await and leaves exactly these timer-less + // channel awaits. + // + // The stalled actor here holds each command — and therefore each reply sender — forever, which + // is the one shape that hangs: dropping the sender would surface as a recv error, not a park. + // + // Time is paused, so the production bound elapses instantly in wall-clock. The OUTER timeout is + // what makes a revert fail cleanly instead of hanging the suite: remove the bound in + // `round_trip` and there is no timer at 30s, auto-advance jumps to the outer 600s, and the + // assert goes red. + // + // Two calls, not one: the second proves the caller was left usable rather than merely returning + // once — which is the property a long-lived daemon task actually needs to keep serving. + #[tokio::test(start_paused = true)] + async fn a_stalled_signer_round_trip_is_bounded_and_leaves_the_caller_usable() { + let (tx, mut rx) = mpsc::channel::(8); + // The stalled actor: receive, then hold. Never answers, never drops a reply sender. + tokio::spawn(async move { + let mut held = Vec::new(); + while let Some(command) = rx.recv().await { + held.push(command); + } + }); + let handle = SignerHandle { + tx, + public_key_hex: "00".repeat(32), + }; + + let outer = std::time::Duration::from_secs(600); + for attempt in 1..=2 { + let call = tokio::time::timeout(outer, handle.public_key_via_actor()); + let outcome = call.await.unwrap_or_else(|_| { + panic!( + "attempt {attempt}: the signer round-trip never returned — an unbounded \ + timer-less await here parks the calling task permanently and silently" + ) + }); + let error = outcome.expect_err("a stalled actor cannot produce a public key"); + assert!( + error.to_string().contains("public_key") + && error.to_string().contains("never answered"), + "attempt {attempt}: the failure must NAME the call and the leg so an operator can \ + see which round-trip stalled, got {error}" + ); + } + } + #[tokio::test(flavor = "current_thread")] async fn actor_serves_pubkey_and_never_the_secret() { let root = temp_home("pubkey"); From 48e1141fded8da0fa4b966bd48790f71d73f4258 Mon Sep 17 00:00:00 2001 From: orveth Date: Sun, 26 Jul 2026 23:40:37 -0700 Subject: [PATCH 2/7] buyer: give the buyer one long-lived authenticated relay session MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The buyer had no persistent relay presence. Every publish, every has_award, every job-view fetch built a fresh Client — fresh MemoryDatabase, fresh WebSocket, fresh NIP-42 handshake — did one thing, and disconnected. That also makes get_job(wait_for) a reconnect-per-iteration poll where a subscription is the correct primitive, which is the dominant term in trade latency. The session is an actor, not a shared Client, because the money write does not run on the daemon's runtime: CdkPaymentEffects builds a current_thread runtime on its own OS thread and awaits the payment gift-wrap there. Channels cross runtimes; Clients do not. So the client lives in one task and every caller reaches it through a handle whose both legs are bounded — an unbounded write actor would reintroduce the signer-park class at a brand-new site. The write path does NOT retry an auth-required rejection. Relay::send_event already waits for authentication and resends. Layering our own retry on top would make ours the third publish of the same event and blur which failure the caller sees. That resend is, however, silently conditional on auto-authentication being enabled, so it is set explicitly and a tooth pins the contract: the same write lands with it on and stays refused with it off, so an SDK bump or an option-drift cleanup goes red instead of quietly removing the retry. An empty accepted-set is a failure, never a quiet success: send_event_to returns Ok even when the relay refused, with the reason in output.failed. Reading only the outer Result would report a refused write as published, and on the money path "we think we sent it" is the ambiguity this daemon exists to remove. spawn returns as soon as the relay is registered, and the actor performs connect and NIP-42 as its first action. The daemon binds its socket after bootstrap and connect-or-spawn allows a cold daemon 10s to appear, so a handshake bounded at 20s on the startup path would let an unreachable — or merely lazily-challenging — relay push the socket past that deadline, and every MCP call would report a daemon that failed to start while it was coming up fine. Co-Authored-By: Claude Opus 5 (1M context) --- crates/mobee-core/src/buyer/mod.rs | 22 + crates/mobee-core/src/buyer/relay.rs | 635 +++++++++++++++++++++++++++ 2 files changed, 657 insertions(+) create mode 100644 crates/mobee-core/src/buyer/relay.rs diff --git a/crates/mobee-core/src/buyer/mod.rs b/crates/mobee-core/src/buyer/mod.rs index d5788ef9..2b0e99a8 100644 --- a/crates/mobee-core/src/buyer/mod.rs +++ b/crates/mobee-core/src/buyer/mod.rs @@ -24,6 +24,7 @@ pub mod client; pub mod lifecycle; pub mod lock; pub mod protocol; +pub mod relay; pub mod reservations; pub mod signer; pub mod store; @@ -78,6 +79,9 @@ pub enum BuyerError { Store(StoreError), Wallet(FundError), Identity(HomeError), + /// The relay could not be REGISTERED — a malformed url or a pool refusal. An unreachable relay + /// is not this: the daemon comes up and serves with the network down. + Relay(String), Io(String), } @@ -88,6 +92,7 @@ impl std::fmt::Display for BuyerError { Self::Store(error) => write!(formatter, "{error}"), Self::Wallet(error) => write!(formatter, "buyer wallet error: {error}"), Self::Identity(error) => write!(formatter, "buyer identity error: {error}"), + Self::Relay(message) => write!(formatter, "buyer relay error: {message}"), Self::Io(message) => write!(formatter, "buyer io error: {message}"), } } @@ -122,6 +127,9 @@ struct BuyerContext { store: BuyerStore, wallet: WalletHandle, signer: SignerHandle, + /// The buyer's one long-lived relay session. Writes and (from the delivery watcher on) + /// subscriptions ride this instead of a fresh `Client` per operation. + relay: relay::RelayHandle, started_at_unix: i64, /// Serializes the money-state-mutating RPCs (`award` reserves, `collect` flips) so a /// reservation's balance/spent snapshot is never read while a concurrent collect is melting. @@ -159,12 +167,21 @@ async fn bootstrap(home: MobeeHome) -> Result<(HomeLock, Arc, Path let signer = signer::spawn(&home)?; + // Registers the relay and hands the session to the actor; it does NOT wait for the socket or + // the NIP-42 handshake, so an unreachable relay cannot delay the daemon past connect-or-spawn's + // readiness deadline (see `relay::spawn`). + let relay_keys = buyer_keys(&home).map_err(BuyerError::Relay)?; + let relay = relay::spawn(relay_keys, &home.config.relay_url) + .await + .map_err(|error| BuyerError::Relay(error.to_string()))?; + let socket_path = home.root.join(SOCKET_FILE); let context = Arc::new(BuyerContext { home, store, wallet, signer, + relay, started_at_unix, money_lock: Mutex::new(()), last_reconcile: Mutex::new(None), @@ -961,6 +978,11 @@ async fn status(context: &BuyerContext, id: Value) -> Response { }, "reconcile": reconcile, "parked_awards": parked_awards, + // The relay the buyer's one long-lived session is bound to. Deliberately NOT a liveness + // probe: `status` is what connect-or-spawn polls to decide the daemon is up, and a probe + // bounded at 10s would push that poll past its own readiness deadline. Liveness belongs + // to the watcher's tick, where a slow answer costs nothing. + "relay": { "url": context.relay.relay_url() }, }), ) } diff --git a/crates/mobee-core/src/buyer/relay.rs b/crates/mobee-core/src/buyer/relay.rs new file mode 100644 index 00000000..3c0c1c3b --- /dev/null +++ b/crates/mobee-core/src/buyer/relay.rs @@ -0,0 +1,635 @@ +//! The buyer's one long-lived relay connection. +//! +//! Every buyer write rides this single authenticated socket, and the delivery watcher's +//! subscriptions will ride it too. Before it, the buyer had no persistent relay presence at all: +//! each publish, each `has_award`, each job-view fetch built a fresh `Client` (fresh +//! `MemoryDatabase`, fresh WebSocket, fresh NIP-42 handshake), did one thing, and disconnected — +//! which also made `get_job(wait_for=…)` a reconnect-per-iteration poll where a subscription is the +//! correct primitive (#175). +//! +//! ## Why an actor, and not a shared `Client` +//! +//! The money write does not run on the daemon's runtime. `CdkPaymentEffects` builds a +//! `new_current_thread` runtime on its own OS thread and awaits the payment gift-wrap there, so a +//! `Client` owned by the daemon's multi-thread runtime cannot simply be handed to it. Channels +//! cross runtimes; `Client`s do not. So the client lives in one task on the daemon runtime and +//! every caller reaches it through [`RelayHandle`] — plain `mpsc` + `oneshot`, runtime-agnostic. +//! +//! Both legs of that round-trip are bounded, for the same reason the signer's are: a timer-less +//! await is the one thing that can park a caller permanently and silently, and this handle is +//! reached from tasks that also own the trade loop. An unbounded write actor would reintroduce the +//! signer-park class at a brand-new site. +//! +//! ## What this layer does NOT do +//! +//! It does not retry an `auth-required:` rejection. `Relay::send_event` already waits for +//! authentication and resends (`nostr-relay-pool` `relay/mod.rs:434-470`), bounded by +//! `WAIT_FOR_AUTHENTICATION_TIMEOUT` + `WAIT_FOR_OK_TIMEOUT`. Layering our own retry on top would +//! make ours the third publish of the same event and blur which failure the caller sees. +//! +//! That SDK resend is, however, silently conditional on `is_auto_authentication_enabled() && +//! has_signer()` — nothing errors or logs if either goes false. So [`spawn`] sets +//! `automatic_authentication(true)` explicitly as a drift guard, and a tooth pins the contract +//! itself: a write refused with `auth-required:` must still land once auth completes. + +use std::time::Duration; + +use nostr_sdk::prelude::{ + Client, Event, Filter, Keys, Kind, RelayOptions, RelayPoolNotification, RelayUrl, + SubscriptionId, +}; +use tokio::sync::{mpsc, oneshot}; + +use crate::relay_auth::{self, AuthWait}; + +/// How long to wait for the socket and for the NIP-42 handshake on it. +const CONNECT_WAIT: Duration = Duration::from_secs(20); + +/// Outer bound on one publish. +/// +/// Sized to sit ABOVE the SDK's own worst case rather than to cut it short: a first `OK` wait +/// (`WAIT_FOR_OK_TIMEOUT`, 10s) + an `auth-required:` re-auth (`WAIT_FOR_AUTHENTICATION_TIMEOUT`, +/// 7s) + the resend's `OK` wait (10s) = 27s of legitimate work. A tighter bound here would abandon +/// writes the SDK was about to land, and on the money path that manufactures exactly the ambiguity +/// (sent? not sent?) this daemon exists to avoid. It is a liveness backstop, not a latency budget. +const PUBLISH_TIMEOUT: Duration = Duration::from_secs(45); + +/// How long the liveness probe waits for its `EOSE`. A `limit(0)` REQ is answered in milliseconds +/// by a healthy relay. +const LIVENESS_PROBE_TIMEOUT: Duration = Duration::from_secs(10); + +/// Bound on one round-trip to the relay actor. +/// +/// Strictly greater than [`PUBLISH_TIMEOUT`]: if the handle gave up first, the actor would still be +/// completing the write while the caller had already been told it failed — the reply would land in +/// a dropped channel and a possibly-accepted write would be reported as a failure. The handle bound +/// exists to catch an actor that is GONE, not to race the work it was asked to do. +const RELAY_CALL_TIMEOUT: Duration = Duration::from_secs(60); + +/// Stable id for the liveness REQ, so a relay `CLOSED` names it. +const LIVENESS_PROBE_SUB_ID: &str = "mobee-buyer-liveness"; + +enum Command { + Publish { + event: Box, + reply: oneshot::Sender>, + }, + /// Ask the relay to serve one trivial REQ on the CURRENT session. + Probe { + reply: oneshot::Sender, + }, + /// Drop the socket and bring a fresh authenticated one up. + Reconnect { + reply: oneshot::Sender>, + }, +} + +/// A write the relay accepted. +#[derive(Debug, Clone)] +pub struct PublishReceipt { + pub event_id: String, + pub relay: String, +} + +/// A write that did not land. Never conflated with success: an empty accepted-set is a failure +/// here, so a caller can only read "published" from a relay that actually said so. +#[derive(Debug, Clone)] +pub enum PublishError { + /// The relay answered and refused. Carries the relay's own reason verbatim. + Refused(String), + /// The write did not complete within [`PUBLISH_TIMEOUT`]. + TimedOut, + /// The write could not be attempted (relay not added, bad url, database error). + Transport(String), +} + +impl std::fmt::Display for PublishError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Refused(reason) => write!(formatter, "relay refused the write: {reason}"), + Self::TimedOut => write!( + formatter, + "relay write did not complete within {PUBLISH_TIMEOUT:?}" + ), + Self::Transport(error) => write!(formatter, "relay write could not be sent: {error}"), + } + } +} + +impl std::error::Error for PublishError {} + +/// The relay actor exited, or failed to answer within [`RELAY_CALL_TIMEOUT`]. Names the call and +/// the leg so an operator sees which round-trip stalled. +#[derive(Debug)] +pub struct RelayActorGone { + call: &'static str, + cause: &'static str, +} + +impl std::fmt::Display for RelayActorGone { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + formatter, + "relay round-trip `{}` did not complete: {}", + self.call, self.cause + ) + } +} + +impl std::error::Error for RelayActorGone {} + +/// The relay connection could not be brought up at all. +#[derive(Debug)] +pub struct RelayBootError(pub String); + +impl std::fmt::Display for RelayBootError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(formatter, "{}", self.0) + } +} + +impl std::error::Error for RelayBootError {} + +/// A cheap, cloneable handle to the relay actor. Safe to hold from any runtime. +#[derive(Clone)] +pub struct RelayHandle { + tx: mpsc::Sender, + relay_url: String, +} + +impl RelayHandle { + /// The relay this buyer is bound to. + pub fn relay_url(&self) -> &str { + &self.relay_url + } + + /// Send one command and await its reply, with BOTH legs bounded — see the module doc. + async fn round_trip( + &self, + call: &'static str, + command: Command, + rx: oneshot::Receiver, + ) -> Result { + match tokio::time::timeout(RELAY_CALL_TIMEOUT, self.tx.send(command)).await { + Ok(Ok(())) => {} + Ok(Err(_)) => { + return Err(RelayActorGone { + call, + cause: "actor exited", + }) + } + Err(_) => { + return Err(RelayActorGone { + call, + cause: "queue stayed full (actor not draining)", + }) + } + } + match tokio::time::timeout(RELAY_CALL_TIMEOUT, rx).await { + Ok(Ok(value)) => Ok(value), + Ok(Err(_)) => Err(RelayActorGone { + call, + cause: "actor dropped the reply", + }), + Err(_) => Err(RelayActorGone { + call, + cause: "actor never answered", + }), + } + } + + /// Publish one signed event on the buyer's authenticated session. + pub async fn publish(&self, event: Event) -> Result, RelayActorGone> { + let (reply, rx) = oneshot::channel(); + self.round_trip( + "publish", + Command::Publish { + event: Box::new(event), + reply, + }, + rx, + ) + .await + } + + /// True when the relay is serving OUR subscriptions on THIS authenticated session. + pub async fn probe(&self) -> Result { + let (reply, rx) = oneshot::channel(); + self.round_trip("probe", Command::Probe { reply }, rx).await + } + + /// Rebuild the session: drop the socket and re-authenticate a fresh one. + pub async fn reconnect(&self) -> Result, RelayActorGone> { + let (reply, rx) = oneshot::channel(); + self.round_trip("reconnect", Command::Reconnect { reply }, rx) + .await + } +} + +/// Register the buyer's relay and spawn the actor that owns the session. +/// +/// Returns as soon as the relay is REGISTERED — it does not wait for the socket or the handshake. +/// That is load-bearing, not an optimisation: the daemon binds its Unix socket after bootstrap, and +/// `mobee`'s connect-or-spawn gives a cold daemon only `SPAWN_READY_TIMEOUT` (10s) to appear +/// (`crates/mobee/src/daemon.rs:21`). Waiting here for a handshake bounded at 20s would let an +/// unreachable — or merely lazily-challenging — relay push the socket past that deadline, and every +/// MCP call would report a daemon that failed to start while it was in fact coming up fine. +/// +/// So the session is brought up by the actor as its FIRST action, before it serves any command: the +/// daemon is responsive immediately, and a caller that publishes during boot simply queues behind +/// the handshake rather than racing it. +/// +/// A relay that cannot be reached is NOT fatal — the daemon must serve `status` with the network +/// down. Only a malformed url or a relay the pool refuses to register fails here; those are +/// configuration errors, not weather. +pub async fn spawn(keys: Keys, relay_url: &str) -> Result { + let client = Client::new(keys.clone()); + // Explicit, not inherited. The SDK's `auth-required:` resend is conditional on this being + // true, and it fails SILENTLY if it is not — see the module doc. + client.automatic_authentication(true); + client + .pool() + .add_relay(relay_url, RelayOptions::default().reconnect(true)) + .await + .map_err(|error| RelayBootError(format!("buyer relay add_relay: {error}")))?; + let parsed = RelayUrl::parse(relay_url) + .map_err(|error| RelayBootError(format!("buyer relay url: {error}")))?; + let relay = client + .relays() + .await + .get(&parsed) + .cloned() + .ok_or_else(|| RelayBootError("buyer relay missing after add_relay".into()))?; + + let public_key = keys.public_key(); + let (tx, mut rx) = mpsc::channel::(64); + let owned_url = relay_url.to_owned(); + tokio::spawn(async move { + connect_and_authenticate(&client, &relay, "boot").await; + while let Some(command) = rx.recv().await { + match command { + Command::Publish { event, reply } => { + let _ = reply.send(publish_bounded(&client, &owned_url, &event).await); + } + Command::Probe { reply } => { + let _ = reply.send( + probe_relay_serves_our_reqs(&client, public_key, LIVENESS_PROBE_TIMEOUT) + .await, + ); + } + Command::Reconnect { reply } => { + let outcome = reconnect_and_authenticate(&client, &relay).await; + if let Ok(wait) = &outcome { + report_auth_wait("reconnect", Ok(*wait)); + } + let _ = reply.send(outcome.map_err(|error| error.to_string())); + } + } + } + }); + + Ok(RelayHandle { + tx, + relay_url: relay_url.to_owned(), + }) +} + +/// Bring the session up on a socket that is not yet connected, and report the handshake. +/// +/// The notification receiver is taken BEFORE `connect` because `Authenticated` is emitted once and +/// is not re-emitted — subscribing after the connect can miss it entirely on a fast relay. +async fn connect_and_authenticate( + client: &Client, + relay: &nostr_sdk::prelude::Relay, + phase: &str, +) { + let mut relay_notifications = relay.notifications(); + client.connect().await; + client.wait_for_connection(CONNECT_WAIT).await; + report_auth_wait( + phase, + relay_auth::wait_for_nip42_auth(&mut relay_notifications, CONNECT_WAIT).await, + ); +} + +/// Log the NIP-42 outcome. UNCONDITIONAL and named: a degraded session is an operator signal, and +/// a path that is silent in its uninteresting case is invisible in its interesting one. +fn report_auth_wait(phase: &str, outcome: Result) { + match outcome { + Ok(AuthWait::Authenticated) => { + eprintln!("buyer relay [{phase}]: NIP-42 authenticated"); + } + Ok(AuthWait::NoChallenge) => { + eprintln!( + "buyer relay [{phase}] DEGRADED: no NIP-42 challenge within {CONNECT_WAIT:?}; \ + proceeding unauthenticated. An auth-gated write will be refused once with \ + `auth-required:` and resent by the SDK after the challenge lands." + ); + } + Err(error) => { + eprintln!( + "buyer relay [{phase}] DEGRADED: NIP-42 did not complete ({error}); writes may be \ + refused until a reconnect re-authenticates" + ); + } + } +} + +/// One publish, bounded, with the relay's own answer preserved. +/// +/// An empty accepted-set is a FAILURE, never a quiet success — the caller must not be able to read +/// "published" out of a relay that never said so. +async fn publish_bounded( + client: &Client, + relay_url: &str, + event: &Event, +) -> Result { + let sent = tokio::time::timeout(PUBLISH_TIMEOUT, client.send_event_to([relay_url], event)).await; + let output = match sent { + Err(_) => return Err(PublishError::TimedOut), + Ok(Err(error)) => return Err(PublishError::Transport(error.to_string())), + Ok(Ok(output)) => output, + }; + if output.success.is_empty() { + let reason = output + .failed + .iter() + .map(|(url, reason)| format!("{url}: {reason}")) + .collect::>() + .join("; "); + return Err(PublishError::Refused(if reason.is_empty() { + "no relay accepted the write and none gave a reason".to_owned() + } else { + reason + })); + } + Ok(PublishReceipt { + event_id: output.val.to_string(), + relay: relay_url.to_owned(), + }) +} + +/// Ask the relay to serve one trivial REQ on the CURRENT session and wait for its `EOSE`. +/// +/// The `EOSE` is a RESPONSE the relay OWES us for a request we made, which is the whole point: a +/// broadcast we merely hope to receive proves nothing by its absence, because the relay may +/// legitimately decline to send it. And it must not be built on observing our own published event +/// — a single `Client` can never be delivered its own echo (the publish saves into the client's own +/// database, and the inbound handler drops anything already present), so such a probe can never +/// succeed at all. +async fn probe_relay_serves_our_reqs( + client: &Client, + buyer_pubkey: nostr_sdk::PublicKey, + timeout: Duration, +) -> bool { + // Receiver BEFORE the REQ — an EOSE that lands first would otherwise be missed. + let mut notifications = client.notifications(); + let probe_id = SubscriptionId::new(LIVENESS_PROBE_SUB_ID); + // `limit(0)` asks for zero stored events, so the relay's only work is the EOSE. Scoped to our + // own offers so the filter is narrow and unambiguous even if it ever did match. + let probe = Filter::new() + .kind(Kind::Custom(crate::kinds::JOB_OFFER_KIND)) + .author(buyer_pubkey) + .limit(0); + if let Err(error) = client.subscribe_with_id(probe_id, probe, None).await { + eprintln!("buyer relay liveness probe: REQ could not be sent ({error})"); + return false; + } + tokio::time::timeout(timeout, async { + loop { + match notifications.recv().await { + Ok(RelayPoolNotification::Message { + message: nostr_sdk::RelayMessage::EndOfStoredEvents(id), + .. + }) if id.to_string() == LIVENESS_PROBE_SUB_ID => return true, + Ok(_) => continue, + // The stream ending is itself a loss of liveness. + Err(_) => return false, + } + } + }) + .await + .unwrap_or(false) +} + +/// Drop the live socket and bring a fresh authenticated one up, returning once NIP-42 has completed +/// on the NEW connection. +/// +/// ORDER IS LOAD-BEARING. `Relay::disconnect` emits `RelayNotification::Shutdown` on the relay's own +/// notification channel; a receiver taken BEFORE the disconnect inherits that Shutdown and the auth +/// wait reads it as "relay shutdown before NIP-42 authentication" — on a socket that in fact +/// authenticated fine. A `broadcast::Receiver` only observes sends made after it subscribes, so +/// taking it AFTER the disconnect cannot inherit our own teardown, while still taking it BEFORE +/// `connect` so the one-shot `Authenticated` cannot be missed. Both halves are required. +async fn reconnect_and_authenticate( + client: &Client, + relay: &nostr_sdk::prelude::Relay, +) -> Result { + client.disconnect().await; + let mut relay_notifications = relay.notifications(); + client.connect().await; + client.wait_for_connection(CONNECT_WAIT).await; + relay_auth::wait_for_nip42_auth(&mut relay_notifications, CONNECT_WAIT).await +} + +#[cfg(test)] +mod tests { + use super::*; + + // TOOTH — the DEPENDENCY CONTRACT the buyer's write path rests on. + // + // A write refused with `auth-required:` must still land once NIP-42 completes. We do not + // implement that resend: `Relay::send_event` does (nostr-relay-pool relay/mod.rs:434-470). But + // it is conditional on `is_auto_authentication_enabled() && has_signer()`, and NOTHING errors or + // logs if either goes false — a "cleanup" that drops `automatic_authentication(true)` from client + // construction would delete the resend silently, and the symptom would surface far from that + // diff, on the money path. So the contract is pinned here rather than assumed. + // + // Both arms are required, and the second is what gives the first meaning: + // 1. built the way `spawn` builds it ⇒ the write LANDS (proving the SDK resend carried it). + // 2. built with auto-auth OFF ⇒ the write is REFUSED (proving arm 1 was not just an + // unauthenticated relay letting everything through — i.e. that the gate is real). + // + // The fixture is `RelayBuilderNip42Mode::Both`, which refuses an EVENT from an unauthenticated + // session with `MachineReadablePrefix::AuthRequired` (nostr-relay-builder local/inner.rs:420-438) + // — the exact wire condition mobee-relay produces. A fixture that served writes unauthenticated + // would make this whole test decorative. + // + // BITE: set `automatic_authentication(false)` in `spawn`, or bump to an SDK whose `send_event` + // no longer resends — either way arm 1 goes red. + // + // Note what the explicit `automatic_authentication(true)` in `spawn` does and does not buy: + // the SDK default is ALREADY true (`pool/options.rs:21`), so merely DELETING that line changes + // nothing today and this tooth would stay green. It is a guard against an upstream default + // flip and against a future caller disabling it — not against local deletion. Arm 2 is what + // demonstrates the option is load-bearing at all. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn an_auth_required_refusal_is_resent_after_auth_and_only_when_auto_auth_is_on() { + use nostr_relay_builder::prelude::{ + LocalRelay, RelayBuilder, RelayBuilderNip42, RelayBuilderNip42Mode, + }; + use nostr_sdk::prelude::EventBuilder; + + let relay_fixture = LocalRelay::new(RelayBuilder::default().nip42(RelayBuilderNip42 { + mode: RelayBuilderNip42Mode::Both, + })); + relay_fixture.run().await.expect("fixture relay run"); + let relay_url = relay_fixture.url().await.to_string(); + + let buyer = Keys::generate(); + let handle = spawn(buyer.clone(), &relay_url) + .await + .expect("buyer relay spawn"); + + let note = EventBuilder::text_note("mobee buyer write-path contract") + .sign(&buyer) + .await + .expect("sign"); + let note_id = note.id; + let receipt = handle + .publish(note) + .await + .expect("actor answered") + .expect("the write must land: the relay refuses it once with auth-required, and the \ + SDK is expected to authenticate and resend it"); + assert_eq!(receipt.event_id, note_id.to_string()); + + // Arm 2 — the drift case. Same fixture, same key, auto-auth OFF: the refusal must STAND. + // Without this arm, arm 1 would still pass against a relay that never enforced anything. + let drifted = Client::new(buyer.clone()); + drifted.automatic_authentication(false); + drifted + .pool() + .add_relay(&relay_url, RelayOptions::default().reconnect(true)) + .await + .expect("add relay"); + drifted.connect().await; + drifted.wait_for_connection(CONNECT_WAIT).await; + let drifted_note = EventBuilder::text_note("mobee buyer write-path drift arm") + .sign(&buyer) + .await + .expect("sign"); + let refused = publish_bounded(&drifted, &relay_url, &drifted_note).await; + assert!( + matches!(refused, Err(PublishError::Refused(_))), + "with automatic_authentication(false) the auth-gated write must stay refused — if this \ + passes, the fixture is not enforcing NIP-42 and arm 1 proves nothing, got {refused:?}" + ); + } + + // TOOTH — an empty accepted-set is a FAILURE, never a quiet success. + // + // `send_event_to` returns `Ok(output)` even when the relay refused: the reason goes into + // `output.failed` and `output.success` is left empty. Reading only the outer `Result` would + // report a refused write as published — and on the money path "we think we sent it" is the + // ambiguity that costs real sats. Covered by arm 2 above at the integration level; asserted here + // on the classifier itself so the rule survives a refactor of the caller. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_relay_that_accepts_nothing_is_reported_as_refused() { + use nostr_relay_builder::prelude::{ + LocalRelay, RelayBuilder, RelayBuilderNip42, RelayBuilderNip42Mode, + }; + use nostr_sdk::prelude::EventBuilder; + + let relay_fixture = LocalRelay::new(RelayBuilder::default().nip42(RelayBuilderNip42 { + mode: RelayBuilderNip42Mode::Both, + })); + relay_fixture.run().await.expect("fixture relay run"); + let relay_url = relay_fixture.url().await.to_string(); + + let buyer = Keys::generate(); + let client = Client::new(buyer.clone()); + client.automatic_authentication(false); + client + .pool() + .add_relay(&relay_url, RelayOptions::default().reconnect(true)) + .await + .expect("add relay"); + client.connect().await; + client.wait_for_connection(CONNECT_WAIT).await; + + let note = EventBuilder::text_note("refused") + .sign(&buyer) + .await + .expect("sign"); + match publish_bounded(&client, &relay_url, ¬e).await { + Err(PublishError::Refused(reason)) => assert!( + !reason.is_empty(), + "a refusal must carry the relay's own reason, or an operator cannot triage it" + ), + other => panic!("a refused write must not read as published, got {other:?}"), + } + } + + // TOOTH — `spawn` must NOT wait for the relay handshake, even when the relay is unreachable. + // + // The daemon binds its Unix socket after bootstrap, and connect-or-spawn gives a cold daemon + // only SPAWN_READY_TIMEOUT (10s, crates/mobee/src/daemon.rs:21) to appear. If the session came + // up inside `spawn`, an unreachable relay would hold the handshake for CONNECT_WAIT (20s), the + // socket would miss that deadline, and EVERY MCP call would report a daemon that failed to + // start — while it was in fact coming up fine. The failure would look like a daemon bug and the + // cause would be a relay one layer away. + // + // 5s is deliberately loose: registration is sub-millisecond, and the regression it guards + // against is 20s. Anything in between means someone moved the handshake back into `spawn`. + // + // BITE: await `connect_and_authenticate` inside `spawn` instead of at the top of the actor, and + // this goes red against the dead port below. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn spawn_returns_immediately_when_the_relay_is_unreachable() { + // Port 1 is privileged and nothing listens there; the dial cannot succeed. + let unreachable = "ws://127.0.0.1:1"; + let handle = tokio::time::timeout( + Duration::from_secs(5), + spawn(Keys::generate(), unreachable), + ) + .await + .expect( + "spawn must not block on the relay handshake — a daemon whose socket waits on an \ + unreachable relay misses connect-or-spawn's 10s readiness deadline", + ) + .expect("registering an unreachable relay is not a configuration error"); + assert_eq!(handle.relay_url(), unreachable); + } + + // TOOTH (#173 class, new site) — the relay handle's round-trip is BOUNDED. + // + // This handle is reached from the payment worker's runtime and from daemon tasks that own the + // trade loop. An unbounded round-trip here would park those callers permanently and silently, + // exactly as the signer's did — a new site for a class we already closed once, which is why it + // is toothed at BOTH sites rather than trusted to the pattern. + // + // Time is paused, so the production bound elapses instantly; the OUTER timeout is what makes a + // revert fail cleanly instead of hanging the suite. + #[tokio::test(start_paused = true)] + async fn a_stalled_relay_actor_cannot_park_the_caller() { + let (tx, mut rx) = mpsc::channel::(8); + // The stalled actor: receive, then hold. Never answers, never drops a reply sender. + tokio::spawn(async move { + let mut held = Vec::new(); + while let Some(command) = rx.recv().await { + held.push(command); + } + }); + let handle = RelayHandle { + tx, + relay_url: "wss://example.invalid".to_owned(), + }; + + let outer = Duration::from_secs(600); + for attempt in 1..=2 { + let call = tokio::time::timeout(outer, handle.probe()); + let outcome = call.await.unwrap_or_else(|_| { + panic!( + "attempt {attempt}: the relay round-trip never returned — an unbounded \ + timer-less await here parks the calling task permanently and silently" + ) + }); + let error = outcome.expect_err("a stalled actor cannot answer a probe"); + assert!( + error.to_string().contains("probe") && error.to_string().contains("never answered"), + "attempt {attempt}: the failure must NAME the call and the leg, got {error}" + ); + } + } +} From 13a833f80244586e99af568867bec4a5dbe47db4 Mon Sep 17 00:00:00 2001 From: orveth Date: Sun, 26 Jul 2026 23:59:44 -0700 Subject: [PATCH 3/7] buyer: wake the daemon's job waits on event arrival instead of polling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit get_job(wait_for=...) slept 400ms and then rebuilt a Client, reconnected, and ran three sequential fetches — every iteration. One ten-second wait cost up to twenty-five fresh WebSocket dials, and a delivery was discovered on the next poll window rather than when it landed. That is the dominant term in trade latency. The subscription is buyer-keyed, not job-keyed. Claims, results and feedback all p-tag the buyer, so ONE static filter covers every job this buyer has or will ever have: no per-job REQ to open when a job is posted, none to close when it settles, no filter to rebuild on reconnect. A dynamic e-tag list would be a subscription-lifecycle problem taken on for no benefit. The subscription is the WAKE; the fetch is the TRUTH. An arriving event means only "something changed for this job" and the view is re-read from the relay. Assembling the view out of the event stream would duplicate the assembly and force trusting a stream that may be partial; this way a missed event costs a re-check, never a wrong answer. A three-second backstop bounds how long a wait can be wrong when an event never arrives at all. The poll variant is untouched, because the CLI has no persistent session to ride. What the two share is the one thing that must never diverge: view_is_ready decides for both. The mechanisms differ deliberately; if the definitions of "ready" ever differed, the same job would read complete to the daemon and pending to the CLI. Lagged is not an error. It means this receiver fell behind and events were dropped for it — so something happened, and the answer is to re-check. Treating it as a failure would turn a busy relay into spurious timeouts that look like a relay fault. Commands are served by a sibling task of the event forwarder: a publish may occupy its full timeout, and if that ran in the forwarding task no job event would reach a waiter for the duration. The job REQ is re-asserted after auth on boot and on every reconnect rather than trusted to replay, because an unauthenticated p-filtered REQ can be CLOSED as restricted, which the SDK treats as removal — leaving a subscription that is gone while the socket still looks healthy. Co-Authored-By: Claude Opus 5 (1M context) --- crates/mobee-core/src/buyer/mod.rs | 6 +- crates/mobee-core/src/buyer/relay.rs | 166 +++++++++++-- crates/mobee-core/src/job_lifecycle.rs | 314 ++++++++++++++++++++++++- 3 files changed, 457 insertions(+), 29 deletions(-) diff --git a/crates/mobee-core/src/buyer/mod.rs b/crates/mobee-core/src/buyer/mod.rs index 2b0e99a8..a84c3293 100644 --- a/crates/mobee-core/src/buyer/mod.rs +++ b/crates/mobee-core/src/buyer/mod.rs @@ -444,7 +444,11 @@ async fn get_job(context: &BuyerContext, id: Value, params: Value) -> Response { timeout_secs: params.timeout_secs, include_display_names: params.include_display_names, }; - match job_lifecycle::get_job_async(&context.home, request).await { + // Subscribe BEFORE the wait begins its first fetch — an event landing in the gap between the + // fetch and the subscribe would be lost, and the wait would then sleep on a view that was + // already stale when it read it. + let events = context.relay.subscribe_events(); + match job_lifecycle::get_job_awaiting_events_async(&context.home, request, events).await { Ok(view) => Response::ok(id, json!(view)), Err(error) => Response::err(id, CODE_INTERNAL, error.to_string()), } diff --git a/crates/mobee-core/src/buyer/relay.rs b/crates/mobee-core/src/buyer/relay.rs index 3c0c1c3b..938debfa 100644 --- a/crates/mobee-core/src/buyer/relay.rs +++ b/crates/mobee-core/src/buyer/relay.rs @@ -32,13 +32,14 @@ //! `automatic_authentication(true)` explicitly as a drift guard, and a tooth pins the contract //! itself: a write refused with `auth-required:` must still land once auth completes. +use std::sync::Arc; use std::time::Duration; use nostr_sdk::prelude::{ Client, Event, Filter, Keys, Kind, RelayOptions, RelayPoolNotification, RelayUrl, SubscriptionId, }; -use tokio::sync::{mpsc, oneshot}; +use tokio::sync::{broadcast, mpsc, oneshot}; use crate::relay_auth::{self, AuthWait}; @@ -69,6 +70,13 @@ const RELAY_CALL_TIMEOUT: Duration = Duration::from_secs(60); /// Stable id for the liveness REQ, so a relay `CLOSED` names it. const LIVENESS_PROBE_SUB_ID: &str = "mobee-buyer-liveness"; +/// Stable id for the buyer's job-event REQ, so a relay `CLOSED` names which subscription died. +const JOB_EVENTS_SUB_ID: &str = "mobee-buyer-jobs"; + +/// Depth of the fan-out channel carrying job events to waiters. A waiter that falls behind sees +/// `Lagged`, which is a re-check signal and NOT an error — see [`RelayHandle::subscribe_events`]. +const JOB_EVENT_CHANNEL_DEPTH: usize = 256; + enum Command { Publish { event: Box, @@ -155,6 +163,7 @@ impl std::error::Error for RelayBootError {} pub struct RelayHandle { tx: mpsc::Sender, relay_url: String, + events: broadcast::Sender>, } impl RelayHandle { @@ -163,6 +172,21 @@ impl RelayHandle { &self.relay_url } + /// Live seller-authored events for this buyer's jobs (claims, results, feedback). + /// + /// Synchronous and cheap by design — no round-trip to the actor. That matters because a waiter + /// MUST subscribe BEFORE it does its catch-up fetch: an event landing in the gap between the + /// fetch and the subscribe is gone, and the waiter then sleeps until its deadline holding a + /// view that was already stale when it read it. Making this a plain call keeps the correct + /// order the natural one to write. + /// + /// `RecvError::Lagged` is NOT an error. It means this receiver fell behind and events were + /// dropped for it — i.e. *something happened*. A waiter must treat it as a re-check signal; + /// treating it as a failure turns a busy relay into spurious timeouts. + pub fn subscribe_events(&self) -> broadcast::Receiver> { + self.events.subscribe() + } + /// Send one command and await its reply, with BOTH legs bounded — see the module doc. async fn round_trip( &self, @@ -263,37 +287,144 @@ pub async fn spawn(keys: Keys, relay_url: &str) -> Result(64); + let (events, _) = broadcast::channel::>(JOB_EVENT_CHANNEL_DEPTH); let owned_url = relay_url.to_owned(); + + // The session task: bring the socket up, subscribe, then FAN OUT events for the rest of its + // life. Commands are served by a sibling task rather than this one, because a publish may + // legitimately occupy PUBLISH_TIMEOUT — and if that ran here, no job event would be forwarded + // for the duration. A delivery arriving during a slow write would reach waiters late or, past + // the channel depth, not at all. + let command_client = client.clone(); + let event_sender = events.clone(); tokio::spawn(async move { connect_and_authenticate(&client, &relay, "boot").await; - while let Some(command) = rx.recv().await { - match command { - Command::Publish { event, reply } => { - let _ = reply.send(publish_bounded(&client, &owned_url, &event).await); - } - Command::Probe { reply } => { - let _ = reply.send( - probe_relay_serves_our_reqs(&client, public_key, LIVENESS_PROBE_TIMEOUT) + // AFTER auth, never before: an unauthenticated `#p` REQ can be CLOSED with `restricted:`, + // which nostr-sdk treats as Remove — the subscription is dropped and no later + // `resubscribe()` brings it back. Ordering here is the difference between a live + // subscription and a permanently deaf one that still looks connected. + subscribe_job_events(&client, public_key).await; + + tokio::spawn(async move { + while let Some(command) = rx.recv().await { + match command { + Command::Publish { event, reply } => { + let _ = + reply.send(publish_bounded(&command_client, &owned_url, &event).await); + } + Command::Probe { reply } => { + let _ = reply.send( + probe_relay_serves_our_reqs( + &command_client, + public_key, + LIVENESS_PROBE_TIMEOUT, + ) .await, - ); - } - Command::Reconnect { reply } => { - let outcome = reconnect_and_authenticate(&client, &relay).await; - if let Ok(wait) = &outcome { - report_auth_wait("reconnect", Ok(*wait)); + ); + } + Command::Reconnect { reply } => { + let outcome = reconnect_and_authenticate(&command_client, &relay).await; + if let Ok(wait) = &outcome { + report_auth_wait("reconnect", Ok(*wait)); + // Re-assert the job REQ on the NEW session rather than trusting the + // pool to replay it, and only after auth has completed — same ordering + // rule as boot. The stable id makes this idempotent. + subscribe_job_events(&command_client, public_key).await; + } + let _ = reply.send(outcome.map_err(|error| error.to_string())); } - let _ = reply.send(outcome.map_err(|error| error.to_string())); } } - } + }); + + forward_job_events(&client, event_sender).await; }); Ok(RelayHandle { tx, relay_url: relay_url.to_owned(), + events, }) } +/// Subscribe to every seller-authored event addressed to this buyer. +/// +/// ONE STATIC FILTER covers every job this buyer has or will ever have, because claims, results and +/// feedback all `p`-tag the buyer (`gateway::claim_draft` / `result_draft` / `error_draft`). So +/// there is no per-job subscription to open when a job is posted, none to close when it settles, +/// and no filter to rebuild on reconnect — a dynamic `#e`-list would be a subscription-lifecycle +/// problem taken on for no benefit. +/// +/// The `#t=mobee` guard keeps a foreign event squatting these kinds from ever being delivered. +async fn subscribe_job_events(client: &Client, buyer_pubkey: nostr_sdk::PublicKey) { + let filter = Filter::new() + .kinds([ + Kind::Custom(crate::kinds::JOB_CLAIM_KIND), + Kind::Custom(crate::kinds::JOB_RESULT_KIND), + Kind::Custom(crate::kinds::JOB_FEEDBACK_KIND), + ]) + .hashtag(crate::gateway::MOBEE_TAG) + .pubkey(buyer_pubkey); + if let Err(error) = client + .subscribe_with_id(SubscriptionId::new(JOB_EVENTS_SUB_ID), filter, None) + .await + { + eprintln!( + "buyer relay: job-event subscription could not be opened ({error}); waiters fall back \ + to their safety re-check until a reconnect restores it" + ); + } +} + +/// Fan out job events to every waiter for the life of the session. +/// +/// Nothing here interprets an event: an arrival means only "something changed for some job", and +/// the waiter re-reads the authoritative view. The subscription is the WAKE; the fetch is the TRUTH. +/// Assembling a view from this stream would duplicate view-assembly and force us to trust a stream +/// that may be partial. +/// +/// No own-echo hazard exists on this path and it must stay that way: every kind here is +/// SELLER-authored, so the buyer never needs to observe its own event — which a single client +/// cannot do anyway. A future "did my own publish land" check must NOT be built on this stream. +async fn forward_job_events(client: &Client, events: broadcast::Sender>) { + let mut notifications = client.notifications(); + loop { + match notifications.recv().await { + Ok(RelayPoolNotification::Message { + message: nostr_sdk::RelayMessage::Closed { subscription_id, message }, + .. + }) if subscription_id.to_string() == JOB_EVENTS_SUB_ID => { + // A CLOSED while the socket is up is the silent-deafness case: connected, and + // serving nothing. Name it — a waiter's safety re-check will carry the load until + // a reconnect re-subscribes, but an operator needs to see why waits got slow. + eprintln!( + "buyer relay: the relay CLOSED our job-event subscription ({message}); waits \ + degrade to the safety re-check until a reconnect re-subscribes" + ); + } + Ok(RelayPoolNotification::Event { + subscription_id, + event, + .. + }) if subscription_id.to_string() == JOB_EVENTS_SUB_ID => { + // A send failure means only that nobody is waiting right now — not a fault. + let _ = events.send(Arc::new(*event)); + } + Ok(_) => continue, + // The notification stream ending means the pool is gone; so is this session. + Err(broadcast::error::RecvError::Closed) => return, + // We fell behind the relay. We cannot know what we missed, so say so: every waiter's + // safety re-check is what recovers the state we dropped. + Err(broadcast::error::RecvError::Lagged(missed)) => { + eprintln!( + "buyer relay: fell behind the relay notification stream, {missed} dropped; \ + waiters recover via their safety re-check" + ); + } + } + } +} + /// Bring the session up on a socket that is not yet connected, and report the handshake. /// /// The notification receiver is taken BEFORE `connect` because `Authenticated` is emitted once and @@ -614,6 +745,7 @@ mod tests { let handle = RelayHandle { tx, relay_url: "wss://example.invalid".to_owned(), + events: broadcast::channel(1).0, }; let outer = Duration::from_secs(600); diff --git a/crates/mobee-core/src/job_lifecycle.rs b/crates/mobee-core/src/job_lifecycle.rs index 0235cb53..2a7a1ca7 100644 --- a/crates/mobee-core/src/job_lifecycle.rs +++ b/crates/mobee-core/src/job_lifecycle.rs @@ -122,12 +122,26 @@ pub struct GetJobRequest { pub include_display_names: bool, } -#[derive(Clone, Debug, PartialEq, Eq)] +#[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum WaitFor { Claim, Result, } +/// Whether a fetched view satisfies the caller's `wait_for` condition. +/// +/// The ONE definition of "ready", shared by every waiter. The buyer daemon waits by subscription +/// and the CLI waits by poll — the mechanisms differ deliberately, but if the two ever disagreed +/// about what they were waiting FOR, the same job would read complete to one and pending to the +/// other. Extracting the predicate rather than the loop is what makes that disagreement impossible +/// to write. +pub fn view_is_ready(view: &JobView, wait_for: WaitFor) -> bool { + match wait_for { + WaitFor::Claim => view.live_claim_id.is_some(), + WaitFor::Result => !view.results.is_empty(), + } +} + impl WaitFor { pub fn parse(raw: &str) -> Result { match raw { @@ -643,22 +657,14 @@ pub async fn get_job_async( let mut view = fetch_job_view_async(home, &keys, &request.job_id, fetch_timeout, now_unix()) .await?; - let ready = match wait_for { - WaitFor::Claim => view.live_claim_id.is_some(), - WaitFor::Result => !view.results.is_empty(), - }; - view.pending = !ready; + view.pending = !view_is_ready(&view, wait_for); maybe_attach_display_names_async(home, &mut view, request.include_display_names).await; return Ok(view); } let this_fetch = fetch_timeout.min(remaining); let mut view = fetch_job_view_async(home, &keys, &request.job_id, this_fetch, now_unix()).await?; - let ready = match wait_for { - WaitFor::Claim => view.live_claim_id.is_some(), - WaitFor::Result => !view.results.is_empty(), - }; - if ready { + if view_is_ready(&view, wait_for) { view.pending = false; maybe_attach_display_names_async(home, &mut view, request.include_display_names).await; return Ok(view); @@ -672,6 +678,105 @@ pub async fn get_job_async( } } +/// How long a subscription-driven wait will sit before re-reading the view anyway. +/// +/// This is a BACKSTOP, not the mechanism. Events resolve the wait in milliseconds; this only bounds +/// how long a wait can be wrong if an event never arrives — a relay that CLOSED our subscription, a +/// fan-out we fell behind on, an event filtered away upstream. Without it, a single missed event +/// costs the caller its whole remaining deadline. +const SAFETY_RECHECK: Duration = Duration::from_secs(3); + +/// `get_job` for a caller that holds a live subscription to this buyer's job events. +/// +/// Same contract and same readiness rule as [`get_job_async`] — [`view_is_ready`] decides for both +/// — but it WAKES on event arrival instead of sleeping 400ms between full reconnect-and-refetch +/// cycles. The poll variant stays exactly as it is for callers with no persistent session (the CLI); +/// this is not a replacement, it is the daemon's path. +/// +/// THE SUBSCRIPTION IS THE WAKE; THE FETCH IS THE TRUTH. An arriving event means only "something +/// changed for this job" — the view is then re-read from the relay. Assembling the view out of the +/// event stream would duplicate the assembly logic and force trusting a stream that may be partial; +/// this way a missed event costs a re-check, never a wrong answer. +/// +/// The caller must have subscribed BEFORE calling: `events` is passed in already-live precisely so +/// the subscribe cannot land after the first fetch and lose an event in the gap. +pub async fn get_job_awaiting_events_async( + home: &MobeeHome, + request: GetJobRequest, + mut events: tokio::sync::broadcast::Receiver>, +) -> Result { + let keys = buyer_keys(home)?; + let fetch_timeout = Duration::from_secs(DEFAULT_FETCH_TIMEOUT_SECS); + + let Some(wait_for) = request.wait_for else { + let mut view = + fetch_job_view_async(home, &keys, &request.job_id, fetch_timeout, now_unix()).await?; + view.pending = false; + maybe_attach_display_names_async(home, &mut view, request.include_display_names).await; + return Ok(view); + }; + + let wait_cap_secs = request + .timeout_secs + .unwrap_or(WAIT_FOR_CAP_SECS) + .min(WAIT_FOR_CAP_SECS); + let deadline = tokio::time::Instant::now() + Duration::from_secs(wait_cap_secs); + + loop { + let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); + let this_fetch = fetch_timeout.min(remaining.max(Duration::from_millis(1))); + let mut view = + fetch_job_view_async(home, &keys, &request.job_id, this_fetch, now_unix()).await?; + if view_is_ready(&view, wait_for) { + view.pending = false; + maybe_attach_display_names_async(home, &mut view, request.include_display_names).await; + return Ok(view); + } + let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); + if remaining.is_zero() { + view.pending = true; + maybe_attach_display_names_async(home, &mut view, request.include_display_names).await; + return Ok(view); + } + await_job_event(&mut events, &request.job_id, remaining.min(SAFETY_RECHECK)).await; + } +} + +/// Wait until an event referencing `job_id` arrives, the window elapses, or the fan-out reports we +/// fell behind. Returns in all three cases — the caller's next act is the same either way: re-read +/// the view. +/// +/// `Lagged` is deliberately NOT an error. It says this receiver missed events, which means +/// something happened; returning here re-checks, whereas treating it as a failure would turn a busy +/// relay into a spurious timeout. `Closed` also returns rather than erroring: the session is gone, +/// and the caller's deadline plus its next fetch is what surfaces that honestly. +async fn await_job_event( + events: &mut tokio::sync::broadcast::Receiver>, + job_id: &str, + window: Duration, +) { + let _ = tokio::time::timeout(window, async { + loop { + match events.recv().await { + Ok(event) if event_references_job(&event, job_id) => return, + Ok(_) => continue, + Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => return, + Err(tokio::sync::broadcast::error::RecvError::Closed) => return, + } + } + }) + .await; +} + +/// True when `event` carries an `e` tag naming this job's offer — the tag every claim, result and +/// feedback roots itself on. +fn event_references_job(event: &nostr_sdk::Event, job_id: &str) -> bool { + event.tags.iter().any(|tag| { + let parts = tag.as_slice(); + parts.first().map(String::as_str) == Some("e") && parts.get(1).map(String::as_str) == Some(job_id) + }) +} + /// Accept a live claim: persist the pay-bind, then publish the `accepted` AWARD (kind-3405). /// Sync entry for CLI/tests — nested call fails fast; MCP uses [`accept_claim_async`]. pub fn accept_claim( @@ -3150,6 +3255,193 @@ mod tests { } } + // The ONE definition of "ready", asserted as a table so the two waiters cannot drift apart. + // The poll path and the subscription path deliberately differ in MECHANISM; if they ever + // differed in what they were waiting FOR, the same job would read complete to the daemon and + // pending to the CLI. Extracting the predicate is what makes that unwritable — this pins it. + #[test] + fn readiness_is_one_shared_definition() { + let mut view = JobView { + job_id: "job".into(), + offer: None, + claims: Vec::new(), + results: Vec::new(), + live_claim_id: None, + accepted: None, + pending: false, + }; + assert!(!view_is_ready(&view, WaitFor::Claim)); + assert!(!view_is_ready(&view, WaitFor::Result)); + + view.live_claim_id = Some("claim".into()); + assert!(view_is_ready(&view, WaitFor::Claim)); + assert!( + !view_is_ready(&view, WaitFor::Result), + "a live claim is not a delivery — waiting for a result must not be satisfied by one" + ); + } + + /// A signed event carrying nothing but an `e` tag. The wake path inspects only that tag, so + /// this is the whole of what a forwarded event contributes: the signal, never the truth. + async fn wake_event(signer: &nostr_sdk::Keys, job_id: &str) -> std::sync::Arc { + use nostr_sdk::prelude::{EventBuilder, Tag}; + let event = EventBuilder::new(nostr_sdk::Kind::Custom(JOB_RESULT_KIND), "") + .tag(Tag::parse(["e", job_id]).expect("e tag")) + .sign(signer) + .await + .expect("sign wake"); + std::sync::Arc::new(event) + } + + // TOOTH — the wake primitive returns for every reason the caller must re-check for, and for + // nothing else. + // + // `Lagged` is the reviewer trap: it means this receiver fell behind and events were DROPPED for + // it — i.e. something happened. Returning re-checks; treating it as a failure would turn a busy + // relay into a spurious timeout, which then looks like a relay fault rather than a client bug. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn the_wake_returns_on_a_match_on_lag_and_on_expiry_but_not_on_an_unrelated_event() { + let seller = nostr_sdk::Keys::generate(); + let (tx, mut rx) = tokio::sync::broadcast::channel(4); + + // A matching event wakes it. + tx.send(wake_event(&seller, "job-a").await).expect("send"); + tokio::time::timeout( + Duration::from_secs(2), + await_job_event(&mut rx, "job-a", Duration::from_secs(30)), + ) + .await + .expect("a matching event must wake the wait"); + + // An event for a DIFFERENT job does not: waking on it would send the caller into a pointless + // re-fetch on every other job's traffic. + tx.send(wake_event(&seller, "job-b").await).expect("send"); + assert!( + tokio::time::timeout( + Duration::from_millis(400), + await_job_event(&mut rx, "job-a", Duration::from_secs(30)), + ) + .await + .is_err(), + "an unrelated job's event must not wake this wait" + ); + + // Lag wakes it — we cannot know what we missed, so the only safe response is to re-check. + for index in 0..8 { + tx.send(wake_event(&seller, &format!("flood-{index}")).await) + .expect("send"); + } + tokio::time::timeout( + Duration::from_secs(2), + await_job_event(&mut rx, "job-a", Duration::from_secs(30)), + ) + .await + .expect("Lagged must wake the wait — it is a re-check signal, not a failure"); + + // And the window expiring returns rather than hanging. + tokio::time::timeout( + Duration::from_secs(2), + await_job_event(&mut rx, "job-a", Duration::from_millis(200)), + ) + .await + .expect("the window must bound the wait"); + } + + // TOOTH — the wait resolves ON EVENT ARRIVAL, not on the safety re-check. + // + // This is the whole point of the piece: `get_job(wait_for=…)` used to sleep 400ms and then + // rebuild a Client, reconnect, and run three sequential fetches, every iteration. Now an event + // wakes it and it re-reads once. + // + // The TIMING assertion is what makes this a tooth rather than a demonstration. The result is + // genuinely on the relay, so a wait with no event wake still succeeds — three seconds later, at + // SAFETY_RECHECK. Asserting only "it returned ready" would pass with the subscription removed + // entirely. Asserting it returned FASTER than the backstop is what proves the event did the work. + // + // The result is published by a SEPARATE identity (a seller), which is both the faithful shape + // and the safe one: a single client can never observe its own published events. + // + // BITE: drop the `tx.send(...)` wake below and this goes red on elapsed, not on correctness. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_wait_resolves_on_arrival_rather_than_on_the_safety_recheck() { + use nostr_relay_builder::prelude::{LocalRelay, RelayBuilder}; + + let relay = LocalRelay::new(RelayBuilder::default()); + relay.run().await.expect("relay run"); + let (root, mut home) = temp_job_home("wait-on-arrival"); + home.config.relay_url = relay.url().await.to_string(); + + let posted = post_job_async( + &home, + PostJobRequest { + task: "wake on arrival".into(), + output: "text/plain".into(), + amount_sats: 2, + seller_pubkey: Some(nostr_sdk::Keys::generate().public_key().to_hex()), + untargeted: false, + deadline_unix: Some(now_unix() + 3_600), + repo: None, + branch: None, + job: JobKind::FromScratch, + }, + ) + .await + .expect("post job"); + let job_id = posted.job_id.clone(); + let buyer_pubkey = buyer_keys(&home).expect("keys").public_key().to_hex(); + + let (tx, rx) = tokio::sync::broadcast::channel(8); + let waiting_home = home.clone(); + let waiting_job = job_id.clone(); + let started = tokio::time::Instant::now(); + let waiter = tokio::spawn(async move { + get_job_awaiting_events_async( + &waiting_home, + GetJobRequest { + job_id: waiting_job, + wait_for: Some(WaitFor::Result), + timeout_secs: Some(30), + include_display_names: false, + }, + rx, + ) + .await + }); + + // Let the catch-up fetch finish and the waiter park on the fan-out, so what we measure is + // the wake and not a race with the first read. + tokio::time::sleep(Duration::from_millis(400)).await; + + let seller = nostr_sdk::Keys::generate(); + let draft = crate::gateway::result_draft( + &job_id, + &buyer_pubkey, + "text/plain", + 2, + "job-hash", + "seller-signature", + "delivered", + None, + &[], + ); + publish_draft_async(&home, &seller, &draft) + .await + .expect("publish the seller result"); + tx.send(wake_event(&seller, &job_id).await).expect("wake"); + + let view = waiter.await.expect("join").expect("wait"); + let elapsed = started.elapsed(); + assert!(!view.pending, "the wait must report the job ready"); + assert!(!view.results.is_empty(), "the delivered result must be in the view"); + assert!( + elapsed < SAFETY_RECHECK, + "the wait must resolve on the EVENT, not on the {SAFETY_RECHECK:?} backstop — took \ + {elapsed:?}; if this fails on timing alone the subscription is not doing the work" + ); + + let _ = std::fs::remove_dir_all(&root); + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn get_job_default_skips_kind_zero_fetch() { use nostr_relay_builder::prelude::{LocalRelay, RelayBuilder}; From 233fadbc0e7cf0efac34bf44ff66e829701ba1b6 Mon Sep 17 00:00:00 2001 From: "worker:mobee-buyer-lazystart" Date: Mon, 27 Jul 2026 00:42:22 -0700 Subject: [PATCH 4/7] buyer: settle delivered work automatically instead of waiting to be asked MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An awarded job used to sit until someone ran collect by hand. The daemon now watches for delivered results and settles them itself, so a seller is paid in seconds rather than whenever the buyer next looks. The watcher adds no money authority. It commits no new money; it converts a reservation the award already created. Both the collect RPC and the watcher go through one `settle_job` — money lock, budget gate, pay-then-flip — so there is a single path to the spend gate rather than two that can drift. Every gate it relies on already lives inside the sealed pay path: the job-hash recomputed from the buyer's signed offer, the offer-authoritative amount, the seller co-signature, the creq check, single-settlement, and the pre-spend commit tip-match. Adding a second verification layer here would be a second thing to keep in agreement, so there isn't one. What the watcher may settle is read from durable state, never from the event that woke it: an arriving result only says "look again", and optionally narrows which job to look at. The work set is derived — an award whose reservation is still `reserved` — so no second lifecycle can disagree with the ledger the budget already trusts. The loop sweeps before it ever waits, which is what collects a delivery that landed while the daemon was down. Awards are recorded where they are published, at the one reserve-then-award seam both the manual and background paths reach the relay through. A store failure there does not fail the award: the event is already public and the reservation already held, so reporting failure would be untrue and would release committed funds. It says what the operator lost instead. The backstop is a fixed-cadence interval rather than a timer re-armed each pass, because a re-armed timer is pushed back by every arriving event — including the ones the watcher ignores — and would go quiet under exactly the steady traffic it exists to survive. Losing the subscription degrades to periodic sweeps rather than stopping; settling never depended on that subscription. Co-Authored-By: Claude Opus 5 (1M context) --- crates/mobee-core/src/buyer/lifecycle.rs | 80 +++- crates/mobee-core/src/buyer/mod.rs | 490 ++++++++++++++++++++++- crates/mobee-core/src/buyer/store.rs | 169 +++++++- crates/mobee-core/src/job_lifecycle.rs | 2 +- 4 files changed, 718 insertions(+), 23 deletions(-) diff --git a/crates/mobee-core/src/buyer/lifecycle.rs b/crates/mobee-core/src/buyer/lifecycle.rs index 2a1c8f4b..d766dcd9 100644 --- a/crates/mobee-core/src/buyer/lifecycle.rs +++ b/crates/mobee-core/src/buyer/lifecycle.rs @@ -199,7 +199,34 @@ where .map_err(AwardError::Reserve)?; match publish().await { - Ok(outcome) => Ok(outcome), + Ok(outcome) => { + // Record the PUBLISHED award. This is the one seam both the manual and the auto award + // path go through, so recording here covers both by construction — recording at the two + // call sites instead would let one drift silently. + // + // A store failure here does NOT fail the award: the 3405 is already public and the + // reservation is already held, so reporting failure would be a lie that also releases + // funds the buyer has genuinely committed. The window is narrow (the `reserve` write + // above went through the same store moments earlier), but it is not nothing, and the + // consequence is specific: without this row the delivery watcher cannot see the job, so + // it will not auto-settle. Say exactly that, unconditionally — an operator reading the + // log must not have to infer why one job stopped settling itself. + if let Err(error) = store.record_award( + job_id, + &outcome.claim_id, + &outcome.award_event_id, + &outcome.seller_pubkey, + amount, + now_unix, + ) { + eprintln!( + "buyer: award for {job_id} published ({}) but recording it failed ({error}); \ + this job will NOT be auto-settled on delivery — collect it manually", + outcome.award_event_id + ); + } + Ok(outcome) + } Err(error) => { // No award reached the relay — reclaim the reservation rather than strand the funds. store @@ -548,6 +575,57 @@ mod tests { store.reservation(&job).expect("read").map(|(state, _)| state), Some(super::super::reservations::ReservationState::Released) ); + // Nothing was published, so nothing may be recorded as awarded — otherwise the delivery + // watcher would sweep a job that has no award on the relay. + assert!( + store.award_record(&job).expect("read").is_none(), + "a failed publish must record no award" + ); + let _ = std::fs::remove_file(&path); + } + + // The published award is recorded at the reserve-then-award SEAM, not at the call sites. Both + // the manual `award` RPC and the background auto-award reach the relay through this one + // function, so recording here is what makes the delivery watcher able to see EITHER — and what + // stops the two paths from drifting apart. Red-on-revert: drop the `record_award` call and the + // job becomes invisible to the watcher (`awarded_unsettled_job_ids` returns empty), which is + // exactly the "awarded but never auto-settled" failure. + #[tokio::test(flavor = "current_thread")] + async fn a_published_award_is_recorded_at_the_single_seam() { + let (store, path) = fresh_store("award-recorded"); + let job = "a".repeat(64); + let claim = "c".repeat(64); + let award_event = "e".repeat(64); + + let outcome = award_with_reservation(&store, &job, 40, 100, u64::MAX, 0, 7, || { + let (job, claim, award_event) = (job.clone(), claim.clone(), award_event.clone()); + async move { + Ok(AwardClaimOutcome { + award_event_id: award_event, + job_id: job, + claim_id: claim, + seller_pubkey: SELLER_HEX.to_owned(), + quoted_mints: Vec::new(), + }) + } + }) + .await + .expect("award"); + assert_eq!(outcome.award_event_id, award_event); + + let recorded = store.award_record(&job).expect("read").expect("award recorded"); + assert_eq!(recorded.award_event_id, award_event); + assert_eq!(recorded.claim_id, claim); + assert_eq!(recorded.seller_pubkey, SELLER_HEX); + assert_eq!(recorded.amount_sats, 40, "the RESERVED amount, not a seller-quoted one"); + assert_eq!(recorded.awarded_at_unix, 7); + + // And the job is now the delivery watcher's work — awarded, reservation still held. + assert_eq!( + store.awarded_unsettled_job_ids().expect("awarded"), + vec![job], + "a published award must put the job in the watcher's work set" + ); let _ = std::fs::remove_file(&path); } diff --git a/crates/mobee-core/src/buyer/mod.rs b/crates/mobee-core/src/buyer/mod.rs index a84c3293..c977b1c6 100644 --- a/crates/mobee-core/src/buyer/mod.rs +++ b/crates/mobee-core/src/buyer/mod.rs @@ -64,6 +64,10 @@ const RELAY_TIMEOUT: Duration = Duration::from_secs(5); /// How often the background auto-award task re-checks the relay for a payable claim, until one /// appears or the offer deadline passes. Bounded polling (no tight spin on a live-but-unpayable claim). const AUTO_AWARD_POLL_INTERVAL: Duration = Duration::from_secs(5); +/// How often the delivery watcher re-checks awarded-unsettled jobs with no event to wake it. The +/// subscription is the fast path; this backstop is what makes a dropped or missed result event a +/// latency cost rather than a stranded payment. +const DELIVERY_RECHECK_INTERVAL: Duration = Duration::from_secs(60); /// Lock file leaf under the home. pub const LOCK_FILE: &str = "buyer.lock"; @@ -232,6 +236,9 @@ pub async fn run(home: MobeeHome) -> Result<(), BuyerError> { } Err(error) => eprintln!("buyer: could not list pending auto-awards to re-arm: {error}"), } + // Start watching for delivered results. Its own first action is a sweep of awarded-unsettled + // jobs, which is what collects a delivery that landed while this daemon was down. + spawn_delivery_watcher(context.clone()); let listener = bind_socket(&socket_path)?; accept_loop(listener, context).await } @@ -581,36 +588,77 @@ struct CollectParams { /// verify integrity, budget-append + wallet melt, materialize) and, ONLY after it succeeds, flip /// the reservation `reserved → spent` via [`lifecycle::settle_after_pay`]. The flip is never /// reached on a pay refusal, so a failed pay never over-states `available`. -async fn collect(context: &BuyerContext, id: Value, params: Value) -> Response { - let params: CollectParams = match serde_json::from_value(params) { - Ok(params) => params, - Err(error) => return Response::err(id, CODE_METHOD_NOT_FOUND, format!("collect params: {error}")), - }; +/// Failure of [`settle_job`]. +#[derive(Debug)] +enum SettleJobError { + /// The budget gate could not be loaded — nothing was attempted. + Gate(String), + /// The sealed pay path refused (or failed). No reservation flip was reached. + Pay(collect::CollectError), + /// The pay landed but the `reserved → spent` flip did not; reconcile converges it on next start. + Store(StoreError), +} + +impl std::fmt::Display for SettleJobError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Gate(message) => write!(formatter, "{message}"), + Self::Pay(error) => write!(formatter, "{error}"), + Self::Store(error) => write!(formatter, "{error}"), + } + } +} +/// The daemon's ONE path to the spend gate: money lock → budget gate → pay-then-flip. +/// +/// Both the `collect` RPC and the delivery watcher call this and nothing else. That is deliberate: +/// it makes "can the watcher reach around a money gate?" answerable by construction instead of by +/// audit. Every gate lives inside the sealed [`collect::collect_async`] this composes — accept-time +/// job-hash recomputation, offer-authoritative amount, seller co-signature, creq verification, +/// single-settlement, the pre-spend commit tip-match, the budget ceiling, and pays-exactly-once — +/// and neither caller can compose a different route to them. +/// +/// The watcher therefore adds NO money authority. It commits no new money; it converts a +/// reservation the award already created. +async fn settle_job( + context: &BuyerContext, + job_id: &str, + out: Option, +) -> Result { // Serialize with award + other collects: at most one wallet-melting op in flight daemon-wide. let _guard = context.money_lock.lock().await; - let mut gate = match BudgetGate::from_home(&context.home) { - Ok(gate) => gate, - Err(error) => return Response::err(id, CODE_INTERNAL, error.to_string()), - }; + let mut gate = + BudgetGate::from_home(&context.home).map_err(|error| SettleJobError::Gate(error.to_string()))?; let request = CollectRequest { - job_id: params.job_id.clone(), - out: params.out, + job_id: job_id.to_owned(), + out, }; // Pay FIRST (append + melt), flip AFTER — the #123/#126 ordering, via the tested seam. - let settled = lifecycle::settle_after_pay( + lifecycle::settle_after_pay( &context.store, - ¶ms.job_id, + job_id, now_unix(), || collect::collect_async(&context.home, &mut gate, request), |outcome| outcome.pay.amount_sats, ) - .await; + .await + .map(|(outcome, _converted)| outcome) + .map_err(|error| match error { + SettleError::Pay(error) => SettleJobError::Pay(error), + SettleError::Store(error) => SettleJobError::Store(error), + }) +} - match settled { - Ok((outcome, _converted)) => Response::ok( +async fn collect(context: &BuyerContext, id: Value, params: Value) -> Response { + let params: CollectParams = match serde_json::from_value(params) { + Ok(params) => params, + Err(error) => return Response::err(id, CODE_METHOD_NOT_FOUND, format!("collect params: {error}")), + }; + + match settle_job(context, ¶ms.job_id, params.out).await { + Ok(outcome) => Response::ok( id, json!({ "pay": { @@ -625,8 +673,8 @@ async fn collect(context: &BuyerContext, id: Value, params: Value) -> Response { "files": outcome.files, }), ), - Err(SettleError::Pay(error)) => Response::err(id, CODE_REFUSED, error.to_string()), - Err(SettleError::Store(error)) => Response::err(id, CODE_INTERNAL, error.to_string()), + Err(error @ SettleJobError::Pay(_)) => Response::err(id, CODE_REFUSED, error.to_string()), + Err(error) => Response::err(id, CODE_INTERNAL, error.to_string()), } } @@ -786,6 +834,157 @@ async fn finalize_auto_award( } } +/// Spawn the delivery watcher — the "seller paid in seconds" half of the trade loop. Subscribes +/// BEFORE the task starts so no result published during scheduling is missed, then hands the +/// receiver to the loop. +/// +/// `subscribe_events` is a synchronous, non-blocking channel handout and `tokio::spawn` returns +/// immediately, so this adds nothing measurable to the daemon's readiness path — which matters, +/// because the socket bind that follows is on a 10s deadline. +fn spawn_delivery_watcher(context: Arc) { + let events = context.relay.subscribe_events(); + tokio::spawn(async move { drive_delivery_watch(&context, events).await }); +} + +/// Watch for delivered results and settle them automatically. +/// +/// Structure mirrors the P2 rule this daemon already lives by — **the subscription is the wake, the +/// durable state is the truth**. An arriving result event is never evidence of anything: it only +/// says "look again", and optionally narrows WHICH job to look at. What may be paid is read from +/// the award/reservation ledger and decided entirely by the sealed pay path, so a forged, replayed, +/// or malformed result event can at worst cost a wasted fetch. +/// +/// The loop starts with a sweep because a delivery that landed while the daemon was DOWN has no +/// event to replay — durable state is the only thing that can find it. +async fn drive_delivery_watch( + context: &Arc, + events: tokio::sync::broadcast::Receiver>, +) { + watch_loop(events, DELIVERY_RECHECK_INTERVAL, |wake| async move { + match wake { + // A result arrived: sweep, narrowed to the jobs that event references. + WatchWake::Delivered(event) => settle_awarded(context, Some(&event)).await, + WatchWake::Sweep | WatchWake::SubscriptionLost => settle_awarded(context, None).await, + } + }) + .await; +} + +/// What woke the watch loop. Modelled explicitly so the loop's control flow is testable without a +/// relay, a wall clock, or a paused clock — the sweep action is injected, and this says why. +#[derive(Debug, Clone, PartialEq, Eq)] +enum WatchWake { + /// A delivered result arrived — sweep, narrowed to the jobs it references. + Delivered(Arc), + /// The backstop fired, or a gap means something may have been missed — sweep everything. + Sweep, + /// The subscription ended; the loop continues on the backstop alone. + SubscriptionLost, +} + +/// The watch loop's control flow, with the sweep action injected. +/// +/// Two properties here are the whole point, and both are the kind that stay silently dead while +/// everything looks fine — so both are toothed directly: +/// +/// 1. The backstop is a fixed-cadence `interval`, NOT a sleep re-armed inside the `select!`. A +/// per-iteration sleep is pushed back by every arriving event, and events the loop does not act +/// on (the `continue` below) would reset it WITHOUT sweeping — so a steady claim/feedback stream +/// would starve the sweep indefinitely, defeating it in exactly the case it exists for: a result +/// event that was missed. `Delay` keeps a slow settle pass from bunching the next ticks. +/// 2. A closed subscription DEGRADES, it does not stop the loop. Settling does not depend on the +/// relay handle at all (the collect path opens its own client), so returning here would strand +/// every future delivery. It drops to timer-only and says so. +async fn watch_loop( + events: tokio::sync::broadcast::Receiver>, + interval: Duration, + mut sweep: S, +) where + S: FnMut(WatchWake) -> Fut, + Fut: std::future::Future, +{ + // A delivery that landed while the daemon was DOWN has no event to replay, so the durable set + // is the only thing that can find it. That is why the loop sweeps before it ever waits. + sweep(WatchWake::Sweep).await; + + let mut backstop = tokio::time::interval(interval); + backstop.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + backstop.tick().await; // The first tick resolves immediately; the boot sweep above WAS it. + + // `None` once the subscription is gone — the loop then runs on the backstop alone. + let mut events = Some(events); + + loop { + let wake = match events.as_mut() { + Some(stream) => tokio::select! { + received = stream.recv() => match received { + Ok(event) => { + if event.kind != nostr_sdk::Kind::Custom(crate::kinds::JOB_RESULT_KIND) { + continue; + } + WatchWake::Delivered(event) + } + // Lagged is NOT an error — the buffer overflowed and a result may have been + // missed. Treating a busy relay as a failure would strand a payment; the right + // response is to widen to a full sweep. + Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => WatchWake::Sweep, + Err(tokio::sync::broadcast::error::RecvError::Closed) => { + eprintln!( + "buyer: delivery watcher lost the relay subscription; falling back to \ + periodic sweeps every {}s (deliveries still settle, just later)", + interval.as_secs() + ); + events = None; + WatchWake::SubscriptionLost + } + }, + _ = backstop.tick() => WatchWake::Sweep, + }, + None => { + backstop.tick().await; + WatchWake::Sweep + } + }; + sweep(wake).await; + } +} + +/// Settle awarded-but-unsettled jobs through the daemon's single spend path. +/// +/// `wake` narrows the sweep to the jobs a just-arrived result references (the fast path); `None` +/// sweeps the whole set (boot, the backstop tick, and after a `Lagged` gap). +async fn settle_awarded(context: &Arc, wake: Option<&nostr_sdk::Event>) { + let jobs = match context.store.awarded_unsettled_job_ids() { + Ok(jobs) => jobs, + Err(error) => { + eprintln!("buyer: delivery watcher could not read awarded jobs ({error}); will retry"); + return; + } + }; + for job_id in jobs { + if let Some(event) = wake { + if !job_lifecycle::event_references_job(event, &job_id) { + continue; + } + } + match settle_job(context, &job_id, None).await { + Ok(outcome) => eprintln!( + "buyer: delivery watcher settled {job_id} — paid {} sat for commit {} ({} file(s))", + outcome.pay.amount_sats, + outcome.commit_oid, + outcome.files.len() + ), + // Nothing delivered yet is the ordinary state of an awarded job, not a failure: the job + // stays in the set and the next event or tick retries. Every OTHER outcome is a real + // refusal — a gate said no — and is named so an operator sees which job stopped and why. + Err(SettleJobError::Pay(collect::CollectError::Lifecycle( + job_lifecycle::JobLifecycleError::NotFound(_), + ))) => {} + Err(error) => eprintln!("buyer: delivery watcher could not settle {job_id}: {error}"), + } + } +} + /// Reconcile every still-`Reserved` job against relay + payment-journal truth: a job the relay no /// longer shows payable (and that has left no funds) is released; a job whose payment journal shows /// a `Closed` attempt is converted to `spent`; an ambiguous (Sent-not-Closed) payment is KEPT (the @@ -1004,6 +1203,261 @@ mod tests { std::env::temp_dir().join(format!("mobee-buyer-mod-{label}-{}-{id}", std::process::id())) } + /// A non-actionable event for the starvation tooth: any kind the watcher does not act on. Job + /// FEEDBACK is the realistic one — it shares the buyer-keyed subscription with results, so a + /// chatty job produces exactly this traffic. + fn non_actionable_event() -> Arc { + use nostr_sdk::prelude::{EventBuilder, Keys}; + Arc::new( + EventBuilder::new(nostr_sdk::Kind::Custom(crate::kinds::JOB_FEEDBACK_KIND), "") + .sign_with_keys(&Keys::generate()) + .expect("sign"), + ) + } + + // ★ LIVENESS TOOTH (a): a steady stream of events the watcher does NOT act on must not starve + // the backstop sweep. + // + // This is the failure that keeps costing us: a liveness mechanism silently dead while + // everything looks fine. A backstop re-armed per loop iteration is pushed back by every + // arriving event — and non-actionable events reset it WITHOUT sweeping — so under steady + // traffic it never fires, and the one case it exists for (a result event we missed) is exactly + // the case it stops covering. A correctness tooth cannot see this: the loop still returns + // correct results for every event it DOES get. + // + // Deterministic by construction — no relay, no paused clock, no wall-clock assertion. The sweep + // action is injected and counted, and the event rate is an order of magnitude faster than the + // backstop, so the two implementations separate hugely: fixed-cadence interval ⇒ many sweeps; + // re-armed sleep ⇒ exactly the one boot sweep, forever. + // + // Red-on-revert: replace the `interval` with `tokio::time::sleep(interval)` inside the select + // and this fails on `sweeps >= 4`, observing 1. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn steady_non_actionable_traffic_does_not_starve_the_backstop_sweep() { + use std::sync::atomic::AtomicUsize; + + let (sender, receiver) = tokio::sync::broadcast::channel(64); + let sweeps = Arc::new(AtomicUsize::new(0)); + + let counter = sweeps.clone(); + let loop_task = tokio::spawn(async move { + watch_loop(receiver, Duration::from_millis(10), move |_wake| { + let counter = counter.clone(); + async move { + counter.fetch_add(1, Ordering::SeqCst); + } + }) + .await; + }); + + // Flood non-actionable events an order of magnitude faster than the backstop for long + // enough that a healthy loop must tick many times. + let flood = tokio::spawn(async move { + for _ in 0..300 { + // A receiver-less send is fine; the loop is the only receiver and may be busy. + let _ = sender.send(non_actionable_event()); + tokio::time::sleep(Duration::from_millis(1)).await; + } + // Hold the sender until the end so the subscription never closes during the flood. + sender + }); + let _sender = flood.await.expect("flood"); + + let observed = sweeps.load(Ordering::SeqCst); + loop_task.abort(); + assert!( + observed >= 4, + "the backstop must keep sweeping under steady non-actionable traffic; saw {observed} \ + sweep(s) across ~300ms at a 10ms cadence (a per-iteration sleep would show exactly 1)" + ); + } + + // ★ LIVENESS TOOTH (b): losing the subscription DEGRADES the watcher to timer-only sweeps — it + // must not stop it. Settling never depended on the relay handle (the collect path opens its own + // client), so returning on `Closed` would strand every delivery from that moment on, silently. + // + // Asserts both halves: the loop observes `SubscriptionLost` exactly once (the arm that emits the + // unconditional operator line — it cannot fire without executing that line), and sweeps continue + // afterwards on the backstop alone. + // + // Red-on-revert: `return` on `RecvError::Closed` and this fails — no post-close sweeps. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_lost_subscription_degrades_to_timer_only_instead_of_stopping() { + use std::sync::atomic::AtomicUsize; + + let (sender, receiver) = tokio::sync::broadcast::channel(8); + let lost = Arc::new(AtomicUsize::new(0)); + let after_loss = Arc::new(AtomicUsize::new(0)); + + let (lost_counter, after_counter) = (lost.clone(), after_loss.clone()); + let loop_task = tokio::spawn(async move { + watch_loop(receiver, Duration::from_millis(10), move |wake| { + let (lost_counter, after_counter) = (lost_counter.clone(), after_counter.clone()); + async move { + if wake == WatchWake::SubscriptionLost { + lost_counter.fetch_add(1, Ordering::SeqCst); + } else if lost_counter.load(Ordering::SeqCst) > 0 { + after_counter.fetch_add(1, Ordering::SeqCst); + } + } + }) + .await; + }); + + // Drop the sender: the subscription closes. + drop(sender); + // Let the backstop tick several times past the close. + tokio::time::sleep(Duration::from_millis(120)).await; + + let (losses, sweeps_after) = (lost.load(Ordering::SeqCst), after_loss.load(Ordering::SeqCst)); + loop_task.abort(); + assert_eq!(losses, 1, "the subscription loss must be observed exactly once, not spun on"); + assert!( + sweeps_after >= 2, + "the watcher must keep sweeping on the backstop after losing the subscription; saw \ + {sweeps_after} sweep(s) in ~120ms at a 10ms cadence" + ); + } + + // ★ THE MONEY TOOTH for the delivery watcher. The watcher settles with no human in the loop, so + // the question that matters is whether its entry to the spend gate is as gated as the RPC's. It + // is, by construction — both call `settle_job` and nothing else — and this proves it end to end: + // an awarded job whose delivered result carries a FORGED seller co-signature must cost NOTHING. + // + // Non-vacuity anchor: the watcher drives the full accept-then-pay path, and accept writes the + // pay-bind BEFORE the pre-pay co-signature gate refuses. So the bind file EXISTING is positive + // evidence the watcher really ran this job — without it, every "nothing happened" assertion + // below would pass just as well if the watcher had never woken at all. + // + // The counterparty events are published from a SEPARATE client than the daemon's own session: + // one nostr-sdk client cannot observe the events it published itself, and that trap bites + // fixtures exactly as it bites production. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn delivery_watcher_refuses_a_forged_delivery_and_spends_nothing() { + use nostr_relay_builder::prelude::{LocalRelay, RelayBuilder}; + use nostr_sdk::prelude::{Client, Keys}; + use nostr_sdk::secp256k1::Message; + + let root = temp_home("watch-forged"); + let _ = std::fs::remove_dir_all(&root); + let mut home = bootstrap_home(&root).expect("bootstrap home"); + + let relay = LocalRelay::new(RelayBuilder::default()); + relay.run().await.expect("relay run"); + let relay_url = relay.url().await.to_string(); + home.config.relay_url = relay_url.clone(); + + let buyer = Keys::parse(&crate::home::read_secret_key_hex(&home).expect("secret")) + .expect("buyer keys"); + let buyer_hex = buyer.public_key().to_hex(); + let seller = Keys::generate(); + let seller_hex = seller.public_key().to_hex(); + let attacker = Keys::generate(); + + let amount = 2u64; + let task = "do a task"; + let output = "text"; + let deadline = now_unix() as u64 + 3_600; + + // Publish offer + claim + forged-cosig result from a client that is NOT the daemon's. + let net = Client::new(Keys::generate()); + net.add_relay(&relay_url).await.expect("add relay"); + net.connect().await; + net.wait_for_connection(Duration::from_secs(5)).await; + let publish = |keys: &Keys, draft: &crate::gateway::EventDraft| { + let builder = crate::gateway::nostr::event_builder(draft).expect("event builder"); + let event = builder.sign_with_keys(keys).expect("sign"); + let net = net.clone(); + let id = event.id; + async move { + net.send_event(&event).await.expect("publish"); + id + } + }; + + let offer_draft = + crate::gateway::OfferDraft::new(task, output, amount, deadline, &seller_hex) + .to_event_draft(); + let job_id = publish(&buyer, &offer_draft).await.to_hex(); + + let creq = crate::gateway::creq::build_seller_creq( + &job_id, + amount, + "sat", + &[crate::home::DEFAULT_MINT_URL.to_string()], + &seller_hex, + ) + .expect("creq"); + let claim_draft = crate::gateway::claim_draft(&job_id, &buyer_hex, &seller_hex, &creq); + let _ = publish(&seller, &claim_draft).await; + + let job_hash = job_lifecycle::job_hash_for_offer(&job_id, task, amount); + let forged = attacker + .sign_schnorr(&Message::from_digest([0x11u8; 32])) + .to_string(); + let git = crate::gateway::GitResultTags { + repo: "https://example.invalid/repo.git", + branch: "main", + commit_sha: &"a".repeat(40), + }; + let result_draft = crate::gateway::result_draft( + &job_id, &buyer_hex, output, amount, &job_hash, &forged, "", Some(git), &[], + ); + let _ = publish(&seller, &result_draft).await; + + let (_lock, context, _socket_path) = bootstrap(home).await.expect("buyer bootstrap"); + + // Seed exactly what the award path writes: the reservation, then the published-award row. + context + .store + .reserve(&job_id, amount, 1_000, u64::MAX, 0, now_unix()) + .expect("reserve"); + context + .store + .record_award(&job_id, &"c".repeat(64), &"e".repeat(64), &seller_hex, amount, now_unix()) + .expect("record award"); + assert_eq!( + context.store.awarded_unsettled_job_ids().expect("awarded"), + vec![job_id.clone()], + "the seeded award must be the watcher's work set" + ); + + // Drive the watcher's settle pass directly — the same call its event and tick paths make. + settle_awarded(&context, None).await; + + // Non-vacuity: the watcher really drove accept-then-pay for this job. + assert!( + context.home.root.join("jobs").join(format!("{job_id}.json")).exists(), + "the watcher must have driven the accept path (no bind ⇒ it never ran this job, and \ + every assertion below would be vacuous)" + ); + // And the money gate refused: nothing spent, no payment journal, no files, and the + // reservation still held (a refused settle must never drop it and over-state `available`). + assert_eq!( + BudgetGate::from_home(&context.home).expect("gate").spent(), + 0, + "a refused delivery must burn zero spend" + ); + assert!( + !context.home.root.join("payment-journal").exists(), + "no payment journal may exist on a pre-pay refusal" + ); + assert!( + !context.home.root.join("results").join(&job_id).exists(), + "a refused delivery must materialize no files" + ); + assert!( + matches!( + context.store.reservation(&job_id).expect("reservation"), + Some((reservations::ReservationState::Reserved, _)) + ), + "a refused settle must leave the reservation reserved" + ); + + let _ = std::fs::remove_dir_all(&root); + drop(relay); + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn status_round_trips_over_the_socket() { let root = temp_home("status"); diff --git a/crates/mobee-core/src/buyer/store.rs b/crates/mobee-core/src/buyer/store.rs index c76e37d4..7cf3489c 100644 --- a/crates/mobee-core/src/buyer/store.rs +++ b/crates/mobee-core/src/buyer/store.rs @@ -29,7 +29,10 @@ use super::reservations::{ /// - v3 — the pending-award intents: the `pending_awards` table (#126/#127 background auto-award). /// One row per posted job the daemon still owes an award; re-armed on restart. Same additive /// forward-only upgrade. -pub const SCHEMA_VERSION: i64 = 3; +/// - v4 — the published-award record: the `awards` table. `pending_awards` tracks the INTENT and +/// its state; this records the award the buyer actually published, keyed by job, carrying the +/// 3405 event id. Same additive forward-only upgrade. +pub const SCHEMA_VERSION: i64 = 4; /// A cloneable handle to the daemon-owned SQLite state. #[derive(Clone)] @@ -117,6 +120,20 @@ impl BuyerStore { reason TEXT, created_at_unix INTEGER NOT NULL, updated_at_unix INTEGER NOT NULL + ); + -- v4: the awards the buyer actually PUBLISHED, one row per job, written at publish + -- time from the single reserve-then-award seam. `pending_awards.state='awarded'` says + -- the intent completed; this says WHICH 3405 carries it. Held locally so the delivery + -- watcher and the boot re-check can enumerate awarded-but-unsettled work from durable + -- state alone — no relay round-trip on the boot path, which is already the tightest + -- deadline the daemon has. + CREATE TABLE IF NOT EXISTS awards ( + job_id TEXT PRIMARY KEY, + claim_id TEXT NOT NULL, + award_event_id TEXT NOT NULL, + seller_pubkey TEXT NOT NULL, + amount_sats INTEGER NOT NULL CHECK (amount_sats >= 0), + awarded_at_unix INTEGER NOT NULL );", )?; // Forward-only, monotone schema-version bump. A fresh DB is stamped at SCHEMA_VERSION; a @@ -517,6 +534,75 @@ impl BuyerStore { .optional()?) } + /// Record the award the buyer PUBLISHED for `job_id`. Called from the single reserve-then-award + /// seam ([`super::lifecycle::award_with_reservation`]) so both the manual and auto award paths + /// are covered by construction — recording at the two call sites instead would let one drift. + /// + /// Idempotent by job: a re-award that republishes the same job overwrites the row rather than + /// failing, matching the re-arm path's own idempotence. + pub fn record_award( + &self, + job_id: &str, + claim_id: &str, + award_event_id: &str, + seller_pubkey: &str, + amount_sats: u64, + now_unix: i64, + ) -> Result<(), StoreError> { + let conn = self.lock()?; + conn.execute( + "INSERT INTO awards + (job_id, claim_id, award_event_id, seller_pubkey, amount_sats, awarded_at_unix) + VALUES (?1, ?2, ?3, ?4, ?5, ?6) + ON CONFLICT(job_id) DO UPDATE SET + claim_id = excluded.claim_id, + award_event_id = excluded.award_event_id, + seller_pubkey = excluded.seller_pubkey, + amount_sats = excluded.amount_sats, + awarded_at_unix = excluded.awarded_at_unix", + params![job_id, claim_id, award_event_id, seller_pubkey, amount_sats as i64, now_unix], + )?; + Ok(()) + } + + /// The published award for `job_id`, if the buyer awarded it. + pub fn award_record(&self, job_id: &str) -> Result, StoreError> { + let conn = self.lock()?; + Ok(conn + .query_row( + "SELECT job_id, claim_id, award_event_id, seller_pubkey, amount_sats, + awarded_at_unix + FROM awards WHERE job_id = ?1", + [job_id], + row_to_award, + ) + .optional()?) + } + + /// Jobs the buyer AWARDED that have not yet settled — the delivery watcher's work set and the + /// boot re-check's sweep set. + /// + /// "Not yet settled" is read from the reservation ledger rather than tracked as a new state: + /// `reserved` is exactly "money committed, not yet paid out", because collect flips it to + /// `spent` only after the pay lands ([`super::lifecycle::settle_after_pay`]) and a release + /// moves it to `released`. Deriving it means there is no second lifecycle that can disagree + /// with the ledger the budget already trusts. + pub fn awarded_unsettled_job_ids(&self) -> Result, StoreError> { + let conn = self.lock()?; + let mut stmt = conn.prepare( + "SELECT a.job_id FROM awards a + JOIN reservations r ON r.job_id = a.job_id + WHERE r.state = 'reserved' + ORDER BY a.awarded_at_unix", + )?; + let rows = stmt.query_map([], |row| row.get::<_, String>(0))?; + let mut jobs = Vec::new(); + for row in rows { + jobs.push(row?); + } + Ok(jobs) + } + /// Every `parked` auto-award intent as `(job_id, reason)` — surfaced in `status` so a buyer sees /// jobs whose award could not be placed rather than silently losing them. pub fn parked_awards(&self) -> Result, StoreError> { @@ -546,6 +632,30 @@ pub struct PendingAward { pub model: Option, } +/// An award the buyer PUBLISHED: the job it commits to, the claim it picked, and the 3405 that +/// carries it. Distinct from [`PendingAward`], which is the intent that preceded it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct AwardRecord { + pub job_id: String, + pub claim_id: String, + pub award_event_id: String, + pub seller_pubkey: String, + pub amount_sats: u64, + pub awarded_at_unix: i64, +} + +/// Map an `awards` row (in the column order both queries select) to an [`AwardRecord`]. +fn row_to_award(row: &rusqlite::Row<'_>) -> rusqlite::Result { + Ok(AwardRecord { + job_id: row.get::<_, String>(0)?, + claim_id: row.get::<_, String>(1)?, + award_event_id: row.get::<_, String>(2)?, + seller_pubkey: row.get::<_, String>(3)?, + amount_sats: row.get::<_, i64>(4)?.max(0) as u64, + awarded_at_unix: row.get::<_, i64>(5)?, + }) +} + /// Read a job's `(state, amount)` from a live transaction. `None` when no row exists. fn read_state( tx: &rusqlite::Transaction<'_>, @@ -655,8 +765,8 @@ mod tests { } // A v1 database (buyer_meta + jobs only, schema_version = 1) is upgraded forward on open: the - // reservations AND pending_awards tables appear and schema_version moves to the current version. - // No data migration, idempotent. + // reservations, pending_awards AND awards tables appear and schema_version moves to the current + // version. No data migration, idempotent. #[test] fn open_migrates_v1_db_forward() { let path = temp_db("migrate"); @@ -682,12 +792,65 @@ mod tests { .put_pending_award(&"a".repeat(64), 10, None, None, 1) .expect("pending award on upgraded db"); assert_eq!(store.list_pending_awards().expect("list").len(), 1); + // The v4 awards table is now usable too — and the job reserved above is immediately + // enumerable as awarded-unsettled, which is the delivery watcher's entire work set. + store + .record_award(&"a".repeat(64), &"c".repeat(64), &"e".repeat(64), &"f".repeat(64), 10, 1) + .expect("record award on upgraded db"); + assert_eq!( + store.awarded_unsettled_job_ids().expect("awarded"), + vec!["a".repeat(64)] + ); // Re-open is idempotent (still current version). let store2 = BuyerStore::open(&path).expect("reopen"); assert_eq!(store2.health().expect("health").schema_version, SCHEMA_VERSION); let _ = std::fs::remove_file(&path); } + // The watcher's work set is DERIVED from the reservation ledger, never tracked as a second + // state: an awarded job counts as unsettled exactly while its reservation is `reserved`, and + // drops out the moment collect flips it to `spent` or a release moves it to `released`. Guards + // the property that makes the watcher safe to re-run — a settled job can never be re-offered + // to the pay path by this query. + #[test] + fn awarded_unsettled_follows_the_reservation_ledger() { + let (store, path) = fresh_store("awarded-unsettled"); + let paid = "a".repeat(64); + let dropped = "b".repeat(64); + let waiting = "c".repeat(64); + + for (job, amount) in [(&paid, 10u64), (&dropped, 20), (&waiting, 30)] { + store.reserve(job, amount, 1_000, NO_BUDGET, 0, 1).expect("reserve"); + store + .record_award(job, &"c".repeat(64), &"e".repeat(64), &"f".repeat(64), amount, 1) + .expect("record award"); + } + // All three are awarded and still reserved ⇒ all three are the watcher's work. + assert_eq!( + store.awarded_unsettled_job_ids().expect("awarded").len(), + 3 + ); + + store.convert_to_spent(&paid, 10, 2).expect("settle"); + store.release(&dropped, 2).expect("release"); + + // Only the job still awaiting delivery remains payable-by-the-watcher. + assert_eq!( + store.awarded_unsettled_job_ids().expect("awarded"), + vec![waiting.clone()] + ); + + // A reservation with no award row is NOT the watcher's business — it never awarded it. + let foreign = "d".repeat(64); + store.reserve(&foreign, 5, 1_000, NO_BUDGET, 0, 3).expect("reserve foreign"); + assert_eq!( + store.awarded_unsettled_job_ids().expect("awarded"), + vec![waiting], + "a reserved job the buyer never awarded must not enter the watcher's work set" + ); + let _ = std::fs::remove_file(&path); + } + // Pending auto-award intent lifecycle: put ⇒ listed as pending; mark_parked ⇒ off the pending // list, state=parked with the surfaced reason (invariant B: park, never silent drop); // mark_awarded ⇒ awarded; re-put ⇒ back to pending with the reason cleared. diff --git a/crates/mobee-core/src/job_lifecycle.rs b/crates/mobee-core/src/job_lifecycle.rs index 2a7a1ca7..47d76050 100644 --- a/crates/mobee-core/src/job_lifecycle.rs +++ b/crates/mobee-core/src/job_lifecycle.rs @@ -770,7 +770,7 @@ async fn await_job_event( /// True when `event` carries an `e` tag naming this job's offer — the tag every claim, result and /// feedback roots itself on. -fn event_references_job(event: &nostr_sdk::Event, job_id: &str) -> bool { +pub(crate) fn event_references_job(event: &nostr_sdk::Event, job_id: &str) -> bool { event.tags.iter().any(|tag| { let parts = tag.as_slice(); parts.first().map(String::as_str) == Some("e") && parts.get(1).map(String::as_str) == Some(job_id) From b7c27407b63f7225266a96abb6ace2456fbf8cba Mon Sep 17 00:00:00 2001 From: "worker:mobee-buyer-lazystart" Date: Mon, 27 Jul 2026 01:08:01 -0700 Subject: [PATCH 5/7] buyer: free a reservation the seller abandoned, and say so on every pass MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A seller that simply stops leaves the buyer's funds committed to a job that will never deliver. The reconcile that resolves this only ran at startup, so the budget stayed pinned until someone restarted the daemon. It now also runs on a slow timer, which frees the reservation within the hour instead. Nothing new decides what may be released. The existing classification already rules it — a paid job converts, an ambiguous payment is never auto-released, and a job is dead only when no payment left funds and the relay no longer shows it payable. That last condition is already time-based rather than dependent on the seller announcing failure: a claim stops being live at the offer deadline whether or not any feedback is ever published. A fresh timeout would have had to re-derive all of that, including the ambiguous-payment care that matters most. Gathering runs without the money lock, since it is per-job relay I/O and would otherwise block trades for seconds; only the apply takes it. That ordering is what makes it impossible to release a reservation in the window after a payment melted but before its reserved-to-spent flip, which would briefly free funds that had already left the wallet. The apply is state-guarded inside its transaction besides, so a disposition that went stale while gathering is a no-op. Every pass is now reported, including the pass that changed nothing. A release moves what the buyer can spend, and a path that prints only when it acts is indistinguishable in a log from one that has stopped running. Released jobs are named, and the examined count is in the line because it is also the number of per-job relay fetches the pass made, so that cost is visible while it grows. Co-Authored-By: Claude Opus 5 (1M context) --- crates/mobee-core/src/buyer/mod.rs | 196 +++++++++++++++++++++++++++-- 1 file changed, 189 insertions(+), 7 deletions(-) diff --git a/crates/mobee-core/src/buyer/mod.rs b/crates/mobee-core/src/buyer/mod.rs index c977b1c6..e63248e3 100644 --- a/crates/mobee-core/src/buyer/mod.rs +++ b/crates/mobee-core/src/buyer/mod.rs @@ -68,6 +68,13 @@ const AUTO_AWARD_POLL_INTERVAL: Duration = Duration::from_secs(5); /// subscription is the fast path; this backstop is what makes a dropped or missed result event a /// latency cost rather than a stranded payment. const DELIVERY_RECHECK_INTERVAL: Duration = Duration::from_secs(60); +/// How often the reservation reconcile re-runs while the daemon serves. This is what frees a +/// reservation stranded by a seller that simply stopped, without waiting for a restart. +/// +/// Deliberately SLOW: a pass makes one relay fetch per still-reserved job, which is the unbounded +/// per-job fetch pattern tracked in #180. Until that moves onto the persistent relay session, the +/// cadence is the thing keeping the load down — raise it only together with that fix. +const RECONCILE_INTERVAL: Duration = Duration::from_secs(600); /// Lock file leaf under the home. pub const LOCK_FILE: &str = "buyer.lock"; @@ -219,12 +226,7 @@ pub async fn run(home: MobeeHome) -> Result<(), BuyerError> { // daemon starts from a converged ledger. A failure is logged, not fatal — an unreachable relay // must not keep the daemon from coming up (the stale reservation is conservative until the next // reconcile). - match reconcile_on_start(&context).await { - Ok(report) => *context.last_reconcile.lock().await = Some(report), - Err(error) => eprintln!( - "buyer: reconcile-on-start did not complete ({error}); serving with the ledger as-is" - ), - } + run_reconcile_pass(&context).await; // Re-arm pending auto-awards left by a prior run: a job posted before a crash still gets its // award with zero manual commands. Each task re-checks the relay for an existing award first // (invariant A), so re-arming never double-awards. @@ -239,6 +241,9 @@ pub async fn run(home: MobeeHome) -> Result<(), BuyerError> { // Start watching for delivered results. Its own first action is a sweep of awarded-unsettled // jobs, which is what collects a delivery that landed while this daemon was down. spawn_delivery_watcher(context.clone()); + // Keep reconciling while we serve, so a reservation stranded by a seller that went away is + // freed within the hour rather than at the next restart. + spawn_reconcile_loop(context.clone()); let listener = bind_socket(&socket_path)?; accept_loop(listener, context).await } @@ -990,7 +995,20 @@ async fn settle_awarded(context: &Arc, wake: Option<&nostr_sdk::Ev /// a `Closed` attempt is converted to `spent`; an ambiguous (Sent-not-Closed) payment is KEPT (the /// phase-3 saga owns it). Pure classification is [`lifecycle::classify_disposition`]; this gathers /// its inputs and applies the batch through [`BuyerStore::reconcile`]. -async fn reconcile_on_start(context: &BuyerContext) -> Result { +/// +/// This runs at boot AND on a slow timer, which is what releases a reservation stranded by a seller +/// that simply stopped — no feedback event required, because a claim stops being live at the offer +/// deadline whether or not the seller ever says so. +/// +/// Gather is done WITHOUT the money lock (it is per-job relay I/O and would block the trade RPCs for +/// seconds); only the apply takes it. That ordering matters: `settle_job` holds the money lock +/// across pay-then-flip, so taking it here makes it impossible to release a reservation in the +/// window after a payment's melt but before its `reserved → spent` flip — which would transiently +/// free funds that had in fact already left the wallet. The apply itself is additionally +/// state-guarded inside its transaction ([`BuyerStore::reconcile`] re-reads each row and acts only +/// on the state it expects), so a disposition that went stale during gather is a no-op rather than a +/// wrong write. +async fn reconcile_reservations(context: &BuyerContext) -> Result { let reserved = context .store .reserved_job_ids() @@ -1027,12 +1045,90 @@ async fn reconcile_on_start(context: &BuyerContext) -> Result) { + match reconcile_reservations(context).await { + Ok(report) => { + report_reconcile(&report); + *context.last_reconcile.lock().await = Some(report); + } + // An unreachable relay must never be fatal: every job it could not verify was treated as + // still-payable, so the ledger is conservative until the next pass. + Err(error) => { + eprintln!("buyer: reconcile pass did not complete ({error}); serving with the ledger as-is") + } + } +} + +/// Report a reconcile pass UNCONDITIONALLY — including the pass that changed nothing. +/// +/// Releasing a reservation is a money-visible decision: it returns funds to `available`. A path that +/// prints only when it acts is indistinguishable, in a log, from a path that has stopped running — +/// so the quiet pass is exactly the one worth printing. Released jobs are named, because "something +/// was released" is not an answer to "which job, and why did my budget move". +/// +/// The examined count is in the line deliberately: it is also the number of per-job relay fetches +/// this pass made, so #180's amplification is visible while it grows instead of when it bites. +fn report_reconcile(report: &ReconcileReport) { + eprintln!("{}", reconcile_line(report)); +} + +/// Build the reconcile report line. Split from the printing so the wording — including the +/// released-nothing case that exists precisely because it is easy to leave out — is directly +/// testable rather than something a reader has to trust. +fn reconcile_line(report: &ReconcileReport) -> String { + let examined = report.released.len() + report.converted.len() + report.kept.len(); + if report.released.is_empty() { + format!( + "buyer: reconcile examined {examined} reserved job(s) — released nothing, converted {}, kept {}", + report.converted.len(), + report.kept.len() + ) + } else { + format!( + "buyer: reconcile examined {examined} reserved job(s) — RELEASED {} (no longer payable \ + on the relay and no funds left: {}), converted {}, kept {}", + report.released.len(), + report.released.join(", "), + report.converted.len(), + report.kept.len() + ) + } +} + +/// Re-run the reconcile on a slow timer for the daemon's lifetime, with the pass injected. +/// +/// Sequential BY CONSTRUCTION: the pass is awaited inside the loop, and the interval's `Delay` +/// behaviour defers a tick that arrives mid-pass. Two passes can therefore never overlap, so the +/// per-job relay fetches cannot compound into each other — which matters while #180 stands, because +/// overlapping passes would multiply exactly the load the slow cadence is holding down. +async fn reconcile_loop(interval: Duration, mut pass: P) +where + P: FnMut() -> Fut, + Fut: std::future::Future, +{ + let mut ticker = tokio::time::interval(interval); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + ticker.tick().await; // Resolves immediately; the boot pass in `run` WAS it. + loop { + ticker.tick().await; + pass().await; + } +} + +fn spawn_reconcile_loop(context: Arc) { + tokio::spawn(async move { + reconcile_loop(RECONCILE_INTERVAL, || run_reconcile_pass(&context)).await; + }); +} + /// Pure reconcile planning: map each reserved job to a disposition from its folded payment progress /// and whether it is still payable. Kept pure (no relay/disk I/O) so the reserved-job → disposition /// mapping is exhaustively testable; [`reconcile_on_start`] gathers the inputs. A job absent from @@ -1203,6 +1299,92 @@ mod tests { std::env::temp_dir().join(format!("mobee-buyer-mod-{label}-{}-{id}", std::process::id())) } + // ★ THE #179/4b TOOTH: the reconcile reports the pass that changed NOTHING. + // + // A release moves the buyer's `available` — it is a money-visible decision — and the pass that + // releases nothing is the one most likely to be dropped as noise. But a path that prints only + // when it acts reads, in a log, exactly like a path that has stopped running, which is how a + // dead release path stays invisible until a reservation strands. So the empty case is asserted + // first here, deliberately. + // + // Red-on-revert: make the empty case return an empty string (or skip the log) and this fails. + #[test] + fn the_reconcile_reports_every_pass_including_the_one_that_did_nothing() { + // Nothing reserved at all: still a line, and it still says what it examined. + let quiet = reconcile_line(&ReconcileReport::default()); + assert!(quiet.contains("examined 0 reserved job(s)"), "unexpected: {quiet}"); + assert!(quiet.contains("released nothing"), "the empty pass must say so: {quiet}"); + + // Jobs examined, none released: the count is still reported, so an operator can see the + // pass ran and how many relay fetches it cost (#180's amplification, visible as it grows). + let busy_but_quiet = ReconcileReport { + kept: vec!["a".repeat(64), "b".repeat(64)], + ..ReconcileReport::default() + }; + let line = reconcile_line(&busy_but_quiet); + assert!(line.contains("examined 2 reserved job(s)"), "unexpected: {line}"); + assert!(line.contains("released nothing"), "unexpected: {line}"); + + // A release NAMES the job and the reason — "something was released" is not an answer to + // "which job, and why did my budget move". + let released = ReconcileReport { + released: vec!["c".repeat(64)], + kept: vec!["d".repeat(64)], + ..ReconcileReport::default() + }; + let line = reconcile_line(&released); + assert!(line.contains("examined 2 reserved job(s)"), "unexpected: {line}"); + assert!(line.contains(&"c".repeat(64)), "the released job must be named: {line}"); + assert!(line.contains("no longer payable"), "the reason must be stated: {line}"); + } + + // ★ NO-OVERLAP TOOTH: reconcile passes must never run concurrently. + // + // Each pass makes one relay fetch per still-reserved job — the unbounded per-job pattern in + // #180 — so overlapping passes would multiply precisely the load the slow cadence exists to + // hold down. `Delay` plus awaiting the pass inside the loop is what prevents it, and both are + // one-word changes away from being wrong (`Burst`, or spawning the pass). + // + // Deterministic: the pass is injected and takes far longer than the tick, so a loop that did + // NOT serialise would immediately show overlap. + // + // Red-on-revert: `tokio::spawn(pass())` instead of awaiting it, and max-in-flight exceeds 1. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn reconcile_passes_never_overlap() { + use std::sync::atomic::AtomicUsize; + + let in_flight = Arc::new(AtomicUsize::new(0)); + let peak = Arc::new(AtomicUsize::new(0)); + let passes = Arc::new(AtomicUsize::new(0)); + + let (f, p, n) = (in_flight.clone(), peak.clone(), passes.clone()); + let task = tokio::spawn(async move { + // Pass duration deliberately several times the tick, so any non-serialising loop + // overlaps on the very first ticks. + reconcile_loop(Duration::from_millis(5), move || { + let (f, p, n) = (f.clone(), p.clone(), n.clone()); + async move { + let now = f.fetch_add(1, Ordering::SeqCst) + 1; + p.fetch_max(now, Ordering::SeqCst); + tokio::time::sleep(Duration::from_millis(25)).await; + f.fetch_sub(1, Ordering::SeqCst); + n.fetch_add(1, Ordering::SeqCst); + } + }) + .await; + }); + + tokio::time::sleep(Duration::from_millis(200)).await; + let (observed_peak, completed) = (peak.load(Ordering::SeqCst), passes.load(Ordering::SeqCst)); + task.abort(); + + assert!(completed >= 2, "the loop must keep running passes; saw {completed}"); + assert_eq!( + observed_peak, 1, + "reconcile passes must never overlap — peak concurrent passes was {observed_peak}" + ); + } + /// A non-actionable event for the starvation tooth: any kind the watcher does not act on. Job /// FEEDBACK is the realistic one — it shares the buyer-keyed subscription with results, so a /// chatty job produces exactly this traffic. From a9ec57e1cbea9fd7fbed5b8217122270b8425f1a Mon Sep 17 00:00:00 2001 From: "worker:mobee-buyer-lazystart" Date: Mon, 27 Jul 2026 02:02:32 -0700 Subject: [PATCH 6/7] buyer: rely on the SDK to recover the job subscription, and pin that it does MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The relay session carried a reconnect command that dropped the socket, re-authenticated, and re-asserted the job subscription in that order. Nothing called it. Every reconnect in the field was always the SDK's own, so that code was not a fallback — it was an untested assumption standing where a verified one should be. What actually happens across a relay restart, measured on a NIP-42-enforcing fixture: the buyer's pubkey-scoped subscription IS refused, because the pool re-issues the request before authentication completes and the relay answers `auth-required:`. That prefix is retryable, so the subscription is kept rather than discarded — the SDK authenticates and re-asserts the same subscription id, and delivery resumes. Re-subscribing ourselves on top of that would be a second request racing the SDK's, which is the same hazard as publishing an event a dependency is already resending. So the command is gone and a test pins the behaviour being relied on instead. It publishes the counterparty's delivery to the replacement relay without waiting for our own session to recover, so the event lands while the subscription is still closed. Passing therefore proves recovery replays what was published during the outage, which is the case that matters: a relay restarts, the seller delivers, and the buyer must still see it. A test that waited for recovery before publishing would have proved only its own timing. Co-Authored-By: Claude Opus 5 (1M context) --- crates/mobee-core/src/buyer/relay.rs | 185 +++++++++++++++++++++------ 1 file changed, 143 insertions(+), 42 deletions(-) diff --git a/crates/mobee-core/src/buyer/relay.rs b/crates/mobee-core/src/buyer/relay.rs index 938debfa..427b8fc3 100644 --- a/crates/mobee-core/src/buyer/relay.rs +++ b/crates/mobee-core/src/buyer/relay.rs @@ -86,10 +86,6 @@ enum Command { Probe { reply: oneshot::Sender, }, - /// Drop the socket and bring a fresh authenticated one up. - Reconnect { - reply: oneshot::Sender>, - }, } /// A write the relay accepted. @@ -241,13 +237,6 @@ impl RelayHandle { let (reply, rx) = oneshot::channel(); self.round_trip("probe", Command::Probe { reply }, rx).await } - - /// Rebuild the session: drop the socket and re-authenticate a fresh one. - pub async fn reconnect(&self) -> Result, RelayActorGone> { - let (reply, rx) = oneshot::channel(); - self.round_trip("reconnect", Command::Reconnect { reply }, rx) - .await - } } /// Register the buyer's relay and spawn the actor that owns the session. @@ -322,17 +311,6 @@ pub async fn spawn(keys: Keys, relay_url: &str) -> Result { - let outcome = reconnect_and_authenticate(&command_client, &relay).await; - if let Ok(wait) = &outcome { - report_auth_wait("reconnect", Ok(*wait)); - // Re-assert the job REQ on the NEW session rather than trusting the - // pool to replay it, and only after auth has completed — same ordering - // rule as boot. The stable id makes this idempotent. - subscribe_job_events(&command_client, public_key).await; - } - let _ = reply.send(outcome.map_err(|error| error.to_string())); - } } } }); @@ -543,26 +521,6 @@ async fn probe_relay_serves_our_reqs( .unwrap_or(false) } -/// Drop the live socket and bring a fresh authenticated one up, returning once NIP-42 has completed -/// on the NEW connection. -/// -/// ORDER IS LOAD-BEARING. `Relay::disconnect` emits `RelayNotification::Shutdown` on the relay's own -/// notification channel; a receiver taken BEFORE the disconnect inherits that Shutdown and the auth -/// wait reads it as "relay shutdown before NIP-42 authentication" — on a socket that in fact -/// authenticated fine. A `broadcast::Receiver` only observes sends made after it subscribes, so -/// taking it AFTER the disconnect cannot inherit our own teardown, while still taking it BEFORE -/// `connect` so the one-shot `Authenticated` cannot be missed. Both halves are required. -async fn reconnect_and_authenticate( - client: &Client, - relay: &nostr_sdk::prelude::Relay, -) -> Result { - client.disconnect().await; - let mut relay_notifications = relay.notifications(); - client.connect().await; - client.wait_for_connection(CONNECT_WAIT).await; - relay_auth::wait_for_nip42_auth(&mut relay_notifications, CONNECT_WAIT).await -} - #[cfg(test)] mod tests { use super::*; @@ -648,6 +606,149 @@ mod tests { ); } + // TOOTH — the SECOND DEPENDENCY CONTRACT this buyer rests on: the SDK recovers our `#p`-gated + // subscription by itself, across a relay restart. + // + // We do NOT re-subscribe on reconnect. We used to carry code that did (disconnect, re-auth, + // re-assert the REQ), and it had NO production caller — so every reconnect in the field was + // always the SDK's, and that code was only masking an unverified assumption. It is deleted, and + // this pins the behaviour we actually stand on instead. + // + // What the relay really does is worth stating, because it looks like breakage in the logs: the + // `#p`-gated REQ IS refused after a reconnect — the relay sends `CLOSED auth-required:` because + // the pool re-issues the REQ before NIP-42 completes. The SDK then authenticates and re-asserts + // the SAME subscription id, and because a fresh REQ also returns matching STORED events, work + // published while we were away is delivered on recovery rather than lost. + // + // The event here is published to the REPLACEMENT relay with NO wait for our session to recover + // first, so it lands while our subscription is still down. That is the field case — a relay + // restarts, the seller delivers, and we must still see it — and it is the property a test that + // waited for recovery before publishing would silently fail to cover. + // + // Arm 1 (baseline delivery before any reconnect) is the non-vacuity anchor: without it, a + // subscription that never delivered anything at all would pass arm 2 by never being tested. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn the_sdk_recovers_our_job_subscription_across_a_relay_restart() { + use nostr_relay_builder::prelude::{ + LocalRelay, RelayBuilder, RelayBuilderNip42, RelayBuilderNip42Mode, + }; + use nostr_sdk::prelude::EventBuilder; + + // Pin a port so the replacement relay comes up at the SAME url the client will retry. + let port = { + let probe = std::net::TcpListener::bind("127.0.0.1:0").expect("ephemeral port"); + probe.local_addr().expect("addr").port() + }; + let fixture = |port: u16| { + LocalRelay::new( + RelayBuilder::default() + .port(port) + .nip42(RelayBuilderNip42 { mode: RelayBuilderNip42Mode::Both }), + ) + }; + + let relay_one = fixture(port); + relay_one.run().await.expect("relay one"); + let relay_url = relay_one.url().await.to_string(); + + let buyer = Keys::generate(); + let handle = spawn(buyer.clone(), &relay_url).await.expect("spawn"); + let mut events = handle.subscribe_events(); + + // A counterparty result event matching the buyer's static filter, from a SEPARATE key. + let seller = Keys::generate(); + let deliver = |label: &'static str| { + let seller = seller.clone(); + let buyer_pk = buyer.public_key(); + let url = relay_url.clone(); + async move { + let event = EventBuilder::new(Kind::Custom(crate::kinds::JOB_RESULT_KIND), label) + .tags([ + nostr_sdk::Tag::hashtag(crate::gateway::MOBEE_TAG), + nostr_sdk::Tag::public_key(buyer_pk), + ]) + .sign(&seller) + .await + .expect("sign"); + let publisher = Client::new(seller.clone()); + publisher + .pool() + .add_relay(&url, RelayOptions::default().reconnect(false)) + .await + .expect("add"); + publisher.connect().await; + publisher.wait_for_connection(CONNECT_WAIT).await; + let sent = publisher.send_event(&event).await; + let _ = publisher.disconnect().await; + (event.id, sent.is_ok()) + } + }; + + // Wait for the session to actually be up before the baseline. `spawn` deliberately does not + // wait for the handshake, and this fixture challenges LAZILY, so boot can sit in NoChallenge + // for CONNECT_WAIT before the job subscription is opened at all. + let mut ready = false; + for _ in 0..30 { + if matches!(handle.probe().await, Ok(true)) { + ready = true; + break; + } + tokio::time::sleep(Duration::from_secs(2)).await; + } + assert!(ready, "the session must come up before the baseline arm means anything"); + + // Baseline: delivery works BEFORE any reconnect. If this fails the probe says nothing. + let (first_id, first_sent) = deliver("before-reconnect").await; + let before = tokio::time::timeout(Duration::from_secs(30), async { + loop { + if let Ok(event) = events.recv().await { + if event.id == first_id { + return true; + } + } + } + }) + .await + .unwrap_or(false); + assert!(first_sent, "the fixture must accept the counterparty's baseline delivery"); + assert!( + before, + "baseline delivery must work, or arm 2 proves nothing — a subscription that never \ + delivered would pass it by never having been exercised" + ); + + // Arm 2 — tear the relay down and bring a fresh one up on the SAME url, then publish + // IMMEDIATELY, without waiting for our own session to recover. The delivery lands while our + // subscription is still closed, so passing this means recovery replayed it. + relay_one.shutdown(); + tokio::time::sleep(Duration::from_secs(2)).await; + let relay_two = fixture(port); + relay_two.run().await.expect("relay two"); + + let (second_id, second_sent) = deliver("during-outage").await; + assert!(second_sent, "the fixture must accept the counterparty's delivery"); + let after = tokio::time::timeout(Duration::from_secs(60), async { + loop { + if let Ok(event) = events.recv().await { + if event.id == second_id { + return true; + } + } + } + }) + .await + .unwrap_or(false); + + assert!( + after, + "the SDK must recover our job subscription after a relay restart and deliver work \ + published while we were away. If this fails, every reconnect silently demotes the \ + delivery watcher to its periodic sweep — deliveries still settle, but seconds become \ + minutes — and the fix is to re-assert the REQ ourselves AFTER auth completes." + ); + relay_two.shutdown(); + } + // TOOTH — an empty accepted-set is a FAILURE, never a quiet success. // // `send_event_to` returns `Ok(output)` even when the relay refused: the reason goes into From 26c523b8d72a07cbe15d562e74f81af93d19181e Mon Sep 17 00:00:00 2001 From: "worker:mobee-buyer-lazystart" Date: Mon, 27 Jul 2026 02:02:32 -0700 Subject: [PATCH 7/7] buyer: prove the reconcile cannot release a reservation mid-settle MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The money lock the reconcile takes for its apply had no test. It closes a real window — a pass whose gather decided a job was dead must not act on that after a settle has melted but before it flips the reservation to spent, or funds that already left the wallet get counted as available again. The assertion is on the reconcile's own report rather than the resulting ledger state, and that distinction is why this was missing rather than merely untested: the end state converges either way, because a released row that turns out paid is converted by the next pass. Any test asking what the ledger looks like afterwards passes with the lock removed. The decision is where the bug is visible. It also asserts up front that the job really does classify as dead, since a job that was never at risk of release would satisfy the rest of the test for no reason. Co-Authored-By: Claude Opus 5 (1M context) --- crates/mobee-core/src/buyer/mod.rs | 89 ++++++++++++++++++++++++++++++ 1 file changed, 89 insertions(+) diff --git a/crates/mobee-core/src/buyer/mod.rs b/crates/mobee-core/src/buyer/mod.rs index e63248e3..17da0ec6 100644 --- a/crates/mobee-core/src/buyer/mod.rs +++ b/crates/mobee-core/src/buyer/mod.rs @@ -1299,6 +1299,95 @@ mod tests { std::env::temp_dir().join(format!("mobee-buyer-mod-{label}-{}-{id}", std::process::id())) } + // ★ THE MELT-TO-FLIP WINDOW TOOTH. The reconcile must not be able to release a reservation while + // a settle is between its wallet melt and its `reserved → spent` flip. + // + // `settle_job` holds the money lock across pay-then-flip. The reconcile GATHERS unlocked (it is + // per-job relay I/O and would block trades for seconds) and takes the lock only for the apply. + // That is the whole protection: a pass whose gather decided "dead" cannot act on that decision + // until the settle has finished moving the amount from `reserved` into `spent`. Without it, a + // pass could free funds that had already left the wallet, and `available` would over-state by + // the amount for as long as the discrepancy stood. + // + // ★ THE ASSERTION IS ON THE REPORT, NOT THE FINAL STATE — and that is the point. The final state + // converges to `spent` either way (a released row that turns out paid is converted by the very + // next pass), so a test that only checked the end state would pass with the lock REMOVED. That + // convergence is exactly what let this ship untoothed: the bug is invisible in the outcome and + // visible only in the decision. + // + // Deterministic: the settle-holder keeps the lock for far longer than a localhost gather takes, + // so the ordering is not a coin flip. And it fails safe — if the gather were somehow slow enough + // to apply after the flip, it would observe `spent` and still keep, passing for the right reason. + // + // Red-on-revert: delete the `money_lock` acquisition in `reconcile_reservations` and the pass + // releases a job whose payment was already in flight. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn the_reconcile_cannot_release_a_reservation_mid_settle() { + use nostr_relay_builder::prelude::{LocalRelay, RelayBuilder}; + + let root = temp_home("mid-settle"); + let _ = std::fs::remove_dir_all(&root); + let mut home = bootstrap_home(&root).expect("bootstrap home"); + + // A relay that serves NO claims, so the classification genuinely reaches `Dead` — with an + // unreachable relay every job is conservatively treated as still-payable and the reconcile + // would never try to release anything, which would make this test vacuous. + let relay = LocalRelay::new(RelayBuilder::default()); + relay.run().await.expect("relay run"); + home.config.relay_url = relay.url().await.to_string(); + + let (_lock, context, _socket) = bootstrap(home).await.expect("buyer bootstrap"); + let job = "a".repeat(64); + context + .store + .reserve(&job, 4, 1_000, u64::MAX, 0, now_unix()) + .expect("reserve"); + + // Confirm the premise: with no bind, no payment journal and no live claim, this job WOULD be + // released by a pass that got to act. Without this the tooth could pass because nothing was + // ever at risk. + let would_release = plan_reconcile( + &[job.clone()], + &BTreeMap::new(), + &BTreeMap::from([(job.clone(), false)]), + ); + assert_eq!( + would_release[&job], + reservations::JobDisposition::Dead, + "premise: this job must classify Dead, or nothing is at risk and the tooth is vacuous" + ); + + // Stand in for a settle that has melted but not yet flipped: hold the money lock, wait long + // enough that any unlocked apply would have already run, then perform the flip. + let settling = { + let context = context.clone(); + let job = job.clone(); + tokio::spawn(async move { + let _guard = context.money_lock.lock().await; + tokio::time::sleep(Duration::from_secs(2)).await; + context.store.convert_to_spent(&job, 4, now_unix()).expect("flip"); + }) + }; + + // Let the settle take the lock first, then run a real pass against it. + tokio::time::sleep(Duration::from_millis(200)).await; + let report = reconcile_reservations(&context).await.expect("reconcile"); + settling.await.expect("settle task"); + + assert!( + !report.released.contains(&job), + "the reconcile released a reservation whose payment was mid-flight — it must wait for \ + the money lock, not act on a gather that went stale. released={:?}", + report.released + ); + assert!( + report.kept.contains(&job), + "the pass should have observed the completed flip and kept the row; report={report:?}" + ); + let _ = std::fs::remove_dir_all(&root); + drop(relay); + } + // ★ THE #179/4b TOOTH: the reconcile reports the pass that changed NOTHING. // // A release moves the buyer's `available` — it is a money-visible decision — and the pass that