diff --git a/src/app/release.rs b/src/app/release.rs index 3b017d8e..80646c1d 100644 --- a/src/app/release.rs +++ b/src/app/release.rs @@ -279,9 +279,25 @@ pub async fn release_action( { Ok(()) => { let client = ctx.nostr_client(); - if client.send_event(&event).await.is_err() { - tracing::warn!("Failed sending child order event for order id: {}; queued for republish by the orderbook reconciler", child_order_id); - mark_orderbook_publish_failed(child_order_id); + // A per-relay rejection resolves to `Ok` with the + // refusing relays in `output.failed`; both that and a + // full send error leave the book divergent somewhere, + // so queue the child for the reconciler. + match client.send_event(&event).await { + Ok(output) if output.failed.is_empty() => {} + Ok(output) => { + tracing::warn!( + "child order event rejected by {} relay(s) for order id: {}: {:?}; queued for republish by the orderbook reconciler", + output.failed.len(), + child_order_id, + output.failed + ); + mark_orderbook_publish_failed(child_order_id); + } + Err(_) => { + tracing::warn!("Failed sending child order event for order id: {}; queued for republish by the orderbook reconciler", child_order_id); + mark_orderbook_publish_failed(child_order_id); + } } } Err(e) => { diff --git a/src/db.rs b/src/db.rs index f46e0723..f77e1ab0 100644 --- a/src/db.rs +++ b/src/db.rs @@ -675,34 +675,6 @@ pub async fn find_order_by_date(pool: &SqlitePool) -> Result, MostroE Ok(order) } -/// All orders currently advertised (or advertisable) as `pending` on the -/// book. Includes `waiting-taker-bond`, which publishes as `pending` on the -/// wire (Phase 1.5), and excludes `waiting-maker-bond`, which was never -/// published. Used by the orderbook reconciler to re-assert the book on -/// relays; only orders whose take window is still open are returned — an -/// expired row is the expiry job's business, not the reconciler's. Legacy -/// rows with `expires_at = 0` have no take-window TTL (mirroring -/// `is_order_take_window_closed`) and stay re-assertable. -pub async fn find_pending_orders_for_reconcile( - pool: &SqlitePool, -) -> Result, MostroError> { - let now = Timestamp::now(); - let orders = sqlx::query_as::<_, Order>( - r#" - SELECT * - FROM orders - WHERE (expires_at >= ?1 OR expires_at = 0) - AND status IN ('pending', 'waiting-taker-bond') - "#, - ) - .bind(now.as_secs() as i64) - .fetch_all(pool) - .await - .map_err(|e| MostroInternalErr(ServiceError::DbAccessError(e.to_string())))?; - - Ok(orders) -} - pub async fn find_order_by_seconds(pool: &SqlitePool) -> Result, MostroError> { let mostro_settings = Settings::get_mostro(); let exp_seconds = mostro_settings.expiration_seconds as u64; @@ -2061,61 +2033,6 @@ mod tests { .unwrap(); } - /// Insert a minimal order row with an explicit status and expiry, for - /// the reconciler full-sweep query. - async fn insert_book_order(pool: &SqlitePool, status: &str, expires_at: i64) { - sqlx::query( - r#" - INSERT INTO orders (id, kind, event_id, status, premium, payment_method, - amount, fiat_code, fiat_amount, created_at, expires_at) - VALUES (?1, 'sell', 'event123', ?2, 0, 'bank', - 100000, 'USD', 100, 1700000000, ?3) - "#, - ) - .bind(uuid::Uuid::new_v4()) - .bind(status) - .bind(expires_at) - .execute(pool) - .await - .unwrap(); - } - - /// The full sweep re-asserts only live pending orders: expired rows are - /// the expiry job's business and post-trade rows are not book entries. - /// Legacy rows with `expires_at = 0` have no take-window TTL and must - /// stay re-assertable (mirroring `is_order_take_window_closed`). - #[tokio::test] - async fn find_pending_orders_for_reconcile_lists_only_live_pending_orders() { - let pool = setup_orders_db().await.unwrap(); - let now = nostr_sdk::prelude::Timestamp::now().as_secs() as i64; - - for (status, expires_at) in [ - ("pending", now + 3_600), // live → listed - ("waiting-taker-bond", now + 600), // publishes as pending → listed - ("pending", 0), // legacy, no TTL → listed - ("pending", now - 60), // expired → expiry job's business - ("active", now + 3_600), // post-trade → not a book entry - ("waiting-maker-bond", now + 600), // never published → skipped - ] { - insert_book_order(&pool, status, expires_at).await; - } - - let listed = super::find_pending_orders_for_reconcile(&pool) - .await - .unwrap(); - let statuses: Vec = listed.iter().map(|o| o.status.clone()).collect(); - - assert_eq!( - listed.len(), - 3, - "live pending-published rows plus legacy no-TTL: {statuses:?}" - ); - assert!(statuses.contains(&"waiting-taker-bond".to_string())); - assert!(listed - .iter() - .any(|o| o.status == "pending" && o.expires_at == 0)); - } - async fn setup_disputes_table(pool: &SqlitePool) { sqlx::query( r#" diff --git a/src/scheduler.rs b/src/scheduler.rs index fcbb4d59..57651e9f 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -1067,11 +1067,6 @@ async fn enforce_escrow_deadline_pass( /// How often the orderbook reconciler drains the failed-publish queue. const ORDERBOOK_RECONCILE_INTERVAL_SECS: u64 = 60; -/// Every this many reconciler ticks, the whole `pending` book is -/// re-asserted on relays (once per hour at the 60 s tick), healing states -/// relays lost for reasons the daemon never saw (dropped events, relay -/// restores from backup, NIP-01 ties lost before the monotonic stamp). -const ORDERBOOK_FULL_SWEEP_EVERY_TICKS: u64 = 60; /// A reconciler republish only proceeds when the order has not been /// stamped for this long. Handlers publish their kind-38383 revision @@ -1083,22 +1078,43 @@ const ORDERBOOK_FULL_SWEEP_EVERY_TICKS: u64 = 60; const ORDERBOOK_QUIESCENT_SECS: u64 = 120; /// One reconciler pass: republish every order whose last kind-38383 -/// publish failed (`util::take_failed_orderbook_publishes`), and — when -/// `full_sweep` — re-assert every live `pending` order from the DB. +/// publish failed (`util::take_failed_orderbook_publishes`). /// /// Every publish goes through `update_order_event_if_quiescent`: an order /// stamped within the last [`ORDERBOOK_QUIESCENT_SECS`] is skipped (queue -/// entries are kept for the next pass), so a sweep can never supersede a -/// transition that published before persisting its CAS. On publish +/// entries are kept for the next pass), so a republish can never supersede +/// a transition that published before persisting its CAS. On publish /// failure the order re-queues itself, so a relay outage self-heals on a -/// later pass. The returned order (fresh `event_id`) is deliberately not -/// persisted, matching the CAS-miss repair path: the DB row's status is -/// the source of truth being re-advertised, not mutated. -async fn reconcile_orderbook_once(pool: &sqlx::SqlitePool, keys: &Keys, full_sweep: bool) { - for (order_id, generation) in util::take_failed_orderbook_publishes() { +/// later pass — up to [`util::MAX_ORDERBOOK_REPUBLISH_ATTEMPTS`] failed +/// attempts, after which the entry is dropped: a persistently-failing +/// relay cannot be converged by republishing (each retry only rewrites +/// the healthy relays' copies with a fresh `created_at`), so the cap +/// trades the retry loop for a residual divergence bounded by NIP-40. +/// The returned order (fresh `event_id`) is deliberately not persisted, +/// matching the CAS-miss repair path: the DB row's status is the source +/// of truth being re-advertised, not mutated. +async fn reconcile_orderbook_once(pool: &sqlx::SqlitePool, keys: &Keys) { + for (order_id, generation, attempts) in util::take_failed_orderbook_publishes() { + if attempts >= util::MAX_ORDERBOOK_REPUBLISH_ATTEMPTS { + warn!( + "orderbook reconciler: giving up on order {order_id} after {attempts} failed \ + publish attempts — check the relay list; the residual divergence is bounded \ + by the event's NIP-40 expiration" + ); + continue; + } match Order::by_id(pool, order_id).await { Ok(Some(order)) => match order.get_order_status() { Ok(status) => { + // Re-seed the drained entry *before* attempting the + // republish: the send path records its own failures via + // `mark_orderbook_publish_failed_at`, which must + // increment on top of the consumed attempts — against + // an absent entry it would restart the count at 1 and + // the attempt cap would never trip. A full success + // clears the entry (its stamp generation is newer), a + // quiescent skip leaves it untouched for the next pass. + util::requeue_orderbook_publish_failure(order_id, generation, attempts); match util::update_order_event_if_quiescent( keys, status, @@ -1109,8 +1125,12 @@ async fn reconcile_orderbook_once(pool: &sqlx::SqlitePool, keys: &Keys, full_swe { Ok(Some(_)) => {} // Stamped too recently — another publication may be - // in flight; keep the failure queued for later. - Ok(None) => util::mark_orderbook_publish_failed_at(order_id, generation), + // in flight; the re-seeded entry keeps the failure + // queued without consuming an attempt. + Ok(None) => {} + // Pre-send failure (tags/ratings/event build): + // consume an attempt so a permanently broken order + // cannot occupy the queue forever. Err(e) => { warn!( "orderbook reconciler: republish of order {order_id} failed: {e}" @@ -1125,71 +1145,39 @@ async fn reconcile_orderbook_once(pool: &sqlx::SqlitePool, keys: &Keys, full_swe Ok(None) => {} Err(e) => { warn!("orderbook reconciler: could not reload order {order_id}: {e}"); - // Transient DB error: keep it queued for the next pass. - util::mark_orderbook_publish_failed_at(order_id, generation); - } - } - } - - if full_sweep { - match crate::db::find_pending_orders_for_reconcile(pool).await { - Ok(orders) => { - info!( - "orderbook reconciler: re-asserting {} pending order(s) on relays", - orders.len() - ); - for order in orders { - match order.get_order_status() { - Ok(status) => { - match util::update_order_event_if_quiescent( - keys, - status, - &order, - ORDERBOOK_QUIESCENT_SECS, - ) - .await - { - Ok(Some(_)) => {} - // Recently stamped — the normal publish - // path owns convergence for this order; a - // lost event is healed by the next sweep. - Ok(None) => {} - Err(e) => warn!( - "orderbook reconciler: re-assert of order {} failed: {e}", - order.id - ), - } - } - Err(e) => warn!( - "orderbook reconciler: order {} has bad status: {e}", - order.id - ), - } - } + // Transient DB error: keep it queued for the next pass + // without consuming an attempt — nothing was published. + util::requeue_orderbook_publish_failure(order_id, generation, attempts); } - Err(e) => warn!("orderbook reconciler: could not list pending orders: {e}"), } } } /// Keeps the public NIP-33 orderbook converged with the DB: drains the -/// failed-publish queue every minute and re-asserts the whole pending book -/// hourly (first sweep right at startup, healing relay state lost across -/// daemon restarts). Without this job a single dropped publish -/// leaves a dead order advertised as `pending` until its NIP-40 expiration. +/// failed-publish queue every minute. Without this job a single dropped +/// publish leaves a dead order advertised as `pending` until its NIP-40 +/// expiration. Orders whose publishes succeeded are never re-sent: a +/// kind-38383 revision is published once, when the order actually +/// transitions. +/// +/// Accepted trade-off: the queue is process-local and only records +/// failures the daemon observed. An entry queued right before a restart +/// is lost, and a relay that silently drops an event (or restores from a +/// backup) is never healed — periodic re-assertion of the whole book was +/// deliberately removed as unintended behavior. NIP-40 expiration bounds +/// any such divergence to the order's real take window. async fn job_orderbook_reconciler(ctx: AppContext) { let keys = ctx.keys().clone(); tokio::spawn(async move { let pool = ctx.pool(); - let mut tick: u64 = 0; loop { - let full_sweep = tick.is_multiple_of(ORDERBOOK_FULL_SWEEP_EVERY_TICKS); - reconcile_orderbook_once(pool, &keys, full_sweep).await; - tick = tick.wrapping_add(1); + // Sleep first: the queue is process-local, so at startup it is + // always empty and an immediate pass could only be a no-op. tokio::time::sleep(tokio::time::Duration::from_secs( ORDERBOOK_RECONCILE_INTERVAL_SECS, )) .await; + reconcile_orderbook_once(pool, &keys).await; } }); } @@ -1649,7 +1637,7 @@ mod tests { let ghost = Uuid::new_v4(); crate::util::mark_orderbook_publish_failed(ghost); - reconcile_orderbook_once(ctx.pool(), &keys, false).await; + reconcile_orderbook_once(ctx.pool(), &keys).await; assert!( !crate::util::is_orderbook_publish_queued(ghost), @@ -1681,7 +1669,7 @@ mod tests { let order = order.create(ctx.pool()).await.unwrap(); crate::util::mark_orderbook_publish_failed(order.id); - reconcile_orderbook_once(ctx.pool(), &keys, false).await; + reconcile_orderbook_once(ctx.pool(), &keys).await; assert!( crate::util::is_orderbook_publish_queued(order.id), @@ -1717,7 +1705,7 @@ mod tests { let _ = crate::util::stamp_orderbook_event(order.id, nostr_sdk::prelude::Timestamp::now()); crate::util::mark_orderbook_publish_failed(order.id); - reconcile_orderbook_once(ctx.pool(), &keys, false).await; + reconcile_orderbook_once(ctx.pool(), &keys).await; assert!( crate::util::is_orderbook_publish_queued(order.id), @@ -1727,6 +1715,139 @@ mod tests { let _ = crate::util::take_failed_orderbook_publishes(); } + /// Pins the invariant the reconciler now promises: an order whose + /// publishes all succeeded is never re-sent. A live `pending` order + /// that is not in the failed-publish queue must come out of a pass + /// untouched — nothing queued and, crucially, no kind-38383 revision + /// stamped for it. + #[tokio::test] + async fn reconciler_never_republishes_order_without_failed_publish() { + let ctx = migrated_ctx().await; + let keys = ctx.keys().clone(); + let _guard = crate::util::ORDERBOOK_QUEUE_TEST_LOCK.lock().await; + + let order = Order { + id: Uuid::new_v4(), + kind: Kind::Sell.to_string(), + status: Status::Pending.to_string(), + fiat_code: "USD".to_string(), + payment_method: "bank".to_string(), + expires_at: nostr_sdk::prelude::Timestamp::now().as_secs() as i64 + 3_600, + ..Default::default() + }; + let order = order.create(ctx.pool()).await.unwrap(); + + reconcile_orderbook_once(ctx.pool(), &keys).await; + + assert!( + !crate::util::is_orderbook_publish_queued(order.id), + "a healthy order must not end up queued by a reconciler pass" + ); + // If the pass had stamped (republished) the order, the monotonic + // registry would hold an entry near `now` and bump this past + // candidate to `last + 1`. Getting the candidate back unchanged + // proves no stamp was created for the order during the pass. + let probe = crate::util::stamp_orderbook_event( + order.id, + nostr_sdk::prelude::Timestamp::from(1_700_000_000), + ); + assert_eq!( + probe.created_at.as_secs(), + 1_700_000_000, + "a reconciler pass must not stamp an order that has no failed publish" + ); + } + + /// The drained entry is re-seeded before the republish attempt, so a + /// failed send increments the consumed-attempt count instead of + /// restarting it at 1 — and a quiescence skip consumes nothing. + #[tokio::test] + async fn reconciler_failed_retry_consumes_exactly_one_attempt() { + let ctx = migrated_ctx().await; + let keys = ctx.keys().clone(); + let _guard = crate::util::ORDERBOOK_QUEUE_TEST_LOCK.lock().await; + + let order = Order { + id: Uuid::new_v4(), + kind: Kind::Sell.to_string(), + status: Status::Canceled.to_string(), + fiat_code: "USD".to_string(), + payment_method: "bank".to_string(), + expires_at: nostr_sdk::prelude::Timestamp::now().as_secs() as i64 + 3_600, + ..Default::default() + }; + let order = order.create(ctx.pool()).await.unwrap(); + crate::util::mark_orderbook_publish_failed(order.id); + assert_eq!(crate::util::orderbook_publish_attempts(order.id), Some(1)); + + // No reachable relays in tests: the republish is attempted and + // fails, which must cost exactly one more attempt. + reconcile_orderbook_once(ctx.pool(), &keys).await; + assert_eq!( + crate::util::orderbook_publish_attempts(order.id), + Some(2), + "a failed republish must increment the consumed attempts, not reset them" + ); + + // The republish just stamped the order, so the next pass hits the + // quiescence guard: requeued without consuming an attempt. + reconcile_orderbook_once(ctx.pool(), &keys).await; + assert_eq!( + crate::util::orderbook_publish_attempts(order.id), + Some(2), + "a quiescence skip must not consume an attempt" + ); + + // Leave the shared queue clean for other tests. + let _ = crate::util::take_failed_orderbook_publishes(); + } + + /// Once an entry has consumed its attempt budget the reconciler drops + /// it instead of republishing again: the retry cannot converge a relay + /// that keeps failing — each extra cycle would only rewrite the healthy + /// relays' copies with a fresh `created_at`. The residual divergence is + /// bounded by the event's NIP-40 expiration. + #[tokio::test] + async fn reconciler_drops_entry_after_max_attempts() { + let ctx = migrated_ctx().await; + let keys = ctx.keys().clone(); + let _guard = crate::util::ORDERBOOK_QUEUE_TEST_LOCK.lock().await; + + let order = Order { + id: Uuid::new_v4(), + kind: Kind::Sell.to_string(), + status: Status::Canceled.to_string(), + fiat_code: "USD".to_string(), + payment_method: "bank".to_string(), + expires_at: nostr_sdk::prelude::Timestamp::now().as_secs() as i64 + 3_600, + ..Default::default() + }; + let order = order.create(ctx.pool()).await.unwrap(); + crate::util::requeue_orderbook_publish_failure( + order.id, + 1, + crate::util::MAX_ORDERBOOK_REPUBLISH_ATTEMPTS, + ); + + reconcile_orderbook_once(ctx.pool(), &keys).await; + + assert!( + !crate::util::is_orderbook_publish_queued(order.id), + "an entry at the attempt cap must be dropped, not retried" + ); + // No stamp may have been created for the dropped entry: probing + // with a past candidate must return it unchanged. + let probe = crate::util::stamp_orderbook_event( + order.id, + nostr_sdk::prelude::Timestamp::from(1_700_000_000), + ); + assert_eq!( + probe.created_at.as_secs(), + 1_700_000_000, + "a dropped entry must not be republished" + ); + } + // ── notify_users_canceled_order ────────────────────────────────────── #[tokio::test] diff --git a/src/util.rs b/src/util.rs index 8ead10e4..558c1681 100644 --- a/src/util.rs +++ b/src/util.rs @@ -498,7 +498,22 @@ async fn finalize_order_publication( .unwrap() .send_event(&event) .await - .map(|_s| ()) + .map(|output| { + // A per-relay rejection resolves to `Ok` with the refusing + // relays in `output.failed`. The publish still stands — the row + // is `Pending` and at least one side of the wire may have it — + // but the divergent relays must be converged, so queue the + // order for the reconciler. + if !output.failed.is_empty() { + tracing::warn!( + "initial orderbook publish rejected by {} relay(s) for order {}: {:?}; queued for republish", + output.failed.len(), + order_id, + output.failed + ); + mark_orderbook_publish_failed(order_id); + } + }) .map_err(|err| { // The row is already `Pending` in the DB; queue it so the // orderbook reconciler retries the publish instead of leaving @@ -1056,28 +1071,64 @@ pub(crate) fn monotonic_order_event_timestamp(order_id: Uuid, candidate: Timesta stamp_orderbook_event(order_id, candidate).created_at } -/// Orders whose latest kind-38383 publish failed (relay send error or no -/// Nostr client), so the DB state and the advertised orderbook diverged. -/// The scheduler's orderbook reconciler drains this map and republishes the -/// current DB state until the wire converges. +/// Orders whose latest kind-38383 publish failed (a relay rejected the +/// event, the send errored, or no Nostr client existed), so the DB state +/// and the advertised orderbook diverged. The scheduler's orderbook +/// reconciler drains this map and republishes the current DB state until +/// the wire converges — or until [`MAX_ORDERBOOK_REPUBLISH_ATTEMPTS`] is +/// reached. +/// +/// Each entry carries `(generation, attempts)`: /// -/// Each entry carries the publication generation that recorded the failure -/// (see [`ORDERBOOK_REGISTRY`]). A successful send clears an entry only if -/// the failure is not newer than itself: without the generation, a slow -/// old send completing *after* a newer publication failed would erase that -/// newer failure, and the state the newer publication carried would never -/// be republished. -static PENDING_ORDERBOOK_REPUBLISH: std::sync::LazyLock>> = - std::sync::LazyLock::new(|| std::sync::Mutex::new(HashMap::new())); - -/// Queue `order_id` for republish, recording the failure at `generation`. -/// A newer failure already recorded for the order is never downgraded. +/// - the publication generation that recorded the failure (see +/// [`ORDERBOOK_REGISTRY`]). A successful send clears an entry only if +/// the failure is not newer than itself: without the generation, a slow +/// old send completing *after* a newer publication failed would erase +/// that newer failure, and the state the newer publication carried +/// would never be republished; +/// - the number of failed publish attempts recorded for this divergence +/// episode. Without the cap, a single persistently-failing relay (offline, +/// rate-limiting, refusing the kind) would put every touched order into a +/// permanent republish cycle: each retry stamps a fresh `created_at` that +/// the *healthy* relays dutifully replace, churning the public book while +/// converging nothing — the retry cannot reach the relay that failed. +static PENDING_ORDERBOOK_REPUBLISH: std::sync::LazyLock< + std::sync::Mutex>, +> = std::sync::LazyLock::new(|| std::sync::Mutex::new(HashMap::new())); + +/// Failed publish attempts after which the reconciler gives up on an +/// order and drops its queue entry. Bounds the churn of a persistent +/// per-relay failure to this many published revisions per episode; the +/// residual divergence is bounded by the event's NIP-40 expiration, the +/// same trade-off the reconciler already accepts for silent relay loss. +/// A later transition that fails again re-queues the order with a fresh +/// attempt budget. +pub(crate) const MAX_ORDERBOOK_REPUBLISH_ATTEMPTS: u8 = 5; + +/// Queue `order_id` for republish, recording one more failed publish +/// attempt at `generation`. A newer failure already recorded for the +/// order is never downgraded. pub(crate) fn mark_orderbook_publish_failed_at(order_id: Uuid, generation: u64) { let mut queue = PENDING_ORDERBOOK_REPUBLISH .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - let entry = queue.entry(order_id).or_insert(generation); - *entry = (*entry).max(generation); + let entry = queue.entry(order_id).or_insert((generation, 0)); + entry.0 = entry.0.max(generation); + entry.1 = entry.1.saturating_add(1); +} + +/// Re-insert a drained entry without consuming an attempt, for reconciler +/// passes that did not publish anything (the quiescence guard refused, or +/// the order row could not be reloaded). Merges with any failure recorded +/// concurrently: neither the generation nor the attempt count is ever +/// downgraded. +pub(crate) fn requeue_orderbook_publish_failure(order_id: Uuid, generation: u64, attempts: u8) { + let mut queue = PENDING_ORDERBOOK_REPUBLISH + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let entry = queue.entry(order_id).or_insert((generation, attempts)); + entry.0 = entry.0.max(generation); + entry.1 = entry.1.max(attempts); } /// Queue `order_id` for republish stamped with a fresh generation, for @@ -1104,7 +1155,7 @@ pub(crate) fn clear_orderbook_publish_failure_up_to(order_id: Uuid, generation: let mut queue = PENDING_ORDERBOOK_REPUBLISH .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - if let Some(&failed_at) = queue.get(&order_id) { + if let Some(&(failed_at, _)) = queue.get(&order_id) { if failed_at <= generation { queue.remove(&order_id); } @@ -1112,13 +1163,15 @@ pub(crate) fn clear_orderbook_publish_failure_up_to(order_id: Uuid, generation: } /// Take the current set of orders awaiting republish (with the generation -/// that recorded each failure), leaving the queue empty. Failed retries -/// re-queue themselves via `update_order_event`. -pub(crate) fn take_failed_orderbook_publishes() -> Vec<(Uuid, u64)> { +/// that recorded each failure and the attempts consumed so far), leaving +/// the queue empty. Failed retries re-queue themselves via +/// `update_order_event`. +pub(crate) fn take_failed_orderbook_publishes() -> Vec<(Uuid, u64, u8)> { PENDING_ORDERBOOK_REPUBLISH .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) .drain() + .map(|(order_id, (generation, attempts))| (order_id, generation, attempts)) .collect() } @@ -1147,6 +1200,16 @@ pub(crate) fn is_orderbook_publish_queued(order_id: Uuid) -> bool { .contains_key(&order_id) } +/// Test-only visibility into an entry's consumed attempt count. +#[cfg(test)] +pub(crate) fn orderbook_publish_attempts(order_id: Uuid) -> Option { + PENDING_ORDERBOOK_REPUBLISH + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(&order_id) + .map(|&(_, attempts)| attempts) +} + pub async fn update_order_event( keys: &Keys, status: Status, @@ -1245,7 +1308,25 @@ async fn update_order_event_stamped( // Only failures recorded by publications stamped no later // than this one may be cleared: a newer concurrent // publication's failure must survive this older success. - Ok(_) => clear_orderbook_publish_failure_up_to(order.id, stamp.generation), + Ok(output) if output.failed.is_empty() => { + clear_orderbook_publish_failure_up_to(order.id, stamp.generation) + } + // A per-relay rejection or timeout resolves to `Ok` with the + // refusing relays in `output.failed` — `send_event` returns + // `Err` only when there was no relay to send to at all. Any + // rejected relay means the book is divergent there, so the + // publish counts as failed and stays queued until a + // republish converges it. + Ok(output) => { + tracing::warn!( + "orderbook publish rejected by {} relay(s) for order {} (status {}): {:?}; queued for republish", + output.failed.len(), + order_updated.id, + status, + output.failed + ); + mark_orderbook_publish_failed_at(order.id, stamp.generation); + } Err(e) => { tracing::warn!( "orderbook publish failed for order {} (status {}): {e}; queued for republish", @@ -2163,7 +2244,7 @@ mod tests { let transition = stamp_orderbook_event(order_id, now); assert!( try_stamp_orderbook_event_quiescent(order_id, now, quiet).is_none(), - "a sweep racing a just-stamped transition must be refused" + "a republish racing a just-stamped transition must be refused" ); assert!( try_stamp_orderbook_event_quiescent( @@ -2175,15 +2256,15 @@ mod tests { "still inside the quiet window" ); - let sweep = try_stamp_orderbook_event_quiescent( + let republish = try_stamp_orderbook_event_quiescent( order_id, Timestamp::from(now.as_secs() + quiet), quiet, ) - .expect("outside the quiet window the sweep must stamp"); - assert!(sweep.created_at > transition.created_at); + .expect("outside the quiet window the republish must stamp"); + assert!(republish.created_at > transition.created_at); assert!( - sweep.generation > transition.generation, + republish.generation > transition.generation, "generation order must match stamp order" ); } @@ -2258,10 +2339,52 @@ mod tests { // A drained entry stays drained until the next failure. let drained = take_failed_orderbook_publishes(); - assert!(drained.iter().any(|(id, _)| *id == order.id)); + assert!(drained.iter().any(|(id, _, _)| *id == order.id)); assert!(!is_orderbook_publish_queued(order.id)); } + /// Attempt accounting: every recorded failure consumes an attempt, a + /// requeue never does (and never downgrades), and a success resets the + /// episode by removing the entry entirely. + #[tokio::test] + async fn republish_attempts_count_failures_not_requeues() { + let _guard = ORDERBOOK_QUEUE_TEST_LOCK.lock().await; + let order_id = Uuid::new_v4(); + + mark_orderbook_publish_failed_at(order_id, 1); + mark_orderbook_publish_failed_at(order_id, 2); + assert_eq!( + orderbook_publish_attempts(order_id), + Some(2), + "each recorded failure must consume one attempt" + ); + + // A requeue preserves the count and never downgrades it. + requeue_orderbook_publish_failure(order_id, 2, 1); + assert_eq!( + orderbook_publish_attempts(order_id), + Some(2), + "a requeue must not downgrade the consumed attempts" + ); + requeue_orderbook_publish_failure(order_id, 2, 4); + assert_eq!( + orderbook_publish_attempts(order_id), + Some(4), + "a requeue must keep the highest attempt count seen" + ); + + // A success removes the entry: the next episode starts fresh. + clear_orderbook_publish_failure_up_to(order_id, 2); + assert_eq!(orderbook_publish_attempts(order_id), None); + mark_orderbook_publish_failed_at(order_id, 3); + assert_eq!( + orderbook_publish_attempts(order_id), + Some(1), + "a new failure after a success starts a fresh attempt budget" + ); + clear_orderbook_publish_failure_up_to(order_id, 3); + } + // ───────────────── take-window gate ───────────────── #[test]