feat(#328): source restore trade-key resync from Action::LastTradeIndex

rust/src/api/orders.rs

@@ -3557,34 +3557,41 @@ pub async fn get_trade_role(order_id: String) -> Result<Option<crate::api::types
     }
 }
 
+/// Coerce a wire trade index (`i64`) into a usable counter value, or `None`.
+///
+/// Trade indexes cross the wire as `i64` (restore payloads, the
+/// `LastTradeIndex` reply). A value that is negative or `>= u32::MAX` is not a
+/// real trade index: negatives are nonsense, and `u32::MAX` is the reserved
+/// terminal index — storing it as the counter would make the next
+/// `derive_trade_key` compute `u32::MAX + 1` and overflow (panic in debug, wrap
+/// to 0 in release, reissuing index 0 — the exact key-reuse the resync
+/// prevents). Such a value is dropped rather than truncated into the counter.
+fn sanitize_trade_index(i: i64) -> Option<u32> {
+    u32::try_from(i).ok().filter(|&v| v < u32::MAX)
+}
+
 /// Highest trade-key index across all recovered orders and disputes (#217).
 ///
 /// The counter must be raised to this so the next `derive_trade_key()` cannot
 /// hand out an index a recovered trade already owns. Returns `None` when the
-/// restore carried no trades (nothing to resync to). Indexes are `i64` on the
-/// wire; a value that is negative or beyond `u32::MAX` is not a real trade
-/// index, so it is dropped rather than truncated into the counter.
+/// restore carried no trades (nothing to resync to).
+///
+/// NOTE (#328): this is only a *lower bound* of the daemon's real counter —
+/// the restore payload lists only non-finalized orders, so a finalized trade
+/// holding a higher index is invisible here. `restore_session` sources its
+/// resync floor from `last_trade_index()` (authoritative) and falls back to
+/// this only when the daemon does not answer.
 fn recovered_max_trade_index(
     info: &mostro_core::message::RestoreSessionInfo,
 ) -> Option<u32> {
-    // A single adapter drops both negatives and any value >= u32::MAX — neither
-    // is a real trade index, and truncating one into a small u32 could corrupt
-    // the counter this exists to protect. u32::MAX itself is dropped: it is the
-    // reserved terminal index, and storing it as the counter would make the next
-    // derive_trade_key compute u32::MAX + 1 and overflow (panic in debug, wrap to
-    // 0 in release — reissuing index 0, the exact key-reuse this resync prevents).
-    // Collapsed into one filter_map so the two conditions can't drift apart.
     let all: Vec<i64> = info
         .restore_orders
         .iter()
         .map(|o| o.trade_index)
         .chain(info.restore_disputes.iter().map(|d| d.trade_index))
         .collect();
     let total = all.len();
-    let valid: Vec<u32> = all
-        .into_iter()
-        .filter_map(|i| u32::try_from(i).ok().filter(|&v| v < u32::MAX))
-        .collect();
+    let valid: Vec<u32> = all.iter().copied().filter_map(sanitize_trade_index).collect();
     // A dropped index is not just an odd value: it means the daemon sent
     // something this client's model does not cover, and a silently-lowered
     // floor produces a later CantDo(InvalidTradeIndex) with no breadcrumb. Warn
@@ -3604,6 +3611,196 @@ fn recovered_max_trade_index(
     valid.into_iter().max()
 }
 
+/// Pick the resync floor as the highest of the daemon counter and the
+/// restore-payload maximum (#328).
+///
+/// The daemon's `LastTradeIndex` answer (`daemon_counter`) is authoritative and
+/// is the real high-water mark, including finalized trades. The payload maximum
+/// is a proven lower bound — every recovered order carries its own index.
+///
+/// Against a consistent daemon the counter is always `>=` the payload maximum:
+/// the daemon raises `last_trade_index` to every index it accepts
+/// (`update_user_trade_index`) and rejects any index it has already seen, so an
+/// order it still returns in the restore payload was necessarily seen at or
+/// below the counter. Taking the max is therefore a no-op in practice — kept as
+/// cheap defense-in-depth so the floor stays correct independent of that
+/// invariant: a stale/partial reply or a daemon bug can never make us resync
+/// below a recovered trade's own index and reuse its key. Returns `None` only
+/// when neither source has a usable index.
+fn resync_floor(
+    daemon_counter: Option<u32>,
+    info: &mostro_core::message::RestoreSessionInfo,
+) -> Option<u32> {
+    let payload_max = recovered_max_trade_index(info);
+    match (daemon_counter, payload_max) {
+        (Some(daemon), Some(payload)) => Some(daemon.max(payload)),
+        (daemon, payload) => daemon.or(payload),
+    }
+}
+
+/// True only for the reply to THIS `LastTradeIndex` request: the action
+/// matches and the daemon echoed our correlation nonce
+/// (`mostro/src/app/last_trade_index.rs` copies `request_id` into the reply).
+///
+/// A replayed reply from an earlier request carries a different nonce — or
+/// none, since this client's own pre-hardening requests sent no id — so
+/// accepting `None` would readmit exactly the replays this guards against.
+/// Strict matching means a daemon that does not echo falls back to the
+/// restore-payload maximum, the same designed path as a silent daemon.
+fn is_matching_last_trade_index_reply(
+    kind: &mostro_core::message::MessageKind,
+    request_id: u64,
+) -> bool {
+    kind.action == mostro_core::message::Action::LastTradeIndex
+        && kind.request_id == Some(request_id)
+}
+
+/// Ask the daemon for its authoritative last-known trade index (#328).
+///
+/// `Action::LastTradeIndex` is account-scoped: the request is signed with the
+/// identity keys for BOTH the Seal and the rumor (see
+/// `actions::last_trade_index`), so it derives NO trade key and the daemon
+/// resolves the account by sender pubkey. The reply is a kind-14 addressed to
+/// the identity pubkey, carrying the counter in `MessageKind::trade_index`.
+///
+/// Returns `Ok(Some(idx))` with the sanitized counter, or `Ok(None)` when the
+/// daemon does not answer within the timeout or the reply carries no usable
+/// index — the caller then falls back to `recovered_max_trade_index`.
+///
+/// This is a self-contained request/reply (own subscription + inline wait, like
+/// mostro-cli's `wait_for_dm`) rather than a reuse of the per-trade
+/// subscription/dispatch path: that path derives its recipient keys from a
+/// trade index and is coupled to trade keys, but this reply is encrypted to the
+/// identity key.
+async fn last_trade_index() -> Result<Option<u32>> {
+    use nostr_sdk::RelayPoolNotification;
+    use crate::rt::time::{timeout, Duration};
+
+    let identity_keys = crate::api::identity::get_active_keys().await?;
+    let identity_pk = identity_keys.public_key();
+    let identity_pk_hex = identity_pk.to_hex();
+    let mostro_pubkey = nostr_sdk::PublicKey::from_hex(&active_mostro_pubkey())?;
+
+    let pool = crate::api::nostr::get_pool()?;
+    let client = pool.client();
+
+    // Grab the notifications receiver BEFORE subscribing so the reply can't
+    // arrive in the gap between subscribe and the first recv.
+    let mut rx = client.notifications();
+
+    // This query, unlike the restore itself, has a fallback (the payload
+    // maximum), so a shorter wait halves the worst-case restore latency
+    // against a silent daemon. Shared by the relay-side auto-close and the
+    // outer wait loop so the two can't drift.
+    const REPLY_TIMEOUT: Duration = Duration::from_secs(5);
+
+    // limit(0): live-only, same rationale as subscribe_daemon_messages — the
+    // reply is published after we subscribe, and we never want a replayed
+    // historical LastTradeIndex to resolve this request.
+    let filter = nostr_sdk::Filter::new()
+        .kind(nostr_sdk::Kind::PrivateDirectMessage)
+        .author(mostro_pubkey)
+        .pubkey(identity_pk)
+        .limit(0);
+    // Auto-close the relay-side subscription — this is a one-shot request/reply
+    // (mostro-cli's wait_for_dm shape), not a long-lived watcher, so the library
+    // issues the CLOSE on every path: after the reply (WaitForEventsAfterEOSE(1))
+    // and, if the daemon never answers, on the timeout. Leaving it to manual
+    // bookkeeping is the leak class #182/#255 address. Auto-close subs are
+    // deliberately excluded from reconnect re-subscription (correct here: if the
+    // socket drops mid-request we fall back to the payload maximum by design).
+    let close_opts = nostr_sdk::prelude::SubscribeAutoCloseOptions::default()
+        .exit_policy(nostr_sdk::prelude::ReqExitPolicy::WaitForEventsAfterEOSE(1))
+        .timeout(Some(REPLY_TIMEOUT));
+    if let Err(e) = client.subscribe(filter, Some(close_opts)).await {
+        log::warn!("[orders] last_trade_index subscribe failed: {e}");
+        return Ok(None);
+    }
+
+    // Correlation nonce, echoed by the daemon in its reply. Without it any
+    // authenticated LastTradeIndex reply resolves this request, so a malicious
+    // relay could replay an old one. Monotonicity caps the damage (the max in
+    // resync_floor means a stale counter degrades to the payload fallback, the
+    // same as a silent daemon) — but the daemon echoes request_id, so binding
+    // the reply to this request costs nothing.
+    let request_id: u64 = {
+        use rand::RngCore;
+        rand::rngs::OsRng.next_u64().max(1) // 0 is indistinguishable from "unset"
+    };
+    let event_json =
+        actions::last_trade_index(&identity_keys, &mostro_pubkey, request_id).await?;
+    publish_event_json(&event_json).await?;
+    crate::api::logging::blog_info(
+        "restore",
+        "LastTradeIndex published — waiting for daemon".to_string(),
+    );
+
+    // Wait for the reply. Bounded by REPLY_TIMEOUT — the same budget the
+    // relay-side auto-close uses, so both give up together.
+    let deadline = REPLY_TIMEOUT;
+    let start = crate::rt::time::Instant::now();
+    loop {
+        let remaining = deadline.saturating_sub(start.elapsed());
+        if remaining.is_zero() {
+            break;
+        }
+        match timeout(remaining, rx.recv()).await {
+            Ok(Ok(RelayPoolNotification::Event { event, .. })) => {
+                if event.kind != nostr_sdk::Kind::PrivateDirectMessage
+                    || event.pubkey != mostro_pubkey
+                {
+                    continue;
+                }
+                let is_for_us = event.tags.iter().any(|t| {
+                    let s = t.as_slice();
+                    s.first().map(|v| v.as_str()) == Some("p")
+                        && s.get(1).map(|v| v.as_str()) == Some(identity_pk_hex.as_str())
+                });
+                if !is_for_us {
+                    continue;
+                }
+                match crate::nostr::transport::unwrap_mostro_message(&identity_keys, &event).await {
+                    Ok(Some(unwrapped)) => {
+                        // Authenticate: the kind-14 author must be the node.
+                        if unwrapped.sender != mostro_pubkey {
+                            continue;
+                        }
+                        let kind = unwrapped.message.get_inner_message_kind();
+                        if !is_matching_last_trade_index_reply(kind, request_id) {
+                            continue;
+                        }
+                        let idx = kind.trade_index.and_then(sanitize_trade_index);
+                        crate::api::logging::blog_info("restore", format!(
+                            "LastTradeIndex reply: trade_index={:?} -> floor={idx:?}",
+                            kind.trade_index
+                        ));
+                        return Ok(idx);
+                    }
+                    Ok(None) => continue,
+                    Err(e) => {
+                        log::warn!("[orders] last_trade_index decrypt failed: {e}");
+                        continue;
+                    }
+                }
+            }
+            Ok(Ok(RelayPoolNotification::Shutdown)) => break,
+            Ok(Err(broadcast::error::RecvError::Lagged(n))) => {
+                log::warn!("[orders] last_trade_index lagged by {n} messages");
+                continue;
+            }
+            Ok(Err(broadcast::error::RecvError::Closed)) => break,
+            Err(_) => break, // timeout
+            Ok(Ok(_)) => continue,
+        }
+    }
+    crate::api::logging::blog_warn(
+        "restore",
+        "LastTradeIndex: no usable daemon reply — falling back to restore payload max"
+            .to_string(),
+    );
+    Ok(None)
+}
+
 /// Send a `RestoreSession` to the active daemon and return the user's active
 /// trades/disputes. Mirrors create_order's send/await, minus the order payload.
 ///
@@ -3677,13 +3874,29 @@ pub async fn restore_session() -> Result<mostro_core::message::RestoreSessionInf
 
     match confirmation {
         Ok(Ok(Wake { reply: DaemonReply::Restored(info), .. })) => {
-            // #217: raise trade_key_index past every recovered trade before
-            // returning, so the next derive_trade_key() can't reuse a key a
-            // recovered trade already owns. Monotonic and idempotent. A persist
-            // failure fails the restore: an un-resynced counter reopens the
-            // key-reuse bug this closes, so silent success would be worse than
-            // a surfaced error the caller can retry.
-            if let Some(floor) = recovered_max_trade_index(&info) {
+            // #217: raise trade_key_index before returning, so the next
+            // derive_trade_key() can't reuse a key a recovered trade already
+            // owns. Monotonic and idempotent. A persist failure fails the
+            // restore: an un-resynced counter reopens the key-reuse bug this
+            // closes, so silent success would be worse than a surfaced error
+            // the caller can retry.
+            //
+            // #328: the authoritative floor is the daemon's LastTradeIndex
+            // counter. The restore payload lists only non-finalized orders, so
+            // recovered_max_trade_index is a lower bound (a finalized trade
+            // holding a higher index is invisible) — kept only as a fallback
+            // for when the daemon does not answer.
+            let daemon_counter = match last_trade_index().await {
+                Ok(idx) => idx,
+                Err(e) => {
+                    crate::api::logging::blog_warn("restore", format!(
+                        "LastTradeIndex request errored ({e}); \
+                         falling back to restore payload max"
+                    ));
+                    None
+                }
+            };
+            if let Some(floor) = resync_floor(daemon_counter, &info) {
                 crate::api::identity::ensure_trade_key_index_at_least(floor).await?;
             }
             Ok(info)
@@ -3813,6 +4026,63 @@ mod tests {
         );
     }
 
+    // ── #328 sanitize_trade_index / resync_floor ─────────────────────────────
+    #[test]
+    fn sanitize_trade_index_drops_negative_and_out_of_range() {
+        assert_eq!(sanitize_trade_index(0), Some(0));
+        assert_eq!(sanitize_trade_index(42), Some(42));
+        assert_eq!(sanitize_trade_index(-1), None);
+        // u32::MAX is the reserved terminal index (dropped to avoid +1 overflow).
+        assert_eq!(sanitize_trade_index(i64::from(u32::MAX)), None);
+        assert_eq!(sanitize_trade_index(i64::from(u32::MAX) - 1), Some(u32::MAX - 1));
+        assert_eq!(sanitize_trade_index(i64::from(u32::MAX) + 1), None);
+    }
+
+    #[test]
+    fn resync_floor_takes_the_higher_of_daemon_counter_and_payload_max() {
+        // The #328 scenario: order X open at index 1 (the only non-finalized
+        // trade the restore returns), order Y canceled at index 2. The payload
+        // max is 1, but the daemon's LastTradeIndex counter is 2 — the real
+        // high-water mark — and must win, or the first new order collides.
+        assert_eq!(resync_floor(Some(2), &restore_info(vec![1], vec![])), Some(2));
+        // Defense-in-depth: never resync below a recovered trade's own index.
+        // A consistent daemon cannot answer below an index it still tracks (it
+        // raised last_trade_index when it accepted that order), so this only
+        // guards a stale/partial reply — the payload lower bound then wins.
+        assert_eq!(resync_floor(Some(1), &restore_info(vec![5], vec![])), Some(5));
+        // Daemon present, payload empty -> the daemon value.
+        assert_eq!(resync_floor(Some(7), &restore_info(vec![], vec![])), Some(7));
+    }
+
+    #[test]
+    fn a_replayed_last_trade_index_reply_is_rejected() {
+        use mostro_core::message::{Action, MessageKind};
+
+        let reply = |request_id: Option<u64>| {
+            MessageKind::new(None, request_id, Some(7), Action::LastTradeIndex, None)
+        };
+        // The genuine reply echoes our nonce.
+        assert!(is_matching_last_trade_index_reply(&reply(Some(42)), 42));
+        // A replay of an earlier request's reply carries a different nonce...
+        assert!(!is_matching_last_trade_index_reply(&reply(Some(41)), 42));
+        // ...or none at all — this client's own earlier requests sent no id,
+        // so their stored replies are exactly the replay material to reject.
+        assert!(!is_matching_last_trade_index_reply(&reply(None), 42));
+        // A different action never matches, even with the right nonce.
+        let other =
+            MessageKind::new(None, Some(42), Some(7), Action::RestoreSession, None);
+        assert!(!is_matching_last_trade_index_reply(&other, 42));
+    }
+
+    #[test]
+    fn resync_floor_falls_back_to_payload_max_when_daemon_is_silent() {
+        // No LastTradeIndex answer (timeout / error): the restore-payload
+        // maximum is the best available lower bound.
+        assert_eq!(resync_floor(None, &restore_info(vec![3, 9], vec![4])), Some(9));
+        // Nothing anywhere -> no resync (None).
+        assert_eq!(resync_floor(None, &restore_info(vec![], vec![])), None);
+    }
+
     fn insert_pending_create(key: &str, request_id: u64) -> tokio::sync::oneshot::Receiver<Wake> {
         let (tx, rx) = tokio::sync::oneshot::channel::<Wake>();
         pending_requests().lock().unwrap().insert(

rust/src/mostro/actions.rs

@@ -13,7 +13,7 @@
 /// arguments — see `api::identity::get_transport_identity_keys`, which
 /// applies the runtime privacy toggle.
 use anyhow::Result;
-use mostro_core::message::{Action, Message, Payload};
+use mostro_core::message::{Action, Message, MessageKind, Payload};
 use nostr_sdk::prelude::*;
 use uuid::Uuid;
 
@@ -419,6 +419,32 @@ pub async fn restore_session(
     wrap_message_first_contact(identity_keys, trade_keys, mostro_pubkey, &msg).await
 }
 
+/// Build and wrap a `LastTradeIndex` request (#328).
+///
+/// Account-scoped: `identity_keys` sign BOTH the Seal and the rumor, so the
+/// daemon resolves the account by sender pubkey and NO trade key is derived.
+/// The reply carries the counter in `MessageKind::trade_index`. Payload must be
+/// `None` (enforced by mostro-core). Mirrors `mostro-cli`'s
+/// `execute_last_trade_index`, which signs with `identity_keys` for both.
+///
+/// `request_id` is the correlation nonce the daemon echoes in its reply
+/// (`mostro/src/app/last_trade_index.rs`) — the caller uses it to reject
+/// replayed replies from earlier requests.
+pub async fn last_trade_index(
+    identity_keys: &Keys,
+    mostro_pubkey: &PublicKey,
+    request_id: u64,
+) -> Result<String> {
+    let msg = Message::Restore(MessageKind::new(
+        None,
+        Some(request_id),
+        None,
+        Action::LastTradeIndex,
+        None,
+    ));
+    wrap_message_first_contact(identity_keys, identity_keys, mostro_pubkey, &msg).await
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;