Moved canonical transaction, NFT, and mempool fact reads into the reconciliation worker's own read-only SQLite snapshot

server/market/reconciliation-commit-worker.js

@@ -1,9 +1,12 @@
 import { DatabaseSync } from 'node:sqlite';
 import { parentPort, workerData } from 'node:worker_threads';
 import { isDeepStrictEqual } from 'node:util';
+import { createHash } from 'node:crypto';
 
 const db = new DatabaseSync(workerData.marketFile);
 db.exec('PRAGMA busy_timeout=5000; PRAGMA synchronous=FULL;');
+const ledger = workerData.ledgerFile ? new DatabaseSync(workerData.ledgerFile, { readOnly: true }) : null;
+ledger?.exec('PRAGMA busy_timeout=5000; PRAGMA query_only=ON;');
 const port = parentPort;
 if (!port) throw new Error('Reconciliation commit worker requires a parent port');
 
@@ -23,6 +26,46 @@ function rows(stage, afterId, limit) {
   return db.prepare(`SELECT id,data FROM ${table} WHERE (? IS NULL OR id>?) ORDER BY id LIMIT ?`)
     .all(afterId, afterId, limit);
 }
+function readLedgerFacts(request, selected) {
+  if (!ledger) return request.fallbackFacts;
+  ledger.exec('BEGIN');
+  try {
+    const tip = ledger.prepare('SELECT hash FROM canonical_blocks ORDER BY height DESC LIMIT 1').get()?.hash ?? request.genesisHash;
+    const height = ledger.prepare('SELECT COALESCE(MAX(height),0) AS n FROM canonical_blocks').get().n;
+    const eventHead = ledger.prepare('SELECT COALESCE(MAX(sequence),0) AS n FROM canonical_events').get().n;
+    const pendingTransactions = ledger.prepare('SELECT id,transaction_json FROM pending_transfers ORDER BY position').all()
+      .map(row => {
+        const tx = JSON.parse(row.transaction_json);
+        const cardIds = tx.type === 'trade' ? [...tx.offered, ...tx.requested].map(leg => leg.cardHash) : [tx.cardHash];
+        return { id: row.id, cardIds };
+      });
+    const pendingIds = pendingTransactions.map(tx => tx.id);
+    const deliveryIds = Object.create(null), transactionIds = new Set(), cardIds = new Set();
+    for (const row of selected) {
+      const record = JSON.parse(row.data);
+      if (request.stage === 'orders') {
+        const id = createHash('sha256').update(JSON.stringify([request.network, 'nft-transfer', request.chainId,
+          record.cardHash, record.seller, record.buyer, record.sequence])).digest('hex');
+        deliveryIds[record.id] = id; transactionIds.add(id); cardIds.add(record.cardHash);
+      } else cardIds.add(record.cardHash);
+    }
+    const deliveryTransactions = Object.create(null);
+    const readTx = ledger.prepare('SELECT height,block_hash FROM canonical_transactions WHERE id=?');
+    for (const id of transactionIds) deliveryTransactions[id] = readTx.get(id) ?? null;
+    const nfts = Object.create(null);
+    const readNft = ledger.prepare(`SELECT n.owner,n.sequence,n.minted_height,b.block_json FROM canonical_nfts n
+      JOIN blocks b ON b.hash=n.block_hash WHERE n.token=?`);
+    for (const card of cardIds) {
+      const nft = readNft.get(card);
+      nfts[card] = nft ? { owner: nft.owner, sequence: nft.sequence, confirmations: height - nft.minted_height + 1 } : null;
+    }
+    ledger.exec('COMMIT');
+    return { height,
+      pendingTransactions, pendingIds, pendingCardIds: [...new Set(pendingTransactions.flatMap(tx => tx.cardIds))],
+      deliveryIds, deliveryTransactions, nfts,
+      snapshot: { tip, eventHead, pendingDigest: JSON.stringify([...pendingIds].sort()) } };
+  } catch (error) { ledger.exec('ROLLBACK'); throw error; }
+}
 function validCard(facts, cardHash, owner, sequence, maturity) {
   const nft = facts.nfts[cardHash];
   return !!nft && nft.owner === owner && nft.sequence === sequence && nft.confirmations >= maturity;
@@ -104,14 +147,23 @@ function saveResults(result) {
 
 port.on('message', request => {
   try {
+    let snapshot, nextJob;
     if (request.operation === 'reconcile-batch') {
       atomic(() => {
         assertCurrentJob(request.expectedJob);
         const selected = rows(request.stage, request.afterId, request.batchSize);
-        const result = request.stage === 'orders' ? processOrders(request, selected) : processListings(request, selected);
+        const facts = readLedgerFacts(request, selected);
+        snapshot = facts.snapshot;
+        const prepared = { ...request, ...facts };
+        const result = request.stage === 'orders' ? processOrders(prepared, selected) : processListings(prepared, selected);
         saveResults(result);
+        nextJob = request.stage === 'orders'
+          ? selected.length < request.batchSize ? { ...request.expectedJob, stage: 'listings', afterId: null }
+            : { ...request.expectedJob, afterId: selected.at(-1).id }
+          : selected.length < request.batchSize ? { ...request.expectedJob, stage: 'finalize', afterId: null }
+            : { ...request.expectedJob, afterId: selected.at(-1).id };
         db.prepare("UPDATE market_meta SET value=? WHERE key='canonical_reconciliation_job'")
-          .run(JSON.stringify(request.nextJob));
+          .run(JSON.stringify(nextJob));
       });
     } else if (request.operation === 'finalize') {
       atomic(() => {
@@ -124,9 +176,9 @@ port.on('message', request => {
         db.prepare("DELETE FROM market_meta WHERE key='canonical_reconciliation_job'").run();
       });
     } else throw new Error('Unknown reconciliation worker operation');
-    port.postMessage({ id: request.id, ok: true });
+    port.postMessage({ id: request.id, ok: true, snapshot, nextJob });
   } catch (error) {
     port.postMessage({ id: request.id, ok: false, error: error instanceof Error ? error.message : 'Worker batch failed' });
   }
 });
-port.on('close', () => db.close());
+port.on('close', () => { ledger?.close(); db.close(); });