diff --git a/docs/custody-sale-rehearsal.md b/docs/custody-sale-rehearsal.md index 3028308..6adcfed 100755 --- a/docs/custody-sale-rehearsal.md +++ b/docs/custody-sale-rehearsal.md @@ -31,11 +31,17 @@ The existing sandbox `mock-sale-settlement` journal and `fixture-complete`/`fixt A submitted history that conflicts with, removes or makes immature an already posted outcome triggers `sale-rehearsal-outcome-reorg`. Business changes roll back, and a redacted durable integrity incident commits while retaining the shared writer lock. All custody instances then reject new mutations; private reads and original signed receipts remain available. The original seller credit/refund stays recorded: no automatic compensation, reversal or second payout. Re-submitting the old mature history cannot clear the incident. Existing schema-owner incident review procedures are not a substitute for resolving chain risk; production resolution requires retained canonical evidence and approved loss policy. -This rehearsal is **not a durable connect/disconnect observer**. It retains first/terminal evidence and the redacted incident reason, not every submitted fork transcript. It cannot detect an intermediate disconnect that is never submitted, even if an outcome later reappears. The caller must retain conflicting evidence for investigation. Production must integrate a replay-safe canonical event inbox/cursor, history anchors, disconnect-before-reconnect handling and persistent per-order review. No monotonic canonical-height guarantee is claimed for this caller-driven seam. +## Local canonical-event inbox foundation + +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. + +Most importantly, the inbox is not connected to candidate transaction IDs, `replayReservedSale()` or settlement decisions. Existing v6 blocks cannot contain candidate sale transactions, and local free transfers are not accepted as sale completion evidence. It therefore **does not create a sale-specific disconnect hold** or detect an unsubmitted intermediate fork that later disappears before a page is observed. The synthetic sale transcript still retains only first/terminal evidence and the redacted incident reason; the caller must retain conflicting transcript evidence. Production must integrate a durable replay-safe canonical consumer with per-order candidate transaction linkage, then persist a review hold for a previously posted outcome. No production sale-observation guarantee is claimed for this increment. ## Validation and remaining gates -TypeScript/Vite Docker build passed; **172 offline regressions and 22 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 27 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; @@ -44,8 +50,9 @@ TypeScript/Vite Docker build passed; **172 offline regressions and 22 PostgreSQL - identical-rounded-allocation but different fee-rate rejection; - 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 validated local-ledger blocks, excluded equal-work branch and a canonical reorg through the adapter (no candidate sale link); - injected terminal-evidence failure → complete rollback and safe retry; corrupted evidence → audit hold. A first bounded-checkpoint implementation attempted metadata UPDATE and was rejected by the real runtime grants. It was corrected to retain insert-only first/terminal evidence; no privilege was widened. Test evidence is for this implementation boundary, not production approval. -**M3 remains open.** Next work is independently reviewed protocol/policy and explicit new-network activation design, then actual PoW/peer/index/undo integration and durable canonical observation tied to financially fenced host-signing intent. Monero deposit confirmation/unlock observation, treasury/signer separation, real withdrawal controls, coordinated recovery, capacity and legal/operational release gates remain separate. The active v6 chain still rejects every candidate sale transaction. Persistent local custody deployments were not switched or restarted by this rehearsal. \ No newline at end of file +**M3 remains open.** Next work is independently reviewed protocol/policy and explicit new-network activation design, then actual PoW/peer/index/undo integration and durable canonical observation tied to financially fenced host-signing intent. Monero deposit confirmation/unlock observation, treasury/signer separation, real withdrawal controls, coordinated recovery, capacity and legal/operational release gates remain separate. The active v6 chain still rejects every candidate sale transaction. Persistent local custody deployments were not switched or restarted by this rehearsal. diff --git a/docs/production-plan.md b/docs/production-plan.md index e649e89..fb9351e 100755 --- a/docs/production-plan.md +++ b/docs/production-plan.md @@ -88,6 +88,8 @@ 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 inbox rehearsal increment, 2026-10-20:** the same gated disposable PostgreSQL instance now stores bounded append-only pages from an in-process `Ledger` adapter, checks stream/anchor/hash/tip/ancestor continuity, permits exact/overlapping/stale consumer retries, and verifies retained history during audits. Integration tests use locally validated blocks to exercise an excluded equal-work branch and a real reorg; tampering or stream changes durably hold mutations. Final build and 27 PostgreSQL tests pass; the offline regression count remains 172. The adapter is not scheduled, its stream is not tamper-proof against a schema owner, it does not match candidate transaction IDs or drive per-order settlement/reorg review, and the inbox's 4,096-event cap/full scan make it explicitly non-production. M3 remains blocked on production observer/checkpoint design, candidate network activation, transaction linkage and per-order reorg handling. + - 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. - Completed-sale inclusion must require the approved evidence that payment is sufficiently confirmed. A peer must not insert a completed sale merely because its local database says paid. Define threshold/policy version, proof/attestation authorization and deterministic validation; no external HTTP calls in consensus validation. diff --git a/docs/sale-lock-candidate.md b/docs/sale-lock-candidate.md index 11cafc9..990de7a 100755 --- a/docs/sale-lock-candidate.md +++ b/docs/sale-lock-candidate.md @@ -73,6 +73,6 @@ Initial candidate evidence: 162 offline tests passed, including 13 candidate tes 1. Independent protocol/consensus review and operator approval of authority, cancellation and numeric policy parameters; final specification and test vectors. 2. Versioned network/manifest and explicit migration/activation strategy. No enabling the candidate under the existing v6 chain identity. 3. Integrate new transaction encodings/state commitments with mint/transfer/trade validation, canonical SQL undo/indexes/events, actual RandomX blocks, peer sync, restart and multi-host fork/partition tests. The current model supplies no PoW, issuance or fork-choice implementation. -4. Production PostgreSQL sale witness/terms storage, financially reserved host-signing workflow, outbox fencing, canonical observer/inbox, atomic terminal postings and reorg holds. The disposable signed-transcript rehearsal exercises terminal postings and durable review, but supplies none of the canonical observer/host-signing integration. Existing sandbox signatures do not authorize the new sale protocol. No signing before final exact order consent. +4. Production PostgreSQL sale witness/terms storage, financially reserved host-signing workflow, outbox fencing, local-ledger canonical observer/inbox, transaction linkage, atomic terminal postings and per-order reorg holds. The disposable signed-transcript rehearsal exercises terminal postings and durable review. A separate bounded PostgreSQL event-inbox rehearsal checks page hashes, anchors, overlapping retries and connect/disconnect tip progression, but is not wired to the ledger and is not linked to a sale. Existing sandbox signatures do not authorize the new sale protocol. No signing before final exact order consent. 5. Publication/review UI, key management/rotation, privacy and public-price disclosure, window/griefing/storage limits and operational recovery. Public terms would reveal sale linkage and amounts; privacy acceptance is a product decision. 6. Monero confirmation/unlock observation, real treasury controls, coordinated financial/node recovery and independently reviewed production release gates. No live wallet, funds or credentials were used in this increment. diff --git a/server/market/canonical-inbox.ts b/server/market/canonical-inbox.ts new file mode 100755 index 0000000..1e51fab --- /dev/null +++ b/server/market/canonical-inbox.ts @@ -0,0 +1,128 @@ +import type { PoolClient } from 'pg'; +import { z } from 'zod'; +import { GENESIS_HASH, sha256 } from '../testnet/consensus.ts'; + +const hex = z.string().regex(/^[a-f0-9]{64}$/); +const MAX_REHEARSAL_EVENTS = 4096; +const eventSchema = z.object({ sequence: z.number().int().positive().safe(), type: z.enum(['connect', 'disconnect']), + height: z.number().int().positive().safe(), hash: hex, previousHash: hex, + transactionIds: z.array(hex).max(64), eventHash: hex }).strict(); +const pageSchema = z.object({ streamId: z.string().uuid(), previousAnchor: hex, anchor: hex, + events: z.array(eventSchema).max(64), cursor: z.number().int().nonnegative().safe(), + head: z.number().int().nonnegative().safe(), caughtUp: z.boolean() }).strict(); +export type CanonicalInboxPage = z.infer; +export type CanonicalInboxResult = { streamId: string; cursor: number; head: number; height: number; tip: string; replayed: boolean }; + +export class CanonicalInboxError extends Error { + readonly code: string; + constructor(code: string) { super(`Canonical event inbox rejected input: ${code}`); this.code = code; } +} +const streamKey = 'canonical-inbox:v1:stream'; +const eventKey = (sequence: number) => `canonical-inbox:v1:event:${sequence.toString().padStart(16, '0')}`; +function storedEvent(text: string): z.infer { + let decoded: unknown; + try { decoded = JSON.parse(text); } catch { throw new CanonicalInboxError('stored-event-json-invalid'); } + const parsed = eventSchema.safeParse(decoded); + if (!parsed.success) throw new CanonicalInboxError('stored-event-invalid'); + return parsed.data; +} +async function anchorAt(client: PoolClient, sequence: number): Promise { + if (sequence === 0) return GENESIS_HASH; + const row = await client.query('SELECT value FROM wk_custody.custody_meta WHERE key=$1', [eventKey(sequence)]); + return row.rows[0] ? storedEvent(row.rows[0].value).eventHash : undefined; +} +function eventDigest(streamId: string, e: z.infer, previousAnchor: string): string { + return sha256(JSON.stringify(['wikideck-canonical-event-v1', streamId, e.sequence, e.type, + e.height, e.hash, e.previousHash, e.transactionIds, previousAnchor])); +} +function applyTip(stack: string[], e: z.infer): void { + if (e.type === 'connect') { + if (e.height !== stack.length + 1 || e.previousHash !== (stack.at(-1) ?? GENESIS_HASH)) throw new CanonicalInboxError('connect-parent-mismatch'); + stack.push(e.hash); return; + } + if (e.height !== stack.length || e.hash !== stack.at(-1) || e.height < 1 || + e.previousHash !== (stack.at(-2) ?? GENESIS_HASH)) throw new CanonicalInboxError('disconnect-ancestry-mismatch'); + stack.pop(); +} +async function readCanonicalHistory(client: PoolClient) { + const rows = await client.query(`SELECT key,value FROM wk_custody.custody_meta + WHERE key=$1 OR key LIKE 'canonical-inbox:v1:event:%' ORDER BY key LIMIT $2`, [streamKey, MAX_REHEARSAL_EVENTS + 2]); + if (rows.rows.length > MAX_REHEARSAL_EVENTS + 1) throw new CanonicalInboxError('rehearsal-event-budget-exceeded'); + const stream = rows.rows.find(r => r.key === streamKey)?.value as string | undefined; + if (rows.rows.length && (!stream || !z.string().uuid().safeParse(stream).success)) throw new CanonicalInboxError('stream-record-invalid'); + const stack: string[] = []; + let cursor = 0, anchor = GENESIS_HASH; + for (const row of rows.rows) { + if (row.key === streamKey) continue; + const match = /^canonical-inbox:v1:event:(\d{16})$/.exec(row.key); + const e = storedEvent(row.value); + if (!match || Number(match[1]) !== cursor + 1 || e.sequence !== cursor + 1) throw new CanonicalInboxError('history-gap-or-invalid-record'); + if (e.eventHash !== eventDigest(stream!, e, anchor)) throw new CanonicalInboxError('event-anchor-invalid'); + applyTip(stack, e); cursor++; anchor = e.eventHash; + } + return { streamId: stream, cursor, anchor, stack }; +} +function assertPageShape(page: CanonicalInboxPage): void { + const finalCursor = page.events.at(-1)?.sequence ?? page.cursor; + if ((page.events.length && page.events[0].sequence < 1) || page.cursor !== finalCursor || + page.head < page.cursor || page.caughtUp !== (page.cursor === page.head) || + (!page.events.length && !page.caughtUp) || + page.anchor !== (page.events.at(-1)?.eventHash ?? page.previousAnchor) || page.events.some((e, i) => + i > 0 && e.sequence !== page.events[i - 1].sequence + 1)) throw new CanonicalInboxError('page-shape-invalid'); +} + +/** Persist only events returned by the local ledger's validated canonical-event API. This verifies + * continuity/integrity, not PoW or provenance; callers must not pass peer-supplied, unvalidated data. + */ +export async function ingestCanonicalPage(client: PoolClient, raw: unknown): Promise { + const parsed = pageSchema.safeParse(raw); + if (!parsed.success) throw new CanonicalInboxError('page-schema-invalid'); + const page = parsed.data; + assertPageShape(page); + if (page.head > MAX_REHEARSAL_EVENTS || page.cursor > MAX_REHEARSAL_EVENTS) throw new CanonicalInboxError('rehearsal-event-budget-exceeded'); + const history = await readCanonicalHistory(client); + const storedStream = history.streamId; + if (storedStream && storedStream !== page.streamId) throw new CanonicalInboxError('stream-changed'); + let { cursor, anchor } = history; + const stack = history.stack; + if (!storedStream) { + if (cursor !== 0 || page.previousAnchor !== GENESIS_HASH || + (page.events.length ? page.events[0].sequence !== 1 : page.cursor !== 0)) { + throw new CanonicalInboxError('initial-history-required'); + } + await client.query('INSERT INTO wk_custody.custody_meta(key,value) VALUES($1,$2)', [streamKey, page.streamId]); + } + if (!page.events.length) { + const savedAnchor = await anchorAt(client, page.cursor); + if (page.cursor > cursor || savedAnchor !== page.anchor || page.previousAnchor !== page.anchor) throw new CanonicalInboxError('empty-cursor-mismatch'); + return { streamId: page.streamId, cursor, head: Math.max(page.head, cursor), height: stack.length, + tip: stack.at(-1) ?? GENESIS_HASH, replayed: true }; + } + const firstSequence = page.events[0].sequence; + if (firstSequence > cursor + 1) throw new CanonicalInboxError('cursor-gap'); + const predecessor = await anchorAt(client, firstSequence - 1); + if (!predecessor || page.previousAnchor !== predecessor) throw new CanonicalInboxError('cursor-anchor-mismatch'); + let priorAnchor = predecessor; + let appended = false; + for (const e of page.events) { + if (e.eventHash !== eventDigest(page.streamId, e, priorAnchor)) throw new CanonicalInboxError('event-hash-invalid'); + if (e.sequence <= cursor) { + const saved = await client.query('SELECT value FROM wk_custody.custody_meta WHERE key=$1', [eventKey(e.sequence)]); + if (!saved.rowCount || JSON.stringify(storedEvent(saved.rows[0].value)) !== JSON.stringify(e)) throw new CanonicalInboxError('replayed-event-mismatch'); + } else { + if (e.sequence !== cursor + 1 || priorAnchor !== anchor) throw new CanonicalInboxError('cursor-anchor-mismatch'); + applyTip(stack, e); + await client.query('INSERT INTO wk_custody.custody_meta(key,value) VALUES($1,$2)', [eventKey(e.sequence), JSON.stringify(e)]); + cursor = e.sequence; anchor = e.eventHash; appended = true; + } + priorAnchor = e.eventHash; + } + if (page.cursor !== page.events.at(-1)!.sequence || page.anchor !== priorAnchor) throw new CanonicalInboxError('cursor-anchor-mismatch'); + return { streamId: page.streamId, cursor, head: Math.max(page.head, cursor), height: stack.length, + tip: stack.at(-1) ?? GENESIS_HASH, replayed: !appended }; +} + +/** Full append-only inbox verification. Expected to run under the caller's coherent audit snapshot. */ +export async function auditCanonicalInbox(client: PoolClient): Promise { + await readCanonicalHistory(client); +} diff --git a/server/market/postgres-custody.ts b/server/market/postgres-custody.ts index d5bee02..2bd96c3 100755 --- a/server/market/postgres-custody.ts +++ b/server/market/postgres-custody.ts @@ -2,6 +2,7 @@ import { Pool, type PoolClient, type PoolConfig } from 'pg'; import { createHash, randomUUID } from 'node:crypto'; import { IS_MAINNET, NETWORK } from '../../shared/network.ts'; import { CHAIN_ID, verifySignature } from '../testnet/consensus.ts'; +import { Ledger } from '../testnet/ledger.ts'; import { ADDITIONAL_MARKET_FEE } from '../../shared/market.ts'; import { allocateXmrSale } from '../../shared/xmr-pricing.ts'; import { parseAtomicXmr } from '../../shared/xmr-amount.ts'; @@ -13,6 +14,7 @@ import { CUSTODY_WRITE_LOCK, verifyCustodySchema } from './postgres-schema.ts'; import { resolvedMarketPolicy, validateMarketPolicy, permitsMarketAction, type MarketPolicy } from './policy.ts'; import { CustodyIntegrityError, auditBusinessLiabilities, auditOrder, auditWithdrawal } from './custody-integrity.ts'; import { replayReservedSale } from './sale-rehearsal.ts'; +import { CanonicalInboxError, auditCanonicalInbox, ingestCanonicalPage } from './canonical-inbox.ts'; export type CustodyOptions = { feeBps: number; card(hash: string, owner: string, sequence: number): { title: string } | undefined; now?: () => number; disposableSaleRehearsal?: boolean } & MarketPolicy; @@ -337,6 +339,38 @@ export class PostgresCustody { return { order: o, action: replay.decision.action, reservationDigest: replay.reservationDigest, stateRoot: replay.stateRoot }; }); } + /** Disposable observer rehearsal only; input must come from Ledger.canonicalEvents(), never peers/HTTP. */ + observeCanonicalEvents(page: unknown) { + check(this.options.disposableSaleRehearsal === true, 'Canonical event inbox rehearsal is unavailable on this custody instance'); + return this.transaction(true, async c => { + check(!this.diagnostics().mutationHold, 'Custody reconciliation hold'); + check(!(await c.query("SELECT 1 FROM wk_custody.custody_meta WHERE key='integrity_incident_v1'")).rowCount, 'Custody integrity incident unresolved'); + try { return await ingestCanonicalPage(c, page); } + catch (error) { + if (error instanceof CanonicalInboxError) throw new CustodyIntegrityError(`canonical-inbox-${error.code}`); + throw error; + } + }); + } + /** Consume one bounded page directly from this process's locally validated ledger; no HTTP/peer feed. */ + async observeLedgerCanonicalEvents(ledger: Ledger) { + check(this.options.disposableSaleRehearsal === true, 'Canonical event inbox rehearsal is unavailable on this custody instance'); + check(ledger instanceof Ledger, 'Canonical inbox requires a local Ledger instance'); + const summary = ledger.summary(); + check(summary.network === NETWORK && summary.chainId === CHAIN_ID, 'Canonical ledger identity mismatch'); + const cursor = await this.transaction(false, async c => { + const row = await c.query(`SELECT key FROM wk_custody.custody_meta + WHERE key LIKE 'canonical-inbox:v1:event:%' ORDER BY key DESC LIMIT 1`); + if (!row.rowCount) return 0; + const match = /^canonical-inbox:v1:event:(\d{16})$/.exec(row.rows[0].key); + if (!match) throw new CustodyIntegrityError('canonical-inbox-cursor-invalid'); + return Number(match[1]); + }); + let page: unknown; + try { page = ledger.canonicalEvents(cursor, 64); } + catch { throw new CustodyIntegrityError('canonical-ledger-cursor-mismatch'); } + return this.observeCanonicalEvents(page); + } audit(): Promise { if (this.auditInFlight) return this.auditInFlight; this.auditState.lastAttemptAt = Date.now(); @@ -359,6 +393,11 @@ export class PostgresCustody { export async function auditPostgres(c: PoolClient): Promise { const incident = await c.query("SELECT 1 FROM wk_custody.custody_meta WHERE key='integrity_incident_v1'"); if (incident.rowCount) throw new CustodyIntegrityError('unresolved-integrity-incident'); + try { await auditCanonicalInbox(c); } + catch (error) { + if (error instanceof CanonicalInboxError) throw new CustodyIntegrityError(`canonical-inbox-${error.code}`); + throw error; + } const bad = await c.query(`SELECT 1 FROM wk_custody.journals j LEFT JOIN wk_custody.entries e ON e.journal=j.id GROUP BY j.id HAVING count(e.position)=0 OR sum(e.debit-e.credit)<>0 LIMIT 1`); if (bad.rowCount) throw new CustodyIntegrityError('unbalanced-journal'); diff --git a/server/market/sale-rehearsal.ts b/server/market/sale-rehearsal.ts index 1447552..205effe 100755 --- a/server/market/sale-rehearsal.ts +++ b/server/market/sale-rehearsal.ts @@ -81,4 +81,4 @@ export function replayReservedSale(order: CustodyOrder, custodianId: string, raw const previous = order.status === 'settled' ? 'settled' : order.status === 'cancelled' ? 'refunded' : 'reserved'; return { input, binding, reservationDigest, stateRoot: saleStateRoot(state, context), decision: saleDecision(state, terms, context, previous), context }; -} \ No newline at end of file +} diff --git a/tests/postgres/custody.test.ts b/tests/postgres/custody.test.ts index 11f4533..2aee904 100755 --- a/tests/postgres/custody.test.ts +++ b/tests/postgres/custody.test.ts @@ -21,12 +21,17 @@ import { createBackup } from '../../server/recovery.ts'; import { CUSTODY_POLICY, custodyMessage, type CustodyAction, type CustodyCommand, type CustodyActivity, type CustodyListing, type CustodyOrder, type CustodyWithdrawal } from '../../shared/custody.ts'; import { NETWORK } from '../../shared/network.ts'; -import { CHAIN_ID, address } from '../../server/testnet/consensus.ts'; +import { CHAIN_ID, GENESIS_HASH, address, sha256 } from '../../server/testnet/consensus.ts'; import { saleContext, termsForCustodyOrder, saleId } from '../../server/testnet/sale-candidate.ts'; import { SALE_CANDIDATE, saleTermsMessage, saleCompletionMessage, saleCancellationMessage, type SaleTransaction } from '../../shared/sale-candidate.ts'; 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 { Ledger } from '../../server/testnet/ledger.ts'; +import { GENESIS_TIME, makeTemplate, miningBlob, mintId, transition } from '../../server/testnet/consensus.ts'; +import { deriveStats } from '../../shared/enrichment.ts'; +import { blockClaimMessage, type NftStateView, type TestnetBlock } from '../../shared/testnet.ts'; if (process.env.WIKIDECK_PG_DISPOSABLE_TEST !== '1') throw new Error('PostgreSQL tests require the isolated disposable compose fixture'); const config: PoolConfig = { host: '127.0.0.1', database: 'custody_test', user: 'postgres', ssl: false, max: 4, @@ -67,6 +72,65 @@ function saleTranscript(order: CustodyOrder, custodianId: string, feeBps = 250) hostAuthorization: authorization(host, saleCancellationMessage(reference)), buyerAuthorization: authorization(buyer, saleCancellationMessage(reference)) }; return { input, complete, cancel, reservationDigest }; } +function canonicalPage(streamId: string, all: CanonicalInboxPage['events'], start: number, count: number): CanonicalInboxPage { + const events = all.slice(start, start + count), cursor = events.at(-1)?.sequence ?? start; + const previousAnchor = start === 0 ? GENESIS_HASH : all[start - 1].eventHash; + const anchor = events.at(-1)?.eventHash ?? previousAnchor; + return { streamId, previousAnchor, anchor, events, cursor, head: all.length, caughtUp: cursor === all.length }; +} +function canonicalEvents(streamId: string) { + const rows: CanonicalInboxPage['events'] = []; + let anchor = GENESIS_HASH; + const add = (type: 'connect' | 'disconnect', height: number, hash: string, previousHash: string, transactionIds: string[] = []) => { + const sequence = rows.length + 1; + const eventHash = sha256(JSON.stringify(['wikideck-canonical-event-v1', streamId, sequence, type, + height, hash, previousHash, transactionIds, anchor])); + rows.push({ sequence, type, height, hash, previousHash, transactionIds, eventHash }); anchor = eventHash; + }; + const a = 'a'.repeat(64), b = 'b'.repeat(64), c = 'c'.repeat(64); + add('connect', 1, a, GENESIS_HASH, ['1'.repeat(64)]); + add('disconnect', 1, a, GENESIS_HASH, ['1'.repeat(64)]); + add('connect', 1, b, GENESIS_HASH, ['2'.repeat(64)]); + add('connect', 2, c, b, ['3'.repeat(64)]); + return rows; +} +function reanchorCanonicalEvents(streamId: string, rows: CanonicalInboxPage['events']) { + let anchor = GENESIS_HASH; + for (const event of rows) { + event.eventHash = sha256(JSON.stringify(['wikideck-canonical-event-v1', streamId, event.sequence, event.type, + event.height, event.hash, event.previousHash, event.transactionIds, anchor])); + anchor = event.eventHash; + } +} +class FixturePow { + async hash(key: string, blob: Uint8Array): Promise { return sha256(Buffer.concat([Buffer.from(key, 'hex'), Buffer.from(blob)])); } + close(): void {} +} +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}`, + revisionId: 1, source: 'live' as const, category: 'Culture' }; +} +function fixtureStats() { + const pageBytes = 2048, languageCount = 3; + return { capturedAt: '2026-10-19T00:00:00.000Z', revisionId: 1, pageBytes, languageCount, + 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) { + 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(); + for (header.nonce = 0; header.nonce < 100000; header.nonce++) { + const blob = miningBlob(header), powHash = await pow.hash(header.key, blob); + if (BigInt(`0x${powHash}`) > BigInt(`0x${header.target}`)) continue; + const hash = sha256(blob); + return { ...header, hash, tokenId: mintId(header), powHash, + signature: miner.sign(blockClaimMessage(hash)), transactions: [] } satisfies TestnetBlock; + } + throw new Error('Canonical fixture nonce search exhausted'); +} const options = { feeBps: 250, card: (token: string, owner: string, sequence: number) => token === hash && owner === seller.actor && sequence === 0 ? { title: 'PostgreSQL fixture' } : undefined }; function envelope(id: string, who: typeof buyer, action: CustodyAction, nonce: string = randomUUID()): CustodyCommand { @@ -422,6 +486,7 @@ test('sale rehearsal requires explicit disposable instance consent and exact tra const o = await f.execute(buyer, { type: 'reserve', listingId: l.id, priceAtomicXmr: l.priceAtomicXmr, feeBps: 250 }) as CustodyOrder; const t1 = saleTranscript(o, f.market.id); assert.throws(() => f.market.observeSaleRehearsal(o.id, t1.input), /unavailable/); + assert.throws(() => f.market.observeCanonicalEvents({}), /unavailable/); const rehearsal = await PostgresCustody.open(f.runtime, { ...options, disposableSaleRehearsal: true }); try { const wrong = structuredClone(t1.input); wrong.lock = { ...t1.input.lock, terms: { ...t1.input.lock.terms, priceAtomicXmr: '1' } } as typeof wrong.lock; @@ -440,6 +505,101 @@ test('sale rehearsal requires explicit disposable instance consent and exact tra } finally { await rehearsal.close(); } }); +test('canonical connect/disconnect inbox resumes bounded pages, retries idempotently, and audits after reconnect', async t => { + const f = await fixture(t); + const observer = await PostgresCustody.open(f.runtime, { ...options, disposableSaleRehearsal: true }); + const streamId = randomUUID(), events = canonicalEvents(streamId); + const first = canonicalPage(streamId, events, 0, 2), second = canonicalPage(streamId, events, 2, 2); + try { + const accepted = await observer.observeCanonicalEvents(first); + assert.equal(accepted.cursor, 2); assert.equal(accepted.height, 0); assert.equal(accepted.tip, GENESIS_HASH); + assert.equal((await observer.observeCanonicalEvents(first)).replayed, true); + // A second consumer may have fetched a larger page before the first writer advanced the inbox. + const overlapping = await observer.observeCanonicalEvents(canonicalPage(streamId, events, 0, 4)); + assert.equal(overlapping.cursor, 4); assert.equal(overlapping.replayed, false); + const stale = await observer.observeCanonicalEvents({ ...first, head: 2, caughtUp: true }); + assert.equal(stale.cursor, 4); assert.equal(stale.head, 4); assert.equal(stale.replayed, true); + const tip = await observer.observeCanonicalEvents(second); + assert.equal(tip.cursor, 4); assert.equal(tip.head, 4); assert.equal(tip.height, 2); assert.equal(tip.tip, 'c'.repeat(64)); + assert.equal(tip.replayed, true); + 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, 4); + await observer.audit(); + } finally { await observer.close(); } + const reopened = await PostgresCustody.open(f.runtime, { ...options, disposableSaleRehearsal: true }); + try { + const replay = await reopened.observeCanonicalEvents(second); + assert.equal(replay.cursor, 4); assert.equal(replay.height, 2); assert.equal(replay.replayed, true); + await reopened.audit(); + } finally { await reopened.close(); } +}); + +test('gated adapter consumes actual validated local Ledger connect and reorg pages', async t => { + const dir = await mkdtemp(join(tmpdir(), 'wikideck-pg-local-ledger-')); + const pow = new FixturePow(), 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 canonical = await fixtureBlock([], {}); + await ledger.submit(canonical); + const first = await observer.observeLedgerCanonicalEvents(ledger); + assert.equal(first.cursor, 1); assert.equal(first.height, 1); assert.equal(first.tip, canonical.hash); + const side = await fixtureBlock([], {}, 50001); + const sideResult = await ledger.submit(side); // Equal-work branch remains noncanonical and emits no event. + assert.equal(sideResult.accepted, true); + assert.equal(ledger.canonicalHead(), 1); + const sideChain = [side]; let sideState = transition({}, side, []), sideChild = side; + let reorgResult = { accepted: true, reorg: false }; + for (let i = 0; i < 16 && !reorgResult.reorg; i++) { + sideChild = await fixtureBlock(sideChain, sideState, 50002 + i); + reorgResult = await ledger.submit(sideChild); + sideState = transition(sideState, sideChild, []); sideChain.push(sideChild); + } + assert.equal(reorgResult.reorg, true, `Expected longer-work side branch to win; events=${JSON.stringify(ledger.canonicalEvents(0))}`); + const expectedCursor = sideChain.length + 2; // initial connect, old-tip disconnect, then each side-branch connect + assert.equal(ledger.canonicalHead(), expectedCursor); + const reorg = await observer.observeLedgerCanonicalEvents(ledger); + assert.equal(reorg.cursor, expectedCursor); assert.equal(reorg.height, sideChain.length); assert.equal(reorg.tip, sideChild.hash); + 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, expectedCursor); + await observer.audit(); +}); + +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); + try { + events[1].previousHash = '9'.repeat(64); + reanchorCanonicalEvents(streamId, events); + await assert.rejects(observer.observeCanonicalEvents(canonicalPage(streamId, events, 0, 2)), /disconnect-ancestry-mismatch/); + 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, 0); + await assert.rejects(PostgresCustody.open(f.runtime, options), /unresolved-integrity/); + } finally { await observer.close(); } +}); + +test('canonical inbox rejects tampered event anchors, persists an integrity hold and refuses another stream', async t => { + const f = await fixture(t), observer = await PostgresCustody.open(f.runtime, { ...options, disposableSaleRehearsal: true }); + const streamId = randomUUID(), events = canonicalEvents(streamId); + try { + const tampered = canonicalPage(streamId, events, 0, 2); + tampered.events[0].transactionIds = ['f'.repeat(64)]; + await assert.rejects(observer.observeCanonicalEvents(tampered), /canonical-inbox-event-hash-invalid/); + 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, 0); + await assert.rejects(PostgresCustody.open(f.runtime, options), /unresolved-integrity/); + } finally { await observer.close(); } +}); + +test('canonical inbox startup audit detects owner-side history mutation and durably fences writers', async t => { + const f = await fixture(t), observer = await PostgresCustody.open(f.runtime, { ...options, disposableSaleRehearsal: true }); + const streamId = randomUUID(), events = canonicalEvents(streamId); + try { + await observer.observeCanonicalEvents(canonicalPage(streamId, events, 0, 2)); + const changed = { ...events[0], transactionIds: ['f'.repeat(64)] }; + await f.admin.query('UPDATE wk_custody.custody_meta SET value=$1 WHERE key=$2', + [JSON.stringify(changed), `canonical-inbox:v1:event:${'1'.padStart(16, '0')}`]); + await assert.rejects(observer.audit(), /canonical-inbox-event-anchor-invalid/); + assert.equal((await f.admin.query("SELECT count(*)::int AS n FROM wk_custody.custody_meta WHERE key='integrity_incident_v1'")).rows[0].n, 1); + await assert.rejects(PostgresCustody.open(f.runtime, options), /unresolved-integrity/); + } finally { await observer.close(); } +}); + test('signed candidate completion settles actual PostgreSQL balances once across pools and reconnect', async t => { const f = await fixture(t); await f.execute(buyer, { type: 'mock-credit' }); const l = await f.execute(seller, { type: 'list', cardHash: hash, sequence: 0, priceAtomicXmr: '100000000000', feeBps: 250 }) as CustodyListing;