WikiDeck/wikideck
3
0
Fork 1
Code Issues 8 Pull Requests Actions Packages Projects 2 Releases Wiki Activity

Reconciliation uses 32-row SQLite slices

Browse Source
This commit is contained in:
eric
2026-10-21 10:29:45 -04:00
parent 6adfda3cf0
commit a127a40293
6 changed files with 44 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/market/store.ts
+6 -5
View File
@@ -52,7 +52,8 @@ const reconciliationJobSchema = z.object({ version: z.literal(1), stage: z.enum(
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;
// Keep each synchronous SQLite slice small enough to limit request-loop stalls.
export const MARKET_RECONCILIATION_BATCH_SIZE = 32;
const RECONCILIATION_JOB_KEY = 'canonical_reconciliation_job';
/** Node-local marketplace, NOT consensus escrow. Its SQLite journal is backed up with the ledger. */
@@ -128,7 +129,7 @@ export class Market {
}
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 }[];
.all(afterId, afterId, MARKET_RECONCILIATION_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;
@@ -270,10 +271,10 @@ export class Market {
}
if (job.stage === 'orders') {
if (rows.length < RECONCILE_BATCH_SIZE) nextJob = { ...job, stage: 'listings', afterId: null };
if (rows.length < MARKET_RECONCILIATION_BATCH_SIZE) nextJob = { ...job, stage: 'listings', afterId: null };
else nextJob = { ...job, afterId: rows.at(-1)!.id };
} else {
if (rows.length < RECONCILE_BATCH_SIZE) {
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')
@@ -305,7 +306,7 @@ export class Market {
}
// 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();
if (!completed && job.stage === 'orders' && rows.length < MARKET_RECONCILIATION_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');
server/testnet/http.ts
+17 -5
View File
@@ -251,17 +251,29 @@ 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.
let stopping=false;
let catchupPaused=false;
let catchupImmediate:NodeJS.Immediate|undefined;
// Drain durable catch-up without page-load dependence. Each pass is a smaller synchronous
// SQLite slice, then setImmediate yields so HTTP and peer work can run between passes.
const advanceMarketCatchup=()=>{
catchupImmediate=undefined;
if(stopping||catchupPaused||!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();}
};
const marketCatchupTimer=setInterval(()=>{
if(!market.hasReconciliationBacklog())return;
try {market.browse();}catch{console.error('Marketplace canonical catch-up failed; operator review required.');}
if(!stopping&&!catchupPaused&&!catchupImmediate&&market.hasReconciliationBacklog()){
catchupImmediate=setImmediate(advanceMarketCatchup);catchupImmediate.unref();
}
},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);clearInterval(marketCatchupTimer);
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.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);});
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);});