104 lines
5.2 KiB
TypeScript
Executable File
104 lines
5.2 KiB
TypeScript
Executable File
import type { Pool } from 'pg';
|
||
import { randomUUID } from 'node:crypto';
|
||
import { z } from 'zod';
|
||
import { MONERO_OBSERVER_WORKER_LOCK } from './postgres-schema.ts';
|
||
import { persistMoneroObservationScan, type PersistObservationScanResult } from './monero-observation-store.ts';
|
||
import type { MoneroObservationScanner } from './monero-observer.ts';
|
||
|
||
const fingerprintSchema = z.string().regex(/^[a-f0-9]{64}$/);
|
||
export const MONERO_OBSERVER_MAX_BLOCKS_PER_PAGE = 100_000;
|
||
export const MONERO_OBSERVER_OVERLAP_BLOCKS = 1_000;
|
||
export const MONERO_OBSERVER_MAX_PAGES_PER_BATCH = 4;
|
||
|
||
export type MoneroObserverBatchResult = {
|
||
walletFingerprint: string;
|
||
pages: PersistObservationScanResult[];
|
||
scanHeight: number;
|
||
walletHeight: number;
|
||
caughtUp: boolean;
|
||
};
|
||
|
||
/**
|
||
* Perform a small bounded catch-up batch. The database observation role supplies the durable cursor;
|
||
* the scanner never receives a spend or address-allocation capability.
|
||
*/
|
||
export async function runMoneroObserverBatch(pool: Pool, walletFingerprint: string, scanner: MoneroObservationScanner,
|
||
options: { now?: () => number; maxPages?: number } = {}): Promise<MoneroObserverBatchResult> {
|
||
fingerprintSchema.parse(walletFingerprint);
|
||
const maxPages = options.maxPages ?? MONERO_OBSERVER_MAX_PAGES_PER_BATCH;
|
||
if (!Number.isSafeInteger(maxPages) || maxPages < 1 || maxPages > MONERO_OBSERVER_MAX_PAGES_PER_BATCH) {
|
||
throw new Error('Invalid Monero observer page budget');
|
||
}
|
||
const pages: PersistObservationScanResult[] = [];
|
||
let walletHeight = 0, scanHeight = 0;
|
||
for (let pageNumber = 0; pageNumber < maxPages; pageNumber++) {
|
||
const state = await pool.query<{ scan_height: string }>(
|
||
'SELECT scan_height::text FROM wk_custody.monero_observer_state WHERE wallet_fingerprint=$1', [walletFingerprint]);
|
||
const expectedScanHeight = state.rowCount ? Number(state.rows[0].scan_height) : 0;
|
||
if (!Number.isSafeInteger(expectedScanHeight) || expectedScanHeight < 0) throw new Error('Invalid durable Monero observer cursor');
|
||
const fromHeight = Math.max(0, expectedScanHeight - MONERO_OBSERVER_OVERLAP_BLOCKS);
|
||
const page = await scanner.scanIncoming(fromHeight, MONERO_OBSERVER_MAX_BLOCKS_PER_PAGE);
|
||
if (page.network !== 'mainnet' || page.fromHeight !== fromHeight || page.throughHeight < expectedScanHeight ||
|
||
page.throughHeight - page.fromHeight >= MONERO_OBSERVER_MAX_BLOCKS_PER_PAGE) {
|
||
throw new Error('Monero observer returned a page outside the worker’s bounded cursor window');
|
||
}
|
||
const result = await persistMoneroObservationScan(pool, { scanId: randomUUID(), walletFingerprint,
|
||
expectedScanHeight, scannedAt: (options.now ?? Date.now)(), page });
|
||
pages.push(result); scanHeight = result.scanHeight; walletHeight = page.walletHeight;
|
||
if (page.throughHeight >= page.walletHeight) break;
|
||
}
|
||
return { walletFingerprint, pages, scanHeight, walletHeight, caughtUp: scanHeight >= walletHeight };
|
||
}
|
||
|
||
export type ObserverWorkerOptions = {
|
||
pollIntervalMs?: number;
|
||
onBatch?: (result: MoneroObserverBatchResult) => void;
|
||
onFailure?: () => void;
|
||
signal?: AbortSignal;
|
||
};
|
||
|
||
function sleep(ms: number, signal?: AbortSignal): Promise<void> {
|
||
if (signal?.aborted) return Promise.resolve();
|
||
return new Promise(resolve => {
|
||
const timer = setTimeout(done, ms); timer.unref();
|
||
function done() { clearTimeout(timer); signal?.removeEventListener('abort', done); resolve(); }
|
||
signal?.addEventListener('abort', done, { once: true });
|
||
});
|
||
}
|
||
|
||
/**
|
||
* Explicit opt-in worker loop. A PostgreSQL session advisory lock prevents duplicate scanner processes;
|
||
* scan failures are redacted and retried with bounded backoff. No journal/credit/release APIs are called.
|
||
*/
|
||
export async function runMoneroObserverWorker(pool: Pool, walletFingerprint: string, scanner: MoneroObservationScanner,
|
||
options: ObserverWorkerOptions = {}): Promise<void> {
|
||
fingerprintSchema.parse(walletFingerprint);
|
||
const pollIntervalMs = options.pollIntervalMs ?? 30_000;
|
||
if (!Number.isSafeInteger(pollIntervalMs) || pollIntervalMs < 1_000 || pollIntervalMs > 15 * 60_000) {
|
||
throw new Error('Invalid Monero observer polling interval');
|
||
}
|
||
const leader = await pool.connect();
|
||
let acquired = false;
|
||
let retryDelay = pollIntervalMs;
|
||
try {
|
||
const lock = await leader.query<{ locked: boolean }>('SELECT pg_try_advisory_lock($1::bigint) AS locked', [MONERO_OBSERVER_WORKER_LOCK]);
|
||
acquired = lock.rows[0]?.locked === true;
|
||
if (!acquired) throw new Error('A Monero observer worker already holds the database lock');
|
||
while (!options.signal?.aborted) {
|
||
try {
|
||
const result = await runMoneroObserverBatch(pool, walletFingerprint, scanner);
|
||
options.onBatch?.(result);
|
||
retryDelay = pollIntervalMs;
|
||
if (!result.caughtUp) continue;
|
||
} catch {
|
||
options.onFailure?.();
|
||
retryDelay = Math.min(Math.max(pollIntervalMs, retryDelay * 2), 5 * 60_000);
|
||
}
|
||
await sleep(retryDelay, options.signal);
|
||
}
|
||
} finally {
|
||
if (acquired) await leader.query('SELECT pg_advisory_unlock($1::bigint)', [MONERO_OBSERVER_WORKER_LOCK]).catch(() => {});
|
||
leader.release();
|
||
}
|
||
}
|