diff --git a/docs/EVENT_ROUTING.md b/docs/EVENT_ROUTING.md index e105a180..282bd1a6 100644 --- a/docs/EVENT_ROUTING.md +++ b/docs/EVENT_ROUTING.md @@ -6,6 +6,41 @@ How Nostr events become actions and side effects. - Source: `src/app.rs:run` - Steps: POW check → signature verify → recency guard → NIP-59 unwrap → parse `mostro_core::Message` → inner verify → `check_trade_index` → dispatch. +## The Inbox Subscription +- Source: `src/inbox.rs` +- Every trade message reaches Mostro over a single long-lived subscription, sent by the event loop (`app::run` / `app::run_cashu`) right after it takes its notification stream — a receiver created after the REQ would miss the relay's `EOSE` — with a stable id (`InboxSubscription`) so that later frames can be attributed to it. Its filter is p-tagged to the node, restricted to the configured transport's event kind, and carries `limit(0)`: only live traffic is wanted. +- `run` consumes the whole notification stream, not just events. `ClientNotification::Message` carries the relay control plane and goes to `InboxKeeper`; `ClientNotification::Shutdown` ends the loop. + +### Recovering a lost ear +A relay can end a subscription at any time by sending `CLOSED`. The nostr-sdk removes the subscription for almost every reason prefix, and a removed subscription is never re-REQ'd, not even after a reconnect — so without handling, one frame from one relay leaves the daemon running, connected, and unable to receive anything. + +The exceptions are `auth-required` and `rate-limited`, which the SDK only *marks*: the subscription stays registered and the SDK re-sends the REQ itself, after the NIP-42 round-trip in the first case and on the next reconnect in the second. The keeper stands down on those two rather than racing it, and whatever the SDK does not get to — a `rate-limited` closure on a connection that never drops — falls to the watchdog below, which is the right pace for a relay that has just asked to be left alone. + +Two mechanisms keep the subscription alive: + +- `InboxKeeper::on_relay_message` reacts to a `CLOSED` naming the inbox by re-sending the REQ to that relay. +- `check_inbox_health`, run every 30 seconds by `job_inbox_watchdog`, audits each connected read relay and re-subscribes any that is no longer serving the inbox. This covers the losses that produce no frame the loop can see: a notification channel that dropped messages under lag, a REQ that failed to go out, a relay added after startup. + +Both go through `resubscribe_relay`, which owns the pacing so neither can bypass it: one per-relay budget in `InboxHealth`, immediate first retry, doubling to a five-minute ceiling, reset when the relay answers with `EOSE`. Because the doublings start well below the 30-second audit interval, a relay that merely lost the inbox is re-subscribed on the next pass, while one that refuses it on principle converges on the five-minute figure rather than drawing a REQ every 30 seconds indefinitely. + +Health is judged by the subscription, never by traffic volume: an instance with no trades in flight is legitimately silent. + +A relay counts as serving the inbox only once **it** has said so, by answering the REQ with an `EOSE` that the event loop recorded in `InboxHealth`. The SDK's own subscription map is not evidence — it records what Mostro sent, so a relay that holds the connection open and quietly drops the REQ still appears subscribed there, and the daemon would resume order timeouts while deaf. The same rule means a relay re-subscribed during an audit does not count until it answers, which costs one interval before recovery is declared and errs toward keeping the timeout clock frozen slightly longer than strictly needed. + +### NIP-42 +The daemon and price clients are built with a `SignerAuthenticator` over the node's keys (`src/util.rs:connect_nostr`). Without it a relay that gates reads behind authentication answers the REQ with `CLOSED "auth-required: …"`, which the SDK treats as permanent. The AUTH event is bound to the relay's challenge and URL, so it cannot be replayed elsewhere. + +### Messages lost while blind are not recovered +`accept_event` rejects anything whose `created_at` is older than ten seconds. A message sent while the inbox was down is therefore already too old to be accepted by the time the subscription returns, and re-subscribing with `since` instead of `limit(0)` would not change that. Whoever sent it has to send it again. + +Because those messages are lost rather than delayed, order timeouts cannot be trusted while the inbox is down — a user who answered on time would look silent. `job_cancel_orders` therefore skips its tick entirely unless an audit has confirmed the daemon is listening (`InboxHealth::is_confirmed_listening`): no slash, no refund, no republish. Startup counts as unconfirmed, since the daemon subscribes before the watchdog's first pass. + +Once the inbox recovers, each order is credited the downtime **it** waited through. `InboxHealth` keeps the wall-clock windows during which the node was deaf; `blind_seconds_since(taken_at)` intersects them with the order's own wait. An order already waiting when a relay went quiet is owed all of that outage; one taken after it ended is owed nothing. The query widens its window by `max_blind_seconds` so no eligible order is missed, and the exact per-order figure decides. + +The credit has to be per order rather than one global allowance: a single figure either under-credits an order that waited through the whole outage or hands the same credit to one taken long afterwards. + +Deferring cannot be unconditional, though. The same pass that slashes a bond is the one that releases it and the one that cancels the seller's hold invoice, so waiting forever on a permanently broken inbox would leave escrows encumbered until CLTV expiry and honest takers' bonds locked indefinitely. After three hours without a confirmed inbox, timed-out orders are unwound anyway — but blamelessly: bonds are released rather than settled (`bond::release_on_timeout_without_slashing`), and the downtime credit is skipped, since by then every waiting order would be owed more than its deadline and nothing would ever be unwound. + ## Dispatch - Router: `src/app.rs:handle_message_action` - Maps `Action` → module function under `src/app/*`. @@ -26,20 +61,57 @@ How Nostr events become actions and side effects. ```mermaid sequenceDiagram participant Relay as Nostr Relay - participant Loop as app.rs (run) + participant EventLoop as app.rs (run) + participant Keeper as InboxKeeper participant Router as handle_message_action participant Mod as app/* participant DB as DB participant LND as LND - Relay-->>Loop: GiftWrap Event - Loop->>Loop: POW + verify + freshness - Loop->>Loop: unwrap + parse Message - Loop->>DB: check_trade_index - Loop->>Router: dispatch(Action) + Relay-->>EventLoop: GiftWrap Event + EventLoop->>EventLoop: POW + verify + freshness + EventLoop->>EventLoop: unwrap + parse Message + EventLoop->>DB: check_trade_index + EventLoop->>Router: dispatch(Action) Router->>Mod: handler(...) par side-effects Mod->>DB: read/write Mod->>LND: hold/settle/cancel/pay end + + Relay-->>EventLoop: CLOSED (inbox subscription) + EventLoop->>Keeper: on_relay_message + Keeper->>Relay: REQ (same subscription id) +``` + +The watchdog runs on its own schedule, independently of the loop above: + +```mermaid +sequenceDiagram + participant Job as job_inbox_watchdog + participant Relay as connected read relays + participant Health as InboxHealth + participant Timeouts as job_cancel_orders + + loop every 30s + Job->>Relay: still serving the inbox subscription? + alt not serving it + Job->>Relay: REQ (same subscription id) + end + alt none were serving it + Job->>Health: Blind + else at least one was + Job->>Health: Listening + end + end + + loop every 60s + Timeouts->>Health: is_confirmed_listening? + alt not confirmed + Timeouts->>Timeouts: skip the tick + else confirmed + Timeouts->>Health: blind_seconds_since(taken_at) per order + Timeouts->>Timeouts: run, each order credited its own downtime + end + end ``` diff --git a/docs/STARTUP_AND_CONFIG.md b/docs/STARTUP_AND_CONFIG.md index 2efad905..2b195169 100644 --- a/docs/STARTUP_AND_CONFIG.md +++ b/docs/STARTUP_AND_CONFIG.md @@ -100,6 +100,12 @@ Configuration is loaded from `~/.mostro/settings.toml` (template: `settings.tpl. - `relays` (Vec): List of Nostr relay URLs for event broadcasting - Default: `['ws://localhost:7000']` - Note: At least one relay required + - Relays that require NIP-42 authentication are supported: Mostro answers the + challenge with its own key. No configuration is needed. + - A relay that ends Mostro's inbox subscription is re-subscribed + automatically, and a watchdog audits the subscription every 30 seconds. If + no relay is serving it, the log carries `Mostro inbox is BLIND` and order + timeouts are held until it recovers. See `docs/EVENT_ROUTING.md`. **Lightning** (`src/config/types.rs:27-46`): - `lnd_cert_file` (String): Path to LND TLS certificate diff --git a/docs/TRANSPORT_V2_SPEC.md b/docs/TRANSPORT_V2_SPEC.md index 39c01ca5..de18519f 100644 --- a/docs/TRANSPORT_V2_SPEC.md +++ b/docs/TRANSPORT_V2_SPEC.md @@ -196,7 +196,7 @@ Minimal daemon integration; **zero handler changes** by design: - `[expiration] dm_days` knob (default 30) in `ExpirationSettings` and the `get_expiration_timestamp_for_kind` fallback (`DM_EVENT_KIND = 14` in `src/config/constants.rs`). -- `src/main.rs` — subscription filter uses `transport.event_kind()`. +- `src/inbox.rs` — subscription filter uses `transport.event_kind()`. - `src/app.rs` — event loop accepts only the configured kind and unwraps via `unwrap_incoming()`. - `src/util.rs send_dm()` — wraps via `wrap_message_with(transport, …)`; diff --git a/src/app.rs b/src/app.rs index 2170c575..a0d07971 100644 --- a/src/app.rs +++ b/src/app.rs @@ -50,6 +50,7 @@ use crate::app::trade_pubkey::trade_pubkey_action; // Core functionality imports use crate::db::add_new_user; use crate::db::is_user_present; +use crate::inbox::{InboxKeeper, InboxSubscription}; use crate::lightning::LndConnector; use crate::spam_gate::SpamGate; use crate::util::enqueue_cant_do_msg; @@ -437,6 +438,11 @@ async fn accept_event( Some((action, message, unwrapped)) } +/// How long to wait before re-attaching to the notification stream after it +/// ended without a shutdown. Long enough that a persistent failure cannot burn +/// a core, short enough that a transient one costs no meaningful deaf time. +const NOTIFICATION_STREAM_RETRY: std::time::Duration = std::time::Duration::from_secs(1); + /// Shared post-dispatch error handling (identical in both loops). A handler /// `Err` is downcast to a `MostroError` and turned into the right reply /// (`manage_errors`) or logged (`warning_msg`); `Ok` is a no-op. Factored out @@ -503,37 +509,69 @@ pub async fn run(ctx: AppContext, ln_client: &mut LndConnector) -> Result<()> { // gate is meaningless for v1 (gift wraps are signed by throwaway keys). let pow_first_contact = ctx.settings().mostro.effective_pow_first_contact(); let gate = gate_for(accepted_kind.as_u16() == crate::config::constants::DM_EVENT_KIND); + // The inbox identity is derived here rather than passed around (see + // `crate::inbox`). + let subscription = InboxSubscription::new(my_keys.public_key(), accepted_kind); + let keeper = InboxKeeper::new(subscription.clone()); + let mut subscribed = false; loop { let mut notifications = client.notifications(); + // The REQ goes out only once this stream exists. A notification + // receiver never sees what was delivered before it was created, so + // subscribing any earlier throws away the relay's EOSE — and every + // event that lands while the rest of the daemon is still booting. + if !subscribed { + subscription.subscribe(client).await?; + subscribed = true; + } + while let Some(notification) = notifications.next().await { - if let ClientNotification::Event { event, .. } = notification { - let Some((action, message, unwrapped)) = accept_event( - &ctx, - &event, - my_keys, - pow, - pow_first_contact, - accepted_kind, - gate, - ) - .await - else { - continue; - }; - let result = handle_message_action( - &action, - message.clone(), - &unwrapped, - my_keys, - ln_client, - &ctx, - ) - .await; - finalize_dispatch(result, message, unwrapped, &action).await; + match notification { + ClientNotification::Event { event, .. } => { + let Some((action, message, unwrapped)) = accept_event( + &ctx, + &event, + my_keys, + pow, + pow_first_contact, + accepted_kind, + gate, + ) + .await + else { + continue; + }; + let result = handle_message_action( + &action, + message.clone(), + &unwrapped, + my_keys, + ln_client, + &ctx, + ) + .await; + finalize_dispatch(result, message, unwrapped, &action).await; + } + ClientNotification::Message { relay_url, message } => { + keeper.on_relay_message(client, &relay_url, &message).await; + } + ClientNotification::Shutdown => return Ok(()), } } + + // The stream ended without a `Shutdown` frame. That frame can be + // missed — the SDK's notification channel silently drops messages when + // the consumer falls behind — and after a shutdown `notifications()` + // hands back an empty stream, so re-taking it unconditionally spins + // this loop at full tilt. Leave when the client is done, and pace the + // retry otherwise. + if client.is_shutdown() { + return Ok(()); + } + tracing::warn!("Nostr notification stream ended without a shutdown; re-attaching"); + tokio::time::sleep(NOTIFICATION_STREAM_RETRY).await; } } @@ -554,30 +592,57 @@ pub async fn run_cashu(ctx: AppContext) -> Result<()> { let accepted_kind = ctx.settings().mostro.transport.event_kind(); let pow_first_contact = ctx.settings().mostro.effective_pow_first_contact(); let gate = gate_for(accepted_kind.as_u16() == crate::config::constants::DM_EVENT_KIND); + let subscription = InboxSubscription::new(my_keys.public_key(), accepted_kind); + let keeper = InboxKeeper::new(subscription.clone()); + let mut subscribed = false; loop { let mut notifications = client.notifications(); + // Subscribe only once the stream exists — see `run`. + if !subscribed { + subscription.subscribe(client).await?; + subscribed = true; + } + while let Some(notification) = notifications.next().await { - if let ClientNotification::Event { event, .. } = notification { - let Some((action, message, unwrapped)) = accept_event( - &ctx, - &event, - my_keys, - pow, - pow_first_contact, - accepted_kind, - gate, - ) - .await - else { - continue; - }; - let result = - dispatch_cashu(&action, message.clone(), &unwrapped, my_keys, &ctx).await; - finalize_dispatch(result, message, unwrapped, &action).await; + match notification { + ClientNotification::Event { event, .. } => { + let Some((action, message, unwrapped)) = accept_event( + &ctx, + &event, + my_keys, + pow, + pow_first_contact, + accepted_kind, + gate, + ) + .await + else { + continue; + }; + let result = + dispatch_cashu(&action, message.clone(), &unwrapped, my_keys, &ctx).await; + finalize_dispatch(result, message, unwrapped, &action).await; + } + ClientNotification::Message { relay_url, message } => { + keeper.on_relay_message(client, &relay_url, &message).await; + } + ClientNotification::Shutdown => return Ok(()), } } + + // The stream ended without a `Shutdown` frame. That frame can be + // missed — the SDK's notification channel silently drops messages when + // the consumer falls behind — and after a shutdown `notifications()` + // hands back an empty stream, so re-taking it unconditionally spins + // this loop at full tilt. Leave when the client is done, and pace the + // retry otherwise. + if client.is_shutdown() { + return Ok(()); + } + tracing::warn!("Nostr notification stream ended without a shutdown; re-attaching"); + tokio::time::sleep(NOTIFICATION_STREAM_RETRY).await; } } diff --git a/src/app/bond/mod.rs b/src/app/bond/mod.rs index c8fe9423..5ae4bd47 100644 --- a/src/app/bond/mod.rs +++ b/src/app/bond/mod.rs @@ -29,8 +29,8 @@ pub use model::Bond; pub use payout::{add_bond_invoice_action, run_bond_payout_cycle}; pub use slash::{ apply_bond_resolution, extract_bond_resolution, notify_bond_slashed, - reconcile_stranded_range_maker_bonds, resolve_range_maker_bond_at_close, - resolve_range_maker_bond_at_close_or_warn, slash_or_release_on_timeout, - validate_bond_resolution, + reconcile_stranded_range_maker_bonds, release_on_timeout_without_slashing, + resolve_range_maker_bond_at_close, resolve_range_maker_bond_at_close_or_warn, + slash_or_release_on_timeout, validate_bond_resolution, }; pub use types::{BondRole, BondSlashReason, BondState}; diff --git a/src/app/bond/slash.rs b/src/app/bond/slash.rs index 42908f42..2f074073 100644 --- a/src/app/bond/slash.rs +++ b/src/app/bond/slash.rs @@ -482,6 +482,19 @@ async fn release_on_timeout(pool: &Pool, order_id: Uuid, republishes: bo } } +/// Resolve a timed-out order's bonds without holding anyone responsible. +/// +/// Used when the timeout cannot be attributed to the user: the daemon's Nostr +/// inbox has been unreachable long enough that waiting any longer would keep +/// hold invoices encumbered until CLTV expiry (see `job_cancel_orders`). The +/// order still has to be unwound, but a silence the node could not hear is not +/// evidence of abandonment, so every bond involved is released rather than +/// settled — the republish-vs-cancel distinction is honoured exactly as in +/// [`slash_or_release_on_timeout`]. +pub async fn release_on_timeout_without_slashing(pool: &Pool, order: &Order) { + release_on_timeout(pool, order.id, order_republishes_on_timeout(order)).await; +} + pub async fn slash_or_release_on_timeout( pool: &Pool, ln_client: &mut L, diff --git a/src/config/mod.rs b/src/config/mod.rs index d141f552..24a71ee9 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -37,7 +37,7 @@ pub static NOSTR_CLIENT: OnceLock = OnceLock::new(); /// /// Kept separate from [`NOSTR_CLIENT`] so `verify_subscriptions(true)` (and /// the REQ `limit` it enforces) applies only to price fetches — not to the -/// daemon's long-lived `.limit(0)` inbox subscription in `main.rs`, where +/// daemon's long-lived `.limit(0)` inbox subscription (`crate::inbox`), where /// pre-EOSE filter verification would reject matching trade messages /// (hermeme, PR #841). pub static PRICE_NOSTR_CLIENT: OnceLock = OnceLock::new(); diff --git a/src/db.rs b/src/db.rs index 1b3fd117..e9eef59c 100644 --- a/src/db.rs +++ b/src/db.rs @@ -10,6 +10,7 @@ use std::collections::HashSet; use std::fs::{set_permissions, Permissions}; use std::path::Path; use std::sync::Arc; +use std::time::Duration; use uuid::Uuid; // Constants for status filtering used across restore session functions @@ -533,10 +534,23 @@ pub async fn find_order_by_date(pool: &SqlitePool) -> Result, MostroE Ok(order) } -pub async fn find_order_by_seconds(pool: &SqlitePool) -> Result, MostroError> { +/// Orders whose waiting deadline has passed and are therefore candidates for +/// the timeout job. +/// +/// `grace` widens the window by the **most** any order could be owed for time +/// the daemon spent unable to receive anything (see +/// [`crate::inbox::InboxHealth::max_blind_seconds`]), so that no order the +/// caller may still have to spare is filtered out here. It deliberately +/// over-selects: what a given order is actually owed depends on when it began +/// waiting, which this query cannot express, so the caller applies the exact +/// per-order figure to the rows returned. +pub async fn find_order_by_seconds( + pool: &SqlitePool, + grace: Duration, +) -> Result, MostroError> { let mostro_settings = Settings::get_mostro(); let exp_seconds = mostro_settings.expiration_seconds as u64; - let expire_time = Timestamp::now() - exp_seconds; + let expire_time = Timestamp::now() - exp_seconds - grace.as_secs(); let order = sqlx::query_as::<_, Order>( r#" SELECT * @@ -5028,7 +5042,7 @@ mod migration_and_query_tests { ) .await; - let stale = find_order_by_seconds(&pool).await.unwrap(); + let stale = find_order_by_seconds(&pool, Duration::ZERO).await.unwrap(); assert_eq!(stale.len(), 1); assert_eq!(stale[0].id, stale_id); } @@ -5072,13 +5086,63 @@ mod migration_and_query_tests { let after = Order::by_id(&pool, id).await.unwrap().unwrap(); assert!(after.taken_at > 0, "take must persist taken_at"); - let stale = find_order_by_seconds(&pool).await.unwrap(); + let stale = find_order_by_seconds(&pool, Duration::ZERO).await.unwrap(); assert!( stale.is_empty(), "a just-taken order must not be timeout-cancel eligible" ); } + #[tokio::test] + async fn find_order_by_seconds_grace_spares_orders_the_daemon_could_not_hear() { + init_test_settings(); + let pool = migrated_pool().await; + + let exp_seconds = Settings::get_mostro().expiration_seconds as i64; + // Taken one minute past the deadline: late by wall time, and the + // caller is about to say the node was deaf for longer than that. + let taken_at = Timestamp::now().as_secs() as i64 - exp_seconds - 60; + let order_id = Uuid::new_v4(); + insert_order( + &pool, + order_id, + "sell", + "waiting-buyer-invoice", + Some(HEX_KEY_A), + Some(HEX_KEY_B), + HEX_KEY_B, + taken_at, + ) + .await; + + // No grace: the order is treated as late, which is what would cancel + // the escrow and slash the bond. + let without_grace = find_order_by_seconds(&pool, Duration::ZERO).await.unwrap(); + assert_eq!(without_grace.len(), 1); + assert_eq!(without_grace[0].id, order_id); + + // Owed more downtime than the order is late by: not late at all. + let with_grace = find_order_by_seconds(&pool, Duration::from_secs(300)) + .await + .unwrap(); + assert!( + with_grace.is_empty(), + "an order cannot be late for a window the daemon spent unable to listen" + ); + + // A grace smaller than the overshoot still selects it: the query only + // has to avoid filtering out rows the caller may spare, and the caller + // decides from each order's own downtime. + let smaller_grace = find_order_by_seconds(&pool, Duration::from_secs(30)) + .await + .unwrap(); + assert_eq!( + smaller_grace.len(), + 1, + "grace only postpones the deadline, it does not remove it" + ); + } + #[tokio::test] async fn find_dispute_by_order_id_finds_and_misses() { let pool = migrated_pool().await; diff --git a/src/inbox.rs b/src/inbox.rs new file mode 100644 index 00000000..2a1b1797 --- /dev/null +++ b/src/inbox.rs @@ -0,0 +1,1934 @@ +//! The daemon's Nostr inbox subscription. +//! +//! Every user action Mostro reacts to — `TakeSell`, `AddInvoice`, `FiatSent`, +//! `Release`, `Dispute`, … — arrives over a **single** long-lived subscription +//! opened once at startup. That subscription is the node's only ear, so this +//! module gives it an identity: a stable id plus the filter that defines it. +//! +//! The id matters because the pieces that keep the inbox alive have to be able +//! to *name* it. A relay's `CLOSED` frame carries a subscription id and nothing +//! else; recognising one as "our inbox just died" — and re-issuing the REQ +//! under the same id — is only possible if the daemon decided the name instead +//! of letting the SDK generate a fresh random one per call. +//! +//! Note that the subscription is deliberately built with `.limit(0)`: it wants +//! live traffic, never stored history. The event loop discards anything whose +//! `created_at` is older than ten seconds anyway (see `accept_event` in +//! `src/app.rs`), so asking a relay for a backlog would only pay for frames +//! that are rejected on arrival. The same ten-second window is why a +//! re-subscribe cannot recover what was missed: whatever a user sent while the +//! inbox was down is already too old to be accepted by the time it could be +//! replayed. Losing the ear loses those messages for good — hence +//! [`InboxKeeper`], which exists to make the outage as short as possible. +//! +//! # Keeping the ear open +//! +//! Relays end subscriptions on their own initiative, and say so with a +//! `CLOSED` frame. The SDK's reaction is unforgiving: for nearly every reason +//! prefix — and for no prefix at all — it *removes* the subscription outright, +//! and removed subscriptions are never re-REQ'd, not even across a reconnect. +//! A single frame from a single relay therefore ends the daemon's ability to +//! hear anything, permanently and without a word: the SDK logs it at `debug`, +//! which release builds filter out (`RUST_LOG=none,mostro=info`). +//! +//! [`InboxKeeper`] closes that hole. It watches the control-plane traffic the +//! event loop used to discard, recognises a `CLOSED` aimed at the inbox, and +//! re-issues the REQ to the relay that sent it — under per-relay backoff, so a +//! relay that refuses the inbox on principle is retried at a decreasing rate +//! instead of being hammered. +//! +//! Two reason prefixes are the exception. `auth-required` and `rate-limited` +//! only *mark* the subscription, leaving it registered for the SDK to re-send +//! by itself, so the keeper stands down on those and lets it: see +//! [`is_provisional_closure`]. +//! +//! Not every way of losing the ear announces itself with a frame, though: the +//! notification channel silently drops messages when the consumer falls +//! behind, a REQ can fail to go out, a relay can be added after startup. +//! [`check_inbox_health`] is the backstop — it asks each connected relay +//! whether it is still serving the subscription, re-subscribes the ones that +//! are not, and records the verdict in [`InboxHealth`] so the rest of the +//! daemon can tell whether Mostro is currently able to hear anything at all. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex, OnceLock}; +use std::time::{Duration, Instant}; + +use nostr_sdk::prelude::*; +use tracing::{debug, error, info, warn}; + +/// Subscription id used for the daemon inbox. +/// +/// Fixed rather than the SDK's per-call random id, so a `CLOSED` frame can be +/// attributed to the inbox and the REQ re-issued under the same name. It is +/// visible to every relay, which costs nothing in privacy: the filter's `#p` +/// tag already names this node. +const INBOX_SUBSCRIPTION_ID: &str = "mostro-inbox"; + +/// Delay before a *second* consecutive re-subscribe to the same relay. +/// +/// The first `CLOSED` is answered immediately — the common case is a transient +/// refusal, and every second of delay is a second of deaf node. Backoff only +/// starts mattering when a relay keeps closing the inbox. +const RESUBSCRIBE_INITIAL_BACKOFF: Duration = Duration::from_secs(2); + +/// Ceiling for the per-relay re-subscribe delay. +/// +/// A relay that has refused the inbox for five minutes straight is not having +/// a hiccup — it is configured to refuse us (NIP-42, a pubkey allowlist, a ban) +/// and the operator has to intervene. Retrying every five minutes keeps the +/// door open for a config change on their side without generating traffic that +/// looks like an attack. +/// +/// This is a real ceiling because [`check_inbox_health`] draws on the same +/// per-relay budget rather than re-sending on every pass: an audit every +/// `INBOX_WATCHDOG_INTERVAL` would otherwise put a hard floor of thirty +/// seconds under it. The doublings still start well below that interval, so a +/// relay that merely lost the inbox is re-subscribed on the next pass and only +/// a persistently refusing one reaches this figure. +const RESUBSCRIBE_MAX_BACKOFF: Duration = Duration::from_secs(300); + +/// Per-relay re-subscribe pacing. +#[derive(Debug)] +struct RelayBackoff { + /// Earliest instant at which another REQ may go out to this relay. + next_attempt_at: Instant, + /// Delay applied after the next attempt; doubles up to the ceiling. + delay: Duration, +} + +/// The daemon's inbox: the subscription every trade message arrives on. +#[derive(Debug, Clone)] +pub struct InboxSubscription { + id: SubscriptionId, + filter: Filter, +} + +impl InboxSubscription { + /// Build the inbox subscription for `mostro_pubkey` on the configured + /// transport's `event_kind` (1059 for protocol v1 gift wraps, 14 for the + /// v2 NIP-44 direct messages — see `docs/TRANSPORT_V2_SPEC.md`). + pub fn new(mostro_pubkey: PublicKey, event_kind: Kind) -> Self { + Self { + id: SubscriptionId::new(INBOX_SUBSCRIPTION_ID), + filter: Filter::new() + .pubkey(mostro_pubkey) + .kind(event_kind) + .limit(0), + } + } + + /// The subscription id relays echo back in `EVENT`, `EOSE` and `CLOSED`. + pub fn id(&self) -> &SubscriptionId { + &self.id + } + + /// The filter defining what the inbox listens for. + pub fn filter(&self) -> &Filter { + &self.filter + } + + /// Send the inbox REQ to every connected relay and report the outcome. + /// + /// The SDK's `Output` marks each relay individually, and a relay that + /// refuses the REQ is not an error for the call as a whole — so a node can + /// come up with a dead ear on some (or every) relay and still look healthy. + /// That verdict is logged here rather than discarded. + pub async fn subscribe(&self, client: &Client) -> Result<(), Error> { + let output = client + .subscribe(self.filter.clone()) + .with_id(self.id.clone()) + .await?; + self.report(&output); + Ok(()) + } + + /// Log which relays took the inbox REQ and which refused it. + fn report(&self, output: &Output) { + for (url, err) in output.failed.iter() { + warn!("Inbox subscription refused by relay {url}: {err}"); + } + + if output.success.is_empty() { + // Not fatal — relays reconnect, and the watchdog retries — but the + // node is deaf until one of them accepts, and that must be said out + // loud. The SDK logs its side at `debug`, which release builds + // filter out entirely (`RUST_LOG=none,mostro=info`). + error!( + "Inbox subscription '{}' was accepted by NO relay: Mostro cannot receive any \ + trade message until this recovers", + self.id + ); + } else { + info!( + "Inbox subscription '{}' active on {} relay(s)", + self.id, + output.success.len() + ); + } + } +} + +/// Keeps the inbox subscription alive across relay-initiated closures. +/// +/// Lives in the event loop, which is the only consumer of the notification +/// stream. All of its mutable state — acknowledgements and re-subscribe +/// pacing — is in [`InboxHealth`], because [`check_inbox_health`] runs from a +/// different task and has to see and share the very same facts. +pub struct InboxKeeper { + subscription: InboxSubscription, + /// Where relay acknowledgements and re-subscribe pacing are recorded. The + /// event loop is the only place an `EOSE` can be observed, but the + /// watchdog is what acts on it, so the facts have to be shared rather than + /// kept here. + health: Option>, +} + +impl InboxKeeper { + pub fn new(subscription: InboxSubscription) -> Self { + Self::with_health(subscription, InboxHealth::global()) + } + + pub fn with_health(subscription: InboxSubscription, health: Option>) -> Self { + Self { + subscription, + health, + } + } + + /// React to one control-plane frame from `relay_url`. + /// + /// Two frames matter for the inbox: `CLOSED`, which means the ear on that + /// relay is gone and has to be re-opened, and `EOSE`, which is a relay + /// confirming it accepted the REQ and is the signal used to clear the + /// backoff. Everything else (`OK`, `NOTICE`, other subscriptions' frames) + /// is not this module's business. + pub async fn on_relay_message( + &self, + client: &Client, + relay_url: &RelayUrl, + message: &RelayMessage<'_>, + ) { + match message { + RelayMessage::Closed { + subscription_id, + message, + } if subscription_id.as_ref() == self.subscription.id() => { + if is_provisional_closure(message) { + // Not the keeper's to answer: the SDK only *marks* these + // two prefixes and re-sends the REQ itself — after the + // NIP-42 round-trip for `auth-required`, on the next + // reconnect for `rate-limited`. Re-issuing the REQ here + // would drop the entry the SDK is about to re-send, race + // its AUTH, and arm a backoff against a relay that is + // behaving exactly as the protocol says it should. + // + // Whatever the SDK does not get to — a `rate-limited` + // closure on a connection that never drops, an + // `auth-required` one this node cannot answer because it + // has no keys — is left to [`check_inbox_health`]: the + // relay is not acknowledged, so the next audit re-sends + // the REQ. That is the right pace for a relay that has + // just asked to be left alone. + info!( + "Relay {relay_url} closed the Mostro inbox subscription provisionally \ + (\"{message}\"); recovery is the SDK's or the watchdog's" + ); + return; + } + warn!("Relay {relay_url} closed the Mostro inbox subscription: \"{message}\""); + self.resubscribe(client, relay_url).await; + } + RelayMessage::EndOfStoredEvents(subscription_id) + if subscription_id.as_ref() == self.subscription.id() => + { + // The relay answered the REQ: it is really serving the inbox, + // which is what the watchdog needs to know, and whatever made + // it fail before is over, so the next failure deserves a prompt + // retry again. + if let Some(health) = &self.health { + if health.note_relay_acknowledged(relay_url) { + info!("Inbox subscription re-established on relay {relay_url}"); + } + } + } + _ => {} + } + } + + /// Re-issue the inbox REQ to a single relay. Pacing is + /// [`resubscribe_relay`]'s job, so the watchdog cannot bypass it. + async fn resubscribe(&self, client: &Client, relay_url: &RelayUrl) { + let relay = match client.relay(relay_url).await { + Ok(Some(relay)) => relay, + Ok(None) => { + warn!("Relay {relay_url} closed the inbox but is no longer in the pool"); + return; + } + Err(e) => { + warn!("Cannot reach relay {relay_url} to re-subscribe the inbox: {e}"); + return; + } + }; + + resubscribe_relay(&relay, &self.subscription, self.health.as_deref()).await; + } +} + +/// Whether a `CLOSED` reason means "not now" rather than "not ever". +/// +/// These are the two prefixes nostr-sdk 0.45.1 maps to `MarkAsClosed` instead +/// of `Remove` (`relay/inner.rs`, the `RelayMessage::Closed` arm), keeping the +/// subscription registered so it can be re-sent without the keeper's help. +/// Every other reason — and no reason at all — removes it, which is what +/// [`InboxKeeper`] exists to undo. +/// +/// `auth-required` is only marked when an authenticator is configured; without +/// one the SDK removes it and no re-REQ follows, but a node in that state has +/// no Nostr keys at all, so the watchdog's pace is the appropriate response. +fn is_provisional_closure(message: &str) -> bool { + matches!( + MachineReadablePrefix::parse(message), + Some(MachineReadablePrefix::AuthRequired) | Some(MachineReadablePrefix::RateLimited) + ) +} + +/// Re-send the inbox REQ to one relay, returning whether one actually went out. +/// +/// Shared by the event-loop keeper (reacting to a `CLOSED`) and the watchdog +/// (finding an ear that went missing without one), so both recover a relay the +/// same way — and, just as importantly, pace it the same way. Backoff and +/// acknowledgement bookkeeping are part of the operation rather than something +/// callers remember to do: +/// +/// - **Pacing.** Both callers draw on one per-relay budget in [`InboxHealth`]. +/// The watchdog would otherwise re-send unconditionally on every pass, +/// putting a hard floor of `INBOX_WATCHDOG_INTERVAL` under a ceiling that +/// claims to be [`RESUBSCRIBE_MAX_BACKOFF`]. Sharing it keeps a transient +/// failure recovering on the very next audit while a relay that refuses the +/// inbox on principle tapers to one REQ every five minutes. +/// - **Acknowledgement.** From the moment a fresh REQ goes out, an earlier +/// `EOSE` says nothing about whether the relay is serving *this* one. A +/// relay that answers, then closes the subscription, then quietly ignores +/// the replacement would otherwise keep its stale credit and read as +/// healthy. +async fn resubscribe_relay( + relay: &Relay, + subscription: &InboxSubscription, + health: Option<&InboxHealth>, +) -> bool { + if let Some(health) = health { + if !health.allow_resubscribe(relay.url()) { + debug!( + "Skipping inbox re-subscribe on relay {}: backing off", + relay.url() + ); + return false; + } + health.note_relay_resubscribed(relay.url()); + } + + // A `CLOSED` does not always remove the subscription: rate-limited and + // auth-required closures only *mark* it, and a marked subscription is + // re-REQ'd no earlier than the next reconnect — which may never come on a + // healthy connection. Dropping the registration first makes the REQ below + // unconditional, instead of being refused as a duplicate id. + let _ = relay.unsubscribe(subscription.id()).await; + + match relay + .subscribe(subscription.filter().clone()) + .with_id(subscription.id().clone()) + .await + { + Ok(_) => info!("Re-sent the inbox subscription to relay {}", relay.url()), + Err(e) => warn!( + "Failed to re-subscribe the inbox on relay {}: {e}", + relay.url() + ), + } + + true +} + +/// Whether the daemon can currently hear anything at all. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum InboxStatus { + /// At least one connected relay is serving the inbox subscription. + Listening, + /// No connected relay is serving it: every message sent to Mostro right + /// now is being lost. + Blind, +} + +/// Process-wide inbox health. `None` until [`InboxHealth::install_global`] +/// runs at startup; consumers treat an absent health record as "listening", so +/// unit tests that never install it behave as before. +static INBOX_HEALTH: OnceLock> = OnceLock::new(); + +/// Why [`InboxHealth::install_global`] refused. Mirrors `spam_gate::InstallError`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum InstallError { + /// A health record is already installed. + AlreadyInstalled, +} + +/// One stretch during which the daemon could not hear. +/// +/// Timestamps are wall-clock seconds, the same base as an order's `taken_at`, +/// because that is what these windows are ultimately intersected against. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct BlindWindow { + start: i64, + /// `None` while the outage is still running. + end: Option, +} + +impl BlindWindow { + /// Seconds of this window that fall inside `[from, to]`. + fn overlap(&self, from: i64, to: i64, now: i64) -> i64 { + let end = self.end.unwrap_or(now); + (end.min(to) - self.start.max(from)).max(0) + } +} + +/// Windows that ended longer ago than this are dropped: no order the timeout +/// job can still be looking at was waiting back then, so they can no longer +/// change any verdict. Generous next to `expiration_seconds` (900s by default) +/// so the bound is never the reason a user loses compensation. +const BLIND_WINDOW_RETENTION_SECS: i64 = 7 * 24 * 3600; + +/// Hard cap on retained windows, so a relay flapping in a tight loop cannot +/// grow this unboundedly between prunes. +const MAX_BLIND_WINDOWS: usize = 512; + +#[derive(Debug)] +struct HealthState { + /// The last verdict an audit reached. `None` until the first one runs: + /// startup is not evidence that the inbox works, and must not be read as + /// such (see [`InboxHealth::is_confirmed_listening`]). + verdict: Option, + /// When the record was installed, which is the earliest moment an outage + /// discovered by the first audit could have begun. + installed_at: i64, + /// Every outage this process has seen, oldest first, pruned by age. + windows: Vec, + /// Relay to the wall-clock second its last `EOSE` for the inbox arrived. + /// + /// A relay is only credited with serving the inbox once it says so. The + /// SDK's own subscription map cannot stand in for this: it records what + /// *we* sent, so a relay that holds the connection open and quietly + /// ignores the REQ still looks subscribed there. + /// + /// The timestamp is what binds the credit to a single websocket session. + /// On reconnect the SDK re-sends the REQ by itself, with no frame this + /// module can observe, so an acknowledgement earned on the previous + /// connection says nothing about the current one — see + /// [`InboxHealth::has_acknowledged_since`]. + acknowledged: HashMap, + /// Re-subscribe pacing, drawn on by the event loop and the watchdog alike. + /// + /// Only holds relays that are currently failing; a relay that answers the + /// REQ is dropped from the map, so the steady state is empty. + backoff: HashMap, +} + +impl HealthState { + fn blind_now(&self) -> Option<&BlindWindow> { + self.windows.last().filter(|w| w.end.is_none()) + } +} + +/// Tracks whether the daemon's ear is open, and when it was not. +/// +/// The scheduler's timeout machinery reads this: an order is only "late" if +/// Mostro was in a position to hear from the user, and the ten-second replay +/// window means a message sent into a dead inbox is gone rather than delayed +/// (see the module docs). Punishing a user for a silence the node itself +/// caused would be unfair, so the timeout clock stops while the inbox is down. +#[derive(Debug)] +pub struct InboxHealth { + state: Mutex, +} + +impl Default for InboxHealth { + fn default() -> Self { + Self::new() + } +} + +impl InboxHealth { + pub fn new() -> Self { + Self::at(now_secs()) + } + + fn at(installed_at: i64) -> Self { + Self { + state: Mutex::new(HealthState { + verdict: None, + installed_at, + windows: Vec::new(), + acknowledged: HashMap::new(), + backoff: HashMap::new(), + }), + } + } + + /// Install as the process-wide health record. + pub fn install_global(self) -> Result<(), InstallError> { + INBOX_HEALTH + .set(Arc::new(self)) + .map_err(|_| InstallError::AlreadyInstalled) + } + + /// The process-wide health record, if one was installed. + pub fn global() -> Option> { + INBOX_HEALTH.get().cloned() + } + + /// Record the current observation, returning the resulting status. + fn observe(&self, status: InboxStatus, now: i64) -> InboxStatus { + let mut state = self.state.lock().expect("inbox health mutex poisoned"); + let first_verdict = state.verdict.is_none(); + state.verdict = Some(status); + + match (status, state.blind_now().is_some()) { + (InboxStatus::Blind, false) => { + // A first audit that finds the inbox deaf has found an outage + // that was already running: the node has not heard anything + // since it came up, so that is when the outage began. + let start = if first_verdict { + state.installed_at + } else { + now + }; + state.windows.push(BlindWindow { start, end: None }); + } + (InboxStatus::Listening, true) => { + if let Some(open) = state.windows.last_mut() { + open.end = Some(now); + } + } + _ => {} + } + + state.windows.retain(|w| match w.end { + Some(end) => end > now - BLIND_WINDOW_RETENTION_SECS, + None => true, + }); + if state.windows.len() > MAX_BLIND_WINDOWS { + let excess = state.windows.len() - MAX_BLIND_WINDOWS; + state.windows.drain(..excess); + } + + status + } + + /// Whether the inbox is deaf right now. + /// + /// A record that has never been audited is not blind — but neither is it + /// known to be listening, which is the question a caller about to act on a + /// user's silence should be asking. See [`Self::is_confirmed_listening`]. + pub fn is_blind(&self) -> bool { + self.state + .lock() + .expect("inbox health mutex poisoned") + .blind_now() + .is_some() + } + + /// Whether an audit has actually confirmed that Mostro can hear. + /// + /// This is the predicate for anything that punishes a user for not + /// answering. It is deliberately false before the first audit: the daemon + /// subscribes at startup and the watchdog's first pass comes later, so + /// between the two there is a window in which a node that never obtained a + /// working inbox would otherwise look healthy and start cancelling orders + /// and slashing bonds over messages it was never in a position to receive. + pub fn is_confirmed_listening(&self) -> bool { + self.state + .lock() + .expect("inbox health mutex poisoned") + .verdict + == Some(InboxStatus::Listening) + } + + /// Seconds the inbox was deaf between `from` and now. + /// + /// This is the compensation a single order is owed, and it is computed per + /// order on purpose. A deadline is wall-clock, but the user is answering a + /// node that has to be listening for the answer to land — and because a + /// message sent into a dead inbox is lost rather than queued (the + /// ten-second replay window, see the module docs), they must send it again + /// once the ear is back. So an order's clock effectively stops for exactly + /// the outages that overlap its own waiting period: an order that was + /// already waiting through an outage is owed all of it, one taken + /// afterwards is owed nothing. + pub fn blind_seconds_since(&self, from: i64) -> i64 { + self.blind_seconds_between(from, now_secs()) + } + + fn blind_seconds_between(&self, from: i64, to: i64) -> i64 { + let state = self.state.lock().expect("inbox health mutex poisoned"); + state.windows.iter().map(|w| w.overlap(from, to, to)).sum() + } + + /// Upper bound on what any order could be owed, for callers that need to + /// widen a query before applying the exact per-order figure. + pub fn max_blind_seconds(&self) -> i64 { + let now = now_secs(); + let state = self.state.lock().expect("inbox health mutex poisoned"); + state + .windows + .iter() + .map(|w| w.end.unwrap_or(now) - w.start) + .sum::() + .max(0) + } + + /// Record that `relay` answered the inbox REQ (an `EOSE` for our + /// subscription), which is the only evidence that it is really serving it. + /// + /// Whatever was making the relay fail is over, so its pacing is reset too + /// and the next failure earns a prompt retry again. Returns whether the + /// relay was being backed off, which is what distinguishes a recovery + /// worth logging from the steady state. + pub fn note_relay_acknowledged(&self, relay: &RelayUrl) -> bool { + self.note_relay_acknowledged_at(relay, now_secs()) + } + + fn note_relay_acknowledged_at(&self, relay: &RelayUrl, at: i64) -> bool { + let mut state = self.state.lock().expect("inbox health mutex poisoned"); + state.acknowledged.insert(relay.clone(), at); + state.backoff.remove(relay).is_some() + } + + /// Whether a re-subscribe to `relay` may go out now, arming the next delay + /// when it may. The first failure for a relay always passes. + /// + /// This is the single pacing budget the event loop and the watchdog share. + /// Under the watchdog's 30-second cadence the doubling only starts to bite + /// once the delay outgrows the interval — so a relay that lost the inbox + /// once is re-subscribed on the very next pass, and only one that keeps + /// refusing tapers to [`RESUBSCRIBE_MAX_BACKOFF`]. + pub fn allow_resubscribe(&self, relay: &RelayUrl) -> bool { + self.allow_resubscribe_at(relay, Instant::now()) + } + + fn allow_resubscribe_at(&self, relay: &RelayUrl, now: Instant) -> bool { + let mut state = self.state.lock().expect("inbox health mutex poisoned"); + match state.backoff.get_mut(relay) { + None => { + state.backoff.insert( + relay.clone(), + RelayBackoff { + next_attempt_at: now + RESUBSCRIBE_INITIAL_BACKOFF, + delay: RESUBSCRIBE_INITIAL_BACKOFF, + }, + ); + true + } + Some(pacing) => { + if now < pacing.next_attempt_at { + return false; + } + pacing.delay = (pacing.delay * 2).min(RESUBSCRIBE_MAX_BACKOFF); + pacing.next_attempt_at = now + pacing.delay; + true + } + } + } + + /// Forget `relay`'s acknowledgement, because the REQ has just been sent + /// again and has yet to be answered. + pub fn note_relay_resubscribed(&self, relay: &RelayUrl) { + self.state + .lock() + .expect("inbox health mutex poisoned") + .acknowledged + .remove(relay); + } + + /// Whether `relay` answered the inbox REQ *on its current connection*. + /// + /// `connected_at` is when the websocket the audit is looking at was + /// established. An acknowledgement older than that was earned on a session + /// that no longer exists: the SDK re-sends the REQ on reconnect of its own + /// accord (`should_resubscribe`), emitting nothing this module can see, so + /// a relay that comes back and then quietly ignores the replacement would + /// otherwise keep reading as healthy on the strength of its old `EOSE`. + /// + /// The comparison is inclusive so that an `EOSE` landing in the same + /// second as the connect still counts. + pub fn has_acknowledged_since(&self, relay: &RelayUrl, connected_at: i64) -> bool { + self.state + .lock() + .expect("inbox health mutex poisoned") + .acknowledged + .get(relay) + .is_some_and(|&at| at >= connected_at) + } + + /// How long the current outage has been running, or zero if listening. + pub fn blind_for_secs(&self) -> i64 { + let now = now_secs(); + self.state + .lock() + .expect("inbox health mutex poisoned") + .blind_now() + .map(|w| (now - w.start).max(0)) + .unwrap_or(0) + } + + /// How long it has been since an audit confirmed Mostro can hear. + /// + /// Zero while listening. Otherwise it counts from the start of the current + /// outage, or — if no audit has ever run — from startup, so a watchdog + /// that never reported cannot leave a caller waiting forever on a verdict + /// that is not coming. + pub fn unconfirmed_for_secs(&self) -> i64 { + let now = now_secs(); + let state = self.state.lock().expect("inbox health mutex poisoned"); + if state.verdict == Some(InboxStatus::Listening) && state.blind_now().is_none() { + return 0; + } + let since = state + .blind_now() + .map(|w| w.start) + .unwrap_or(state.installed_at); + (now - since).max(0) + } +} + +/// Wall-clock seconds, the base an order's `taken_at` is recorded in. +fn now_secs() -> i64 { + Timestamp::now().as_secs() as i64 +} + +/// Check every read relay, re-subscribing any that is not serving the inbox, +/// and record the verdict in the process-wide [`InboxHealth`]. +/// +/// The health question is asked of the *subscription*, not of traffic: a node +/// with no trades in flight is legitimately silent, so treating quiet as +/// failure would raise false alarms on an idle instance and, worse, would stop +/// the timeout machinery for no reason. +/// +/// A relay counts as serving the inbox only when **it** has said so, by +/// answering the REQ with an `EOSE` the event loop recorded, *on the websocket +/// session it is currently on*. The SDK's own subscription map is not +/// evidence: it records what Mostro sent, so a relay that keeps the connection +/// open and quietly drops the REQ still appears subscribed there — and the +/// daemon would resume timeouts while deaf. Neither is an acknowledgement from +/// an earlier connection: the SDK re-sends the REQ by itself after a reconnect +/// and the relay may ignore that one, which is why the check is +/// [`InboxHealth::has_acknowledged_since`] against the relay's `connected_at` +/// rather than a plain membership test. +/// +/// It also means a relay re-subscribed during this audit does not count until +/// it answers, which costs one interval before recovery is declared and keeps +/// the error on the safe side: the timeout clock stays frozen slightly longer +/// than strictly needed rather than restarting too early. +pub async fn check_inbox_health(client: &Client, subscription: &InboxSubscription) -> InboxStatus { + check_inbox_health_with(client, subscription, InboxHealth::global()).await +} + +/// [`check_inbox_health`] against an explicit health record, so tests do not +/// have to install the process-wide one. +async fn check_inbox_health_with( + client: &Client, + subscription: &InboxSubscription, + health: Option>, +) -> InboxStatus { + let relays = client + .relays() + .with_capabilities(RelayCapabilities::READ) + .await; + + let mut listening = 0usize; + let mut retried = 0usize; + + for (url, relay) in relays.iter() { + if !relay.status().is_connected() { + continue; + } + let registered = relay.subscription(subscription.id()).await.is_some(); + let connected_at = relay.stats().connected_at().as_secs() as i64; + // Without a health record there is nowhere to have stored an + // acknowledgement, so fall back to registration alone. + let acknowledged = health + .as_ref() + .map(|h| h.has_acknowledged_since(url, connected_at)) + .unwrap_or(true); + + if registered && acknowledged { + listening += 1; + } else { + // Not serving it: a CLOSED the event loop never saw (the + // notification channel drops frames when it lags), a REQ that + // failed to go out, a relay re-added after startup — or one that + // took the REQ and never answered it. + warn!("Relay {url} is connected but not serving the Mostro inbox; re-subscribing"); + if resubscribe_relay(relay, subscription, health.as_deref()).await { + retried += 1; + } + } + } + + let status = if listening > 0 { + InboxStatus::Listening + } else { + InboxStatus::Blind + }; + + if let Some(health) = &health { + let was_blind = health.is_blind(); + health.observe(status, now_secs()); + + match (was_blind, status) { + (false, InboxStatus::Blind) => error!( + "Mostro inbox is BLIND: no connected relay is serving subscription '{}'. \ + Trade messages sent now are lost, and order timeouts are on hold until it \ + recovers", + subscription.id() + ), + (true, InboxStatus::Listening) => info!( + "Mostro inbox recovered: subscription '{}' is live on {listening} relay(s)", + subscription.id() + ), + (true, InboxStatus::Blind) => { + warn!("Mostro inbox still blind ({retried} relay(s) retried this round)") + } + (false, InboxStatus::Listening) => { + debug!("Inbox healthy on {listening} relay(s)"); + } + } + } + + status +} + +#[cfg(test)] +mod tests { + use super::*; + + fn pubkey() -> PublicKey { + Keys::generate().public_key() + } + + /// A keeper backed by its own health record: all of its state — pacing and + /// acknowledgements alike — now lives there, so tests need the handle too. + fn keeper() -> (InboxKeeper, Arc) { + let health = Arc::new(InboxHealth::at(T0)); + let keeper = InboxKeeper::with_health( + InboxSubscription::new(pubkey(), Kind::GiftWrap), + Some(health.clone()), + ); + (keeper, health) + } + + /// Whether `health` is currently pacing re-subscribes to `url`. + fn backing_off(health: &InboxHealth, url: &RelayUrl) -> bool { + health + .state + .lock() + .expect("inbox health mutex poisoned") + .backoff + .contains_key(url) + } + + fn relay_url(url: &str) -> RelayUrl { + RelayUrl::parse(url).expect("valid relay url") + } + + /// Whether `health` holds any acknowledgement for `url` at all, for the + /// tests that care about the record rather than the connection it belongs + /// to. + fn acked(health: &InboxHealth, url: &RelayUrl) -> bool { + health.has_acknowledged_since(url, 0) + } + + #[test] + fn filter_matches_the_subscription_the_daemon_has_always_used() { + let key = pubkey(); + + for kind in [Kind::GiftWrap, Kind::PrivateDirectMessage] { + let inbox = InboxSubscription::new(key, kind); + // The pre-existing `main.rs` filter, spelled out: p-tagged to this + // node, one transport kind, no stored history. + let expected = Filter::new().pubkey(key).kind(kind).limit(0); + assert_eq!( + inbox.filter(), + &expected, + "inbox filter drifted from the daemon's historical subscription" + ); + } + } + + #[test] + fn id_is_stable_across_instances() { + // A CLOSED frame can only be attributed to the inbox if the id is the + // same one the REQ went out under — including after a re-subscribe, + // which builds a fresh `InboxSubscription`. + let key = pubkey(); + let first = InboxSubscription::new(key, Kind::PrivateDirectMessage); + let second = InboxSubscription::new(key, Kind::PrivateDirectMessage); + + assert_eq!(first.id(), second.id()); + assert_eq!(first.id().to_string(), INBOX_SUBSCRIPTION_ID); + } + + #[test] + fn transport_kind_selects_what_the_inbox_hears() { + let key = pubkey(); + + let v1 = InboxSubscription::new(key, Kind::GiftWrap); + let v2 = InboxSubscription::new(key, Kind::PrivateDirectMessage); + + assert_ne!(v1.filter(), v2.filter()); + assert_eq!(v1.filter().kinds.as_ref().unwrap().len(), 1); + assert!(v1 + .filter() + .kinds + .as_ref() + .unwrap() + .contains(&Kind::GiftWrap)); + assert!(v2 + .filter() + .kinds + .as_ref() + .unwrap() + .contains(&Kind::PrivateDirectMessage)); + } + + // ───────────────────────────── backoff pacing ───────────────────────────── + + #[test] + fn first_closure_from_a_relay_retries_immediately() { + let health = InboxHealth::at(T0); + let relay = relay_url("ws://relay.example"); + + assert!( + health.allow_resubscribe_at(&relay, Instant::now()), + "a first CLOSED must be answered at once: every delay is deaf time" + ); + } + + #[test] + fn repeat_closures_are_paced_and_back_off() { + let health = InboxHealth::at(T0); + let relay = relay_url("ws://relay.example"); + let start = Instant::now(); + + assert!(health.allow_resubscribe_at(&relay, start)); + // A relay that closes again right away must not pull a second REQ. + assert!(!health.allow_resubscribe_at(&relay, start)); + assert!(!health.allow_resubscribe_at(&relay, start + Duration::from_secs(1))); + + // Past the first delay it retries, and the next wait is longer. + assert!(health.allow_resubscribe_at(&relay, start + RESUBSCRIBE_INITIAL_BACKOFF)); + assert!(!health.allow_resubscribe_at(&relay, start + RESUBSCRIBE_INITIAL_BACKOFF * 2)); + assert!(health.allow_resubscribe_at(&relay, start + RESUBSCRIBE_INITIAL_BACKOFF * 3)); + } + + #[test] + fn backoff_is_capped() { + let health = InboxHealth::at(T0); + let relay = relay_url("ws://relay.example"); + let mut now = Instant::now(); + + // Drive it well past the ceiling. + for _ in 0..20 { + assert!(health.allow_resubscribe_at(&relay, now)); + now += RESUBSCRIBE_MAX_BACKOFF * 2; + } + + assert_eq!( + health + .state + .lock() + .unwrap() + .backoff + .get(&relay) + .expect("state kept") + .delay, + RESUBSCRIBE_MAX_BACKOFF, + "a hostile relay must still be retried every {RESUBSCRIBE_MAX_BACKOFF:?}" + ); + } + + #[test] + fn backoff_is_per_relay() { + let health = InboxHealth::at(T0); + let hostile = relay_url("ws://hostile.example"); + let healthy = relay_url("ws://healthy.example"); + let now = Instant::now(); + + assert!(health.allow_resubscribe_at(&hostile, now)); + assert!(!health.allow_resubscribe_at(&hostile, now)); + // One misbehaving relay must not delay recovery on another. + assert!(health.allow_resubscribe_at(&healthy, now)); + } + + #[test] + fn an_acknowledgement_clears_the_pacing_for_the_next_failure() { + let health = InboxHealth::at(T0); + let relay = relay_url("ws://relay.example"); + let now = Instant::now(); + + assert!(health.allow_resubscribe_at(&relay, now)); + assert!(!health.allow_resubscribe_at(&relay, now)); + + assert!( + health.note_relay_acknowledged(&relay), + "clearing a live backoff entry is what marks a recovery" + ); + assert!( + health.allow_resubscribe_at(&relay, now), + "a relay that answered starts over: the next failure is a fresh one" + ); + + // The steady state has nothing to clear, so nothing to report either. + health.note_relay_acknowledged(&relay); + assert!(!health.note_relay_acknowledged(&relay)); + } + + #[test] + fn the_watchdog_cadence_recovers_promptly_and_only_then_tapers() { + // The point of sharing one budget: an audit every + // `INBOX_WATCHDOG_INTERVAL` must still re-subscribe a relay that + // simply lost the inbox, while a relay that refuses it converges on + // the advertised ceiling instead of drawing a REQ every 30 seconds + // forever. + let health = InboxHealth::at(T0); + let relay = relay_url("ws://hostile.example"); + let tick = Duration::from_secs(crate::scheduler::INBOX_WATCHDOG_INTERVAL); + let mut now = Instant::now(); + + assert!( + health.allow_resubscribe_at(&relay, now), + "the pass that first notices the loss must act on it" + ); + for pass in 1..=3 { + now += tick; + assert!( + health.allow_resubscribe_at(&relay, now), + "pass {pass}: a delay still under the audit interval must not skip a retry" + ); + } + + // Once the doubling outgrows the interval, passes start being skipped. + let mut attempts = 0; + for _ in 0..40 { + now += tick; + if health.allow_resubscribe_at(&relay, now) { + attempts += 1; + } + } + assert!( + attempts < 40, + "a relay that keeps refusing must stop drawing a REQ on every pass" + ); + assert_eq!( + health + .state + .lock() + .unwrap() + .backoff + .get(&relay) + .expect("state kept") + .delay, + RESUBSCRIBE_MAX_BACKOFF + ); + } + + // ───────────────────────── control-plane handling ───────────────────────── + + #[tokio::test] + async fn eose_for_the_inbox_clears_the_backoff() { + let client = crate::util::mostro_nostr_client_options(None).build(); + let (keeper, health) = keeper(); + let relay = relay_url("ws://relay.example"); + + assert!(health.allow_resubscribe(&relay)); + assert!(backing_off(&health, &relay)); + + let eose = RelayMessage::EndOfStoredEvents(std::borrow::Cow::Owned( + keeper.subscription.id().clone(), + )); + keeper.on_relay_message(&client, &relay, &eose).await; + + assert!( + !backing_off(&health, &relay), + "an accepted REQ must reset the pacing for the next failure" + ); + } + + #[test] + fn only_the_two_prefixes_the_sdk_recovers_from_are_provisional() { + // Mirrors the `RelayMessage::Closed` arm of nostr-sdk 0.45.1: these + // two map to `MarkAsClosed`, everything else to `Remove`. A future + // bump that changes the split has to change this list with it. + assert!(is_provisional_closure( + "auth-required: we only serve authenticated users" + )); + assert!(is_provisional_closure("rate-limited: slow down")); + + for permanent in [ + "blocked: you are banned", + "restricted: not for you", + "error: go away", + "invalid: bad filter", + "unsupported: no such filter", + "pow: 24 bits required", + "duplicate: already have it", + "", + "we are closing this one", + ] { + assert!( + !is_provisional_closure(permanent), + "{permanent:?} removes the subscription, so the keeper has to re-send the REQ" + ); + } + } + + #[tokio::test] + async fn a_provisional_closure_is_left_to_the_sdk() { + let client = crate::util::mostro_nostr_client_options(None).build(); + let health = Arc::new(InboxHealth::at(T0)); + let subscription = InboxSubscription::new(pubkey(), Kind::GiftWrap); + let keeper = InboxKeeper::with_health(subscription.clone(), Some(health.clone())); + let relay = relay_url("ws://relay.example"); + + health.note_relay_acknowledged(&relay); + + for reason in ["auth-required: please auth", "rate-limited: slow down"] { + let closed = RelayMessage::Closed { + subscription_id: std::borrow::Cow::Owned(subscription.id().clone()), + message: std::borrow::Cow::Borrowed(reason), + }; + keeper.on_relay_message(&client, &relay, &closed).await; + } + + assert!( + !backing_off(&health, &relay), + "a relay the SDK will re-REQ by itself must not be put on the shared backoff" + ); + assert!( + acked(&health, &relay), + "the subscription is still registered and still answered, so the credit stands" + ); + } + + #[tokio::test] + async fn a_permanent_closure_is_still_the_keepers_to_answer() { + use nostr_sdk::local_relay::LocalRelay; + + let relay = LocalRelay::builder().build(); + relay.run().await.expect("run local relay"); + let url = relay.url().await; + + let (keeper, health) = keeper(); + let client = crate::util::mostro_nostr_client_options(None).build(); + client.add_relay(url.clone()).await.expect("add_relay"); + client.connect().await; + health.note_relay_acknowledged(&url); + + let closed = RelayMessage::Closed { + subscription_id: std::borrow::Cow::Owned(keeper.subscription.id().clone()), + message: std::borrow::Cow::Borrowed("blocked: you are banned"), + }; + keeper.on_relay_message(&client, &url, &closed).await; + + assert!( + backing_off(&health, &url), + "a CLOSED the SDK removes the subscription for must still arm the keeper" + ); + assert!( + !acked(&health, &url), + "the replacement REQ has yet to be answered, so the old EOSE cannot vouch for it" + ); + + relay.shutdown(); + } + + #[tokio::test] + async fn frames_for_other_subscriptions_are_ignored() { + let client = crate::util::mostro_nostr_client_options(None).build(); + let (keeper, health) = keeper(); + let relay = relay_url("ws://relay.example"); + + // Mostro's price provider and NIP-33 queries share these relays; their + // CLOSED frames must not touch the inbox's state. + let other = RelayMessage::Closed { + subscription_id: std::borrow::Cow::Owned(SubscriptionId::new("someone-else")), + message: std::borrow::Cow::Borrowed("error: not yours"), + }; + keeper.on_relay_message(&client, &relay, &other).await; + + assert!( + !backing_off(&health, &relay), + "a CLOSED for another subscription must not be treated as an inbox failure" + ); + } + + // ────────────────────────── end-to-end regression ───────────────────────── + + /// Rejects the first REQ it sees and admits every one after it: a relay + /// having a bad moment, which is exactly the case the daemon used to never + /// recover from. + #[derive(Debug, Default)] + struct RejectFirstQuery { + seen: std::sync::atomic::AtomicUsize, + } + + impl nostr_sdk::local_relay::QueryPolicy for RejectFirstQuery { + fn admit_query<'a>( + &'a self, + _query: &'a mut Filter, + _addr: &'a std::net::SocketAddr, + ) -> std::pin::Pin< + Box< + dyn std::future::Future + + Send + + 'a, + >, + > { + Box::pin(async move { + let first = self.seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0; + if first { + nostr_sdk::local_relay::QueryPolicyResult::reject( + MachineReadablePrefix::Error, + "subscription refused", + ) + } else { + nostr_sdk::local_relay::QueryPolicyResult::Accept + } + }) + } + } + + /// A gift wrap addressed to `recipient` — one p-tag, as the transport (and + /// the relay's own validation) requires. + fn wrap_for(recipient: PublicKey) -> Event { + EventBuilder::new(Kind::GiftWrap, "sealed") + .tag(Tag::public_key(recipient)) + .finalize(&Keys::generate()) + .expect("sign gift wrap") + } + + #[tokio::test] + async fn closed_inbox_is_resubscribed_and_hears_again() { + use futures::StreamExt; + use nostr_sdk::local_relay::LocalRelay; + + let relay = LocalRelay::builder() + .query_policy(RejectFirstQuery::default()) + .build(); + relay.run().await.expect("run local relay"); + let url = relay.url().await; + + let mostro = Keys::generate(); + let subscription = InboxSubscription::new(mostro.public_key(), Kind::GiftWrap); + + let client = crate::util::mostro_nostr_client_options(None).build(); + client.add_relay(url.clone()).await.expect("add_relay"); + client.connect().await; + + let mut notifications = client.notifications(); + // The relay CLOSEs this one; without the keeper the ear is gone here. + subscription.subscribe(&client).await.expect("subscribe"); + + let publisher = ClientBuilder::default().build(); + publisher.add_relay(url.clone()).await.expect("add_relay"); + publisher.connect().await; + + let keeper = InboxKeeper::new(subscription.clone()); + let wanted = wrap_for(mostro.public_key()); + let wanted_id = wanted.id; + let mut published = false; + + let heard = tokio::time::timeout(Duration::from_secs(20), async { + while let Some(notification) = notifications.next().await { + match notification { + ClientNotification::Event { event, .. } => { + if event.id == wanted_id { + return true; + } + } + ClientNotification::Message { relay_url, message } => { + keeper.on_relay_message(&client, &relay_url, &message).await; + // Publish only once the inbox is confirmed live again, + // so the event cannot be mistaken for stored history. + if !published && matches!(&*message, RelayMessage::EndOfStoredEvents(_)) { + published = true; + publisher.send_event(&wanted).await.expect("publish"); + } + } + ClientNotification::Shutdown => return false, + } + } + false + }) + .await + .expect("timed out: the inbox never recovered from the relay's CLOSED"); + + assert!( + heard, + "after a relay CLOSED the inbox, Mostro must re-subscribe and receive again" + ); + + relay.shutdown(); + } + + #[tokio::test] + async fn a_receiver_created_after_the_req_misses_its_eose() { + use futures::StreamExt; + use nostr_sdk::local_relay::LocalRelay; + + // Why the event loop must subscribe *after* taking its notification + // stream: the SDK delivers nothing that predates the receiver, so a + // REQ sent earlier loses its EOSE — and any event arriving meanwhile. + let relay = LocalRelay::builder().build(); + relay.run().await.expect("run local relay"); + let url = relay.url().await; + + let subscription = InboxSubscription::new(pubkey(), Kind::GiftWrap); + let client = crate::util::mostro_nostr_client_options(None).build(); + client.add_relay(url.clone()).await.expect("add_relay"); + client.connect().await; + + // Subscribe first, listen second — the order this module avoids. + subscription.subscribe(&client).await.expect("subscribe"); + tokio::time::sleep(Duration::from_millis(500)).await; + let mut late = client.notifications(); + + let saw_eose = tokio::time::timeout(Duration::from_secs(2), async { + while let Some(notification) = late.next().await { + if let ClientNotification::Message { message, .. } = notification { + if let RelayMessage::EndOfStoredEvents(id) = &*message { + if id.as_ref() == subscription.id() { + return true; + } + } + } + } + false + }) + .await + .unwrap_or(false); + + assert!( + !saw_eose, + "SDK behaviour changed: a late receiver now sees earlier frames, so the \ + subscribe-after-stream ordering in `app::run` could be relaxed" + ); + + relay.shutdown(); + } + + #[tokio::test] + async fn without_the_keeper_a_closed_inbox_stays_dead() { + use nostr_sdk::local_relay::LocalRelay; + + // The defect this module exists for: the SDK drops a CLOSED + // subscription outright, and nothing re-issues the REQ. Pinned here so + // that a future SDK bump changing this behaviour is noticed. + let relay = LocalRelay::builder() + .query_policy(RejectFirstQuery::default()) + .build(); + relay.run().await.expect("run local relay"); + let url = relay.url().await; + + let subscription = InboxSubscription::new(pubkey(), Kind::GiftWrap); + let client = crate::util::mostro_nostr_client_options(None).build(); + client.add_relay(url.clone()).await.expect("add_relay"); + client.connect().await; + subscription.subscribe(&client).await.expect("subscribe"); + + // Give the relay time to answer with CLOSED. + tokio::time::sleep(Duration::from_secs(2)).await; + + assert!( + !client.subscriptions().await.contains_key(subscription.id()), + "SDK behaviour changed: a CLOSED subscription is no longer dropped, \ + so the keeper's premise needs revisiting" + ); + + relay.shutdown(); + } + + // ───────────────────────────── health record ───────────────────────────── + + /// Health observations are wall-clock based, so tests drive a fixed origin + /// rather than the real clock. + const T0: i64 = 1_700_000_000; + + #[test] + fn health_records_an_outage_from_first_blindness_to_recovery() { + let health = InboxHealth::at(T0); + + assert!(!health.is_blind(), "a fresh record starts out listening"); + + health.observe(InboxStatus::Blind, T0); + assert!(health.is_blind()); + + // Staying blind must not restart the clock — the outage began at the + // first observation, and that is what an order is owed. + health.observe(InboxStatus::Blind, T0 + 30); + assert!(health.is_blind()); + + health.observe(InboxStatus::Listening, T0 + 90); + assert!(!health.is_blind()); + + assert_eq!( + health.blind_seconds_between(T0, T0 + 90), + 90, + "the recorded outage must span the whole blind window" + ); + } + + #[test] + fn health_is_not_listening_until_an_audit_says_so() { + let health = InboxHealth::at(T0); + + // Startup is not evidence. Between `main` subscribing and the + // watchdog's first pass, a node whose inbox never worked would + // otherwise process timeouts as if it had been listening all along. + assert!( + !health.is_confirmed_listening(), + "an unaudited record must not authorise acting on a user's silence" + ); + assert!( + !health.is_blind(), + "nor should it claim an outage it has not observed" + ); + + health.observe(InboxStatus::Listening, T0); + assert!(health.is_confirmed_listening()); + } + + #[test] + fn a_blind_first_audit_dates_the_outage_from_startup() { + let health = InboxHealth::at(T0); + + // The watchdog's first pass comes some time after boot. Finding the + // inbox deaf then means it was deaf for that whole stretch, not just + // from the moment somebody looked. + health.observe(InboxStatus::Blind, T0 + 30); + health.observe(InboxStatus::Listening, T0 + 90); + + assert_eq!( + health.blind_seconds_between(T0, T0 + 90), + 90, + "the outage must be dated from startup, not from the first audit" + ); + } + + #[test] + fn a_node_that_was_never_blind_owes_nothing() { + let health = InboxHealth::at(T0); + health.observe(InboxStatus::Listening, T0); + + assert_eq!(health.blind_seconds_between(T0, T0 + 10_000), 0); + assert_eq!(health.max_blind_seconds(), 0); + } + + // ──────────────────── what a single order is owed ──────────────────── + + #[test] + fn an_order_is_owed_only_the_downtime_it_waited_through() { + let health = InboxHealth::at(T0); + // One outage: [T0+100, T0+400], five minutes. + health.observe(InboxStatus::Listening, T0); + health.observe(InboxStatus::Blind, T0 + 100); + health.observe(InboxStatus::Listening, T0 + 400); + + let now = T0 + 1_000; + + // Waiting since before it started: owed the whole outage. + assert_eq!(health.blind_seconds_between(T0, now), 300); + // Taken midway through: owed only the remainder. + assert_eq!(health.blind_seconds_between(T0 + 250, now), 150); + // Taken after it ended: owed nothing. This is what a single global + // allowance got wrong — it credited orders that never lost a second. + assert_eq!(health.blind_seconds_between(T0 + 500, now), 0); + } + + #[test] + fn compensation_does_not_evaporate_as_time_passes() { + let health = InboxHealth::at(T0); + health.observe(InboxStatus::Listening, T0); + health.observe(InboxStatus::Blind, T0 + 100); + health.observe(InboxStatus::Listening, T0 + 400); + + // The debt an order carries is a property of when it waited, not of + // how long ago the outage was. A decaying allowance wore off at the + // same rate the deadline advanced, so it compensated almost nothing. + for probe in [400, 700, 5_000, 50_000] { + assert_eq!( + health.blind_seconds_between(T0, T0 + probe), + 300, + "an order waiting since T0 is owed the outage regardless of when we ask" + ); + } + } + + #[test] + fn an_order_waiting_through_an_outage_survives_its_nominal_deadline() { + // The regression in full: 900s timeout, an order taken at T0, and a + // 300s outage right at the start. Under the old decaying allowance + // this order was cancelled at ~T0+900, having had only 600s of + // listening time. + let health = InboxHealth::at(T0); + health.observe(InboxStatus::Blind, T0); + health.observe(InboxStatus::Listening, T0 + 300); + + let exp_seconds = 900i64; + let late_at = |now: i64| { + let owed = health.blind_seconds_between(T0, now); + (now - T0) >= exp_seconds + owed + }; + + assert!(!late_at(T0 + 900), "cancelled after only 600s of listening"); + assert!(!late_at(T0 + 1_199)); + assert!( + late_at(T0 + 1_200), + "and it must still expire once it has had its full 900s" + ); + } + + #[test] + fn consecutive_outages_accumulate_their_debt() { + let health = InboxHealth::at(T0); + health.observe(InboxStatus::Listening, T0); + health.observe(InboxStatus::Blind, T0 + 100); + health.observe(InboxStatus::Listening, T0 + 200); + health.observe(InboxStatus::Blind, T0 + 240); + health.observe(InboxStatus::Listening, T0 + 290); + + assert_eq!( + health.blind_seconds_between(T0, T0 + 1_000), + 150, + "an order waiting through both outages is owed both" + ); + assert_eq!( + health.blind_seconds_between(T0 + 210, T0 + 1_000), + 50, + "one taken between them is owed only the second" + ); + } + + #[test] + fn an_ongoing_outage_counts_up_to_now() { + let health = InboxHealth::at(T0); + health.observe(InboxStatus::Listening, T0); + health.observe(InboxStatus::Blind, T0 + 100); + + assert_eq!(health.blind_seconds_between(T0, T0 + 400), 300); + assert_eq!(health.blind_seconds_between(T0, T0 + 900), 800); + } + + #[test] + fn stale_windows_are_pruned() { + let health = InboxHealth::at(T0); + health.observe(InboxStatus::Listening, T0); + health.observe(InboxStatus::Blind, T0 + 100); + health.observe(InboxStatus::Listening, T0 + 200); + + // Far past the retention horizon, the old window is dropped rather + // than accumulating for the life of the process. + let much_later = T0 + BLIND_WINDOW_RETENTION_SECS + 1_000; + health.observe(InboxStatus::Listening, much_later); + + assert_eq!(health.blind_seconds_between(T0, much_later), 0); + assert!(health.state.lock().expect("lock").windows.is_empty()); + } + + #[test] + fn unconfirmed_time_counts_from_the_outage_or_from_startup() { + // What bounds how long the timeout job may defer. It has to answer + // even when no audit ever ran, or a watchdog that died would park the + // job on a verdict that is never coming. + let never_audited = InboxHealth::at(now_secs() - 120); + assert!( + never_audited.unconfirmed_for_secs() >= 120, + "with no verdict at all, the clock runs from startup" + ); + + let healthy = InboxHealth::at(now_secs()); + healthy.observe(InboxStatus::Listening, now_secs()); + assert_eq!( + healthy.unconfirmed_for_secs(), + 0, + "a confirmed inbox owes no waiting" + ); + + let blind = InboxHealth::at(now_secs() - 600); + blind.observe(InboxStatus::Listening, now_secs() - 600); + blind.observe(InboxStatus::Blind, now_secs() - 300); + assert!( + (300..=310).contains(&blind.unconfirmed_for_secs()), + "while blind it runs from the start of the outage, got {}", + blind.unconfirmed_for_secs() + ); + } + + #[test] + fn health_ignores_repeated_healthy_observations() { + let health = InboxHealth::at(T0); + + health.observe(InboxStatus::Listening, T0); + health.observe(InboxStatus::Listening, T0 + 30); + + assert!(!health.is_blind()); + assert_eq!( + health.blind_seconds_between(T0, T0 + 30), + 0, + "a node that was never blind has no outage to compensate for" + ); + } + + // ──────────────────────────────── watchdog ──────────────────────────────── + + #[tokio::test] + async fn watchdog_resubscribes_a_relay_that_lost_the_inbox() { + use nostr_sdk::local_relay::LocalRelay; + + // Accepts every REQ: the point here is the *missing* subscription, not + // a refusing relay. + let relay = LocalRelay::builder().build(); + relay.run().await.expect("run local relay"); + let url = relay.url().await; + + let subscription = InboxSubscription::new(pubkey(), Kind::GiftWrap); + let client = crate::util::mostro_nostr_client_options(None).build(); + client.add_relay(url.clone()).await.expect("add_relay"); + client.connect().await; + subscription.subscribe(&client).await.expect("subscribe"); + tokio::time::sleep(Duration::from_millis(500)).await; + + // Simulate the ear vanishing without a CLOSED the loop could see — + // a frame dropped by a lagging notification channel looks like this. + client + .unsubscribe(subscription.id()) + .await + .expect("drop the subscription"); + assert!(!client.subscriptions().await.contains_key(subscription.id())); + + let health = Arc::new(InboxHealth::at(T0)); + + // The audit re-subscribes, but does not yet claim to be listening: a + // REQ that just went out proves nothing about whether the relay will + // honour it. + assert_eq!( + check_inbox_health_with(&client, &subscription, Some(health.clone())).await, + InboxStatus::Blind, + "a relay re-subscribed during this audit must not count as listening yet" + ); + assert!( + client.subscriptions().await.contains_key(subscription.id()), + "the inbox subscription must be back after the audit" + ); + + // The relay answers the new REQ; in the daemon this is the event loop + // seeing the EOSE and recording it. + health.note_relay_acknowledged(&url); + + assert_eq!( + check_inbox_health_with(&client, &subscription, Some(health)).await, + InboxStatus::Listening, + "a relay that answered the REQ is serving the inbox" + ); + + relay.shutdown(); + } + + #[tokio::test] + async fn a_closed_frame_invalidates_the_relays_earlier_acknowledgement() { + use nostr_sdk::local_relay::LocalRelay; + + // A relay can answer the REQ, later close the subscription, and then + // ignore the replacement. Its old EOSE must not carry over: the + // watchdog would see the re-registered subscription plus stale credit + // and resume order timeouts while the node is deaf. + let relay = LocalRelay::builder().build(); + relay.run().await.expect("run local relay"); + let url = relay.url().await; + + let client = crate::util::mostro_nostr_client_options(None).build(); + client.add_relay(url.clone()).await.expect("add_relay"); + client.connect().await; + + let health = Arc::new(InboxHealth::at(T0)); + let subscription = InboxSubscription::new(pubkey(), Kind::GiftWrap); + let keeper = InboxKeeper::with_health(subscription.clone(), Some(health.clone())); + + // The relay answered an earlier REQ. + health.note_relay_acknowledged(&url); + assert!(acked(&health, &url)); + + // Now it closes the subscription; the keeper re-sends the REQ. + let closed = RelayMessage::Closed { + subscription_id: std::borrow::Cow::Owned(subscription.id().clone()), + message: std::borrow::Cow::Borrowed("error: go away"), + }; + keeper.on_relay_message(&client, &url, &closed).await; + + assert!( + !acked(&health, &url), + "an EOSE from before the CLOSED must not vouch for the replacement REQ" + ); + + relay.shutdown(); + } + + #[test] + fn an_acknowledgement_does_not_survive_the_connection_it_was_earned_on() { + // A websocket drop and reconnect leaves no trace the keeper can act + // on: there is no relay-status `ClientNotification` in nostr-sdk + // 0.45.1, and the SDK silently re-sends the REQ by itself + // (`should_resubscribe`). If the relay then ignores that replacement, + // the only thing standing between a deaf node and resumed slashing is + // the acknowledgement expiring with its session. + let health = InboxHealth::at(T0); + let url = relay_url("ws://relay.example"); + + health.note_relay_acknowledged_at(&url, T0 + 100); + + assert!(health.has_acknowledged_since(&url, T0 + 50)); + assert!( + health.has_acknowledged_since(&url, T0 + 100), + "an EOSE landing in the same second as the connect must still count" + ); + assert!( + !health.has_acknowledged_since(&url, T0 + 101), + "credit earned on a previous connection must not vouch for this one" + ); + } + + #[tokio::test] + async fn watchdog_does_not_trust_an_acknowledgement_from_a_previous_connection() { + use nostr_sdk::local_relay::LocalRelay; + + // The audit-level counterpart: the subscription is registered and the + // relay has an acknowledgement on file, but it predates the current + // websocket session, which is exactly what a reconnect leaves behind. + let relay = LocalRelay::builder().build(); + relay.run().await.expect("run local relay"); + let url = relay.url().await; + + let subscription = InboxSubscription::new(pubkey(), Kind::GiftWrap); + let client = crate::util::mostro_nostr_client_options(None).build(); + client.add_relay(url.clone()).await.expect("add_relay"); + client.connect().await; + subscription.subscribe(&client).await.expect("subscribe"); + tokio::time::sleep(Duration::from_millis(500)).await; + + let health = Arc::new(InboxHealth::at(T0)); + health.note_relay_acknowledged_at(&url, T0); + + assert!( + client.subscriptions().await.contains_key(subscription.id()), + "precondition: the SDK has the subscription registered" + ); + assert_eq!( + check_inbox_health_with(&client, &subscription, Some(health.clone())).await, + InboxStatus::Blind, + "a stale acknowledgement plus a live registration must not read as listening" + ); + + // An EOSE on the current connection is what settles it. + health.note_relay_acknowledged(&url); + assert_eq!( + check_inbox_health_with(&client, &subscription, Some(health)).await, + InboxStatus::Listening + ); + + relay.shutdown(); + } + + #[tokio::test] + async fn watchdog_does_not_trust_a_relay_that_never_answered() { + use nostr_sdk::local_relay::LocalRelay; + + // A relay can hold the connection open and quietly drop the REQ. The + // SDK still lists the subscription, because that map records what + // Mostro sent, not what the relay agreed to serve. Treating it as + // proof would resume order timeouts against a deaf node. + let relay = LocalRelay::builder().build(); + relay.run().await.expect("run local relay"); + let url = relay.url().await; + + let subscription = InboxSubscription::new(pubkey(), Kind::GiftWrap); + let client = crate::util::mostro_nostr_client_options(None).build(); + client.add_relay(url.clone()).await.expect("add_relay"); + client.connect().await; + subscription.subscribe(&client).await.expect("subscribe"); + tokio::time::sleep(Duration::from_millis(500)).await; + + let health = Arc::new(InboxHealth::at(T0)); + + assert!( + client.subscriptions().await.contains_key(subscription.id()), + "precondition: the SDK has the subscription registered" + ); + assert_eq!( + check_inbox_health_with(&client, &subscription, Some(health.clone())).await, + InboxStatus::Blind, + "local registration alone must not count as the relay serving the inbox" + ); + + // Once it does answer, the same registration is finally evidence. + health.note_relay_acknowledged(&url); + assert_eq!( + check_inbox_health_with(&client, &subscription, Some(health)).await, + InboxStatus::Listening + ); + + relay.shutdown(); + } + + #[tokio::test] + async fn keeper_records_the_acknowledgement_the_watchdog_reads() { + // The EOSE is only observable from the event loop, while the watchdog + // is what acts on it — this is the handoff between the two. + let client = crate::util::mostro_nostr_client_options(None).build(); + let health = Arc::new(InboxHealth::at(T0)); + let subscription = InboxSubscription::new(pubkey(), Kind::GiftWrap); + let keeper = InboxKeeper::with_health(subscription.clone(), Some(health.clone())); + let url = relay_url("ws://relay.example"); + + assert!(!acked(&health, &url)); + + let eose = + RelayMessage::EndOfStoredEvents(std::borrow::Cow::Owned(subscription.id().clone())); + keeper.on_relay_message(&client, &url, &eose).await; + + assert!( + acked(&health, &url), + "an EOSE for the inbox must be recorded as the relay serving it" + ); + + // A frame for someone else's subscription proves nothing about ours. + let other_url = relay_url("ws://other.example"); + let other = RelayMessage::EndOfStoredEvents(std::borrow::Cow::Owned(SubscriptionId::new( + "not-the-inbox", + ))); + keeper.on_relay_message(&client, &other_url, &other).await; + assert!(!acked(&health, &other_url)); + } + + #[tokio::test] + async fn watchdog_stays_blind_against_a_relay_that_keeps_closing() { + use nostr_sdk::local_relay::LocalRelay; + + /// Refuses every REQ, always. + #[derive(Debug)] + struct RejectAllQueries; + + impl nostr_sdk::local_relay::QueryPolicy for RejectAllQueries { + fn admit_query<'a>( + &'a self, + _query: &'a mut Filter, + _addr: &'a std::net::SocketAddr, + ) -> std::pin::Pin< + Box< + dyn std::future::Future + + Send + + 'a, + >, + > { + Box::pin(async { + nostr_sdk::local_relay::QueryPolicyResult::reject( + MachineReadablePrefix::Blocked, + "no subscriptions here", + ) + }) + } + } + + let relay = LocalRelay::builder().query_policy(RejectAllQueries).build(); + relay.run().await.expect("run local relay"); + let url = relay.url().await; + + let subscription = InboxSubscription::new(pubkey(), Kind::GiftWrap); + let client = crate::util::mostro_nostr_client_options(None).build(); + client.add_relay(url.clone()).await.expect("add_relay"); + client.connect().await; + subscription.subscribe(&client).await.expect("subscribe"); + + let health = Arc::new(InboxHealth::at(T0)); + + // However many rounds it runs, a relay that keeps closing the inbox + // never makes the node look healthy — this is what keeps the timeout + // machinery paused while trade messages are being lost. + for round in 0..3 { + tokio::time::sleep(Duration::from_millis(500)).await; + assert_eq!( + check_inbox_health_with(&client, &subscription, Some(health.clone())).await, + InboxStatus::Blind, + "round {round}: a relay that refuses every REQ must never read as listening" + ); + } + + // The audit draws on the same budget the event loop does, so a relay + // that refuses on principle is paced towards `RESUBSCRIBE_MAX_BACKOFF` + // instead of being handed a REQ on every pass, forever. Skipping the + // retry must not soften the verdict: the node is still deaf here. + assert!( + backing_off(&health, &url), + "repeated refusals must accumulate on the shared re-subscribe budget" + ); + assert!(!health.allow_resubscribe(&url)); + assert_eq!( + check_inbox_health_with(&client, &subscription, Some(health)).await, + InboxStatus::Blind, + "a pass that backs off instead of re-sending must still report the inbox deaf" + ); + + relay.shutdown(); + } + + #[tokio::test] + async fn watchdog_reports_blind_when_no_relay_serves_the_inbox() { + // A client with no relays at all is the limit case of every relay + // being down: nothing can deliver a trade message. + let client = crate::util::mostro_nostr_client_options(None).build(); + let subscription = InboxSubscription::new(pubkey(), Kind::GiftWrap); + + assert_eq!( + check_inbox_health(&client, &subscription).await, + InboxStatus::Blind + ); + } + + #[tokio::test] + async fn watchdog_ignores_a_disconnected_relay() { + use nostr_sdk::local_relay::LocalRelay; + + let live = LocalRelay::builder().build(); + live.run().await.expect("run local relay"); + let live_url = live.url().await; + + let client = crate::util::mostro_nostr_client_options(None).build(); + client.add_relay(live_url.clone()).await.expect("add_relay"); + // Never connects: a relay that is down must not be re-subscribed on + // every tick, nor drag the verdict to blind while another one serves. + client + .add_relay("ws://127.0.0.1:1") + .await + .expect("add_relay"); + client.connect().await; + + let subscription = InboxSubscription::new(pubkey(), Kind::GiftWrap); + subscription.subscribe(&client).await.expect("subscribe"); + tokio::time::sleep(Duration::from_millis(500)).await; + + assert_eq!( + check_inbox_health(&client, &subscription).await, + InboxStatus::Listening, + "one healthy relay is enough to keep hearing" + ); + + live.shutdown(); + } +} diff --git a/src/main.rs b/src/main.rs index 6a636497..9eac8db5 100644 --- a/src/main.rs +++ b/src/main.rs @@ -6,6 +6,7 @@ pub mod config; pub mod db; pub mod escrow; pub mod flow; +pub mod inbox; pub mod lightning; pub mod lnurl; pub mod messages; @@ -28,6 +29,7 @@ use crate::config::{ get_db_pool, Settings, DB_POOL, LN_STATUS, MESSAGE_QUEUES, MOSTRO_CONFIG, NOSTR_CLIENT, }; use crate::db::find_held_invoices; +use crate::inbox::InboxHealth; use crate::lightning::LnStatus; use crate::lightning::LndConnector; use crate::rpc::RpcServer; @@ -104,10 +106,11 @@ async fn main() -> Result<()> { support protocol v2. See https://github.com/MostroP2P/mostro/issues/786" ); } - let subscription = Filter::new() - .pubkey(mostro_keys.public_key()) - .kind(transport.event_kind()) - .limit(0); + // Install the inbox health record before anything can observe the inbox, + // so the watchdog and the scheduler read the same one from their own tasks. + if InboxHealth::new().install_global().is_err() { + tracing::warn!("Inbox health record already installed"); + } let client = match get_nostr_client() { Ok(client) => client, @@ -118,8 +121,11 @@ async fn main() -> Result<()> { } }; - // Client subscription - client.subscribe(subscription).await?; + // The inbox REQ is sent by the event loop (`app::run` / `app::run_cashu`), + // which subscribes only after its notification stream exists. Sending it + // from here would put it ahead of any receiver, and the SDK delivers + // nothing that predates one — the relay's EOSE and any trade message + // arriving during the rest of this boot would be dropped on the floor. // Publish NIP-01 kind 0 metadata event let mostro_settings = Settings::get_mostro(); diff --git a/src/price/providers/nostr.rs b/src/price/providers/nostr.rs index 7c55e231..dac6c798 100644 --- a/src/price/providers/nostr.rs +++ b/src/price/providers/nostr.rs @@ -715,7 +715,7 @@ mod tests { #[tokio::test] #[ignore = "hits a real Nostr relay; run explicitly for manual verification"] async fn live_relay_fetch_returns_real_rates() { - let client = crate::util::price_nostr_client_options().build(); + let client = crate::util::price_nostr_client_options(None).build(); client .add_relay("wss://relay.mostro.network") .await diff --git a/src/scheduler.rs b/src/scheduler.rs index 5256c691..1eadf4d1 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -19,8 +19,9 @@ use nostr_sdk::prelude::EventBuilder; use nostr_sdk::prelude::{FinalizeEvent, Kind as NostrKind, Nip65Tag, Tag}; use std::collections::HashSet; use std::sync::Arc; +use std::time::Duration; use tokio::sync::RwLock; -use tracing::{error, info, warn}; +use tracing::{debug, error, info, warn}; use util::{enqueue_order_msg, get_nostr_relays, send_dm, update_order_event}; pub async fn start_scheduler(ctx: AppContext) { @@ -54,10 +55,58 @@ pub async fn start_scheduler(ctx: AppContext) { job_update_bitcoin_prices().await; job_flush_messages_queue(ctx.clone()).await; job_refresh_active_pubkeys(ctx.clone()).await; + job_inbox_watchdog(ctx.clone()).await; info!("Scheduler Started"); } +/// Longest the timeout job defers to an inbox that has not been confirmed +/// listening. +/// +/// Holding timeouts is the right call for an outage, but it cannot be +/// unconditional: the same pass that slashes a bond is the one that releases +/// it, and the one that cancels the seller's hold invoice. Waiting forever on +/// a permanently broken inbox would leave escrows encumbered until CLTV expiry +/// and honest takers' bonds locked indefinitely. Three hours is far longer +/// than any transient relay problem and far shorter than the CLTV horizon, so +/// an operator has time to notice while the funds never become hostage to it. +const MAX_UNCONFIRMED_INBOX_PAUSE_SECS: i64 = 3 * 3600; + +/// How often the inbox watchdog audits the subscription across relays. +/// +/// Short enough that a lost ear is measured in seconds rather than the 60s +/// timeout tick, long enough that it is not a source of traffic on its own. +/// Hardcoded like the other maintenance intervals in this module. +/// +/// Visible to `crate::inbox` because the two pacing knobs interact: the audit +/// draws on the same per-relay budget as the event loop, so this interval is +/// what decides how many doublings pass before `RESUBSCRIBE_MAX_BACKOFF` is +/// the thing actually limiting retries. +pub(crate) const INBOX_WATCHDOG_INTERVAL: u64 = 30; + +/// Audit the daemon's Nostr inbox and re-subscribe any relay that stopped +/// serving it (see `crate::inbox`). +/// +/// The event loop already reacts to a `CLOSED` frame, but only to frames it +/// actually receives — the SDK's notification channel drops them when the +/// consumer lags, and some ways of losing a subscription produce no frame at +/// all. This job is the backstop, and the only thing that notices when *every* +/// relay has gone quiet. +async fn job_inbox_watchdog(ctx: AppContext) { + #[allow(deprecated)] + let event_kind = ctx.settings().mostro.transport.event_kind(); + let subscription = crate::inbox::InboxSubscription::new(ctx.keys().public_key(), event_kind); + + tokio::spawn(async move { + loop { + // Sleep first: at startup the event loop has just subscribed, and a + // REQ still in flight would look exactly like a missing one. + tokio::time::sleep(tokio::time::Duration::from_secs(INBOX_WATCHDOG_INTERVAL)).await; + crate::inbox::check_inbox_health(ctx.nostr_client(), &subscription).await; + } + }); +} + /// Periodically rebuild the protocol-v2 anti-spam gate's active-trade-pubkey /// cache from the DB (spec §6 Phase 2). Status mutations are scattered across /// many handlers with no single choke-point, so a periodic full reload is the @@ -486,7 +535,75 @@ async fn job_cancel_orders(ctx: AppContext) { loop { info!("Check for order to republish for late actions of users"); - if let Ok(older_orders_list) = crate::db::find_order_by_seconds(pool).await { + // A timeout means "the user did not answer in time", and that + // conclusion is only sound while Mostro can hear. With the inbox + // down, an answer that was sent is simply never delivered — and + // the ten-second replay window means it is lost, not queued — so + // acting on the deadline would cancel escrows and slash bonds over + // the node's own deafness. Skip the whole tick: cancel, refund, + // republish and slash all rest on the same unsound premise. + // + // The test is "an audit confirmed we can hear", not "no audit has + // reported deafness". This job's first tick runs immediately while + // the watchdog's comes later, so an unaudited record would let a + // node that never obtained a working inbox act on that window. + // + // Waiting cannot be unconditional, though. A permanently deaf + // inbox is an operator problem, and holding every tick forever + // turns it into a second one: hold invoices stay encumbered until + // CLTV expiry and honest takers never get their bonds back, since + // the same pass that slashes is the one that releases. Past + // `MAX_UNCONFIRMED_INBOX_PAUSE_SECS` the orders are unwound + // anyway — but blamelessly, which is the part that matters. + let health = crate::inbox::InboxHealth::global(); + let mut blameless = false; + if let Some(health) = health.as_ref() { + if !health.is_confirmed_listening() { + let deaf_for = health.unconfirmed_for_secs(); + if deaf_for < MAX_UNCONFIRMED_INBOX_PAUSE_SECS { + warn!( + "scheduler_timeout: inbox not confirmed listening for {deaf_for}s, holding order timeouts" + ); + tokio::time::sleep(tokio::time::Duration::from_secs(60)).await; + continue; + } + warn!( + "scheduler_timeout: inbox unconfirmed for {deaf_for}s, past the \ + {MAX_UNCONFIRMED_INBOX_PAUSE_SECS}s bound; unwinding timed-out orders \ + WITHOUT slashing so escrows and bonds are not held until CLTV expiry" + ); + blameless = true; + } + } + + // Compensation for inbox downtime is per order, and applied in two + // steps. The query widens its window by the most any order could + // be owed, so nothing eligible is missed; the exact figure — the + // downtime that overlaps *this* order's own wait — then decides. + // A single global allowance cannot do this: it would either + // under-credit an order that waited through the whole outage or + // hand the same credit to one taken long after it ended. + // + // The credit is skipped once the pause bound is passed. By then + // the outage is hours deep, so every waiting order would be owed + // more than its deadline and none would ever be unwound — which is + // the state this branch exists to escape. Nobody is punished for + // it: `blameless` releases the bonds instead of settling them. + let max_grace = if blameless { + 0 + } else { + health.as_ref().map(|h| h.max_blind_seconds()).unwrap_or(0) + }; + if max_grace > 0 { + info!( + "scheduler_timeout: up to {max_grace}s of inbox downtime is credited against order deadlines" + ); + } + + if let Ok(older_orders_list) = + crate::db::find_order_by_seconds(pool, Duration::from_secs(max_grace.max(0) as u64)) + .await + { for order in older_orders_list.into_iter() { // The tick-start snapshot may be stale by the time this // iteration is reached — re-read and re-confirm before @@ -498,6 +615,23 @@ async fn job_cancel_orders(ctx: AppContext) { else { continue; }; + + // Give this order back the downtime it actually waited + // through. Orders taken after the outage are owed nothing + // and fall through unchanged. + if let Some(health) = health.as_ref().filter(|_| !blameless) { + let owed = health.blind_seconds_since(order.taken_at); + let waited = + nostr_sdk::prelude::Timestamp::now().as_secs() as i64 - order.taken_at; + if waited < exp_seconds as i64 + owed { + debug!( + "scheduler_timeout: order {} not late yet; {owed}s of its wait was inbox downtime", + order.id + ); + continue; + } + } + // Check if order is a sell order and Buyer is not sending the invoice for too much time. // Same if seller is not paying hold invoice if order.status == Status::WaitingBuyerInvoice.to_string() @@ -584,39 +718,50 @@ async fn job_cancel_orders(ctx: AppContext) { // `order` is the pre-mutation snapshot — its // waiting status and trade pubkeys are intact, // which the §3.1 buyer/seller → bond mapping needs. - match bond::slash_or_release_on_timeout( - pool, - &mut ln_client, - &order, - Settings::get_bond(), - ) - .await - { - Ok(Some(slashed)) => { - bond::notify_bond_slashed(&order, &slashed).await; - } - Ok(None) => {} - Err(e) => { - // `Err` from `slash_or_release_on_timeout` is a DB-read - // failure (e.g. `find_active_bonds_for_order` / - // `timeout_slash_confirmed` couldn't read the bond - // rows), so we don't yet know whether the slash - // applies. Falling through to cancel/republish - // would persist the order out of - // `find_order_by_seconds`'s waiting-state - // eligibility window, and the next tick would - // never re-evaluate it — losing the slash whose - // applicability we couldn't even determine. - // `continue` keeps the order eligible so the - // next tick re-runs the full path (the slash - // primitive is idempotent on a settled HTLC and - // a `PendingPayout` bond, so a retry that - // finds the work already done is a no-op). - tracing::warn!( - "scheduler_timeout: bond slash/release errored for {} ({}); skipping cancel/republish so next tick retries", - order.id, e - ); - continue; + // + // Past the pause bound (`blameless`) none of that + // applies: the node has been unable to hear for hours, + // so the timeout says nothing about the user. The order + // is still unwound — otherwise the escrow sits until + // CLTV expiry — but every bond is released rather than + // settled. + if blameless { + bond::release_on_timeout_without_slashing(pool, &order).await; + } else { + match bond::slash_or_release_on_timeout( + pool, + &mut ln_client, + &order, + Settings::get_bond(), + ) + .await + { + Ok(Some(slashed)) => { + bond::notify_bond_slashed(&order, &slashed).await; + } + Ok(None) => {} + Err(e) => { + // `Err` from `slash_or_release_on_timeout` is a DB-read + // failure (e.g. `find_active_bonds_for_order` / + // `timeout_slash_confirmed` couldn't read the bond + // rows), so we don't yet know whether the slash + // applies. Falling through to cancel/republish + // would persist the order out of + // `find_order_by_seconds`'s waiting-state + // eligibility window, and the next tick would + // never re-evaluate it — losing the slash whose + // applicability we couldn't even determine. + // `continue` keeps the order eligible so the + // next tick re-runs the full path (the slash + // primitive is idempotent on a settled HTLC and + // a `PendingPayout` bond, so a retry that + // finds the work already done is a no-op). + tracing::warn!( + "scheduler_timeout: bond slash/release errored for {} ({}); skipping cancel/republish so next tick retries", + order.id, e + ); + continue; + } } } diff --git a/src/util.rs b/src/util.rs index 0e8c8736..02e30da4 100644 --- a/src/util.rs +++ b/src/util.rs @@ -1423,16 +1423,36 @@ async fn update_order_event_stamped( Ok(Some(order_updated)) } +/// The identity Mostro authenticates with on NIP-42 relays, when it has one. +/// +/// Keys are set by `settings_init()` long before any client is built, so in a +/// running daemon this is always `Some`; it is `None` only in tests that skip +/// the configuration bootstrap. Saying so out loud matters because the failure +/// it causes is silent — an auth-gated relay simply stops delivering. +fn nip42_identity() -> Option<&'static Keys> { + match get_keys() { + Ok(keys) => Some(keys), + Err(e) => { + tracing::warn!( + "Nostr keys unavailable ({e}); this client cannot answer NIP-42 challenges and \ + will be refused by relays that require authentication" + ); + None + } + } +} + pub async fn connect_nostr() -> Result { let nostr_settings = Settings::get_nostr(); // Daemon inbox client: shared size limits, but **no** - // `verify_subscriptions`. The long-lived `.limit(0)` subscription in - // `main.rs` must not count pre-EOSE frames against a zero limit — that - // would drop matching trade messages before dispatch (hermeme, PR #841). + // `verify_subscriptions`. The long-lived `.limit(0)` inbox subscription + // (`crate::inbox`) must not count pre-EOSE frames against a zero limit — + // that would drop matching trade messages before dispatch (hermeme, PR + // #841). // Price queries use [`connect_price_nostr`] / [`PRICE_NOSTR_CLIENT`] with // verification enabled instead. - let client = mostro_nostr_client_options().build(); + let client = mostro_nostr_client_options(nip42_identity()).build(); // Add relays for relay in nostr_settings.relays.iter() { @@ -1452,7 +1472,7 @@ pub async fn connect_nostr() -> Result { /// daemon client, with [`price_nostr_client_options`]). pub async fn connect_price_nostr() -> Result { let nostr_settings = Settings::get_nostr(); - let client = price_nostr_client_options().build(); + let client = price_nostr_client_options(nip42_identity()).build(); for relay in nostr_settings.relays.iter() { client @@ -1499,23 +1519,45 @@ fn mostro_nostr_relay_limits() -> RelayLimits { limits } -fn client_options_from_policy(policy: MostroNostrClientPolicy) -> ClientBuilder { +fn client_options_from_policy( + policy: MostroNostrClientPolicy, + authenticate_as: Option<&Keys>, +) -> ClientBuilder { let mut builder = ClientBuilder::new().relay_limits(mostro_nostr_relay_limits()); if policy.verify_subscriptions { builder = builder.verify_subscriptions(true); } + if let Some(keys) = authenticate_as { + // NIP-42. Without an authenticator the SDK cannot answer a relay's + // AUTH challenge, and a relay that gates reads behind it answers the + // REQ with `CLOSED "auth-required: …"` — which the SDK then treats as + // permanent, dropping the subscription for good. With one, the closure + // is provisional: the client authenticates and the REQ is re-sent. + // + // Authenticating costs no privacy Mostro has not already spent. The + // AUTH event is bound to that relay's challenge and URL, so it cannot + // be replayed elsewhere, and the node publishes orders signed with this + // very key to these very relays. + builder = builder.authenticator(SignerAuthenticator::new(keys.clone())); + } builder } /// Process-wide daemon Nostr [`ClientBuilder`] (inbox / publishing). -pub(crate) fn mostro_nostr_client_options() -> ClientBuilder { - client_options_from_policy(daemon_nostr_client_policy()) +/// +/// `authenticate_as` is the identity used for NIP-42; passing `None` builds a +/// client that cannot read from auth-gated relays. +pub(crate) fn mostro_nostr_client_options(authenticate_as: Option<&Keys>) -> ClientBuilder { + client_options_from_policy(daemon_nostr_client_policy(), authenticate_as) } /// Price-provider Nostr [`ClientBuilder`]: size limits plus subscription /// filter verification, scoped away from the daemon inbox client. -pub(crate) fn price_nostr_client_options() -> ClientBuilder { - client_options_from_policy(price_nostr_client_policy()) +/// +/// Shares the daemon's relay list, so it needs the same NIP-42 identity: an +/// auth-gated relay blinds the price feed exactly like it blinds the inbox. +pub(crate) fn price_nostr_client_options(authenticate_as: Option<&Keys>) -> ClientBuilder { + client_options_from_policy(price_nostr_client_policy(), authenticate_as) } /// Which caller drove `show_hold_invoice`, and therefore which order @@ -3637,7 +3679,7 @@ mod tests { #[test] fn nostr_client_policies_scope_verify_to_price_only() { - // Daemon inbox must not enable verify_subscriptions (main.rs limit(0)). + // Daemon inbox must not enable verify_subscriptions (inbox limit(0)). assert!( !daemon_nostr_client_policy().verify_subscriptions, "daemon client must leave verify_subscriptions off" @@ -3647,8 +3689,79 @@ mod tests { "price client must enable verify_subscriptions" ); // Helpers remain constructible (SDK copies the flag into RelayOptions). - let _daemon = mostro_nostr_client_options().build(); - let _price = price_nostr_client_options().build(); + let _daemon = mostro_nostr_client_options(None).build(); + let _price = price_nostr_client_options(None).build(); + // Both accept a NIP-42 identity: they share a relay list, so an + // auth-gated relay blinds the price feed exactly like the inbox. + let keys = Keys::generate(); + let _daemon_auth = mostro_nostr_client_options(Some(&keys)).build(); + let _price_auth = price_nostr_client_options(Some(&keys)).build(); + } + + /// A local relay that requires NIP-42 authentication before it will serve + /// reads — the shape that silently blinded the daemon. + async fn nip42_read_gated_relay() -> nostr_sdk::local_relay::LocalRelay { + let relay = nostr_sdk::local_relay::LocalRelay::builder() + .nip42(nostr_sdk::local_relay::LocalRelayBuilderNip42::read()) + .build(); + relay.run().await.expect("run nip42 relay"); + relay + } + + #[tokio::test] + async fn auth_gated_relay_keeps_the_subscription_only_when_authenticating() { + use std::time::Duration; + + let relay = nip42_read_gated_relay().await; + let url = relay.url().await; + + let keys = Keys::generate(); + let seeded = EventBuilder::new(nostr::event::Kind::TextNote, "gated") + .finalize(&keys) + .expect("sign"); + relay.add_event(seeded).await.expect("seed event"); + + let filter = Filter::new().kind(nostr::event::Kind::TextNote).limit(0); + let id = SubscriptionId::new("nip42-probe"); + + // Without an identity the relay answers `auth-required`, and the SDK + // drops the subscription for good: the daemon is deaf on this relay + // with no way back. + let anonymous = mostro_nostr_client_options(None).build(); + anonymous.add_relay(url.clone()).await.expect("add_relay"); + anonymous.connect().await; + anonymous + .subscribe(filter.clone()) + .with_id(id.clone()) + .await + .expect("subscribe"); + + // With one, the closure is provisional: the client answers the + // challenge and the REQ is re-sent, so the subscription survives. + let authenticated = mostro_nostr_client_options(Some(&keys)).build(); + authenticated + .add_relay(url.clone()) + .await + .expect("add_relay"); + authenticated.connect().await; + authenticated + .subscribe(filter) + .with_id(id.clone()) + .await + .expect("subscribe"); + + tokio::time::sleep(Duration::from_secs(2)).await; + + assert!( + !anonymous.subscriptions().await.contains_key(&id), + "SDK behaviour changed: an auth-required CLOSED no longer drops the subscription" + ); + assert!( + authenticated.subscriptions().await.contains_key(&id), + "a client with a NIP-42 identity must keep its subscription on an auth-gated relay" + ); + + relay.shutdown(); } #[tokio::test] @@ -3666,7 +3779,7 @@ mod tests { .expect("mock relay"); let url = mock.url().await; - let price_client = price_nostr_client_options().build(); + let price_client = price_nostr_client_options(None).build(); price_client .add_relay(url.clone()) .await @@ -3695,14 +3808,14 @@ mod tests { use nostr_sdk::local_relay::MockRelay; use std::time::Duration; - // Clean mock (no random flood): mirrors main.rs `.limit(0)` inbox — + // Clean mock (no random flood): mirrors the `.limit(0)` inbox — // history is skipped, but live matching events after EOSE must arrive. // Daemon options intentionally omit verify_subscriptions so pre-EOSE // frames are not counted against limit 0 (hermeme, PR #841). let mock = MockRelay::run().await.expect("mock relay"); let url = mock.url().await; - let daemon = mostro_nostr_client_options().build(); + let daemon = mostro_nostr_client_options(None).build(); daemon.add_relay(url.clone()).await.expect("add_relay"); daemon.connect().await; @@ -3769,7 +3882,7 @@ mod tests { seeder.send_event(&event).await.expect("seed"); } - let price_client = price_nostr_client_options().build(); + let price_client = price_nostr_client_options(None).build(); price_client.add_relay(url).await.expect("add_relay"); price_client.connect().await;