diff --git a/lib/features/order/providers/trade_state_provider.dart b/lib/features/order/providers/trade_state_provider.dart index 0afd3443..f3722344 100644 --- a/lib/features/order/providers/trade_state_provider.dart +++ b/lib/features/order/providers/trade_state_provider.dart @@ -1,3 +1,4 @@ +import 'dart:async'; import 'package:flutter/foundation.dart'; import 'package:flutter_riverpod/flutter_riverpod.dart'; import 'package:mostro/src/rust/api/orders.dart' as orders_api; @@ -25,33 +26,107 @@ final tradeAmountProvider = } }); -/// Live order status for a single trade, polled from the order book every 2 s. +/// Live order status for a single trade. /// -/// Starts with an immediate fetch (no initial delay) so the first emission -/// reflects the real relay status. When the order is no longer in the in-memory -/// order book (e.g. after cancellation), falls back to the persisted trade DB -/// so terminal statuses like Canceled are reflected in the UI. +/// Push-first: emits an immediate status from the order book (falling back to +/// the persisted trade DB when the order has left the in-memory book), then +/// reflects each `on_trade_updated` push for this `orderId` as it arrives, so +/// the UI reacts to daemon-driven status changes in real time instead of on a +/// fixed 2 s poll. A long reconnection-fallback poll runs only to reconcile a +/// push that was dropped or missed (app resumed from background, stream +/// reconnect); the daemon stays the authority either way. final tradeStatusProvider = StreamProvider.family.autoDispose((ref, orderId) async* { - while (true) { - final info = await orders_api.getOrder(orderId: orderId); - if (info != null) { - yield info.status; - if (_isTerminal(info.status)) return; - } else { - // Order removed from in-memory book — check the persisted trade DB. - final trades = await orders_api.listTrades(); - final trade = trades.where((t) => t.order.id == orderId).firstOrNull; - if (trade != null) { - yield trade.order.status; - // Terminal status — no need to keep polling. - if (_isTerminal(trade.order.status)) return; - } + // Push-first: a single event stream carries both push updates for THIS order + // (bridged from the shared [tradeUpdatesProvider] via ref.listen, so one relay + // subscription feeds every watched trade and tests can drive it through + // `tradeUpdatesProvider.overrideWith`) and periodic reconnection-fallback + // ticks. Merging both into one stream means a single subscription drains them + // in order — no abandoned `moveNext()` futures, no busy-looping. + final events = StreamController<_StatusEvent>(); + + final sub = ref.listen>(tradeUpdatesProvider, + (_, next) { + final u = next.valueOrNull; + if (u != null && u.orderId == orderId && !events.isClosed) { + events.add(_PushEvent(u.status)); + } + }); + + final ticker = Timer.periodic(_reconnectPoll, (_) { + if (!events.isClosed) events.add(const _FallbackTick()); + }); + + ref.onDispose(() { + sub.close(); + ticker.cancel(); + events.close(); + }); + + // Immediate first emission — current status, same DB fallback as before for + // orders already gone from the in-memory book. + OrderStatus? last = await _currentStatus(orderId); + if (last != null) { + yield last; + if (_isTerminal(last)) return; + } + + // Drain the merged stream. A push carries the new status directly; a fallback + // tick triggers a reconciliation fetch. Only distinct statuses are emitted. + await for (final event in events.stream) { + final status = switch (event) { + _PushEvent(:final status) => status, + _FallbackTick() => await _currentStatus(orderId), + }; + if (status != null && status != last) { + last = status; + yield status; + if (_isTerminal(status)) return; } - await Future.delayed(const Duration(seconds: 2)); } }); +/// Internal event type merged into [tradeStatusProvider]'s single stream: either +/// a pushed status or a periodic fallback tick that triggers a reconciliation +/// fetch. +sealed class _StatusEvent { + const _StatusEvent(); +} + +class _PushEvent extends _StatusEvent { + const _PushEvent(this.status); + final OrderStatus status; +} + +class _FallbackTick extends _StatusEvent { + const _FallbackTick(); +} + +/// Long fallback interval for [tradeStatusProvider]: pushes carry the real-time +/// signal, so this only reconciles a dropped/missed update rather than driving +/// the UI (was a 2 s poll before the push-first migration). +const _reconnectPoll = Duration(seconds: 30); + +/// Current status for [orderId]: the in-memory order book first, then the +/// persisted trade DB when the order has been removed from the book (e.g. after +/// a cancellation wipe). Returns null when neither knows the order. +/// +/// A bridge/DB failure yields null rather than propagating: the reconciliation +/// fetch is best-effort, so a transient failure must not tear down the whole +/// status stream — the next push or fallback tick recovers. (This also lets the +/// provider run in tests without `RustLib.init()`, where these calls fail.) +Future _currentStatus(String orderId) async { + try { + final info = await orders_api.getOrder(orderId: orderId); + if (info != null) return info.status; + final trades = await orders_api.listTrades(); + return trades.where((t) => t.order.id == orderId).firstOrNull?.order.status; + } catch (e, st) { + debugPrint('[tradeStatusProvider] status fetch failed for $orderId: $e\n$st'); + return null; + } +} + /// Trade lifecycle updates pushed from Rust (daemon-driven cancellations). /// /// Complements [tradeStatusProvider]'s polling, which cannot observe a diff --git a/rust/src/api/orders.rs b/rust/src/api/orders.rs index 9fb47953..1e3aee14 100644 --- a/rust/src/api/orders.rs +++ b/rust/src/api/orders.rs @@ -1375,6 +1375,12 @@ pub async fn cancel_order(order_id: String) -> Result<()> { log::warn!("[orders] failed to optimistically update cancel status for {order_id}: {e}"); } } + // Push the optimistic Canceled to the trade-status stream: the order is + // gone from the book and the daemon's gift-wrap confirmation may arrive + // much later (or never, if the app closes), so a push-first listener needs + // this signal now — the old 2 s poll saw the DB write, the push channel + // must too. + emit_trade_update(&order_id, crate::api::types::OrderStatus::Canceled); crate::api::logging::blog_info( "orders", @@ -4783,6 +4789,44 @@ mod tests { )); } + /// #272: a client-initiated cancel emits an optimistic `Canceled` TradeUpdate + /// for the cancelled order. `cancel_order` itself needs trade keys, identity + /// and the relay (not available in a unit test), so this covers the emit + /// contract it relies on: the same `emit_trade_update` call cancel_order makes + /// after its optimistic book/DB update. The emit is independent of the DB + /// write, so it still fires when that write fails — here there is no DB at + /// all, mirroring that path. + /// + /// `on_trade_updated` is a process-wide broadcast, so a concurrent test may + /// interleave its own emits; the subscriber therefore filters for this + /// test's unique order id rather than assuming the first update is ours. + #[tokio::test] + async fn client_cancel_emits_canceled_update() { + let oid = "order-cancel-272-unique"; + let mut stream = on_trade_updated().await.unwrap(); + // The exact call cancel_order performs after its optimistic update. + emit_trade_update(oid, crate::api::types::OrderStatus::Canceled); + + // Drain until our order's update arrives (skipping any interleaved emits + // from concurrently-running tests on the shared broadcast channel). + let deadline = std::time::Duration::from_secs(2); + let found = tokio::time::timeout(deadline, async { + loop { + let update = stream.next().await.expect("channel stays open"); + if update.order_id == oid { + return update; + } + } + }) + .await + .expect("cancel must push a Canceled update for our order"); + + assert!(matches!( + found.status, + crate::api::types::OrderStatus::Canceled + )); + } + /// The sweep only acts on positive daemon signals: pending republish /// (wipe for takers, resync for makers) and outright cancellation; /// absence from the book or ambiguous statuses leave the trade alone. diff --git a/specs/004-mostro-p2p-client/contracts/orders.md b/specs/004-mostro-p2p-client/contracts/orders.md index f412e3d6..563b95f4 100644 --- a/specs/004-mostro-p2p-client/contracts/orders.md +++ b/specs/004-mostro-p2p-client/contracts/orders.md @@ -221,7 +221,19 @@ TradeUpdate { ``` ### on_order_status_changed(order_id: String) → Stream -Emits when a specific order's status changes. +**Superseded by `on_trade_updated()`.** A separate per-order status +stream is not implemented: `on_trade_updated()` already emits a +`TradeUpdate { order_id, status }` on daemon-driven status transitions and +on client-initiated optimistic cancellation. Clients filter by `order_id`. +`tradeStatusProvider` consumes that push channel directly (with a low-frequency +reconnection fallback), so a dedicated single-order status stream would +duplicate it. + +Note: `cancel_order` emits an optimistic `Canceled` immediately after the client +action — before the daemon's gift-wrap confirmation, and even if the local DB +write fails (the order is already gone from the book, so the push is the only +timely signal). Consumers must therefore not treat every `Canceled` update as +daemon-confirmed state. ### on_trade_step_changed() → Stream Emits when the active trade's step changes. Used to update the