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 d5788ef9..17da0ec6 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; @@ -63,6 +64,17 @@ 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); +/// 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"; @@ -78,6 +90,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 +103,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 +138,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 +178,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), @@ -198,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. @@ -215,6 +238,12 @@ 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()); + // 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 } @@ -427,7 +456,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()), } @@ -560,36 +593,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), + }) +} + +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 settled { - Ok((outcome, _converted)) => Response::ok( + match settle_job(context, ¶ms.job_id, params.out).await { + Ok(outcome) => Response::ok( id, json!({ "pay": { @@ -604,8 +678,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()), } } @@ -765,12 +839,176 @@ 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 /// 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() @@ -807,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 @@ -961,6 +1277,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() }, }), ) } @@ -978,6 +1299,436 @@ 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 + // 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. + 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/relay.rs b/crates/mobee-core/src/buyer/relay.rs new file mode 100644 index 00000000..427b8fc3 --- /dev/null +++ b/crates/mobee-core/src/buyer/relay.rs @@ -0,0 +1,868 @@ +//! 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::sync::Arc; +use std::time::Duration; + +use nostr_sdk::prelude::{ + Client, Event, Filter, Keys, Kind, RelayOptions, RelayPoolNotification, RelayUrl, + SubscriptionId, +}; +use tokio::sync::{broadcast, 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"; + +/// 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, + reply: oneshot::Sender>, + }, + /// Ask the relay to serve one trivial REQ on the CURRENT session. + Probe { + 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, + events: broadcast::Sender>, +} + +impl RelayHandle { + /// The relay this buyer is bound to. + pub fn relay_url(&self) -> &str { + &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, + 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 + } +} + +/// 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 (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; + // 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, + ); + } + } + } + }); + + 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 +/// 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) +} + +#[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 — 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 + // `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(), + events: broadcast::channel(1).0, + }; + + 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}" + ); + } + } +} 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"); 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 0235cb53..47d76050 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. +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) + }) +} + /// 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};