Skip to content

fix: stop re-publishing the pending orderbook hourly and at startup #888

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
grunch merged 3 commits into from
Aug 18, 2026
+361 −184
Merged

fix: stop re-publishing the pending orderbook hourly and at startup #888

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
86f00b4
fix: stop re-publishing the pending orderbook hourly and at startup
Catrya Aug 17, 2026
2b68eab
fix: queue per-relay orderbook publish rejections for republish
Catrya Aug 18, 2026
cd55800
fix: cap orderbook republish attempts per failure episode
Catrya Aug 18, 2026
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
83 changes: 0 additions & 83 deletions src/db.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -675,34 +675,6 @@ pub async fn find_order_by_date(pool: &SqlitePool) -> Result<Vec<Order>, 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(

@grunch grunch Aug 17, 2026

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.

The created_at churn has a smaller root cause than the sweep.

The first and strongest justification in the body is that every re-assert stamps a fresh created_at because the monotonic registry is in-memory. That is accurate, but the sweep is the symptom rather than the cause:

  • orders.created_at is a persisted column — visible in the INSERT of the very test helper this PR removes;
  • ORDERBOOK_REGISTRY.last_ts (util.rs:960-976) simply starts empty, and its own doc admits it: "Across a restart the map starts empty";
  • update_order_event_stamped (util.rs:1219-1223) always passes Timestamp::now() as the candidate, in both StampPolicy arms, so nothing ever reuses the persisted value.

Seeding the registry from orders.created_at at startup — or reusing the persisted created_at when the published status has not changed — makes a re-assert idempotent: same d tag, same content, same created_at -> same event id -> relays dedupe it outright. That kills all three harms the body lists (the churn, the 24x/day traffic, and the orders.event_id drift, which disappears because the id no longer changes) without giving up the healing.

Caveat I will name myself: order_to_tags includes reputation data via get_ratings_for_pending_order, which can legitimately change between re-asserts and would produce a new id. That is a real update, not churn, so it does not undermine the approach.

Even if the deletion is still the preferred outcome, this deserves a sentence in the body on why the idempotent-re-assert path was rejected — right now the body reads as though re-publication is inherently churn-producing, and it is not.

pool: &SqlitePool,
) -> Result<Vec<Order>, 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<Vec<Order>, MostroError> {
let mostro_settings = Settings::get_mostro();
let exp_seconds = mostro_settings.expiration_seconds as u64;
Expand Down Expand Up @@ -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<String> = 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#"
Expand Down
74 changes: 13 additions & 61 deletions src/scheduler.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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`).

@grunch grunch Aug 17, 2026

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.

The failed-publish queue only fills on Err, so it cannot be the safety net this PR credits it as.

This doc line is now the whole convergence story, so it is worth checking what actually feeds that queue. At util.rs:1244-1258:

match client.send_event(&event).await {
    Ok(_) => clear_orderbook_publish_failure_up_to(order.id, stamp.generation),
    Err(e) => { warn!(...); mark_orderbook_publish_failed_at(order.id, stamp.generation); }
}

In nostr-sdk 0.45.1 (Cargo.toml:49), send_event resolves to Result<SendEventOutput, Error> where SendEventOutput = Output<EventId, EventSendStatus, String> (client/api/send_event.rs:19), and Output carries both success: HashMap<RelayUrl, S> and failed: HashMap<RelayUrl, F> (client/api/output.rs:14-21). Err is reserved for the "no relays to send to" class — Error::not_found and gossip_not_configured (client/api/send_event.rs:301,413), plus the pool's NoRelaysSpecified / NoRelays / RelayNotFound.

So a per-relay rejection or timeout lands in Ok, inside a failed map this code never reads — and that same Ok(_) arm additionally clears whatever queue entry the order already had. Relay A accepts, relay B rejects: nothing is queued, and the book stays divergent on B indefinitely.

That is precisely the gap the hourly sweep was papering over by brute force. Removing the sweep promotes it to the primary divergence path, with nothing left to heal it. The fix is short:

Ok(output) if output.failed.is_empty() =>
    clear_orderbook_publish_failure_up_to(order.id, stamp.generation),
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) => { /* unchanged */ }

This is pre-existing from #872, so it can land as a companion PR — but it should land with this one or before it, because this PR's risk assessment depends on it.

One caveat I will state rather than overclaim: I verified the failed map and the Err conditions in the SDK directly. The "every relay fails" case very likely also returns Ok — that is how nostr-relay-pool 0.44.3 behaves, which is the version I have vendored — but I could not confirm it against the exact pool 0.45.1 pins. The patch above covers both readings either way.

///
/// 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() {
Expand Down Expand Up @@ -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

@grunch grunch Aug 17, 2026

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.

This removes the only recovery for publishes lost across a restart, and the PR body does not say so.

PENDING_ORDERBOOK_REPUBLISH (util.rs:1070) is a LazyLock<Mutex<HashMap<Uuid, u64>>> — process-local, and empty again after every restart. The doc line this PR deletes named exactly that case: "first sweep right at startup, healing relay state lost across daemon restarts".

Two classes lose all coverage:

  • Queue lost across a restart. A publish fails at T and is queued in memory; the daemon restarts before the next 60s drain. The entry is gone, and the order stays divergent on relays for its entire take window.
  • Silent relay loss. A relay ACKs and later drops the event, or restores from a backup. The daemon never observed a failure, so nothing was ever queued, and nothing ever heals it.

The body states "Everything else from #872 stays intact, including the two layers that actually keep the book converged". Those two layers do not cover these: the queue only covers observed failures, and NIP-40 caps how long a stale event lives but cannot restore a missing one.

To be clear, I think this is a defensible trade — the author of #872 confirms the re-publication was never intended, the harms you list are real, and NIP-40 bounds the blast radius to the order's actual take window. My ask is narrow: state the trade explicitly here and in the body, instead of leaving it implicit behind "everything else stays intact". The deleted comment is the only place it was written down.

/// 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;

@grunch grunch Aug 17, 2026

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.

The first pass at startup is now a guaranteed no-op. (nit)

The loop calls reconcile_orderbook_once before its first sleep. That ordering was meaningful while tick == 0 triggered the startup full sweep; with the sweep gone, the in-memory queue is always empty at process start, so this first call can only ever do nothing.

Harmless either way — but moving the sleep to the top of the loop, or leaving a one-line comment, saves the next reader from working out what the startup pass is supposed to accomplish.

tokio::time::sleep(tokio::time::Duration::from_secs(
ORDERBOOK_RECONCILE_INTERVAL_SECS,
))
Expand Down Expand Up @@ -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;

@grunch grunch Aug 17, 2026

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.

No test pins the newly documented invariant.

The three retained tests all exercise the queue drain — ghost order, requeue while relays are unreachable, quiescence guard. None of them asserts the guarantee this PR newly documents at scheduler.rs:1132: "Orders whose publishes succeeded are never re-sent".

A cheap regression test closes that: seed a live pending order that is not in the failed-publish queue, run reconcile_orderbook_once, and assert no stamp was created for it. The existing test helpers already reach far enough — is_orderbook_publish_queued plus the registry are enough to observe it.

Deletions are the changes that regress most quietly. Without a test, nothing stops someone from re-adding a periodic sweep in six months for what will look like a perfectly good reason.


assert!(
!crate::util::is_orderbook_publish_queued(ghost),
Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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),
Expand Down
10 changes: 5 additions & 5 deletions src/util.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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"
);
}
Expand Down