Added opt-in canonical observer polling lifecycle

server/market/canonical-observer.ts

@@ -0,0 +1,87 @@
+import { Ledger } from '../testnet/ledger.ts';
+import type { PostgresCustody } from './postgres-custody.ts';
+
+export type CanonicalObserverState = 'stopped' | 'running' | 'failed' | 'review-required';
+export type CanonicalObserverStatus = { state: CanonicalObserverState; cursor: number; head: number; height: number;
+  tip: string; caughtUp: boolean; failureCode: string | null };
+
+/** Bounded polling loop for the explicitly disposable PostgreSQL rehearsal only.
+ * It reads local Ledger events; it does not verify candidate consensus or trigger custody payouts.
+ */
+export class CanonicalObserver {
+  private readonly custody: PostgresCustody;
+  private readonly ledger: Ledger;
+  private readonly intervalMs: number;
+  private readonly pagesPerCycle: number;
+  private active = false;
+  private timer: ReturnType<typeof setTimeout> | undefined;
+  private cycle: Promise<void> | undefined;
+  private current: CanonicalObserverStatus = { state: 'stopped', cursor: 0, head: 0, height: 0,
+    tip: '', caughtUp: false, failureCode: null };
+
+  constructor(custody: PostgresCustody, ledger: Ledger, options: { intervalMs?: number; pagesPerCycle?: number } = {}) {
+    if (!custody.rehearsalObserverEnabled) throw new Error('Canonical observer requires disposable rehearsal custody');
+    if (process.env.WIKIDECK_CUSTODY_CANONICAL_REHEARSAL !== '1' || process.env.WIKIDECK_PG_DISPOSABLE_TEST !== '1' ||
+        process.env.NODE_ENV === 'production') throw new Error('Canonical observer requires explicit non-production disposable runtime opt-in');
+    if (!(ledger instanceof Ledger)) throw new Error('Canonical observer requires a local Ledger instance');
+    this.custody = custody; this.ledger = ledger;
+    this.intervalMs = options.intervalMs ?? 1000; this.pagesPerCycle = options.pagesPerCycle ?? 4;
+    if (!Number.isSafeInteger(this.intervalMs) || this.intervalMs < 25 || this.intervalMs > 60000 ||
+        !Number.isSafeInteger(this.pagesPerCycle) || this.pagesPerCycle < 1 || this.pagesPerCycle > 16) {
+      throw new Error('Invalid canonical observer bounds');
+    }
+  }
+
+  status(): CanonicalObserverStatus { return { ...this.current }; }
+
+  async start(): Promise<CanonicalObserverStatus> {
+    if (this.active) { await this.cycle; return this.status(); }
+    if (this.current.state === 'failed' || this.current.state === 'review-required') return this.status();
+    this.active = true; this.current = { ...this.current, state: 'running', failureCode: null };
+    await this.runCycle();
+    return this.status();
+  }
+
+  /** Deterministic wake-up hook; the scheduled poll and manual wake are serialized. */
+  async pollNow(): Promise<CanonicalObserverStatus> {
+    if (!this.active) throw new Error('Canonical observer is not running');
+    if (this.timer) { clearTimeout(this.timer); this.timer = undefined; }
+    await this.runCycle();
+    return this.status();
+  }
+
+  async close(): Promise<void> {
+    this.active = false;
+    if (this.timer) { clearTimeout(this.timer); this.timer = undefined; }
+    await this.cycle;
+    if (this.current.state === 'running') this.current = { ...this.current, state: 'stopped' };
+  }
+
+  private runCycle(): Promise<void> {
+    if (this.cycle) return this.cycle;
+    const work = (async () => {
+      for (let page = 0; page < this.pagesPerCycle && this.active; page++) {
+        try {
+          const result = await this.custody.observeLedgerCanonicalEvents(this.ledger);
+          this.current = { state: result.reviewRequired ? 'review-required' : 'running', cursor: result.cursor,
+            head: result.head, height: result.height, tip: result.tip, caughtUp: result.cursor >= result.head,
+            failureCode: result.reviewRequired ? 'posted-sale-terminal-disconnected' : null };
+          if (result.reviewRequired) { this.active = false; break; }
+          if (result.cursor >= result.head) break;
+        } catch {
+          this.current = { ...this.current, state: 'failed', failureCode: 'canonical-observer-failed' };
+          this.active = false;
+          break;
+        }
+      }
+    })();
+    this.cycle = work.finally(() => {
+      this.cycle = undefined;
+      if (this.active && !this.timer) {
+        this.timer = setTimeout(() => { this.timer = undefined; void this.runCycle(); }, this.intervalMs);
+        this.timer.unref();
+      }
+    });
+    return this.cycle;
+  }
+}

server/market/custody-storage.ts

@@ -8,13 +8,21 @@ import { custodyStorage, postgresConfig } from './postgres-config.ts';
 /** No automatic SQLite reinterpretation or fallback after a database connection/configuration failure. */
 export async function openCustodyStorage(directory: string, options: CustodyOptions, env = process.env): Promise<CustodyMarket | PostgresCustody> {
   const storage = custodyStorage(env), marker = join(directory, 'testnet', 'custody-postgres.json');
+  const observerMode = env.WIKIDECK_CUSTODY_CANONICAL_REHEARSAL;
+  if (observerMode !== undefined && observerMode !== '0' && observerMode !== '1') {
+    throw new Error('WIKIDECK_CUSTODY_CANONICAL_REHEARSAL must be exactly 0 or 1');
+  }
+  if (observerMode === '1' && (env.NODE_ENV === 'production' || storage !== 'postgres' || env.WIKIDECK_PG_DISPOSABLE_TEST !== '1')) {
+    throw new Error('Canonical observer rehearsal requires PostgreSQL and the disposable database test guard');
+  }
+  const custodyOptions = observerMode === '1' ? { ...options, disposableSaleRehearsal: true } : options;
   await mkdir(join(directory, 'testnet'), { recursive: true, mode: 0o700 });
   let existing: { version: number; custodianId: string } | undefined;
   try { existing = JSON.parse(await readFile(marker, 'utf8')); }
   catch (e) { if ((e as NodeJS.ErrnoException).code !== 'ENOENT') throw e; }
   if (storage === 'sqlite') {
     if (existing) throw new Error('Custody data directory is bound to PostgreSQL; SQLite fallback prohibited');
-    return CustodyMarket.open(directory, options);
+    return CustodyMarket.open(directory, custodyOptions);
   }
   const source = join(directory, 'testnet', 'custody-sandbox.sqlite');
   let frozen: string | undefined;
@@ -27,7 +35,7 @@ export async function openCustodyStorage(directory: string, options: CustodyOpti
       if (!frozen) throw new Error('Existing SQLite custody requires explicit stopped-node import before PostgreSQL activation');
     } finally { db.close(); }
   } catch (e) { if ((e as NodeJS.ErrnoException).code !== 'ENOENT') throw e; }
-  const market = await PostgresCustody.open(await postgresConfig(env), options);
+  const market = await PostgresCustody.open(await postgresConfig(env), custodyOptions);
   try {
     if (frozen && await market.importSource() !== frozen) throw new Error('PostgreSQL import does not match local frozen custody source');
     if (existing && (existing.version !== 1 || existing.custodianId !== market.id)) throw new Error('PostgreSQL custodian differs from data-directory binding');

server/market/postgres-custody.ts

@@ -104,6 +104,7 @@ export class PostgresCustody {
     return { enabled: true, mode: 'sandbox', policy: CUSTODY_POLICY, network: NETWORK, chainId: CHAIN_ID,
       custodianId: this.id, feeBps: this.options.feeBps, ...resolvedMarketPolicy(this.options), liveDeposits: false, liveWithdrawals: false, nftSettlement: false };
   }
+  get rehearsalObserverEnabled(): boolean { return this.options.disposableSaleRehearsal === true; }
   private async transaction<T>(write: boolean, fn: (c: PoolClient) => Promise<T>): Promise<T> {
     check(this.active < 16, 'Custody database busy; retry the original command shortly');
     this.active++;