From 7025c373905b002d368029d7d7582c32b2315401 Mon Sep 17 00:00:00 2001 From: Eric Meehan Date: Tue, 20 Oct 2026 20:11:57 -0400 Subject: [PATCH] Added opt-in canonical observer polling lifecycle --- compose.postgres-test.yaml | 2 + docs/custody-sale-rehearsal.md | 7 ++- docs/production-plan.md | 2 +- server/market/canonical-observer.ts | 87 +++++++++++++++++++++++++++++ server/market/custody-storage.ts | 12 +++- server/market/postgres-custody.ts | 1 + server/testnet/http.ts | 25 +++++++-- tests/postgres/custody.test.ts | 79 ++++++++++++++++++++++++-- 8 files changed, 200 insertions(+), 15 deletions(-) create mode 100755 server/market/canonical-observer.ts diff --git a/compose.postgres-test.yaml b/compose.postgres-test.yaml index 2444749..7b287c8 100755 --- a/compose.postgres-test.yaml +++ b/compose.postgres-test.yaml @@ -22,6 +22,8 @@ services: condition: service_healthy environment: WIKIDECK_PG_DISPOSABLE_TEST: "1" + WIKIDECK_CUSTODY_CANONICAL_REHEARSAL: "1" + NODE_ENV: test command: ["npx", "tsx", "--test", "--test-concurrency=1", "tests/postgres/*.test.ts"] networks: fixture: diff --git a/docs/custody-sale-rehearsal.md b/docs/custody-sale-rehearsal.md index ff90500..47e698a 100755 --- a/docs/custody-sale-rehearsal.md +++ b/docs/custody-sale-rehearsal.md @@ -35,7 +35,9 @@ A submitted history that conflicts with, removes or makes immature an already po The same explicitly disposable PostgreSQL instance now has `observeCanonicalEvents()` and `observeLedgerCanonicalEvents(ledger)`. The ledger adapter requires an actual in-process `Ledger` object, checks its network/chain identity, and fetches one page at a time from its locally validated [canonical event API](../server/testnet/canonical-index.ts)—no caller/HTTP/peer page can enter through that adapter. The inbox is append-only and bounded. It checks stream identity, page limits, predecessor anchors, event hashes, sequence continuity, connect parent/tip and tip-first disconnect rules against a reconstructed canonical hash stack. It accepts exact retries and overlapping pages from concurrent consumers only after matching their persisted overlap; stale retries report the durable cursor/head, not the stale page head. The startup/periodic financial audit replays and verifies the stored inbox; a changed stream, gap or malformed/tampered event creates the normal durable integrity hold. Inbox records use namespaced `custody_meta` keys and the existing runtime role; there is no schema migration or mutable cursor. -The adapter is not automatically scheduled: there is no HTTP route or background worker. PostgreSQL tests include synthetic event pages for malformed/retry cases and one integration using blocks accepted by a local `Ledger` configured with a **test-only fake PoW engine**, including an equal-work side branch and a reorg. This tests the local adapter/API boundary, not real RandomX validation, external provenance or production multi-node behavior. Hash-anchor consistency is not proof of honest observation; the stream ID and unkeyed hashes are stored beside the events, so a schema owner able to coherently rewrite the complete history can also recompute the hashes. The included corruption test detects a partial edit, not a fully rewritten valid history. Production must anchor custody checkpoints outside the database's mutable trust domain. The disposable inbox is capped at 4,096 total events, stores the append-only event history and fully rescans it during audits. That is intentionally a small-fixture design, not production storage/cursor architecture or a scale claim. +An opt-in `CanonicalObserver` now schedules one bounded poll cycle on the server runtime, resumes from its durable cursor after restart, and stops/drains before PostgreSQL and the ledger close. Each cycle processes at most four pages of 64 events; lag is exposed on the private operations listener. Poll failures or sale-review events stop the loop fail-closed; it does not retry a failed observer automatically. The runtime opt-in requires `WIKIDECK_CUSTODY_CANONICAL_REHEARSAL=1`, PostgreSQL custody and `WIKIDECK_PG_DISPOSABLE_TEST=1`; the configured and actual database must also be `custody_test`, and `NODE_ENV=production` is explicitly rejected. Invalid/missing combinations fail closed, including direct construction of the polling controller. Shutdown stops scheduling immediately and drains the current page before custody/ledger close. This is **not enabled in the persistent sandbox or any regular deployment** and has no public HTTP control endpoint. + +PostgreSQL tests include synthetic event pages for malformed/retry cases and local `Ledger` blocks configured with a **test-only fake PoW engine**, including an equal-work side branch and a reorg. They exercise the adapter and bounded scheduler lifecycle, not real RandomX validation, external provenance or production multi-node behavior. Hash-anchor consistency is not proof of honest observation; the stream ID and unkeyed hashes are stored beside the events, so a schema owner able to coherently rewrite the complete history can also recompute the hashes. The included corruption test detects a partial edit, not a fully rewritten valid history. Production must anchor custody checkpoints outside the database's mutable trust domain. The disposable inbox is capped at 4,096 total events, stores the append-only event history and fully rescans it during audits. That is intentionally a small-fixture design, not production storage/cursor architecture or a scale claim. The synthetic PostgreSQL event-page seam now links an explicitly terminal candidate transaction ID to its posted order. A later **submitted disconnect event** containing that exact ID commits both the canonical event page and a durable per-order `sale-reorg-review` record plus global `posted-sale-terminal-disconnected` mutation hold in one transaction; the original journal is not reversed or paid again. This remains true if the transaction reappears later in the same replacement-branch page: review does not auto-clear on apparent re-inclusion. Generic incident resolution re-audits the retained review record and therefore cannot clear it. This exercises order-to-event linkage and atomic review behavior. The hold only detects disconnects actually delivered to this seam; it cannot detect an unseen intermediate fork. @@ -43,7 +45,7 @@ The local-ledger adapter is **not** linked to these terminal candidate IDs: acti ## Validation and remaining gates -TypeScript/Vite Docker build passed; **172 offline regressions and 28 PostgreSQL integration tests passed, zero failures/skips**. [Candidate tests](../tests/sale-candidate.test.ts) cover signature compatibility/recipient tampering and candidate transition rules. [PostgreSQL tests](../tests/postgres/custody.test.ts) add: +TypeScript/Vite Docker build passed; **172 offline regressions and 29 PostgreSQL integration tests passed, zero failures/skips**. [Candidate tests](../tests/sale-candidate.test.ts) cover signature compatibility/recipient tampering and candidate transition rules. [PostgreSQL tests](../tests/postgres/custody.test.ts) add: - explicit instance/database consent and exact signed reservation matching; - seed-binding changes, malformed/cyclic/oversized collections, and denial of ordinary cancellation after binding; @@ -53,6 +55,7 @@ TypeScript/Vite Docker build passed; **172 offline regressions and 28 PostgreSQL - one-atomic settlement and signed custom-recipient revenue across current-policy changes; - conflicting outcome → durable cross-instance/restart hold without a second credit; - anchored canonical-event connect/disconnect paging, overlapping/stale consumer retries, reconnect, partial history mutation audit, forged disconnect-ancestor rejection and tamper-triggered hold; actual locally validated blocks using test-only fake PoW, excluded equal-work branch and a canonical reorg through the adapter (no candidate sale link); +- opt-in HTTP runtime startup, initial observer checkpoint, 64-event bounded catch-up, clean shutdown, durable restart resume and invalid-flag rejection; - synthetic exact candidate-terminal transaction disconnect/reappearance in one page → committed event history plus order-linked durable hold, generic incident resolution denied, one preserved seller credit and no second payout; - injected terminal-evidence failure → complete rollback and safe retry; corrupted evidence → audit hold. diff --git a/docs/production-plan.md b/docs/production-plan.md index 7351f2f..784fcbd 100755 --- a/docs/production-plan.md +++ b/docs/production-plan.md @@ -88,7 +88,7 @@ Blocked on P0 ADR; use P1 amount types and P2 durable events. Consensus work is **Accounting rehearsal increment, 2026-10-20:** [signed candidate → PostgreSQL reservation resolution](custody-sale-rehearsal.md) now exercises actual restricted-role SQL terminal postings from bounded synthetic transcripts, behind explicit disposable database/instance guards and without an HTTP route. Exact terms/hosting recipients and reservation commitments are checked; mature completion credits seller/fees once, cancellation/non-admission releases buyer funds, and conflicting post-settlement evidence durably holds mutations rather than repaying. Insert-only first/terminal evidence is audited. Final build, 172 offline tests and 22 PostgreSQL integration tests passed. This supersedes the earlier absence of any balance-posting test integration, **not** the production/activation boundary: caller-supplied seeds/blocks are not canonical evidence, intermediate unseen disconnects cannot be detected, and no signer, canonical inbox, live deposit/withdrawal or current-chain paid sale is enabled. M3 remains open. -**Canonical/per-order reorg rehearsal increment, 2026-10-20:** the gated disposable PostgreSQL observer stores bounded pages from an in-process `Ledger` adapter, checks stream/anchor/hash/tip/ancestor continuity, supports retries, and audits retained history. A local-ledger test using fake PoW covers an excluded equal-work branch and reorg. A separate synthetic-event test links the deterministic candidate terminal transaction ID to its posted order; disconnecting that ID (even followed by same-page re-inclusion) commits the event history and durable order review hold atomically, without undoing or repeating the payout. Generic incident resolution cannot clear a sale-reorg record. Final build and 28 PostgreSQL tests pass; offline regressions remain 172. The worker is not scheduled, active v6 cannot emit candidate IDs, and synthetic linkage is not canonical sale evidence. Stream hashes are not tamper-proof against the schema owner; 4,096-event cap/full scan are non-production. M3 remains blocked on reviewed candidate activation, actual candidate IDs from a production local ledger, observer/checkpoint design and production readiness. +**Canonical/per-order reorg rehearsal increment, 2026-10-20:** the gated disposable PostgreSQL observer stores bounded pages from an in-process `Ledger` adapter, checks stream/anchor/hash/tip/ancestor continuity, supports retries, and audits retained history. A local-ledger test using fake PoW covers an excluded equal-work branch and reorg. A separate synthetic-event test links the deterministic candidate terminal transaction ID to its posted order; disconnecting that ID (even followed by same-page re-inclusion) commits the event history and durable order review hold atomically, without undoing or repeating the payout. Generic incident resolution cannot clear a sale-reorg record. The server now has an opt-in polling lifecycle, restricted to PostgreSQL `custody_test`, both disposable flags and non-production mode; it drains on shutdown, resumes the persistent cursor after restart and stops on review/failure. Final build, 172 offline regressions and 29 PostgreSQL tests pass. The worker remains disabled in all regular deployments, active v6 cannot emit candidate IDs, and synthetic linkage is not canonical sale evidence. Stream hashes are not tamper-proof against the schema owner; 4,096-event cap/full scan are non-production. M3 remains blocked on reviewed candidate activation, actual candidate IDs from an activated local ledger, production observer/checkpoint design and production readiness. - If a reviewed trustless design is selected: implement its actual cross-chain validity/evidence rules, compatible cryptographic primitives, both-party failure/refund paths and commission semantics. Run maintained reference vectors, negative tests, fuzz/property tests and specialist audit before claiming atomicity. - If a custodial/attested fallback is explicitly approved: define chain-enforced NFT lock, order identity, authorized releaser, cancellation/refund/expiry and stale-signature rules. Separate host control keys from seller wallet and processor payout keys. Reserve/refund procedures must not permit both buyer refund and later seller delivery/release. Describe who can steal, censor or lock assets and how recovery is authorized. diff --git a/server/market/canonical-observer.ts b/server/market/canonical-observer.ts new file mode 100755 index 0000000..8f3b87b --- /dev/null +++ b/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 | undefined; + private cycle: Promise | 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 { + 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 { + 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 { + 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 { + 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; + } +} diff --git a/server/market/custody-storage.ts b/server/market/custody-storage.ts index 3ce5bab..002d87c 100755 --- a/server/market/custody-storage.ts +++ b/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 { 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'); diff --git a/server/market/postgres-custody.ts b/server/market/postgres-custody.ts index 7f508c2..09c8f15 100755 --- a/server/market/postgres-custody.ts +++ b/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(write: boolean, fn: (c: PoolClient) => Promise): Promise { check(this.active < 16, 'Custody database busy; retry the original command shortly'); this.active++; diff --git a/server/testnet/http.ts b/server/testnet/http.ts index bda9dbf..f700b90 100644 --- a/server/testnet/http.ts +++ b/server/testnet/http.ts @@ -20,7 +20,8 @@ import { marketCommandV3Schema } from '../../shared/market-v3.ts'; import { custodyCommandSchema } from '../../shared/custody.ts'; import { CustodyMarket, CustodyError, custodyMode } from '../market/custody.ts'; import { openCustodyStorage } from '../market/custody-storage.ts'; -import type { PostgresCustody } from '../market/postgres-custody.ts'; +import { PostgresCustody } from '../market/postgres-custody.ts'; +import { CanonicalObserver } from '../market/canonical-observer.ts'; import { validateMarketPolicy } from '../market/policy.ts'; import { IS_MAINNET, PROFILE } from '../../shared/network.ts'; import { assertLaunchAllowed, launchManifest } from '../launch.ts'; @@ -32,6 +33,14 @@ assertLaunchAllowed(); const paymentConfig = marketConfig(); // Fail before opening node resources if live payments are requested. await validateMarketPolicy(paymentConfig); // Reject invalid recipient checksum/network before opening node resources. const custodyConfig = custodyMode(); // Explicit opt-in sandbox only. Unknown/live modes fail before opening resources. +const canonicalObserverMode = process.env.WIKIDECK_CUSTODY_CANONICAL_REHEARSAL; +if (canonicalObserverMode !== undefined && canonicalObserverMode !== '0' && canonicalObserverMode !== '1') { + throw new Error('WIKIDECK_CUSTODY_CANONICAL_REHEARSAL must be exactly 0 or 1'); +} +if (canonicalObserverMode === '1' && (process.env.NODE_ENV === 'production' || custodyConfig !== 'sandbox' || + process.env.WIKIDECK_CUSTODY_STORAGE !== 'postgres' || process.env.WIKIDECK_PG_DISPOSABLE_TEST !== '1')) { + throw new Error('Canonical observer rehearsal requires sandbox custody, PostgreSQL, and disposable test isolation'); +} const limits = resourceConfig(); const proxyTrust=trustedProxies(); const opsPort=operationsPort(); @@ -57,6 +66,7 @@ let market: Market; try { market = await Market.open(dataDir, ledger, paymentConfig); } catch(e) {peers.close();ledger.close();engine.close();await lock.close();await unlink(lockPath);throw e;} let custody: CustodyMarket | PostgresCustody | undefined; +let custodyObserver: CanonicalObserver | undefined; try { if (custodyConfig === 'sandbox') custody = await openCustodyStorage(dataDir, { ...paymentConfig, card: (hash, owner, sequence) => { @@ -65,7 +75,12 @@ try { ledger.pendingTransactions().some(tx => transactionCardIds(tx).includes(hash))) return undefined; return { title: nft.block.article.title }; } }); -} catch(e) {market.close();peers.close();ledger.close();engine.close();await lock.close();await unlink(lockPath);throw e;} + if (canonicalObserverMode === '1') { + if (!(custody instanceof PostgresCustody)) throw new Error('Canonical observer rehearsal requires PostgreSQL custody'); + custodyObserver = new CanonicalObserver(custody, ledger); + if ((await custodyObserver.start()).state !== 'running') throw new Error('Canonical observer rehearsal failed to start; inspect custody diagnostics'); + } +} catch(e) {await custodyObserver?.close();await custody?.close();market.close();peers.close();ledger.close();engine.close();await lock.close();await unlink(lockPath);throw e;} const app = express(); app.disable('x-powered-by'); if(proxyTrust.length)app.set('trust proxy',proxyTrust); @@ -226,6 +241,7 @@ const operations=startOperations(opsPort,()=>({network:NETWORK,paymentMode:'mock ledger:ledger.diagnostics(),native:engine.diagnostics(), peers:{configured:peers.status().length,connected:peers.status().filter(p=>p.connected).length}, rpc:{api:apiBudget.status(),p2p:peerBudget.status()},market:market.diagnostics(), + canonicalObserver:custodyObserver?.status(), custody: custody && 'diagnostics' in custody ? custody.diagnostics() : { backend: custody ? 'sqlite-sandbox' : 'off' }}),resolve(dataDir,PROFILE)); server.maxConnections=256; server.headersTimeout=10000; @@ -238,7 +254,8 @@ marketTimer.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); + const observerDrain=custodyObserver?.close()??Promise.resolve(); const timeout=setTimeout(()=>process.exit(1),25000);timeout.unref(); - server.close(async()=>{await custody?.close();market.close();ledger.close();engine.close();await lock.close();await unlink(lockPath).catch(()=>{});clearTimeout(timeout);process.exit(0);}); + 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);operations.close();peers.close();await custody?.close();market.close();ledger.close();engine.close();await lock.close();await unlink(lockPath).catch(()=>{});process.exit(1);}); +server.on('error',async()=>{clearInterval(marketTimer);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);}); diff --git a/tests/postgres/custody.test.ts b/tests/postgres/custody.test.ts index afd5ab3..615a8e4 100755 --- a/tests/postgres/custody.test.ts +++ b/tests/postgres/custody.test.ts @@ -28,6 +28,7 @@ import { SALE_CANDIDATE, saleTermsMessage, saleCompletionMessage, saleCancellati import { replayReservedSale, type SaleRehearsal } from '../../server/market/sale-rehearsal.ts'; import { keccak } from 'hash-wasm'; import type { CanonicalInboxPage } from '../../server/market/canonical-inbox.ts'; +import { CanonicalObserver } from '../../server/market/canonical-observer.ts'; import { Ledger } from '../../server/testnet/ledger.ts'; import { GENESIS_TIME, makeTemplate, miningBlob, mintId, transition } from '../../server/testnet/consensus.ts'; import { deriveStats } from '../../shared/enrichment.ts'; @@ -106,6 +107,9 @@ class FixturePow { async hash(key: string, blob: Uint8Array): Promise { return sha256(Buffer.concat([Buffer.from(key, 'hex'), Buffer.from(blob)])); } close(): void {} } +class EasyFixturePow extends FixturePow { + async hash(_key: string, _blob: Uint8Array): Promise { return '0'.repeat(64); } +} function fixtureArticle(pageId = 50000) { return { pageId, title: 'Canonical inbox fixture', extract: `Synthetic local-only canonical observer test fixture ${pageId}.`, image: 'https://upload.wikimedia.org/wikipedia/commons/test-image.jpg', url: `https://en.wikipedia.org/?curid=${pageId}`, @@ -117,13 +121,13 @@ function fixtureStats() { languagesTruncated: false, pageviews: { status: 'unavailable' as const, months: [], latestMonth: '2026-09', total: null, dailyAverage: null, trendPercent: null }, stats: deriveStats(pageBytes, languageCount, null) }; } -async function fixtureBlock(ancestors: TestnetBlock[], state: NftStateView, pageId = 50000) { +async function fixtureBlock(ancestors: TestnetBlock[], state: NftStateView, pageId = 50000, proof = new FixturePow()) { const miner = seller; const template = makeTemplate(ancestors, state, miner.actor, miner.publicKey, [], fixtureArticle(pageId), fixtureStats(), GENESIS_TIME + (ancestors.length + 1) * 30); - const header = structuredClone(template.header), pow = new FixturePow(); + const header = structuredClone(template.header); for (header.nonce = 0; header.nonce < 100000; header.nonce++) { - const blob = miningBlob(header), powHash = await pow.hash(header.key, blob); + const blob = miningBlob(header), powHash = await proof.hash(header.key, blob); if (BigInt(`0x${powHash}`) > BigInt(`0x${header.target}`)) continue; const hash = sha256(blob); return { ...header, hash, tokenId: mintId(header), powHash, @@ -276,9 +280,10 @@ test('SQLite cutover preserves identity, history, prepared attempts, ordering an test('PostgreSQL HTTP custody integration, safe storage binding and backup refusal use the real runtime role', { timeout: 45000 }, async t => { const f = await fixture(t), dir = await mkdtemp(join(tmpdir(), 'custody-pg-http-')); const urlFile = join(dir, 'pg-url'); await writeFile(urlFile, 'postgresql://custody_runtime@127.0.0.1/custody_test', { mode: 0o600 }); - const env = { ...process.env, HOST: '127.0.0.1', PORT: '0', NODE_ENV: 'production', WIKIDECK_DATA_DIR: dir, + const env = { ...process.env, HOST: '127.0.0.1', PORT: '0', NODE_ENV: 'test', WIKIDECK_DATA_DIR: dir, WIKIDECK_CUSTODY_MODE: 'sandbox', WIKIDECK_CUSTODY_STORAGE: 'postgres', WIKIDECK_CUSTODY_PG_URL_FILE: urlFile, - WIKIDECK_CUSTODY_PG_LOCAL_TEST: '1', WIKIDECK_PAYMENT_MODE: 'mock', WIKIDECK_PEERS: '' }; + WIKIDECK_CUSTODY_PG_LOCAL_TEST: '1', WIKIDECK_PG_DISPOSABLE_TEST: '1', WIKIDECK_CUSTODY_CANONICAL_REHEARSAL: '1', + WIKIDECK_PAYMENT_MODE: 'mock', WIKIDECK_PEERS: '' }; const child = spawn(process.execPath, ['--experimental-strip-types', 'server/index.ts'], { env, stdio: ['ignore', 'pipe', 'pipe'] }); const closed = new Promise(resolve => child.once('close', () => resolve())); t.after(async () => { if (child.exitCode === null) child.kill('SIGTERM'); await closed; await rm(dir, { recursive: true, force: true }); }); @@ -289,12 +294,14 @@ test('PostgreSQL HTTP custody integration, safe storage binding and backup refus child.once('exit', () => { clearTimeout(timer); reject(new Error('HTTP startup failed')); }); }); const page = await (await fetch(`${base}/api/custody`)).json(); assert.equal(page.info.custodianId, f.market.id); + const observerStream = await f.admin.query("SELECT value FROM wk_custody.custody_meta WHERE key='canonical-inbox:v1:stream'"); + assert.match(observerStream.rows[0].value, /^[a-f0-9-]{36}$/); const command = envelope(f.market.id, buyer, { type: 'mock-credit' }); const response = await fetch(`${base}/api/custody/command`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(command) }); assert.equal(response.status, 200); assert.equal((await response.json()).availableAtomicXmr, '1000000000000'); assert.match(response.headers.get('cache-control') ?? '', /no-store/); assert.equal(JSON.parse(await readFile(join(dir, 'testnet', 'custody-postgres.json'), 'utf8')).custodianId, f.market.id); - child.kill('SIGTERM'); await closed; + child.kill('SIGTERM'); await closed; assert.equal(child.exitCode, 0, 'opt-in observer should drain before custody shutdown'); await assert.rejects(createBackup(dir, join(dir, '..', `refused-${randomUUID()}`)), /coordinated/); await assert.rejects(openCustodyStorage(dir, options, { WIKIDECK_CUSTODY_STORAGE: 'sqlite' }), /fallback prohibited/); }); @@ -303,6 +310,14 @@ test('configuration requires secret files, verified TLS, literal backend, and lo const dir = await mkdtemp(join(tmpdir(), 'custody-pg-config-')); t.after(() => rm(dir, { recursive: true, force: true })); const file = join(dir, 'connection'); assert.equal(custodyStorage({}), 'sqlite'); assert.throws(() => custodyStorage({ WIKIDECK_CUSTODY_STORAGE: 'other' })); + await assert.rejects(openCustodyStorage(dir, options, { WIKIDECK_CUSTODY_STORAGE: 'sqlite', + WIKIDECK_CUSTODY_CANONICAL_REHEARSAL: '1', WIKIDECK_PG_DISPOSABLE_TEST: '1' }), /requires PostgreSQL/); + await assert.rejects(openCustodyStorage(dir, options, { WIKIDECK_CUSTODY_STORAGE: 'postgres', + WIKIDECK_CUSTODY_CANONICAL_REHEARSAL: '1' }), /disposable database test guard/); + await assert.rejects(openCustodyStorage(dir, options, { NODE_ENV: 'production', WIKIDECK_CUSTODY_STORAGE: 'postgres', + WIKIDECK_CUSTODY_CANONICAL_REHEARSAL: '1', WIKIDECK_PG_DISPOSABLE_TEST: '1' }), /requires PostgreSQL/); + await assert.rejects(openCustodyStorage(dir, options, { WIKIDECK_CUSTODY_STORAGE: 'postgres', + WIKIDECK_CUSTODY_CANONICAL_REHEARSAL: 'yes', WIKIDECK_PG_DISPOSABLE_TEST: '1' }), /exactly 0 or 1/); await assert.rejects(postgresConfig({}), /configuration/); await writeFile(file, 'postgresql://user:secret@database.example/custody'); const secure = await postgresConfig({ WIKIDECK_CUSTODY_PG_URL_FILE: file }); assert.equal((secure.ssl as { rejectUnauthorized: boolean }).rejectUnauthorized, true); @@ -562,6 +577,58 @@ test('gated adapter consumes actual validated local Ledger connect and reorg pag await observer.audit(); }); +test('scheduled observer bounds catch-up, stops cleanly, and resumes its durable cursor after restart', async t => { + const dir = await mkdtemp(join(tmpdir(), 'wikideck-pg-observer-runner-')); + const pow = new EasyFixturePow(), ledger = await Ledger.open(dir, pow), f = await fixture(t); + const observer = await PostgresCustody.open(f.runtime, { ...options, disposableSaleRehearsal: true }); + t.after(async () => { await observer.close(); ledger.close(); pow.close(); await rm(dir, { recursive: true, force: true }); }); + const chain: TestnetBlock[] = []; let state: NftStateView = {}; + for (let i = 0; i < 66; i++) { + const block = await fixtureBlock(chain, state, 51000 + i, pow); + await ledger.submit(block); state = transition(state, block, []); chain.push(block); + } + const firstRunner = new CanonicalObserver(observer, ledger, { intervalMs: 60000, pagesPerCycle: 1 }); + try { + const priorMode = process.env.NODE_ENV, priorGate = process.env.WIKIDECK_CUSTODY_CANONICAL_REHEARSAL; + try { + process.env.NODE_ENV = 'production'; + assert.throws(() => new CanonicalObserver(observer, ledger), /non-production/); + process.env.NODE_ENV = 'test'; delete process.env.WIKIDECK_CUSTODY_CANONICAL_REHEARSAL; + assert.throws(() => new CanonicalObserver(observer, ledger), /non-production/); + } finally { + if (priorMode === undefined) delete process.env.NODE_ENV; else process.env.NODE_ENV = priorMode; + if (priorGate === undefined) delete process.env.WIKIDECK_CUSTODY_CANONICAL_REHEARSAL; + else process.env.WIKIDECK_CUSTODY_CANONICAL_REHEARSAL = priorGate; + } + const first = await firstRunner.start(); + assert.equal(first.state, 'running'); assert.equal(first.cursor, 64); assert.equal(first.head, 66); assert.equal(first.caughtUp, false); + const caughtUp = await firstRunner.pollNow(); + assert.equal(caughtUp.cursor, 66); assert.equal(caughtUp.caughtUp, true); assert.equal(caughtUp.tip, chain.at(-1)!.hash); + await firstRunner.close(); assert.equal(firstRunner.status().state, 'stopped'); + const next = await fixtureBlock(chain, state, 52000, pow); + await ledger.submit(next); + assert.equal((await f.admin.query("SELECT count(*)::int AS n FROM wk_custody.custody_meta WHERE key LIKE 'canonical-inbox:v1:event:%'")).rows[0].n, 66); + await assert.rejects(firstRunner.pollNow(), /not running/); + const restarted = new CanonicalObserver(observer, ledger, { intervalMs: 60000, pagesPerCycle: 1 }); + try { + const resumed = await restarted.start(); + assert.equal(resumed.cursor, 67); assert.equal(resumed.head, 67); assert.equal(resumed.caughtUp, true); + assert.equal(resumed.tip, next.hash); + } finally { await restarted.close(); } + assert.equal((await f.admin.query("SELECT count(*)::int AS n FROM wk_custody.custody_meta WHERE key LIKE 'canonical-inbox:v1:event:%'")).rows[0].n, 67); + chain.push(next); state = transition(state, next, []); + const automatic = new CanonicalObserver(observer, ledger, { intervalMs: 25, pagesPerCycle: 1 }); + try { + assert.equal((await automatic.start()).cursor, 67); + const automaticBlock = await fixtureBlock(chain, state, 52001, pow); + await ledger.submit(automaticBlock); + const deadline = Date.now() + 3000; + while (automatic.status().cursor < 68 && Date.now() < deadline) await new Promise(resolve => setTimeout(resolve, 10)); + assert.equal(automatic.status().cursor, 68, 'scheduled poll did not consume the new ledger event'); + } finally { await automatic.close(); } + } finally { await firstRunner.close(); } +}); + test('canonical inbox rejects a disconnected block with a forged but rehashed ancestor', async t => { const f = await fixture(t), observer = await PostgresCustody.open(f.runtime, { ...options, disposableSaleRehearsal: true }); const streamId = randomUUID(), events = canonicalEvents(streamId);