Worker-owned reconciliation commits

tests/market.test.ts

@@ -37,7 +37,7 @@ function chainFixture() {
     return id;
   } };
 }
-async function fixture(t: { after(fn: () => Promise<void>): void }, indexed = false) {
+async function fixture(t: { after(fn: () => Promise<void>): 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>): 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>): 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<ReturnType<typeof fixture>>, maxPasses = 100): Promise<void> {
+  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));