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..8007076a --- /dev/null +++ b/crates/mobee-core/src/seller_node/p_gate_relay_fixture.rs @@ -0,0 +1,347 @@ +//! 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::{Value, json}; +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, 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, +} + +#[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() + } + + /// 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(), + }); + } + } + + /// 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 && 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..b7420eff 100644 --- a/crates/mobee-core/src/seller_node/run.rs +++ b/crates/mobee-core/src/seller_node/run.rs @@ -281,6 +281,136 @@ 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 + ) +} + +/// 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 +/// 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 +532,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 +689,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 +750,22 @@ 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." - ), - Err(error) => return Err(NodeError::Relay(format!("NIP-42 auth: {error}"))), - } + 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 +775,7 @@ impl SellerNodeRunner { publisher, relay_url, seller_pubkey, + boot_auth, }) } @@ -658,24 +826,46 @@ 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 +959,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 +1000,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() + && 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 +1097,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 +1146,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!( + "{}", + unknown_close_diagnostic( + &id, + last_backfill_at.elapsed().as_secs(), + last_authenticated_at.elapsed().as_secs(), + nip42_authed, + ) + ); + 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 +1216,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 +1294,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 +1489,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 +3377,681 @@ 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"); + // 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 + .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() + > before + }) + .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" + ); + } + + // 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); + } + + /// 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. 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!( + removed, + "`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); + } + + /// 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"); + // `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 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); + } + + /// 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"; + + /// 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"); + // `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(); + + // 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" + ); + + // (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); + } + + /// 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"); + // `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(); + + 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" + ); + + }) + .await; + 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"); + } }