fix: keep the Nostr inbox alive across relay closures

docs/EVENT_ROUTING.md

@@ -6,6 +6,43 @@ 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/mod.rs` (subscription, keeper, watchdog audit) and `src/inbox/health.rs` (the health record the scheduler reads)
+- 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 the REQ for those two rather than racing it.
+
+It does not stand down on the health verdict. `MarkAsClosed` leaves the entry in the SDK's subscription map, so the relay keeps *looking* subscribed, and the SDK does not always get around to re-sending: `rate-limited` has no retry timer at all (only the next reconnect, which never comes on a healthy connection), and an `auth-required` whose AUTH the relay then rejects ends without a re-subscribe. The relay's acknowledgement is therefore dropped either way, which hands the case to the watchdog below — the right pace for a relay that has just asked to be left alone, and the difference between recovering and being silently deaf for the life of the connection.
+
+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. `find_order_by_seconds` selects on the nominal deadline alone — the credit is applied only in the scheduler's loop, order by order, where it can only ever spare.
+
+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. A failed release keeps the order in its waiting state so the next tick retries, exactly as the slashing path does: cancelling first would take the order out of `find_order_by_seconds`'s eligibility window with the bond still `Locked` and nothing left to look at it again.
+
 ## Dispatch
 - Router: `src/app.rs:handle_message_action`
 - Maps `Action` → module function under `src/app/*`.
@@ -26,20 +63,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
 ```

docs/STARTUP_AND_CONFIG.md

@@ -100,6 +100,12 @@ Configuration is loaded from `~/.mostro/settings.toml` (template: `settings.tpl.
 - `relays` (Vec<String>): 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

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/mod.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, …)`;

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;
     }
 }
 

src/app/bond/flow.rs

@@ -642,17 +642,25 @@ pub async fn release_bonds_for_order_or_warn(
     }
 }
 
-/// Like [`release_bonds_for_order_or_warn`] but **retains the maker's
-/// bond** — the waiting-timeout republish path (see [`release_active_bonds`]).
-/// The maker's `Locked` bond stays put because the order returns to the
-/// book with the maker still committed; only the abandoning taker side is
-/// released.
+/// Like [`release_bonds_for_order`] but **retains the maker's bond** — the
+/// waiting-timeout republish path (see [`release_active_bonds`]). The maker's
+/// `Locked` bond stays put because the order returns to the book with the
+/// maker still committed; only the abandoning taker side is released.
+pub async fn release_taker_bonds_for_order(
+    pool: &Pool<Sqlite>,
+    order_id: Uuid,
+) -> Result<(), MostroError> {
+    release_active_bonds(pool, order_id, true).await
+}
+
+/// Best-effort [`release_taker_bonds_for_order`], for the call sites that
+/// cannot act on the failure anyway (see [`release_bonds_for_order_or_warn`]).
 pub async fn release_taker_bonds_for_order_or_warn(
     pool: &Pool<Sqlite>,
     order_id: Uuid,
     context: &'static str,
 ) {
-    if let Err(e) = release_active_bonds(pool, order_id, true).await {
+    if let Err(e) = release_taker_bonds_for_order(pool, order_id).await {
         warn!("{context}: bond release failed for {}: {}", order_id, e);
     }
 }