From e48081f0412648d74e72e4e6906902aeb5c53ec2 Mon Sep 17 00:00:00 2001 From: Eric Meehan Date: Tue, 20 Oct 2026 14:35:45 -0400 Subject: [PATCH] Canonical indexes and recoverable market observation --- docs/canonical-indexes.md | 49 +++++++ docs/production-plan.md | 6 +- package.json | 1 + scripts/benchmark-canonical-index.ts | 59 +++++++++ server/market/store.ts | 92 ++++++++++--- server/testnet/canonical-index.ts | 186 +++++++++++++++++++++++++++ server/testnet/http.ts | 30 +++-- server/testnet/ledger.ts | 36 +++--- tests/api.test.ts | 4 + tests/market.test.ts | 82 +++++++++++- tests/testnet-ledger.test.ts | 92 ++++++++++++- 11 files changed, 583 insertions(+), 54 deletions(-) create mode 100755 docs/canonical-indexes.md create mode 100755 scripts/benchmark-canonical-index.ts create mode 100755 server/testnet/canonical-index.ts diff --git a/docs/canonical-indexes.md b/docs/canonical-indexes.md new file mode 100755 index 0000000..0ea2648 --- /dev/null +++ b/docs/canonical-indexes.md @@ -0,0 +1,49 @@ +# M2 increment — canonical indexes and recoverable market observation + +Status: implemented local infrastructure, **not full M2 acceptance or production authorization**. Evidence date: 2026-10-20. Related: [production plan](production-plan.md), [custodial design](custodial-market-design.md). + +## Delivered + +- [Canonical index](../server/testnet/canonical-index.ts) stores current NFT ownership/sequence, canonical mint/block identity and transaction inclusion in SQLite. Owner/mint ordering and transaction indexes replace whole-chain scans for deck, NFT and marketplace queries. +- [Ledger persistence](../server/testnet/ledger.ts) commits accepted blocks, pending transactions, index connect/disconnect undo and canonical events in **one SQLite transaction**. A losing side branch or mempool submission does not create a canonical event. Disconnects run tip-first; connects run ancestor-first. No await separates the database commit from updating the in-memory tip. +- Events contain sequence, connect/disconnect type, block identity/height/parent, transaction IDs and a hash chained to the previous event. An independently generated local stream ID binds the history; it is **not consensus data or a public proof of honest observation**. Pages contain at most 64 events and no article payloads. Financial/chain history is not silently pruned. +- On first open, original blocks are replayed and independently proof/state validated before additive local index backfill. Later opens compare indexed block identity, ownership, transaction inclusion and event continuity/anchors to that validated chain. Inconsistent state fails closed; existing consensus version, genesis, signatures and free-transfer/trade rules are unchanged. +- [Market reconciliation](../server/market/store.ts) uses indexed current delivery/ownership queries and a persisted `(stream ID, sequence, anchor)` checkpoint. Every observed disconnect of a previously completed order causes a durable review hold, even if delivery reappeared before polling. Review is never automatically repaid. Hold updates and checkpoint advancement commit together in the market database. +- Catch-up consumes one bounded event page per reconciliation pass. While behind, ordinary market mutation/payout/expiry decisions stop; diagnostic/read access remains available and exposes event cursor/head/backlog. Periodic polling and subsequent requests advance catch-up. No claim of instant catch-up or background-thread isolation. +- Restore mismatch (different stream, different history anchor or a cursor ahead of the restored ledger) is refused rather than resetting the cursor. Restore the matching stopped-node ledger/market backup; any alternative recovery requires reviewed evidence, not deleting a checkpoint. +- [HTTP NFT lookup and custody card checks](../server/testnet/http.ts) use the same index. Wallet deck pagination accepts an optional `tip` query; stale tips return 409. Existing callers without `tip` remain compatible. There is no new public event/admin endpoint, no payment observer and no NFT-lock signer. + +## Tests and measurements + +The final Docker test-target build passed TypeScript/Vite compilation. All **149 tests passed, zero failures/skips**, including native RandomX multi-process HTTP/peer/restart and backup rehearsal. New failure evidence includes: + +- indexed ownership/sequence versus full replay for transfers, multi-card trades and forks; +- append/disconnect/reconnect ordering, side-branch/mempool exclusion, bounded defensive-copy pages and durable restart; +- injected event-write failure rolling back blocks, indexes and mempool together; +- additive migration of a prior-schema copy, refusing corrupt ownership and event records; +- production market adapter exercised with full snapshots forbidden; +- multi-page catch-up and persisted restart checkpoint; +- injected cursor-write failure rolling back the payout hold, followed by restart recovery; +- numeric cursor reuse with a different historical hash rejected; +- real HTTP NFT results and stale-tip deck pagination. + +[Benchmark script](../scripts/benchmark-canonical-index.ts), available as `npm run benchmark:index -- 10000`, creates/removes its own temporary SQL database. Input is bounded to 1–100,000 synthetic blocks; **these are not valid mined blocks** and are never submitted to a ledger/peer. It measures only derived index behavior, not PoW, network throughput or operating capacity. + +Observed in the development Docker environment: + +| Metric | Result | +| --- | --- | +| Fixture | 10,000 NFTs/blocks, 9,998 transfers, 16 owner identities | +| SQL fixture build (single transaction) | 5,096 ms | +| 1,000 query groups (NFT + 24-item owner page + optional transaction lookup), median / p95 | 0.619 / 1.167 ms | +| Read all events, maximum 64 per page | 157 pages, 54.5 ms total | +| Verify derived indexes and anchored history | 524 ms | +| Database plus WAL before close | 76,856,784 bytes | + +These numbers are a reproducible **baseline**, not acceptance SLOs: host CPU/storage and production concurrency were not selected, outputs include synthetic article payloads, warm caches apply, one transaction amortizes insertion fsync, and real validation/startup/HTTP/market workloads are excluded. They do not establish a supported production envelope. + +## Remaining gates + +M2 still needs target-hardware/data/SLO decisions, market-order incremental scheduling rather than scanning every existing bounded order/listing, bounded worker/event-loop latency under load, production financial database/schema and measured recovery/retention/storage growth. Ledger PoW replay and DAG state memory still scale with retained history; indexed queries do not solve those costs. Startup validation checks undo-record counts but not every possible future undo mutation against a second reconstructed undo model; comprehensive corruption/fuzz coverage and independent review remain required. + +Custodial reservations still stop at `awaiting-lock`. This increment does **not** consume ordinary transfers as custodial settlement evidence: doing that would bypass the reviewed sale-lock/cancel protocol. Real Monero deposit/withdrawal integration, production treasury keys, host attestations and a reviewed versioned NFT lock/completion/cancellation specification remain blocked. No real funds, credentials, external processor or genesis change were used. diff --git a/docs/production-plan.md b/docs/production-plan.md index 93c2349..5ffd343 100755 --- a/docs/production-plan.md +++ b/docs/production-plan.md @@ -63,6 +63,8 @@ Acceptance: exact-amount property/boundary tests, versioned signature tests, zer Can run alongside P0/P1. Addresses SEC-01/SEC-06 rather than adding more fixed caps. +**Implemented M2 increment, 2026-10-20:** [canonical indexes and durable market observation](canonical-indexes.md). Canonical ownership/transaction SQL indexes now serve NFT, deck, custody availability and mock-market queries; per-block undo and anchored connect/disconnect events commit with ledger updates. Market checkpoint/reorg holds persist atomically and bounded catch-up blocks new mutations. Final TypeScript/Vite build and 149 offline tests passed. A disposable 10,000-NFT SQL baseline is recorded, **not** a target-hardware SLO or complete M2 acceptance. Production financial storage, incremental order scheduling, retained-DAG capacity and operator dataset/latency targets remain open. + - Indexed canonical ownership, NFT/transfer lookup and paged wallet/explorer APIs: browser signing must not require loading all history. Tip-pinned queries detect reorgs and preserve complete ownership semantics. - Incremental market reconciliation from durable canonical/reorg events. Maintain rollback-safe indexes and a recoverable cursor; catch-up after restart must not miss delivery or payout holds. Bound worker time/concurrency; move expensive work off the request event loop where appropriate. - A versioned production market/settlement schema with append-only financial events and balanced accounting entries for buyer liabilities, host revenue, author liabilities, seller liabilities, paid/refunded amounts and actual network fees. Derived views are rebuildable; financial history and idempotency keys are not discarded to reclaim space. @@ -138,7 +140,7 @@ Acceptance: signed release evidence packet; recorded unresolved accepted risks w First implementation increment: write the M0 threat model/ADR research questions and P1 XMR domain specification, then build exact amount/property tests. Do not spend the next increment adding another disconnected “live” scaffold. Each milestone must integrate with its actual callers, persistence, recovery, UI and failure tests before being described as delivered. -Next increment status: M0 source investigation and [concrete settlement/evidence choices](settlement-feasibility.md) remain open. M1 now has a signed v3 XMR-native mock marketplace path; the public USD-to-XMR rate was removed. Version-2 historical USD orders are preserved and can continue under their original terms; open USD listings are legacy read-only until seller retirement/relisting. This is not live settlement, a completed migration dry-run, or production milestone acceptance; payment and payouts remain mock-only, and live auctions remain a separate release gate. +Current increment status: the operator selected custodial implementation after [settlement research](settlement-feasibility.md). M1 has a signed v3 XMR-native mock path, with historical v2 USD orders preserved. [ADR 002](custodial-market-design.md) records the implemented isolated custody accounting/UI foundation; production custody is not complete. The next M2 infrastructure increment is [canonical indexing and durable reconciliation](canonical-indexes.md), now built and regression-tested. Next engineering gates include the remaining production accounting/worker envelope and the separately reviewed NFT lock/completion/cancellation specification (M3); no live Monero integration or consensus activation is authorized by passing these tests. Migration dry-run, final fee/risk policy, capacity SLOs and independent review remain open. ## 5. Acceptance evidence and release discipline @@ -178,4 +180,4 @@ Not every value is needed to begin engineering; each becomes a blocking decision - [Resource controls](node-resources.md), [wallet lifecycle](wallet-security.md), [calibration evidence](consensus-calibration.md), [operations/recovery](operations-recovery.md). - [Offline SHKeeper contract/state-machine work](shkeeper-settlement.md) — reusable tests/ideas, not the selected production architecture. -This plan supersedes the earlier **production pricing direction** of USD listings, but does not claim the current USD/mock runtime has already migrated. Keep historical evidence/versioned terms intact; update live-facing copy when the integrated XMR-native implementation is actually delivered. +This plan supersedes the earlier USD production-pricing direction. New mock listings already use native XMR; historical USD terms remain intact. Custody and M2 implementation status are explicitly bounded above; no sandbox or infrastructure milestone implies live settlement or launch readiness. diff --git a/package.json b/package.json index 5aa738a..2e998f8 100644 --- a/package.json +++ b/package.json @@ -11,6 +11,7 @@ "test:rpc-load": "node --experimental-strip-types scripts/rpc-load.ts", "analyze:consensus": "node --experimental-strip-types scripts/consensus-analysis.ts", "benchmark:randomx": "node --experimental-strip-types scripts/benchmark-randomx.ts", + "benchmark:index": "node --experimental-strip-types scripts/benchmark-canonical-index.ts", "node:backup": "node --experimental-strip-types scripts/node-backup.ts", "ops:check": "node --experimental-strip-types scripts/ops-check.ts", "miner": "tsx server/testnet/miner.ts", diff --git a/scripts/benchmark-canonical-index.ts b/scripts/benchmark-canonical-index.ts new file mode 100755 index 0000000..ee01d58 --- /dev/null +++ b/scripts/benchmark-canonical-index.ts @@ -0,0 +1,59 @@ +import { DatabaseSync } from 'node:sqlite'; +import { mkdtemp, rm, stat } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { performance } from 'node:perf_hooks'; +import { CanonicalIndex } from '../server/testnet/canonical-index.ts'; +import { CHAIN_ID, GENESIS_HASH, sha256, transactionId } from '../server/testnet/consensus.ts'; +import { NETWORK, type TestnetBlock, type NftState, type NftTransfer } from '../shared/testnet.ts'; + +// Synthetic LOCAL SQL dataset only. Not valid mined blocks; never submitted to Ledger or a peer. +// This measures derived-index costs, not RandomX validation, consensus security or production SLOs. +const text = process.argv[2] ?? '10000'; +if (!/^[1-9]\d{0,5}$/.test(text) || Number(text) > 100000) throw new Error('Fixture size must be 1..100000'); +const count = Number(text), queries = 1000; +const directory = await mkdtemp(join(tmpdir(), 'wikideck-index-benchmark-')); +const file = join(directory, 'fixture.sqlite'), db = new DatabaseSync(file); +try { + db.exec('PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL; PRAGMA foreign_keys=ON; CREATE TABLE blocks(hash TEXT PRIMARY KEY,block_json TEXT NOT NULL)'); + const index = new CanonicalIndex(db), chain: TestnetBlock[] = [], state: NftState = Object.create(null); + index.initialize([], state); + const owners = Array.from({ length: 16 }, (_, i) => sha256(`fixture-owner-${i}`)); + const insert = db.prepare('INSERT INTO blocks VALUES(?,?)'); + const start = performance.now(); + db.exec('BEGIN IMMEDIATE'); + try { + for (let height = 1; height <= count; height++) { + const hash = sha256(`fixture-block-${height}`), token = sha256(`fixture-token-${height}`), owner = owners[height % owners.length]; + const txs: NftTransfer[] = height > 2 ? [{ network: NETWORK, chainId: CHAIN_ID, cardHash: chain[height - 3].tokenId, + from: state[chain[height - 3].tokenId].owner, to: owner, sequence: 0, publicKey: '', signature: '' }] : []; + const block = { height, hash, tokenId: token, previousHash: chain.at(-1)?.hash ?? GENESIS_HASH, owner, + article: { title: `Synthetic index fixture ${height}`, pageId: height, extract: 'x'.repeat(512) }, transactions: txs } as unknown as TestnetBlock; + state[token] = { owner, sequence: 0, mintedHeight: height, pageId: height }; + for (const tx of txs) state[tx.cardHash] = { ...state[tx.cardHash], owner: tx.to, sequence: 1 }; + insert.run(hash, JSON.stringify(block)); index.connect(block); chain.push(block); + } + db.exec('COMMIT'); + } catch (error) { db.exec('ROLLBACK'); throw error; } + const buildMs = performance.now() - start; + const latency: number[] = []; + for (let i = 0; i < queries; i++) { + const block = chain[(i * 7919) % count], before = performance.now(); + const nft = index.nft(block.tokenId, count); + if (nft?.sequence !== state[block.tokenId].sequence || nft.owner !== state[block.tokenId].owner) throw new Error('Fixture ownership mismatch'); + index.deck(owners[i % owners.length], 0, 24, count); + if (block.transactions.length && !index.transaction(transactionId(block.transactions[0]))) throw new Error('Fixture transaction missing'); + latency.push(performance.now() - before); + } + let cursor = 0, batches = 0; + const consumeStart = performance.now(); + do { const page = index.events(cursor, 64); cursor = page.cursor; batches++; } while (cursor < index.head()); + const eventReadMs = performance.now() - consumeStart; + const verifyStart = performance.now(); index.initialize(chain, state); const verifyMs = performance.now() - verifyStart; + const bytes = (await stat(file)).size + (await stat(file + '-wal')).size; + latency.sort((a, b) => a - b); + console.log(JSON.stringify({ fixture: 'synthetic-derived-index-only', blocks: count, nfts: count, + transactions: Math.max(0, count - 2), queries, buildMs, lookupGroupP50Ms: latency[499], lookupGroupP95Ms: latency[949], + eventBatches: batches, maxEventsPerBatch: 64, eventReadMs, startupVerifyMs: verifyMs, databaseAndWalBytes: bytes, + warning: 'No PoW, HTTP load, real assets, target hardware or production SLO certification.' }, null, 2)); +} finally { db.close(); await rm(directory, { recursive: true, force: true }); } diff --git a/server/market/store.ts b/server/market/store.ts index 260aba1..96dd97d 100755 --- a/server/market/store.ts +++ b/server/market/store.ts @@ -11,8 +11,17 @@ import { isXmrListing, isXmrOrder, marketCommandV3Schema, marketMessageV3, type MarketOrderRecord, type MarketOrderV3 } from '../../shared/market-v3.ts'; import { CHAIN_ID, transactionId, transferSchema, verifySignature } from '../testnet/consensus.ts'; import { mockQuote, mockXmrQuote, paymentMode } from './payments.ts'; +import { RpcError } from '../resources.ts'; +import type { CanonicalEvents, IndexedNft } from '../testnet/canonical-index.ts'; +import { z } from 'zod'; -type Chain = { snapshot(): TestnetSnapshot; summary?(): { tip: string; revision: number }; submitTransfer(raw: unknown): Promise }; +type Chain = { snapshot(): TestnetSnapshot; summary?(): { tip: string; revision: number; height?: number }; + submitTransfer(raw: unknown): Promise; nft?(token: string): IndexedNft | undefined; + canonicalTransaction?(id: string): { height: number; blockHash: string } | undefined; + canonicalHead?(): number; canonicalEvents?(after: number, limit?: number): CanonicalEvents; + pendingTransactions?(): TestnetSnapshot['pendingTransfers'] }; +type ChainView = { height: number; pendingTransfers: TestnetSnapshot['pendingTransfers']; + nft(token: string): IndexedNft | undefined; transaction(id: string): { height: number; blockHash: string } | undefined }; type Config = { feeBps: number; confirmations: number }; export function marketConfig(env: NodeJS.ProcessEnv = process.env): Config { paymentMode(env); @@ -28,12 +37,17 @@ export function marketConfig(env: NodeJS.ProcessEnv = process.env): Config { } const PAYMENT_WINDOW = 30 * 60_000; const DELIVERY_WINDOW = 48 * 60 * 60_000; +const cursorSchema = z.object({ version: z.literal(1), streamId: z.string().uuid(), + cursor: z.number().int().nonnegative().safe(), anchor: z.string().regex(/^[a-f0-9]{64}$/) }).strict(); /** Node-local marketplace, NOT consensus escrow. Its SQLite journal is backed up with the ledger. */ export class Market { private queue: Promise = Promise.resolve(); private queued = 0; private reconciled: { tip: string; revision: number; at: number } | undefined; + private eventCursor = 0; + private eventCheckpoint: z.infer | undefined; + private eventBacklog = false; private readonly db: DatabaseSync; private readonly chain: Chain; readonly config: Config; @@ -41,6 +55,11 @@ export class Market { private readonly now: () => number; private constructor(db: DatabaseSync, chain: Chain, config: Config, id: string, now: () => number) { this.db = db; this.chain = chain; this.config = config; this.id = id; this.now = now; + const cursor = db.prepare("SELECT value FROM market_meta WHERE key='canonical_event_cursor'").get() as { value: string } | undefined; + if (cursor) { + this.eventCheckpoint = cursorSchema.parse(JSON.parse(cursor.value)); + this.eventCursor = this.eventCheckpoint.cursor; + } } static async open(directory: string, chain: Chain, config = marketConfig(), now = Date.now): Promise { @@ -77,7 +96,8 @@ export class Market { close() { this.db.close(); } diagnostics() { const states=this.db.prepare("SELECT json_extract(data,'$.status') AS status,COUNT(*) AS count FROM orders GROUP BY json_extract(data,'$.status')").all(); - return { queued:this.queued,lastReconciledAt:this.reconciled?.at??null,orderStates:states }; + return { queued:this.queued,lastReconciledAt:this.reconciled?.at??null,orderStates:states, + canonicalEventCursor:this.eventCursor,canonicalEventHead:this.chain.canonicalHead?.()??null,eventBacklog:this.eventBacklog }; } info(): MarketInfo { return { network: NETWORK, chainId: CHAIN_ID, marketId: this.id, paymentMode: 'mock', marketVersion: 3, feeBps: this.config.feeBps, @@ -110,11 +130,25 @@ export class Market { try { const result = fn(); this.db.exec('COMMIT'); return result; } catch (error) { this.db.exec('ROLLBACK'); throw error; } } - private card(snapshot: TestnetSnapshot, hash: string, owner: string, sequence: number): boolean { - const block = snapshot.blocks.find(b => b.tokenId === hash); - return !!block && testnetOwnerOf(snapshot, hash) === owner && - snapshot.transfers.filter(tx => transactionCardIds(tx).includes(hash)).length === sequence && - snapshot.blocks.length + 1 - block.height >= CONSENSUS.coinbaseMaturity; + private view(): ChainView { + if (this.chain.nft && this.chain.canonicalTransaction && this.chain.pendingTransactions && this.chain.summary) { + const height = this.chain.summary().height; + if (height === undefined) throw new Error('Indexed chain height missing'); + return { height, pendingTransfers: this.chain.pendingTransactions(), nft: token => this.chain.nft!(token), + transaction: id => this.chain.canonicalTransaction!(id) }; + } + // Compatibility boundary for unit-test/legacy adapters only. The actual node always supplies indexed queries. + const s = this.chain.snapshot(); + const txs = new Map(s.blocks.flatMap(b => b.transactions.map(tx => [transactionId(tx), { height: b.height, blockHash: b.hash }] as const))); + return { height: s.blocks.length, pendingTransfers: s.pendingTransfers, transaction: id => txs.get(id), nft: token => { + const block = s.blocks.find(b => b.tokenId === token); + return block ? { block, owner: testnetOwnerOf(s, token)!, sequence: s.transfers.filter(tx => transactionCardIds(tx).includes(token)).length, + confirmations: s.blocks.length + 1 - block.height } : undefined; + } }; + } + private card(view: ChainView, hash: string, owner: string, sequence: number): boolean { + const nft = view.nft(hash); + return !!nft && nft.owner === owner && nft.sequence === sequence && nft.confirmations >= CONSENSUS.coinbaseMaturity; } private deliveryId(order: MarketOrderRecord): string { // Identity excludes signature bytes. Recognize exact sale delivery through ANY submission route, @@ -124,17 +158,34 @@ export class Market { } /** Reconcile from canonical transaction identity, never from a mempool acknowledgement. */ private reconcile() { - const s = this.chain.snapshot(), now = this.now(); - const deliveries = new Map(); - for (const b of s.blocks) for (const tx of b.transactions) deliveries.set(transactionId(tx), b); + const events = this.chain.canonicalEvents?.(this.eventCursor, 64); + if (events && this.eventCheckpoint && (events.streamId !== this.eventCheckpoint.streamId || + events.previousAnchor !== this.eventCheckpoint.anchor)) throw new RpcError(409, 'Canonical event history changed; restore matching ledger and market backups.'); + const s = this.view(), now = this.now(); const pendingIds = new Set(s.pendingTransfers.map(transactionId)); this.atomic(() => { + // Process every disconnect, even if the same delivery reappeared before this consumer caught up. + // Never infer that a current tip erases an intermediate loss of previously settled evidence. + const disconnected = new Set(events?.events.filter(e => e.type === 'disconnect').flatMap(e => e.transactionIds) ?? []); + if (events) { + for (const o of this.rows('orders')) { + if (o.status === 'complete' && disconnected.has(this.deliveryId(o))) { + o.status = 'review'; o.payout = 'review'; + if (o.additionalFee) o.additionalFee.payout = 'review'; + this.saveOrder(o); + } + } + this.db.prepare('INSERT INTO market_meta VALUES(?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value') + .run('canonical_event_cursor', JSON.stringify({ version: 1, streamId: events.streamId, cursor: events.cursor, anchor: events.anchor })); + // Catch-up is bounded per pass. No new payout, timeout/refund decision or mutation while events remain. + if (!events.caughtUp) return; + } for (const o of this.rows('orders')) { const l = this.listing(o.listingId); if (o.status === 'refunded' || o.status === 'expired') continue; const deliveryId = this.deliveryId(o); - const delivery = deliveries.get(deliveryId); - const confirmations = delivery ? s.blocks.length - delivery.height + 1 : 0; + const delivery = s.transaction(deliveryId); + const confirmations = delivery ? s.height - delivery.height + 1 : 0; o.deliveryConfirmations = confirmations; if (o.status === 'complete' || o.status === 'review') { if (!delivery || confirmations < o.requiredConfirmations) { @@ -174,8 +225,12 @@ export class Market { } this.db.prepare('DELETE FROM receipts WHERE expires < ?').run(now - 86400_000); }); + if (events) { + this.eventCursor = events.cursor; this.eventBacklog = !events.caughtUp; + this.eventCheckpoint = { version: 1, streamId: events.streamId, cursor: events.cursor, anchor: events.anchor }; + } const summary = this.chain.summary?.(); - if (summary) this.reconciled = { ...summary, at: now }; + if (summary && !this.eventBacklog) this.reconciled = { ...summary, at: now }; } browse(offset = 0): { listings: MarketListingRecord[]; next: number | null; info: MarketInfo } { if (!Number.isSafeInteger(offset) || offset < 0 || offset > 10000) throw new Error('Invalid marketplace page'); @@ -215,6 +270,7 @@ export class Market { const summary = this.chain.summary?.(), last = this.reconciled; if (a.type !== 'activity' || !summary || !last || summary.tip !== last.tip || summary.revision !== last.revision || this.now() - last.at >= 1000) this.reconcile(); + if (this.eventBacklog && a.type !== 'activity') throw new RpcError(503, 'Market canonical reconciliation is catching up; retry shortly.'); if (a.type === 'activity') return { orders: (this.db.prepare("SELECT data FROM orders WHERE json_extract(data,'$.seller')=? OR json_extract(data,'$.buyer')=? ORDER BY rowid DESC") .all(c.actor,c.actor) as { data: string }[]).map(r=>JSON.parse(r.data) as MarketOrderRecord), @@ -240,7 +296,7 @@ export class Market { return value; } private mutateV2(c: MarketCommand): MarketActivityV3 | MarketListingRecord | MarketOrderRecord { - const a = c.action, now = this.now(), s = this.chain.snapshot(); + const a = c.action, now = this.now(), s = this.view(); if (a.type === 'list') { throw new Error('USD listing terms are retired; create a new native-XMR listing'); } @@ -260,7 +316,7 @@ export class Market { o.status = 'paid'; o.deliverBy = now + DELIVERY_WINDOW; } else { if (o.status !== 'refund-required' || s.pendingTransfers.some(tx => transactionCardIds(tx).includes(o.cardHash)) || - s.blocks.some(b => b.transactions.some(tx => transactionId(tx) === this.deliveryId(o)))) throw new Error('Refund requires failed delivery and no pending/canonical sale transfer'); + s.transaction(this.deliveryId(o))) throw new Error('Refund requires failed delivery and no pending/canonical sale transfer'); o.status = 'refunded'; const l = this.listing(o.listingId); l.status = 'cancelled'; this.saveListing(l); } this.saveOrder(o); return o; @@ -275,7 +331,7 @@ export class Market { } private mutateV3(c: MarketCommandV3): MarketActivityV3 | MarketListingRecord | MarketOrderRecord { - const a = c.action, now = this.now(), s = this.chain.snapshot(); + const a = c.action, now = this.now(), s = this.view(); if (a.type === 'list') { if (a.feeBps !== this.config.feeBps) throw new Error('Market fee changed; review and sign again'); if (!this.card(s, a.cardHash, c.actor, a.sequence) || s.pendingTransfers.some(tx => transactionCardIds(tx).includes(a.cardHash))) throw new Error('Card is missing, stale, immature, or already pending'); @@ -283,7 +339,7 @@ export class Market { if (this.rows('listings').filter(l => l.seller === c.actor && ['open','reserved'].includes(l.status)).length >= 100) throw new Error('Limit: 100 active listings per seller'); if ((this.db.prepare('SELECT COUNT(*) AS n FROM listings').get() as { n: number }).n >= 10000) throw new Error('Marketplace journal limit reached; operator archival required'); if (this.db.prepare("SELECT 1 FROM listings WHERE card=? AND status IN ('open','reserved')").get(a.cardHash)) throw new Error('Card already listed'); - const mint = s.blocks.find(b => b.tokenId === a.cardHash)!; + const mint = s.nft(a.cardHash)!.block; const l: MarketListingV3 = { id: randomUUID(), termsVersion: 3, cardHash: a.cardHash, sequence: a.sequence, seller: c.actor, title: mint.article.title, image: mint.article.image!, kind: a.kind, priceAtomicXmr: a.priceAtomicXmr, feeBps: a.feeBps, additionalFee: { ...ADDITIONAL_MARKET_FEE }, @@ -337,7 +393,7 @@ export class Market { order.status = 'paid'; order.deliverBy = now + DELIVERY_WINDOW; } else { if (order.status !== 'refund-required' || s.pendingTransfers.some(tx => transactionCardIds(tx).includes(order.cardHash)) || - s.blocks.some(b => b.transactions.some(tx => transactionId(tx) === this.deliveryId(order)))) throw new Error('Refund requires failed delivery and no pending/canonical sale transfer'); + s.transaction(this.deliveryId(order))) throw new Error('Refund requires failed delivery and no pending/canonical sale transfer'); order.status = 'refunded'; const listing = this.listing(order.listingId); listing.status = 'cancelled'; this.saveListing(listing); } this.saveOrder(order); return order; diff --git a/server/testnet/canonical-index.ts b/server/testnet/canonical-index.ts new file mode 100755 index 0000000..4460fe8 --- /dev/null +++ b/server/testnet/canonical-index.ts @@ -0,0 +1,186 @@ +import type { DatabaseSync } from 'node:sqlite'; +import { randomUUID } from 'node:crypto'; +import { isNftTrade, transactionCardIds, type NftState, type TestnetBlock } from '../../shared/testnet.ts'; +import { GENESIS_HASH, sha256, transactionId } from './consensus.ts'; +import { RpcError } from '../resources.ts'; + +export type CanonicalEvent = { sequence: number; type: 'connect' | 'disconnect'; + height: number; hash: string; previousHash: string; transactionIds: string[]; eventHash: string }; +export type CanonicalEvents = { streamId: string; previousAnchor: string; anchor: string; + events: CanonicalEvent[]; cursor: number; head: number; caughtUp: boolean }; +type NftRow = { token: string; owner: string; sequence: number; minted_height: number; block_hash: string }; +export type IndexedNft = { block: TestnetBlock; owner: string; sequence: number; confirmations: number }; + +/** Local derived indexes, NOT consensus. Writes share the ledger's block/pending transaction. */ +export class CanonicalIndex { + private readonly db: DatabaseSync; + private streamId = ''; + constructor(db: DatabaseSync) { this.db = db; } + + initialize(chain: TestnetBlock[], state: NftState): void { + this.db.exec('BEGIN IMMEDIATE'); + try { + this.db.exec(`CREATE TABLE IF NOT EXISTS canonical_meta(key TEXT PRIMARY KEY,value TEXT NOT NULL); + CREATE TABLE IF NOT EXISTS canonical_blocks(height INTEGER PRIMARY KEY,hash TEXT NOT NULL UNIQUE REFERENCES blocks(hash)); + CREATE TABLE IF NOT EXISTS canonical_nfts(token TEXT PRIMARY KEY,owner TEXT NOT NULL,sequence INTEGER NOT NULL, + minted_height INTEGER NOT NULL,block_hash TEXT NOT NULL REFERENCES blocks(hash)); + CREATE INDEX IF NOT EXISTS canonical_owner ON canonical_nfts(owner,minted_height,token); + CREATE TABLE IF NOT EXISTS canonical_transactions(id TEXT PRIMARY KEY,height INTEGER NOT NULL,block_hash TEXT NOT NULL); + CREATE INDEX IF NOT EXISTS canonical_tx_height ON canonical_transactions(height); + CREATE TABLE IF NOT EXISTS canonical_undo(height INTEGER PRIMARY KEY,data TEXT NOT NULL); + CREATE TABLE IF NOT EXISTS canonical_events(sequence INTEGER PRIMARY KEY AUTOINCREMENT,type TEXT NOT NULL, + height INTEGER NOT NULL,hash TEXT NOT NULL,previous_hash TEXT NOT NULL,transactions_json TEXT NOT NULL, + previous_event_hash TEXT NOT NULL,event_hash TEXT NOT NULL); + CREATE TRIGGER IF NOT EXISTS canonical_events_no_update BEFORE UPDATE ON canonical_events + BEGIN SELECT RAISE(ABORT,'Immutable canonical event'); END; + CREATE TRIGGER IF NOT EXISTS canonical_events_no_delete BEFORE DELETE ON canonical_events + BEGIN SELECT RAISE(ABORT,'Immutable canonical event'); END;`); + const version = this.db.prepare("SELECT value FROM canonical_meta WHERE key='version'").get() as { value: string } | undefined; + if (!version) { + // Additive backfill only after Ledger.open has independently validated all persisted proofs/state. + const count = this.db.prepare('SELECT COUNT(*) AS n FROM canonical_events').get() as { n: number }; + if (count.n || (this.db.prepare('SELECT COUNT(*) AS n FROM canonical_blocks').get() as { n: number }).n) { + throw new Error('Partial canonical index; offline operator review required'); + } + this.streamId = randomUUID(); + this.db.prepare('INSERT INTO canonical_meta VALUES(?,?)').run('stream_id', this.streamId); + for (const block of chain) this.connect(block); + this.db.prepare('INSERT INTO canonical_meta VALUES(?,?)').run('version', '1'); + } else if (version.value !== '1') throw new Error('Canonical index version mismatch'); + else { + const row = this.db.prepare("SELECT value FROM canonical_meta WHERE key='stream_id'").get() as { value: string } | undefined; + if (!row || !/^[a-f0-9]{8}(-[a-f0-9]{4}){3}-[a-f0-9]{12}$/.test(row.value)) throw new Error('Canonical stream identity missing'); + this.streamId = row.value; + } + this.verify(chain, state); + this.db.exec('COMMIT'); + } catch (error) { this.db.exec('ROLLBACK'); throw error; } + } + private verify(chain: TestnetBlock[], state: NftState): void { + const blocks = this.db.prepare('SELECT height,hash FROM canonical_blocks ORDER BY height').all() as { height: number; hash: string }[]; + if (blocks.length !== chain.length || blocks.some((b, i) => b.height !== chain[i].height || b.hash !== chain[i].hash)) { + throw new Error('Canonical block index mismatch'); + } + const nfts = this.db.prepare('SELECT * FROM canonical_nfts').all() as NftRow[]; + const expected = new Map(chain.map(b => [b.tokenId, b])); + if (nfts.length !== expected.size) throw new Error('Canonical NFT index mismatch'); + for (const row of nfts) { + const nft = state[row.token], block = expected.get(row.token); + if (!nft || !block || row.owner !== nft.owner || row.sequence !== nft.sequence || + row.minted_height !== block.height || row.block_hash !== block.hash) throw new Error('Canonical ownership index mismatch'); + } + const transactions = this.db.prepare('SELECT id,height,block_hash FROM canonical_transactions').all() as { id: string; height: number; block_hash: string }[]; + const expectedTxs = new Map(chain.flatMap(b => b.transactions.map(tx => [transactionId(tx), b] as const))); + if (transactions.length !== expectedTxs.size || transactions.some(row => { + const block = expectedTxs.get(row.id); return !block || row.height !== block.height || row.block_hash !== block.hash; + })) throw new Error('Canonical transaction index mismatch'); + const undo = this.db.prepare('SELECT COUNT(*) AS n FROM canonical_undo').get() as { n: number }; + if (undo.n !== chain.length) throw new Error('Canonical undo index mismatch'); + // A cursor is only useful if history has no gaps and describes the same validated DAG transitions. + const stack: string[] = []; + let sequence = 0, anchor = GENESIS_HASH; + for (const raw of this.db.prepare(`SELECT e.*,b.block_json FROM canonical_events e + LEFT JOIN blocks b ON b.hash=e.hash ORDER BY e.sequence`).iterate() as Iterable<{ + sequence: number; type: string; height: number; hash: string; previous_hash: string; + transactions_json: string; block_json: string | null; previous_event_hash: string; event_hash: string }>) { + const block = raw.block_json ? JSON.parse(raw.block_json) as TestnetBlock : undefined; + if (raw.sequence !== ++sequence || !block || block.height !== raw.height || block.previousHash !== raw.previous_hash || + raw.transactions_json !== JSON.stringify(block.transactions.map(transactionId))) throw new Error('Canonical event integrity mismatch'); + if (raw.previous_event_hash !== anchor || raw.event_hash !== this.eventHash(raw.sequence, raw.type, + block, anchor)) throw new Error('Canonical event anchor mismatch'); + anchor = raw.event_hash; + if (raw.type === 'connect') { + if (raw.height !== stack.length + 1 || raw.previous_hash !== (stack.at(-1) ?? GENESIS_HASH)) throw new Error('Canonical event connect mismatch'); + stack.push(raw.hash); + } else if (raw.type === 'disconnect') { + if (raw.height !== stack.length || stack.pop() !== raw.hash) throw new Error('Canonical event disconnect mismatch'); + } else throw new Error('Unknown canonical event'); + } + if (stack.length !== chain.length || stack.some((hash, i) => hash !== chain[i].hash)) throw new Error('Canonical event tip mismatch'); + } + private event(type: CanonicalEvent['type'], block: TestnetBlock) { + const prior = this.db.prepare('SELECT sequence,event_hash FROM canonical_events ORDER BY sequence DESC LIMIT 1').get() as + { sequence: number; event_hash: string } | undefined; + const sequence = (prior?.sequence ?? 0) + 1, anchor = prior?.event_hash ?? GENESIS_HASH; + this.db.prepare('INSERT INTO canonical_events(sequence,type,height,hash,previous_hash,transactions_json,previous_event_hash,event_hash) VALUES(?,?,?,?,?,?,?,?)') + .run(sequence, type, block.height, block.hash, block.previousHash, JSON.stringify(block.transactions.map(transactionId)), + anchor, this.eventHash(sequence, type, block, anchor)); + } + private eventHash(sequence: number, type: string, block: TestnetBlock, previousAnchor: string): string { + return sha256(JSON.stringify(['wikideck-canonical-event-v1', this.streamId, sequence, type, + block.height, block.hash, block.previousHash, block.transactions.map(transactionId), previousAnchor])); + } + connect(block: TestnetBlock): void { + const tokens = new Set([block.tokenId, ...block.transactions.flatMap(transactionCardIds)]); + const undo = [...tokens].map(token => [token, this.db.prepare('SELECT * FROM canonical_nfts WHERE token=?').get(token) ?? null]); + this.db.prepare('INSERT INTO canonical_undo VALUES(?,?)').run(block.height, JSON.stringify(undo)); + this.db.prepare('INSERT INTO canonical_blocks VALUES(?,?)').run(block.height, block.hash); + this.db.prepare('INSERT INTO canonical_nfts VALUES(?,?,?,?,?)').run(block.tokenId, block.owner, 0, block.height, block.hash); + const update = this.db.prepare('UPDATE canonical_nfts SET owner=?,sequence=sequence+1 WHERE token=?'); + for (const tx of block.transactions) { + if (isNftTrade(tx)) { + for (const leg of tx.offered) update.run(tx.counterparty, leg.cardHash); + for (const leg of tx.requested) update.run(tx.proposer, leg.cardHash); + } else update.run(tx.to, tx.cardHash); + this.db.prepare('INSERT INTO canonical_transactions VALUES(?,?,?)').run(transactionId(tx), block.height, block.hash); + } + this.event('connect', block); + } + disconnect(block: TestnetBlock): void { + const row = this.db.prepare('SELECT data FROM canonical_undo WHERE height=?').get(block.height) as { data: string } | undefined; + if (!row) throw new Error('Missing canonical undo record'); + for (const [token, prior] of JSON.parse(row.data) as [string, NftRow | null][]) { + this.db.prepare('DELETE FROM canonical_nfts WHERE token=?').run(token); + if (prior) this.db.prepare('INSERT INTO canonical_nfts VALUES(?,?,?,?,?)') + .run(prior.token, prior.owner, prior.sequence, prior.minted_height, prior.block_hash); + } + this.db.prepare('DELETE FROM canonical_transactions WHERE height=?').run(block.height); + this.db.prepare('DELETE FROM canonical_blocks WHERE height=?').run(block.height); + this.db.prepare('DELETE FROM canonical_undo WHERE height=?').run(block.height); + this.event('disconnect', block); + } + head(): number { + return (this.db.prepare('SELECT COALESCE(MAX(sequence),0) AS n FROM canonical_events').get() as { n: number }).n; + } + events(after: number, limit = 64): CanonicalEvents { + const head = this.head(); + if (!Number.isSafeInteger(after) || after < 0 || after > head || !Number.isSafeInteger(limit) || limit < 1 || limit > 64) { + throw new RpcError(409, 'Invalid canonical event cursor; restore/reconcile local consumer state.'); + } + const rows = this.db.prepare('SELECT * FROM canonical_events WHERE sequence>? ORDER BY sequence LIMIT ?').all(after, limit) as + { sequence: number; type: CanonicalEvent['type']; height: number; hash: string; previous_hash: string; + transactions_json: string; event_hash: string }[]; + if (rows.some((r, i) => r.sequence !== after + i + 1)) throw new Error('Canonical event history gap'); + const events = rows.map(r => ({ sequence: r.sequence, type: r.type, height: r.height, hash: r.hash, + previousHash: r.previous_hash, transactionIds: JSON.parse(r.transactions_json) as string[], eventHash: r.event_hash })); + const cursor = events.at(-1)?.sequence ?? after; + if (cursor < head && !events.length) throw new Error('Missing canonical event history'); + const previousAnchor = after === 0 ? GENESIS_HASH : (this.db.prepare('SELECT event_hash FROM canonical_events WHERE sequence=?').get(after) as { event_hash: string } | undefined)?.event_hash; + if (!previousAnchor) throw new Error('Missing canonical cursor anchor'); + return { streamId: this.streamId, previousAnchor, anchor: events.at(-1)?.eventHash ?? previousAnchor, + events, cursor, head, caughtUp: cursor === head }; + } + nft(token: string, height: number): IndexedNft | undefined { + const row = this.db.prepare(`SELECT n.owner,n.sequence,n.minted_height,b.block_json FROM canonical_nfts n + JOIN blocks b ON b.hash=n.block_hash WHERE n.token=?`).get(token) as + { owner: string; sequence: number; minted_height: number; block_json: string } | undefined; + return row ? { block: JSON.parse(row.block_json), owner: row.owner, sequence: row.sequence, + confirmations: height - row.minted_height + 1 } : undefined; + } + transaction(id: string): { height: number; blockHash: string } | undefined { + const row = this.db.prepare('SELECT height,block_hash FROM canonical_transactions WHERE id=?').get(id) as + { height: number; block_hash: string } | undefined; + return row ? { height: row.height, blockHash: row.block_hash } : undefined; + } + deck(owner: string, offset: number, limit: number, height: number) { + const total = (this.db.prepare('SELECT COUNT(*) AS n FROM canonical_nfts WHERE owner=?').get(owner) as { n: number }).n; + const rows = this.db.prepare(`SELECT n.owner,n.sequence,n.minted_height,b.block_json FROM canonical_nfts n + JOIN blocks b ON b.hash=n.block_hash WHERE n.owner=? ORDER BY n.minted_height,n.token LIMIT ? OFFSET ?`) + .all(owner, limit, offset) as { owner: string; sequence: number; minted_height: number; block_json: string }[]; + return { total, nfts: rows.map(r => ({ block: JSON.parse(r.block_json) as TestnetBlock, + sequence: r.sequence, confirmations: height - r.minted_height + 1 })) }; + } + tip(): string { + return (this.db.prepare('SELECT hash FROM canonical_blocks ORDER BY height DESC LIMIT 1').get() as { hash: string } | undefined)?.hash ?? GENESIS_HASH; + } +} diff --git a/server/testnet/http.ts b/server/testnet/http.ts index 1b6e263..55ddf5d 100644 --- a/server/testnet/http.ts +++ b/server/testnet/http.ts @@ -6,7 +6,7 @@ import { fileURLToPath } from 'node:url'; import { mkdir, open, unlink } from 'node:fs/promises'; import { z } from 'zod'; import { NETWORK, CONSENSUS } from '../../shared/testnet.ts'; -import { testnetOwnerOf, transactionCardIds } from '../../shared/testnet.ts'; +import { transactionCardIds } from '../../shared/testnet.ts'; import { CHAIN_ID, blockSchema, tradeSchema, transferSchema, transactionSchema } from './consensus.ts'; import { RandomX } from './randomx.ts'; import { Ledger } from './ledger.ts'; @@ -56,11 +56,10 @@ let custody: CustodyMarket | undefined; try { if (custodyConfig === 'sandbox') custody = await CustodyMarket.open(dataDir, { feeBps: paymentConfig.feeBps, card: (hash, owner, sequence) => { - const s = ledger.snapshot(), b = s.blocks.find(block => block.tokenId === hash); - if (!b || testnetOwnerOf(s, hash) !== owner || s.blocks.length + 1 - b.height < CONSENSUS.coinbaseMaturity || - s.transfers.filter(tx => transactionCardIds(tx).includes(hash)).length !== sequence || - s.pendingTransfers.some(tx => transactionCardIds(tx).includes(hash))) return undefined; - return { title: b.article.title }; + const nft = ledger.nft(hash); + if (!nft || nft.owner !== owner || nft.confirmations < CONSENSUS.coinbaseMaturity || nft.sequence !== sequence || + 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;} const app = express(); @@ -138,6 +137,9 @@ app.get('/api/decks/:address', rateLimit({windowMs:60000,limit:60,standardHeader res.status(400).json({error:'Invalid deck page'});return; } res.setHeader('Cache-Control','no-store'); + if(req.query.tip!==undefined && req.query.tip!==ledger.summary().tip){ + res.status(409).json({error:'Canonical tip changed; restart deck pagination.'});return; + } res.json(ledger.deck(req.params.address,offset,limit)); }); // Optional live data is intentionally not used by genesis, block validation or NFT ownership. @@ -156,13 +158,15 @@ app.get('/p2p/blocks',(req,res)=>{ }); app.get('/p2p/mempool',(_req,res)=>res.json({transactions:ledger.pendingTransactions()})); app.get('/api/nfts/:id', (req,res)=>{ - const s=ledger.snapshot();const block=s.blocks.find(b=>b.tokenId===req.params.id); - if(!block){res.status(404).json({error:'NFT not found on the canonical chain'});return;} - const owner=testnetOwnerOf(s,block.tokenId)??block.owner; - const sequence=s.transfers.filter(tx=>transactionCardIds(tx).includes(block.tokenId)).length; - res.json({network:NETWORK,chainId:CHAIN_ID,tokenId:block.tokenId,owner,sequence,blockHash:block.hash, - confirmations:s.blocks.length-block.height+1,article:block.article,statsSnapshot:block.statsSnapshot, - stateRoot:s.blocks.at(-1)?.stateRoot}); + if(typeof req.params.id!=='string'||!/^[a-f0-9]{64}$/.test(req.params.id)){ + res.status(400).json({error:'Invalid NFT identifier'});return; + } + const nft=ledger.nft(req.params.id); + if(!nft){res.status(404).json({error:'NFT not found on the canonical chain'});return;} + const block=nft.block; + res.json({network:NETWORK,chainId:CHAIN_ID,tokenId:block.tokenId,owner:nft.owner,sequence:nft.sequence,blockHash:block.hash, + confirmations:nft.confirmations,article:block.article,statsSnapshot:block.statsSnapshot, + stateRoot:ledger.block(ledger.summary().tip)?.stateRoot}); }); const templateLimit=rateLimit({windowMs:60000,limit:20,standardHeaders:'draft-8',legacyHeaders:false}); app.post('/api/mining/template', templateLimit, async (req,res)=>{ diff --git a/server/testnet/ledger.ts b/server/testnet/ledger.ts index 8f1b00e..25499b6 100644 --- a/server/testnet/ledger.ts +++ b/server/testnet/ledger.ts @@ -13,6 +13,7 @@ import type { Article } from '../../shared/protocol.ts'; import type { MintStatsSnapshot } from '../../shared/enrichment.ts'; import { PROFILE } from '../../shared/network.ts'; import { RpcError } from '../resources.ts'; +import { CanonicalIndex } from './canonical-index.ts'; const MAX_PENDING = 256; const MAX_DECK_PAGE = 24; @@ -39,11 +40,14 @@ export class Ledger { private readonly file: string; private readonly db: DatabaseSync; private readonly pow: PowEngine; + private readonly index: CanonicalIndex; private closed = false; private revision = 0; private reorgs = 0; private canonicalCache: { tip: string; blocks: TestnetBlock[] } | undefined; - private constructor(file: string, db: DatabaseSync, pow: PowEngine) { this.file = file; this.db = db; this.pow = pow; } + private constructor(file: string, db: DatabaseSync, pow: PowEngine) { + this.file = file; this.db = db; this.pow = pow; this.index = new CanonicalIndex(db); + } static async open(dir: string, pow: PowEngine): Promise { const directory = join(dir, PROFILE); @@ -141,6 +145,7 @@ export class Ledger { ledger.pending.push(tx); } ledger.pendingState = state; + ledger.index.initialize(ledger.canonical(), materializeState(ledger.tipState())); return ledger; } catch (error) { db.close(); @@ -192,7 +197,9 @@ export class Ledger { [...disconnected.flatMap(b => b.transactions), ...this.pending]) : undefined; const pending = filtered?.pending ?? this.pending; // Nothing visible changes until the atomic replacement succeeds. - await this.persist(node.block, pending); + const oldHashes = new Set(oldChain.map(b => b.hash)); + const connected = wins ? chain.filter(b => !oldHashes.has(b.hash)) : []; + this.persist(node.block, pending, disconnected, connected); this.nodes.set(node.block.hash, node); if (wins) this.tip = node.block.hash; if (disconnected.length) this.reorgs++; @@ -217,7 +224,7 @@ export class Ledger { if (this.pending.length >= MAX_PENDING) throw new Error('Testnet mempool limit: 256 transfers'); const nextPendingState = applyTransactionOverlay(this.pendingState, tx, this.height() + 1); const pending = [...this.pending, tx]; - await this.persist(undefined, pending); + this.persist(undefined, pending); this.pending = pending; this.pendingState = nextPendingState; this.revision++; @@ -246,6 +253,10 @@ export class Ledger { } peerStatus() { return { network: NETWORK, chainId: CHAIN_ID, size: this.size, tip: this.tip }; } pendingTransactions(): TestnetTransaction[] { return structuredClone(this.pending); } + canonicalEvents(after: number, limit = 64) { return this.index.events(after, limit); } + canonicalHead(): number { return this.index.head(); } + nft(token: string) { return this.index.nft(token, this.height()); } + canonicalTransaction(id: string) { return this.index.transaction(id); } legacySnapshot(maxBlocks: number): TestnetSnapshot { // Reject before any full-chain copy/serialization. Never truncate ownership history silently. if (this.height() > maxBlocks) throw new RpcError(413, 'Full snapshot disabled at this height; use /api/summary and /api/canonical pages.'); @@ -263,18 +274,10 @@ export class Ledger { if (!Number.isSafeInteger(offset) || offset < 0 || !Number.isSafeInteger(limit) || limit < 1 || limit > MAX_DECK_PAGE) { throw new Error('Invalid deck page'); } - const blocks = this.canonical(); - const byToken = new Map(blocks.map(block => [block.tokenId, block])); - const state = materializeState(this.tipState()); - const owned = Object.entries(state).filter(([, nft]) => nft.owner === address) - .flatMap(([tokenId, nft]) => { - const block = byToken.get(tokenId); - return block ? [{ block, sequence: nft.sequence, confirmations: blocks.length - block.height + 1 }] : []; - }).sort((a, b) => a.block.height - b.block.height || a.block.tokenId.localeCompare(b.block.tokenId)); - const nfts = owned.slice(offset, offset + limit); - const next = offset + nfts.length < owned.length ? offset + nfts.length : null; + const { total, nfts } = this.index.deck(address, offset, limit, this.height()); + const next = offset + nfts.length < total ? offset + nfts.length : null; return structuredClone({ network: NETWORK, chainId: CHAIN_ID, address, tip: this.tip, - offset, limit, total: owned.length, next, nfts }); + offset, limit, total, next, nfts }); } private height(): number { return this.nodes.get(this.tip)?.block.height ?? 0; } @@ -322,7 +325,8 @@ export class Ledger { this.queue = result.catch(() => undefined).finally(() => { this.queued--; }); return result; } - private async persist(block: TestnetBlock | undefined, pendingTransfers: TestnetTransaction[]): Promise { + private persist(block: TestnetBlock | undefined, pendingTransfers: TestnetTransaction[], + disconnected: TestnetBlock[] = [], connected: TestnetBlock[] = []): void { this.db.exec('BEGIN IMMEDIATE'); try { if (block) this.db.prepare('INSERT INTO blocks(hash, previous_hash, block_json) VALUES(?, ?, ?)') @@ -330,6 +334,8 @@ export class Ledger { this.db.exec('DELETE FROM pending_transfers'); const insertPending = this.db.prepare('INSERT INTO pending_transfers(position, id, transaction_json) VALUES(?, ?, ?)'); pendingTransfers.forEach((tx, index) => insertPending.run(index, transactionId(tx), JSON.stringify(tx))); + for (const old of [...disconnected].reverse()) this.index.disconnect(old); + for (const next of connected) this.index.connect(next); this.db.exec('COMMIT'); } catch (error) { this.db.exec('ROLLBACK'); diff --git a/tests/api.test.ts b/tests/api.test.ts index 78367d3..4db1902 100644 --- a/tests/api.test.ts +++ b/tests/api.test.ts @@ -288,6 +288,10 @@ test('native RandomX HTTP: three blocks, mined transfer, peer gossip/sync and du assert.equal(aliceDeck.tip, block2.hash); assert.equal(aliceDeck.total, 2); assert.deepEqual(aliceDeck.nfts.map(item => item.block.tokenId), [block1.tokenId, block2.tokenId]); + assert.equal((await json<{ owner: string; sequence: number }>(first.base, `/api/nfts/${block1.tokenId}`)).owner, alice.owner); + await request(first.base, '/api/nfts/not-a-token', 400); + await request(first.base, `/api/decks/${alice.owner}?tip=${block1.hash}`, 409); + assert.equal((await json<{ total: number }>(first.base, `/api/decks/${alice.owner}?tip=${block2.hash}&limit=1`)).total, 2); const market = await json<{ info: MarketInfo; listings: MarketListingV3[] }>(first.base, '/api/market'); assert.equal(market.info.paymentMode, 'mock'); assert.deepEqual(market.listings, []); diff --git a/tests/market.test.ts b/tests/market.test.ts index 5112de1..f92e42e 100755 --- a/tests/market.test.ts +++ b/tests/market.test.ts @@ -10,7 +10,8 @@ import { mockQuote, paymentMode } from '../server/market/payments.ts'; import { ADDITIONAL_MARKET_FEE, marketMessage, parseUsd, type MarketAction } from '../shared/market.ts'; import { marketActionV3Schema, marketMessageV3, type MarketActionV3, type MarketActivityV3, type MarketListingV3, type MarketOrderV3 } from '../shared/market-v3.ts'; -import { NETWORK, nftTransferMessage, type NftTransfer, type TestnetSnapshot, type TestnetBlock } from '../shared/testnet.ts'; +import { NETWORK, nftTransferMessage, testnetOwnerOf, transactionCardIds, type NftTransfer, type TestnetSnapshot, type TestnetBlock } from '../shared/testnet.ts'; +import type { CanonicalEvent } from '../server/testnet/canonical-index.ts'; import { address, CHAIN_ID, GENESIS_HASH, transferId, verifySignature } from '../server/testnet/consensus.ts'; function identity() { @@ -35,11 +36,37 @@ function chainFixture() { return id; } }; } -async function fixture(t: { after(fn: () => Promise): void }) { +async function fixture(t: { after(fn: () => Promise): void }, indexed = false) { const dir = await mkdtemp(join(tmpdir(), 'wikideck-market-')); const chain = chainFixture(); + // Test-only indexed adapter: production indexes/events are exercised separately by Ledger tests. + const events: CanonicalEvent[] = []; + const streamId = randomUUID(); + const adapter = indexed ? { ...chain, + snapshot: (): TestnetSnapshot => { throw new Error('Indexed market must not request a full chain snapshot'); }, + summary: () => ({ tip: chain.state.tip, revision: events.length, height: chain.state.blocks.length }), + pendingTransactions: () => structuredClone(chain.state.pendingTransfers), + nft: (token: string) => { + const b = chain.state.blocks.find(b => b.tokenId === token); + return b ? { block: b, owner: testnetOwnerOf(chain.state, token)!, + sequence: chain.state.transfers.filter(tx => transactionCardIds(tx).includes(token)).length, + confirmations: chain.state.blocks.length + 1 - b.height } : undefined; + }, + canonicalTransaction: (id: string) => { + const b = chain.state.blocks.find(b => b.transactions.some(tx => transferId(tx as NftTransfer) === id)); + return b ? { height: b.height, blockHash: b.hash ?? 'f'.repeat(64) } : undefined; + }, + canonicalHead: () => events.length, + canonicalEvents: (after: number, limit = 64) => { + if (after > events.length) throw new Error('Canonical cursor ahead of restored ledger'); + const page = events.slice(after, after + limit), cursor = page.at(-1)?.sequence ?? after; + return { streamId, previousAnchor: after ? events[after - 1].eventHash : GENESIS_HASH, + anchor: page.at(-1)?.eventHash ?? (after ? events[after - 1].eventHash : GENESIS_HASH), + events: page, cursor, head: events.length, caughtUp: cursor === events.length }; + }, + } : chain; let time = 1_800_000_000_000; - let market = await Market.open(dir, chain, { feeBps: 250, confirmations: 2 }, () => time); + let market = await Market.open(dir, adapter, { feeBps: 250, confirmations: 2 }, () => time); t.after(async () => { market.close(); await rm(dir, { recursive: true, force: true }); }); const envelope = (who: typeof alice, action: MarketAction) => { const terms = { network: NETWORK, chainId: CHAIN_ID, marketId: market.id, actor: who.actor, @@ -76,12 +103,57 @@ async function fixture(t: { after(fn: () => Promise): void }) { chain.state.blocks.push({ ...chain.state.blocks[0], height, transactions: [tx] }); chain.state.transfers.push(tx); chain.state.pendingTransfers = []; }; - return { dir, chain, envelope, envelopeV3, run, list, buy, orders, delivery, confirm, + return { dir, chain, events, envelope, envelopeV3, run, list, buy, orders, delivery, confirm, get market() { return market; }, get now() { return time; }, advance: (ms: number) => { time += ms; }, - restart: async () => { market.close(); market = await Market.open(dir, chain, + restart: async () => { market.close(); market = await Market.open(dir, adapter, { feeBps: 500, confirmations: 5 }, () => time); } }; } +test('indexed market uses bounded durable event catch-up and never requests full snapshots', async t => { + const f = await fixture(t, true); + for (let i = 1; i <= 130; i++) f.events.push({ sequence: i, type: 'connect', height: i, + hash: 'd'.repeat(64), previousHash: 'e'.repeat(64), transactionIds: [], eventHash: i.toString(16).padStart(64, '0') }); + await assert.rejects(f.list(), /catching up/); + assert.equal(f.market.diagnostics().canonicalEventCursor, 64); + f.market.browse(); assert.equal(f.market.diagnostics().canonicalEventCursor, 128); + const l = await f.list(); assert.ok(l.id); + assert.equal(f.market.diagnostics().canonicalEventCursor, 130); + assert.equal(f.market.diagnostics().eventBacklog, false); + await f.restart(); assert.equal(f.market.diagnostics().canonicalEventCursor, 130); + f.market.browse(); assert.equal(f.market.diagnostics().canonicalEventCursor, 130); + const anchor = f.events[129].eventHash; + f.events[129].eventHash = 'f'.repeat(64); f.advance(1001); + assert.throws(() => f.market.browse(), /history changed/); + f.events[129].eventHash = anchor; + f.events.length = 0; + assert.throws(() => f.market.browse(), /cursor ahead/); +}); + +test('disconnect cursor and payout hold commit together and survive a crash/restart', async t => { + const f = await fixture(t, true), listing = await f.list(), order = await f.buy(listing); + await f.run(bob, { type: 'mock-pay', orderId: order.id }); + const delivery = f.delivery(order); f.confirm(delivery); + f.chain.state.blocks.push({ ...f.chain.state.blocks[0], height: f.chain.state.blocks.length + 1 }); + f.events.push({ sequence: 1, type: 'connect', height: 3, hash: 'd'.repeat(64), previousHash: 'e'.repeat(64), transactionIds: [transferId(delivery)], eventHash: '1'.repeat(64) }); + assert.equal((await f.orders())[0].status, 'complete'); + // The same delivery may already be canonical again; the intermediate disconnect must not be lost. + f.events.push({ sequence: 2, type: 'disconnect', height: 3, hash: 'd'.repeat(64), previousHash: 'e'.repeat(64), transactionIds: [transferId(delivery)], eventHash: '2'.repeat(64) }); + const db = new DatabaseSync(join(f.dir, 'testnet', 'market.sqlite')); + try { + db.exec("CREATE TRIGGER fail_cursor BEFORE UPDATE ON market_meta WHEN NEW.key='canonical_event_cursor' BEGIN SELECT RAISE(ABORT,'injected cursor failure'); END"); + assert.throws(() => f.market.browse(), /injected cursor failure/); + const persisted = db.prepare('SELECT data FROM orders WHERE id=?').get(order.id) as { data: string }; + assert.equal(JSON.parse(persisted.data).status, 'complete'); + assert.equal(f.market.diagnostics().canonicalEventCursor, 1); + db.exec('DROP TRIGGER fail_cursor'); + } finally { db.close(); } + await f.restart(); + const reviewed = (await f.orders())[0]; + assert.equal(reviewed.status, 'review'); assert.equal(reviewed.payout, 'review'); + assert.equal(f.market.diagnostics().canonicalEventCursor, 2); + await f.restart(); assert.equal((await f.orders())[0].status, 'review'); +}); + test('USD and mock XMR accounting use bounded integer amounts and exact fee splitting', () => { assert.equal(parseUsd('15.01'), 1501); assert.equal(parseUsd('0.01'), 1); for (const invalid of ['0', '-1', '1e3', '1.001', ' 1', 'NaN', '1000001', '01.00']) assert.throws(() => parseUsd(invalid)); diff --git a/tests/testnet-ledger.test.ts b/tests/testnet-ledger.test.ts index 3d8990f..4be1189 100644 --- a/tests/testnet-ledger.test.ts +++ b/tests/testnet-ledger.test.ts @@ -12,7 +12,7 @@ import { Ledger } from '../server/testnet/ledger.ts'; import { PeerNetwork } from '../server/testnet/peers.ts'; import { CHAIN_ID, GENESIS_HASH, GENESIS_TIME, MAX_TARGET, address, makeTemplate, miningBlob, mintId, sha256, stateRoot, transferId, transition, workFor } from '../server/testnet/consensus.ts'; -import { NETWORK, CONSENSUS, blockClaimMessage, nftTransferMessage, testnetOwnerOf, +import { NETWORK, CONSENSUS, blockClaimMessage, nftTransferMessage, testnetOwnerOf, transactionCardIds, nftTradeMessage, type MiningTemplate, type NftState, type NftStateLayer, type NftStateView, type NftTrade, type NftTransfer, type TestnetBlock } from '../shared/testnet.ts'; import { deriveStats } from '../shared/enrichment.ts'; @@ -259,6 +259,11 @@ test('signed multi-card trade is validated, mempooled, mined, and applied atomic for (const block of offered) assert.equal(testnetOwnerOf(after, block.tokenId), bob.owner); for (const block of requested) assert.equal(testnetOwnerOf(after, block.tokenId), alice.owner); assert.deepEqual(after.transfers.at(-1), bundle); + for (const block of after.blocks) { + const indexed = ledger.nft(block.tokenId)!; + assert.equal(indexed.owner, testnetOwnerOf(after, block.tokenId)); + assert.equal(indexed.sequence, after.transfers.filter(tx => transactionCardIds(tx).includes(block.tokenId)).length); + } const deck = ledger.deck(alice.owner, 0, 2); assert.equal(deck.total, after.blocks.filter(block => testnetOwnerOf(after, block.tokenId) === alice.owner).length); assert.equal(deck.nfts.length, 2); @@ -291,14 +296,99 @@ test('reorg rolls back NFT ownership and requeues only still-valid disconnected assert.equal(testnetOwnerOf(snapshot, committed.tokenId), undefined); assert.deepEqual(snapshot.transfers, []); assert.deepEqual(snapshot.pendingTransfers, [tx, tx2]); // Orphan issuance is discarded. + assert.equal(ledger.nft(minted.tokenId)!.owner, alice.owner); + assert.equal(ledger.nft(minted.tokenId)!.sequence, 0); + assert.equal(ledger.nft(committed.tokenId), undefined); + assert.equal(ledger.canonicalTransaction(transferId(tx)), undefined); assert.ok(ledger.block(committed.hash)); const restored = await Ledger.open(dir, new FakePow()); t.after(async () => { restored.close(); }); assert.deepEqual(restored.snapshot(), snapshot); + assert.equal(restored.canonicalHead(), ledger.canonicalHead()); + assert.deepEqual(restored.canonicalEvents(0), ledger.canonicalEvents(0)); const next = await mine(ledger.template(bob.owner, bob.publicKey, articleFor(ledger.snapshot().blocks), statsSnapshot()), bob); await ledger.submit(next); assert.equal(testnetOwnerOf(ledger.snapshot(), minted.tokenId), carol.owner); + assert.equal(ledger.nft(minted.tokenId)!.owner, carol.owner); + assert.equal(ledger.nft(minted.tokenId)!.sequence, 2); +}); + +test('canonical event pages record reorg order durably but exclude side branches, duplicates and mempool', async t => { + const { ledger, file, dir } = await fixture(t), original = branch(); + const first = await extend(original); await ledger.submit(first); + const second = await extend(original); await ledger.submit(second); + const fork: Branch = structuredClone(original); + const tx = transfer(first.tokenId); + await ledger.submitTransfer(tx); + const beforePending = ledger.canonicalHead(); + assert.equal(beforePending, 2); + const third = await extend(original, alice, [tx]); await ledger.submit(third); + assert.equal(ledger.canonicalTransaction(transferId(tx))?.blockHash, third.hash); + const competing = await extend(fork, bob); await ledger.submit(competing); + assert.equal(ledger.canonicalHead(), 3); // Equal-work branch does not change canonical state. + const winner = await extend(fork, bob); await ledger.submit(winner); + assert.equal(ledger.canonicalHead(), 6); + assert.equal(ledger.canonicalTransaction(transferId(tx)), undefined); + const events = ledger.canonicalEvents(0).events; + assert.deepEqual(events.map(e => [e.type, e.hash]), [ + ['connect', first.hash], ['connect', second.hash], ['connect', third.hash], + ['disconnect', third.hash], ['connect', competing.hash], ['connect', winner.hash], + ]); + assert.deepEqual(events[3].transactionIds, [transferId(tx)]); + const page = ledger.canonicalEvents(2, 1); + assert.equal(page.cursor, 3); assert.equal(page.caughtUp, false); assert.equal(page.head, 6); + page.events[0].transactionIds.length = 0; + assert.equal(ledger.canonicalEvents(2, 1).events[0].transactionIds.length, 1); + for (const [after, limit] of [[-1, 1], [7, 1], [0, 65], [0, 0], [0.5, 1]]) { + assert.throws(() => ledger.canonicalEvents(after, limit), /cursor/); + } + await ledger.submit(winner); assert.equal(ledger.canonicalHead(), 6); + ledger.close(); + const reopened = await Ledger.open(dir, new FakePow()); + try { + assert.equal(reopened.canonicalHead(), 6); assert.deepEqual(reopened.canonicalEvents(0).events, events); + const db = new DatabaseSync(file); + try { assert.throws(() => db.exec("UPDATE canonical_events SET type='connect'"), /Immutable/); } finally { db.close(); } + } finally { reopened.close(); } +}); + +test('block, mempool, canonical indexes and events roll back together on storage failure', async t => { + const { ledger, file } = await fixture(t), chain = branch(); + const candidate = await extend(chain), db = new DatabaseSync(file); + try { + db.exec("CREATE TRIGGER fail_event BEFORE INSERT ON canonical_events BEGIN SELECT RAISE(ABORT,'injected storage failure'); END"); + await assert.rejects(ledger.submit(candidate), /injected storage failure/); + assert.equal(ledger.size, 0); assert.equal(ledger.canonicalHead(), 0); assert.equal(ledger.nft(candidate.tokenId), undefined); + assert.equal((db.prepare('SELECT COUNT(*) AS n FROM blocks').get() as { n: number }).n, 0); + db.exec('DROP TRIGGER fail_event'); + await ledger.submit(candidate); + assert.equal(ledger.nft(candidate.tokenId)!.owner, alice.owner); assert.equal(ledger.canonicalHead(), 1); + } finally { db.close(); } +}); + +test('validated existing ledgers backfill indexes once and refuse corrupt ownership or event history', async t => { + const { ledger, file, dir } = await fixture(t), chain = branch(); + await ledger.submit(await extend(chain)); await ledger.submit(await extend(chain)); + const expected = ledger.snapshot(); ledger.close(); + const db = new DatabaseSync(file); + try { + // Simulate the prior schema: preserve original blocks/metadata/pending exactly, remove only new derived tables. + db.exec('DROP TABLE canonical_undo; DROP TABLE canonical_transactions; DROP TABLE canonical_nfts; DROP TABLE canonical_blocks; DROP TABLE canonical_events; DROP TABLE canonical_meta;'); + } finally { db.close(); } + const migrated = await Ledger.open(dir, new FakePow()); + assert.deepEqual(migrated.snapshot(), expected); assert.equal(migrated.canonicalHead(), 2); migrated.close(); + const corrupt = new DatabaseSync(file); + try { corrupt.prepare('UPDATE canonical_nfts SET owner=? WHERE token=?').run(bob.owner, chain.blocks[0].tokenId); } + finally { corrupt.close(); } + await assert.rejects(Ledger.open(dir, new FakePow()), /ownership index mismatch/); + const repair = new DatabaseSync(file); + try { + repair.prepare('UPDATE canonical_nfts SET owner=? WHERE token=?').run(alice.owner, chain.blocks[0].tokenId); + repair.exec('DROP TRIGGER canonical_events_no_update'); + repair.exec("UPDATE canonical_events SET transactions_json='[\"wrong\"]' WHERE sequence=1"); + } finally { repair.close(); } + await assert.rejects(Ledger.open(dir, new FakePow()), /event integrity mismatch/); }); test('reorg removes mempool conflicts and does not requeue already committed transfers', async t => {