diff --git a/rust/src/api/orders.rs b/rust/src/api/orders.rs index d9fcbb8a..8087205f 100644 --- a/rust/src/api/orders.rs +++ b/rust/src/api/orders.rs @@ -27,27 +27,7 @@ fn trade_key_map() -> &'static std::sync::RwLock> { TRADE_KEY_MAP.get_or_init(|| std::sync::RwLock::new(HashMap::new())) } -/// Maps `trade_pubkey_hex` → `trade_key_index` for newly created maker orders. -/// -/// The daemon assigns its own UUID to a new order and publishes it as a Kind -/// 38383 event signed by the daemon (not the maker), so there is no way to -/// derive the real order ID from the Kind 38383 event itself. Instead we keep -/// a short-lived entry keyed by the trade public key and call -/// `resolve_maker_order` once the daemon's gift-wrapped acknowledgement arrives -/// and the real order ID is known. -static PENDING_MAKER_KEYS: OnceLock>> = OnceLock::new(); - -fn pending_maker_keys() -> &'static std::sync::RwLock> { - PENDING_MAKER_KEYS.get_or_init(|| std::sync::RwLock::new(HashMap::new())) -} - -fn store_pending_maker_key(trade_pubkey_hex: &str, index: u32) { - if let Ok(mut map) = pending_maker_keys().write() { - map.insert(trade_pubkey_hex.to_string(), index); - } -} - -// ── Daemon confirmation channel for create_order ───────────────────────────── +// ── Pending create_order bookkeeping ───────────────────────────────────────── /// Result sent by the gift-wrap handler to the waiting `create_order` call. enum DaemonConfirmation { @@ -57,61 +37,135 @@ enum DaemonConfirmation { Rejected { reason: String, message: String }, } -/// Maps `trade_pubkey_hex` → oneshot sender for the pending `create_order` call. -static PENDING_CONFIRMATIONS: OnceLock>>> = OnceLock::new(); - -fn pending_confirmations() -> &'static std::sync::Mutex>> { - PENDING_CONFIRMATIONS.get_or_init(|| std::sync::Mutex::new(HashMap::new())) +/// Everything one `create_order` attempt needs tracked until the daemon's +/// reply is consumed. +/// +/// `request_id` is the correlation nonce sent in the outgoing `new-order` +/// message; the daemon echoes it in both the `NewOrder` confirmation and any +/// `CantDo` rejection. Only a reply carrying the matching nonce may resolve +/// or consume this record — stale events replayed by relays carry a different +/// (or no) `request_id` and must leave every part of it in place for the +/// genuine reply. Keeping the waiter channel, the trade index, and the +/// local-UUID bridge in one record keyed by the attempt's fresh trade key +/// means an uncorrelated event cannot consume state belonging to a live (or +/// concurrent) create. +struct PendingCreate { + request_id: u64, + trade_index: u32, + /// Locally-generated UUID the order was created under before the daemon + /// assigned the real one. Bridged to the daemon UUID on confirmation. + local_uuid: String, + /// Content fingerprint (see `order_content_key`) — lets the Kind 38383 + /// subscription find this record when the daemon's public event arrives + /// (that event carries neither our trade pubkey nor a request_id). + content_key: String, + /// `Some` while the `create_order` call is blocked waiting. The 10s + /// timeout detaches only this sender and leaves the rest of the record, + /// so a genuine late reply still reconciles trade-key and id bindings + /// instead of being indistinguishable from a stale replay. + tx: Option>, } -/// Maps `content_key` → `local_uuid` for newly created maker orders. +/// Maps `trade_pubkey_hex` → the pending `create_order` attempt. /// -/// At create time the order is added to the order book with a locally-generated -/// UUID. When the daemon later publishes its own Kind 38383 event (with a -/// daemon-assigned UUID), the subscription loop uses this map to remove the -/// stale local entry so there is only one entry in the book — keyed by the -/// daemon's UUID — and `cancel_order` sends the correct ID. -static PENDING_LOCAL_IDS: OnceLock>> = OnceLock::new(); - -fn pending_local_ids() -> &'static std::sync::RwLock> { - PENDING_LOCAL_IDS.get_or_init(|| std::sync::RwLock::new(HashMap::new())) +/// The daemon assigns its own UUID to a new order and publishes it as a Kind +/// 38383 event signed by the daemon (not the maker), so the real order ID is +/// only learnable from the gift-wrapped acknowledgement; this record carries +/// the correlation state needed to consume that acknowledgement safely. +static PENDING_CREATES: OnceLock>> = OnceLock::new(); + +fn pending_creates() -> &'static std::sync::Mutex> { + PENDING_CREATES.get_or_init(|| std::sync::Mutex::new(HashMap::new())) +} + +/// True when a daemon reply carrying `got` may resolve a waiter that expects +/// `expected`. Replies must echo the exact nonce — `None` (stale replays, +/// unsolicited events) never matches. +fn request_id_matches(expected: u64, got: Option) -> bool { + got == Some(expected) } -fn store_pending_local_id(content_key: &str, local_uuid: &str) { - if let Ok(mut map) = pending_local_ids().write() { - map.insert(content_key.to_string(), local_uuid.to_string()); +/// Remove and return the pending create for `trade_pubkey_hex` **only** when +/// `got` echoes its `request_id`. A mismatched or absent id leaves the record +/// in place: relays can replay historical events, and a stale +/// `NewOrder`/`CantDo` must not confirm, reject, or reconcile a create — the +/// genuine reply (carrying the nonce) arrives later and finds the record. +fn take_matching_create(trade_pubkey_hex: &str, got: Option) -> Option { + let mut map = pending_creates().lock().ok()?; + match map.get(trade_pubkey_hex) { + Some(p) if request_id_matches(p.request_id, got) => map.remove(trade_pubkey_hex), + Some(_) => { + crate::api::logging::blog_debug("gift-wrap", format!( + "request_id {got:?} does not match pending create for trade={} — \ + leaving record for the genuine reply", + &trade_pubkey_hex[..8] + )); + None + } + None => None, } } -fn take_pending_local_id(content_key: &str) -> Option { - pending_local_ids() - .write() - .ok() - .and_then(|mut m| m.remove(content_key)) +/// Remove and return the pending create whose content fingerprint equals +/// `content_key` — used by the Kind 38383 subscription to bridge the local +/// UUID once the daemon's public event arrives. Records with a live waiter +/// (`tx` is `Some`) are left alone: the in-flight `create_order` call owns +/// the reconciliation and must still find its record when the kind-14 +/// acknowledgement lands. +fn take_pending_create_by_content_key(content_key: &str) -> Option { + let mut map = pending_creates().lock().ok()?; + let key = map + .iter() + .find(|(_, p)| p.content_key == content_key && p.tx.is_none()) + .map(|(k, _)| k.clone())?; + map.remove(&key) } -/// Resolve a pending maker order once the daemon's gift-wrapped acknowledgement -/// is processed and the real order ID becomes known. -/// -/// Moves the entry from `PENDING_MAKER_KEYS` into the regular `TRADE_KEY_MAP` -/// so that subsequent maker actions (e.g. cancel) can find the trade key via -/// `get_trade_key_index(real_order_id)`. -pub async fn resolve_maker_order(order_id: &str, trade_pubkey_hex: &str) { - let index = pending_maker_keys() - .read() - .ok() - .and_then(|m| m.get(trade_pubkey_hex).copied()); - if let Some(idx) = index { - if let Ok(mut map) = pending_maker_keys().write() { - map.remove(trade_pubkey_hex); +/// Detach the waiter channel from the pending create for `trade_pubkey_hex`, +/// leaving the record itself in place. Called on the 10s timeout: the caller +/// stops waiting, but the record must survive so a genuine late reply still +/// reconciles (and a stale replay still cannot). +fn detach_create_waiter(trade_pubkey_hex: &str) { + if let Ok(mut m) = pending_creates().lock() { + if let Some(p) = m.get_mut(trade_pubkey_hex) { + p.tx = None; } - store_trade_key_index(order_id, idx).await; - log::info!("[orders] resolved maker order={order_id} trade_index={idx}"); - } else { - log::warn!("[orders] resolve_maker_order: no pending entry for pubkey={trade_pubkey_hex}"); } } +/// Drop the pending create for `trade_pubkey_hex` unconditionally — publish +/// failure rollback, or end of the per-trade subscription's lifetime. +fn remove_pending_create(trade_pubkey_hex: &str) { + if let Ok(mut m) = pending_creates().lock() { + m.remove(trade_pubkey_hex); + } +} + +/// Local UUID of the pending create for `trade_pubkey_hex`, if any — a +/// read-only peek used to decide whether a stored order id is ours to rebind. +fn pending_local_uuid_for(trade_pubkey_hex: &str) -> Option { + pending_creates() + .lock() + .ok()? + .get(trade_pubkey_hex) + .map(|p| p.local_uuid.clone()) +} + +/// True when the order id stored for a trade may be rebound to `incoming_id`. +/// +/// Only the locally-generated UUID of this trade key's own pending create is +/// ever ours to rebind (local → daemon). A stored id that is not that UUID is +/// either already the daemon's (nothing to do) or belongs to an earlier life +/// of a reused trade key — rebinding it to whatever id an incoming event +/// carries would let a stale replay corrupt a confirmed order. +fn may_reconcile_stored_id( + stored_id: &str, + incoming_id: &str, + pending_local_uuid: Option<&str>, +) -> bool { + stored_id != incoming_id && pending_local_uuid == Some(stored_id) +} + /// Build a stable content key for a maker order. /// /// The key is stored in `TRADE_KEY_MAP` at creation time (prefixed with @@ -357,7 +411,10 @@ static PROCESSED_GW: OnceLock bool { - const MAX_ENTRIES: usize = 128; + // Sized for the global feed's history replay on reused keys: a mass + // replay longer than this window would evict ids that a slower relay + // may still redeliver within the same session. + const MAX_ENTRIES: usize = 512; let deque = PROCESSED_GW.get_or_init(|| std::sync::Mutex::new(std::collections::VecDeque::new())); let mut guard = match deque.lock() { Ok(g) => g, @@ -483,15 +540,13 @@ pub async fn create_order(params: NewOrderParams) -> Result { ¶ms_for_dispatch.payment_method, ); - // Register all bookkeeping entries before publishing the event. + // Register the trade-key mappings before publishing the event. // The daemon can respond with a Kind 38383 event within milliseconds; if // we stored these after publish the subscription loop could arrive before // the keys are written and miss the fingerprint match entirely. store_trade_key_index(&order.id, trade_index).await; // local UUID fallback store_trade_key_index(&ck, trade_index).await; // content fingerprint let trade_pk_hex = sender_keys.public_key().to_hex(); - store_pending_maker_key(&trade_pk_hex, trade_index); - store_pending_local_id(&ck, &order.id); // DO NOT add to order book or DB yet — wait for daemon confirmation first. // This avoids a phantom "pending" order when the daemon rejects (CantDo). @@ -499,20 +554,42 @@ pub async fn create_order(params: NewOrderParams) -> Result { let mostro_pubkey = nostr_sdk::PublicKey::from_hex(&active_mostro_pubkey())?; let identity_keys = crate::api::identity::get_transport_identity_keys(&sender_keys).await?; + + // Correlation nonce for this create attempt. The daemon echoes it in its + // reply (NewOrder or CantDo); only a reply carrying it may resolve the + // confirmation below. + let request_id: u64 = { + use rand::RngCore; + rand::rngs::OsRng.next_u64().max(1) // 0 is indistinguishable from "unset" + }; + let event_json = actions::new_order( &identity_keys, &sender_keys, &mostro_pubkey, ¶ms_for_dispatch, trade_index, + request_id, ) .await?; - // Set up the confirmation channel AFTER building the event but BEFORE - // publishing, so the entry is in the map before any response can arrive. + // Register the pending-create record AFTER building the event but BEFORE + // publishing, so it is in the map before any response can arrive. The + // record carries everything the dispatcher needs to consume the daemon's + // reply: the waiter channel, and the correlation/bridging state that must + // only ever be touched by a reply echoing this attempt's request_id. let (conf_tx, conf_rx) = tokio::sync::oneshot::channel::(); - if let Ok(mut map) = pending_confirmations().lock() { - map.insert(trade_pk_hex.clone(), conf_tx); + if let Ok(mut map) = pending_creates().lock() { + map.insert( + trade_pk_hex.clone(), + PendingCreate { + request_id, + trade_index, + local_uuid: order.id.clone(), + content_key: ck.clone(), + tx: Some(conf_tx), + }, + ); } // Subscribe to gift-wrap responses AFTER registering the confirmation @@ -523,9 +600,7 @@ pub async fn create_order(params: NewOrderParams) -> Result { if let Err(e) = publish_event_json(&event_json).await { // Rollback all in-memory bookkeeping on publish failure. if let Ok(mut m) = trade_key_map().write() { m.remove(&order.id); m.remove(&ck); } - if let Ok(mut m) = pending_maker_keys().write() { m.remove(&trade_pk_hex); } - if let Ok(mut m) = pending_local_ids().write() { m.remove(&ck); } - if let Ok(mut m) = pending_confirmations().lock() { m.remove(&trade_pk_hex); } + remove_pending_create(&trade_pk_hex); return Err(e); } @@ -542,9 +617,15 @@ pub async fn create_order(params: NewOrderParams) -> Result { conf_rx, ).await; - // Clean up the pending entry regardless of outcome. - if let Ok(mut map) = pending_confirmations().lock() { - map.remove(&trade_pk_hex); + // On success or rejection the dispatcher already consumed the record + // (take_matching_create). On timeout, detach only the waiter channel and + // leave the record in place: a genuine late reply must still be able to + // reconcile the trade-key and id bindings, and only the echoed nonce can + // consume what remains — a stale replay still cannot. The record's + // lifetime is bounded by the per-trade subscription (see + // subscribe_gift_wraps), which removes it when the subscription ends. + if !matches!(confirmation, Ok(Ok(_))) { + detach_create_waiter(&trade_pk_hex); } // Resolve the daemon's verdict. The order only exists once the daemon @@ -949,8 +1030,8 @@ pub async fn cancel_order(order_id: String) -> Result<()> { /// responses. /// /// Called immediately after creating a new maker order. Handles: -/// - `Action::NewOrder` — daemon confirmed the order; bridges daemon UUID into -/// `TRADE_KEY_MAP` via `resolve_maker_order`. +/// - `Action::NewOrder` — daemon confirmed the order; consumes the pending +/// create record and bridges the daemon UUID into `TRADE_KEY_MAP`. /// - All other actions are logged (full trade-session routing is Phase 7+). /// /// The relay subscription is established synchronously (awaited) before returning, @@ -987,10 +1068,26 @@ pub(crate) async fn subscribe_gift_wraps(trade_pubkey: nostr_sdk::PublicKey, tra // Protocol v2: kind-14 NIP-44 replies authored by Mostro, p-tagged to // this trade key. + // + // `limit(0)` makes this a live-only subscription: relays return no + // stored events, only events published after subscribe. In normal + // operation the key is freshly derived and has no history — the guard + // protects the cases where key reuse happens anyway: a mnemonic + // re-imported on another device resets the trade key counter to 0 (no + // last-trade-index resync yet), re-deriving keys whose full reply + // history sits on the relays; any future counter regression does the + // same. Replayed replies from an earlier life of the key are what used + // to falsely resolve waiting create_order calls. + // mostro-cli (`wait_for_dm`) and MostriX (waiter subscriptions) use the + // same pattern for the same purpose. Unlike a `since` cutoff, `limit(0)` + // never touches live events, so it cannot drop the genuine reply when + // the client clock runs ahead of the daemon's. Offline catch-up is the + // global feed's job (see subscribe_node_filters), which replays history. let filter = nostr_sdk::Filter::new() .kind(nostr_sdk::Kind::PrivateDirectMessage) .author(mostro_pubkey) - .pubkey(trade_pubkey); + .pubkey(trade_pubkey) + .limit(0); if let Err(e) = client.subscribe(filter, None).await { log::warn!("[orders] subscribe_gift_wraps subscribe failed: {e}"); return; @@ -1075,6 +1172,12 @@ pub(crate) async fn subscribe_gift_wraps(trade_pubkey: nostr_sdk::PublicKey, tra Ok(Ok(_)) => continue, } } + + // The subscription bounds the pending-create record's lifetime: once + // no reply can be delivered here anymore, a still-unconsumed record + // (create timed out and no genuine late reply ever arrived) is dead + // state — drop it. + remove_pending_create(&trade_pubkey_hex); }); } @@ -1137,12 +1240,15 @@ async fn dispatch_mostro_message( } // Centralized response validation: catches malformed `request_id` fields - // and flags `CantDo` responses. We pass `None` because the app does not - // yet track outstanding request_ids per action (see issue #101 §5). + // and flags `CantDo` responses. We still pass `None` here on purpose: + // request_id correlation happens at the waiter arms below (via + // `take_matching_create`) because `validate_response` short-circuits + // on `CantDo` BEFORE comparing request_ids, so it cannot distinguish a + // stale replayed rejection from the genuine one. // // `MostroCantDo` is NOT a reason to drop the message — the `Action::CantDo` // arm below is what unblocks `create_order` callers waiting on a - // `pending_confirmations` oneshot. Without propagating it, rejected orders + // pending-create oneshot. Without propagating it, rejected orders // time out and fall back to the optimistic local-ID path, leaving phantom // pending orders in the book. match mostro_core::nip59::validate_response(&msg, None) { @@ -1183,16 +1289,26 @@ async fn dispatch_mostro_message( )); // Reconcile local UUID → daemon UUID if needed. Gift wrap actions - // arrive with the daemon's order ID, but if create_order timed out the - // order book and DB still use the local UUID. Reconcile before any - // status update so that update_order_status / update_trade_fields find - // the order by the daemon ID. + // arrive with the daemon's order ID, but if the create's acknowledgement + // was missed the trade-key bookkeeping still uses the local UUID. + // Reconcile before any status update so that update_order_status / + // update_trade_fields find the order by the daemon ID. + // + // Gated by ownership: only the local UUID recorded in this trade key's + // own pending create may ever be rebound. Without the gate, any event + // carrying an old order id for a reused trade index (stale replays after + // a mnemonic re-import) would rebind a confirmed order's id — daemon → + // daemon — corrupting the order book, the trade row, and the trade-key + // mapping in one stroke. Cold start loses nothing: the pending map is + // empty after a restart, and the Kind 38383 fingerprint path owns maker + // recovery there. if let Some(daemon_id) = &kind.id { let did = daemon_id.to_string(); if order_book().get_order(&did).await.is_none() { if let Some(db) = crate::db::app_db::db() { if let Ok(Some(local_id)) = db.get_order_id_by_trade_index(trade_index).await { - if local_id != did { + let owned = pending_local_uuid_for(trade_pubkey_hex); + if may_reconcile_stored_id(&local_id, &did, owned.as_deref()) { log::info!( "[orders] reconciling order ID: local={local_id} → daemon={did}" ); @@ -1220,54 +1336,48 @@ async fn dispatch_mostro_message( Action::NewOrder => { if let Some(order_id) = &kind.id { let daemon_id = order_id.to_string(); - resolve_maker_order(&daemon_id, trade_pubkey_hex).await; - - // If create_order is waiting for confirmation, notify it. - // The caller handles UUID replacement. Otherwise (cold-start / - // reconnect), do the replacement here. - let conf_tx = pending_confirmations() - .lock() - .ok() - .and_then(|mut m| m.remove(trade_pubkey_hex)); - if let Some(tx) = conf_tx { - let _ = tx.send(DaemonConfirmation::Confirmed { - daemon_id: daemon_id.clone(), - }); - crate::api::logging::blog_info("gift-wrap", format!( - "NewOrder: notified waiting create_order daemon={daemon_id}" - )); - } else { - // No waiting caller — either cold start, reconnect, or - // create_order timed out. Reconcile the local UUID with - // the daemon UUID if the order was persisted under a local ID. - let local_id = pending_local_ids() - .write() - .ok() - .and_then(|mut m| { - // Find the entry whose local_uuid is in the order book. - let key = m.iter() - .find(|(_, uuid)| uuid.as_str() != daemon_id) - .map(|(k, _)| k.clone()); - key.and_then(|k| m.remove(&k)) + + // Consume the pending create ONLY when this reply echoes its + // request_id. Everything the reply is allowed to touch — the + // trade-key binding, the waiter channel, the local→daemon id + // bridge — lives in that one record, so a stale replay or a + // foreign reply (mismatched/absent nonce) touches nothing and + // the genuine reply still finds the record intact. + if let Some(pending) = take_matching_create(trade_pubkey_hex, kind.request_id) { + // Bind the daemon UUID to this attempt's trade index so + // subsequent maker actions (e.g. cancel) can find the key. + store_trade_key_index(&daemon_id, pending.trade_index).await; + + if let Some(tx) = pending.tx { + // create_order is still waiting — the caller handles + // UUID adoption and persistence. + let _ = tx.send(DaemonConfirmation::Confirmed { + daemon_id: daemon_id.clone(), }); - if let Some(local_id) = local_id { - if order_book().get_order(&local_id).await.is_some() { - let mut info = order_book().get_order(&local_id).await.unwrap(); - order_book().remove_order(&local_id).await; - info.id = daemon_id.clone(); - order_book().upsert_order(info).await; - if let Some(db) = crate::db::app_db::db() { - let _ = db.update_trade_order_id(&local_id, &daemon_id).await; - } - crate::api::logging::blog_info("gift-wrap", format!( - "NewOrder: late reconciliation local={local_id} → daemon={daemon_id}" - )); - } + crate::api::logging::blog_info("gift-wrap", format!( + "NewOrder: notified waiting create_order daemon={daemon_id}" + )); } else { + // Genuine reply after the 10s timeout: the caller + // already returned NoDaemonResponse and persisted + // nothing, so there is no local order to rebind — + // the trade-key binding above plus the Kind 38383 + // fingerprint path restore maker ownership. crate::api::logging::blog_info("gift-wrap", format!( - "NewOrder: daemon order={daemon_id} confirmed (no local order to reconcile)" + "NewOrder: late confirmation for timed-out create \ + local={} daemon={daemon_id}", + pending.local_uuid )); } + } else { + // Cold start / reconnect (no record — in-memory state is + // empty after a restart), or an uncorrelated event that + // must not consume anything. The Kind 38383 fingerprint + // path owns maker-order recovery in both cases. + crate::api::logging::blog_info("gift-wrap", format!( + "NewOrder: daemon order={daemon_id} with no matching \ + pending create — leaving state untouched" + )); } } else { log::warn!("[orders] gift-wrap NewOrder has no order id"); @@ -1530,20 +1640,28 @@ async fn dispatch_mostro_message( other => format!("Order rejected by Mostro: {other}"), }; - // If create_order is waiting, notify it — the caller handles - // cleanup and returns an error to Dart. Otherwise ignore stale events. - let conf_tx = pending_confirmations() - .lock() - .ok() - .and_then(|mut m| m.remove(trade_pubkey_hex)); - if let Some(tx) = conf_tx { - crate::api::logging::blog_warn("gift-wrap", format!( - "CantDo: reason={reason} — notifying waiting create_order" - )); - let _ = tx.send(DaemonConfirmation::Rejected { reason, message }); + // Consume the pending create ONLY when this rejection echoes its + // request_id — a genuine rejection ends the attempt, so the whole + // record goes with it. Stale replayed CantDo events (no or foreign + // request_id) touch nothing and leave the record for the genuine + // reply. + if let Some(pending) = take_matching_create(trade_pubkey_hex, kind.request_id) { + if let Some(tx) = pending.tx { + crate::api::logging::blog_warn("gift-wrap", format!( + "CantDo: reason={reason} — notifying waiting create_order" + )); + let _ = tx.send(DaemonConfirmation::Rejected { reason, message }); + } else { + // Genuine rejection after the 10s timeout: the caller + // already returned NoDaemonResponse and persisted nothing, + // so dropping the record is the only cleanup needed. + crate::api::logging::blog_warn("gift-wrap", format!( + "CantDo: reason={reason} — late rejection for timed-out create" + )); + } } else { crate::api::logging::blog_debug("gift-wrap", format!( - "CantDo: reason={reason} — no waiting caller, ignoring stale event" + "CantDo: reason={reason} — no matching pending create, ignoring stale event" )); } } @@ -1896,6 +2014,11 @@ async fn subscribe_node_filters( // Kind-14 NIP-44 replies authored by Mostro for all known trade pubkeys. // The author pin disambiguates from NIP-17 peer chat (also kind 14). + // + // Deliberately NO `since` here: this is the offline catch-up channel — + // after any downtime it must replay the full stored history so status + // changes and late reconciliations are never lost. Only the ephemeral + // per-trade subscription (subscribe_gift_wraps) carries a cutoff. if !trade_pubkeys.is_empty() { let dm_filter = nostr_sdk::Filter::new() .kind(nostr_sdk::Kind::PrivateDirectMessage) @@ -2083,8 +2206,11 @@ async fn ingest_order_event(event: &nostr_sdk::Event) { // book optimistically (see `create_order`), so there // is nothing to remove here — just bridge the local // UUID → daemon UUID in the DB so tradeStatusProvider - // polls with the real order ID. - if let Some(local_id) = take_pending_local_id(&ck) { + // polls with the real order ID. Only records without a + // live waiter are taken: an in-flight create_order owns + // its own reconciliation via the kind-14 acknowledgement. + if let Some(pending) = take_pending_create_by_content_key(&ck) { + let local_id = pending.local_uuid; if let Some(db) = crate::db::app_db::db() { if let Err(e) = db.update_trade_order_id(&local_id, &info.id).await @@ -2316,6 +2442,137 @@ mod tests { use crate::api::types::TradeRole; use crate::mostro::session::session_manager; + // ── request_id correlation ──────────────────────────────────────────────── + + #[test] + fn request_id_only_matches_the_exact_nonce() { + assert!(request_id_matches(42, Some(42))); + assert!(!request_id_matches(42, Some(41))); + // Stale replayed events carry no request_id — they must never match. + assert!(!request_id_matches(42, None)); + } + + fn insert_pending_create(key: &str, request_id: u64) -> tokio::sync::oneshot::Receiver { + let (tx, rx) = tokio::sync::oneshot::channel::(); + pending_creates().lock().unwrap().insert( + key.to_string(), + PendingCreate { + request_id, + trade_index: 3, + local_uuid: format!("local-{key}"), + content_key: format!("content:{key}"), + tx: Some(tx), + }, + ); + rx + } + + /// A reply with a foreign or missing request_id must leave the record in + /// place so the genuine reply can still resolve it; only the echoed nonce + /// consumes it. + #[tokio::test] + async fn take_matching_create_ignores_stale_events() { + let key = "test-take-matching-create-pubkey"; + let mut rx = insert_pending_create(key, 7); + + // Stale replay (no request_id) and foreign reply: record untouched. + assert!(take_matching_create(key, None).is_none()); + assert!(take_matching_create(key, Some(99)).is_none()); + assert!(pending_creates().lock().unwrap().contains_key(key)); + assert!(rx.try_recv().is_err()); // nothing sent + + // Genuine reply: record consumed exactly once, waiter still attached. + let pending = take_matching_create(key, Some(7)).expect("must match"); + let tx = pending.tx.expect("waiter must still be attached"); + let _ = tx.send(DaemonConfirmation::Confirmed { + daemon_id: "d".to_string(), + }); + assert!(!pending_creates().lock().unwrap().contains_key(key)); + assert!(take_matching_create(key, Some(7)).is_none()); + } + + /// After the 10s timeout only the waiter channel is detached; the record + /// survives so the genuine late reply still matches — and stale events + /// still cannot consume it. + #[tokio::test] + async fn late_genuine_reply_matches_after_timeout() { + let key = "test-late-reply-pubkey"; + let _rx = insert_pending_create(key, 11); + + detach_create_waiter(key); + assert!(pending_creates().lock().unwrap().contains_key(key)); + + // Stale events still bounce off the detached record. + assert!(take_matching_create(key, None).is_none()); + assert!(take_matching_create(key, Some(99)).is_none()); + + // The genuine late reply consumes it: no waiter, but the bridging + // state (trade index, local uuid) is intact for reconciliation. + let pending = take_matching_create(key, Some(11)).expect("must match"); + assert!(pending.tx.is_none()); + assert_eq!(pending.trade_index, 3); + assert_eq!(pending.local_uuid, format!("local-{key}")); + assert!(!pending_creates().lock().unwrap().contains_key(key)); + } + + /// Concurrent creates each own their record: a reply correlated to one + /// attempt must never consume state belonging to another. + #[tokio::test] + async fn concurrent_creates_do_not_cross_consume() { + let key_a = "test-concurrent-a-pubkey"; + let key_b = "test-concurrent-b-pubkey"; + let _rx_a = insert_pending_create(key_a, 21); + let _rx_b = insert_pending_create(key_b, 22); + + // A's nonce only ever matches A's record, under either key. + assert!(take_matching_create(key_b, Some(21)).is_none()); + let pending = take_matching_create(key_a, Some(21)).expect("must match A"); + assert_eq!(pending.local_uuid, format!("local-{key_a}")); + + // B is untouched and still consumable by its own nonce. + let pending = take_matching_create(key_b, Some(22)).expect("must match B"); + assert_eq!(pending.local_uuid, format!("local-{key_b}")); + } + + /// Only the pending create's own local UUID may be rebound to an incoming + /// event's order id; a stored id that is already a daemon's (or belongs to + /// an earlier life of a reused trade key) must never be rebound. + #[test] + fn stored_id_reconciles_only_when_owned_by_the_pending_create() { + // The legitimate case: the stored id is this create's local UUID. + assert!(may_reconcile_stored_id("local-1", "daemon-1", Some("local-1"))); + // Already the incoming id: nothing to rebind. + assert!(!may_reconcile_stored_id("daemon-1", "daemon-1", Some("local-1"))); + // Stored id is a confirmed daemon id — a stale replay carrying an old + // order id for the same (reused) trade index must not rebind it. + assert!(!may_reconcile_stored_id("daemon-1", "old-daemon-9", Some("local-1"))); + // No pending create for this trade key (cold start / uncorrelated + // event): never rebind here. + assert!(!may_reconcile_stored_id("local-1", "daemon-1", None)); + } + + /// The Kind 38383 path matches by content fingerprint, but must leave + /// records with a live waiter alone — the in-flight create_order call owns + /// that reconciliation. + #[tokio::test] + async fn content_key_lookup_skips_live_waiters() { + let key = "test-content-key-pubkey"; + let ck = format!("content:{key}"); + let _rx = insert_pending_create(key, 31); + + // Live waiter attached: the 38383 path must not consume the record. + assert!(take_pending_create_by_content_key(&ck).is_none()); + + // After the timeout detaches the waiter, the fingerprint match takes it. + detach_create_waiter(key); + let pending = take_pending_create_by_content_key(&ck).expect("must match"); + assert_eq!(pending.local_uuid, format!("local-{key}")); + assert!(!pending_creates().lock().unwrap().contains_key(key)); + + // Unknown fingerprints never match anything. + assert!(take_pending_create_by_content_key("content:unknown").is_none()); + } + // ── Helper ──────────────────────────────────────────────────────────────── fn dummy_order_info(id: &str) -> crate::api::types::OrderInfo { diff --git a/rust/src/frb_generated.rs b/rust/src/frb_generated.rs index 1bfaccbf..10a3f72c 100644 --- a/rust/src/frb_generated.rs +++ b/rust/src/frb_generated.rs @@ -45,7 +45,7 @@ flutter_rust_bridge::frb_generated_boilerplate!( default_rust_auto_opaque = RustAutoOpaqueMoi, ); pub(crate) const FLUTTER_RUST_BRIDGE_CODEGEN_VERSION: &str = "2.11.1"; -pub(crate) const FLUTTER_RUST_BRIDGE_CODEGEN_CONTENT_HASH: i32 = 426373470; +pub(crate) const FLUTTER_RUST_BRIDGE_CODEGEN_CONTENT_HASH: i32 = -1966610045; // Section: executor @@ -3364,49 +3364,6 @@ fn wire__crate__api__nostr__remove_relay_impl( }, ) } -fn wire__crate__api__orders__resolve_maker_order_impl( - port_: flutter_rust_bridge::for_generated::MessagePort, - ptr_: flutter_rust_bridge::for_generated::PlatformGeneralizedUint8ListPtr, - rust_vec_len_: i32, - data_len_: i32, -) { - FLUTTER_RUST_BRIDGE_HANDLER.wrap_async::( - flutter_rust_bridge::for_generated::TaskInfo { - debug_name: "resolve_maker_order", - port: Some(port_), - mode: flutter_rust_bridge::for_generated::FfiCallMode::Normal, - }, - move || { - let message = unsafe { - flutter_rust_bridge::for_generated::Dart2RustMessageSse::from_wire( - ptr_, - rust_vec_len_, - data_len_, - ) - }; - let mut deserializer = - flutter_rust_bridge::for_generated::SseDeserializer::new(message); - let api_order_id = ::sse_decode(&mut deserializer); - let api_trade_pubkey_hex = ::sse_decode(&mut deserializer); - deserializer.end(); - move |context| async move { - transform_result_sse::<_, ()>( - (move || async move { - let output_ok = Result::<_, ()>::Ok({ - crate::api::orders::resolve_maker_order( - &api_order_id, - &api_trade_pubkey_hex, - ) - .await; - })?; - Ok(output_ok) - })() - .await, - ) - } - }, - ) -} fn wire__crate__api__orders__restart_orders_subscription_impl( port_: flutter_rust_bridge::for_generated::MessagePort, ptr_: flutter_rust_bridge::for_generated::PlatformGeneralizedUint8ListPtr, @@ -5622,47 +5579,46 @@ fn pde_ffi_dispatcher_primary_impl( ), 80 => wire__crate__api__orders__release_order_impl(port, ptr, rust_vec_len, data_len), 81 => wire__crate__api__nostr__remove_relay_impl(port, ptr, rust_vec_len, data_len), - 82 => wire__crate__api__orders__resolve_maker_order_impl(port, ptr, rust_vec_len, data_len), - 83 => wire__crate__api__orders__restart_orders_subscription_impl( + 82 => wire__crate__api__orders__restart_orders_subscription_impl( port, ptr, rust_vec_len, data_len, ), - 84 => wire__crate__api__orders__send_fiat_sent_impl(port, ptr, rust_vec_len, data_len), - 85 => wire__crate__api__messages__send_file_impl(port, ptr, rust_vec_len, data_len), - 86 => wire__crate__api__orders__send_invoice_impl(port, ptr, rust_vec_len, data_len), - 87 => wire__crate__api__messages__send_message_impl(port, ptr, rust_vec_len, data_len), - 88 => wire__crate__api__settings__set_active_mostro_node_impl( + 83 => wire__crate__api__orders__send_fiat_sent_impl(port, ptr, rust_vec_len, data_len), + 84 => wire__crate__api__messages__send_file_impl(port, ptr, rust_vec_len, data_len), + 85 => wire__crate__api__orders__send_invoice_impl(port, ptr, rust_vec_len, data_len), + 86 => wire__crate__api__messages__send_message_impl(port, ptr, rust_vec_len, data_len), + 87 => wire__crate__api__settings__set_active_mostro_node_impl( port, ptr, rust_vec_len, data_len, ), - 89 => wire__crate__api__settings__set_default_fiat_code_impl( + 88 => wire__crate__api__settings__set_default_fiat_code_impl( port, ptr, rust_vec_len, data_len, ), - 90 => wire__crate__api__settings__set_default_lightning_address_impl( + 89 => wire__crate__api__settings__set_default_lightning_address_impl( port, ptr, rust_vec_len, data_len, ), - 91 => wire__crate__api__settings__set_language_impl(port, ptr, rust_vec_len, data_len), - 92 => { + 90 => wire__crate__api__settings__set_language_impl(port, ptr, rust_vec_len, data_len), + 91 => { wire__crate__api__settings__set_logging_enabled_impl(port, ptr, rust_vec_len, data_len) } - 93 => { + 92 => { wire__crate__api__reputation__set_privacy_mode_impl(port, ptr, rust_vec_len, data_len) } - 94 => wire__crate__api__settings__set_theme_impl(port, ptr, rust_vec_len, data_len), - 95 => wire__crate__api__disputes__submit_evidence_impl(port, ptr, rust_vec_len, data_len), - 96 => wire__crate__api__reputation__submit_rating_impl(port, ptr, rust_vec_len, data_len), - 97 => wire__crate__api__orders__subscribe_orders_impl(port, ptr, rust_vec_len, data_len), - 98 => wire__crate__api__orders__take_order_impl(port, ptr, rust_vec_len, data_len), + 93 => wire__crate__api__settings__set_theme_impl(port, ptr, rust_vec_len, data_len), + 94 => wire__crate__api__disputes__submit_evidence_impl(port, ptr, rust_vec_len, data_len), + 95 => wire__crate__api__reputation__submit_rating_impl(port, ptr, rust_vec_len, data_len), + 96 => wire__crate__api__orders__subscribe_orders_impl(port, ptr, rust_vec_len, data_len), + 97 => wire__crate__api__orders__take_order_impl(port, ptr, rust_vec_len, data_len), _ => unreachable!(), } } diff --git a/rust/src/mostro/actions.rs b/rust/src/mostro/actions.rs index 375f1ca7..b2e57401 100644 --- a/rust/src/mostro/actions.rs +++ b/rust/src/mostro/actions.rs @@ -23,6 +23,11 @@ use crate::nostr::gift_wrap; /// Build and wrap a NewOrder MostroMessage. /// +/// `request_id` is a caller-generated correlation nonce: the daemon echoes it +/// in the `NewOrder` confirmation and in any `CantDo` rejection, which is how +/// `create_order` tells the genuine reply apart from stale relay-replayed +/// events addressed to the same trade key. +/// /// Returns the NIP-59 Gift Wrap event JSON ready for publication. pub async fn new_order( identity_keys: &Keys, @@ -30,6 +35,7 @@ pub async fn new_order( mostro_pubkey: &PublicKey, params: &NewOrderParams, trade_index: u32, + request_id: u64, ) -> Result { use mostro_core::order::{Kind, SmallOrder, Status}; @@ -62,7 +68,13 @@ pub async fn new_order( ); let payload = Some(Payload::Order(small_order)); - let msg = Message::new_order(None, None, Some(trade_index as i64), Action::NewOrder, payload); + let msg = Message::new_order( + None, + Some(request_id), + Some(trade_index as i64), + Action::NewOrder, + payload, + ); wrap_message(identity_keys, trade_keys, mostro_pubkey, &msg).await } @@ -315,3 +327,52 @@ async fn wrap_message( gift_wrap::wrap_mostro_message(identity_keys, trade_keys, mostro_pubkey, msg, pow).await?; Ok(event.as_json()) } + +#[cfg(test)] +mod tests { + use super::*; + + /// The outgoing new-order message must carry the caller's request_id — + /// it is the correlation nonce the daemon echoes in its reply, and + /// `create_order` relies on it to tell the genuine reply apart from + /// stale relay-replayed events. + #[tokio::test] + async fn new_order_carries_request_id_and_trade_index() { + let identity_keys = Keys::generate(); + let trade_keys = Keys::generate(); + let mostro_keys = Keys::generate(); + + let params = NewOrderParams { + kind: OrderKind::Sell, + fiat_amount: Some(100.0), + fiat_amount_min: None, + fiat_amount_max: None, + fiat_code: "USD".to_string(), + payment_method: "cashapp".to_string(), + premium: 0.0, + amount_sats: None, + }; + + let json = new_order( + &identity_keys, + &trade_keys, + &mostro_keys.public_key(), + ¶ms, + 3, + 42, + ) + .await + .unwrap(); + + let event = Event::from_json(&json).unwrap(); + let unwrapped = gift_wrap::unwrap_mostro_message(&mostro_keys, &event) + .await + .unwrap() + .expect("message must decrypt for the recipient"); + + let kind = unwrapped.message.get_inner_message_kind(); + assert_eq!(kind.request_id, Some(42)); + assert_eq!(kind.trade_index, Some(3)); + assert!(matches!(kind.action, Action::NewOrder)); + } +}