From 6b11a262f32e829e27ae2102f36e3c3aeb3c6b98 Mon Sep 17 00:00:00 2001 From: "worker:mobee-recoveryfix" Date: Mon, 27 Jul 2026 13:14:03 -0700 Subject: [PATCH 1/6] seller: order the recovery resubscribe after NIP-42, and give the open-pool re-arm an owned schedule MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two seller recovery-loop fixes (#189 client half, #190). They land together because they share the CLOSED-handling arm and the relay fixture the teeth need; splitting them would produce two commits neither of which compiles alone. #189 — the recovery path let p-gated REQs out before NIP-42. nostr-sdk re-sends every REGISTERED subscription as its first act on socket-up (relay/inner.rs:748-752), before that connection has any auth state; auth happens later, in the ingester (:936). mobee-relay evaluates its p-gate against the empty authed pubkey of that unauthenticated session and answers `restricted:` — the permanent prefix — where the truth is the retryable `auth-required:`. nostr-sdk takes it at its word and DELETES the subscription (:1028), so the post-auth resubscribe at :941 cannot see it and never restores it. Carrying registrations across the reconnect therefore killed the kind-1059 money leg on every recovery. The registrations are now dropped BEFORE the new socket comes up, so that first resubscribe has nothing to send and the REQs go out after auth — the order boot always had. A failed reconnect re-registers them, because the SDK's background reconnect is a real recovery path and can only restore what it still knows. Plus the belt the ordering fix cannot cover: the SDK's own background reconnect resubscribes pre-auth too, and that path has no hook. A `restricted:` CLOSED of a subscription whose filters all pin #p to our OWN pubkey, on an authenticated session, is that race and not a gate violation — so it is re-issued once per authenticated session. The taxonomy is untouched: `restricted:` stays permanent-class, a genuine wrong-#p refusal cannot reach the branch (we author these filters), and a second refusal falls through to the paced recovery. Also: a CLOSED for a subscription id we never registered no longer forces a recovery. It cannot be a leg of ours going deaf, and escalating it was costing a full reconnect per cycle — which, before the ordering fix, re-closed the 1059 leg. It is logged with the age of the last backfill and of the last NIP-42 auth, which is what the relay owner needs to tell our own transient `fetch_events` REQ (generated id, same cadence) from a relay-side auth-TTL sweep. #190 — the open-pool re-arm waited on a trigger nothing guarantees. The degrade re-armed only via `open_pool_degraded = false` in the recovery-success arm, so a seat that degrades and then stays healthy has no path back. The re-arm now rides the wrap-backfill tick: owned, unconditional, and not disableable the way the heartbeat is. Acceptance is the relay's EOSE — a response NIP-01 owes us — never our own send having succeeded, and a refusal doubles a capped backoff so a permanently refusing relay costs one REQ per cap interval rather than one per tick. An attempt that draws no verdict at all is treated as a refusal, so there is no timer-less park. This half is defence in depth, not a repair for an observed seat: the reported stuck specimen was withdrawn — every seat seen degraded was flapping on the #189 sawtooth. The gap is structural, and it survives the #189 fix. Teeth (crates/mobee-core/src/seller_node/p_gate_relay_fixture.rs): a relay fixture that answers a p-gated REQ from an unauthenticated session with `restricted:`, as the deployed relay does. The nostr-relay-builder fixture says `auth-required:`, which nostr-sdk keeps and restores by itself, so against it every ordering passes. Co-Authored-By: Claude Opus 5 --- Cargo.lock | 2 + crates/mobee-core/Cargo.toml | 8 +- crates/mobee-core/src/seller_node/mod.rs | 2 + .../src/seller_node/p_gate_relay_fixture.rs | 356 +++++++ crates/mobee-core/src/seller_node/run.rs | 967 +++++++++++++++++- 5 files changed, 1288 insertions(+), 47 deletions(-) create mode 100644 crates/mobee-core/src/seller_node/p_gate_relay_fixture.rs diff --git a/Cargo.lock b/Cargo.lock index 114b4ec3..9d796513 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2601,6 +2601,7 @@ dependencies = [ "cdk", "cdk-sqlite", "config", + "futures-util", "getrandom 0.3.4", "git2", "hex", @@ -2615,6 +2616,7 @@ dependencies = [ "serde_json", "sha2 0.11.0", "tokio", + "tokio-tungstenite", "toml", "url", "uuid", diff --git a/crates/mobee-core/Cargo.toml b/crates/mobee-core/Cargo.toml index 9ed7bd95..31032b93 100644 --- a/crates/mobee-core/Cargo.toml +++ b/crates/mobee-core/Cargo.toml @@ -51,6 +51,9 @@ required-features = ["git-delivery"] [dev-dependencies] async-trait = "0.1.89" cdk-sqlite = "=0.17.2" +# Sink/Stream helpers for the p-gate relay fixture's websocket split. Already in the workspace lock +# via nostr-sdk; no new resolution. +futures-util = { version = "0.3", default-features = false } # In-process NIP-01 relay for the seller-daemon live-delivery integration test. Same nostr 0.44 # line as `nostr-sdk` above, so its `Event`/`Filter` types unify with the daemon's. nostr-relay-builder = "0.44" @@ -58,6 +61,9 @@ nostr-relay-builder = "0.44" # rcgen self-signed cert. ring-backed on both — aws-lc-rs is not in the workspace lock. rcgen = { version = "0.13", default-features = false, features = ["ring"] } rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12"] } -tokio = { version = "1.52.0", features = ["macros", "rt", "rt-multi-thread", "time", "test-util"] } +tokio = { version = "1.52.0", features = ["macros", "net", "rt", "rt-multi-thread", "time", "test-util"] } +# The p-gate relay fixture speaks raw NIP-01 over a websocket, because it has to answer a p-gated REQ +# with the `restricted:` prefix — see the module docs for why nostr-relay-builder cannot. +tokio-tungstenite = { version = "0.26", default-features = false, features = ["handshake"] } url = "2.5.8" uuid = "1.23.5" diff --git a/crates/mobee-core/src/seller_node/mod.rs b/crates/mobee-core/src/seller_node/mod.rs index 3d572f89..80162ab8 100644 --- a/crates/mobee-core/src/seller_node/mod.rs +++ b/crates/mobee-core/src/seller_node/mod.rs @@ -22,6 +22,8 @@ pub mod buzz; pub mod ingester; pub mod lock; pub mod outbox; +#[cfg(test)] +mod p_gate_relay_fixture; pub mod publisher; pub mod run; pub mod roster; diff --git a/crates/mobee-core/src/seller_node/p_gate_relay_fixture.rs b/crates/mobee-core/src/seller_node/p_gate_relay_fixture.rs new file mode 100644 index 00000000..87d45e6e --- /dev/null +++ b/crates/mobee-core/src/seller_node/p_gate_relay_fixture.rs @@ -0,0 +1,356 @@ +//! An in-process relay that reproduces mobee-relay's `#p`-gate wire behaviour, and records what the +//! client actually sent. +//! +//! The `nostr-relay-builder` fixture used elsewhere in this crate cannot express the failure in #189. +//! Its NIP-42 read gate answers an unauthenticated `REQ` with the `auth-required:` prefix +//! (`local/inner.rs:961-989`), which nostr-sdk classifies as `MarkAsClosed` — the subscription stays +//! in the registry and the post-auth `resubscribe()` restores it automatically. The bug under test is +//! precisely what happens when the relay says `restricted:` instead: nostr-sdk classifies that as +//! `Remove` (`relay/inner.rs:1028`), deletes the subscription, and nothing ever restores it. A +//! fixture that self-heals would report every ordering as green. +//! +//! So this speaks the deployed relay's rule directly: a `REQ` carrying a `#p` filter is refused with +//! `restricted:` unless the session has completed NIP-42 **and** the `#p` value is the authenticated +//! pubkey. That one rule covers both cases the teeth need to tell apart — the pre-auth race (right +//! `#p`, no auth yet) and a genuine gate violation (someone else's `#p`, fully authed). +//! +//! Every `REQ` is recorded with the session's auth state at the moment it arrived, which is the +//! property #189 is actually about: a REQ that reaches the relay before AUTH does. + +use std::collections::VecDeque; +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; + +use futures_util::{SinkExt, StreamExt}; +use serde_json::{json, Value}; +use tokio::sync::Mutex; + +/// What the relay answered a `REQ` with. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(super) enum Verdict { + /// Served — the end-of-stored-events the client waits for. + Eose, + /// Refused, with the full reason string (prefix included). + Closed(String), +} + +/// One `REQ` as the relay saw it. +#[derive(Debug, Clone)] +pub(super) struct ReqRecord { + pub subscription_id: String, + /// Whether NIP-42 had completed on this socket when the `REQ` arrived. `false` here on a + /// `#p`-pinned subscription IS the #189 bug — the client let a p-gated REQ out before auth. + pub authenticated: bool, + /// Whether any filter pinned `#p`. + pub p_pinned: bool, + /// How many filters rode this one `REQ`. The grouped offer REQ carries 2 (targeted + open-pool); + /// the degraded targeted-only shape carries 1. + pub filter_count: usize, + /// Whether at least one filter carried NO `#p` — i.e. the un-pinned open-pool half is present. + pub has_unpinned_filter: bool, + pub verdict: Verdict, +} + +/// A refusal the test has armed for the relay to emit on the next matching `REQ`, once. +#[derive(Debug, Clone)] +struct ForcedClose { + subscription_id: String, + reason: String, + /// Match only a `REQ` that carries an un-pinned filter. Without this, a refusal armed for the + /// open-pool half would be spent on the targeted-only re-subscribe that follows a rejection. + require_unpinned: bool, +} + +#[derive(Debug, Default)] +struct Controls { + /// Refusals queued by the test, consumed one per matching `REQ`. + forced: Mutex>, + /// Writer for the most recently accepted socket, so a test can push an UNSOLICITED `CLOSED` — + /// which is what the deployed relay does, and what neither a REQ nor a policy hook can model. + live: Mutex>>>, +} + +/// A running fixture relay. Dropping it stops accepting new connections. +pub(super) struct PGateRelay { + url: String, + transcript: Arc>>, + controls: Arc, + /// Sockets accepted so far. A reconnect is the only thing that increments this, which makes + /// "no reconnect was required" an observable rather than an inference. + connections: Arc, + _accept: tokio::task::JoinHandle<()>, +} + +impl PGateRelay { + /// Bind on an ephemeral port and start serving. + /// + /// `auth_delay` is how long the relay withholds its NIP-42 challenge after a socket opens. The + /// deployed relay challenges immediately, but a client that fires REQs on socket-up loses the + /// race regardless of how fast the challenge is; holding it open makes that deterministic + /// instead of timing-dependent, so the tooth cannot pass by being lucky. + pub(super) async fn start(auth_delay: Duration) -> Self { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind fixture relay"); + let addr: SocketAddr = listener.local_addr().expect("fixture relay addr"); + let transcript: Arc>> = Arc::new(Mutex::new(Vec::new())); + let controls: Arc = Arc::new(Controls::default()); + + let connections = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let accept = tokio::spawn({ + let transcript = Arc::clone(&transcript); + let controls = Arc::clone(&controls); + let connections = Arc::clone(&connections); + async move { + while let Ok((stream, _)) = listener.accept().await { + connections.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + let transcript = Arc::clone(&transcript); + let controls = Arc::clone(&controls); + tokio::spawn(async move { + let _ = serve_connection(stream, auth_delay, transcript, controls).await; + }); + } + } + }); + + Self { + url: format!("ws://{addr}"), + transcript, + controls, + connections, + _accept: accept, + } + } + + /// How many sockets this relay has accepted. + pub(super) fn connections(&self) -> usize { + self.connections.load(std::sync::atomic::Ordering::SeqCst) + } + + /// Push an UNSOLICITED `CLOSED` for `subscription_id` down the live socket — the deployed relay's + /// behaviour when it drops a subscription out from under a client that is otherwise healthy. + pub(super) async fn close_now(&self, subscription_id: &str, reason: &str) { + let live = self.controls.live.lock().await.clone(); + if let Some(writer) = live { + let _ = send(&writer, json!(["CLOSED", subscription_id, reason])).await; + } + } + + pub(super) fn url(&self) -> String { + self.url.clone() + } + + /// Arm one refusal: the next `REQ` for `subscription_id` is answered `CLOSED` with `reason`, + /// whatever the session's auth state. Used to induce a degrade on an otherwise healthy seat. + pub(super) async fn force_close_once(&self, subscription_id: &str, reason: &str) { + self.controls.forced.lock().await.push_back(ForcedClose { + subscription_id: subscription_id.to_string(), + reason: reason.to_string(), + require_unpinned: false, + }); + } + + /// Refuse the next `count` `REQ`s for `subscription_id` that carry an un-pinned filter — i.e. a + /// relay that keeps rejecting the open-pool half while serving the targeted one. + pub(super) async fn refuse_unpinned(&self, subscription_id: &str, count: usize, reason: &str) { + let mut forced = self.controls.forced.lock().await; + for _ in 0..count { + forced.push_back(ForcedClose { + subscription_id: subscription_id.to_string(), + reason: reason.to_string(), + require_unpinned: true, + }); + } + } + + /// Every `REQ` the relay has seen, in arrival order. + pub(super) async fn reqs(&self) -> Vec { + self.transcript.lock().await.clone() + } + + /// Every `REQ` seen for one subscription id. + pub(super) async fn reqs_for(&self, subscription_id: &str) -> Vec { + self.reqs() + .await + .into_iter() + .filter(|record| record.subscription_id == subscription_id) + .collect() + } + + /// Wait until `predicate` holds over the transcript, or give up. Returns whether it held — + /// the caller asserts, so a timeout reads as the failure it is rather than a hang. + pub(super) async fn wait_until(&self, timeout: Duration, predicate: F) -> bool + where + F: Fn(&[ReqRecord]) -> bool, + { + tokio::time::timeout(timeout, async { + loop { + if predicate(&self.reqs().await) { + return; + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + }) + .await + .is_ok() + } +} + +/// Serve one client socket for its whole life. +async fn serve_connection( + stream: tokio::net::TcpStream, + auth_delay: Duration, + transcript: Arc>>, + controls: Arc, +) -> Result<(), Box> { + let ws = tokio_tungstenite::accept_async(stream).await?; + let (writer, mut reader) = ws.split(); + let writer = Arc::new(Mutex::new(writer)); + *controls.live.lock().await = Some(Arc::clone(&writer)); + + // The challenge goes out on its own task so the delay never blocks reading: the REQs we are here + // to observe arrive DURING that window. + let challenge = "mobee-recoveryfix-challenge"; + tokio::spawn({ + let writer = Arc::clone(&writer); + async move { + tokio::time::sleep(auth_delay).await; + let _ = send(&writer, json!(["AUTH", challenge])).await; + } + }); + + // Per-session NIP-42 state. `None` until the client answers the challenge. + let mut authed_pubkey: Option = None; + + while let Some(message) = reader.next().await { + let message = message?; + let text = match message { + tokio_tungstenite::tungstenite::Message::Text(text) => text, + tokio_tungstenite::tungstenite::Message::Close(_) => break, + _ => continue, + }; + let Ok(frame) = serde_json::from_str::>(&text) else { + continue; + }; + match frame.first().and_then(Value::as_str) { + Some("AUTH") => { + let Some(event) = frame.get(1) else { continue }; + // Kind 22242 is the NIP-42 auth event. Checking it keeps the fixture honest: a client + // that authenticated with something else has not authenticated. + if event.get("kind").and_then(Value::as_u64) == Some(22242) { + authed_pubkey = event + .get("pubkey") + .and_then(Value::as_str) + .map(str::to_string); + } + let id = event.get("id").and_then(Value::as_str).unwrap_or_default(); + send(&writer, json!(["OK", id, authed_pubkey.is_some(), ""])).await?; + } + Some("EVENT") => { + let Some(event) = frame.get(1) else { continue }; + let id = event.get("id").and_then(Value::as_str).unwrap_or_default(); + send(&writer, json!(["OK", id, true, ""])).await?; + } + Some("REQ") => { + let Some(subscription_id) = + frame.get(1).and_then(Value::as_str).map(str::to_string) + else { + continue; + }; + let filters: Vec<&Value> = frame.iter().skip(2).collect(); + let pinned: Vec> = filters + .iter() + .map(|filter| { + filter + .get("#p") + .and_then(Value::as_array) + .and_then(|values| values.first()) + .and_then(Value::as_str) + }) + .collect(); + + let verdict = decide( + &subscription_id, + &pinned, + authed_pubkey.as_deref(), + &controls, + ) + .await; + + + transcript.lock().await.push(ReqRecord { + subscription_id: subscription_id.clone(), + authenticated: authed_pubkey.is_some(), + p_pinned: pinned.iter().any(Option::is_some), + filter_count: filters.len(), + has_unpinned_filter: pinned.iter().any(Option::is_none), + verdict: verdict.clone(), + }); + + match verdict { + Verdict::Eose => { + send(&writer, json!(["EOSE", subscription_id])).await?; + } + Verdict::Closed(reason) => { + send(&writer, json!(["CLOSED", subscription_id, reason])).await?; + } + } + } + _ => {} + } + } + Ok(()) +} + +/// The deployed relay's rule, and the whole reason this fixture exists. +/// +/// A `#p`-pinned filter is refused with the PERMANENT-class `restricted:` prefix unless the session +/// is authenticated as that very pubkey. mobee-relay reaches this verdict by evaluating its p-gate +/// against an empty authed pubkey on an unauthenticated connection (`req.rs:208`), which is why the +/// pre-auth race and a genuine wrong-`#p` request are indistinguishable on the wire — the exact +/// ambiguity the client-side fix has to resolve without softening the taxonomy. +async fn decide( + subscription_id: &str, + pinned: &[Option<&str>], + authed_pubkey: Option<&str>, + controls: &Controls, +) -> Verdict { + let has_unpinned = pinned.iter().any(Option::is_none); + let mut forced = controls.forced.lock().await; + if let Some(index) = forced.iter().position(|entry| { + entry.subscription_id == subscription_id && (!entry.require_unpinned || has_unpinned) + }) { + let entry = forced.remove(index).expect("index just found"); + return Verdict::Closed(entry.reason); + } + drop(forced); + + for value in pinned.iter().flatten() { + if authed_pubkey != Some(*value) { + return Verdict::Closed( + "restricted: p-gated events require #p matching your pubkey".to_string(), + ); + } + } + Verdict::Eose +} + +type Writer = futures_util::stream::SplitSink< + tokio_tungstenite::WebSocketStream, + tokio_tungstenite::tungstenite::Message, +>; + +async fn send( + writer: &Arc>, + value: Value, +) -> Result<(), Box> { + writer + .lock() + .await + .send(tokio_tungstenite::tungstenite::Message::Text( + value.to_string().into(), + )) + .await?; + Ok(()) +} diff --git a/crates/mobee-core/src/seller_node/run.rs b/crates/mobee-core/src/seller_node/run.rs index 3bc9d67d..61a2e6c2 100644 --- a/crates/mobee-core/src/seller_node/run.rs +++ b/crates/mobee-core/src/seller_node/run.rs @@ -281,6 +281,113 @@ fn subscription_label(id: &str) -> &'static str { } } +/// Whether `id` names a long-lived subscription this node registers. Anything else — a transient +/// `fetch_events` REQ, a stale generation, a relay-side artefact — is not a leg of ours, so its +/// closure cannot make us deaf. +fn is_our_subscription(id: &str) -> bool { + matches!( + id, + OFFER_SUB_ID | AWARD_SUB_ID | WRAP_SUB_ID | LIVENESS_PROBE_SUB_ID + ) +} + +/// Whether EVERY filter on this subscription pins `#p` to our own pubkey. +/// +/// This is the precondition for reading a `restricted:` CLOSED as the #189 pre-auth race instead of a +/// gate violation, and the CLOSED-prefix taxonomy stays load-bearing everywhere else: `restricted:` +/// remains permanent-class, and the SDK's `Remove` classification is not softened. The carve-out is +/// sound because mobee-relay's p-gate has exactly two ways to refuse a `#p` filter — the `#p` names +/// somebody else, or the connection had no authenticated pubkey to compare it against. We author +/// these filters from `self.seller_pubkey`, so the first is impossible by construction for the ids +/// below; only the second remains, and the second is retryable once auth exists. A subscription +/// carrying ANY un-pinned filter is excluded, because there the refusal may genuinely be about the +/// un-pinned half — that case has its own repair, the targeted-only degrade. +fn subscription_pins_only_our_pubkey(id: &str, claim_open_pool: bool) -> bool { + match id { + AWARD_SUB_ID | WRAP_SUB_ID => true, + OFFER_SUB_ID => !claim_open_pool, + _ => false, + } +} + +/// Owned ticks to wait before the next open-pool re-arm attempt, after `rejections` consecutive +/// refusals (#190). +/// +/// Doubling, capped: a relay that permanently refuses the un-pinned filter must cost one REQ per cap +/// interval, never one REQ per tick. Zero rejections means "attempt on the next tick" — the first +/// try after a degrade is not delayed, because the degrade itself is usually collateral from the +/// #189 race rather than a real refusal of the open-pool half. +fn open_pool_rearm_cooldown_ticks(rejections: u32) -> u32 { + /// Ceiling on the backoff, in owned ticks. + const MAX_COOLDOWN_TICKS: u32 = 12; + match rejections { + 0 => 0, + n => (1u32 << (n - 1).min(31)).min(MAX_COOLDOWN_TICKS), + } +} + +/// Open-pool degrade bookkeeping (#190). Absent = the open-pool half is live. +/// +/// The re-arm this drives is DEFENCE IN DEPTH, not a repair for an observed stuck seat: the reported +/// specimen was withdrawn — every seat seen degraded was flapping on the #189 sawtooth, not stuck. +/// The gap it closes is structural rather than field-observed. A seat that degrades and then never +/// recovers has no path back, because the only re-arm was `open_pool_degraded = false` in the +/// recovery-success arm; a healthy seat produces no recoveries, so it would hold the degraded shape +/// indefinitely. That reasoning survives the #189 fix, which is why the owned schedule stays. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct OpenPoolDegrade { + /// Consecutive re-arm attempts the relay refused. + rejections: u32, + /// Owned ticks still to skip before the next attempt. + cooldown_ticks: u32, + /// An attempt is on the wire, awaiting the relay's verdict: `EOSE` re-arms, `CLOSED` rejects. + attempt_pending: bool, +} + +impl OpenPoolDegrade { + /// Freshly degraded: attempt the re-arm on the very next owned tick. + fn new() -> Self { + Self { + rejections: 0, + cooldown_ticks: 0, + attempt_pending: false, + } + } + + /// What the next owned tick should do. + fn on_tick(&mut self) -> RearmStep { + if self.attempt_pending { + // The previous attempt drew neither an EOSE nor a CLOSED within a full tick. Treat the + // silence as a refusal rather than waiting on it: an attempt with no verdict pending is + // exactly the timer-less park this fix exists to remove. + self.reject(); + return RearmStep::Wait; + } + if self.cooldown_ticks > 0 { + self.cooldown_ticks -= 1; + return RearmStep::Wait; + } + self.attempt_pending = true; + RearmStep::Attempt + } + + /// The relay refused (or ignored) the re-arm. + fn reject(&mut self) { + self.attempt_pending = false; + self.rejections = self.rejections.saturating_add(1); + self.cooldown_ticks = open_pool_rearm_cooldown_ticks(self.rejections); + } +} + +/// What an owned tick does about a degraded open-pool half. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum RearmStep { + /// Send the grouped offer REQ again. + Attempt, + /// Still cooling down, or still waiting on the last attempt's verdict. + Wait, +} + /// Ask the relay to serve one trivial REQ on the CURRENT session and wait for its `EOSE`. True means /// the relay is answering OUR subscriptions on THIS authenticated connection — the exact property the /// #150 watchdog needs, and the one thing a heartbeat cannot demonstrate. @@ -402,6 +509,33 @@ async fn reconnect_and_authenticate( relay_auth::wait_for_nip42_auth(&mut relay_notifications, CONNECT_WAIT).await } +/// Leave the SDK with nothing to re-`REQ` when the next socket comes up, and return how many +/// registrations survived — which must be zero. +/// +/// `RelayPool::unsubscribe_all` is best-effort by construction: the relay-level loop removes each id +/// from the map and then sends its `CLOSE`, propagating the first send error with `?` +/// (`relay/inner.rs:1724-1736`), so one failed send leaves every remaining id registered. A single +/// leftover registration is the whole #189 hazard — it is the thing that gets re-sent pre-auth — so +/// the relay's own view is swept afterwards. `Relay::unsubscribe` removes before it sends, so the +/// sweep empties the map whether or not the socket can carry the `CLOSE`. +async fn clear_subscription_registrations( + client: &Client, + relay: &nostr_sdk::prelude::Relay, +) -> usize { + client.unsubscribe_all().await; + for id in relay.subscriptions().await.keys() { + let _ = relay.unsubscribe(id).await; + } + let leftover = relay.subscriptions().await.len(); + if leftover > 0 { + eprintln!( + "seller node WARN: {leftover} subscription registration(s) survived the pre-reconnect \ + clear; they will be re-sent before NIP-42 completes" + ); + } + leftover +} + /// The refusal reason `on_offer` logs when an offer fails to parse. /// /// A cross-version offer is a DISTINCT refusal from a malformed one (#146 / #117 refusal taxonomy): @@ -532,6 +666,9 @@ pub struct SellerNodeRunner { publisher: RelayPublisher, relay_url: String, seller_pubkey: nostr_sdk::PublicKey, + /// Outcome of the boot NIP-42 handshake, which seeds the run loop's view of whether the current + /// socket is authenticated. `NoChallenge` is not authentication. + boot_auth: AuthWait, } impl SellerNodeRunner { @@ -590,15 +727,23 @@ impl SellerNodeRunner { let mut relay_notifications = relay.notifications(); client.connect().await; client.wait_for_connection(CONNECT_WAIT).await; - match relay_auth::wait_for_nip42_auth(&mut relay_notifications, CONNECT_WAIT).await { - Ok(AuthWait::Authenticated) => eprintln!("seller node relay authenticated (NIP-42)"), - Ok(AuthWait::NoChallenge) => eprintln!( - "seller node WARN: no NIP-42 challenge within {CONNECT_WAIT:?}; proceeding \ - (auto-auth stays ON — a challenge on the REQ still authenticates). p-gated kind-1059 \ - receive may be degraded until auth completes." - ), + let boot_auth = match relay_auth::wait_for_nip42_auth(&mut relay_notifications, CONNECT_WAIT) + .await + { + Ok(AuthWait::Authenticated) => { + eprintln!("seller node relay authenticated (NIP-42)"); + AuthWait::Authenticated + } + Ok(AuthWait::NoChallenge) => { + eprintln!( + "seller node WARN: no NIP-42 challenge within {CONNECT_WAIT:?}; proceeding \ + (auto-auth stays ON — a challenge on the REQ still authenticates). p-gated \ + kind-1059 receive may be degraded until auth completes." + ); + AuthWait::NoChallenge + } Err(error) => return Err(NodeError::Relay(format!("NIP-42 auth: {error}"))), - } + }; let publisher = RelayPublisher::new(node.signer().clone(), client.clone(), &relay_url); @@ -608,6 +753,7 @@ impl SellerNodeRunner { publisher, relay_url, seller_pubkey, + boot_auth, }) } @@ -658,24 +804,44 @@ impl SellerNodeRunner { /// (see [`probe_relay_serves_our_reqs`]), so that REQ could only ever have returned nothing. /// Liveness is asserted by the probe instead. async fn subscribe_all(&self, since: Option) -> Result<(), NodeError> { - let apply = |filter: Filter| match since { - Some(cursor) => filter.since(cursor), - None => filter, - }; - self.subscribe_offers(since, self.claim_open_pool()).await?; - let award_filter = apply( - Filter::new() + for id in [OFFER_SUB_ID, AWARD_SUB_ID, WRAP_SUB_ID] { + self.subscribe_one(id, since).await?; + } + Ok(()) + } + + /// Issue (or re-issue) the REQ for ONE named subscription, so a single leg can be repaired + /// without re-dialing the relay or disturbing the others. + async fn subscribe_one( + &self, + id: &str, + since: Option, + ) -> Result<(), NodeError> { + // The offer REQ has its own entry point: it is the only subscription with a meaningful + // partial form, and it carries two filters rather than one. + if id == OFFER_SUB_ID { + return self.subscribe_offers(since, self.claim_open_pool()).await; + } + let base = match id { + AWARD_SUB_ID => Filter::new() .kind(Kind::Custom(JOB_AWARD_KIND)) .hashtag(crate::gateway::MOBEE_TAG) .pubkey(self.seller_pubkey), - ); - let wrap_filter = apply(Filter::new().kind(Kind::GiftWrap).pubkey(self.seller_pubkey)); - for (id, filter) in [(AWARD_SUB_ID, award_filter), (WRAP_SUB_ID, wrap_filter)] { - self.client - .subscribe_with_id(nostr_sdk::SubscriptionId::new(id), filter, None) - .await - .map_err(|error| NodeError::Relay(format!("subscribe {id}: {error}")))?; - } + WRAP_SUB_ID => Filter::new().kind(Kind::GiftWrap).pubkey(self.seller_pubkey), + other => { + return Err(NodeError::Relay(format!( + "subscribe {other}: not one of ours" + ))) + } + }; + let filter = match since { + Some(cursor) => base.since(cursor), + None => base, + }; + self.client + .subscribe_with_id(nostr_sdk::SubscriptionId::new(id), filter, None) + .await + .map_err(|error| NodeError::Relay(format!("subscribe {id}: {error}")))?; Ok(()) } @@ -769,12 +935,31 @@ impl SellerNodeRunner { // probe. Seeded to "now" so a healthy node never trips before its first probe. let mut last_liveness_seen = tokio::time::Instant::now(); let mut last_liveness_seen_unix = now_unix(); - // Whether the offer REQ is currently running in its degraded targeted-only shape after a - // relay `CLOSED` (re-armed by the next successful recovery). - let mut open_pool_degraded = false; + // Set while the offer REQ is running in its degraded targeted-only shape after a relay + // `CLOSED`. Carries its own re-arm schedule (#190) — see [`OpenPoolDegrade`]. + let mut open_pool: Option = None; // A repair the CLOSED arm has asked for, run on the next heartbeat tick through the ONE // paced recovery path rather than an off-cadence ad-hoc resubscribe. let mut forced_recovery: Option = None; + // NIP-42 state of the CURRENT socket, and when it was last established. + // + // Tracked here because `Authenticated` is a RELAY notification that never becomes a pool + // notification (`relay/inner.rs:418` maps it to `None`), so the pool stream the loop already + // watches cannot see it. Seeded from the boot handshake. Both stale readings are bounded and + // safe: stale-false only declines a cheap retry and falls through to the paced recovery, + // while stale-true spends the single retry this session allows and then does the same. + let mut nip42_authed = matches!(self.boot_auth, AuthWait::Authenticated); + let mut last_authenticated_at = tokio::time::Instant::now(); + let mut relay_notifications = relay.notifications(); + // Subscriptions that have already spent their one post-auth retry on this session (#189 + // belt). Cleared whenever a new session authenticates, so the budget is per-session and can + // never become a loop. + let mut restricted_retry_used: std::collections::HashSet = + std::collections::HashSet::new(); + // When the last periodic wrap backfill ran. Reported alongside an unknown-id `CLOSED` so the + // relay owner can tell a refusal of our transient `fetch_events` REQ (which uses a generated + // id, and runs on exactly this cadence) from a relay-side sweep of a stale generation. + let mut last_backfill_at = tokio::time::Instant::now(); // Which path actually restored the receive leg. Manual recovery and the SDK's background // reconnect were previously indistinguishable in the log — which is how a manual path that // never once succeeded went unnoticed (#171). The next answered probe names it. @@ -791,6 +976,38 @@ impl SellerNodeRunner { // the positive signal external supervision watches. _ = wrap_backfill_tick.tick() => { self.run_wrap_backfill().await; + last_backfill_at = tokio::time::Instant::now(); + // #190: the open-pool half is re-armed on THIS tick, which is owned and + // unconditional. It rides the backfill rather than the heartbeat because the + // heartbeat is disableable by config, and a repair must not depend on a tick that + // may never fire. Acceptance is the relay's EOSE below — a response the protocol + // owes us — never the fact that we managed to send the REQ. + if let Some(state) = open_pool.as_mut() { + if state.on_tick() == RearmStep::Attempt { + let overlap = nostr_sdk::Timestamp::from( + last_liveness_seen_unix + .saturating_sub(STALL_OVERLAP_MARGIN_SECS as i64) + .max(0) as u64, + ); + match self.subscribe_offers(Some(overlap), true).await { + Ok(()) => eprintln!( + "seller node RELAY-CLOSED RE-ARM: retrying the open-pool half of \ + the offer subscription (attempt after {} rejection(s), \ + since={} overlap); the relay's EOSE confirms it", + state.rejections, + overlap.as_secs() + ), + Err(error) => { + state.reject(); + eprintln!( + "seller node RELAY-CLOSED RE-ARM failed to send ({error}); \ + next attempt in {} backfill tick(s)", + state.cooldown_ticks + ); + } + } + } + } continue; } // The heartbeat tick rides the SAME loop (never a blocking side-thread). Probe first, @@ -856,7 +1073,7 @@ impl SellerNodeRunner { last_liveness_seen = tokio::time::Instant::now(); last_liveness_seen_unix = now_unix(); // The full set was re-subscribed, so the open-pool half is back. - open_pool_degraded = false; + open_pool = None; manual_recovery_succeeded = true; eprintln!( "seller node RELAY-STALL recovery SUCCEEDED (attempts={attempts}, \ @@ -905,18 +1122,64 @@ impl SellerNodeRunner { "seller node RELAY-CLOSED: relay closed the {label} subscription \ (id={id}): {reason}" ); + + // An id we never registered cannot be a leg of ours going deaf, so it + // must not cost a reconnect — and escalating it did exactly that. Field + // seats open every cycle with a CLOSED for an unknown id; that forced a + // full recovery, and the recovery then re-closed the 1059 leg. A + // self-inflicted sawtooth on a socket that was never broken. + // + // The two ages are for the relay owner, who cannot see either from the + // server side. Our periodic wrap backfill uses `fetch_events`, which + // GENERATES its subscription id (`pool/mod.rs:815`) and runs on exactly + // this cadence, so a small backfill age implicates our own transient REQ; + // an auth age near the relay's NIP-42 TTL instead implicates a + // re-challenge sweep closing auth-scoped subs from the pre-expiry + // generation. + if !is_our_subscription(&id) { + eprintln!( + "seller node RELAY-CLOSED UNKNOWN-ID: id={id} was never in our \ + registry (ours: {OFFER_SUB_ID}, {AWARD_SUB_ID}, {WRAP_SUB_ID}, \ + {LIVENESS_PROBE_SUB_ID}); no recovery forced. \ + last_backfill={}s ago, last_nip42_auth={}s ago, \ + authed={nip42_authed}", + last_backfill_at.elapsed().as_secs(), + last_authenticated_at.elapsed().as_secs() + ); + continue; + } + + // Whether the offer REQ currently on the wire carries the un-pinned + // open-pool filter: either it was never dropped, or a re-arm attempt has + // just put it back. This is what decides whether a refusal can be ABOUT + // the un-pinned half — while degraded to targeted-only, it cannot. + let offer_req_carries_unpinned = self.claim_open_pool() + && open_pool.is_none_or(|state| state.attempt_pending); + // The offer REQ is the one subscription with a meaningful partial form: // drop the un-pinned open-pool filter and re-subscribe targeted-only, so // a relay that refuses the grouped REQ still leaves targeted claiming // alive rather than taking the whole offer leg down. - if id == OFFER_SUB_ID && self.claim_open_pool() && !open_pool_degraded { + if id == OFFER_SUB_ID && offer_req_carries_unpinned { + // A CLOSED landing while a re-arm attempt is on the wire IS that + // attempt's verdict, and it is what advances the backoff (#190). + let refused = open_pool.as_mut().map(|state| { + state.reject(); + (state.rejections, state.cooldown_ticks) + }); match self.subscribe_offers(None, false).await { Ok(()) => { - open_pool_degraded = true; + let (rejections, cooldown) = refused.unwrap_or_else(|| { + open_pool = Some(OpenPoolDegrade::new()); + (0, 0) + }); eprintln!( "seller node RELAY-CLOSED DEGRADE: offer subscription \ - re-armed TARGETED-ONLY (open-pool half dropped; \ - re-armed on the next successful recovery)" + re-armed TARGETED-ONLY (open-pool half dropped after \ + {rejections} consecutive refusal(s); the open-pool half \ + is retried on the \ + {wrap_backfill_interval_secs}s backfill tick, next \ + attempt in {cooldown} tick(s) — no reconnect required)" ); } Err(error) => { @@ -929,12 +1192,74 @@ impl SellerNodeRunner { )); } } - } else { - // Awards / 1059 / heartbeat have no partial form — repair them - // through the one paced recovery path so nothing re-dials the relay - // off-cadence. - forced_recovery = - Some(format!("relay CLOSED the {label} subscription: {reason}")); + continue; + } + + // #189 BELT. A `restricted:` CLOSED of a subscription whose filters all + // pin `#p` to our OWN pubkey, on a session that has authenticated, is the + // pre-auth REQ race — not a gate violation. It arrives mostly from the + // SDK's own background reconnect, which resubscribes on socket-up before + // AUTH exists (`relay/inner.rs:748-752`) and is not a path we can order + // from out here. So re-issue that ONE REQ, at most once per authenticated + // session (`insert` returns false the second time, and the budget is + // cleared only when a NEW session authenticates). The taxonomy is not + // softened: a genuine wrong-`#p` `restricted:` cannot reach this branch, + // because we author these filters from our own pubkey — and a second + // refusal falls through to the paced recovery below rather than looping. + let restricted = matches!( + nostr_sdk::prelude::MachineReadablePrefix::parse(&reason), + Some(nostr_sdk::prelude::MachineReadablePrefix::Restricted) + ); + if restricted + && nip42_authed + && subscription_pins_only_our_pubkey(&id, offer_req_carries_unpinned) + && restricted_retry_used.insert(id.clone()) + { + let overlap = nostr_sdk::Timestamp::from( + last_liveness_seen_unix + .saturating_sub(STALL_OVERLAP_MARGIN_SECS as i64) + .max(0) as u64, + ); + match self.subscribe_one(&id, Some(overlap)).await { + Ok(()) => { + eprintln!( + "seller node RELAY-CLOSED RETRY: the {label} \ + subscription pins #p to our OWN pubkey and this session \ + authenticated {}s ago, so `restricted:` here is the \ + pre-auth REQ race (#189) and not a gate violation; \ + re-subscribed ONCE with since={} overlap", + last_authenticated_at.elapsed().as_secs(), + overlap.as_secs() + ); + continue; + } + Err(error) => eprintln!( + "seller node RELAY-CLOSED retry failed ({error}); forcing \ + full recovery on the next heartbeat tick" + ), + } + } + + // Awards / 1059 / probe have no partial form — repair them through the + // one paced recovery path so nothing re-dials the relay off-cadence. + forced_recovery = + Some(format!("relay CLOSED the {label} subscription: {reason}")); + } + // An EOSE for the offer subscription while a re-arm attempt is on the wire is + // the relay ACCEPTING the grouped REQ. Acceptance is read from this response + // — which NIP-01 owes us — and never from our own send having succeeded: a + // REQ that left the socket proves nothing about whether the relay took it. + Ok(RelayPoolNotification::Message { + message: nostr_sdk::RelayMessage::EndOfStoredEvents(eose_id), + .. + }) if eose_id.to_string() == OFFER_SUB_ID => { + if open_pool.is_some_and(|state| state.attempt_pending) { + open_pool = None; + eprintln!( + "seller node RELAY-CLOSED RE-ARMED: the open-pool half of the \ + offer subscription is live again (the relay served the grouped \ + REQ); no reconnect was required" + ); } } Ok(_) => {} @@ -945,6 +1270,34 @@ impl SellerNodeRunner { } } } + // The relay's OWN notification stream, watched only to know whether the current + // socket has completed NIP-42. `Authenticated` never reaches the pool stream above + // (`relay/inner.rs:418` maps it to `None`), so this is the only way to see it. + relay_event = relay_notifications.recv() => { + use nostr_sdk::pool::RelayNotification; + match relay_event { + Ok(RelayNotification::Authenticated) => { + nip42_authed = true; + last_authenticated_at = tokio::time::Instant::now(); + // A newly authenticated session earns a fresh retry budget: the budget + // exists to bound retries WITHIN a session, not to spend one forever. + restricted_retry_used.clear(); + } + Ok(RelayNotification::AuthenticationFailed) => nip42_authed = false, + // A socket that went away takes its NIP-42 state with it — whatever comes + // back starts unauthenticated. + Ok(RelayNotification::RelayStatus { status }) + if status != nostr_sdk::prelude::RelayStatus::Connected => + { + nip42_authed = false; + } + Ok(RelayNotification::Shutdown) => nip42_authed = false, + Ok(_) => {} + // Lagging this stream costs only auth-state precision, and both stale + // readings are bounded (see the declaration). Never go deaf over it. + Err(_) => {} + } + } } } Ok(()) @@ -1112,20 +1465,56 @@ impl SellerNodeRunner { } } - /// Tear down the silently-dead connection and rebuild it: reconnect, re-run NIP-42 (the p-gated - /// kind-1059 resubscribe depends on it, same as boot), then resubscribe ALL filters with - /// `since = overlap`. Clears the stale subscriptions only AFTER a successful reconnect+auth so a - /// failed recovery never leaves the node reconnected-but-deaf. + /// Tear down the silently-dead connection and rebuild it: drop the stale registrations, reconnect, + /// re-run NIP-42 (the p-gated kind-1059 resubscribe depends on it, same as boot), then resubscribe + /// ALL filters with `since = overlap`. + /// + /// CLEARING BEFORE THE RECONNECT IS THE WHOLE OF #189. `RelayInner::post_connection` re-sends every + /// registered `REQ` as its first act on socket-up (`relay/inner.rs:748-752`), before that + /// connection has any NIP-42 state at all; auth only happens later, in the ingester + /// (`inner.rs:936`). mobee-relay evaluates its p-gate against the empty authed pubkey of that + /// unauthenticated session and answers `restricted:` — the PERMANENT prefix — where the truth is + /// the retryable `auth-required:`. nostr-sdk takes `restricted:` at its word and DELETES the + /// subscription (`inner.rs:1028` → `remove_subscription`), so the post-auth `resubscribe()` at + /// `inner.rs:941` cannot see it and never restores it. Carrying registrations across the socket + /// boundary therefore kills the kind-1059 money leg on every single recovery. With nothing + /// registered, that first resubscribe has nothing to send and the REQs go out below — after auth, + /// the same order boot has always had. async fn reconnect_and_resubscribe( &self, relay: &nostr_sdk::prelude::Relay, overlap_since: nostr_sdk::Timestamp, ) -> Result<(), NodeError> { - reconnect_and_authenticate(&self.client, relay) - .await - .map_err(|error| NodeError::Relay(format!("reconnect NIP-42 auth: {error}")))?; - self.client.unsubscribe_all().await; - self.subscribe_all(Some(overlap_since)).await + clear_subscription_registrations(&self.client, relay).await; + match reconnect_and_authenticate(&self.client, relay).await { + Ok(AuthWait::Authenticated) => self.subscribe_all(Some(overlap_since)).await, + Ok(AuthWait::NoChallenge) => { + // Same posture as boot: proceed, loudly. Auto-auth stays on, so a challenge raised on + // the REQ itself still authenticates — but a p-gated resubscribe issued before that + // completes is exactly the condition above, so say so rather than report a clean + // recovery. + eprintln!( + "seller node WARN: recovery saw no NIP-42 challenge within {CONNECT_WAIT:?}; \ + resubscribing anyway (auto-auth stays ON). p-gated kind-1059 receive may be \ + degraded until auth completes." + ); + self.subscribe_all(Some(overlap_since)).await + } + Err(error) => { + // The registrations are gone and the new socket never authenticated. Put them back: + // the SDK's own background reconnect is a real recovery path in the field (the run + // loop distinguishes it in the RESTORED line) and it can only restore subscriptions it + // still knows about. Re-registering makes a failed attempt no worse than not having + // tried; the next heartbeat tick retries the whole recovery. + if let Err(restore) = self.subscribe_all(Some(overlap_since)).await { + eprintln!( + "seller node WARN: subscriptions could not be restored after a failed \ + recovery ({restore}); the next heartbeat tick retries" + ); + } + Err(NodeError::Relay(format!("reconnect NIP-42 auth: {error}"))) + } + } } /// Consider one offer event: parse it, apply the money-safety gates, and — if admitted — journal @@ -2964,4 +3353,490 @@ mod tests { ); let _ = std::fs::remove_dir_all(&root); } + + // ---- #189 / #190 recovery teeth ------------------------------------------------------------ + // + // These drive the REAL paths against [`p_gate_relay_fixture`], which answers a `#p`-gated REQ + // from an unauthenticated session with the permanent-class `restricted:` prefix exactly as + // mobee-relay does. The nostr-relay-builder fixture used above cannot express this: it says + // `auth-required:`, which nostr-sdk keeps and restores by itself, so every ordering would pass. + + use crate::seller_node::p_gate_relay_fixture::{PGateRelay, ReqRecord, Verdict}; + + /// Generous enough that a slow box never flakes, short enough that a real failure fails fast. + const FIXTURE_WAIT: Duration = Duration::from_secs(15); + + /// A throwaway home per test. Unique per test name AND process so a parallel run never collides + /// on the exclusive home lock. + fn throwaway_root(label: &str) -> std::path::PathBuf { + let root = std::env::temp_dir().join(format!( + "mobee-recoveryfix-{label}-{}", + std::process::id() + )); + let _ = std::fs::remove_dir_all(&root); + root + } + + /// Boot a real runner against the fixture relay. + async fn boot_against( + root: &std::path::Path, + fixture: &PGateRelay, + claim_open_pool: bool, + ) -> SellerNodeRunner { + let mut home = crate::home::bootstrap(root).expect("bootstrap home"); + home.config.relay_url = fixture.url(); + home.config.seller = Some(seller_cfg(1, claim_open_pool)); + SellerNodeRunner::boot(home) + .await + .expect("boot the node against the fixture relay") + } + + /// The relay handle the recovery path takes. + async fn relay_handle(runner: &SellerNodeRunner) -> nostr_sdk::prelude::Relay { + runner + .client + .relays() + .await + .get(&RelayUrl::parse(&runner.relay_url).expect("relay url")) + .cloned() + .expect("relay handle") + } + + /// Every REQ that reached the relay before that session had completed NIP-42, on a filter the + /// relay p-gates. This set being non-empty IS #189. + fn p_gated_before_auth(reqs: &[ReqRecord]) -> Vec<&ReqRecord> { + reqs.iter() + .filter(|record| record.p_pinned && !record.authenticated) + .collect() + } + + /// Every REQ the relay refused with the permanent-class prefix — each one a subscription + /// nostr-sdk has deleted from its registry and will never restore. + fn permanently_removed(reqs: &[ReqRecord]) -> Vec<&ReqRecord> { + reqs.iter() + .filter(|record| { + matches!(&record.verdict, Verdict::Closed(reason) if reason.starts_with("restricted:")) + }) + .collect() + } + + /// TOOTH #189 (a) — THE ORDERING. A recovery whose AUTH lands well after the socket does must + /// still put every REQ on the wire AFTER NIP-42, leaving all four subscriptions live and nothing + /// permanently removed. + /// + /// The fixture withholds its challenge for 400ms, so the pre-auth window is wide and the outcome + /// is decided by ordering rather than luck. + /// + /// RED ON REVERT: move `clear_subscription_registrations` back to AFTER + /// `reconnect_and_authenticate` in `reconnect_and_resubscribe` and this goes red — the SDK's + /// `post_connection` resubscribe (`relay/inner.rs:748-752`) puts all three registered REQs on the + /// new socket immediately, the fixture refuses the p-gated ones `restricted:`, and both + /// assertions below fire. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn recovery_puts_no_p_gated_req_on_the_wire_before_nip42_completes() { + let fixture = PGateRelay::start(Duration::from_millis(400)).await; + let root = throwaway_root("order"); + let runner = boot_against(&root, &fixture, false).await; + let relay = relay_handle(&runner).await; + + runner.subscribe_all(None).await.expect("boot subscribe"); + assert!( + fixture + .wait_until(FIXTURE_WAIT, |reqs| { + [OFFER_SUB_ID, AWARD_SUB_ID, WRAP_SUB_ID] + .iter() + .all(|id| reqs.iter().any(|r| r.subscription_id == *id)) + }) + .await, + "harness check: the boot subscriptions must reach the relay before we induce a recovery" + ); + + runner + .reconnect_and_resubscribe(&relay, nostr_sdk::Timestamp::from(0)) + .await + .expect("recovery must succeed against a relay that authenticates"); + + assert!( + fixture + .wait_until(FIXTURE_WAIT, |reqs| { + [OFFER_SUB_ID, AWARD_SUB_ID, WRAP_SUB_ID] + .iter() + .all(|id| reqs.iter().filter(|r| r.subscription_id == *id).count() >= 2) + }) + .await, + "the recovery must re-issue every REQ" + ); + + // The fourth subscription: the liveness probe, which only exists on a session the relay is + // actually serving. Asserting it here is what makes "all four end live" true rather than + // three-plus-an-assumption. + assert!( + probe_relay_serves_our_reqs(&runner.client, runner.seller_pubkey, FIXTURE_WAIT).await, + "the liveness probe must answer on the recovered session" + ); + + let reqs = fixture.reqs().await; + assert!( + p_gated_before_auth(&reqs).is_empty(), + "a p-gated REQ reached the relay before NIP-42 completed — that is #189: {:?}", + p_gated_before_auth(&reqs) + ); + assert!( + permanently_removed(&reqs).is_empty(), + "the relay permanently removed a subscription (`restricted:`), so the money leg is dead \ + until the next backfill: {:?}", + permanently_removed(&reqs) + ); + for id in [OFFER_SUB_ID, AWARD_SUB_ID, WRAP_SUB_ID, LIVENESS_PROBE_SUB_ID] { + let last = fixture + .reqs_for(id) + .await + .pop() + .unwrap_or_else(|| panic!("no REQ recorded for {id}")); + assert_eq!( + last.verdict, + Verdict::Eose, + "{id} must end the recovery LIVE (served), not closed" + ); + } + + runner.client.disconnect().await; + let _ = std::fs::remove_dir_all(&root); + } + + /// TOOTH #189 (c) — the money leg survives REPEATED recoveries, not just the first. A one-shot + /// ordering fix that degrades after a few cycles would still pin settlement to the 300s backfill + /// on the reconnect-heavy hosts where this was found. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn wraps_subscription_survives_ten_consecutive_reconnects() { + let fixture = PGateRelay::start(Duration::from_millis(120)).await; + let root = throwaway_root("tenreconnects"); + let runner = boot_against(&root, &fixture, false).await; + let relay = relay_handle(&runner).await; + runner.subscribe_all(None).await.expect("boot subscribe"); + + for cycle in 1..=10 { + runner + .reconnect_and_resubscribe(&relay, nostr_sdk::Timestamp::from(0)) + .await + .unwrap_or_else(|error| panic!("recovery {cycle} failed: {error}")); + assert!( + fixture + .wait_until(FIXTURE_WAIT, |reqs| { + reqs.iter() + .filter(|r| r.subscription_id == WRAP_SUB_ID) + .count() + > cycle + }) + .await, + "recovery {cycle} did not re-issue the kind-1059 REQ" + ); + let wraps = fixture.reqs_for(WRAP_SUB_ID).await; + let last = wraps.last().expect("a wrap REQ exists"); + assert_eq!( + last.verdict, + Verdict::Eose, + "the kind-1059 money leg was refused on recovery {cycle}: {last:?}" + ); + assert!( + last.authenticated, + "recovery {cycle} sent the kind-1059 REQ on an unauthenticated session" + ); + } + + assert!( + p_gated_before_auth(&fixture.reqs().await).is_empty(), + "ten recoveries must not leak a single pre-auth p-gated REQ" + ); + + runner.client.disconnect().await; + let _ = std::fs::remove_dir_all(&root); + } + + /// TOOTH #189 (b) — THE TAXONOMY MUST NOT SOFTEN. A genuine gate violation — a REQ for somebody + /// else's `#p` — is still refused `restricted:`, still deleted by the SDK, and stays deleted. + /// + /// The belt cannot reach it by construction, and both halves of that are asserted: the id is not + /// one of ours, and `subscription_pins_only_our_pubkey` refuses it even if it were. Collapse the + /// belt's own-`#p` guard into a bare `restricted:` check and this goes red. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn a_genuine_wrong_p_restricted_stays_removed() { + let fixture = PGateRelay::start(Duration::from_millis(0)).await; + let root = throwaway_root("wrongp"); + let runner = boot_against(&root, &fixture, false).await; + let relay = relay_handle(&runner).await; + + let stranger = Keys::generate().public_key(); + let foreign_id = "someone-elses-gift-wraps"; + runner + .client + .subscribe_with_id( + nostr_sdk::SubscriptionId::new(foreign_id), + Filter::new().kind(Kind::GiftWrap).pubkey(stranger), + None, + ) + .await + .expect("send the offending REQ"); + + assert!( + fixture + .wait_until(FIXTURE_WAIT, |reqs| reqs + .iter() + .any(|r| r.subscription_id == foreign_id)) + .await, + "harness check: the offending REQ must reach the relay" + ); + let refusal = fixture + .reqs_for(foreign_id) + .await + .pop() + .expect("the offending REQ was recorded"); + assert!( + matches!(&refusal.verdict, Verdict::Closed(reason) if reason.starts_with("restricted:")), + "a wrong-#p REQ must still be refused `restricted:`, authenticated or not: {refusal:?}" + ); + assert!( + refusal.authenticated, + "harness check: this refusal must come from an AUTHENTICATED session, otherwise it \ + proves nothing about a genuine violation" + ); + + // The SDK deleted it, and nothing in the client puts it back. + assert!( + !relay + .subscriptions() + .await + .keys() + .any(|id| id.to_string() == foreign_id), + "`restricted:` must remain permanent-class: the subscription stays removed" + ); + assert!( + !is_our_subscription(foreign_id), + "the belt only ever considers our own subscription ids" + ); + assert!( + !subscription_pins_only_our_pubkey(foreign_id, false), + "and even then only ids whose every filter pins #p to our OWN pubkey" + ); + assert_eq!( + fixture.reqs_for(foreign_id).await.len(), + 1, + "the offending REQ must never be retried" + ); + + runner.client.disconnect().await; + let _ = std::fs::remove_dir_all(&root); + } + + /// One owned tick, for the loop teeth below. Both #190 loop teeth set the SAME value, so running + /// them in parallel cannot make them disagree. + const TEST_BACKFILL_SECS: &str = "1"; + + /// Drive the backfill tick fast enough to observe. This is the documented test-only seam; no + /// production path sets it. + fn use_fast_backfill_tick() { + unsafe { std::env::set_var(WRAP_BACKFILL_INTERVAL_ENV, TEST_BACKFILL_SECS) }; + } + + /// Offer REQs that carried the un-pinned open-pool filter — i.e. the grouped shape, armed. + fn grouped_offer_reqs(reqs: &[ReqRecord]) -> Vec<&ReqRecord> { + reqs.iter() + .filter(|record| record.subscription_id == OFFER_SUB_ID && record.has_unpinned_filter) + .collect() + } + + /// TOOTH #190 (a) + (b) — THE OWNED RE-ARM. Drop the open-pool half on a seat that is perfectly + /// healthy and never reconnects; the open-pool half must come back on its own within one owned + /// tick, and the targeted half must never be disturbed while that happens. + /// + /// This is the only proof Fix 2 has. The reported stuck specimen was withdrawn — every seat seen + /// degraded in the field was flapping on the #189 sawtooth — so the quiet-seat case is reasoned, + /// not observed, and this tooth is what stands in for the observation. + /// + /// RED ON REVERT: delete the `open_pool` block from the `wrap_backfill_tick` arm (the hookup) and + /// this goes red — nothing else re-arms without a recovery, and no recovery ever happens here. + /// A state-machine-only test would stay green under that revert, which is why this drives the + /// real loop. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn open_pool_rearms_on_an_owned_tick_without_any_reconnect() { + use_fast_backfill_tick(); + let fixture = PGateRelay::start(Duration::from_millis(0)).await; + let root = throwaway_root("rearm"); + let mut home = crate::home::bootstrap(&root).expect("bootstrap home"); + home.config.relay_url = fixture.url(); + home.config.seller = Some(seller_cfg(1, true)); + // The watchdog is off so nothing but the CLOSED under test can move the node. A recovery + // would re-arm the open-pool half for the wrong reason and the tooth would prove nothing. + home.config.seller_heartbeat.enabled = false; + let runner = SellerNodeRunner::boot(home).await.expect("boot runner"); + let loop_handle = tokio::spawn(async move { runner.run().await }); + + assert!( + fixture + .wait_until(FIXTURE_WAIT, |reqs| !grouped_offer_reqs(reqs).is_empty()) + .await, + "harness check: the seat must boot with the open-pool half ARMED, or there is nothing \ + to degrade" + ); + let connections_before = fixture.connections(); + let grouped_before = grouped_offer_reqs(&fixture.reqs().await).len(); + + // The degrade, exactly as the field sees it: an unsolicited CLOSED on a healthy socket. + fixture + .close_now( + OFFER_SUB_ID, + "restricted: p-gated events require #p matching your pubkey", + ) + .await; + + assert!( + fixture + .wait_until(FIXTURE_WAIT, |reqs| grouped_offer_reqs(reqs).len() + > grouped_before) + .await, + "the open-pool half was never re-armed: a healthy seat that degrades has no recovery to \ + wait for, which is #190" + ); + + assert_eq!( + fixture.connections(), + connections_before, + "the re-arm must not cost a reconnect — that dependency IS the bug" + ); + + // (b) The targeted half is never disturbed: every offer REQ ever sent, degraded or grouped, + // carries the `#p == self` filter. A degrade that dropped it would stop targeted claiming. + let offers = fixture.reqs_for(OFFER_SUB_ID).await; + assert!(offers.len() >= 3, "expected boot + degrade + re-arm REQs"); + for req in &offers { + assert!( + req.p_pinned, + "an offer REQ went out without the targeted #p filter: {req:?}" + ); + assert_eq!( + req.verdict, + Verdict::Eose, + "the relay refused an offer REQ it should have served: {req:?}" + ); + } + + loop_handle.abort(); + let _ = std::fs::remove_dir_all(&root); + } + + /// TOOTH #190 (c) — a relay that keeps refusing the open-pool half must cost a REQ per BACKOFF, + /// never a REQ per tick. With a 1s owned tick and refusals armed, the doubling schedule (attempt, + /// skip 1, skip 2, skip 4, …) has to hold the attempt count far below the tick count. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn repeated_open_pool_rejection_backs_off_and_never_hot_loops() { + use_fast_backfill_tick(); + let fixture = PGateRelay::start(Duration::from_millis(0)).await; + let root = throwaway_root("backoff"); + let mut home = crate::home::bootstrap(&root).expect("bootstrap home"); + home.config.relay_url = fixture.url(); + home.config.seller = Some(seller_cfg(1, true)); + home.config.seller_heartbeat.enabled = false; + let runner = SellerNodeRunner::boot(home).await.expect("boot runner"); + let loop_handle = tokio::spawn(async move { runner.run().await }); + + assert!( + fixture + .wait_until(FIXTURE_WAIT, |reqs| !grouped_offer_reqs(reqs).is_empty()) + .await, + "harness check: the seat must boot with the open-pool half armed" + ); + // Every grouped REQ from here on is refused; the targeted-only re-subscribe is still served. + fixture + .refuse_unpinned( + OFFER_SUB_ID, + 12, + "restricted: p-gated events require #p matching your pubkey", + ) + .await; + let grouped_before = grouped_offer_reqs(&fixture.reqs().await).len(); + + fixture + .close_now( + OFFER_SUB_ID, + "restricted: p-gated events require #p matching your pubkey", + ) + .await; + + // Twelve owned ticks. Un-backed-off, that is twelve attempts; the schedule allows at most + // four (t+0, +2, +5, +10). + tokio::time::sleep(Duration::from_secs(12)).await; + let attempts = grouped_offer_reqs(&fixture.reqs().await).len() - grouped_before; + assert!( + attempts >= 1, + "the re-arm must still be attempted — backoff is not abandonment" + ); + assert!( + attempts <= 5, + "the open-pool re-arm hot-looped: {attempts} attempts over ~12 owned ticks, which is a \ + REQ per tick against a relay that has refused every one" + ); + + // The targeted half kept working throughout — a backing-off re-arm must not starve claiming. + let served_targeted = fixture + .reqs_for(OFFER_SUB_ID) + .await + .into_iter() + .filter(|req| !req.has_unpinned_filter && req.verdict == Verdict::Eose) + .count(); + assert!( + served_targeted >= 1, + "the targeted-only offer subscription must stay live across the backoff" + ); + + loop_handle.abort(); + let _ = std::fs::remove_dir_all(&root); + } + + /// The backoff arithmetic itself: doubling, capped, and never zero after a refusal — a zero + /// cooldown at any rejection count would be the hot loop the loop tooth above forbids. + #[test] + fn open_pool_rearm_backoff_doubles_and_stays_capped() { + assert_eq!( + open_pool_rearm_cooldown_ticks(0), + 0, + "the first attempt after a degrade is not delayed" + ); + let schedule: Vec = (1..=8).map(open_pool_rearm_cooldown_ticks).collect(); + assert_eq!(schedule, vec![1, 2, 4, 8, 12, 12, 12, 12]); + for rejections in 1..=64 { + assert!( + open_pool_rearm_cooldown_ticks(rejections) >= 1, + "a refused re-arm must always cost at least one skipped tick" + ); + assert!( + open_pool_rearm_cooldown_ticks(rejections) <= 12, + "the backoff must stay capped so a re-arm is never abandoned" + ); + } + } + + /// The degrade state machine, including the case a timer-less design would park on: an attempt + /// that draws no verdict at all. Silence must advance the backoff, never wait forever. + #[test] + fn a_rearm_attempt_with_no_verdict_is_treated_as_a_refusal() { + let mut state = OpenPoolDegrade::new(); + assert_eq!(state.on_tick(), RearmStep::Attempt, "first tick attempts"); + assert!(state.attempt_pending); + + // No EOSE, no CLOSED — the relay simply said nothing. + assert_eq!( + state.on_tick(), + RearmStep::Wait, + "a pending attempt is not re-sent on top of itself" + ); + assert!( + !state.attempt_pending, + "silence must resolve the attempt rather than leave it pending forever" + ); + assert_eq!(state.rejections, 1); + assert_eq!(state.cooldown_ticks, 1); + + assert_eq!(state.on_tick(), RearmStep::Wait, "cooling down"); + assert_eq!(state.on_tick(), RearmStep::Attempt, "then attempting again"); + } } From 262b23afd8932ccaaf3c996d46312036c753a7d2 Mon Sep 17 00:00:00 2001 From: "worker:mobee-recoveryfix" Date: Mon, 27 Jul 2026 13:27:28 -0700 Subject: [PATCH 2/6] seller: pin the unknown-id CLOSED behaviour with a tooth that actually bites MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The downgrade shipped in the previous commit — a CLOSED for a subscription id we never registered is logged and does nothing else — had no test holding it down, so a refactor could restore the escalation silently. Two teeth now do. The behavioural one runs the real loop WITH the watchdog enabled, because with it disabled "no reconnect happened" is true even when the escalation is restored; and it waits long enough for a reconnect to COMPLETE against the fixture (~6s), because a shorter window reads a recovery in progress as a recovery that never happened. The first draft waited 4s and passed under revert — decorative. It now fails with "a CLOSED for an id we never registered forced a reconnect". The diagnostic itself moved into `unknown_close_diagnostic` so its content is pinned rather than drifting: the relay owner reads that line to separate our own transient `fetch_events` REQ (generated id, same 300s cadence) from a relay-side NIP-42 TTL sweep, and it is useless to them if either age goes missing. Also corrected an overclaim in the #190 re-arm tooth: with the watchdog off, "no reconnect" is the test's PREMISE, not an observation — and saying so is what makes the tooth's real claim legible, which is that the re-arm cannot have come from the recovery path because there is no recovery path in that test. Co-Authored-By: Claude Opus 5 --- .../src/seller_node/p_gate_relay_fixture.rs | 1 - crates/mobee-core/src/seller_node/run.rs | 157 +++++++++++++++++- 2 files changed, 149 insertions(+), 9 deletions(-) diff --git a/crates/mobee-core/src/seller_node/p_gate_relay_fixture.rs b/crates/mobee-core/src/seller_node/p_gate_relay_fixture.rs index 87d45e6e..1c1317b3 100644 --- a/crates/mobee-core/src/seller_node/p_gate_relay_fixture.rs +++ b/crates/mobee-core/src/seller_node/p_gate_relay_fixture.rs @@ -278,7 +278,6 @@ async fn serve_connection( ) .await; - transcript.lock().await.push(ReqRecord { subscription_id: subscription_id.clone(), authenticated: authed_pubkey.is_some(), diff --git a/crates/mobee-core/src/seller_node/run.rs b/crates/mobee-core/src/seller_node/run.rs index 61a2e6c2..3c797cce 100644 --- a/crates/mobee-core/src/seller_node/run.rs +++ b/crates/mobee-core/src/seller_node/run.rs @@ -291,6 +291,29 @@ fn is_our_subscription(id: &str) -> bool { ) } +/// The diagnostic for a `CLOSED` naming a subscription id we never registered. +/// +/// A function rather than an inline `eprintln!` because this line is field-facing: the relay owner +/// reads it to tell two hypotheses apart, and neither is visible from the server side. Our periodic +/// wrap backfill calls `fetch_events`, which GENERATES its subscription id (`pool/mod.rs:815`) and +/// runs on exactly the cadence these closes appear on — so a small `last_backfill` age implicates our +/// own transient REQ. A `last_nip42_auth` age near the relay's NIP-42 TTL instead implicates a +/// re-challenge sweep closing auth-scoped subscriptions from the pre-expiry generation. Being a +/// function, its content is pinned by a test instead of drifting silently. +fn unknown_close_diagnostic( + id: &str, + last_backfill_secs: u64, + last_nip42_auth_secs: u64, + authed: bool, +) -> String { + format!( + "seller node RELAY-CLOSED UNKNOWN-ID: id={id} was never in our registry (ours: \ + {OFFER_SUB_ID}, {AWARD_SUB_ID}, {WRAP_SUB_ID}, {LIVENESS_PROBE_SUB_ID}); no recovery \ + forced. last_backfill={last_backfill_secs}s ago, \ + last_nip42_auth={last_nip42_auth_secs}s ago, authed={authed}" + ) +} + /// Whether EVERY filter on this subscription pins `#p` to our own pubkey. /// /// This is the precondition for reading a `restricted:` CLOSED as the #189 pre-auth race instead of a @@ -1138,13 +1161,13 @@ impl SellerNodeRunner { // generation. if !is_our_subscription(&id) { eprintln!( - "seller node RELAY-CLOSED UNKNOWN-ID: id={id} was never in our \ - registry (ours: {OFFER_SUB_ID}, {AWARD_SUB_ID}, {WRAP_SUB_ID}, \ - {LIVENESS_PROBE_SUB_ID}); no recovery forced. \ - last_backfill={}s ago, last_nip42_auth={}s ago, \ - authed={nip42_authed}", - last_backfill_at.elapsed().as_secs(), - last_authenticated_at.elapsed().as_secs() + "{}", + unknown_close_diagnostic( + &id, + last_backfill_at.elapsed().as_secs(), + last_authenticated_at.elapsed().as_secs(), + nip42_authed, + ) ); continue; } @@ -3628,6 +3651,119 @@ mod tests { let _ = std::fs::remove_dir_all(&root); } + /// TOOTH — an unknown-id `CLOSED` is INERT beyond its log line. Escalating one cost a reconnect + /// per cycle on a socket that was never broken, and that reconnect is what re-closed the money + /// leg. Pinned here so a refactor cannot quietly restore the escalation. + /// + /// Two things make this tooth bite rather than decorate. The watchdog is ENABLED, so a forced + /// recovery has a tick to run on — with it off, "no reconnect happened" would be true even with + /// the escalation restored. And the window is long enough for a reconnect to COMPLETE against + /// this fixture (~6s), because a window shorter than that reads a recovery still in progress as + /// a recovery that never happened. A first draft of this tooth waited 4s and passed under + /// revert; the wait below returns early when the socket count moves, so the red path is fast and + /// only the green path pays the full window. + /// + /// RED ON REVERT: drop the `!is_our_subscription` early return so the unknown id falls through + /// to `forced_recovery`, and the socket-count assertion fires. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn an_unknown_id_closed_costs_no_reconnect_and_no_resubscribe() { + use_fast_backfill_tick(); + let fixture = PGateRelay::start(Duration::from_millis(0)).await; + let root = throwaway_root("unknownid"); + let mut home = crate::home::bootstrap(&root).expect("bootstrap home"); + home.config.relay_url = fixture.url(); + home.config.seller = Some(seller_cfg(1, false)); + home.config.seller_heartbeat.enabled = true; + home.config.seller_heartbeat.interval_secs = 1; + let runner = SellerNodeRunner::boot(home).await.expect("boot runner"); + let loop_handle = tokio::spawn(async move { runner.run().await }); + + assert!( + fixture + .wait_until(FIXTURE_WAIT, |reqs| reqs + .iter() + .any(|r| r.subscription_id == WRAP_SUB_ID)) + .await, + "harness check: the seat must be up before we close something it never registered" + ); + // The watchdog must be demonstrably live, or "no reconnect" is just a dead loop. + assert!( + fixture + .wait_until(FIXTURE_WAIT, |reqs| reqs + .iter() + .any(|r| r.subscription_id == LIVENESS_PROBE_SUB_ID)) + .await, + "harness check: the heartbeat watchdog must be ticking, otherwise a forced recovery \ + could not have fired even if one had been requested" + ); + let connections_before = fixture.connections(); + + let stranger_id = "some-subscription-we-never-registered"; + fixture + .close_now( + stranger_id, + "restricted: p-gated events require #p matching your pubkey", + ) + .await; + + let escalated = tokio::time::timeout(Duration::from_secs(20), async { + loop { + if fixture.connections() != connections_before { + return; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + }) + .await + .is_ok(); + assert!( + !escalated, + "a CLOSED for an id we never registered forced a reconnect — that is a reconnect per \ + cycle on a socket that was never broken" + ); + assert!( + fixture.reqs_for(stranger_id).await.is_empty(), + "we must never REQ a subscription id that was never ours" + ); + // Still alive and still watching: inert about the close, not inert about liveness. + let probes_before = fixture.reqs_for(LIVENESS_PROBE_SUB_ID).await.len(); + assert!( + fixture + .wait_until(FIXTURE_WAIT, |reqs| reqs + .iter() + .filter(|r| r.subscription_id == LIVENESS_PROBE_SUB_ID) + .count() + > probes_before) + .await, + "the node must keep probing after an unknown-id CLOSED" + ); + + loop_handle.abort(); + let _ = std::fs::remove_dir_all(&root); + } + + /// The unknown-id line is field-facing: the relay owner reads it to separate our own transient + /// `fetch_events` REQ from a relay-side auth-TTL sweep, so both ages have to actually be in it. + #[test] + fn the_unknown_close_diagnostic_carries_both_ages_and_the_auth_state() { + let line = unknown_close_diagnostic("deadbeef", 7, 301, true); + assert!(line.starts_with("seller node RELAY-CLOSED UNKNOWN-ID:")); + for expected in [ + "id=deadbeef", + "last_backfill=7s ago", + "last_nip42_auth=301s ago", + "authed=true", + "no recovery forced", + WRAP_SUB_ID, + ] { + assert!( + line.contains(expected), + "the unknown-id diagnostic must carry {expected:?}, or the relay owner cannot tell \ + the two hypotheses apart: {line}" + ); + } + } + /// One owned tick, for the loop teeth below. Both #190 loop teeth set the SAME value, so running /// them in parallel cannot make them disagree. const TEST_BACKFILL_SECS: &str = "1"; @@ -3698,10 +3834,15 @@ mod tests { wait for, which is #190" ); + // Not an observation but the test's PREMISE, and the reason it proves anything: with the + // watchdog off there is no recovery path in this process at all, so the re-arm above cannot + // have come from one. `open_pool_degraded = false` in the recovery-success arm — the only + // re-arm before this fix — is unreachable here. assert_eq!( fixture.connections(), connections_before, - "the re-arm must not cost a reconnect — that dependency IS the bug" + "harness check: nothing may reconnect in this test, or the re-arm could be the old \ + recovery path in disguise" ); // (b) The targeted half is never disturbed: every offer REQ ever sent, degraded or grouped, From 2b396045024708b6926ef15d8c740a851acb96f1 Mon Sep 17 00:00:00 2001 From: "worker:mobee-recoveryfix" Date: Mon, 27 Jul 2026 13:52:46 -0700 Subject: [PATCH 3/6] seller: make the recovery teeth survive a full-suite parallel run, and drop one that asserted the fix away MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both failures came out of `cargo test --workspace`; both targeted runs had been green, which is the point worth recording — a module-filtered green is not the same evidence. The ten-reconnect tooth asserted "zero pre-auth p-gated REQs across all ten cycles". That contradicts what the fix claims. The SDK's background reconnect resubscribes before AUTH and has no hook, and a recovery whose reconnect fails re-registers ON PURPOSE so the SDK can still rescue us — both put a pre-auth REQ on the wire by design, which is the entire reason the retry belt exists. The assertion was asserting the fix away, and it flaked under parallelism because the thing it forbade does happen. Removed. The per-cycle assertions are the real claim and they still bite: under revert, "the kind-1059 money leg was refused on recovery 1 ... authenticated: false". The zero-leak claim now lives only where it is true — the single controlled recovery in the ordering tooth. The same tooth also raced its own boot REQ against the first clear, and counted wrap REQs by cycle index, which a background reconnect's own REQ would misread as this cycle's. It now waits for the boot REQ to land and counts against the previous cycle. The wrong-#p tooth read the SDK's subscription registry once, racing the SDK's processing of the CLOSED against the fixture's record of the REQ — opposite sides of the socket. Bounded wait instead. Plus PR-BODY.md, and the clippy/rustfmt cleanups on the touched files: a let-chain collapse in the re-arm tick, the fixture's unused forced-close variant removed, and `filter_count` put to work asserting that both offer shapes ride ONE subscription. Co-Authored-By: Claude Opus 5 --- PR-BODY.md | 197 ++++++++++++++++++ .../src/seller_node/p_gate_relay_fixture.rs | 30 +-- crates/mobee-core/src/seller_node/run.rs | 156 +++++++++----- 3 files changed, 308 insertions(+), 75 deletions(-) create mode 100644 PR-BODY.md diff --git a/PR-BODY.md b/PR-BODY.md new file mode 100644 index 00000000..5b6fa784 --- /dev/null +++ b/PR-BODY.md @@ -0,0 +1,197 @@ +# Seller recovery: order the resubscribe after NIP-42, and give the open-pool re-arm an owned schedule + +Closes the client half of #189 and #190. + +Three behavioural changes. The first two are the charter's; the third is a defect found while +instrumenting the first, and it is called out separately below because it rides a two-fix PR. + +--- + +## 1. #189 — the recovery path let p-gated REQs out before NIP-42 + +**Mechanism, read out of the SDK source rather than inferred.** `RelayInner::post_connection` calls +`resubscribe()` as its first act on every socket-up (`relay/inner.rs:748-752`), before that +connection has any NIP-42 state; auth happens later, in the ingester (`:936`), which then +resubscribes a second time (`:941`). mobee-relay evaluates its p-gate against the empty authed +pubkey of that unauthenticated session and answers `restricted:` — the permanent prefix — where the +truth is the retryable `auth-required:`. + +That misclassification is not cosmetic, and the SDK's CLOSED taxonomy is why: + +| relay says | nostr-sdk does | consequence | +|---|---|---| +| `auth-required:` | `MarkAsClosed` (`:1023-1027`) | stays in the registry; the post-auth `resubscribe()` at `:941` **restores it automatically** | +| `restricted:` | `Remove` (`:1028`) | **deleted**; `:941` cannot see it and never restores it | +| *(empty reason)* | `Remove` (`:1029-1034`) | same permanent deletion | + +So carrying subscription registrations across a reconnect killed the kind-1059 money leg on every +recovery. (The relay half is owned elsewhere and is being patched independently — the +misclassification lives in the fork's open-read path, where a pending-auth connection collapses to +an anonymous empty pubkey before the p-gate runs. No relay code is touched here, and this fix does +not depend on that one landing.) **The fix is ordering:** the registrations are now dropped *before* the new socket comes +up, so that first resubscribe has nothing to send and the REQs go out after auth — the order boot +has always had. A failed reconnect re-registers them, because the SDK's own background reconnect is +a real recovery path in the field (the run loop distinguishes it in the `RESTORED` line) and can +only restore what it still knows about. + +**Plus the belt, which is not redundancy.** `RelayOptions::reconnect(true)` means the SDK also +reconnects on its own, entirely inside the SDK, with no hook — and that path will always resubscribe +pre-auth. So: a `restricted:` CLOSED of a subscription whose filters *all* pin `#p` to our own +pubkey, on a session that has authenticated, is re-issued once per authenticated session. + +**The CLOSED-prefix taxonomy is not softened.** `restricted:` stays permanent-class. A genuine +wrong-`#p` refusal cannot reach the branch — we author these filters from our own pubkey, so the +only way the relay can refuse one is by having no authenticated pubkey to compare against. A +subscription carrying any un-pinned filter is excluded, because there the refusal may genuinely be +about the un-pinned half. A second refusal falls through to the paced recovery rather than looping. +`subscription_pins_only_our_pubkey` carries the argument in a doc comment. + +NIP-42 state is tracked from a new `relay.notifications()` arm, because `Authenticated` never +becomes a pool notification (`relay/inner.rs:418` maps it to `None`) — the pool stream the run loop +already watches structurally cannot see it. Both stale readings of that flag are bounded and safe: +stale-false only declines a cheap retry and falls through to the paced recovery; stale-true spends +the one retry the session allows and then does the same. + +## 2. #190 — the open-pool re-arm waited on a trigger nothing guarantees + +The degrade re-armed only via `open_pool_degraded = false` in the recovery-success arm +(`run.rs:859`), so a seat that degrades and then stays healthy has no path back. + +The re-arm now rides the **wrap-backfill tick**. Not the heartbeat: the heartbeat is disableable by +config, and a repair must not depend on a tick that may never fire; the backfill tick is +unconditional. Acceptance is the relay's **EOSE on the offer subscription** — a response NIP-01 owes +us — never the fact that our send succeeded, because a REQ that left the socket proves nothing about +whether the relay took it. A refusal doubles a capped backoff, so a permanently refusing relay costs +one REQ per cap interval rather than one per tick, and an attempt that draws *no* verdict at all is +treated as a refusal so there is no timer-less park. + +`run.rs:859` stays — a full resubscribe genuinely does restore the grouped REQ. It was only ever the +*sole* path; this adds an owned one alongside it. + +**Scope, stated honestly.** This half is defence in depth, not a repair for an observed seat. The +reported stuck specimen was withdrawn: every seat seen degraded in the field was flapping on the +#189 sawtooth, not stuck. The quiet-seat case follows from `run.rs:859` but nobody has observed it. +The gap is structural, and it survives the #189 fix — which is why the owned schedule stays. + +## 3. An unknown-id CLOSED no longer forces a recovery + +Field seats open every cycle with a CLOSED naming a subscription id the client never registered. It +was escalated to a full recovery, and the recovery then re-closed the 1059 leg: **escalating an +unknown-id CLOSED cost a reconnect per cycle on a socket that was never broken.** A subscription id +we never registered cannot be a leg of ours going deaf, so it is now logged and nothing else. +Genuine deafness stays covered by the EOSE liveness probe, which is unchanged. + +**This is only safe because §2 exists, and they must land together.** Rocky's field data shows the +unknown-close-triggered recovery was also what re-armed the open-pool half — by accident. Removing +the escalation removes that accidental rescue, and the owned re-arm in §2 is what covers it now. +The no-reconnect tooth is what proves the replacement works without the reconnect. + +**A candidate for what the unknown id is — a lead, not a claim.** Our own periodic wrap backfill +calls `client.fetch_events`, which *generates* its subscription id (`pool/mod.rs:815`) and runs on +exactly the 300s cadence these closes appear on. The relay owner has since enumerated every periodic +mechanism on the relay side and none fits, which places the source outside the relay — client-side +or a fronting proxy's idle timeout — and makes our own transient REQ the leading candidate rather +than merely a plausible one. It is still not asserted here. The new log line carries the age of the +last backfill *and* the age of the last successful NIP-42 auth, and is the instrument that settles +it: a small backfill age implicates our own REQ, and the auth age bounds anything session-scoped. + +--- + +## The fixture, and a trap worth naming + +The teeth needed a relay this repo did not have. `nostr-relay-builder`'s NIP-42 read gate answers an +unauthenticated REQ with **`auth-required:`** (`local/inner.rs:961-989`) — which nostr-sdk keeps and +restores by itself. **Against that fixture every ordering passes and the tooth is decorative.** The +next person to test anything in this area will hit the same wall. + +`crates/mobee-core/src/seller_node/p_gate_relay_fixture.rs` speaks the deployed rule directly — a +`#p` filter is refused `restricted:` unless the session is authenticated as that very pubkey — which +covers both cases the teeth must tell apart: the pre-auth race (right `#p`, no auth yet) and a +genuine violation (someone else's `#p`, fully authed). It records every REQ with the session's auth +state *at arrival*, and counts sockets, so "no reconnect was required" is an observable rather than +an inference. + +## Teeth + +Nine new tests, all against the real paths. + +| tooth | what it pins | +|---|---| +| `recovery_puts_no_p_gated_req_on_the_wire_before_nip42_completes` | #189(a): AUTH held 400ms past the socket; all four subscriptions end live, zero permanent removals | +| `a_genuine_wrong_p_restricted_stays_removed` | #189(b): the taxonomy does not soften — wrong-`#p` is refused, deleted, never retried | +| `wraps_subscription_survives_ten_consecutive_reconnects` | #189(c): the money leg survives repeated recoveries, not just the first | +| `open_pool_rearms_on_an_owned_tick_without_any_reconnect` | #190(a)+(b): re-arm within one owned tick, targeted half never disturbed | +| `repeated_open_pool_rejection_backs_off_and_never_hot_loops` | #190(c): ≤5 attempts over ~12 owned ticks against a relay refusing every one | +| `open_pool_rearm_backoff_doubles_and_stays_capped` | the backoff arithmetic; never zero after a refusal | +| `a_rearm_attempt_with_no_verdict_is_treated_as_a_refusal` | silence advances the backoff instead of parking | +| `an_unknown_id_closed_costs_no_reconnect_and_no_resubscribe` | §3: inert about the close, not inert about liveness | +| `the_unknown_close_diagnostic_carries_both_ages_and_the_auth_state` | the field-facing line keeps what the relay owner needs | + +### Red-on-revert, strong form, `rc=101` each + +``` +1(a) move clear_subscription_registrations back after reconnect_and_authenticate + panicked: a p-gated REQ reached the relay before NIP-42 completed — that is #189: + [ReqRecord { subscription_id: "mobee-awards", authenticated: false, p_pinned: true, + verdict: Closed("restricted: p-gated events require #p matching your pubkey") }, ...] + +2(a) disable the open_pool block in the wrap-backfill arm (the hookup only) + panicked: the open-pool half was never re-armed: a healthy seat that degrades has no + recovery to wait for, which is #190 + +3 drop the !is_our_subscription early return + panicked: a CLOSED for an id we never registered forced a reconnect — that is a reconnect + per cycle on a socket that was never broken +``` + +**One of these teeth did not bite on the first try, and the fix is worth recording.** The unknown-id +tooth originally slept 4s and then asserted no reconnect had happened. A reconnect against this +fixture takes ~6s, so under revert the recovery was still in flight and the tooth passed — a +decorative tooth. It now waits for the socket count to move with a 20s ceiling, returning early on +the red path. Any "X did not happen" assertion has to outlast the time X takes to happen. + +## Gates + +- `cargo test --workspace` — **rc=0, 607 tests** (555 in the `mobee-core` suite). Baseline on the + untouched tip: rc=0 / **598 tests**, run on a frozen checkout of `origin/dev`. The delta is exactly + the nine teeth above. +- **Two of those teeth failed their first full-suite run, and both were the test's fault.** The + ten-reconnect tooth asserted "zero pre-auth p-gated REQs across all ten cycles" — which contradicts + what this fix claims, since the SDK's background reconnect and the deliberate re-register on a + failed recovery both put pre-auth REQs on the wire by design. That assertion was asserting the fix + away; it is gone, the per-cycle claim stands, and the bite was re-verified afterwards. The same + tooth also raced its own boot REQ against the first clear. Recorded because a module-filtered green + is not the same evidence as a full-suite-parallel green — both passed the former. +- The #169-arc teeth in this region all re-ran green and are unmodified by this diff: + `liveness_probe_answers_only_on_an_authenticated_session`, + `reconnect_reauthenticates_and_delivery_resumes_in_process`, + `wrap_backfill_cursor_clamps_to_the_oldest_unsettled_delivery_and_fails_closed`. +- **Per-file `rustfmt`, with the whole-tree skew measured rather than asserted.** Under this + toolchain the *untouched* `run.rs` from `origin/dev` already produces **83** `rustfmt --edition + 2024 --check` diffs, almost all import-ordering from the 2024 style edition. The set-difference of + normalised diff hunks between the pristine file and mine is **empty** — this diff contributes zero + formatting deviations. `p_gate_relay_fixture.rs` is `rc=0` outright, being new. CI runs neither + `fmt` nor `clippy` (`.github/workflows/ci.yml` is build+test), and whole-tree formatting is a + fleet-level call, not this slice's. Precedent: the crossmint slice's `PR-BODY.md`. +- **`clippy`** (`--no-default-features --features gateway,git-delivery,wallet --all-targets -D + warnings`): 57 pre-existing errors across the workspace. Two name `run.rs` — `:2377` and `:2447` — + both in test code this diff does not touch. Zero findings attributable to this change. + +## Field validation + +Pre-registered by the rocky fleet *before* the fix, which is the right order. + +**BEFORE**, three seats mid-flap: 42 recoveries, `attempts=1` on all, degrades 1:1 with recoveries, +inter-event gap exactly 300s in 38 of 39 intervals, deaf window ~10.7s per cycle, open-pool armed +window ≈0s. That 300s regularity is external confirmation of the `run.rs:859` mechanism: the +heartbeat tick servicing a queued `forced_recovery` is what paces the flap. + +**Expected after**: degrades → 0, recoveries → 0 or genuine, state FULL across several consecutive +300s windows. + +**One guard on reading that result.** The success signal here is the *absence* of `RELAY-CLOSED` +lines, and an absence is only evidence if the line could still have appeared. **This diff renames no +existing log line** — every field-facing line keeps its prefix, and the changed `DEGRADE` line keeps +`seller node RELAY-CLOSED DEGRADE:` while only its trailing parenthetical now states the real +schedule. The after-run should confirm those lines still exist in the new build before reading their +absence as success. diff --git a/crates/mobee-core/src/seller_node/p_gate_relay_fixture.rs b/crates/mobee-core/src/seller_node/p_gate_relay_fixture.rs index 1c1317b3..8007076a 100644 --- a/crates/mobee-core/src/seller_node/p_gate_relay_fixture.rs +++ b/crates/mobee-core/src/seller_node/p_gate_relay_fixture.rs @@ -23,7 +23,7 @@ use std::sync::Arc; use std::time::Duration; use futures_util::{SinkExt, StreamExt}; -use serde_json::{json, Value}; +use serde_json::{Value, json}; use tokio::sync::Mutex; /// What the relay answered a `REQ` with. @@ -52,14 +52,16 @@ pub(super) struct ReqRecord { pub verdict: Verdict, } -/// A refusal the test has armed for the relay to emit on the next matching `REQ`, once. +/// A refusal the test has armed, spent on the next `REQ` for this subscription that carries an +/// un-pinned filter. +/// +/// Matching only the un-pinned shape is load-bearing: a refusal armed for the open-pool half would +/// otherwise be spent on the targeted-only re-subscribe that immediately follows a rejection, and +/// the backoff tooth would be measuring the wrong REQ. #[derive(Debug, Clone)] struct ForcedClose { subscription_id: String, reason: String, - /// Match only a `REQ` that carries an un-pinned filter. Without this, a refusal armed for the - /// open-pool half would be spent on the targeted-only re-subscribe that follows a rejection. - require_unpinned: bool, } #[derive(Debug, Default)] @@ -141,16 +143,6 @@ impl PGateRelay { self.url.clone() } - /// Arm one refusal: the next `REQ` for `subscription_id` is answered `CLOSED` with `reason`, - /// whatever the session's auth state. Used to induce a degrade on an otherwise healthy seat. - pub(super) async fn force_close_once(&self, subscription_id: &str, reason: &str) { - self.controls.forced.lock().await.push_back(ForcedClose { - subscription_id: subscription_id.to_string(), - reason: reason.to_string(), - require_unpinned: false, - }); - } - /// Refuse the next `count` `REQ`s for `subscription_id` that carry an un-pinned filter — i.e. a /// relay that keeps rejecting the open-pool half while serving the targeted one. pub(super) async fn refuse_unpinned(&self, subscription_id: &str, count: usize, reason: &str) { @@ -159,7 +151,6 @@ impl PGateRelay { forced.push_back(ForcedClose { subscription_id: subscription_id.to_string(), reason: reason.to_string(), - require_unpinned: true, }); } } @@ -317,9 +308,10 @@ async fn decide( ) -> Verdict { let has_unpinned = pinned.iter().any(Option::is_none); let mut forced = controls.forced.lock().await; - if let Some(index) = forced.iter().position(|entry| { - entry.subscription_id == subscription_id && (!entry.require_unpinned || has_unpinned) - }) { + if let Some(index) = forced + .iter() + .position(|entry| entry.subscription_id == subscription_id && has_unpinned) + { let entry = forced.remove(index).expect("index just found"); return Verdict::Closed(entry.reason); } diff --git a/crates/mobee-core/src/seller_node/run.rs b/crates/mobee-core/src/seller_node/run.rs index 3c797cce..013d01fe 100644 --- a/crates/mobee-core/src/seller_node/run.rs +++ b/crates/mobee-core/src/seller_node/run.rs @@ -750,23 +750,22 @@ impl SellerNodeRunner { let mut relay_notifications = relay.notifications(); client.connect().await; client.wait_for_connection(CONNECT_WAIT).await; - let boot_auth = match relay_auth::wait_for_nip42_auth(&mut relay_notifications, CONNECT_WAIT) - .await - { - Ok(AuthWait::Authenticated) => { - eprintln!("seller node relay authenticated (NIP-42)"); - AuthWait::Authenticated - } - Ok(AuthWait::NoChallenge) => { - eprintln!( - "seller node WARN: no NIP-42 challenge within {CONNECT_WAIT:?}; proceeding \ + let boot_auth = + match relay_auth::wait_for_nip42_auth(&mut relay_notifications, CONNECT_WAIT).await { + Ok(AuthWait::Authenticated) => { + eprintln!("seller node relay authenticated (NIP-42)"); + AuthWait::Authenticated + } + Ok(AuthWait::NoChallenge) => { + eprintln!( + "seller node WARN: no NIP-42 challenge within {CONNECT_WAIT:?}; proceeding \ (auto-auth stays ON — a challenge on the REQ still authenticates). p-gated \ kind-1059 receive may be degraded until auth completes." - ); - AuthWait::NoChallenge - } - Err(error) => return Err(NodeError::Relay(format!("NIP-42 auth: {error}"))), - }; + ); + AuthWait::NoChallenge + } + Err(error) => return Err(NodeError::Relay(format!("NIP-42 auth: {error}"))), + }; let publisher = RelayPublisher::new(node.signer().clone(), client.clone(), &relay_url); @@ -850,11 +849,13 @@ impl SellerNodeRunner { .kind(Kind::Custom(JOB_AWARD_KIND)) .hashtag(crate::gateway::MOBEE_TAG) .pubkey(self.seller_pubkey), - WRAP_SUB_ID => Filter::new().kind(Kind::GiftWrap).pubkey(self.seller_pubkey), + WRAP_SUB_ID => Filter::new() + .kind(Kind::GiftWrap) + .pubkey(self.seller_pubkey), other => { return Err(NodeError::Relay(format!( "subscribe {other}: not one of ours" - ))) + ))); } }; let filter = match since { @@ -1005,29 +1006,29 @@ impl SellerNodeRunner { // heartbeat is disableable by config, and a repair must not depend on a tick that // may never fire. Acceptance is the relay's EOSE below — a response the protocol // owes us — never the fact that we managed to send the REQ. - if let Some(state) = open_pool.as_mut() { - if state.on_tick() == RearmStep::Attempt { - let overlap = nostr_sdk::Timestamp::from( - last_liveness_seen_unix - .saturating_sub(STALL_OVERLAP_MARGIN_SECS as i64) - .max(0) as u64, - ); - match self.subscribe_offers(Some(overlap), true).await { - Ok(()) => eprintln!( - "seller node RELAY-CLOSED RE-ARM: retrying the open-pool half of \ - the offer subscription (attempt after {} rejection(s), \ - since={} overlap); the relay's EOSE confirms it", - state.rejections, - overlap.as_secs() - ), - Err(error) => { - state.reject(); - eprintln!( - "seller node RELAY-CLOSED RE-ARM failed to send ({error}); \ - next attempt in {} backfill tick(s)", - state.cooldown_ticks - ); - } + if let Some(state) = open_pool.as_mut() + && state.on_tick() == RearmStep::Attempt + { + let overlap = nostr_sdk::Timestamp::from( + last_liveness_seen_unix + .saturating_sub(STALL_OVERLAP_MARGIN_SECS as i64) + .max(0) as u64, + ); + match self.subscribe_offers(Some(overlap), true).await { + Ok(()) => eprintln!( + "seller node RELAY-CLOSED RE-ARM: retrying the open-pool half of \ + the offer subscription (attempt after {} rejection(s), since={} \ + overlap); the relay's EOSE confirms it", + state.rejections, + overlap.as_secs() + ), + Err(error) => { + state.reject(); + eprintln!( + "seller node RELAY-CLOSED RE-ARM failed to send ({error}); next \ + attempt in {} backfill tick(s)", + state.cooldown_ticks + ); } } } @@ -3392,10 +3393,8 @@ mod tests { /// A throwaway home per test. Unique per test name AND process so a parallel run never collides /// on the exclusive home lock. fn throwaway_root(label: &str) -> std::path::PathBuf { - let root = std::env::temp_dir().join(format!( - "mobee-recoveryfix-{label}-{}", - std::process::id() - )); + let root = + std::env::temp_dir().join(format!("mobee-recoveryfix-{label}-{}", std::process::id())); let _ = std::fs::remove_dir_all(&root); root } @@ -3510,7 +3509,12 @@ mod tests { until the next backfill: {:?}", permanently_removed(&reqs) ); - for id in [OFFER_SUB_ID, AWARD_SUB_ID, WRAP_SUB_ID, LIVENESS_PROBE_SUB_ID] { + for id in [ + OFFER_SUB_ID, + AWARD_SUB_ID, + WRAP_SUB_ID, + LIVENESS_PROBE_SUB_ID, + ] { let last = fixture .reqs_for(id) .await @@ -3537,8 +3541,22 @@ mod tests { let runner = boot_against(&root, &fixture, false).await; let relay = relay_handle(&runner).await; runner.subscribe_all(None).await.expect("boot subscribe"); + // The boot REQ must have LANDED before the first recovery clears the registrations, or the + // clear races it and the first cycle measures a REQ that was never sent. + assert!( + fixture + .wait_until(FIXTURE_WAIT, |reqs| reqs + .iter() + .any(|r| r.subscription_id == WRAP_SUB_ID)) + .await, + "harness check: the boot kind-1059 REQ must reach the relay first" + ); for cycle in 1..=10 { + // Counted against the previous cycle rather than the cycle index: the SDK's background + // reconnect can issue a wrap REQ of its own at any point, and an absolute count would + // read that as this cycle's. + let before = fixture.reqs_for(WRAP_SUB_ID).await.len(); runner .reconnect_and_resubscribe(&relay, nostr_sdk::Timestamp::from(0)) .await @@ -3549,7 +3567,7 @@ mod tests { reqs.iter() .filter(|r| r.subscription_id == WRAP_SUB_ID) .count() - > cycle + > before }) .await, "recovery {cycle} did not re-issue the kind-1059 REQ" @@ -3567,10 +3585,14 @@ mod tests { ); } - assert!( - p_gated_before_auth(&fixture.reqs().await).is_empty(), - "ten recoveries must not leak a single pre-auth p-gated REQ" - ); + // Deliberately NOT asserting "zero pre-auth p-gated REQs across all ten cycles". That would + // contradict what the fix claims: the SDK's own background reconnect resubscribes before + // AUTH (`relay/inner.rs:748-752`) and has no hook, and a recovery whose reconnect fails + // re-registers on purpose so the SDK can still rescue us — both put a pre-auth REQ on the + // wire by design, which is exactly why the retry belt exists. The per-cycle assertions above + // are the real claim: after every recovery the money leg ends up SERVED on an AUTHENTICATED + // session. The blanket form flaked here under full-suite parallelism, and it deserved to. + // The single controlled recovery in the ordering tooth is where the zero-leak claim belongs. runner.client.disconnect().await; let _ = std::fs::remove_dir_all(&root); @@ -3624,13 +3646,27 @@ mod tests { proves nothing about a genuine violation" ); - // The SDK deleted it, and nothing in the client puts it back. + // The SDK deleted it, and nothing in the client puts it back. Waited for rather than read + // once: the relay recording the REQ and the SDK processing the CLOSED are different sides of + // the socket, so a bare read races the removal under load — which is a flaky test, not a + // finding. + let removed = tokio::time::timeout(FIXTURE_WAIT, async { + loop { + if !relay + .subscriptions() + .await + .keys() + .any(|id| id.to_string() == foreign_id) + { + return; + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + }) + .await + .is_ok(); assert!( - !relay - .subscriptions() - .await - .keys() - .any(|id| id.to_string() == foreign_id), + removed, "`restricted:` must remain permanent-class: the subscription stays removed" ); assert!( @@ -3859,6 +3895,14 @@ mod tests { Verdict::Eose, "the relay refused an offer REQ it should have served: {req:?}" ); + // Both shapes ride ONE subscription: grouped is targeted + un-pinned, degraded is + // targeted alone. A third shape would mean the two filters had been split across + // subscriptions, which delivers stored offers but never live ones. + let expected = if req.has_unpinned_filter { 2 } else { 1 }; + assert_eq!( + req.filter_count, expected, + "an offer REQ carried an unexpected filter count: {req:?}" + ); } loop_handle.abort(); From e396c8d2d2aceaf1de79da98ec49aff8df5cad7e Mon Sep 17 00:00:00 2001 From: "worker:mobee-recoveryfix" Date: Mon, 27 Jul 2026 13:56:21 -0700 Subject: [PATCH 4/6] docs: state the field contract as invariants, and make step 0 grep the prefix not the sentence MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cumulative recovery counts grow with uptime and are meaningless across a rebuild, so the contract is now the per-cycle invariant set (deg==rec 1:1, interval exactly 300s, attempts=1/outage ~10s, state TARGETED-ONLY) with pass = deg 0 + state FULL sustained across several consecutive 300s windows. Step 0 is the executable rename guard, with the one correction that matters: the DEGRADE line's trailing parenthetical CHANGED on purpose — #190 requires it to state the real re-arm schedule — so a check matching the old sentence aborts on a correct build. Verified against the binary: the prefix 'seller node RELAY-CLOSED DEGRADE:' is present, 're-armed on the next successful recovery' is gone. Measured literal counts are included, with the note that eprintln! splits a message at each placeholder, so a count describes placeholder placement rather than call sites and must be calibrated against a binary rather than remembered. Co-Authored-By: Claude Opus 5 --- PR-BODY.md | 61 +++++++++++++++++++++++++++++++++++++++++------------- 1 file changed, 47 insertions(+), 14 deletions(-) diff --git a/PR-BODY.md b/PR-BODY.md index 5b6fa784..b35c658b 100644 --- a/PR-BODY.md +++ b/PR-BODY.md @@ -181,17 +181,50 @@ the red path. Any "X did not happen" assertion has to outlast the time X takes t Pre-registered by the rocky fleet *before* the fix, which is the right order. -**BEFORE**, three seats mid-flap: 42 recoveries, `attempts=1` on all, degrades 1:1 with recoveries, -inter-event gap exactly 300s in 38 of 39 intervals, deaf window ~10.7s per cycle, open-pool armed -window ≈0s. That 300s regularity is external confirmation of the `run.rs:859` mechanism: the -heartbeat tick servicing a queued `forced_recovery` is what paces the flap. - -**Expected after**: degrades → 0, recoveries → 0 or genuine, state FULL across several consecutive -300s windows. - -**One guard on reading that result.** The success signal here is the *absence* of `RELAY-CLOSED` -lines, and an absence is only evidence if the line could still have appeared. **This diff renames no -existing log line** — every field-facing line keeps its prefix, and the changed `DEGRADE` line keeps -`seller node RELAY-CLOSED DEGRADE:` while only its trailing parenthetical now states the real -schedule. The after-run should confirm those lines still exist in the new build before reading their -absence as success. +**The contract is the invariant set, not cumulative counts.** Cumulative totals grow with uptime and +say nothing across a rebuild, so the BEFORE state is characterised by the shape of each cycle, +scoped per-run (the after-script asserts `boots=1`): + +| invariant (BEFORE, flapping) | expected AFTER | +|---|---| +| degrades == recoveries, 1:1 | degrades = 0 | +| inter-event interval **exactly 300s** | no periodic degrade/recovery cycle | +| `attempts=1`, outage ~10s per cycle | recoveries 0, or genuine and non-periodic | +| offer state **TARGETED-ONLY** | state **FULL**, sustained across several consecutive 300s windows | + +That 300s regularity is external confirmation of the `run.rs:859` mechanism: the heartbeat tick +servicing a queued `forced_recovery` is what paces the flap. + +### Step 0 — prove the lines can still appear, before reading their absence + +The success signal is the *absence* of `RELAY-CLOSED` lines, and an absence is only evidence if the +line could still have been emitted. The after-script therefore checks the literals in the shipped +binary first and aborts if any is missing. **Grep the stable PREFIX, not the full sentence**, and +here is why that is not pedantry: + +``` +$ strings | grep -cF 'seller node RELAY-CLOSED DEGRADE:' # 1 — anchor present +$ strings | grep -cF 're-armed on the next successful recovery' # 0 — GONE, on purpose +``` + +The `DEGRADE` line's trailing parenthetical **changed by design** — #190 requires it to state the +real re-arm schedule instead of the old "re-armed on the next successful recovery". A step 0 +matching that old sentence aborts on a correct build. The prefix is the contract; the tail is not. + +For the same reason, expected literal counts must be calibrated against a binary, not remembered. +Measured on this branch (`cargo build -p mobee --features wallet`, `strings | grep -cF`): + +``` +1 seller node RELAY-CLOSED: 1 seller node RELAY-CLOSED DEGRADE: +1 seller node RELAY-CLOSED degrade failed +1 RELAY-RECOVERY triggered 1 recovery SUCCEEDED +1 RESTORED via MANUAL 2 wrap backfill (periodic) +``` + +Note these are counts of `strings` lines, and `eprintln!` splits a message at each `{}` placeholder — +so a count is a property of where the placeholders fall, not of how many call sites exist. If the +pre-registered expectations were taken from a differently-built binary they will not match this one, +and the mismatch is not a rename. + +New literals this PR adds, all following the existing shape, none replacing anything: +`RELAY-CLOSED RETRY:`, `RELAY-CLOSED RE-ARM`, `RELAY-CLOSED RE-ARMED:`, `RELAY-CLOSED UNKNOWN-ID:`. From 5d835dcd05492d8254f18e884e71bfa0c899ed37 Mon Sep 17 00:00:00 2001 From: "worker:mobee-recoveryfix" Date: Mon, 27 Jul 2026 14:38:37 -0700 Subject: [PATCH 5/6] docs: pass on deg over a window, with a positive parse control MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two amendments from the fleet's self-test of the after-run, folded into the contract so the durable record carries them rather than a comment. state is the weak signal: on the known-bad build a seat sampled FULL, caught inside the few-second armed window between the resubscribe and the next close, so a point sample can read FULL against a 100% degrade rate. PASS now hangs on deg = 0 across >=3 consecutive 300s windows per seat, which cannot be caught mid-cycle; state is corroborating only. And the pass signal is a zero, which a regex that matches nothing produces just as readily. The script now requires a must-be-present line (wrap backfill) to parse non-zero before any zero counts — the denominator, without which a checker cannot tell 'nothing broken' from 'nothing examined'. Step 0 guards the vocabulary in the binary; this guards the parsing. Neither substitutes for the other. Co-Authored-By: Claude Opus 5 --- PR-BODY.md | 17 +++++++++++++++-- 1 file changed, 15 insertions(+), 2 deletions(-) diff --git a/PR-BODY.md b/PR-BODY.md index b35c658b..c71698ec 100644 --- a/PR-BODY.md +++ b/PR-BODY.md @@ -187,14 +187,27 @@ scoped per-run (the after-script asserts `boots=1`): | invariant (BEFORE, flapping) | expected AFTER | |---|---| -| degrades == recoveries, 1:1 | degrades = 0 | +| degrades == recoveries, 1:1 | **`deg = 0` across ≥3 consecutive 300s windows (~15 min) per seat** | | inter-event interval **exactly 300s** | no periodic degrade/recovery cycle | | `attempts=1`, outage ~10s per cycle | recoveries 0, or genuine and non-periodic | -| offer state **TARGETED-ONLY** | state **FULL**, sustained across several consecutive 300s windows | +| offer state **TARGETED-ONLY** | state **FULL** — corroborating only, see below | That 300s regularity is external confirmation of the `run.rs:859` mechanism: the heartbeat tick servicing a queued `forced_recovery` is what paces the flap. +**`state` is the weak signal; `deg` over a window is the strong one.** On the known-bad build a seat +sampled `FULL` — caught inside the few-second armed window between the resubscribe and the next +close. A point sample can therefore read `FULL` against a 100% degrade rate. `deg` counted over a +window cannot be caught mid-cycle, which is why the pass criterion hangs on it and not on `state`. + +**The pass signal is a zero, so the parser needs a positive control.** `deg = 0` is +indistinguishable from a regex that matches nothing, so the after-script requires a +must-be-present line (`wrap backfill`, the node's unconditional periodic line) to parse non-zero +before it will read any zero as meaningful — measured controls 41/35/43 on the current fleet. This +is the denominator: a checker that only reports failures cannot tell "nothing broken" from "nothing +examined". The binary-literal guard in step 0 protects the *vocabulary*; this protects the *parsing*. +Both are needed, and neither substitutes for the other. + ### Step 0 — prove the lines can still appear, before reading their absence The success signal is the *absence* of `RELAY-CLOSED` lines, and an absence is only evidence if the From 9d288be5c8237bd95ffc69b6f6dcef02f65aeae9 Mon Sep 17 00:00:00 2001 From: "worker:mobee-recoveryfix" Date: Mon, 27 Jul 2026 15:03:27 -0700 Subject: [PATCH 6/6] seller: drive the loop teeth on a LocalSet so they compile under acp+wallet MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Gate regression, mine. The three loop teeth `tokio::spawn` the runner, and under the `acp` feature the runner holds an AcpDriver whose std mpsc Receiver is !Sync — so `run()`'s future is !Send and the spawn fails to COMPILE. Nine errors on `cargo test -p mobee-core --features acp,gateway,git-delivery,wallet --lib --no-run`; base dev compiles that target rc=0. A LocalSet keeps the loop on this thread, which is also the truer shape: the node runs its loop as one task, not spread across a work-stealing pool. No production code moves — the defect was entirely in how the tests drive it. Why my gates missed it: none of them compiled acp and wallet TOGETHER. The workspace default has no acp; the -p acp run had no wallet. Each feature was covered and the combination was not, which is exactly the shape where a cfg-gated item stays invisible — a gate must compile every SHIPPED combo, and acp+wallet is the seller's real config. That command is now in the gate set. Also drops PR-BODY.md from the repo root: it is a process artifact that does not belong on dev, and removing it dissolves the fold-later bookkeeping — the live PR description is now the only copy of the contract. Co-Authored-By: Claude Opus 5 --- PR-BODY.md | 243 --------------- crates/mobee-core/src/seller_node/run.rs | 364 ++++++++++++----------- 2 files changed, 197 insertions(+), 410 deletions(-) delete mode 100644 PR-BODY.md diff --git a/PR-BODY.md b/PR-BODY.md deleted file mode 100644 index c71698ec..00000000 --- a/PR-BODY.md +++ /dev/null @@ -1,243 +0,0 @@ -# Seller recovery: order the resubscribe after NIP-42, and give the open-pool re-arm an owned schedule - -Closes the client half of #189 and #190. - -Three behavioural changes. The first two are the charter's; the third is a defect found while -instrumenting the first, and it is called out separately below because it rides a two-fix PR. - ---- - -## 1. #189 — the recovery path let p-gated REQs out before NIP-42 - -**Mechanism, read out of the SDK source rather than inferred.** `RelayInner::post_connection` calls -`resubscribe()` as its first act on every socket-up (`relay/inner.rs:748-752`), before that -connection has any NIP-42 state; auth happens later, in the ingester (`:936`), which then -resubscribes a second time (`:941`). mobee-relay evaluates its p-gate against the empty authed -pubkey of that unauthenticated session and answers `restricted:` — the permanent prefix — where the -truth is the retryable `auth-required:`. - -That misclassification is not cosmetic, and the SDK's CLOSED taxonomy is why: - -| relay says | nostr-sdk does | consequence | -|---|---|---| -| `auth-required:` | `MarkAsClosed` (`:1023-1027`) | stays in the registry; the post-auth `resubscribe()` at `:941` **restores it automatically** | -| `restricted:` | `Remove` (`:1028`) | **deleted**; `:941` cannot see it and never restores it | -| *(empty reason)* | `Remove` (`:1029-1034`) | same permanent deletion | - -So carrying subscription registrations across a reconnect killed the kind-1059 money leg on every -recovery. (The relay half is owned elsewhere and is being patched independently — the -misclassification lives in the fork's open-read path, where a pending-auth connection collapses to -an anonymous empty pubkey before the p-gate runs. No relay code is touched here, and this fix does -not depend on that one landing.) **The fix is ordering:** the registrations are now dropped *before* the new socket comes -up, so that first resubscribe has nothing to send and the REQs go out after auth — the order boot -has always had. A failed reconnect re-registers them, because the SDK's own background reconnect is -a real recovery path in the field (the run loop distinguishes it in the `RESTORED` line) and can -only restore what it still knows about. - -**Plus the belt, which is not redundancy.** `RelayOptions::reconnect(true)` means the SDK also -reconnects on its own, entirely inside the SDK, with no hook — and that path will always resubscribe -pre-auth. So: a `restricted:` CLOSED of a subscription whose filters *all* pin `#p` to our own -pubkey, on a session that has authenticated, is re-issued once per authenticated session. - -**The CLOSED-prefix taxonomy is not softened.** `restricted:` stays permanent-class. A genuine -wrong-`#p` refusal cannot reach the branch — we author these filters from our own pubkey, so the -only way the relay can refuse one is by having no authenticated pubkey to compare against. A -subscription carrying any un-pinned filter is excluded, because there the refusal may genuinely be -about the un-pinned half. A second refusal falls through to the paced recovery rather than looping. -`subscription_pins_only_our_pubkey` carries the argument in a doc comment. - -NIP-42 state is tracked from a new `relay.notifications()` arm, because `Authenticated` never -becomes a pool notification (`relay/inner.rs:418` maps it to `None`) — the pool stream the run loop -already watches structurally cannot see it. Both stale readings of that flag are bounded and safe: -stale-false only declines a cheap retry and falls through to the paced recovery; stale-true spends -the one retry the session allows and then does the same. - -## 2. #190 — the open-pool re-arm waited on a trigger nothing guarantees - -The degrade re-armed only via `open_pool_degraded = false` in the recovery-success arm -(`run.rs:859`), so a seat that degrades and then stays healthy has no path back. - -The re-arm now rides the **wrap-backfill tick**. Not the heartbeat: the heartbeat is disableable by -config, and a repair must not depend on a tick that may never fire; the backfill tick is -unconditional. Acceptance is the relay's **EOSE on the offer subscription** — a response NIP-01 owes -us — never the fact that our send succeeded, because a REQ that left the socket proves nothing about -whether the relay took it. A refusal doubles a capped backoff, so a permanently refusing relay costs -one REQ per cap interval rather than one per tick, and an attempt that draws *no* verdict at all is -treated as a refusal so there is no timer-less park. - -`run.rs:859` stays — a full resubscribe genuinely does restore the grouped REQ. It was only ever the -*sole* path; this adds an owned one alongside it. - -**Scope, stated honestly.** This half is defence in depth, not a repair for an observed seat. The -reported stuck specimen was withdrawn: every seat seen degraded in the field was flapping on the -#189 sawtooth, not stuck. The quiet-seat case follows from `run.rs:859` but nobody has observed it. -The gap is structural, and it survives the #189 fix — which is why the owned schedule stays. - -## 3. An unknown-id CLOSED no longer forces a recovery - -Field seats open every cycle with a CLOSED naming a subscription id the client never registered. It -was escalated to a full recovery, and the recovery then re-closed the 1059 leg: **escalating an -unknown-id CLOSED cost a reconnect per cycle on a socket that was never broken.** A subscription id -we never registered cannot be a leg of ours going deaf, so it is now logged and nothing else. -Genuine deafness stays covered by the EOSE liveness probe, which is unchanged. - -**This is only safe because §2 exists, and they must land together.** Rocky's field data shows the -unknown-close-triggered recovery was also what re-armed the open-pool half — by accident. Removing -the escalation removes that accidental rescue, and the owned re-arm in §2 is what covers it now. -The no-reconnect tooth is what proves the replacement works without the reconnect. - -**A candidate for what the unknown id is — a lead, not a claim.** Our own periodic wrap backfill -calls `client.fetch_events`, which *generates* its subscription id (`pool/mod.rs:815`) and runs on -exactly the 300s cadence these closes appear on. The relay owner has since enumerated every periodic -mechanism on the relay side and none fits, which places the source outside the relay — client-side -or a fronting proxy's idle timeout — and makes our own transient REQ the leading candidate rather -than merely a plausible one. It is still not asserted here. The new log line carries the age of the -last backfill *and* the age of the last successful NIP-42 auth, and is the instrument that settles -it: a small backfill age implicates our own REQ, and the auth age bounds anything session-scoped. - ---- - -## The fixture, and a trap worth naming - -The teeth needed a relay this repo did not have. `nostr-relay-builder`'s NIP-42 read gate answers an -unauthenticated REQ with **`auth-required:`** (`local/inner.rs:961-989`) — which nostr-sdk keeps and -restores by itself. **Against that fixture every ordering passes and the tooth is decorative.** The -next person to test anything in this area will hit the same wall. - -`crates/mobee-core/src/seller_node/p_gate_relay_fixture.rs` speaks the deployed rule directly — a -`#p` filter is refused `restricted:` unless the session is authenticated as that very pubkey — which -covers both cases the teeth must tell apart: the pre-auth race (right `#p`, no auth yet) and a -genuine violation (someone else's `#p`, fully authed). It records every REQ with the session's auth -state *at arrival*, and counts sockets, so "no reconnect was required" is an observable rather than -an inference. - -## Teeth - -Nine new tests, all against the real paths. - -| tooth | what it pins | -|---|---| -| `recovery_puts_no_p_gated_req_on_the_wire_before_nip42_completes` | #189(a): AUTH held 400ms past the socket; all four subscriptions end live, zero permanent removals | -| `a_genuine_wrong_p_restricted_stays_removed` | #189(b): the taxonomy does not soften — wrong-`#p` is refused, deleted, never retried | -| `wraps_subscription_survives_ten_consecutive_reconnects` | #189(c): the money leg survives repeated recoveries, not just the first | -| `open_pool_rearms_on_an_owned_tick_without_any_reconnect` | #190(a)+(b): re-arm within one owned tick, targeted half never disturbed | -| `repeated_open_pool_rejection_backs_off_and_never_hot_loops` | #190(c): ≤5 attempts over ~12 owned ticks against a relay refusing every one | -| `open_pool_rearm_backoff_doubles_and_stays_capped` | the backoff arithmetic; never zero after a refusal | -| `a_rearm_attempt_with_no_verdict_is_treated_as_a_refusal` | silence advances the backoff instead of parking | -| `an_unknown_id_closed_costs_no_reconnect_and_no_resubscribe` | §3: inert about the close, not inert about liveness | -| `the_unknown_close_diagnostic_carries_both_ages_and_the_auth_state` | the field-facing line keeps what the relay owner needs | - -### Red-on-revert, strong form, `rc=101` each - -``` -1(a) move clear_subscription_registrations back after reconnect_and_authenticate - panicked: a p-gated REQ reached the relay before NIP-42 completed — that is #189: - [ReqRecord { subscription_id: "mobee-awards", authenticated: false, p_pinned: true, - verdict: Closed("restricted: p-gated events require #p matching your pubkey") }, ...] - -2(a) disable the open_pool block in the wrap-backfill arm (the hookup only) - panicked: the open-pool half was never re-armed: a healthy seat that degrades has no - recovery to wait for, which is #190 - -3 drop the !is_our_subscription early return - panicked: a CLOSED for an id we never registered forced a reconnect — that is a reconnect - per cycle on a socket that was never broken -``` - -**One of these teeth did not bite on the first try, and the fix is worth recording.** The unknown-id -tooth originally slept 4s and then asserted no reconnect had happened. A reconnect against this -fixture takes ~6s, so under revert the recovery was still in flight and the tooth passed — a -decorative tooth. It now waits for the socket count to move with a 20s ceiling, returning early on -the red path. Any "X did not happen" assertion has to outlast the time X takes to happen. - -## Gates - -- `cargo test --workspace` — **rc=0, 607 tests** (555 in the `mobee-core` suite). Baseline on the - untouched tip: rc=0 / **598 tests**, run on a frozen checkout of `origin/dev`. The delta is exactly - the nine teeth above. -- **Two of those teeth failed their first full-suite run, and both were the test's fault.** The - ten-reconnect tooth asserted "zero pre-auth p-gated REQs across all ten cycles" — which contradicts - what this fix claims, since the SDK's background reconnect and the deliberate re-register on a - failed recovery both put pre-auth REQs on the wire by design. That assertion was asserting the fix - away; it is gone, the per-cycle claim stands, and the bite was re-verified afterwards. The same - tooth also raced its own boot REQ against the first clear. Recorded because a module-filtered green - is not the same evidence as a full-suite-parallel green — both passed the former. -- The #169-arc teeth in this region all re-ran green and are unmodified by this diff: - `liveness_probe_answers_only_on_an_authenticated_session`, - `reconnect_reauthenticates_and_delivery_resumes_in_process`, - `wrap_backfill_cursor_clamps_to_the_oldest_unsettled_delivery_and_fails_closed`. -- **Per-file `rustfmt`, with the whole-tree skew measured rather than asserted.** Under this - toolchain the *untouched* `run.rs` from `origin/dev` already produces **83** `rustfmt --edition - 2024 --check` diffs, almost all import-ordering from the 2024 style edition. The set-difference of - normalised diff hunks between the pristine file and mine is **empty** — this diff contributes zero - formatting deviations. `p_gate_relay_fixture.rs` is `rc=0` outright, being new. CI runs neither - `fmt` nor `clippy` (`.github/workflows/ci.yml` is build+test), and whole-tree formatting is a - fleet-level call, not this slice's. Precedent: the crossmint slice's `PR-BODY.md`. -- **`clippy`** (`--no-default-features --features gateway,git-delivery,wallet --all-targets -D - warnings`): 57 pre-existing errors across the workspace. Two name `run.rs` — `:2377` and `:2447` — - both in test code this diff does not touch. Zero findings attributable to this change. - -## Field validation - -Pre-registered by the rocky fleet *before* the fix, which is the right order. - -**The contract is the invariant set, not cumulative counts.** Cumulative totals grow with uptime and -say nothing across a rebuild, so the BEFORE state is characterised by the shape of each cycle, -scoped per-run (the after-script asserts `boots=1`): - -| invariant (BEFORE, flapping) | expected AFTER | -|---|---| -| degrades == recoveries, 1:1 | **`deg = 0` across ≥3 consecutive 300s windows (~15 min) per seat** | -| inter-event interval **exactly 300s** | no periodic degrade/recovery cycle | -| `attempts=1`, outage ~10s per cycle | recoveries 0, or genuine and non-periodic | -| offer state **TARGETED-ONLY** | state **FULL** — corroborating only, see below | - -That 300s regularity is external confirmation of the `run.rs:859` mechanism: the heartbeat tick -servicing a queued `forced_recovery` is what paces the flap. - -**`state` is the weak signal; `deg` over a window is the strong one.** On the known-bad build a seat -sampled `FULL` — caught inside the few-second armed window between the resubscribe and the next -close. A point sample can therefore read `FULL` against a 100% degrade rate. `deg` counted over a -window cannot be caught mid-cycle, which is why the pass criterion hangs on it and not on `state`. - -**The pass signal is a zero, so the parser needs a positive control.** `deg = 0` is -indistinguishable from a regex that matches nothing, so the after-script requires a -must-be-present line (`wrap backfill`, the node's unconditional periodic line) to parse non-zero -before it will read any zero as meaningful — measured controls 41/35/43 on the current fleet. This -is the denominator: a checker that only reports failures cannot tell "nothing broken" from "nothing -examined". The binary-literal guard in step 0 protects the *vocabulary*; this protects the *parsing*. -Both are needed, and neither substitutes for the other. - -### Step 0 — prove the lines can still appear, before reading their absence - -The success signal is the *absence* of `RELAY-CLOSED` lines, and an absence is only evidence if the -line could still have been emitted. The after-script therefore checks the literals in the shipped -binary first and aborts if any is missing. **Grep the stable PREFIX, not the full sentence**, and -here is why that is not pedantry: - -``` -$ strings | grep -cF 'seller node RELAY-CLOSED DEGRADE:' # 1 — anchor present -$ strings | grep -cF 're-armed on the next successful recovery' # 0 — GONE, on purpose -``` - -The `DEGRADE` line's trailing parenthetical **changed by design** — #190 requires it to state the -real re-arm schedule instead of the old "re-armed on the next successful recovery". A step 0 -matching that old sentence aborts on a correct build. The prefix is the contract; the tail is not. - -For the same reason, expected literal counts must be calibrated against a binary, not remembered. -Measured on this branch (`cargo build -p mobee --features wallet`, `strings | grep -cF`): - -``` -1 seller node RELAY-CLOSED: 1 seller node RELAY-CLOSED DEGRADE: -1 seller node RELAY-CLOSED degrade failed -1 RELAY-RECOVERY triggered 1 recovery SUCCEEDED -1 RESTORED via MANUAL 2 wrap backfill (periodic) -``` - -Note these are counts of `strings` lines, and `eprintln!` splits a message at each `{}` placeholder — -so a count is a property of where the placeholders fall, not of how many call sites exist. If the -pre-registered expectations were taken from a differently-built binary they will not match this one, -and the mismatch is not a rename. - -New literals this PR adds, all following the existing shape, none replacing anything: -`RELAY-CLOSED RETRY:`, `RELAY-CLOSED RE-ARM`, `RELAY-CLOSED RE-ARMED:`, `RELAY-CLOSED UNKNOWN-ID:`. diff --git a/crates/mobee-core/src/seller_node/run.rs b/crates/mobee-core/src/seller_node/run.rs index 013d01fe..b7420eff 100644 --- a/crates/mobee-core/src/seller_node/run.rs +++ b/crates/mobee-core/src/seller_node/run.rs @@ -3712,68 +3712,78 @@ mod tests { home.config.seller_heartbeat.enabled = true; home.config.seller_heartbeat.interval_secs = 1; let runner = SellerNodeRunner::boot(home).await.expect("boot runner"); - let loop_handle = tokio::spawn(async move { runner.run().await }); - - assert!( - fixture - .wait_until(FIXTURE_WAIT, |reqs| reqs - .iter() - .any(|r| r.subscription_id == WRAP_SUB_ID)) - .await, - "harness check: the seat must be up before we close something it never registered" - ); - // The watchdog must be demonstrably live, or "no reconnect" is just a dead loop. - assert!( - fixture - .wait_until(FIXTURE_WAIT, |reqs| reqs - .iter() - .any(|r| r.subscription_id == LIVENESS_PROBE_SUB_ID)) - .await, - "harness check: the heartbeat watchdog must be ticking, otherwise a forced recovery \ - could not have fired even if one had been requested" - ); - let connections_before = fixture.connections(); + // `run()` is NOT `Send` under the `acp` feature — the runner holds an `AcpDriver` + // whose std mpsc `Receiver` is `!Sync` — so `tokio::spawn` fails to COMPILE on the + // seller's real feature combo (`acp` + `wallet`), while compiling fine on the + // workspace default. A `LocalSet` keeps the loop on this thread, which is also the + // truer shape: the node runs its loop as one task, not spread across a pool. + let local = tokio::task::LocalSet::new(); + let loop_handle = local.spawn_local(async move { runner.run().await }); + local + .run_until(async { + + assert!( + fixture + .wait_until(FIXTURE_WAIT, |reqs| reqs + .iter() + .any(|r| r.subscription_id == WRAP_SUB_ID)) + .await, + "harness check: the seat must be up before we close something it never registered" + ); + // The watchdog must be demonstrably live, or "no reconnect" is just a dead loop. + assert!( + fixture + .wait_until(FIXTURE_WAIT, |reqs| reqs + .iter() + .any(|r| r.subscription_id == LIVENESS_PROBE_SUB_ID)) + .await, + "harness check: the heartbeat watchdog must be ticking, otherwise a forced recovery \ + could not have fired even if one had been requested" + ); + let connections_before = fixture.connections(); - let stranger_id = "some-subscription-we-never-registered"; - fixture - .close_now( - stranger_id, - "restricted: p-gated events require #p matching your pubkey", - ) - .await; + let stranger_id = "some-subscription-we-never-registered"; + fixture + .close_now( + stranger_id, + "restricted: p-gated events require #p matching your pubkey", + ) + .await; - let escalated = tokio::time::timeout(Duration::from_secs(20), async { - loop { - if fixture.connections() != connections_before { - return; - } - tokio::time::sleep(Duration::from_millis(50)).await; - } - }) - .await - .is_ok(); - assert!( - !escalated, - "a CLOSED for an id we never registered forced a reconnect — that is a reconnect per \ - cycle on a socket that was never broken" - ); - assert!( - fixture.reqs_for(stranger_id).await.is_empty(), - "we must never REQ a subscription id that was never ours" - ); - // Still alive and still watching: inert about the close, not inert about liveness. - let probes_before = fixture.reqs_for(LIVENESS_PROBE_SUB_ID).await.len(); - assert!( - fixture - .wait_until(FIXTURE_WAIT, |reqs| reqs - .iter() - .filter(|r| r.subscription_id == LIVENESS_PROBE_SUB_ID) - .count() - > probes_before) - .await, - "the node must keep probing after an unknown-id CLOSED" - ); + let escalated = tokio::time::timeout(Duration::from_secs(20), async { + loop { + if fixture.connections() != connections_before { + return; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + }) + .await + .is_ok(); + assert!( + !escalated, + "a CLOSED for an id we never registered forced a reconnect — that is a reconnect per \ + cycle on a socket that was never broken" + ); + assert!( + fixture.reqs_for(stranger_id).await.is_empty(), + "we must never REQ a subscription id that was never ours" + ); + // Still alive and still watching: inert about the close, not inert about liveness. + let probes_before = fixture.reqs_for(LIVENESS_PROBE_SUB_ID).await.len(); + assert!( + fixture + .wait_until(FIXTURE_WAIT, |reqs| reqs + .iter() + .filter(|r| r.subscription_id == LIVENESS_PROBE_SUB_ID) + .count() + > probes_before) + .await, + "the node must keep probing after an unknown-id CLOSED" + ); + }) + .await; loop_handle.abort(); let _ = std::fs::remove_dir_all(&root); } @@ -3841,70 +3851,80 @@ mod tests { // would re-arm the open-pool half for the wrong reason and the tooth would prove nothing. home.config.seller_heartbeat.enabled = false; let runner = SellerNodeRunner::boot(home).await.expect("boot runner"); - let loop_handle = tokio::spawn(async move { runner.run().await }); - - assert!( - fixture - .wait_until(FIXTURE_WAIT, |reqs| !grouped_offer_reqs(reqs).is_empty()) - .await, - "harness check: the seat must boot with the open-pool half ARMED, or there is nothing \ - to degrade" - ); - let connections_before = fixture.connections(); - let grouped_before = grouped_offer_reqs(&fixture.reqs().await).len(); - - // The degrade, exactly as the field sees it: an unsolicited CLOSED on a healthy socket. - fixture - .close_now( - OFFER_SUB_ID, - "restricted: p-gated events require #p matching your pubkey", - ) - .await; + // `run()` is NOT `Send` under the `acp` feature — the runner holds an `AcpDriver` + // whose std mpsc `Receiver` is `!Sync` — so `tokio::spawn` fails to COMPILE on the + // seller's real feature combo (`acp` + `wallet`), while compiling fine on the + // workspace default. A `LocalSet` keeps the loop on this thread, which is also the + // truer shape: the node runs its loop as one task, not spread across a pool. + let local = tokio::task::LocalSet::new(); + let loop_handle = local.spawn_local(async move { runner.run().await }); + local + .run_until(async { + + assert!( + fixture + .wait_until(FIXTURE_WAIT, |reqs| !grouped_offer_reqs(reqs).is_empty()) + .await, + "harness check: the seat must boot with the open-pool half ARMED, or there is nothing \ + to degrade" + ); + let connections_before = fixture.connections(); + let grouped_before = grouped_offer_reqs(&fixture.reqs().await).len(); - assert!( - fixture - .wait_until(FIXTURE_WAIT, |reqs| grouped_offer_reqs(reqs).len() - > grouped_before) - .await, - "the open-pool half was never re-armed: a healthy seat that degrades has no recovery to \ - wait for, which is #190" - ); + // The degrade, exactly as the field sees it: an unsolicited CLOSED on a healthy socket. + fixture + .close_now( + OFFER_SUB_ID, + "restricted: p-gated events require #p matching your pubkey", + ) + .await; + + assert!( + fixture + .wait_until(FIXTURE_WAIT, |reqs| grouped_offer_reqs(reqs).len() + > grouped_before) + .await, + "the open-pool half was never re-armed: a healthy seat that degrades has no recovery to \ + wait for, which is #190" + ); - // Not an observation but the test's PREMISE, and the reason it proves anything: with the - // watchdog off there is no recovery path in this process at all, so the re-arm above cannot - // have come from one. `open_pool_degraded = false` in the recovery-success arm — the only - // re-arm before this fix — is unreachable here. - assert_eq!( - fixture.connections(), - connections_before, - "harness check: nothing may reconnect in this test, or the re-arm could be the old \ - recovery path in disguise" - ); + // Not an observation but the test's PREMISE, and the reason it proves anything: with the + // watchdog off there is no recovery path in this process at all, so the re-arm above cannot + // have come from one. `open_pool_degraded = false` in the recovery-success arm — the only + // re-arm before this fix — is unreachable here. + assert_eq!( + fixture.connections(), + connections_before, + "harness check: nothing may reconnect in this test, or the re-arm could be the old \ + recovery path in disguise" + ); - // (b) The targeted half is never disturbed: every offer REQ ever sent, degraded or grouped, - // carries the `#p == self` filter. A degrade that dropped it would stop targeted claiming. - let offers = fixture.reqs_for(OFFER_SUB_ID).await; - assert!(offers.len() >= 3, "expected boot + degrade + re-arm REQs"); - for req in &offers { - assert!( - req.p_pinned, - "an offer REQ went out without the targeted #p filter: {req:?}" - ); - assert_eq!( - req.verdict, - Verdict::Eose, - "the relay refused an offer REQ it should have served: {req:?}" - ); - // Both shapes ride ONE subscription: grouped is targeted + un-pinned, degraded is - // targeted alone. A third shape would mean the two filters had been split across - // subscriptions, which delivers stored offers but never live ones. - let expected = if req.has_unpinned_filter { 2 } else { 1 }; - assert_eq!( - req.filter_count, expected, - "an offer REQ carried an unexpected filter count: {req:?}" - ); - } + // (b) The targeted half is never disturbed: every offer REQ ever sent, degraded or grouped, + // carries the `#p == self` filter. A degrade that dropped it would stop targeted claiming. + let offers = fixture.reqs_for(OFFER_SUB_ID).await; + assert!(offers.len() >= 3, "expected boot + degrade + re-arm REQs"); + for req in &offers { + assert!( + req.p_pinned, + "an offer REQ went out without the targeted #p filter: {req:?}" + ); + assert_eq!( + req.verdict, + Verdict::Eose, + "the relay refused an offer REQ it should have served: {req:?}" + ); + // Both shapes ride ONE subscription: grouped is targeted + un-pinned, degraded is + // targeted alone. A third shape would mean the two filters had been split across + // subscriptions, which delivers stored offers but never live ones. + let expected = if req.has_unpinned_filter { 2 } else { 1 }; + assert_eq!( + req.filter_count, expected, + "an offer REQ carried an unexpected filter count: {req:?}" + ); + } + }) + .await; loop_handle.abort(); let _ = std::fs::remove_dir_all(&root); } @@ -3922,57 +3942,67 @@ mod tests { home.config.seller = Some(seller_cfg(1, true)); home.config.seller_heartbeat.enabled = false; let runner = SellerNodeRunner::boot(home).await.expect("boot runner"); - let loop_handle = tokio::spawn(async move { runner.run().await }); - - assert!( - fixture - .wait_until(FIXTURE_WAIT, |reqs| !grouped_offer_reqs(reqs).is_empty()) - .await, - "harness check: the seat must boot with the open-pool half armed" - ); - // Every grouped REQ from here on is refused; the targeted-only re-subscribe is still served. - fixture - .refuse_unpinned( - OFFER_SUB_ID, - 12, - "restricted: p-gated events require #p matching your pubkey", - ) - .await; - let grouped_before = grouped_offer_reqs(&fixture.reqs().await).len(); - - fixture - .close_now( - OFFER_SUB_ID, - "restricted: p-gated events require #p matching your pubkey", - ) - .await; + // `run()` is NOT `Send` under the `acp` feature — the runner holds an `AcpDriver` + // whose std mpsc `Receiver` is `!Sync` — so `tokio::spawn` fails to COMPILE on the + // seller's real feature combo (`acp` + `wallet`), while compiling fine on the + // workspace default. A `LocalSet` keeps the loop on this thread, which is also the + // truer shape: the node runs its loop as one task, not spread across a pool. + let local = tokio::task::LocalSet::new(); + let loop_handle = local.spawn_local(async move { runner.run().await }); + local + .run_until(async { + + assert!( + fixture + .wait_until(FIXTURE_WAIT, |reqs| !grouped_offer_reqs(reqs).is_empty()) + .await, + "harness check: the seat must boot with the open-pool half armed" + ); + // Every grouped REQ from here on is refused; the targeted-only re-subscribe is still served. + fixture + .refuse_unpinned( + OFFER_SUB_ID, + 12, + "restricted: p-gated events require #p matching your pubkey", + ) + .await; + let grouped_before = grouped_offer_reqs(&fixture.reqs().await).len(); - // Twelve owned ticks. Un-backed-off, that is twelve attempts; the schedule allows at most - // four (t+0, +2, +5, +10). - tokio::time::sleep(Duration::from_secs(12)).await; - let attempts = grouped_offer_reqs(&fixture.reqs().await).len() - grouped_before; - assert!( - attempts >= 1, - "the re-arm must still be attempted — backoff is not abandonment" - ); - assert!( - attempts <= 5, - "the open-pool re-arm hot-looped: {attempts} attempts over ~12 owned ticks, which is a \ - REQ per tick against a relay that has refused every one" - ); + fixture + .close_now( + OFFER_SUB_ID, + "restricted: p-gated events require #p matching your pubkey", + ) + .await; + + // Twelve owned ticks. Un-backed-off, that is twelve attempts; the schedule allows at most + // four (t+0, +2, +5, +10). + tokio::time::sleep(Duration::from_secs(12)).await; + let attempts = grouped_offer_reqs(&fixture.reqs().await).len() - grouped_before; + assert!( + attempts >= 1, + "the re-arm must still be attempted — backoff is not abandonment" + ); + assert!( + attempts <= 5, + "the open-pool re-arm hot-looped: {attempts} attempts over ~12 owned ticks, which is a \ + REQ per tick against a relay that has refused every one" + ); - // The targeted half kept working throughout — a backing-off re-arm must not starve claiming. - let served_targeted = fixture - .reqs_for(OFFER_SUB_ID) - .await - .into_iter() - .filter(|req| !req.has_unpinned_filter && req.verdict == Verdict::Eose) - .count(); - assert!( - served_targeted >= 1, - "the targeted-only offer subscription must stay live across the backoff" - ); + // The targeted half kept working throughout — a backing-off re-arm must not starve claiming. + let served_targeted = fixture + .reqs_for(OFFER_SUB_ID) + .await + .into_iter() + .filter(|req| !req.has_unpinned_filter && req.verdict == Verdict::Eose) + .count(); + assert!( + served_targeted >= 1, + "the targeted-only offer subscription must stay live across the backoff" + ); + }) + .await; loop_handle.abort(); let _ = std::fs::remove_dir_all(&root); }