diff --git a/docs/canonical-indexes.md b/docs/canonical-indexes.md index 77fc000..84beca7 100755 --- a/docs/canonical-indexes.md +++ b/docs/canonical-indexes.md @@ -10,13 +10,13 @@ Status: implemented local infrastructure, **not full M2 acceptance or production - 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. - Order and listing review now uses stable-ID keyset batches capped at 32 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. Requests advance one bounded slice, and an unref'd HTTP-lifecycle pump drains remaining work through `setImmediate` turns without depending on page loads. This is cooperative yielding, **not a worker thread**: one slice still runs synchronous SQLite work on the server event loop. There is no claim of instant catch-up, a throughput SLO, or horizontal-worker safety. +- While any job/event backlog exists, ordinary market mutations and time-dependent payout/expiry decisions stop; diagnostic/read access remains available. Requests advance one bounded slice, and an unref'd HTTP-lifecycle pump drains remaining work without depending on page loads. The production HTTP process enables a single persistent Node worker thread to own atomic batch writes and final checkpoint commits; the main thread retains the chain queries, per-row reconciliation decisions and event/revision fences. This reduces SQLite write/commit work on the request loop but is **not complete CPU isolation**. Worker failure leaves the durable job/cursor intact, fails closed, and requires retry/restart; there is no horizontal-worker safety claim. - 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. - [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 **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: +The final Docker test-target build passed TypeScript/Vite compilation. All **175 offline tests passed, zero failures/skips**, including native RandomX multi-process HTTP/peer/restart and backup rehearsal. The focused market suite passed **24 tests**; the disposable PostgreSQL suite passed **31 tests** in the preceding integration run. 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; @@ -27,6 +27,7 @@ The final Docker test-target build passed TypeScript/Vite compilation. All **174 - 32-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; +- worker-owned batch/finalize transaction with injected failure, mutation fence and restart continuation; - numeric cursor reuse with a different historical hash rejected; - real HTTP NFT results and stale-tip deck pagination. @@ -45,10 +46,10 @@ Observed in the development Docker environment: These numbers are a reproducible **baseline**, not acceptance SLOs: host CPU/storage and production concurrency were not selected, outputs include synthetic article payloads, warm caches apply, one transaction amortizes insertion fsync, and real validation/startup/HTTP/market workloads are excluded. They do not establish a supported production envelope. -The synthetic market-journal harness [benchmark](../scripts/benchmark-market-reconciliation.ts), available as `npm run benchmark:market -- 10000 [terminal|confirmed-delivery]`, measures two bounded reconciliation paths in its own temporary SQLite directory. With 32-row slices, the development Docker host processed 10,000 terminal orders plus 10,000 listings in 2,549 ms (about 7,845 records/s), with pass-latency p50/p95/max of 3.44/6.48/42.13 ms, sampled event-loop-delay p95/max of 14.97/48.99 ms at 10ms resolution, and an 11,927,552-byte SQLite file. The confirmed-delivery fixture transitioned 10,000 paid orders and reserved listings to complete/sold in 4,667 ms (about 4,286 records/s), with pass-latency p50/p95/max of 6.88/14.30/62.12 ms, sampled event-loop-delay p95/max of 20.63/62.23 ms, and a 12,832,768-byte SQLite file. The 32-row size increases pass count (625 vs. 157 at 128 rows) and reduces typical delay, but these runs still show long-tail event-loop stalls; a smaller batch is not equivalent to worker isolation. The canonical transaction IDs are in-memory adapter fixtures, not validated Ledger evidence. These are not target-hardware throughput/SLO claims and exclude concurrent HTTP traffic, other server work, PostgreSQL, actual chain validation and adversarial workloads. +The synthetic market-journal harness [benchmark](../scripts/benchmark-market-reconciliation.ts), available as `npm run benchmark:market -- 10000 [terminal|confirmed-delivery] [worker]`, measures terminal and confirmed-delivery reconciliation in its own temporary SQLite directory. With synchronous 32-row commits, 10,000 terminal orders plus 10,000 listings took 2,549 ms (about 7,845 records/s; sampled event-loop-delay p95/max 14.97/48.99 ms). The confirmed-delivery path took 4,667 ms (about 4,286 records/s; 20.63/62.23 ms). With worker-owned commits, the terminal fixture took 2,782 ms (about 7,190 records/s; sampled event-loop-delay p95/max 11.17/16.11 ms), and confirmed delivery took 5,762 ms (about 3,471 records/s; 13.39/24.25 ms). Worker IPC/transaction cost increases elapsed time, while sampled maximum main-loop delay fell in these runs; measurements are noisy and not an SLO. Maximum sampled pass duration with the worker was 25.15 ms for confirmed delivery, since row evaluation and chain lookups remain on the main thread. Canonical transaction IDs are in-memory adapter fixtures, not validated Ledger evidence. These runs are not target-hardware capacity claims and exclude concurrent HTTP traffic, PostgreSQL, actual chain validation and adversarial workloads. ## Remaining gates -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. +M2 still needs target-hardware/data/SLO decisions, populated long-journal/large-event benchmarks, HTTP load tests during worker commits, production financial database/schema and measured recovery/retention/storage growth. The worker moves writes/final commits off-loop, but row processing/chain reads remain on-loop; queue isolation, multi-process fencing and a demonstrated production throughput envelope remain open. 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. diff --git a/docs/production-plan.md b/docs/production-plan.md index da94781..824c00b 100755 --- a/docs/production-plan.md +++ b/docs/production-plan.md @@ -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 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 advances in durable, stable-ID keyset batches of at most 32 records per table, resumes after restart, re-fences when the chain tip/revision changes, and drains through a yielding HTTP-lifecycle pump 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. Synthetic 10,000-order/listing runs processed terminal records in 2,549 ms (event-loop-delay p95/max 14.97/48.99 ms) and paid orders with confirmed-delivery adapter fixtures in 4,667 ms (20.63/62.23 ms). Slicing reduces typical per-pass delay but increases total work and still has long-tail stalls; it is not worker-thread isolation or target-hardware evidence. Queue isolation, target-scale/concurrent throughput and latency, production financial storage/recovery, retained-DAG capacity and operator dataset/SLO 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 advances in durable, stable-ID keyset batches of at most 32 records per table, resumes after restart, and re-fences when the chain tip/revision changes. The HTTP process now uses a single worker thread for atomic batch writes and final cursor commits; main-thread chain reads and row evaluation remain. Final TypeScript/Vite build and **175 offline tests passed**; focused market tests passed 24/24; the disposable PostgreSQL suite passed 31/31 in the preceding integration run. Synthetic 10,000-order/listing tests showed worker-backed confirmed-delivery elapsed time 5,762 ms and sampled event-loop-delay p95/max 13.39/24.25 ms, versus 4,667 ms and 20.63/62.23 ms with main-thread commits. This suggests a responsiveness/throughput trade-off on this development host, not an SLO. Remaining: real concurrent HTTP load, further main-thread CPU isolation, multi-process fencing, target hardware/data, production financial storage/recovery, retained-DAG capacity and operator SLO decisions. **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. **Local bounded implementation delivered:** rollback-safe canonical event pages, durable 32-row order/listing keyset jobs, tip/revision re-fencing, restart continuation, mutation fencing, and a cooperative background pump that yields between batches; regression tests cover appends/reorgs and injected SQLite failures. **Still open:** true worker-thread isolation (the pump still runs synchronous SQLite on the main event loop), coordination of a single worker across processes, and lag/throughput/request-latency plus overload/recovery measurements at declared order/event volumes. +- Incremental market reconciliation from durable canonical/reorg events. **Local bounded implementation delivered:** rollback-safe canonical event pages, durable 32-row order/listing keyset jobs, tip/revision re-fencing, restart continuation, mutation fencing, and a single worker thread owning atomic batch writes/finalization. Regression tests cover reorgs, injected worker commit failures and restart recovery. **Still open:** move more row evaluation/chain lookup work off the request event loop, measure concurrent HTTP latency and queue lag, coordinate/fence processes, and establish throughput/overload/recovery limits at declared order/event volumes. - 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. diff --git a/scripts/benchmark-market-reconciliation.ts b/scripts/benchmark-market-reconciliation.ts index d3b7000..66e847b 100755 --- a/scripts/benchmark-market-reconciliation.ts +++ b/scripts/benchmark-market-reconciliation.ts @@ -15,6 +15,7 @@ const text = process.argv[2] ?? '10000'; if (!/^[1-9]\d{0,5}$/.test(text) || Number(text) > 100000) throw new Error('Fixture size must be 1..100000 orders'); const count = Number(text), scenario = process.argv[3] ?? 'terminal'; if (!['terminal', 'confirmed-delivery'].includes(scenario)) throw new Error('Scenario must be terminal or confirmed-delivery'); +const workerEnabled = process.argv[4] === 'worker'; const directory = await mkdtemp(join(tmpdir(), 'wikideck-market-reconciliation-')); const streamId = randomUUID(), tip = 'a'.repeat(64); const deliveryRows = new Map(); @@ -33,7 +34,7 @@ try { events: [], cursor: after, head: 0, caughtUp: true }), submitTransfer: async () => { throw new Error('Benchmark never submits transfers'); }, }; - market = await Market.open(directory, chain, { feeBps: 250, confirmations: 2 }); + market = await Market.open(directory, chain, { feeBps: 250, confirmations: 2 }, Date.now, { worker: workerEnabled }); const db = new DatabaseSync(join(directory, NETWORK.includes('mainnet') ? 'mainnet' : 'testnet', 'market.sqlite')); try { const insertListing = db.prepare('INSERT INTO listings(id,card,status,data) VALUES(?,?,?,?)'); @@ -62,25 +63,30 @@ try { } finally { db.close(); } const latencies: number[] = []; - const maxPasses = Math.ceil(count / MARKET_RECONCILIATION_BATCH_SIZE) * 2 + 4; + const deadline = Date.now() + Math.max(30_000, count * 20); const eventLoopDelay = monitorEventLoopDelay({ resolution: 10 }); eventLoopDelay.enable(); const start = performance.now(); + let started = false; try { - for (let pass = 0; pass < maxPasses && (pass === 0 || market.hasReconciliationBacklog()); pass++) { - const before = performance.now(); - market.browse(); - latencies.push(performance.now() - before); + while (Date.now() < deadline && (!started || market.hasReconciliationBacklog() || market.isReconciliationWorkerBusy())) { + if (!workerEnabled || !market.isReconciliationWorkerBusy()) { + const before = performance.now(); + market.browse(); + latencies.push(performance.now() - before); + started = true; + } await yieldToEventLoop(); // Permit the delay monitor to observe synchronous batch stalls. + if (workerEnabled && market.isReconciliationWorkerBusy()) await new Promise(resolve => setTimeout(resolve, 1)); } } finally { eventLoopDelay.disable(); } const elapsedMs = performance.now() - start; - if (market.hasReconciliationBacklog()) throw new Error(`Reconciliation did not finish within ${maxPasses} bounded passes`); - market.close(); market = undefined; + if (!started || market.hasReconciliationBacklog() || market.isReconciliationWorkerBusy()) throw new Error('Reconciliation did not finish before benchmark deadline'); + await market.close(); market = undefined; const bytes = (await stat(join(directory, NETWORK.includes('mainnet') ? 'mainnet' : 'testnet', 'market.sqlite'))).size; latencies.sort((a, b) => a - b); const percentile = (p: number) => latencies[Math.min(latencies.length - 1, Math.ceil(latencies.length * p) - 1)]; - console.log(JSON.stringify({ fixture: `synthetic-${scenario}-orders-and-listings`, orders: count, listings: count, + console.log(JSON.stringify({ fixture: `synthetic-${scenario}-orders-and-listings`, workerEnabled, orders: count, listings: count, batchSizePerTable: MARKET_RECONCILIATION_BATCH_SIZE, reconciliationPasses: latencies.length, elapsedMs, rowsPerSecond: Math.round((count * 2 * 1000) / elapsedMs), passLatencyP50Ms: percentile(0.50), passLatencyP95Ms: percentile(0.95), passLatencyMaxMs: latencies.at(-1), sqliteBytes: bytes, diff --git a/scripts/worker-probe.ts b/scripts/worker-probe.ts new file mode 100755 index 0000000..4db9376 --- /dev/null +++ b/scripts/worker-probe.ts @@ -0,0 +1 @@ +// Temporary runtime probe removed after verification. diff --git a/server/market/reconciliation-commit-worker.js b/server/market/reconciliation-commit-worker.js new file mode 100755 index 0000000..7c0e610 --- /dev/null +++ b/server/market/reconciliation-commit-worker.js @@ -0,0 +1,52 @@ +import { DatabaseSync } from 'node:sqlite'; +import { parentPort, workerData } from 'node:worker_threads'; +import { isDeepStrictEqual } from 'node:util'; + +const db = new DatabaseSync(workerData.marketFile); +db.exec('PRAGMA busy_timeout=5000; PRAGMA synchronous=FULL;'); +const port = parentPort; +if (!port) throw new Error('Reconciliation commit worker requires a parent port'); + +function atomic(fn) { + db.exec('BEGIN IMMEDIATE'); + try { const result = fn(); db.exec('COMMIT'); return result; } + catch (error) { db.exec('ROLLBACK'); throw error; } +} +function assertCurrentJob(expected) { + const row = db.prepare("SELECT value FROM market_meta WHERE key='canonical_reconciliation_job'").get(); + if (!row || !isDeepStrictEqual(JSON.parse(row.value), expected)) { + throw new Error('Reconciliation job changed before worker commit'); + } +} + +port.on('message', request => { + try { + if (request.operation === 'persist-batch') { + atomic(() => { + assertCurrentJob(request.expectedJob); + const order = db.prepare(`INSERT INTO orders(id,data) VALUES(?,?) + ON CONFLICT(id) DO UPDATE SET data=excluded.data WHERE orders.data<>excluded.data`); + for (const row of request.orders) order.run(row.id, row.data); + const listing = db.prepare(`INSERT INTO listings(id,card,status,data) VALUES(?,?,?,?) + ON CONFLICT(id) DO UPDATE SET status=excluded.status,data=excluded.data WHERE listings.data<>excluded.data`); + for (const row of request.listings) listing.run(row.id, row.card, row.status, row.data); + db.prepare("UPDATE market_meta SET value=? WHERE key='canonical_reconciliation_job'") + .run(JSON.stringify(request.nextJob)); + }); + } else { + atomic(() => { + assertCurrentJob(request.expectedJob); + db.prepare('DELETE FROM receipts WHERE expires < ?').run(request.observedAt - 86400_000); + if (request.cursor) { + db.prepare(`INSERT INTO market_meta(key,value) VALUES('canonical_event_cursor',?) + ON CONFLICT(key) DO UPDATE SET value=excluded.value`).run(request.cursor); + } + db.prepare("DELETE FROM market_meta WHERE key='canonical_reconciliation_job'").run(); + }); + } + port.postMessage({ id: request.id, ok: true }); + } catch (error) { + port.postMessage({ id: request.id, ok: false, error: error instanceof Error ? error.message : 'Worker batch failed' }); + } +}); +port.on('close', () => db.close()); diff --git a/server/market/reconciliation-commit-worker.ts b/server/market/reconciliation-commit-worker.ts new file mode 100755 index 0000000..c55854e --- /dev/null +++ b/server/market/reconciliation-commit-worker.ts @@ -0,0 +1 @@ +// The runtime worker is reconciliation-commit-worker.js so it is independent of TS loaders. diff --git a/server/market/store.ts b/server/market/store.ts index 0f50670..ffc2de0 100755 --- a/server/market/store.ts +++ b/server/market/store.ts @@ -2,6 +2,7 @@ import { DatabaseSync } from 'node:sqlite'; import { randomUUID } from 'node:crypto'; import { mkdir, chmod } from 'node:fs/promises'; import { join } from 'node:path'; +import { Worker } from 'node:worker_threads'; import { NETWORK, CONSENSUS, nftTransferMessage, testnetOwnerOf, transactionCardIds, type TestnetSnapshot, type NftTransfer } from '../../shared/testnet.ts'; import { ADDITIONAL_MARKET_FEE, marketCommandSchema, marketMessage, type MarketCommand, type MarketListing, @@ -48,7 +49,7 @@ const canonicalEventSchema = z.object({ sequence: z.number().int().positive().sa 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']), +const reconciliationJobSchema = z.object({ version: z.literal(1), stage: z.enum(['orders', 'listings', 'finalize']), 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(); @@ -65,13 +66,39 @@ export class Market { private eventCheckpoint: z.infer | undefined; private eventBacklog = false; private reconciliationJob: z.infer | undefined; + private readonly reconciliationWorker?: Worker; + private workerRequestId = 0; + private readonly workerRequests = new Map(); + private reconciliationWorkerBusy = false; + private reconciliationWorkerError: string | undefined; + private stagedReconciliationWrites: { orders: Map; + listings: Map } | undefined; private readonly db: DatabaseSync; private readonly chain: Chain; readonly config: Config; readonly id: string; private readonly now: () => number; - private constructor(db: DatabaseSync, chain: Chain, config: Config, id: string, now: () => number) { + private constructor(db: DatabaseSync, chain: Chain, config: Config, id: string, now: () => number, worker?: Worker) { this.db = db; this.chain = chain; this.config = config; this.id = id; this.now = now; + this.reconciliationWorker = worker; + worker?.on('message', (message: { id: number; ok: boolean; error?: string }) => { + const pending = this.workerRequests.get(message.id); + if (!pending) return; + this.workerRequests.delete(message.id); + if (message.ok) pending.resolve(); else pending.reject(new Error(message.error ?? 'Reconciliation worker failed')); + }); + worker?.on('error', error => { + this.reconciliationWorkerError = 'Reconciliation worker failed'; + for (const pending of this.workerRequests.values()) pending.reject(error); + this.workerRequests.clear(); + }); + worker?.on('exit', code => { + if (!this.workerRequests.size) return; + const error = new Error(`Reconciliation worker exited with code ${code}`); + this.reconciliationWorkerError = 'Reconciliation worker stopped unexpectedly'; + for (const pending of this.workerRequests.values()) pending.reject(error); + this.workerRequests.clear(); + }); const cursor = db.prepare("SELECT value FROM market_meta WHERE key='canonical_event_cursor'").get() as { value: string } | undefined; if (cursor) { this.eventCheckpoint = cursorSchema.parse(JSON.parse(cursor.value)); @@ -82,12 +109,14 @@ export class Market { else this.eventBacklog = !!chain.canonicalHead && this.eventCursor < chain.canonicalHead(); } - static async open(directory: string, chain: Chain, config = marketConfig(), now = Date.now): Promise { + static async open(directory: string, chain: Chain, config = marketConfig(), now = Date.now, + reconciliation: { worker?: boolean } = {}): Promise { await validateMarketPolicy(config); const folder = join(directory, NETWORK.includes('mainnet') ? 'mainnet' : 'testnet'); await mkdir(folder, { recursive: true, mode: 0o700 }); const file = join(folder, 'market.sqlite'); const db = new DatabaseSync(file); + let worker: Worker | undefined; try { db.exec(`PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL; PRAGMA busy_timeout=5000; CREATE TABLE IF NOT EXISTS market_meta(key TEXT PRIMARY KEY, value TEXT NOT NULL); @@ -111,14 +140,22 @@ export class Market { db.prepare('INSERT INTO market_meta VALUES(?,?)').run('identity', JSON.stringify({ version: 1, network: NETWORK, chainId: CHAIN_ID, id })); } if (process.platform !== 'win32') await chmod(file, 0o600); - return new Market(db, chain, config, id, now); - } catch (error) { db.close(); throw error; } + worker = reconciliation.worker ? new Worker(new URL('./reconciliation-commit-worker.js', import.meta.url), + { workerData: { marketFile: file }, execArgv: [] }) : undefined; + worker?.unref(); + return new Market(db, chain, config, id, now, worker); + } catch (error) { if (worker) await worker.terminate(); db.close(); throw error; } } - close() { this.db.close(); } + async close(): Promise { + if (this.reconciliationWorker) await this.reconciliationWorker.terminate(); + this.db.close(); + } + isReconciliationWorkerBusy(): boolean { return this.reconciliationWorkerBusy; } diagnostics() { const states=this.db.prepare("SELECT json_extract(data,'$.status') AS status,COUNT(*) AS count FROM orders GROUP BY json_extract(data,'$.status')").all(); return { queued:this.queued,lastReconciledAt:this.reconciled?.at??null,orderStates:states, - canonicalEventCursor:this.eventCursor,canonicalEventHead:this.chain.canonicalHead?.()??null,eventBacklog:this.eventBacklog }; + canonicalEventCursor:this.eventCursor,canonicalEventHead:this.chain.canonicalHead?.()??null,eventBacklog:this.eventBacklog, + reconciliationWorkerBusy:this.reconciliationWorkerBusy,reconciliationWorkerFailed:!!this.reconciliationWorkerError }; } hasReconciliationBacklog(): boolean { return this.eventBacklog || !!this.reconciliationJob; } info(): MarketInfo { @@ -142,10 +179,18 @@ export class Market { return JSON.parse(row.data) as MarketOrderRecord; } private saveListing(l: MarketListingRecord) { + if (this.stagedReconciliationWrites) { + this.stagedReconciliationWrites.listings.set(l.id, l); + return; + } this.db.prepare('INSERT INTO listings VALUES(?,?,?,?) ON CONFLICT(id) DO UPDATE SET status=excluded.status,data=excluded.data WHERE listings.data<>excluded.data') .run(l.id, l.cardHash, l.status, JSON.stringify(l)); } private saveOrder(o: MarketOrderRecord) { + if (this.stagedReconciliationWrites) { + this.stagedReconciliationWrites.orders.set(o.id, o); + return; + } this.db.prepare('INSERT INTO orders VALUES(?,?) ON CONFLICT(id) DO UPDATE SET data=excluded.data WHERE orders.data<>excluded.data').run(o.id, JSON.stringify(o)); } private atomic(fn: () => T): T { @@ -179,8 +224,77 @@ export class Market { return transactionId({ network: NETWORK, chainId: CHAIN_ID, cardHash: order.cardHash, from: order.seller, to: order.buyer, sequence: order.sequence, publicKey: '', signature: '' }); } + private workerRequest(request: Record): Promise { + const worker = this.reconciliationWorker; + if (!worker) return Promise.reject(new Error('Reconciliation worker is unavailable')); + const id = ++this.workerRequestId; + return new Promise((resolve, reject) => { + this.workerRequests.set(id, { resolve, reject }); + try { worker.postMessage({ ...request, id }); } + catch (error) { this.workerRequests.delete(id); reject(error instanceof Error ? error : new Error('Worker request failed')); } + }); + } + private scheduleWorkerContinuation(): void { + setImmediate(() => { if (!this.reconciliationWorkerBusy && this.hasReconciliationBacklog()) this.reconcile(); }).unref(); + } + private persistWorkerBatch(job: z.infer, nextJob: z.infer, + writes: { orders: Map; listings: Map }): void { + if (this.reconciliationWorkerBusy) return; + this.reconciliationWorkerBusy = true; + void this.workerRequest({ operation: 'persist-batch', expectedJob: job, nextJob, + orders: [...writes.orders.values()].map(order => ({ id: order.id, data: JSON.stringify(order) })), + listings: [...writes.listings.values()].map(listing => ({ id: listing.id, card: listing.cardHash, + status: listing.status, data: JSON.stringify(listing) })) }) + .then(() => { this.reconciliationJob = nextJob; this.reconciliationWorkerError = undefined; this.eventBacklog = true; }) + .catch(error => { this.reconciliationWorkerError = error instanceof Error ? error.message : 'Worker commit failed'; this.eventBacklog = true; }) + .finally(() => { + this.reconciliationWorkerBusy = false; + if (!this.reconciliationWorkerError) this.scheduleWorkerContinuation(); + }); + } + private finalizeWorkerJob(job: z.infer): void { + const summary = this.chain.summary?.(), eventPage = job.eventPage as CanonicalEvents | null; + const currentHead = this.chain.canonicalHead?.() ?? eventPage?.head ?? 0; + if ((summary && (job.tip !== summary.tip || job.revision !== summary.revision)) || + (eventPage && currentHead > eventPage.head)) { + 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 restarted = reconciliationJobSchema.parse({ version: 1, stage: 'orders', afterId: null, observedAt: this.now(), + tip: summary?.tip ?? null, revision: summary?.revision ?? null, eventPage: events ?? null }); + this.atomic(() => this.db.prepare('UPDATE market_meta SET value=? WHERE key=?') + .run(JSON.stringify(restarted), RECONCILIATION_JOB_KEY)); + this.reconciliationJob = restarted; this.eventBacklog = true; this.reconciliationWorkerError = undefined; + this.scheduleWorkerContinuation(); + return; + } + if (this.reconciliationWorkerBusy) return; + this.reconciliationWorkerBusy = true; + const cursor = eventPage ? JSON.stringify({ version: 1, streamId: eventPage.streamId, + cursor: eventPage.cursor, anchor: eventPage.anchor }) : null; + void this.workerRequest({ operation: 'finalize', expectedJob: job, cursor, observedAt: job.observedAt }) + .then(() => { + this.reconciliationJob = undefined; + if (eventPage) { + this.eventCursor = eventPage.cursor; + this.eventCheckpoint = { version: 1, streamId: eventPage.streamId, cursor: eventPage.cursor, anchor: eventPage.anchor }; + } + const current = this.chain.summary?.(); + const head = this.chain.canonicalHead?.() ?? eventPage?.head ?? 0; + this.eventBacklog = (!!eventPage && (!eventPage.caughtUp || head > eventPage.cursor)) || + (!!current && (current.tip !== job.tip || current.revision !== job.revision)); + if (current && !this.eventBacklog) this.reconciled = { ...current, at: this.now() }; + this.reconciliationWorkerError = undefined; + }) + .catch(error => { this.reconciliationWorkerError = error instanceof Error ? error.message : 'Worker finalize failed'; this.eventBacklog = true; }) + .finally(() => { + this.reconciliationWorkerBusy = false; + if (!this.reconciliationWorkerError && this.hasReconciliationBacklog()) this.scheduleWorkerContinuation(); + }); + } /** Reconcile one durable keyset batch from canonical transaction identity, never mempool acknowledgement. */ private reconcile() { + if (this.reconciliationWorkerBusy) return; const now = this.now(); const summary = this.chain.summary?.(); if (!this.reconciliationJob) { @@ -209,14 +323,38 @@ export class Market { this.reconciliationJob = job; this.eventBacklog = true; } + if (job.stage === 'finalize') { + if (this.reconciliationWorker) { this.finalizeWorkerJob(job); return; } + const eventPage = job.eventPage as CanonicalEvents | null; + this.atomic(() => { + 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 })); + this.db.prepare('DELETE FROM market_meta WHERE key=?').run(RECONCILIATION_JOB_KEY); + }); + this.reconciliationJob = undefined; + if (eventPage) { + this.eventCursor = eventPage.cursor; + this.eventCheckpoint = { version: 1, streamId: eventPage.streamId, cursor: eventPage.cursor, anchor: eventPage.anchor }; + } + const head = this.chain.canonicalHead?.() ?? eventPage?.head ?? 0; + this.eventBacklog = !!eventPage && (!eventPage.caughtUp || head > eventPage.cursor); + const current = this.chain.summary?.(); + if (current && !this.eventBacklog) this.reconciled = { ...current, at: now }; + return; + } 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 | undefined; let completed = false; + const workerWrites = this.reconciliationWorker ? { orders: new Map(), + listings: new Map() } : undefined; + this.stagedReconciliationWrites = workerWrites; - this.atomic(() => { + try { this.atomic(() => { if (job.stage === 'orders' && rows.length) { for (const row of rows) { const o = JSON.parse(row.data) as MarketOrderRecord; @@ -275,18 +413,28 @@ export class Market { else nextJob = { ...job, afterId: rows.at(-1)!.id }; } else { if (rows.length < MARKET_RECONCILIATION_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 })); + if (this.reconciliationWorker) nextJob = { ...job, stage: 'finalize', afterId: null }; + else { + 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 })); + } + this.db.prepare('DELETE FROM market_meta WHERE key=?').run(RECONCILIATION_JOB_KEY); + completed = true; } - this.db.prepare('DELETE FROM market_meta WHERE key=?').run(RECONCILIATION_JOB_KEY); - completed = true; } else nextJob = { ...job, afterId: rows.at(-1)!.id }; } - if (nextJob) this.db.prepare('UPDATE market_meta SET value=? WHERE key=?').run(JSON.stringify(nextJob), RECONCILIATION_JOB_KEY); - }); + if (nextJob && !this.reconciliationWorker) this.db.prepare('UPDATE market_meta SET value=? WHERE key=?').run(JSON.stringify(nextJob), RECONCILIATION_JOB_KEY); + }); } finally { this.stagedReconciliationWrites = undefined; } + + if (this.reconciliationWorker) { + if (!nextJob) throw new Error('Worker reconciliation batch did not produce a continuation'); + if (!workerWrites) throw new Error('Worker reconciliation writes were not staged'); + this.persistWorkerBatch(job, nextJob, workerWrites); + return; + } if (completed) { this.reconciliationJob = undefined; @@ -347,8 +495,9 @@ 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 (this.hasReconciliationBacklog() || a.type !== 'activity' || !summary || !last || summary.tip !== last.tip || summary.revision !== last.revision || - this.now() - last.at >= RECONCILIATION_FRESH_MS) this.reconcile(); + const stale = !summary || !last || summary.tip !== last.tip || summary.revision !== last.revision || + this.now() - last.at >= RECONCILIATION_FRESH_MS; + if (this.hasReconciliationBacklog() || stale || (!this.reconciliationWorker && a.type !== 'activity')) 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") diff --git a/server/testnet/http.ts b/server/testnet/http.ts index daf08c8..7a84128 100644 --- a/server/testnet/http.ts +++ b/server/testnet/http.ts @@ -63,7 +63,7 @@ try { peers = new PeerNetwork({ ledger, urls:(process.env.WIKIDECK_PEERS ?? ''). allowPrivate:process.env.WIKIDECK_ALLOW_PRIVATE_PEERS === '1' }); } catch(e) {ledger.close();engine.close();await lock.close();await unlink(lockPath);throw e;} let market: Market; -try { market = await Market.open(dataDir, ledger, paymentConfig); } +try { market = await Market.open(dataDir, ledger, paymentConfig, Date.now, { worker: true }); } catch(e) {peers.close();ledger.close();engine.close();await lock.close();await unlink(lockPath);throw e;} let custody: CustodyMarket | PostgresCustody | undefined; let custodyObserver: CanonicalObserver | undefined; @@ -80,7 +80,7 @@ try { custodyObserver = new CanonicalObserver(custody, ledger); if ((await custodyObserver.start()).state !== 'running') throw new Error('Canonical observer rehearsal failed to start; inspect custody diagnostics'); } -} catch(e) {await custodyObserver?.close();await custody?.close();market.close();peers.close();ledger.close();engine.close();await lock.close();await unlink(lockPath);throw e;} +} catch(e) {await custodyObserver?.close();await custody?.close();await market.close();peers.close();ledger.close();engine.close();await lock.close();await unlink(lockPath);throw e;} const app = express(); app.disable('x-powered-by'); if(proxyTrust.length)app.set('trust proxy',proxyTrust); @@ -258,10 +258,12 @@ let catchupImmediate:NodeJS.Immediate|undefined; // SQLite slice, then setImmediate yields so HTTP and peer work can run between passes. const advanceMarketCatchup=()=>{ catchupImmediate=undefined; - if(stopping||catchupPaused||!market.hasReconciliationBacklog())return; + if(stopping||catchupPaused||market.isReconciliationWorkerBusy()||!market.hasReconciliationBacklog())return; try {market.browse();} catch {catchupPaused=true;console.error('Marketplace canonical catch-up paused; operator review required.');return;} - if(market.hasReconciliationBacklog()){catchupImmediate=setImmediate(advanceMarketCatchup);catchupImmediate.unref();} + if(market.hasReconciliationBacklog()&&!market.isReconciliationWorkerBusy()){ + catchupImmediate=setImmediate(advanceMarketCatchup);catchupImmediate.unref(); + } }; const marketCatchupTimer=setInterval(()=>{ if(!stopping&&!catchupPaused&&!catchupImmediate&&market.hasReconciliationBacklog()){ @@ -274,6 +276,6 @@ for(const signal of ['SIGTERM','SIGINT']as const)process.once(signal,()=>{ if(catchupImmediate)clearImmediate(catchupImmediate); 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.close(async()=>{await observerDrain;await custody?.close();await market.close();ledger.close();engine.close();await lock.close();await unlink(lockPath).catch(()=>{});clearTimeout(timeout);process.exit(0);}); }); -server.on('error',async()=>{stopping=true;clearInterval(marketTimer);clearInterval(marketCatchupTimer);if(catchupImmediate)clearImmediate(catchupImmediate);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()=>{stopping=true;clearInterval(marketTimer);clearInterval(marketCatchupTimer);if(catchupImmediate)clearImmediate(catchupImmediate);operations.close();peers.close();await custodyObserver?.close();await custody?.close();await market.close();ledger.close();engine.close();await lock.close();await unlink(lockPath).catch(()=>{});process.exit(1);}); diff --git a/tests/api.test.ts b/tests/api.test.ts index 4db1902..fc7f8d1 100644 --- a/tests/api.test.ts +++ b/tests/api.test.ts @@ -127,12 +127,20 @@ test('native RandomX HTTP: three blocks, mined transfer, peer gossip/sync and du async function request(base: string, path: string, expected = 200, body?: unknown, headers: Record = {}): Promise { - const response = await fetch(`${base}${path}`, { - method: body === undefined ? 'GET' : 'POST', redirect: 'error', - signal: AbortSignal.any([t.signal, AbortSignal.timeout(30_000)]), - headers: { ...(body === undefined ? {} : { 'Content-Type': 'application/json' }), ...headers }, - body: body === undefined ? undefined : JSON.stringify(body), - }); + const until = Date.now() + 20_000; + let response: Response; + do { + response = await fetch(`${base}${path}`, { + method: body === undefined ? 'GET' : 'POST', redirect: 'error', + signal: AbortSignal.any([t.signal, AbortSignal.timeout(30_000)]), + headers: { ...(body === undefined ? {} : { 'Content-Type': 'application/json' }), ...headers }, + body: body === undefined ? undefined : JSON.stringify(body), + }); + if (path !== '/api/market/command' || response.status !== 503 || Date.now() >= until || + !(await response.clone().text()).includes('reconciliation is catching up')) break; + const retryAfter = Number(response.headers.get('Retry-After')); + await delay(Number.isFinite(retryAfter) && retryAfter > 0 ? retryAfter * 1000 : 25, undefined, { signal: t.signal }); + } while (Date.now() < until); if (response.status !== expected) { assert.fail(`${path}: expected ${expected}, got ${response.status}: ${await response.text()}`); } @@ -360,7 +368,9 @@ test('native RandomX HTTP: three blocks, mined transfer, peer gossip/sync and du assert.deepEqual(canonical.blocks, [block1, block2, block3]); assert.deepEqual(canonical.transfers, [transfer]); assert.deepEqual(canonical.pendingTransfers, []); - const settling = await marketCommand(first.base, market.info, bob, { type: 'activity' }); + const settling = await waitFor('Worker market delivery reconciliation', + () => marketCommand(first.base, market.info, bob, { type: 'activity' }), + activity => activity.orders[0]?.deliveryConfirmations === 1); assert.equal(settling.orders[0].deliveryConfirmations, 1); assert.equal(settling.orders[0].payout, 'not-started'); assert.equal((await json<{ listings: MarketListingV3[] }>(second.base, '/api/market')).listings.length, 0); diff --git a/tests/market.test.ts b/tests/market.test.ts index 30f2859..2628421 100755 --- a/tests/market.test.ts +++ b/tests/market.test.ts @@ -37,7 +37,7 @@ function chainFixture() { return id; } }; } -async function fixture(t: { after(fn: () => Promise): void }, indexed = false) { +async function fixture(t: { after(fn: () => Promise): void }, indexed = false, useWorker = false) { const dir = await mkdtemp(join(tmpdir(), 'wikideck-market-')); const chain = chainFixture(); // Test-only indexed adapter: production indexes/events are exercised separately by Ledger tests. @@ -67,8 +67,8 @@ async function fixture(t: { after(fn: () => Promise): void }, indexed = fa }, } : chain; let time = 1_800_000_000_000; - let market = await Market.open(dir, adapter, { feeBps: 250, confirmations: 2 }, () => time); - t.after(async () => { market.close(); await rm(dir, { recursive: true, force: true }); }); + let market = await Market.open(dir, adapter, { feeBps: 250, confirmations: 2 }, () => time, { worker: useWorker }); + t.after(async () => { await market.close(); await rm(dir, { recursive: true, force: true }); }); const envelope = (who: typeof alice, action: MarketAction) => { const terms = { network: NETWORK, chainId: CHAIN_ID, marketId: market.id, actor: who.actor, publicKey: who.publicKey, nonce: randomUUID(), expiresAt: time + 60000, action }; @@ -106,8 +106,17 @@ async function fixture(t: { after(fn: () => Promise): void }, indexed = fa }; return { dir, chain, events, envelope, envelopeV3, run, list, buy, orders, delivery, confirm, get market() { return market; }, get now() { return time; }, advance: (ms: number) => { time += ms; }, - restart: async (policy: MarketPolicy = {}) => { market.close(); market = await Market.open(dir, adapter, - { feeBps: 500, confirmations: 5, ...policy }, () => time); } }; + restart: async (policy: MarketPolicy = {}) => { await market.close(); market = await Market.open(dir, adapter, + { feeBps: 500, confirmations: 5, ...policy }, () => time, { worker: useWorker }); } }; +} + +async function drainWorkerReconciliation(f: Awaited>, maxPasses = 100): Promise { + for (let pass = 0; pass < maxPasses; pass++) { + f.market.browse(); + if (!f.market.hasReconciliationBacklog() && !f.market.isReconciliationWorkerBusy()) return; + await new Promise(resolve => setTimeout(resolve, 1)); + } + assert.fail('Worker reconciliation did not drain within the bounded test wait'); } test('mock market switches block new listings/bids/checkout while allowing paid-order delivery and private activity', async t => { const f = await fixture(t), l = await f.list(), o = await f.buy(l); @@ -263,6 +272,45 @@ test('market reconciliation batch failure rolls back order effects and resumes f assert.equal(f.market.hasReconciliationBacklog(), false); }); +test('worker-backed reconciliation commits reorg holds atomically and resumes after restart', async t => { + const f = await fixture(t, true, true); + await drainWorkerReconciliation(f); + const 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) }); + await f.orders(); await drainWorkerReconciliation(f); + 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_worker_review BEFORE UPDATE ON orders WHEN json_extract(NEW.data,'$.status')='review' BEGIN SELECT RAISE(ABORT,'injected worker batch failure'); END"); + } finally { db.close(); } + f.market.browse(); + for (let i = 0; i < 100 && (f.market.isReconciliationWorkerBusy() || !f.market.diagnostics().reconciliationWorkerFailed); i++) { + await new Promise(resolve => setTimeout(resolve, 1)); + } + assert.equal(f.market.diagnostics().reconciliationWorkerFailed, true); + await assert.rejects(f.list(), /catching up/); // Do not admit new business while the worker commit is uncertain. + const check = new DatabaseSync(join(f.dir, 'testnet', 'market.sqlite')); + try { + assert.equal(JSON.parse((check.prepare('SELECT data FROM orders WHERE id=?').get(order.id) as { data: string }).data).status, 'complete'); + assert.equal(f.market.diagnostics().canonicalEventCursor, 1); + check.exec('DROP TRIGGER fail_worker_review'); + } finally { check.close(); } + + await f.restart(); + await drainWorkerReconciliation(f); + const reviewed = (await f.orders())[0]; + assert.equal(reviewed.status, 'review'); assert.equal(reviewed.payout, '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));