import { mkdir, open, rename, unlink } from 'node:fs/promises'; import { join } from 'node:path'; import { randomUUID } from 'node:crypto'; import { setTimeout as delay } from 'node:timers/promises'; import { z } from 'zod'; import { NETWORK, CONSENSUS, type TestnetBlock, type NftTransfer, type NftState, type MiningTemplate, type TestnetSnapshot } from '../../shared/testnet.ts'; import { CHAIN_ID, GENESIS_HASH, applyTransfer, blockSchema, makeTemplate, transferId, transferSchema, validateBlock, workFor } from './consensus.ts'; import type { PowEngine } from './randomx.ts'; import type { Article } from '../../shared/protocol.ts'; import type { MintStatsSnapshot } from '../../shared/enrichment.ts'; const MAX_BLOCKS = 10_000; const MAX_PENDING = 256; const MAX_FILE = 128 * 1024 * 1024; const diskSchema = z.object({ version: z.literal(1), network: z.literal(NETWORK), chainId: z.literal(CHAIN_ID), blocks: z.array(blockSchema).max(MAX_BLOCKS), pendingTransfers: z.array(transferSchema).max(MAX_PENDING) }).strict(); type Node = { block: TestnetBlock; state: NftState; work: bigint }; /** Experimental, bounded DAG. The supplied engine is always used, including on restart. * One process must own a data directory; there is no cross-process writer locking. */ export class Ledger { private readonly nodes = new Map(); private tip = GENESIS_HASH; private pending: NftTransfer[] = []; private queue: Promise = Promise.resolve(); private queued = 0; /** Application-owned connectivity metadata; never part of consensus or persistence. */ peers = 0; private readonly file: string; private readonly pow: PowEngine; private constructor(file: string, pow: PowEngine) { this.file = file; this.pow = pow; } static async open(dir: string, pow: PowEngine): Promise { const directory = join(dir, 'testnet'); await mkdir(directory, { recursive: true, mode: 0o700 }); const ledger = new Ledger(join(directory, 'ledger.json'), pow); let handle; try { handle = await open(ledger.file, 'r'); } catch (error) { if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error; await ledger.persist([], []); return ledger; } let text: string; try { if ((await handle.stat()).size > MAX_FILE) throw new Error('Testnet ledger file exceeds 128 MiB'); // Bounded reads also protect against a file growing after stat(). const chunks: Buffer[] = []; let size = 0; for (;;) { const buffer = Buffer.alloc(Math.min(64 * 1024, MAX_FILE + 1 - size)); const { bytesRead } = await handle.read(buffer, 0, buffer.length, null); if (!bytesRead) break; size += bytesRead; if (size > MAX_FILE) throw new Error('Testnet ledger file exceeds 128 MiB'); chunks.push(buffer.subarray(0, bytesRead)); } text = Buffer.concat(chunks).toString('utf8'); } finally { await handle.close(); } const disk = diskSchema.parse(JSON.parse(text)); // Mismatches/corruption fail closed, never reset. for (const raw of disk.blocks) { if (ledger.nodes.has(raw.hash)) throw new Error('Duplicate persisted block'); const node = await ledger.validate(raw); ledger.nodes.set(node.block.hash, node); if (node.work > ledger.tipWork()) ledger.tip = node.block.hash; } const state = structuredClone(ledger.tipState()); const seen = new Set(ledger.canonical().flatMap(b => b.transactions.map(transferId))); for (const tx of disk.pendingTransfers) { const id = transferId(tx); if (seen.has(id)) throw new Error('Duplicate persisted transfer'); applyTransfer(state, tx, ledger.height() + 1); seen.add(id); } ledger.pending = disk.pendingTransfers; return ledger; } get size(): number { return this.nodes.size; } get allBlocks(): TestnetBlock[] { return this.inventory(0, MAX_BLOCKS); } block(hash: string): TestnetBlock | undefined { const block = this.nodes.get(hash)?.block; return block ? structuredClone(block) : undefined; } /** Append-only insertion order is topological, including all valid side branches. * HTTP callers should request limit=32 and return next=offset+length or null. */ inventory(offset: number, limit: number): TestnetBlock[] { if (!Number.isSafeInteger(offset) || offset < 0 || !Number.isSafeInteger(limit) || limit < 1 || limit > MAX_BLOCKS) { throw new Error('Invalid inventory range'); } return structuredClone(Array.from(this.nodes.values()).slice(offset, offset + limit).map(n => n.block)); } async submit(raw: unknown): Promise<{ accepted: boolean; reorg: boolean }> { const input = structuredClone(raw); return this.serialize(async () => { const parsed = blockSchema.parse(input); if (this.nodes.has(parsed.hash)) return { accepted: false, reorg: false }; if (this.size >= MAX_BLOCKS) throw new Error('Experimental testnet limit: 10000 stored blocks'); const node = await this.validate(parsed); const wins = node.work > this.tipWork(); const oldChain = this.canonical(); const chain = wins ? [...this.branch(node.block.previousHash), node.block] : oldChain; const hashes = new Set(chain.map(b => b.hash)); const disconnected = wins ? oldChain.filter(b => !hashes.has(b.hash)) : []; const pending = wins ? this.filterPending(node.state, chain, [...disconnected.flatMap(b => b.transactions), ...this.pending]) : this.pending; // Nothing visible changes until the atomic replacement succeeds. await this.persist([...Array.from(this.nodes.values(), n => n.block), node.block], pending); this.nodes.set(node.block.hash, node); if (wins) this.tip = node.block.hash; this.pending = pending; return { accepted: true, reorg: disconnected.length > 0 }; }); } async submitTransfer(raw: unknown): Promise { const input = structuredClone(raw); return this.serialize(async () => { const tx = transferSchema.parse(input); const id = transferId(tx); if (this.pending.some(t => transferId(t) === id)) return id; if (this.pending.length >= MAX_PENDING) throw new Error('Testnet mempool limit: 256 transfers'); const state = structuredClone(this.tipState()); for (const pending of this.pending) applyTransfer(state, pending, this.height() + 1); applyTransfer(state, tx, this.height() + 1); // No speculative issuance. const pending = [...this.pending, tx]; await this.persist(Array.from(this.nodes.values(), n => n.block), pending); this.pending = pending; return id; }); } template(owner: string, publicKey: string, article: Article, statsSnapshot: MintStatsSnapshot): MiningTemplate { return makeTemplate(this.canonical(), this.tipState(), owner, publicKey, structuredClone(this.pending.slice(0, CONSENSUS.maxTransactions)), structuredClone(article), structuredClone(statsSnapshot)); } snapshot(): TestnetSnapshot { const blocks = this.canonical(); return structuredClone({ network: NETWORK, chainId: CHAIN_ID, genesis: GENESIS_HASH, tip: this.tip, cumulativeWork: this.tipWork().toString(), blocks, transfers: blocks.flatMap(b => b.transactions), pendingTransfers: this.pending, peers: this.peers, algorithm: 'RandomX', maturity: CONSENSUS.coinbaseMaturity }); } private height(): number { return this.nodes.get(this.tip)?.block.height ?? 0; } private tipState(): NftState { return this.nodes.get(this.tip)?.state ?? {}; } private tipWork(): bigint { return this.nodes.get(this.tip)?.work ?? 0n; } private branch(hash: string): TestnetBlock[] { const result: TestnetBlock[] = []; while (hash !== GENESIS_HASH) { const node = this.nodes.get(hash); if (!node) throw new Error('Unknown parent'); result.push(node.block); hash = node.block.previousHash; } return result.reverse(); } private canonical(): TestnetBlock[] { return this.branch(this.tip); } private async validate(raw: TestnetBlock): Promise { const ancestors = this.branch(raw.previousHash); const parent = this.nodes.get(raw.previousHash); const { block, state } = await validateBlock(raw, ancestors, parent?.state ?? {}, this.pow); return { block, state, work: (parent?.work ?? 0n) + workFor(block.target) }; } private filterPending(parent: NftState, chain: TestnetBlock[], candidates: NftTransfer[]): NftTransfer[] { const state = structuredClone(parent); const seen = new Set(chain.flatMap(b => b.transactions.map(transferId))); const pending: NftTransfer[] = []; for (const tx of candidates) { if (pending.length === MAX_PENDING) break; const id = transferId(tx); if (seen.has(id)) continue; try { applyTransfer(state, tx, chain.length + 1); } catch { continue; } // Disconnected issuance, immature coins and conflicts are not replayable. seen.add(id); pending.push(tx); } return pending; } private serialize(operation: () => Promise): Promise { if (this.queued >= 256) return Promise.reject(new Error('Testnet mutation queue full')); this.queued++; const result = this.queue.then(operation); this.queue = result.catch(() => undefined).finally(() => { this.queued--; }); return result; } private async persist(blocks: TestnetBlock[], pendingTransfers: NftTransfer[]): Promise { const data = JSON.stringify({ version: 1, network: NETWORK, chainId: CHAIN_ID, blocks, pendingTransfers }); if (Buffer.byteLength(data) > MAX_FILE) throw new Error('Testnet ledger file exceeds 128 MiB'); const temp = `${this.file}.${randomUUID()}.tmp`; try { const handle = await open(temp, 'wx', 0o600); try { await handle.writeFile(data); await handle.sync(); } finally { await handle.close(); } // Windows scanners can briefly hold the destination open without delete sharing. // Retry only transient sharing failures; never unlink the old durable ledger. for (let attempt = 0; ; attempt++) { try { await rename(temp, this.file); break; } catch (error) { const code = (error as NodeJS.ErrnoException).code; if (process.platform !== 'win32' || attempt >= 5 || !['EPERM', 'EACCES', 'EBUSY'].includes(code ?? '')) throw error; await delay(10 * 2 ** attempt); } } } finally { await unlink(temp).catch(() => undefined); } } }