Bounded reconciliation increment

docs/canonical-indexes.md

@@ -1,6 +1,6 @@
 # M2 increment — canonical indexes and recoverable market observation
 
-Status: implemented local infrastructure, **not full M2 acceptance or production authorization**. Evidence date: 2026-10-20. Related: [production plan](production-plan.md), [custodial design](custodial-market-design.md).
+Status: implemented local infrastructure, **not full M2 acceptance or production authorization**. Evidence date: 2026-10-21. Related: [production plan](production-plan.md), [custodial design](custodial-market-design.md).
 
 ## Delivered
 
@@ -9,13 +9,15 @@ Status: implemented local infrastructure, **not full M2 acceptance or production
 - Events contain sequence, connect/disconnect type, block identity/height/parent, transaction IDs and a hash chained to the previous event. An independently generated local stream ID binds the history; it is **not consensus data or a public proof of honest observation**. Pages contain at most 64 events and no article payloads. Financial/chain history is not silently pruned.
 - On first open, original blocks are replayed and independently proof/state validated before additive local index backfill. Later opens compare indexed block identity, ownership, transaction inclusion and event continuity/anchors to that validated chain. Inconsistent state fails closed; existing consensus version, genesis, signatures and free-transfer/trade rules are unchanged.
 - [Market reconciliation](../server/market/store.ts) uses indexed current delivery/ownership queries and a persisted `(stream ID, sequence, anchor)` checkpoint. Every observed disconnect of a previously completed order causes a durable review hold, even if delivery reappeared before polling. Review is never automatically repaid. Hold updates and checkpoint advancement commit together in the market database.
-- Catch-up consumes one bounded event page per reconciliation pass. While behind, ordinary market mutation/payout/expiry decisions stop; diagnostic/read access remains available and exposes event cursor/head/backlog. Periodic polling and subsequent requests advance catch-up. No claim of instant catch-up or background-thread isolation.
+- Order and listing review now uses stable-ID keyset batches capped at 128 rows each. A strict, durable job record stores stage, continuation ID, observed time, chain tip/revision, and the bounded canonical event page. Each batch commits row effects and its continuation atomically; the canonical event cursor advances only after both tables finish. A chain tip/revision change during a multi-batch scan rewinds the job and captures a fresh event page, preventing a mixed-view scan from being marked caught up. Completion checks the current event head so events appended during the scan remain behind the write fence.
+- While any job/event backlog exists, ordinary market mutations and time-dependent payout/expiry decisions stop; diagnostic/read access remains available. Subsequent requests advance bounded work, and the HTTP lifecycle has an unref'd 250ms poller that drives catch-up independently of user page loads. The poller is not a separately isolated worker: SQLite work is synchronous on the server event loop. There is no claim of instant catch-up, a throughput SLO, or horizontal-worker safety.
+- Reconciliation restart, mid-job event append/reorg fencing, disconnect review, and injected SQLite batch/cursor failures are tested. A failed batch leaves both order effects and durable keyset progress unchanged; restart resumes committed work. Restore mismatch (different stream, different history anchor or a cursor ahead of the restored ledger) is refused rather than resetting the cursor. Restore the matching stopped-node ledger/market backup; any alternative recovery requires reviewed evidence, not deleting a checkpoint.
 - Restore mismatch (different stream, different history anchor or a cursor ahead of the restored ledger) is refused rather than resetting the cursor. Restore the matching stopped-node ledger/market backup; any alternative recovery requires reviewed evidence, not deleting a checkpoint.
 - [HTTP NFT lookup and custody card checks](../server/testnet/http.ts) use the same index. Wallet deck pagination accepts an optional `tip` query; stale tips return 409. Existing callers without `tip` remain compatible. There is no new public event/admin endpoint, no payment observer and no NFT-lock signer.
 
 ## Tests and measurements
 
-The final Docker test-target build passed TypeScript/Vite compilation. All **149 tests passed, zero failures/skips**, including native RandomX multi-process HTTP/peer/restart and backup rehearsal. New failure evidence includes:
+The final Docker test-target build passed TypeScript/Vite compilation. All **174 offline tests passed, zero failures/skips**, including native RandomX multi-process HTTP/peer/restart and backup rehearsal. The focused market suite passed **23 tests**; the disposable PostgreSQL suite passed **31 tests**. New failure evidence includes:
 
 - indexed ownership/sequence versus full replay for transfers, multi-card trades and forks;
 - append/disconnect/reconnect ordering, side-branch/mempool exclusion, bounded defensive-copy pages and durable restart;
@@ -23,7 +25,9 @@ The final Docker test-target build passed TypeScript/Vite compilation. All **149
 - additive migration of a prior-schema copy, refusing corrupt ownership and event records;
 - production market adapter exercised with full snapshots forbidden;
 - multi-page catch-up and persisted restart checkpoint;
-- injected cursor-write failure rolling back the payout hold, followed by restart recovery;
+- 128-row order/listing keyset progress, restart continuation, and a new canonical event arriving during an in-flight job;
+- chain revision change fencing that resets continuation and refreshes the event page before the scan continues;
+- injected batch-write failure rolling back order effects and keyset progress, plus cursor-write failure rolling back checkpoint advancement while preserving committed review holds;
 - numeric cursor reuse with a different historical hash rejected;
 - real HTTP NFT results and stale-tip deck pagination.
 
@@ -44,6 +48,6 @@ These numbers are a reproducible **baseline**, not acceptance SLOs: host CPU/sto
 
 ## Remaining gates
 
-M2 still needs target-hardware/data/SLO decisions, market-order incremental scheduling rather than scanning every existing bounded order/listing, bounded worker/event-loop latency under load, production financial database/schema and measured recovery/retention/storage growth. Ledger PoW replay and DAG state memory still scale with retained history; indexed queries do not solve those costs. Startup validation checks undo-record counts but not every possible future undo mutation against a second reconstructed undo model; comprehensive corruption/fuzz coverage and independent review remain required.
+M2 still needs target-hardware/data/SLO decisions, populated long-journal/large-event benchmarks, bounded worker/event-loop latency under load, production financial database/schema and measured recovery/retention/storage growth. The new scheduler bounds work per pass, but does not yet provide queue isolation, multi-worker coordination, or a demonstrated production throughput envelope. Ledger PoW replay and DAG state memory still scale with retained history; indexed queries do not solve those costs. Startup validation checks undo-record counts but not every possible future undo mutation against a second reconstructed undo model; comprehensive corruption/fuzz coverage and independent review remain required.
 
 Custodial reservations still stop at `awaiting-lock`. This increment does **not** consume ordinary transfers as custodial settlement evidence: doing that would bypass the reviewed sale-lock/cancel protocol. Real Monero deposit/withdrawal integration, production treasury keys, host attestations and a reviewed versioned NFT lock/completion/cancellation specification remain blocked. No real funds, credentials, external processor or genesis change were used.

docs/production-plan.md

@@ -1,6 +1,6 @@
 # Production delivery plan
 
-Status: active engineering plan; **not launch or payment authorization**. Updated 2026-10-20. The objective is a production service with explicit guarantees, measurable limits and accountable operations—not relabeling the existing testnet. Existing implementations are reusable groundwork; findings are closed only by reviewed implementation and evidence.
+Status: active engineering plan; **not launch or payment authorization**. Updated 2026-10-21. The objective is a production service with explicit guarantees, measurable limits and accountable operations—not relabeling the existing testnet. Existing implementations are reusable groundwork; findings are closed only by reviewed implementation and evidence.
 
 ## 1. Decisions agreed with the operator
 
@@ -63,7 +63,7 @@ Acceptance: exact-amount property/boundary tests, versioned signature tests, zer
 
 Can run alongside P0/P1. Addresses SEC-01/SEC-06 rather than adding more fixed caps.
 
-**Implemented M2 increment, 2026-10-20:** [canonical indexes and durable market observation](canonical-indexes.md). Canonical ownership/transaction SQL indexes now serve NFT, deck, custody availability and mock-market queries; per-block undo and anchored connect/disconnect events commit with ledger updates. Market checkpoint/reorg holds persist atomically and bounded catch-up blocks new mutations. Final TypeScript/Vite build and 149 offline tests passed. A disposable 10,000-NFT SQL baseline is recorded, **not** a target-hardware SLO or complete M2 acceptance. Production financial storage, incremental order scheduling, retained-DAG capacity and operator dataset/latency targets remain open.
+**Implemented M2 increments, 2026-10-20–21:** [canonical indexes and durable market observation](canonical-indexes.md). Canonical ownership/transaction SQL indexes now serve NFT, deck, custody availability and mock-market queries; per-block undo and anchored connect/disconnect events commit with ledger updates. Market checkpoint/reorg holds persist atomically and bounded catch-up blocks new mutations. Order/listing reconciliation now advances in durable, stable-ID keyset batches of at most 128 records per table, resumes after restart, re-fences when the chain tip/revision changes, and runs from an HTTP lifecycle poller rather than relying only on page loads. Final TypeScript/Vite build and **174 offline tests passed**; focused market tests passed 23/23 and the disposable PostgreSQL integration suite passed 31/31. A disposable 10,000-NFT SQL baseline is recorded, **not** a target-hardware SLO or complete M2 acceptance. Queue isolation, target-scale throughput/latency, production financial storage/recovery, retained-DAG capacity and operator dataset/SLO targets remain open.
 
 **Next financial-storage increment:** [ADR 003 — PostgreSQL](postgres-financial-storage.md) selects and integrates the custody SQL backend with existing HTTP callers, exact constrained accounting, explicit migration/runtime roles, source-freezing SQLite import and dump/restore rehearsal. The adapter remains on the fictitious sandbox domain; it is not a production deposit/withdrawal worker. SQLite node-only backup deliberately refuses PostgreSQL-bound financial state. Production hardware/SLO tests, coordinated backup/PITR/failover, real payment schema/worker controls and independent review remain open, so this does not close M2/M4 or authorize funds.
 
@@ -72,7 +72,7 @@ Can run alongside P0/P1. Addresses SEC-01/SEC-06 rather than adding more fixed c
 **Deployment increment:** [persistent local custody stack](persistent-custody-sandbox.md) now starts PostgreSQL and the actual app together, with private TLS/SCRAM networking, separate owner/runtime roles, explicit migrations and scoped persistent credential/data volumes. A signed fictitious credit and custodian identity survived container/network removal and recreation; the server rejected plaintext authentication. Existing deployments and SQLite custody were not switched. Production certificate rotation, coordinated backup/restore and external operating review remain open; all XMR remains simulated.
 
 - Indexed canonical ownership, NFT/transfer lookup and paged wallet/explorer APIs: browser signing must not require loading all history. Tip-pinned queries detect reorgs and preserve complete ownership semantics.
-- Incremental market reconciliation from durable canonical/reorg events. Maintain rollback-safe indexes and a recoverable cursor; catch-up after restart must not miss delivery or payout holds. Bound worker time/concurrency; move expensive work off the request event loop where appropriate.
+- Incremental market reconciliation from durable canonical/reorg events. **Local bounded implementation delivered:** rollback-safe canonical event pages, durable 128-row order/listing keyset jobs, tip/revision re-fencing, restart continuation, mutation fencing, and background catch-up polling; regression tests cover appends/reorgs and injected SQLite failures. **Still open:** move synchronous SQLite batches off the HTTP event loop, coordinate a single worker across processes, measure lag/throughput and request latency at declared order/event volumes, and demonstrate overload/recovery limits before production claims.
 - A versioned production market/settlement schema with append-only financial events and balanced accounting entries for buyer liabilities, host revenue, author liabilities, seller liabilities, paid/refunded amounts and actual network fees. Derived views are rebuildable; financial history and idempotency keys are not discarded to reclaim space.
 - Durable inbox/outbox, processor attempts and reconciliation states; no network operation inside a long DB transaction. Store intent before external side effects; unknown POST outcomes stay quarantined until evidenced resolution. Per-market isolation, actor-scoped paged reads, queue/backpressure and tenant/global exposure budgets.
 - Database architecture ADR: retain a measured single-writer SQLite deployment if it meets declared limits, or choose a production DB/worker topology and supported migration. Do not claim horizontal safety merely by launching multiple workers against local state. Start with one authoritative writer per market.
@@ -155,7 +155,7 @@ Acceptance: signed release evidence packet; recorded unresolved accepted risks w
 
 First implementation increment: write the M0 threat model/ADR research questions and P1 XMR domain specification, then build exact amount/property tests. Do not spend the next increment adding another disconnected “live” scaffold. Each milestone must integrate with its actual callers, persistence, recovery, UI and failure tests before being described as delivered.
 
-Current increment status: the operator selected custodial implementation after [settlement research](settlement-feasibility.md). M1 has a signed v3 XMR-native mock path, with historical v2 USD orders preserved. [ADR 002](custodial-market-design.md) records the implemented isolated custody accounting/UI foundation; production custody is not complete. The next M2 infrastructure increment is [canonical indexing and durable reconciliation](canonical-indexes.md), now built and regression-tested. Next engineering gates include the remaining production accounting/worker envelope and the separately reviewed NFT lock/completion/cancellation specification (M3); no live Monero integration or consensus activation is authorized by passing these tests. Migration dry-run, final fee/risk policy, capacity SLOs and independent review remain open.
+Current increment status: the operator selected custodial implementation after [settlement research](settlement-feasibility.md). M1 has a signed v3 XMR-native mock path, with historical v2 USD orders preserved. [ADR 002](custodial-market-design.md) records the implemented isolated custody accounting/UI foundation; production custody is not complete. [Canonical indexing and durable reconciliation](canonical-indexes.md) now includes bounded, restartable, chain-revision-fenced market catch-up and has passed focused, offline, and disposable PostgreSQL regression suites. This closes one incremental-scheduling implementation gap, **not** M2 acceptance. Next gates include target-scale benchmarks, off-event-loop worker design, production accounting/recovery envelope and the separately reviewed NFT lock/completion/cancellation specification (M3); no live Monero integration or consensus activation is authorized by passing these tests. Migration dry-run, final fee/risk policy, capacity SLOs and independent review remain open.
 
 ## 5. Acceptance evidence and release discipline
 

server/market/store.ts

@@ -38,8 +38,22 @@ export function marketConfig(env: NodeJS.ProcessEnv = process.env): Config {
 }
 const PAYMENT_WINDOW = 30 * 60_000;
 const DELIVERY_WINDOW = 48 * 60 * 60_000;
+const RECONCILIATION_FRESH_MS = 15_000;
 const cursorSchema = z.object({ version: z.literal(1), streamId: z.string().uuid(),
   cursor: z.number().int().nonnegative().safe(), anchor: z.string().regex(/^[a-f0-9]{64}$/) }).strict();
+const canonicalEventSchema = z.object({ sequence: z.number().int().positive().safe(), type: z.enum(['connect', 'disconnect']),
+  height: z.number().int().positive().safe(), hash: z.string().regex(/^[a-f0-9]{64}$/),
+  previousHash: z.string().regex(/^[a-f0-9]{64}$/), transactionIds: z.array(z.string().regex(/^[a-f0-9]{64}$/)).max(64),
+  eventHash: z.string().regex(/^[a-f0-9]{64}$/) }).strict();
+const canonicalEventPageSchema = z.object({ streamId: z.string().uuid(), previousAnchor: z.string().regex(/^[a-f0-9]{64}$/),
+  anchor: z.string().regex(/^[a-f0-9]{64}$/), events: z.array(canonicalEventSchema).max(64),
+  cursor: z.number().int().nonnegative().safe(), head: z.number().int().nonnegative().safe(), caughtUp: z.boolean() }).strict();
+const reconciliationJobSchema = z.object({ version: z.literal(1), stage: z.enum(['orders', 'listings']),
+  afterId: z.string().nullable(), observedAt: z.number().int().nonnegative().safe(),
+  tip: z.string().regex(/^[a-f0-9]{64}$/).nullable(), revision: z.number().int().nonnegative().safe().nullable(),
+  eventPage: canonicalEventPageSchema.nullable() }).strict();
+const RECONCILE_BATCH_SIZE = 128;
+const RECONCILIATION_JOB_KEY = 'canonical_reconciliation_job';
 
 /** Node-local marketplace, NOT consensus escrow. Its SQLite journal is backed up with the ledger. */
 export class Market {
@@ -49,6 +63,7 @@ export class Market {
   private eventCursor = 0;
   private eventCheckpoint: z.infer<typeof cursorSchema> | undefined;
   private eventBacklog = false;
+  private reconciliationJob: z.infer<typeof reconciliationJobSchema> | undefined;
   private readonly db: DatabaseSync;
   private readonly chain: Chain;
   readonly config: Config;
@@ -61,6 +76,9 @@ export class Market {
       this.eventCheckpoint = cursorSchema.parse(JSON.parse(cursor.value));
       this.eventCursor = this.eventCheckpoint.cursor;
     }
+    const job = db.prepare('SELECT value FROM market_meta WHERE key=?').get(RECONCILIATION_JOB_KEY) as { value: string } | undefined;
+    if (job) { this.reconciliationJob = reconciliationJobSchema.parse(JSON.parse(job.value)); this.eventBacklog = true; }
+    else this.eventBacklog = !!chain.canonicalHead && this.eventCursor < chain.canonicalHead();
   }
 
   static async open(directory: string, chain: Chain, config = marketConfig(), now = Date.now): Promise<Market> {
@@ -101,15 +119,16 @@ export class Market {
     return { queued:this.queued,lastReconciledAt:this.reconciled?.at??null,orderStates:states,
       canonicalEventCursor:this.eventCursor,canonicalEventHead:this.chain.canonicalHead?.()??null,eventBacklog:this.eventBacklog };
   }
+  hasReconciliationBacklog(): boolean { return this.eventBacklog || !!this.reconciliationJob; }
   info(): MarketInfo {
     return { network: NETWORK, chainId: CHAIN_ID, marketId: this.id, paymentMode: 'mock', marketVersion: 3, feeBps: this.config.feeBps,
       ...resolvedMarketPolicy(this.config),
       additionalFee: { ...ADDITIONAL_MARKET_FEE },
       deliveryConfirmations: this.config.confirmations, serverTime: this.now() };
   }
-  private rows<T>(table: 'listings' | 'orders'): T[] {
-    return (this.db.prepare(`SELECT data FROM ${table} ORDER BY rowid DESC LIMIT 10000`).all() as { data: string }[])
-      .map(row => JSON.parse(row.data) as T);
+  private nextReconciliationRows(table: 'listings' | 'orders', afterId: string | null): { id: string; data: string }[] {
+    return this.db.prepare(`SELECT id,data FROM ${table} WHERE (? IS NULL OR id>?) ORDER BY id LIMIT ?`)
+      .all(afterId, afterId, RECONCILE_BATCH_SIZE) as { id: string; data: string }[];
   }
   private listing(id: string): MarketListingRecord {
     const row = this.db.prepare('SELECT data FROM listings WHERE id=?').get(id) as { data: string } | undefined;
@@ -159,88 +178,142 @@ export class Market {
     return transactionId({ network: NETWORK, chainId: CHAIN_ID, cardHash: order.cardHash,
       from: order.seller, to: order.buyer, sequence: order.sequence, publicKey: '', signature: '' });
   }
-  /** Reconcile from canonical transaction identity, never from a mempool acknowledgement. */
+  /** Reconcile one durable keyset batch from canonical transaction identity, never mempool acknowledgement. */
   private reconcile() {
-    const events = this.chain.canonicalEvents?.(this.eventCursor, 64);
-    if (events && this.eventCheckpoint && (events.streamId !== this.eventCheckpoint.streamId ||
-      events.previousAnchor !== this.eventCheckpoint.anchor)) throw new RpcError(409, 'Canonical event history changed; restore matching ledger and market backups.');
-    const s = this.view(), now = this.now();
-    const pendingIds = new Set(s.pendingTransfers.map(transactionId));
+    const now = this.now();
+    const summary = this.chain.summary?.();
+    if (!this.reconciliationJob) {
+      const events = this.chain.canonicalEvents?.(this.eventCursor, 64);
+      if (events && this.eventCheckpoint && (events.streamId !== this.eventCheckpoint.streamId ||
+        events.previousAnchor !== this.eventCheckpoint.anchor)) throw new RpcError(409, 'Canonical event history changed; restore matching ledger and market backups.');
+      const job = reconciliationJobSchema.parse({ version: 1, stage: 'orders', afterId: null, observedAt: now,
+        tip: summary?.tip ?? null, revision: summary?.revision ?? null, eventPage: events ?? null });
+      this.atomic(() => this.db.prepare('INSERT INTO market_meta(key,value) VALUES(?,?)')
+        .run(RECONCILIATION_JOB_KEY, JSON.stringify(job)));
+      this.reconciliationJob = job; this.eventBacklog = true;
+    }
+
+    let job = this.reconciliationJob;
+    if (summary && (job.tip !== summary.tip || job.revision !== summary.revision)) {
+      // Canonical state changed while a multi-batch scan was running. Rewind the
+      // keyset and bind a fresh event page so every row is checked against one
+      // observed chain revision before the canonical cursor can advance.
+      const events = this.chain.canonicalEvents?.(this.eventCursor, 64);
+      if (events && this.eventCheckpoint && (events.streamId !== this.eventCheckpoint.streamId ||
+        events.previousAnchor !== this.eventCheckpoint.anchor)) throw new RpcError(409, 'Canonical event history changed; restore matching ledger and market backups.');
+      job = reconciliationJobSchema.parse({ version: 1, stage: 'orders', afterId: null, observedAt: now,
+        tip: summary.tip, revision: summary.revision, eventPage: events ?? null });
+      this.atomic(() => this.db.prepare('UPDATE market_meta SET value=? WHERE key=?')
+        .run(JSON.stringify(job), RECONCILIATION_JOB_KEY));
+      this.reconciliationJob = job;
+      this.eventBacklog = true;
+    }
+    const s = this.view(), pendingIds = new Set(s.pendingTransfers.map(transactionId));
+    const eventPage = job.eventPage as CanonicalEvents | null;
+    const disconnected = new Set(eventPage?.events.filter(e => e.type === 'disconnect').flatMap(e => e.transactionIds) ?? []);
+    const rows = this.nextReconciliationRows(job.stage, job.afterId);
+    let nextJob: z.infer<typeof reconciliationJobSchema> | undefined;
+    let completed = false;
+
     this.atomic(() => {
-      // Process every disconnect, even if the same delivery reappeared before this consumer caught up.
-      // Never infer that a current tip erases an intermediate loss of previously settled evidence.
-      const disconnected = new Set(events?.events.filter(e => e.type === 'disconnect').flatMap(e => e.transactionIds) ?? []);
-      if (events) {
-        for (const o of this.rows<MarketOrderRecord>('orders')) {
+      if (job.stage === 'orders' && rows.length) {
+        for (const row of rows) {
+          const o = JSON.parse(row.data) as MarketOrderRecord;
+          const l = this.listing(o.listingId);
+          // A disconnect of completed delivery is sticky even if the same transaction has reappeared.
           if (o.status === 'complete' && disconnected.has(this.deliveryId(o))) {
             o.status = 'review'; o.payout = 'review';
             if (o.additionalFee) o.additionalFee.payout = 'review';
             this.saveOrder(o);
           }
-        }
-        this.db.prepare('INSERT INTO market_meta VALUES(?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value')
-          .run('canonical_event_cursor', JSON.stringify({ version: 1, streamId: events.streamId, cursor: events.cursor, anchor: events.anchor }));
-        // Catch-up is bounded per pass. No new payout, timeout/refund decision or mutation while events remain.
-        if (!events.caughtUp) return;
-      }
-      for (const o of this.rows<MarketOrderRecord>('orders')) {
-        const l = this.listing(o.listingId);
-        if (o.status === 'refunded' || o.status === 'expired') continue;
-        const deliveryId = this.deliveryId(o);
-        const delivery = s.transaction(deliveryId);
-        const confirmations = delivery ? s.height - delivery.height + 1 : 0;
-        o.deliveryConfirmations = confirmations;
-        if (o.status === 'complete' || o.status === 'review') {
-          if (!delivery || confirmations < o.requiredConfirmations) {
+          if (o.status === 'refunded' || o.status === 'expired') continue;
+          const deliveryId = this.deliveryId(o);
+          const delivery = s.transaction(deliveryId);
+          const confirmations = delivery ? s.height - delivery.height + 1 : 0;
+          o.deliveryConfirmations = confirmations;
+          if (o.status === 'complete' || o.status === 'review') {
+            if (!delivery || confirmations < o.requiredConfirmations) {
+              o.status = 'review'; o.payout = 'review';
+              if (o.additionalFee) o.additionalFee.payout = 'review';
+            }
+            this.saveOrder(o); continue; // Never automatically pay a second time after a reorganization.
+          }
+          if (delivery && o.status === 'awaiting-payment') {
             o.status = 'review'; o.payout = 'review';
             if (o.additionalFee) o.additionalFee.payout = 'review';
+          } else if (delivery) {
+            o.transferId = deliveryId; o.status = 'delivery-pending';
+            if (confirmations >= o.requiredConfirmations) {
+              o.status = 'complete'; o.payout = 'mock-paid'; l.status = 'sold';
+              if (o.additionalFee) o.additionalFee.payout = 'mock-paid';
+            }
+          } else if (o.status === 'awaiting-payment' && job.observedAt >= o.payBy) {
+            o.status = 'expired'; l.status = l.kind === 'fixed' ? 'open' : 'expired';
+          } else if (o.status !== 'awaiting-payment') {
+            const pending = pendingIds.has(deliveryId);
+            const conflict = s.pendingTransfers.some(tx => transactionCardIds(tx).includes(o.cardHash) && transactionId(tx) !== deliveryId);
+            if (!this.card(s, o.cardHash, o.seller, o.sequence) || conflict || (job.observedAt >= o.deliverBy && !pending)) o.status = 'refund-required';
+            else if (o.status === 'delivery-pending' && !pending) o.status = 'paid';
           }
-          this.saveOrder(o); continue; // Never automatically pay a second time after a reorganization.
+          this.saveOrder(o); this.saveListing(l);
         }
-        if (delivery && o.status === 'awaiting-payment') {
-          // A gift/direct delivery is not evidence that payment was received. Never pay an unpaid order.
-          o.status = 'review'; o.payout = 'review';
-          if (o.additionalFee) o.additionalFee.payout = 'review';
-        } else if (delivery) {
-          o.transferId = deliveryId;
-          o.status = 'delivery-pending';
-          if (confirmations >= o.requiredConfirmations) {
-            o.status = 'complete'; o.payout = 'mock-paid'; l.status = 'sold';
-            if (o.additionalFee) o.additionalFee.payout = 'mock-paid';
+      } else if (job.stage === 'listings' && rows.length) {
+        for (const row of rows) {
+          const l = JSON.parse(row.data) as MarketListingRecord;
+          if (l.status !== 'open') continue;
+          if (l.additionalFee?.bps !== ADDITIONAL_MARKET_FEE.bps || l.additionalFee?.address !== ADDITIONAL_MARKET_FEE.address ||
+              !this.card(s, l.cardHash, l.seller, l.sequence) || s.pendingTransfers.some(tx => transactionCardIds(tx).includes(l.cardHash))) l.status = 'invalid';
+          else if (l.kind === 'auction' && job.observedAt >= l.endsAt! + PAYMENT_WINDOW) l.status = 'expired';
+          else if (l.kind === 'auction' && job.observedAt >= l.endsAt! && !l.highestBid) l.status = 'expired';
+          this.saveListing(l);
+        }
+      }
+
+      if (job.stage === 'orders') {
+        if (rows.length < RECONCILE_BATCH_SIZE) nextJob = { ...job, stage: 'listings', afterId: null };
+        else nextJob = { ...job, afterId: rows.at(-1)!.id };
+      } else {
+        if (rows.length < RECONCILE_BATCH_SIZE) {
+          this.db.prepare('DELETE FROM receipts WHERE expires < ?').run(job.observedAt - 86400_000);
+          if (eventPage) {
+            this.db.prepare('INSERT INTO market_meta(key,value) VALUES(?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value')
+              .run('canonical_event_cursor', JSON.stringify({ version: 1, streamId: eventPage.streamId,
+                cursor: eventPage.cursor, anchor: eventPage.anchor }));
           }
-        } else if (o.status === 'awaiting-payment' && now >= o.payBy) {
-          o.status = 'expired'; l.status = l.kind === 'fixed' ? 'open' : 'expired';
-        } else if (o.status !== 'awaiting-payment') {
-          const pending = pendingIds.has(deliveryId);
-          const conflict = s.pendingTransfers.some(tx => transactionCardIds(tx).includes(o.cardHash) && transactionId(tx) !== deliveryId);
-          if (!this.card(s, o.cardHash, o.seller, o.sequence) || conflict || (now >= o.deliverBy && !pending)) o.status = 'refund-required';
-          else if (o.status === 'delivery-pending' && !pending) o.status = 'paid';
-        }
-        this.saveOrder(o); this.saveListing(l);
+          this.db.prepare('DELETE FROM market_meta WHERE key=?').run(RECONCILIATION_JOB_KEY);
+          completed = true;
+        } else nextJob = { ...job, afterId: rows.at(-1)!.id };
       }
-      for (const l of this.rows<MarketListingRecord>('listings')) {
-        if (l.status !== 'open') continue;
-        if (l.additionalFee?.bps !== ADDITIONAL_MARKET_FEE.bps || l.additionalFee?.address !== ADDITIONAL_MARKET_FEE.address ||
-          !this.card(s, l.cardHash, l.seller, l.sequence) || s.pendingTransfers.some(tx => transactionCardIds(tx).includes(l.cardHash))) l.status = 'invalid';
-        else if (l.kind === 'auction' && now >= l.endsAt! + PAYMENT_WINDOW) l.status = 'expired';
-        else if (l.kind === 'auction' && now >= l.endsAt! && !l.highestBid) l.status = 'expired';
-        this.saveListing(l);
-      }
-      this.db.prepare('DELETE FROM receipts WHERE expires < ?').run(now - 86400_000);
+      if (nextJob) this.db.prepare('UPDATE market_meta SET value=? WHERE key=?').run(JSON.stringify(nextJob), RECONCILIATION_JOB_KEY);
     });
-    if (events) {
-      this.eventCursor = events.cursor; this.eventBacklog = !events.caughtUp;
-      this.eventCheckpoint = { version: 1, streamId: events.streamId, cursor: events.cursor, anchor: events.anchor };
+
+    if (completed) {
+      this.reconciliationJob = undefined;
+      if (eventPage) {
+        this.eventCursor = eventPage.cursor;
+        this.eventCheckpoint = { version: 1, streamId: eventPage.streamId, cursor: eventPage.cursor, anchor: eventPage.anchor };
+        // The ledger can append events while large journals are processed over many bounded calls.
+        // Do not mark the market reconciled to a newer tip until those post-page events are consumed.
+        const currentHead = this.chain.canonicalHead?.() ?? eventPage.head;
+        this.eventBacklog = !eventPage.caughtUp || currentHead > eventPage.cursor;
+      } else this.eventBacklog = false;
+      const current = this.chain.summary?.();
+      if (current && !this.eventBacklog) this.reconciled = { ...current, at: now };
+    } else {
+      this.reconciliationJob = nextJob;
+      this.eventBacklog = true;
     }
-    const summary = this.chain.summary?.();
-    if (summary && !this.eventBacklog) this.reconciled = { ...summary, at: now };
+    // Empty/short order pages can advance to the listing stage in the same bounded call. This
+    // avoids rejecting the first command on an empty market while still capping work at 2×128 rows.
+    if (!completed && job.stage === 'orders' && rows.length < RECONCILE_BATCH_SIZE) this.reconcile();
   }
   browse(offset = 0): { listings: MarketListingRecord[]; next: number | null; info: MarketInfo } {
     if (!Number.isSafeInteger(offset) || offset < 0 || offset > 10000) throw new Error('Invalid marketplace page');
     // Reconciliation must not interleave with delivery submission while its chain await is outstanding.
     const summary = this.chain.summary?.();
     const last = this.reconciled;
-    if (!this.queued && (!summary || !last || summary.tip !== last.tip || summary.revision !== last.revision || this.now() - last.at >= 1000)) this.reconcile();
+    if (!this.queued && (this.hasReconciliationBacklog() || !summary || !last || summary.tip !== last.tip ||
+      summary.revision !== last.revision || this.now() - last.at >= RECONCILIATION_FRESH_MS)) this.reconcile();
     if (!resolvedMarketPolicy(this.config).marketEnabled) return { listings: [], next: null, info: this.info() };
     const rows = this.db.prepare("SELECT data FROM listings WHERE status IN ('open','reserved') ORDER BY rowid DESC LIMIT 25 OFFSET ?")
       .all(offset) as { data: string }[];
@@ -273,8 +346,8 @@ export class Market {
     if (!permitsMarketAction(this.config, c.action.type)) throw new Error('New market trading is disabled by this host; existing activity and resolution remain available');
     const a = c.action;
     const summary = this.chain.summary?.(), last = this.reconciled;
-    if (a.type !== 'activity' || !summary || !last || summary.tip !== last.tip || summary.revision !== last.revision ||
-      this.now() - last.at >= 1000) this.reconcile();
+    if (this.hasReconciliationBacklog() || a.type !== 'activity' || !summary || !last || summary.tip !== last.tip || summary.revision !== last.revision ||
+      this.now() - last.at >= RECONCILIATION_FRESH_MS) this.reconcile();
     if (this.eventBacklog && a.type !== 'activity') throw new RpcError(503, 'Market canonical reconciliation is catching up; retry shortly.');
     if (a.type === 'activity') return {
       orders: (this.db.prepare("SELECT data FROM orders WHERE json_extract(data,'$.seller')=? OR json_extract(data,'$.buyer')=? ORDER BY rowid DESC")
@@ -343,7 +416,9 @@ export class Market {
       if ((a.hostFeeAddress ?? ADDITIONAL_MARKET_FEE.address) !== hostFeeAddress) throw new Error('Host fee recipient changed; review and sign again');
       if (!this.card(s, a.cardHash, c.actor, a.sequence) || s.pendingTransfers.some(tx => transactionCardIds(tx).includes(a.cardHash))) throw new Error('Card is missing, stale, immature, or already pending');
       if (a.kind === 'fixed' ? a.endsAt !== null : a.endsAt === null || a.endsAt < now + 60_000 || a.endsAt > now + 30 * 86400_000) throw new Error('Auction duration must be 1 minute to 30 days; fixed listings have no end');
-      if (this.rows<MarketListingRecord>('listings').filter(l => l.seller === c.actor && ['open','reserved'].includes(l.status)).length >= 100) throw new Error('Limit: 100 active listings per seller');
+      const activeListings = this.db.prepare(`SELECT COUNT(*) AS n FROM listings
+        WHERE json_extract(data,'$.seller')=? AND status IN ('open','reserved')`).get(c.actor) as { n: number };
+      if (activeListings.n >= 100) throw new Error('Limit: 100 active listings per seller');
       if ((this.db.prepare('SELECT COUNT(*) AS n FROM listings').get() as { n: number }).n >= 10000) throw new Error('Marketplace journal limit reached; operator archival required');
       if (this.db.prepare("SELECT 1 FROM listings WHERE card=? AND status IN ('open','reserved')").get(a.cardHash)) throw new Error('Card already listed');
       const mint = s.nft(a.cardHash)!.block;

server/testnet/http.ts

@@ -251,11 +251,17 @@ server.setTimeout(30000,socket=>socket.destroy());
 // Advance simulated payout/expiry state even when no browser is open. No real funds are sent.
 const marketTimer=setInterval(()=>{try {market.browse();}catch{console.error('Marketplace reconciliation failed; operator review required.');}},15000);
 marketTimer.unref();
+// When durable reconciliation is behind, advance bounded keyset pages without requiring a browser request.
+const marketCatchupTimer=setInterval(()=>{
+  if(!market.hasReconciliationBacklog())return;
+  try {market.browse();}catch{console.error('Marketplace canonical catch-up failed; operator review required.');}
+},250);
+marketCatchupTimer.unref();
 let stopping=false;
 for(const signal of ['SIGTERM','SIGINT']as const)process.once(signal,()=>{
-  if(stopping)return;stopping=true;peers.close();operations.close();clearInterval(marketTimer);
+  if(stopping)return;stopping=true;peers.close();operations.close();clearInterval(marketTimer);clearInterval(marketCatchupTimer);
   const observerDrain=custodyObserver?.close()??Promise.resolve();
   const timeout=setTimeout(()=>process.exit(1),25000);timeout.unref();
   server.close(async()=>{await observerDrain;await custody?.close();market.close();ledger.close();engine.close();await lock.close();await unlink(lockPath).catch(()=>{});clearTimeout(timeout);process.exit(0);});
 });
-server.on('error',async()=>{clearInterval(marketTimer);operations.close();peers.close();await custodyObserver?.close();await custody?.close();market.close();ledger.close();engine.close();await lock.close();await unlink(lockPath).catch(()=>{});process.exit(1);});
+server.on('error',async()=>{clearInterval(marketTimer);clearInterval(marketCatchupTimer);operations.close();peers.close();await custodyObserver?.close();await custody?.close();market.close();ledger.close();engine.close();await lock.close();await unlink(lockPath).catch(()=>{});process.exit(1);});

tests/market.test.ts

@@ -133,20 +133,84 @@ test('indexed market uses bounded durable event catch-up and never requests full
   await f.restart(); assert.equal(f.market.diagnostics().canonicalEventCursor, 130);
   f.market.browse(); assert.equal(f.market.diagnostics().canonicalEventCursor, 130);
   const anchor = f.events[129].eventHash;
-  f.events[129].eventHash = 'f'.repeat(64); f.advance(1001);
+  f.events[129].eventHash = 'f'.repeat(64); f.advance(16_000);
   assert.throws(() => f.market.browse(), /history changed/);
   f.events[129].eventHash = anchor;
   f.events.length = 0;
   assert.throws(() => f.market.browse(), /cursor ahead/);
 });
 
-test('disconnect cursor and payout hold commit together and survive a crash/restart', async t => {
+test('market reconciliation keyset progress is bounded, durable and resumes after restart', async t => {
+  const f = await fixture(t, true), db = new DatabaseSync(join(f.dir, 'testnet', 'market.sqlite'));
+  const ids = Array.from({ length: 260 }, (_, i) => `00000000-0000-4000-8000-${(i + 1).toString(16).padStart(12, '0')}`);
+  try {
+    const insertListing = db.prepare('INSERT INTO listings VALUES(?,?,?,?)');
+    const insertOrder = db.prepare('INSERT INTO orders VALUES(?,?)');
+    for (const [i, id] of ids.entries()) {
+      const listingId = `10000000-0000-4000-8000-${(i + 1).toString(16).padStart(12, '0')}`;
+      const cardHash = (i + 1).toString(16).padStart(64, '0');
+      const target = i === ids.length - 1;
+      insertListing.run(listingId, cardHash, target ? 'sold' : 'cancelled', JSON.stringify({ id: listingId, seller: alice.actor,
+        cardHash, sequence: 0, status: target ? 'sold' : 'cancelled', kind: 'fixed' }));
+      const order = target ? { id, termsVersion: 3, listingId, seller: alice.actor, buyer: bob.actor, cardHash, sequence: 0,
+        priceAtomicXmr: '100000000000', feeBps: 250, createdAt: f.now, payBy: f.now + 1000, deliverBy: f.now + 2000,
+        status: 'complete', payout: 'mock-paid', deliveryConfirmations: 2, requiredConfirmations: 2, transferId: '0'.repeat(64) }
+        : { id, listingId, seller: alice.actor, buyer: bob.actor, cardHash, sequence: 0, status: 'refunded', createdAt: f.now };
+      insertOrder.run(id, JSON.stringify(order));
+    }
+  } finally { db.close(); }
+  const disconnectedTx = transferId({ network: NETWORK, chainId: CHAIN_ID,
+    cardHash: (ids.length).toString(16).padStart(64, '0'), from: alice.actor, to: bob.actor, sequence: 0, publicKey: '', signature: '' });
+  f.events.push({ sequence: 1, type: 'disconnect', height: 1, hash: 'd'.repeat(64), previousHash: GENESIS_HASH,
+    transactionIds: [disconnectedTx], eventHash: 'e'.repeat(64) });
+
+  // First browse starts reconciliation and visits only the first fixed-size keyset page.
+  f.market.browse();
+  const progress = () => {
+    const conn = new DatabaseSync(join(f.dir, 'testnet', 'market.sqlite'));
+    try { return JSON.parse((conn.prepare('SELECT value FROM market_meta WHERE key=?').get('canonical_reconciliation_job') as { value: string }).value); }
+    finally { conn.close(); }
+  };
+  assert.equal(progress().stage, 'orders'); assert.equal(progress().afterId, ids[127]);
+  assert.equal(f.market.hasReconciliationBacklog(), true);
+  await assert.rejects(f.list(), /catching up/); // second 128-order batch; no new listing is admitted.
+  assert.equal(progress().stage, 'orders'); assert.equal(progress().afterId, ids[255]);
+  // The ledger advances while this durable multi-batch job is in flight.
+  f.events.push({ sequence: 2, type: 'connect', height: 2, hash: 'f'.repeat(64), previousHash: 'd'.repeat(64),
+    transactionIds: [], eventHash: '2'.repeat(64) });
+  f.market.browse(); // Tip/revision changed: rewind and bind the updated event page.
+  assert.equal(progress().eventPage.cursor, 2);
+  assert.equal(progress().afterId, ids[127]);
+  f.market.browse(); // Second order batch of the refreshed scan.
+  f.market.browse(); // Remaining four orders, then at most one listing page.
+  assert.equal(f.market.diagnostics().canonicalEventCursor, 0);
+  assert.equal(f.market.hasReconciliationBacklog(), true, 'new events appended during job must stay behind the work fence');
+  const reviewDb = new DatabaseSync(join(f.dir, 'testnet', 'market.sqlite'));
+  try {
+    const row = reviewDb.prepare('SELECT data FROM orders WHERE id=?').get(ids[259]) as { data: string };
+    assert.equal(JSON.parse(row.data).status, 'review'); // Disconnect reached the tail batch before the cursor advances.
+  } finally { reviewDb.close(); }
+  assert.equal(f.market.diagnostics().canonicalEventCursor, 0);
+
+  // Simulate process death after committed batch progress, then resume using the SQLite job cursor.
+  await f.restart(); assert.equal(f.market.diagnostics().eventBacklog, true);
+  for (let i = 0; i < 12 && f.market.hasReconciliationBacklog(); i++) f.market.browse();
+  assert.equal(f.market.hasReconciliationBacklog(), false);
+  const conn = new DatabaseSync(join(f.dir, 'testnet', 'market.sqlite'));
+  try { assert.equal(conn.prepare('SELECT 1 FROM market_meta WHERE key=?').get('canonical_reconciliation_job'), undefined); }
+  finally { conn.close(); }
+  assert.equal(f.market.diagnostics().canonicalEventCursor, 2);
+});
+
+test('bounded disconnect review batch commits before checkpoint advance and resumes after restart', async t => {
   const f = await fixture(t, true), listing = await f.list(), order = await f.buy(listing);
   await f.run(bob, { type: 'mock-pay', orderId: order.id });
   const delivery = f.delivery(order); f.confirm(delivery);
   f.chain.state.blocks.push({ ...f.chain.state.blocks[0], height: f.chain.state.blocks.length + 1 });
   f.events.push({ sequence: 1, type: 'connect', height: 3, hash: 'd'.repeat(64), previousHash: 'e'.repeat(64), transactionIds: [transferId(delivery)], eventHash: '1'.repeat(64) });
-  assert.equal((await f.orders())[0].status, 'complete');
+  const initiallyCompleted = (await f.orders())[0];
+  assert.equal(initiallyCompleted.status, 'complete', JSON.stringify({ order: initiallyCompleted,
+    job: new DatabaseSync(join(f.dir, 'testnet', 'market.sqlite')).prepare('SELECT value FROM market_meta WHERE key=?').get('canonical_reconciliation_job') }));
   // The same delivery may already be canonical again; the intermediate disconnect must not be lost.
   f.events.push({ sequence: 2, type: 'disconnect', height: 3, hash: 'd'.repeat(64), previousHash: 'e'.repeat(64), transactionIds: [transferId(delivery)], eventHash: '2'.repeat(64) });
   const db = new DatabaseSync(join(f.dir, 'testnet', 'market.sqlite'));
@@ -154,7 +218,9 @@ test('disconnect cursor and payout hold commit together and survive a crash/rest
     db.exec("CREATE TRIGGER fail_cursor BEFORE UPDATE ON market_meta WHEN NEW.key='canonical_event_cursor' BEGIN SELECT RAISE(ABORT,'injected cursor failure'); END");
     assert.throws(() => f.market.browse(), /injected cursor failure/);
     const persisted = db.prepare('SELECT data FROM orders WHERE id=?').get(order.id) as { data: string };
-    assert.equal(JSON.parse(persisted.data).status, 'complete');
+    assert.equal(JSON.parse(persisted.data).status, 'review');
+    const job = JSON.parse((db.prepare("SELECT value FROM market_meta WHERE key='canonical_reconciliation_job'").get() as { value: string }).value);
+    assert.equal(job.stage, 'listings'); // Order effects and work cursor committed atomically; event anchor still has not advanced.
     assert.equal(f.market.diagnostics().canonicalEventCursor, 1);
     db.exec('DROP TRIGGER fail_cursor');
   } finally { db.close(); }
@@ -165,6 +231,33 @@ test('disconnect cursor and payout hold commit together and survive a crash/rest
   await f.restart(); assert.equal((await f.orders())[0].status, 'review');
 });
 
+test('market reconciliation batch failure rolls back order effects and resumes from durable keyset', async t => {
+  const f = await fixture(t, true), listing = await f.list(), order = await f.buy(listing);
+  await f.run(bob, { type: 'mock-pay', orderId: order.id });
+  const delivery = f.delivery(order); f.confirm(delivery);
+  f.chain.state.blocks.push({ ...f.chain.state.blocks[0], height: 4, transactions: [] });
+  f.events.push({ sequence: 1, type: 'connect', height: 3, hash: 'd'.repeat(64), previousHash: 'e'.repeat(64),
+    transactionIds: [transferId(delivery)], eventHash: '1'.repeat(64) });
+  assert.equal((await f.orders())[0].status, 'complete');
+  f.events.push({ sequence: 2, type: 'disconnect', height: 3, hash: 'd'.repeat(64), previousHash: 'e'.repeat(64),
+    transactionIds: [transferId(delivery)], eventHash: '2'.repeat(64) });
+  const db = new DatabaseSync(join(f.dir, 'testnet', 'market.sqlite'));
+  try {
+    db.exec("CREATE TRIGGER fail_review_batch BEFORE UPDATE ON orders WHEN json_extract(NEW.data,'$.status')='review' BEGIN SELECT RAISE(ABORT,'injected reconciliation batch failure'); END");
+    assert.throws(() => f.market.browse(), /injected reconciliation batch failure/);
+    const row = db.prepare('SELECT data FROM orders WHERE id=?').get(order.id) as { data: string };
+    assert.equal(JSON.parse(row.data).status, 'complete');
+    const job = JSON.parse((db.prepare("SELECT value FROM market_meta WHERE key='canonical_reconciliation_job'").get() as { value: string }).value);
+    assert.equal(job.stage, 'orders'); assert.equal(job.afterId, null);
+    assert.equal(f.market.diagnostics().canonicalEventCursor, 1);
+    db.exec('DROP TRIGGER fail_review_batch');
+  } finally { db.close(); }
+  await f.restart();
+  assert.equal((await f.orders())[0].status, 'review');
+  assert.equal(f.market.diagnostics().canonicalEventCursor, 2);
+  assert.equal(f.market.hasReconciliationBacklog(), false);
+});
+
 test('USD and mock XMR accounting use bounded integer amounts and exact fee splitting', () => {
   assert.equal(parseUsd('15.01'), 1501); assert.equal(parseUsd('0.01'), 1);
   for (const invalid of ['0', '-1', '1e3', '1.001', ' 1', 'NaN', '1000001', '01.00']) assert.throws(() => parseUsd(invalid));