From 86f00b43f42a85b79b962ed5bfd242083fa0f9bc Mon Sep 17 00:00:00 2001 From: Catrya <140891948+Catrya@users.noreply.github.com> Date: Mon, 17 Aug 2026 15:37:17 -0600 Subject: [PATCH 1/3] fix: stop re-publishing the pending orderbook hourly and at startup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The orderbook reconciler (introduced in #872) re-asserted every live pending / waiting-taker-bond order on relays once per hour, with the first sweep at daemon startup. This full re-publication was never the intended behavior (confirmed by the author of #872): an order's kind-38383 event should be published when the order actually transitions, and then left alone. The sweep was also actively harmful: - Every re-assert stamped a fresh created_at (the monotonic registry is in-memory, so after a restart nothing bounds the new stamp), so no pending order ever appeared older than about an hour on clients and sorting the book by date was meaningless. - Every pending order was re-sent to every relay 24 times a day — pointless traffic that relays may read as spam. - The re-published event id was deliberately not persisted, so orders.event_id drifted from the event actually live on relays. The sweep never healed the ghost order class the reconciler exists for either: its query only listed DB-pending rows, so an order that is terminal in the DB while a relay still advertises it as pending was never touched. That class remains covered by the two layers that actually fix it, which this commit keeps intact: the failed-publish queue (drained every 60 s, retrying only sends that are known to have failed) and the NIP-40 expiration tag capping any stale pending event at the order's real take window. --- src/db.rs | 83 ------------------------------------------------ src/scheduler.rs | 74 ++++++++---------------------------------- src/util.rs | 10 +++--- 3 files changed, 18 insertions(+), 149 deletions(-) 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..d087b28d 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,18 +1078,17 @@ 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) { +async fn reconcile_orderbook_once(pool: &sqlx::SqlitePool, keys: &Keys) { for (order_id, generation) in util::take_failed_orderbook_publishes() { match Order::by_id(pool, order_id).await { Ok(Some(order)) => match order.get_order_status() { @@ -1130,62 +1124,20 @@ async fn reconcile_orderbook_once(pool: &sqlx::SqlitePool, keys: &Keys, full_swe } } } - - 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 - ), - } - } - } - 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. 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); + reconcile_orderbook_once(pool, &keys).await; tokio::time::sleep(tokio::time::Duration::from_secs( ORDERBOOK_RECONCILE_INTERVAL_SECS, )) @@ -1649,7 +1601,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 +1633,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 +1669,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), diff --git a/src/util.rs b/src/util.rs index 8ead10e4..a0e1cfec 100644 --- a/src/util.rs +++ b/src/util.rs @@ -2163,7 +2163,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 +2175,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" ); } From 2b68eab01200c8e147d9fbecba0299ca281e551b Mon Sep 17 00:00:00 2001 From: Catrya <140891948+Catrya@users.noreply.github.com> Date: Mon, 17 Aug 2026 20:03:13 -0600 Subject: [PATCH 2/3] fix: queue per-relay orderbook publish rejections for republish send_event returns Err only when there was no relay to send to at all; a per-relay rejection or timeout resolves to Ok with the refusing relays in output.failed, which the publish paths never read - and the Ok arm even cleared the order's existing queue entry. Relay A accepts, relay B rejects: nothing was queued and the book stayed divergent on B indefinitely. All three queue feeders now treat a non-empty failed map as a failed publish: update_order_event_stamped, the initial pending publish, and the range child-order publish. The reconciler's doc now states the queue's limits explicitly (process local, observed failures only, divergence bounded by NIP-40), the loop sleeps before its first pass instead of draining an always-empty queue, and a new regression test pins that an order with no failed publish is never re-sent. --- src/app/release.rs | 22 ++++++++++++++++--- src/scheduler.rs | 54 +++++++++++++++++++++++++++++++++++++++++++++- src/util.rs | 37 +++++++++++++++++++++++++++++-- 3 files changed, 107 insertions(+), 6 deletions(-) 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/scheduler.rs b/src/scheduler.rs index d087b28d..a5cc1773 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -1132,16 +1132,25 @@ async fn reconcile_orderbook_once(pool: &sqlx::SqlitePool, keys: &Keys) { /// 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(); loop { - reconcile_orderbook_once(pool, &keys).await; + // 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; } }); } @@ -1679,6 +1688,49 @@ 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" + ); + } + // ── notify_users_canceled_order ────────────────────────────────────── #[tokio::test] diff --git a/src/util.rs b/src/util.rs index a0e1cfec..8ce68da0 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 @@ -1245,7 +1260,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", From cd55800f17dc34549bf12aaf317a207fb60ad8c2 Mon Sep 17 00:00:00 2001 From: Catrya <140891948+Catrya@users.noreply.github.com> Date: Tue, 18 Aug 2026 10:00:17 -0600 Subject: [PATCH 3/3] fix: cap orderbook republish attempts per failure episode Treating a partially-rejected send as a failed publish reopened a loop with no exit: one persistently-failing relay (offline, rate-limiting, refusing the kind) put every touched order into a permanent ~120s republish cycle, each retry stamping a fresh created_at that only the healthy relays applied - churning the public book at ~720 revisions per day per order while converging nothing, since the retry cannot reach the relay that failed. Queue entries now carry (generation, attempts). Every recorded publish failure consumes an attempt; after 5 the reconciler drops the entry with a loud warning and the residual divergence is bounded by the event's NIP-40 expiration - the same trade-off already accepted for silent relay loss. Reconciler passes that publish nothing (quiescence skip, transient DB error) requeue without consuming, the drained entry is re-seeded before each retry so the count survives the drain, a fully-successful publish still clears the entry, and a later failure starts a fresh budget. --- src/scheduler.rs | 133 +++++++++++++++++++++++++++++++++++++++++++--- src/util.rs | 134 +++++++++++++++++++++++++++++++++++++++-------- 2 files changed, 237 insertions(+), 30 deletions(-) diff --git a/src/scheduler.rs b/src/scheduler.rs index a5cc1773..57651e9f 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -1085,14 +1085,36 @@ const ORDERBOOK_QUIESCENT_SECS: u64 = 120; /// 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. +/// 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) in util::take_failed_orderbook_publishes() { + 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, @@ -1103,8 +1125,12 @@ async fn reconcile_orderbook_once(pool: &sqlx::SqlitePool, keys: &Keys) { { 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}" @@ -1119,8 +1145,9 @@ async fn reconcile_orderbook_once(pool: &sqlx::SqlitePool, keys: &Keys) { 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); + // 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); } } } @@ -1731,6 +1758,96 @@ mod tests { ); } + /// 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 8ce68da0..558c1681 100644 --- a/src/util.rs +++ b/src/util.rs @@ -1071,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 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. +/// Each entry carries `(generation, attempts)`: +/// +/// - 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 @@ -1119,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); } @@ -1127,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() } @@ -1162,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, @@ -2291,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]