Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
104 changes: 104 additions & 0 deletions rust/src/api/identity.rs
Original file line number Diff line number Diff line change
Expand Up @@ -399,6 +399,83 @@ async fn derive_trade_key_with<S: Storage>(
})
}

/// Raise `trade_key_index` to at least `floor`, never lowering it (#217).
///
/// A restore recovers trades that already occupy trade-key indexes; without
/// this, the next `derive_trade_key()` would hand out an index a recovered
/// trade already owns — reusing a key the daemon has bound. The bump is
/// monotonic: a stale or partial `RestoreData`, or one that arrives after the
/// counter has already advanced, must never rewind it. Idempotent — applying
/// the same recovered set twice changes nothing.
///
/// Persisted under the same discipline as `derive_trade_key`: if the counter
/// moves, the write must succeed or the call fails, so the advance is durable
/// (a bumped-but-unpersisted counter would regress on the next restart).
pub(crate) async fn ensure_trade_key_index_at_least(floor: u32) -> Result<()> {
let db = crate::db::app_db::db();
// Same durable-storage precondition as derive_trade_key: on native, refuse
// to advance the counter when there is no store, because the _with core
// would otherwise bump and publish the raised index WITHOUT persisting it
// (the `if let Some(db)` save is skipped) — and publication is best-effort,
// so a session loss would reload a stale pre-resync index and reopen the
// key-reuse bug this closes (#249).
#[cfg(not(target_arch = "wasm32"))]
require_durable_storage(db)?;

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(low) Unreachable from the only caller — the comment presents it as load-bearing.

restore_session calls derive_trade_key() as its first statement, and that already runs require_durable_storage and returns early on native. So by the time ensure_trade_key_index_at_least runs, a store is guaranteed to exist and this check can never fire in production.

Defence-in-depth is fine and I would keep it, but the comment ("the _with core would otherwise bump and publish the raised index WITHOUT persisting it") describes a path no caller can reach today. Worth a short note that it guards future callers rather than the current one, so a reader does not go looking for the scenario it prevents.

// Web is exempt for the same reason derive_trade_key is: `init_db` is never
// called there and IndexedDB has no save_identity yet, so the published
// index is web's durable record via the Flutter mirror until #233 lands.
#[cfg(target_arch = "wasm32")]
if db.is_none() {
log::warn!(
"[identity] no local store on web — the resynced trade-key counter is durable only through the Flutter mirror"

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(medium) Broken line continuation — this ships a run of literal spaces into the log.

The \ + newline continuation was lost when this was adapted from derive_trade_key (line 336, which has it right). The literal is currently:

"[identity] no local store on web — the resynced trade-key counter              is durable only through the Flutter mirror"

So the emitted line reads ...counter is durable.... Restore the continuation:

log::warn!(
    "[identity] no local store on web — the resynced trade-key counter is \
     durable only through the Flutter mirror"
);

Only reachable on wasm, which is why neither cargo test nor clippy caught it.

);
}
ensure_trade_key_index_at_least_with(db, trade_key_index_tx(), floor).await
}

/// Testable core of [`ensure_trade_key_index_at_least`]: takes an explicit store
/// and publish channel so tests can inject a failing store and a private channel,
/// mirroring `derive_trade_key` / `derive_trade_key_with`.
async fn ensure_trade_key_index_at_least_with<S: Storage>(
db: Option<&S>,
tx: &broadcast::Sender<u32>,
floor: u32,
) -> Result<()> {
let mut guard = identity_lock().write().await;
let state = guard.as_mut().ok_or_else(|| anyhow!("NoIdentity"))?;
let current = state.identity_info.trade_key_index;
let raised = current.max(floor);
if raised == current {
// Already ahead of (or level with) the recovered set — no-op, no write.
return Ok(());
Comment thread
coderabbitai[bot] marked this conversation as resolved.
}
state.identity_info.trade_key_index = raised;
if let Some(db) = db {
if let Err(e) = db.save_identity(&state.identity_info).await {
// Roll back the in-memory bump on a failed persist. Without this, a
// retried restore with the same floor would see `raised == current`,
// take the no-op short-circuit above, and return Ok(()) WITHOUT ever
// re-attempting the write — silently leaving the durable counter
// un-raised and reopening the key-reuse bug this closes. (Unlike
// derive_trade_key_with, which safely keeps its forward mutation
// because it has no idempotency short-circuit to defeat.)
state.identity_info.trade_key_index = current;

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(high) This rollback is the subtlest logic in the PR and nothing tests it.

The comment right above correctly explains why the rollback is load-bearing: without it the idempotency short-circuit at the top swallows the retry and the durable counter silently stays un-raised. That is a good catch — and it is exactly the kind of reasoning that needs a test pinning it, because a future refactor that "simplifies" the short-circuit or drops the restore-on-failure line will not fail anything.

The doc comment on this function even advertises the seam:

takes an explicit store and publish channel so tests can inject a failing store and a private channel

No test in identity.rs injects a failing store — the only Storage used in tests is temp_store, which succeeds. Please add one (a Storage impl whose save_identity returns Err) asserting the three properties this branch exists for:

  1. the call returns Err with the StorageError: marker;
  2. the in-memory trade_key_index is unchanged afterwards (get_identity() still reports the old value);
  3. a retry with the same floor against a working store actually performs the write — i.e. the short-circuit was not poisoned.

(3) is the one that would regress silently, and it is the reason this rollback was written.

return Err(anyhow!(
"StorageError: failed to persist resynced trade_key_index {raised}: {e}"
));
Comment thread
codaMW marked this conversation as resolved.
}
}
// Only after the primary record is durable: mirror to secure storage the
// same way derive_trade_key_with does, so a later loss of mostro.db still
// reloads the resynced counter rather than a stale pre-restore index (#249).
publish_index(tx, raised);
crate::api::logging::blog_info(
"restore",
format!("trade_key_index resynced {current} -> {raised} from recovered trades"),
);
Ok(())
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.

/// Re-derive an existing trade key by index.
pub async fn get_trade_key(index: u32) -> Result<TradeKeyInfo> {
let guard = identity_lock().read().await;
Expand Down Expand Up @@ -707,6 +784,33 @@ mod tests {
let current = get_identity().await.unwrap().unwrap();
assert_eq!(current.trade_key_index, 22);

// #217 resync — asserted here (not a separate #[tokio::test]) so it
// shares the single identity_lock lifecycle and can't race it. Uses the
// `_with` core so publications land on this test's private channel.
// Never lowers: a floor below current is a no-op — no write, no publish.
ensure_trade_key_index_at_least_with(Some(&db), &tx, 10).await.unwrap();
assert_eq!(get_identity().await.unwrap().unwrap().trade_key_index, 22);
assert!(
published.rx.try_recv().is_err(),
"a no-op resync must not publish",
);
// Raises to the recovered max, persists, and publishes to the mirror.
ensure_trade_key_index_at_least_with(Some(&db), &tx, 50).await.unwrap();
assert_eq!(get_identity().await.unwrap().unwrap().trade_key_index, 50);
assert_eq!(published.next().await.unwrap(), 50);
assert_eq!(db.get_identity().await.unwrap().unwrap().trade_key_index, 50);
// Idempotent: the same floor again changes nothing and publishes nothing.
ensure_trade_key_index_at_least_with(Some(&db), &tx, 50).await.unwrap();
assert_eq!(get_identity().await.unwrap().unwrap().trade_key_index, 50);
assert!(
published.rx.try_recv().is_err(),
"an idempotent resync must not publish again",
);
// Regression (the bug #217 fixes): the next derived key is FRESH —
// index 51, past every recovered trade — not a reused recovered index.
let after = derive_trade_key_with(Some(&db), &tx).await.unwrap();
assert_eq!(after.index, 51);

crate::api::logging::forward_log(log::Level::Info, "identity_probe", "before delete");

delete_identity().await.unwrap();
Expand Down
123 changes: 122 additions & 1 deletion rust/src/api/orders.rs
Original file line number Diff line number Diff line change
Expand Up @@ -3071,6 +3071,35 @@ pub async fn get_trade_role(order_id: String) -> Result<Option<crate::api::types
}
}

/// 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.
fn recovered_max_trade_index(
info: &mostro_core::message::RestoreSessionInfo,
) -> Option<u32> {
info.restore_orders
.iter()
.map(|o| o.trade_index)
.chain(info.restore_disputes.iter().map(|d| d.trade_index))
// `filter_map` with `try_from` drops both negatives and any value
// beyond `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 also dropped: it is reserved as the
// terminal index, because 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). 4 billion trades is not reachable
// in practice, but the floor must never be a value the counter cannot
// advance past.
.filter_map(|i| u32::try_from(i).ok())
.filter(|&i| i < u32::MAX)

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(medium) Invalid indexes are dropped silently — that is a silent failure on protocol drift.

A negative trade_index, or one at/beyond u32::MAX, is not just an odd value: it means the daemon sent something this client's model does not cover. Dropping it without a trace means the counter can end up lower than the true recovered maximum and the resulting CantDo(InvalidTradeIndex) on the next order will have no breadcrumb pointing back here.

The filtering decision is right; the silence is not. Suggest counting the drops and emitting blog_warn("restore", ...) when the count is non-zero. The degenerate case matters most: when every index is invalid this returns None, restore_session skips the resync entirely, and the restore reports success — the one path where a log is the only evidence anything happened.

Minor, same expression: the two adapters can collapse into one, which also removes the chance of the second filter drifting from the first —

.filter_map(|i| u32::try_from(i).ok().filter(|&v| v < u32::MAX))

.max()
}

/// 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.
///
Expand Down Expand Up @@ -3143,7 +3172,18 @@ pub async fn restore_session() -> Result<mostro_core::message::RestoreSessionInf
}

match confirmation {
Ok(Ok(DaemonReply::Restored(info))) => Ok(info),
Ok(Ok(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) {

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(high) The resync cannot cover the index this very function already consumed — see point 1 of the summary.

By the time this line runs, derive_trade_key() at the top of restore_session has already handed out current + 1 from the un-resynced counter. On the actual recovery path (import_from_mnemonic(recover = true)load_identity_from_mnemonic(words, 0, ...)) that is index 1 on a fresh install — a key a recovered trade already owns.

And because the bump lives only in this arm, the Rejected / timeout paths below leave the counter advanced by one with no resync: the next order then goes out at index 2, which the daemon already bound. That is the failure #217 describes, still reachable.

(medium, same line) Failing the whole restore on a persist error discards info entirely — the oneshot is consumed, so a retry costs another daemon round-trip and another consumed index. The trade-off is argued well in the comment and I would not block on it, but with #219 about to drive this from the UI it is worth deciding now whether the caller should instead receive the recovered trades plus an explicit "resync failed, do not create orders" signal. Surfacing the data and the hazard separately gives the UI something to act on; an opaque Err gives it a retry button and nothing else.

crate::api::identity::ensure_trade_key_index_at_least(floor).await?;
}
Ok(info)
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Ok(Ok(DaemonReply::Rejected { reason, message })) => {
crate::api::logging::blog_warn("orders", format!(
"restore_session rejected: {reason} — {message}"
Expand Down Expand Up @@ -3187,6 +3227,55 @@ mod tests {
);
}

// ── #217 recovered_max_trade_index ────────────────────────────────────────
fn restored_order(trade_index: i64) -> mostro_core::message::RestoredOrdersInfo {
mostro_core::message::RestoredOrdersInfo {
order_id: uuid::Uuid::new_v4(),
trade_index,
status: "active".to_string(),
}
}

fn restored_dispute(trade_index: i64) -> mostro_core::message::RestoredDisputesInfo {
mostro_core::message::RestoredDisputesInfo {
dispute_id: uuid::Uuid::new_v4(),
order_id: uuid::Uuid::new_v4(),
trade_index,
status: "initiated".to_string(),
initiator: None,
solver_pubkey: None,
}
}

fn restore_info(
orders: Vec<i64>,
disputes: Vec<i64>,
) -> mostro_core::message::RestoreSessionInfo {
mostro_core::message::RestoreSessionInfo {
restore_orders: orders.into_iter().map(restored_order).collect(),
restore_disputes: disputes.into_iter().map(restored_dispute).collect(),
}
}

#[test]
fn recovered_max_is_none_when_nothing_was_restored() {
assert_eq!(recovered_max_trade_index(&restore_info(vec![], vec![])), None);
}

#[test]
fn recovered_max_spans_orders_and_disputes() {
// Max lives in disputes here — the fn must consider both collections.
assert_eq!(
recovered_max_trade_index(&restore_info(vec![3, 7], vec![12, 5])),
Some(12)
);
// ...and the other way round.
assert_eq!(
recovered_max_trade_index(&restore_info(vec![40, 9], vec![2])),
Some(40)
);
}

#[test]
fn a_dispute_message_without_a_peer_payload_yields_no_solver() {
use mostro_core::message::Payload;
Expand All @@ -3198,6 +3287,38 @@ mod tests {
assert_eq!(admin_pubkey_from_payload(Some(&Payload::Amount(42))), None);
}

#[test]
fn recovered_max_drops_negative_and_out_of_range_indexes() {

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(low) Test placement. This one is separated from its two siblings (recovered_max_is_none_when_nothing_was_restored, recovered_max_spans_orders_and_disputes) by the unrelated a_dispute_message_without_a_peer_payload_yields_no_solver. Move it up under the ── #217 recovered_max_trade_index ── banner so the section the banner introduces is actually contiguous.

// A negative index is not a real trade index — dropped, not counted.
assert_eq!(
recovered_max_trade_index(&restore_info(vec![-1, 8], vec![-99])),
Some(8)
);
// Beyond u32::MAX: dropped rather than truncated into a small counter.
let huge = i64::from(u32::MAX) + 1;
assert_eq!(
recovered_max_trade_index(&restore_info(vec![huge, 4], vec![])),
Some(4)
);
// u32::MAX itself is dropped — reserved as the terminal index, since
// storing it would make the next derive_trade_key overflow on +1.
let terminal = i64::from(u32::MAX);
assert_eq!(
recovered_max_trade_index(&restore_info(vec![terminal, 4], vec![])),
Some(4)
);
// Only u32::MAX present -> None (no safe floor to resync to).
assert_eq!(
recovered_max_trade_index(&restore_info(vec![terminal], vec![])),
None
);
// All invalid -> None (nothing safe to resync to).
assert_eq!(
recovered_max_trade_index(&restore_info(vec![-1], vec![huge])),
None
);
}

fn insert_pending_create(key: &str, request_id: u64) -> tokio::sync::oneshot::Receiver<DaemonReply> {
let (tx, rx) = tokio::sync::oneshot::channel::<DaemonReply>();
pending_requests().lock().unwrap().insert(
Expand Down
Loading