Implemented and validated the separate Monero observation-worker path

server/market/monero-observer-worker.ts

@@ -0,0 +1,103 @@
+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();
+  }
+}