From 268a8c1a75a4e80831da3e62a71b46138037d7e3 Mon Sep 17 00:00:00 2001 From: jagdeep sidhu Date: Sat, 24 Oct 2026 23:08:26 -0700 Subject: [PATCH] Keep requested privacy pools current with public withdrawal data --- .env.example | 5 +- docs/privacy-index.md | 61 ++++++++--- lib/appFactory.js | 1 + lib/privacyIndex.js | 160 +++++++++++++++++++++------ lib/privacyRpc.js | 58 ++++++---- lib/privacyStore.js | 40 ++++++- tests/privacyIndex.test.js | 218 ++++++++++++++++++++++++++++++++++--- tests/privacyRpc.test.js | 49 ++++++++- 8 files changed, 496 insertions(+), 96 deletions(-) diff --git a/.env.example b/.env.example index 75e7170..7398aef 100644 --- a/.env.example +++ b/.env.example @@ -97,7 +97,8 @@ SYSCOIN_SENTRY_GENESIS_HASH= # Optional public Tornado Classic ETH pool index. Disabled by default. # Set BOTH values to opt in. The RPC must support Ethereum mainnet finalized # headers, historical state and eth_getLogs from the 2019 deployment blocks. -# HTTPS only; no wallet keys or caller-supplied endpoints. No RPC runs until a -# public snapshot request arrives. See docs/privacy-index.md for lifecycle. +# HTTPS only; no wallet keys or caller-supplied endpoints. First request opts a +# pool into backfill and periodic refresh; previously used pools resume on restart. +# Unused pools stay idle. See docs/privacy-index.md for lifecycle and feed trust. PRIVACY_INDEX_ENABLED=false PRIVACY_INDEX_RPC_URL= diff --git a/docs/privacy-index.md b/docs/privacy-index.md index ba2e5fd..69378dd 100644 --- a/docs/privacy-index.md +++ b/docs/privacy-index.md @@ -1,7 +1,7 @@ # Public Classic ETH pool snapshots -`GET /api/privacy/v1/pools/:poolId/snapshot` supplies public deposit data for -Pali's local Tornado Classic Merkle-tree reconstruction. It never receives a +`GET /api/privacy/v1/pools/:poolId/snapshot` supplies public deposit and withdrawal data for +Pali's local Tornado Classic Merkle-tree reconstruction and note-status checks. It never receives a note, commitment, nullifier, leaf position, account address, signing key, RPC URL or arbitrary contract address. Every query string or request body is rejected. Only the following Ethereum mainnet pools are accepted: @@ -24,22 +24,31 @@ RPC. URLs containing credentials or fragments are rejected. A provider API token may be in its path/query; it never enters responses or application logs. The RPC must support `finalized` headers, historical `eth_call`/`eth_getCode`, and `eth_getLogs` from the deployment blocks. Scans start with 5,000-block -ranges and shrink on provider range/response-size failures. There is no +ranges and shrink on provider range/response-size failures. After three sparse +successful full ranges, the scanner doubles the range toward the original +5,000-block cap. The cap is not increased based on an assumed provider capability. There is no fallback to unfinalized data or to a different provider. A provider which cannot serve even one block of historical logs needs a suitable endpoint before enabling. -No RPC is sent at startup. The first request starts a job, bounded to 16 chunks -and a 45-second scheduling budget (an already-started RPC can take another -10 seconds). A global queue runs only one job at a time; requests for the same -pool share it. Further requests resume incomplete work. Successful snapshots -are checked for refresh at most once a minute; failures back off for 30 seconds. +An unused pool sends no RPC. Its first request durably activates it and starts +a job, bounded to 16 chunks and a 45-second scheduling budget (an already-started +RPC can take another 10 seconds). A global queue runs only one job at a time; +requests for the same pool share it. Activated pools continue unfinished backfill +in bounded jobs separated by one second, then refresh once a minute without +requiring another client request. Failures back off for 30 seconds. On restart, +only previously activated pools resume. The single scheduled timer is unreferenced; +`index.close()` cancels future work and waits for the current bounded job. The +standalone app exposes its index handle for test/server lifecycle cleanup. Per-process IP limits allow 60 requests/minute and apply IPv6 subnet bucketing. Run one writer process for this file store; multiple workers require explicit shared locking and shared request limits before deployment. The backend process needs write access to `data/privacy-index/`, resolved from its working directory. Checkpoints and complete snapshots live in separate -files with checksums and are published by same-directory atomic rename. They +files with checksums and are published by same-directory atomic rename. An +allowlisted per-pool activation marker preserves opt-in intent even if the first +RPC fails. Existing snapshot/checkpoint filenames also identify previously used +pools during migration; unknown filenames activate nothing. They contain only public data, can be rebuilt, and never change `sysnode.db` or its schema. The existing deployment preserves `data/`. The feature needs no SQL migration, but the opt-in environment values and nginx route must be configured @@ -66,14 +75,26 @@ HTTP 200 contains only: ```text { version: 1, chainId: 1, poolAddress, asOfBlockNumber, asOfBlockHash, - leafCount, root, leaves: ["0x<64 lowercase hex digits>", ...] } + leafCount, root, leaves: ["0x<64 lowercase hex digits>", ...], + withdrawnNullifiers: ["0x<64 lowercase hex digits>", ...] } ``` -Deposits are ordered by canonical event position and must have contiguous +The additive `withdrawnNullifiers` field is required in new cache files. Old +deposit-only snapshots and version-1 checkpoints are discarded and rebuilt, +never interpreted as an empty spent list. Checkpoints now use internal version 2. +Each array is bounded to 2^20 entries, withdrawals cannot outnumber deposits, and +the complete serialized snapshot/file is bounded to 160 MiB. Existing clients +which project the deposit fields can keep using public version 1; clients using +spent-note status must require the added field. + +A single OR-topic request scans Deposit and Withdrawal events over the same +block range. Deposits are ordered by canonical event position and must have contiguous indices starting at zero. The server validates field ranges, the pool's 20-level configuration and denomination, and matches total count and root metadata at the pinned finalized block. It rejects removed/malformed logs, -missing indices, count mismatches and changed block anchors. JSON-RPC responses +missing indices, duplicate commitments/nullifiers, inconsistent event ordering, +count mismatches and changed block anchors. Withdrawal nullifiers are unique +and appear in event order at the same snapshot block as the deposits. JSON-RPC responses have a 10-second timeout, 4-MiB bound, no redirects and exact response-ID checks. The server does **not** independently reconstruct the MiMC root or authenticate @@ -81,6 +102,16 @@ the configured RPC cryptographically. Pali must reconstruct the tree and check the returned root against its own trusted chain observation before use. This service is a public-data accelerator, not a replacement for wallet verification. +**The deposit root does not authenticate completeness of the withdrawal/nullifier +feed.** A missing withdrawal can pass the deposit-count/root checks. Spent-status +users additionally rely on the configured RPC and scanner returning all canonical +Withdrawal events; checksums and block fences do not remove that trust assumption. +An old or partial snapshot must not be called a current unspent result. No API +accepts a note or a per-user nullifier to resolve this on the server. +Even a fresh finalized snapshot does not include withdrawals in newer blocks or +the mempool; absence from its list is an observation at that block, not a +guarantee that a withdrawal will succeed now. + Complete snapshots remain immutable while a refresh is incomplete or fails. `X-Privacy-Index-State` reports `ready`, `refreshing`, or `stale`; the block/hash in the response remain the observation anchor, not the current HTTP time. @@ -90,7 +121,8 @@ Strong ETags support `If-None-Match`/304 with the same freshness header. A persisted snapshot is withheld after restart until its block/hash, on-chain count and root are checked again. Known checkpoint/snapshot reorgs -discard the affected pool's files and restart from deployment. A gap discards +discard the affected pool's data files and restart from deployment; its +activation marker is retained. A gap discards unfinished work and retries from the last complete snapshot or deployment. An RPC outage does not destroy a previously validated in-memory snapshot; cold starts during an outage return 503. Invalid/corrupt files are never served. @@ -106,5 +138,6 @@ npm run audit:prod Tests use synthetic RPCs and temporary directories and never make mainnet writes. They cover restart/resume, gaps, reorgs, unavailable sources, -deduplication, atomic snapshot availability, malformed inputs and anonymous +deduplication, withdrawal coverage and old-cache rebuilds, periodic refresh, +timer cleanup, atomic snapshot availability, malformed inputs and anonymous HTTP/ETag behavior. No deployment or live pool claim is part of these tests. diff --git a/lib/appFactory.js b/lib/appFactory.js index ff33b49..75faeb2 100644 --- a/lib/appFactory.js +++ b/lib/appFactory.js @@ -551,6 +551,7 @@ function createApp({ runAtomic: services.runAtomic, recoveryMonitor, recoveryRepository: services.recoveryRepository, + privacyIndex, }; } diff --git a/lib/privacyIndex.js b/lib/privacyIndex.js index 6f96791..776859d 100644 --- a/lib/privacyIndex.js +++ b/lib/privacyIndex.js @@ -2,25 +2,37 @@ const { createHash } = require('crypto'); const { performance } = require('perf_hooks'); -const { getPool, MAX_LEAVES } = require('./privacyRegistry'); +const { getPool, MAX_LEAVES, fieldWord } = require('./privacyRegistry'); const { createPrivacyRpc } = require('./privacyRpc'); const { createPrivacyStore } = require('./privacyStore'); -// One process-wide queue. Only a request can enqueue work; no boot-time RPC, -// timers, wallet records or account-specific queries are involved. +// One process-wide queue. A public request activates a pool durably; only those +// pools resume after restart and receive subsequent bounded background refresh. function createPrivacyIndex({ rpc, store = createPrivacyStore(), now = Date.now, chunkBlocks = 5000, maxChunks = 16, jobBudgetMs = 45000, - refreshMs = 60000, retryMs = 30000, log = () => {} } = {}) { + refreshMs = 60000, retryMs = 30000, continueMs = 1000, + autoStart = true, setTimeoutImpl = setTimeout, clearTimeoutImpl = clearTimeout, + log = () => {} } = {}) { if (!rpc) throw new Error('privacy_rpc_required'); + for (const [value, cap] of [[chunkBlocks, 5000], [maxChunks, 16], [jobBudgetMs, 45000]]) { + if (!Number.isSafeInteger(value) || value < 1 || value > cap) throw new Error('privacy_invalid_work_limit'); + } const states = new Map(); + const active = new Set(); const queue = []; let running = null; + let startup = Promise.resolve(); + let timer = null; + let started = false; + let closed = false; + let restoreFailed = false; async function stateFor(pool) { if (!states.has(pool.id)) { const state = { snapshot: null, encoded: null, etag: null, checkpoint: null, candidate: null, syncing: false, retryAt: 0, error: false }; state.loaded = Promise.all([store.load(pool.id, 'snapshot'), store.load(pool.id, 'checkpoint')]) - .then(([snapshot, checkpoint]) => { state.candidate = snapshot; state.checkpoint = checkpoint; }); + .then(([snapshot, checkpoint]) => { state.candidate = snapshot; state.checkpoint = checkpoint; }) + .catch((error) => { states.delete(pool.id); throw error; }); states.set(pool.id, state); } const state = states.get(pool.id); @@ -38,90 +50,121 @@ function createPrivacyIndex({ rpc, store = createPrivacyStore(), now = Date.now, await store.remove(pool.id, 'snapshot'); await store.remove(pool.id, 'checkpoint'); } - async function verifySaved(pool, snapshot, head) { + async function verifySaved(pool, snapshot, head, jobRpc) { if (snapshot.asOfBlockNumber > head.number) throw new Error('privacy_reorg'); - const header = await rpc.block(snapshot.asOfBlockNumber); + const header = await jobRpc.block(snapshot.asOfBlockNumber); if (header.hash !== snapshot.asOfBlockHash) throw new Error('privacy_reorg'); - const observed = await rpc.state(pool, header); + const observed = await jobRpc.state(pool, header); if (observed.nextIndex !== snapshot.leafCount || observed.root !== snapshot.root) throw new Error('privacy_reorg'); } async function sync(pool, state) { const deadline = performance.now() + jobBudgetMs; + const beforeRequest = () => { + if (closed) throw new Error('privacy_index_closed'); + if (performance.now() >= deadline) throw new Error('privacy_budget_exhausted'); + }; + const jobRpc = Object.fromEntries(['head', 'block', 'state', 'events', 'assertBlock'].map((name) => [name, (...args) => { + beforeRequest(); + return rpc[name](...args, { beforeRequest }); + }])); try { - const head = await rpc.head(); + const head = await jobRpc.head(); if (head.number < pool.deploymentBlock) throw new Error('privacy_pool_not_deployed'); // In-memory snapshots are also checked before extending. On restart the // saved snapshot is withheld until this independent canonical read passes. const saved = state.snapshot || state.candidate; if (saved) { - await verifySaved(pool, saved, head); + await verifySaved(pool, saved, head, jobRpc); if (!state.snapshot) publish(state, saved); } let checkpoint = state.checkpoint; if (checkpoint && state.snapshot && checkpoint.targetBlock < state.snapshot.asOfBlockNumber) checkpoint = null; if (checkpoint) { if (checkpoint.targetBlock > head.number || - (await rpc.block(checkpoint.targetBlock)).hash !== checkpoint.targetHash || + (await jobRpc.block(checkpoint.targetBlock)).hash !== checkpoint.targetHash || (checkpoint.scannedBlock >= pool.deploymentBlock && - (await rpc.block(checkpoint.scannedBlock)).hash !== checkpoint.scannedHash)) throw new Error('privacy_reorg'); + (await jobRpc.block(checkpoint.scannedBlock)).hash !== checkpoint.scannedHash)) throw new Error('privacy_reorg'); } else if (state.snapshot && state.snapshot.asOfBlockNumber === head.number) { state.error = false; state.retryAt = now() + refreshMs; return; } else { checkpoint = { - version: 1, poolAddress: pool.address, + version: 2, poolAddress: pool.address, scannedBlock: state.snapshot?.asOfBlockNumber ?? pool.deploymentBlock - 1, scannedHash: state.snapshot?.asOfBlockHash ?? null, targetBlock: head.number, targetHash: head.hash, leaves: state.snapshot ? state.snapshot.leaves.slice() : [], + withdrawnNullifiers: state.snapshot ? state.snapshot.withdrawnNullifiers.slice() : [], }; } state.checkpoint = checkpoint; - const target = await rpc.block(checkpoint.targetBlock); + const target = await jobRpc.block(checkpoint.targetBlock); if (target.hash !== checkpoint.targetHash) throw new Error('privacy_reorg'); - const expected = await rpc.state(pool, target); + const expected = await jobRpc.state(pool, target); if (checkpoint.leaves.length > expected.nextIndex) throw new Error('privacy_index_gap'); - let blockSpan = Math.min(checkpoint.blockSpan || chunkBlocks, chunkBlocks); - for (let chunk = 0; checkpoint.scannedBlock < target.number && chunk < maxChunks && performance.now() < deadline; chunk++) { + const leafSet = new Set(checkpoint.leaves); + const withdrawnSet = new Set(checkpoint.withdrawnNullifiers); + let blockSpan = Math.min(checkpoint.blockSpan || state.blockSpan || chunkBlocks, chunkBlocks); + let sparseChunks = 0; + for (let chunk = 0; !closed && checkpoint.scannedBlock < target.number && chunk < maxChunks && performance.now() < deadline; chunk++) { const from = checkpoint.scannedBlock + 1; let to, boundary, rows; for (;;) { to = Math.min(target.number, from + blockSpan - 1); - boundary = await rpc.block(to); - try { rows = await rpc.deposits(pool, from, to); break; } + boundary = await jobRpc.block(to); + try { rows = await jobRpc.events(pool, from, to); break; } catch (error) { // Read-only provider size/range limits can be retried on a smaller // public range. The job deadline and a one-block floor bound this. if (!['privacy_rpc_unavailable', 'privacy_rpc_rejected', 'privacy_logs_too_large'].includes(error.message) || to === from || performance.now() >= deadline) throw error; blockSpan = Math.max(1, Math.floor((to - from + 1) / 2)); + sparseChunks = 0; } } // Validate the complete chunk before mutating its checkpoint. let next = checkpoint.leaves.length; let previous = null; + const leaves = [], nullifiers = []; for (const row of rows) { - if (row.leafIndex !== next || next >= expected.nextIndex || next >= MAX_LEAVES || - (previous && row.blockNumber === previous.blockNumber && row.logIndex <= previous.logIndex) || + if (row.blockNumber < from || row.blockNumber > to || + (previous && (row.blockNumber < previous.blockNumber || + (row.blockNumber === previous.blockNumber && (row.logIndex <= previous.logIndex || + row.txIndex < previous.txIndex || row.blockHash !== previous.blockHash || + (row.txIndex === previous.txIndex && row.txHash !== previous.txHash))))) || (row.blockNumber === boundary.number && row.blockHash !== boundary.hash)) throw new Error('privacy_index_gap'); - previous = row; next++; + if (row.kind === 'deposit') { + fieldWord(row.leaf); + if (row.leafIndex !== next || next >= expected.nextIndex || next >= MAX_LEAVES || leafSet.has(row.leaf)) throw new Error('privacy_index_gap'); + leafSet.add(row.leaf); leaves.push(row.leaf); next++; + } else if (row.kind === 'withdrawal') { + fieldWord(row.nullifierHash); + if (withdrawnSet.has(row.nullifierHash) || withdrawnSet.size >= next) throw new Error('privacy_index_gap'); + withdrawnSet.add(row.nullifierHash); nullifiers.push(row.nullifierHash); + } else throw new Error('privacy_index_gap'); + previous = row; } - await rpc.assertBlock(boundary); - checkpoint.leaves.push(...rows.map((row) => row.leaf)); + await jobRpc.assertBlock(boundary); + checkpoint.leaves.push(...leaves); + checkpoint.withdrawnNullifiers.push(...nullifiers); checkpoint.scannedBlock = to; checkpoint.scannedHash = boundary.hash; - checkpoint.blockSpan = blockSpan; + sparseChunks = rows.length <= 128 && to - from + 1 === blockSpan ? sparseChunks + 1 : 0; + if (sparseChunks >= 3) { blockSpan = Math.min(chunkBlocks, blockSpan * 2); sparseChunks = 0; } + checkpoint.blockSpan = state.blockSpan = blockSpan; } // A reorg during any read cannot publish a mixed-fork view. - await rpc.assertBlock(target); + await jobRpc.assertBlock(target); if (checkpoint.scannedBlock === target.number) { if (checkpoint.leaves.length !== expected.nextIndex) throw new Error('privacy_index_gap'); const snapshot = { version: 1, chainId: 1, poolAddress: pool.address, asOfBlockNumber: target.number, asOfBlockHash: target.hash, - leafCount: checkpoint.leaves.length, root: expected.root, leaves: checkpoint.leaves }; + leafCount: checkpoint.leaves.length, root: expected.root, leaves: checkpoint.leaves, + withdrawnNullifiers: checkpoint.withdrawnNullifiers }; // The root comes from the pool at this exact block. Wallets must rebuild // MiMC independently before trusting these public RPC-provided leaves. + // That root does NOT authenticate completeness of the withdrawal feed. await store.save(pool.id, 'snapshot', snapshot); publish(state, snapshot); state.checkpoint = null; @@ -129,10 +172,15 @@ function createPrivacyIndex({ rpc, store = createPrivacyStore(), now = Date.now, state.retryAt = now() + refreshMs; } else { await store.save(pool.id, 'checkpoint', checkpoint); - state.retryAt = 0; // the next explicit poll may advance another bounded job + state.retryAt = now() + continueMs; } state.error = false; } catch (error) { + if (['privacy_budget_exhausted', 'privacy_index_closed'].includes(error.message)) { + if (state.checkpoint) await store.save(pool.id, 'checkpoint', state.checkpoint); + state.retryAt = now() + continueMs; + return; + } if (error.message === 'privacy_reorg') await reset(pool, state); if (error.message === 'privacy_index_gap') { // Missing new logs invalidate only unfinished work. A previously @@ -146,24 +194,62 @@ function createPrivacyIndex({ rpc, store = createPrivacyStore(), now = Date.now, } } function enqueue(pool, state) { - if (state.syncing || now() < state.retryAt) return; + if (closed || state.syncing || now() < state.retryAt) return; state.syncing = true; queue.push({ pool, state }); if (running) return; running = (async () => { - while (queue.length) { + while (!closed && queue.length) { const next = queue.shift(); try { await sync(next.pool, next.state); } catch (_) { next.state.error = true; next.state.retryAt = now() + retryMs; } finally { next.state.syncing = false; } } - })().finally(() => { running = null; }); + })().finally(() => { running = null; scheduleNext(); }); + } + async function activate(pool, state) { + if (!state.activation) state.activation = store.activate(pool.id).then(() => { active.add(pool.id); }) + .catch((error) => { state.activation = null; throw error; }); + await state.activation; + } + function scheduleNext() { + if (timer) { clearTimeoutImpl(timer); timer = null; } + if (!started || closed || running || (!active.size && !restoreFailed)) return; + const wait = restoreFailed ? retryMs : Math.min(...[...active].map((id) => Math.max(0, (states.get(id)?.retryAt || 0) - now()))); + timer = setTimeoutImpl(() => { timer = null; return tick(); }, Math.max(1, wait)); + if (timer && typeof timer.unref === 'function') timer.unref(); + } + async function restore() { + for (const id of await store.listActive()) { + const pool = getPool(id); + if (!pool || closed) continue; + const state = await stateFor(pool); + await activate(pool, state); + enqueue(pool, state); + } + restoreFailed = false; + } + async function tick() { + if (closed) return; + try { + if (restoreFailed) await restore(); + for (const id of active) enqueue(getPool(id), await stateFor(getPool(id))); + } catch (_) { restoreFailed = true; log('privacy_resume_failed', { code: 'privacy_storage_unavailable' }); } + finally { scheduleNext(); } + } + function start() { + if (started || closed) return startup; + started = true; + startup = restore().catch(() => { restoreFailed = true; log('privacy_resume_failed', { code: 'privacy_storage_unavailable' }); }) + .finally(scheduleNext); + return startup; } async function read(id) { + if (closed) return { status: 503, body: { error: 'privacy_index_closed' } }; const pool = getPool(id); if (!pool) return { status: 404, body: { error: 'unknown_privacy_pool' } }; let state; - try { state = await stateFor(pool); } + try { await startup; state = await stateFor(pool); await activate(pool, state); } catch (_) { return { status: 503, body: { error: 'privacy_index_unavailable' } }; } enqueue(pool, state); if (state.snapshot) return { status: 200, body: state.snapshot, encoded: state.encoded, etag: state.etag, @@ -172,7 +258,15 @@ function createPrivacyIndex({ rpc, store = createPrivacyStore(), now = Date.now, return { status: 202, body: { phase: 'syncing', scannedBlock: state.checkpoint?.scannedBlock ?? pool.deploymentBlock - 1, startBlock: pool.deploymentBlock, targetBlock: state.checkpoint?.targetBlock ?? pool.deploymentBlock - 1 } }; } - return { read, whenIdle: () => running || Promise.resolve() }; + async function whenIdle() { await startup; while (running) await running; } + async function close() { + closed = true; + if (timer) { clearTimeoutImpl(timer); timer = null; } + for (const item of queue.splice(0)) item.state.syncing = false; + await whenIdle(); + } + if (autoStart) start(); + return { read, whenIdle, start, tick, close }; } function configuredPrivacyIndex({ enabled = process.env.PRIVACY_INDEX_ENABLED, diff --git a/lib/privacyRpc.js b/lib/privacyRpc.js index 5acba0b..d251f6b 100644 --- a/lib/privacyRpc.js +++ b/lib/privacyRpc.js @@ -6,12 +6,14 @@ const { MAX_LEAVES, word, fieldWord } = require('./privacyRegistry'); const ABI = new Interface([ 'event Deposit(bytes32 indexed commitment,uint32 leafIndex,uint256 timestamp)', + 'event Withdrawal(address to,bytes32 nullifierHash,address indexed relayer,uint256 fee)', 'function levels() view returns (uint32)', 'function denomination() view returns (uint256)', 'function nextIndex() view returns (uint32)', 'function getLastRoot() view returns (bytes32)', ]); const DEPOSIT_TOPIC = ABI.getEvent('Deposit').topicHash; +const WITHDRAWAL_TOPIC = ABI.getEvent('Withdrawal').topicHash; const MAX_RPC_BYTES = 4 * 1024 * 1024; function quantity(value) { @@ -26,7 +28,10 @@ function createPrivacyRpc(rpcUrl, { post = axios.post, now = Date.now } = {}) { try { url = new URL(rpcUrl); } catch (_) { throw new Error('privacy_rpc_url_invalid'); } if (url.protocol !== 'https:' || url.username || url.password || url.hash) throw new Error('privacy_rpc_url_invalid'); let id = 0; - async function call(method, params) { + async function call(method, params, { beforeRequest } = {}) { + // A logical read such as state() uses several HTTP requests. Check the + // caller's close/deadline fence before every one, not just at method entry. + if (beforeRequest) beforeRequest(); const current = ++id; let response; try { @@ -43,30 +48,30 @@ function createPrivacyRpc(rpcUrl, { post = axios.post, now = Date.now } = {}) { if (!Object.hasOwn(body, 'result')) throw new Error('privacy_rpc_invalid_response'); return body.result; } - async function block(tag) { - const value = await call('eth_getBlockByNumber', [typeof tag === 'number' ? toQuantity(tag) : tag, false]); + async function block(tag, context) { + const value = await call('eth_getBlockByNumber', [typeof tag === 'number' ? toQuantity(tag) : tag, false], context); if (!value || typeof value !== 'object') throw new Error('privacy_missing_block'); const result = { number: quantity(value.number), hash: word(value.hash), parentHash: word(value.parentHash), timestamp: quantity(value.timestamp) }; if (typeof tag === 'number' && result.number !== tag) throw new Error('privacy_wrong_block'); return result; } - async function assertBlock(header) { - if ((await block(header.number)).hash !== header.hash) throw new Error('privacy_reorg'); + async function assertBlock(header, context) { + if ((await block(header.number, context)).hash !== header.hash) throw new Error('privacy_reorg'); } - async function head() { - if (quantity(await call('eth_chainId', [])) !== 1) throw new Error('privacy_wrong_chain'); + async function head(context) { + if (quantity(await call('eth_chainId', [], context)) !== 1) throw new Error('privacy_wrong_chain'); // Never silently substitute latest/safe for finalized. - const result = await block('finalized'); + const result = await block('finalized', context); const age = now() - result.timestamp * 1000; if (age < -30000 || age > 30 * 60 * 1000) throw new Error('privacy_stale_chain'); return result; } - async function state(pool, header) { + async function state(pool, header, context) { const tag = toQuantity(header.number); - const code = await call('eth_getCode', [pool.address, tag]); + const code = await call('eth_getCode', [pool.address, tag], context); if (typeof code !== 'string' || !/^0x(?:[0-9a-f]{2})+$/.test(code)) throw new Error('privacy_missing_pool'); async function view(name) { - const result = await call('eth_call', [{ to: pool.address, data: ABI.encodeFunctionData(name), gas: '0x186a0' }, tag]); + const result = await call('eth_call', [{ to: pool.address, data: ABI.encodeFunctionData(name), gas: '0x186a0' }, tag], context); word(result); try { return ABI.decodeFunctionResult(name, result)[0]; } catch (_) { throw new Error('privacy_invalid_pool_state'); } } @@ -76,30 +81,39 @@ function createPrivacyRpc(rpcUrl, { post = axios.post, now = Date.now } = {}) { const root = fieldWord(await view('getLastRoot')); if (levels !== 20n || denomination !== BigInt(pool.denomination) || !Number.isSafeInteger(nextIndex) || nextIndex < 0 || nextIndex > MAX_LEAVES) throw new Error('privacy_pool_mismatch'); - await assertBlock(header); + await assertBlock(header, context); return { nextIndex, root }; } - async function deposits(pool, from, to) { - const rows = await call('eth_getLogs', [{ address: pool.address, fromBlock: toQuantity(from), toBlock: toQuantity(to), topics: [DEPOSIT_TOPIC] }]); + async function events(pool, from, to, context) { + const rows = await call('eth_getLogs', [{ address: pool.address, fromBlock: toQuantity(from), toBlock: toQuantity(to), topics: [[DEPOSIT_TOPIC, WITHDRAWAL_TOPIC]] }], context); if (!Array.isArray(rows)) throw new Error('privacy_invalid_logs'); if (rows.length > 10000) throw new Error('privacy_logs_too_large'); return rows.map((row) => { if (!row || row.removed !== false || row.address !== pool.address || - !Array.isArray(row.topics) || row.topics.length !== 2 || row.topics[0] !== DEPOSIT_TOPIC || - typeof row.data !== 'string' || !/^0x[0-9a-f]{128}$/.test(row.data)) throw new Error('privacy_invalid_log'); - fieldWord(row.topics[1]); + !Array.isArray(row.topics) || row.topics.length !== 2 || + ![DEPOSIT_TOPIC, WITHDRAWAL_TOPIC].includes(row.topics[0]) || + typeof row.data !== 'string' || !/^0x[0-9a-f]+$/.test(row.data) || + row.data.length !== (row.topics[0] === DEPOSIT_TOPIC ? 130 : 194)) throw new Error('privacy_invalid_log'); + word(row.topics[1]); + if (row.topics[0] === WITHDRAWAL_TOPIC && + (BigInt(row.topics[1]) >= (1n << 160n) || !/^0x0{24}/.test(row.data))) throw new Error('privacy_invalid_log'); const blockNumber = quantity(row.blockNumber); if (blockNumber < from || blockNumber > to) throw new Error('privacy_invalid_log'); let parsed; try { parsed = ABI.parseLog(row); } catch (_) { throw new Error('privacy_invalid_log'); } - const leafIndex = Number(parsed.args.leafIndex); - if (!Number.isSafeInteger(leafIndex) || leafIndex < 0 || leafIndex >= MAX_LEAVES) throw new Error('privacy_invalid_log'); - return { leaf: fieldWord(parsed.args.commitment), leafIndex, blockNumber, + const position = { blockNumber, blockHash: word(row.blockHash), txHash: word(row.transactionHash), txIndex: quantity(row.transactionIndex), logIndex: quantity(row.logIndex) }; + if (parsed.name === 'Deposit') { + const leafIndex = Number(parsed.args.leafIndex); + if (!Number.isSafeInteger(leafIndex) || leafIndex < 0 || leafIndex >= MAX_LEAVES) throw new Error('privacy_invalid_log'); + return { ...position, kind: 'deposit', leaf: fieldWord(parsed.args.commitment), leafIndex }; + } + if (parsed.args.fee > BigInt(pool.denomination)) throw new Error('privacy_invalid_log'); + return { ...position, kind: 'withdrawal', nullifierHash: fieldWord(parsed.args.nullifierHash) }; }).sort((a, b) => a.blockNumber - b.blockNumber || a.txIndex - b.txIndex || a.logIndex - b.logIndex); } - return { head, block, assertBlock, state, deposits }; + return { head, block, assertBlock, state, events }; } -module.exports = { createPrivacyRpc, ABI, DEPOSIT_TOPIC, quantity, MAX_RPC_BYTES }; +module.exports = { createPrivacyRpc, ABI, DEPOSIT_TOPIC, WITHDRAWAL_TOPIC, quantity, MAX_RPC_BYTES }; diff --git a/lib/privacyStore.js b/lib/privacyStore.js index fc8afe7..e910f2c 100644 --- a/lib/privacyStore.js +++ b/lib/privacyStore.js @@ -4,19 +4,27 @@ const fs = require('fs/promises'); const path = require('path'); const { createHash, randomBytes } = require('crypto'); const { getPool, MAX_LEAVES, word, fieldWord } = require('./privacyRegistry'); -const MAX_FILE_BYTES = 80 * 1024 * 1024; +const MAX_FILE_BYTES = 160 * 1024 * 1024; + +function validateNullifiers(values, leafCount) { + if (!Array.isArray(values) || values.length > leafCount || new Set(values).size !== values.length) throw new Error('privacy_invalid_nullifiers'); + values.forEach(fieldWord); +} function validateSnapshot(value, pool) { if (!value || value.version !== 1 || value.chainId !== 1 || value.poolAddress !== pool.address || !Number.isSafeInteger(value.asOfBlockNumber) || value.asOfBlockNumber < pool.deploymentBlock || !Array.isArray(value.leaves) || value.leaves.length > MAX_LEAVES || value.leafCount !== value.leaves.length) throw new Error('privacy_invalid_snapshot'); word(value.asOfBlockHash); fieldWord(value.root); value.leaves.forEach(fieldWord); + if (new Set(value.leaves).size !== value.leaves.length) throw new Error('privacy_invalid_snapshot'); + validateNullifiers(value.withdrawnNullifiers, value.leafCount); // Return a strict public projection, never expose file bookkeeping. return { version: 1, chainId: 1, poolAddress: pool.address, asOfBlockNumber: value.asOfBlockNumber, - asOfBlockHash: value.asOfBlockHash, leafCount: value.leaves.length, root: value.root, leaves: value.leaves }; + asOfBlockHash: value.asOfBlockHash, leafCount: value.leaves.length, root: value.root, leaves: value.leaves, + withdrawnNullifiers: value.withdrawnNullifiers }; } function validateCheckpoint(value, pool) { - if (!value || value.version !== 1 || value.poolAddress !== pool.address || + if (!value || value.version !== 2 || value.poolAddress !== pool.address || !Number.isSafeInteger(value.scannedBlock) || value.scannedBlock < pool.deploymentBlock - 1 || !Number.isSafeInteger(value.targetBlock) || value.targetBlock < value.scannedBlock || !Array.isArray(value.leaves) || value.leaves.length > MAX_LEAVES) throw new Error('privacy_invalid_checkpoint'); @@ -25,12 +33,14 @@ function validateCheckpoint(value, pool) { if (value.blockSpan !== undefined && (!Number.isSafeInteger(value.blockSpan) || value.blockSpan < 1 || value.blockSpan > 5000)) throw new Error('privacy_invalid_checkpoint'); if (value.scannedBlock === pool.deploymentBlock - 1 && value.leaves.length) throw new Error('privacy_invalid_checkpoint'); value.leaves.forEach(fieldWord); + if (new Set(value.leaves).size !== value.leaves.length) throw new Error('privacy_invalid_checkpoint'); + validateNullifiers(value.withdrawnNullifiers, value.leaves.length); return value; } function createPrivacyStore(directory = path.resolve('data/privacy-index')) { function file(id, kind) { - if (!getPool(id) || !['snapshot', 'checkpoint'].includes(kind)) throw new Error('privacy_unknown_pool'); + if (!getPool(id) || !['snapshot', 'checkpoint', 'active'].includes(kind)) throw new Error('privacy_unknown_pool'); return path.join(directory, `${id}.${kind}.json`); } async function load(id, kind) { @@ -43,6 +53,10 @@ function createPrivacyStore(directory = path.resolve('data/privacy-index')) { !envelope.value || typeof envelope.value !== 'object' || Array.isArray(envelope.value)) throw new Error('privacy_file_envelope'); const data = JSON.stringify(envelope.value); if (envelope.sha256 !== createHash('sha256').update(data).digest('hex')) throw new Error('privacy_file_checksum'); + if (kind === 'active') { + if (envelope.value.version !== 1 || envelope.value.poolId !== id) throw new Error('privacy_invalid_active_pool'); + return envelope.value; + } return kind === 'snapshot' ? validateSnapshot(envelope.value, getPool(id)) : validateCheckpoint(envelope.value, getPool(id)); } catch (error) { if (error.code === 'ENOENT') return null; @@ -53,7 +67,9 @@ function createPrivacyStore(directory = path.resolve('data/privacy-index')) { } async function save(id, kind, value) { const target = file(id, kind); - if (kind === 'snapshot') validateSnapshot(value, getPool(id)); else validateCheckpoint(value, getPool(id)); + if (kind === 'snapshot') validateSnapshot(value, getPool(id)); + else if (kind === 'checkpoint') validateCheckpoint(value, getPool(id)); + else if (value.version !== 1 || value.poolId !== id) throw new Error('privacy_invalid_active_pool'); const data = JSON.stringify(value); const encoded = JSON.stringify({ sha256: createHash('sha256').update(data).digest('hex'), value }); if (Buffer.byteLength(encoded) > MAX_FILE_BYTES) throw new Error('privacy_file_too_large'); @@ -76,7 +92,19 @@ function createPrivacyStore(directory = path.resolve('data/privacy-index')) { async function remove(id, kind) { try { await fs.unlink(file(id, kind)); } catch (error) { if (error.code !== 'ENOENT') throw new Error('privacy_storage_unavailable'); } } - return { load, save, remove }; + async function listActive() { + let names; + try { names = await fs.readdir(directory); } + catch (error) { if (error.code === 'ENOENT') return []; throw new Error('privacy_storage_unavailable'); } + // Old files identify previously requested pools even when their deposit-only + // payload now requires a rebuild. An unrelated filename activates nothing. + return [...new Set(names.flatMap((name) => { + const match = /^(eth-(?:0\.1|1|10|100))\.(?:active|snapshot|checkpoint)\.json$/.exec(name); + return match && getPool(match[1]) ? [match[1]] : []; + }))]; + } + return { load, save, remove, listActive, + activate: (id) => save(id, 'active', { version: 1, poolId: id }) }; } module.exports = { createPrivacyStore, validateSnapshot, validateCheckpoint, MAX_FILE_BYTES }; diff --git a/tests/privacyIndex.test.js b/tests/privacyIndex.test.js index c7061a6..525defc 100644 --- a/tests/privacyIndex.test.js +++ b/tests/privacyIndex.test.js @@ -3,9 +3,11 @@ const fs = require('fs/promises'); const os = require('os'); const path = require('path'); +const { createHash } = require('crypto'); const { createPrivacyIndex, configuredPrivacyIndex } = require('../lib/privacyIndex'); const { createPrivacyStore } = require('../lib/privacyStore'); const { POOLS } = require('../lib/privacyRegistry'); +const { createPrivacyRpc, ABI } = require('../lib/privacyRpc'); const pool = POOLS[0]; const word = (n) => `0x${BigInt(n).toString(16).padStart(64, '0')}`; @@ -13,29 +15,30 @@ function fixture({ count = 3, blocks = 4 } = {}) { const start = pool.deploymentBlock; const state = { height: start + blocks - 1, fork: 0, fail: false, gap: false, extra: false, reads: 0, active: 0, peak: 0 }; const header = (n) => ({ number: n, hash: word(n + state.fork * 100000000), parentHash: word(n - 1), timestamp: 1700000000 }); - const deposits = Array.from({ length: count }, (_, i) => ({ leaf: word(i + 1), leafIndex: i, + const deposits = Array.from({ length: count }, (_, i) => ({ kind: 'deposit', leaf: word(i + 1), leafIndex: i, blockNumber: start + i, blockHash: header(start + i).hash, txHash: word(i + 100), txIndex: 0, logIndex: i })); + const withdrawals = []; const rpc = { async head() { state.reads++; if (state.fail) throw new Error('privacy_rpc_unavailable'); return header(state.height); }, async block(n) { if (state.fail) throw new Error('privacy_rpc_unavailable'); return header(n); }, async assertBlock(h) { if (h.hash !== header(h.number).hash) throw new Error('privacy_reorg'); }, async state(_pool, h) { return { nextIndex: deposits.filter((d) => d.blockNumber <= h.number).length, root: word(1234) }; }, - async deposits(_pool, from, to) { + async events(_pool, from, to) { state.active++; state.peak = Math.max(state.peak, state.active); await new Promise((resolve) => setImmediate(resolve)); state.active--; - const rows = deposits.filter((d) => d.blockNumber >= from && d.blockNumber <= to); + const rows = [...deposits, ...withdrawals].filter((d) => d.blockNumber >= from && d.blockNumber <= to).sort((a, b) => a.blockNumber - b.blockNumber || a.logIndex - b.logIndex); return state.gap ? rows.filter((d) => d.leafIndex !== 1) : rows; }, }; - return { rpc, state, deposits, header }; + return { rpc, state, deposits, withdrawals, header }; } describe('public privacy index lifecycle', () => { let dir; beforeEach(async () => { dir = await fs.mkdtemp(path.join(os.tmpdir(), 'privacy-index-test-')); }); afterEach(async () => { await fs.rm(dir, { recursive: true, force: true }); }); - const setup = (dir, rpc, extra = {}) => createPrivacyIndex({ rpc, store: createPrivacyStore(dir), refreshMs: 0, retryMs: 0, ...extra }); + const setup = (dir, rpc, extra = {}) => createPrivacyIndex({ rpc, store: createPrivacyStore(dir), autoStart: false, refreshMs: 0, retryMs: 0, continueMs: 0, ...extra }); test('disabled config performs no RPC or file IO and opt-in validates configuration', () => { expect(configuredPrivacyIndex({ enabled: 'false', rpcUrl: 'broken' })).toBeNull(); @@ -57,7 +60,7 @@ describe('public privacy index lifecycle', () => { expect(ready.status).toBe(200); expect(ready.body).toEqual({ version: 1, chainId: 1, poolAddress: pool.address, asOfBlockNumber: f.state.height, asOfBlockHash: f.header(f.state.height).hash, - leafCount: 3, root: word(1234), leaves: [word(1), word(2), word(3)] }); + leafCount: 3, root: word(1234), leaves: [word(1), word(2), word(3)], withdrawnNullifiers: [] }); }); test('simultaneous polls share one job, and other pools cannot run RPC scans in parallel', async () => { @@ -82,8 +85,8 @@ describe('public privacy index lifecycle', () => { test('provider range limits shrink the public scan range without losing progress', async () => { const f = fixture({ blocks: 16 }); - const original = f.rpc.deposits; - f.rpc.deposits = jest.fn(async (p, from, to) => { + const original = f.rpc.events; + f.rpc.events = jest.fn(async (p, from, to) => { if (to - from + 1 > 2) throw new Error('privacy_rpc_unavailable'); return original(p, from, to); }); @@ -91,7 +94,7 @@ describe('public privacy index lifecycle', () => { await index.read(pool.id); await index.whenIdle(); const ready = await index.read(pool.id); await index.whenIdle(); expect(ready.status).toBe(200); expect(ready.body.leafCount).toBe(3); - expect(f.rpc.deposits.mock.calls.length).toBeGreaterThan(8); + expect(f.rpc.events.mock.calls.length).toBeGreaterThan(8); }); test('incomplete gaps are never served as a complete tree and resets are persistent', async () => { @@ -103,8 +106,8 @@ describe('public privacy index lifecycle', () => { }); test('missing trailing deposit fails final count despite contiguous earlier indices', async () => { - const f = fixture(); const original = f.rpc.deposits; - f.rpc.deposits = async (...args) => (await original(...args)).filter((r) => r.leafIndex !== 2); + const f = fixture(); const original = f.rpc.events; + f.rpc.events = async (...args) => (await original(...args)).filter((r) => r.leafIndex !== 2); const index = setup(dir, f.rpc); await index.read(pool.id); await index.whenIdle(); expect((await index.read(pool.id)).status).toBe(503); await index.whenIdle(); }); @@ -127,10 +130,10 @@ describe('public privacy index lifecycle', () => { await index.read(pool.id); await index.whenIdle(); const old = await index.read(pool.id); await index.whenIdle(); f.state.height += 2; - f.deposits.push({ leaf: word(99), leafIndex: 3, blockNumber: f.state.height, + f.deposits.push({ kind: 'deposit', leaf: word(99), leafIndex: 3, blockNumber: f.state.height, blockHash: f.header(f.state.height).hash, txHash: word(999), txIndex: 0, logIndex: 0 }); - const original = f.rpc.deposits; - f.rpc.deposits = async (...args) => (await original(...args)).filter((r) => r.leafIndex !== 3); + const original = f.rpc.events; + f.rpc.events = async (...args) => (await original(...args)).filter((r) => r.leafIndex !== 3); await index.read(pool.id); await index.whenIdle(); const stale = await index.read(pool.id); await index.whenIdle(); expect(stale.status).toBe(200); expect(stale.phase).toBe('stale'); expect(stale.body).toEqual(old.body); @@ -145,8 +148,8 @@ describe('public privacy index lifecycle', () => { }); test('an in-flight target reorg cannot publish a mixed-fork snapshot', async () => { - const f = fixture(); const original = f.rpc.deposits; - f.rpc.deposits = async (...args) => { const rows = await original(...args); f.state.fork = 1; return rows; }; + const f = fixture(); const original = f.rpc.events; + f.rpc.events = async (...args) => { const rows = await original(...args); f.state.fork = 1; return rows; }; const index = setup(dir, f.rpc); await index.read(pool.id); await index.whenIdle(); expect(await createPrivacyStore(dir).load(pool.id, 'snapshot')).toBeNull(); }); @@ -177,4 +180,187 @@ describe('public privacy index lifecycle', () => { const stillReady = await index.read(pool.id); await index.whenIdle(); expect(stillReady.status).toBe(200); expect(stillReady.body).toEqual(old.body); }); + + test('withdrawals are published with deposits at the same pinned block', async () => { + const f = fixture(); + f.withdrawals.push({ kind: 'withdrawal', nullifierHash: word(88), blockNumber: f.state.height, + blockHash: f.header(f.state.height).hash, txHash: word(188), txIndex: 1, logIndex: 8 }); + const index = setup(dir, f.rpc, { chunkBlocks: 1 }); + await index.read(pool.id); await index.whenIdle(); + const ready = await index.read(pool.id); await index.whenIdle(); + expect(ready.body.withdrawnNullifiers).toEqual([word(88)]); + expect(ready.body.asOfBlockNumber).toBe(f.state.height); + }); + + test('duplicate withdrawal nullifiers across ranges cannot publish a snapshot', async () => { + const f = fixture(); + for (const n of [pool.deploymentBlock + 1, pool.deploymentBlock + 2]) { + f.withdrawals.push({ kind: 'withdrawal', nullifierHash: word(88), blockNumber: n, + blockHash: f.header(n).hash, txHash: word(n), txIndex: 1, logIndex: 8 }); + } + const index = setup(dir, f.rpc, { chunkBlocks: 1 }); + await index.read(pool.id); await index.whenIdle(); + expect((await index.read(pool.id)).status).toBe(503); await index.whenIdle(); + expect(await createPrivacyStore(dir).load(pool.id, 'snapshot')).toBeNull(); + }); + + test('old deposit-only snapshots and checkpoints require a full rebuild', async () => { + const old = { version: 1, chainId: 1, poolAddress: pool.address, asOfBlockNumber: pool.deploymentBlock + 3, + asOfBlockHash: word(pool.deploymentBlock + 3), leafCount: 3, root: word(1234), leaves: [word(1), word(2), word(3)] }; + const oldCheckpoint = { version: 1, poolAddress: pool.address, scannedBlock: old.asOfBlockNumber, + scannedHash: old.asOfBlockHash, targetBlock: old.asOfBlockNumber, targetHash: old.asOfBlockHash, leaves: old.leaves }; + for (const [kind, value] of [['snapshot', old], ['checkpoint', oldCheckpoint]]) { + await fs.writeFile(path.join(dir, `${pool.id}.${kind}.json`), JSON.stringify({ value, + sha256: createHash('sha256').update(JSON.stringify(value)).digest('hex') })); + } + const f = fixture(); + f.withdrawals.push({ kind: 'withdrawal', nullifierHash: word(88), blockNumber: f.state.height, + blockHash: f.header(f.state.height).hash, txHash: word(188), txIndex: 1, logIndex: 8 }); + const index = setup(dir, f.rpc); + expect((await index.read(pool.id)).status).toBe(202); await index.whenIdle(); + expect((await index.read(pool.id)).body.withdrawnNullifiers).toEqual([word(88)]); await index.whenIdle(); + }); + + test('adaptive ranges grow back after sparse successful full ranges', async () => { + const f = fixture({ count: 0, blocks: 30 }); + const original = f.rpc.events; + let calls = 0; + f.rpc.events = jest.fn(async (p, from, to) => { + if (++calls === 1) throw new Error('privacy_rpc_rejected'); + return original(p, from, to); + }); + const index = setup(dir, f.rpc, { chunkBlocks: 8 }); + await index.read(pool.id); await index.whenIdle(); + const spans = f.rpc.events.mock.calls.map(([, from, to]) => to - from + 1); + expect(spans.slice(0, 5)).toEqual([8, 4, 4, 4, 8]); + }); + + function scheduler() { + let now = 0; + const timers = new Set(); + return { timers, now: () => now, + setTimeoutImpl: (fn, delay) => { const t = { fn, at: now + delay, unref: jest.fn() }; timers.add(t); return t; }, + clearTimeoutImpl: (t) => timers.delete(t), + advance: async (ms) => { now += ms; const due = [...timers].filter((t) => t.at <= now); for (const t of due) { timers.delete(t); await t.fn(); } }, + }; + } + + test('unused pools stay idle; first request enables durable periodic refresh and close stops it', async () => { + const f = fixture(); const s = scheduler(); + const index = setup(dir, f.rpc, { ...s, autoStart: true, refreshMs: 50, continueMs: 10 }); + await index.whenIdle(); expect(f.state.reads).toBe(0); expect(s.timers.size).toBe(0); + await index.read(pool.id); await index.whenIdle(); + expect([...s.timers][0].unref).toHaveBeenCalled(); + f.state.height++; + f.withdrawals.push({ kind: 'withdrawal', nullifierHash: word(77), blockNumber: f.state.height, + blockHash: f.header(f.state.height).hash, txHash: word(177), txIndex: 1, logIndex: 8 }); + await s.advance(50); await index.whenIdle(); + expect((await index.read(pool.id)).body.withdrawnNullifiers).toEqual([word(77)]); + expect(f.state.reads).toBe(2); + await index.close(); expect(s.timers.size).toBe(0); + await s.advance(500); expect(f.state.reads).toBe(2); + expect((await index.read(pool.id)).status).toBe(503); + }); + + test('restart resumes incomplete activated pools without another user request', async () => { + const f = fixture(); + const first = setup(dir, f.rpc, { chunkBlocks: 1, maxChunks: 1 }); + await first.read(pool.id); await first.whenIdle(); await first.close(); + const s = scheduler(); + const resumed = setup(dir, f.rpc, { ...s, autoStart: true, chunkBlocks: 1, maxChunks: 1, continueMs: 10, refreshMs: 100 }); + await resumed.whenIdle(); + expect(f.state.reads).toBe(2); + await s.advance(10); await resumed.whenIdle(); + await s.advance(10); await resumed.whenIdle(); + expect((await resumed.read(pool.id)).status).toBe(200); + expect(await createPrivacyStore(dir).listActive()).toEqual([pool.id]); + await resumed.close(); + }); + + test('close during an in-flight read does not start another RPC or rearm a timer', async () => { + const f = fixture(); const s = scheduler(); + let release; + const original = f.rpc.head; + f.rpc.head = jest.fn(async () => { await new Promise((resolve) => { release = resolve; }); return original(); }); + f.rpc.block = jest.fn(f.rpc.block); + const index = setup(dir, f.rpc, { ...s, autoStart: true }); + await index.read(pool.id); + const closing = index.close(); + release(); await closing; + expect(f.rpc.block).not.toHaveBeenCalled(); + expect(s.timers.size).toBe(0); + }); + + test('the job deadline also bounds metadata reads before a scan begins', async () => { + const f = fixture(); + f.rpc.head = async () => { await new Promise((resolve) => setTimeout(resolve, 10)); return f.header(f.state.height); }; + f.rpc.block = jest.fn(f.rpc.block); + const index = setup(dir, f.rpc, { jobBudgetMs: 1 }); + await index.read(pool.id); await index.whenIdle(); + expect(f.rpc.block).not.toHaveBeenCalled(); + expect(await createPrivacyStore(dir).load(pool.id, 'snapshot')).toBeNull(); + await index.close(); + }); + + function adapterFixture({ pauseCode = true } = {}) { + let releaseCode, enteredCode; + const codeEntered = new Promise((resolve) => { enteredCode = resolve; }); + const codeReleased = new Promise((resolve) => { releaseCode = resolve; }); + const calls = []; + const post = async (_url, payload) => { + calls.push(payload.method); + let result; + if (payload.method === 'eth_chainId') result = '0x1'; + else if (payload.method === 'eth_getBlockByNumber') { + const n = payload.params[0] === 'finalized' ? pool.deploymentBlock + 3 : Number(BigInt(payload.params[0])); + result = { number: `0x${n.toString(16)}`, hash: word(n), parentHash: word(n - 1), timestamp: '0x6553f100' }; + } else if (payload.method === 'eth_getCode') { + enteredCode(); + if (pauseCode) await codeReleased; + result = '0x6000'; + } else if (payload.method === 'eth_call') { + const fn = ABI.parseTransaction(payload.params[0]).name; + result = ABI.encodeFunctionResult(fn, [{ levels: 20, denomination: pool.denomination, nextIndex: 0, getLastRoot: word(1234) }[fn]]); + } else if (payload.method === 'eth_getLogs') result = []; + else throw new Error('unexpected_fixture_rpc'); + return { status: 200, data: { jsonrpc: '2.0', id: payload.id, result } }; + }; + return { rpc: createPrivacyRpc('https://archive.example', { post, now: () => 1700000000000 }), + calls, codeEntered, releaseCode }; + } + + test('close fences nested actual-adapter requests after an in-flight eth_getCode', async () => { + const f = adapterFixture(); const index = setup(dir, f.rpc); + await index.read(pool.id); await f.codeEntered; + const before = [...f.calls]; + const closing = index.close(); + f.releaseCode(); await closing; + expect(before.at(-1)).toBe('eth_getCode'); + expect(f.calls).toEqual(before); + expect(f.calls).not.toContain('eth_call'); + expect(await createPrivacyStore(dir).load(pool.id, 'snapshot')).toBeNull(); + }); + + test('deadline fences nested actual-adapter requests after an in-flight eth_getCode', async () => { + const f = adapterFixture(); const index = setup(dir, f.rpc, { jobBudgetMs: 200 }); + await index.read(pool.id); await f.codeEntered; + const before = [...f.calls]; + await new Promise((resolve) => setTimeout(resolve, 225)); + f.releaseCode(); await index.whenIdle(); + expect(f.calls).toEqual(before); + expect(f.calls).not.toContain('eth_call'); + expect(await createPrivacyStore(dir).load(pool.id, 'snapshot')).toBeNull(); + await index.close(); + }); + + test('healthy actual-adapter control still reads pool state and publishes the snapshot', async () => { + const f = adapterFixture({ pauseCode: false }); const index = setup(dir, f.rpc, { refreshMs: 60000 }); + await index.read(pool.id); await index.whenIdle(); + const ready = await index.read(pool.id); await index.whenIdle(); + expect(ready.status).toBe(200); + expect(ready.body).toMatchObject({ leafCount: 0, withdrawnNullifiers: [] }); + expect(f.calls.filter((method) => method === 'eth_call')).toHaveLength(4); + expect(f.calls).toContain('eth_getLogs'); + await index.close(); + }); }); diff --git a/tests/privacyRpc.test.js b/tests/privacyRpc.test.js index ced7d27..ed2bfd7 100644 --- a/tests/privacyRpc.test.js +++ b/tests/privacyRpc.test.js @@ -1,6 +1,6 @@ 'use strict'; -const { createPrivacyRpc, ABI, DEPOSIT_TOPIC, quantity, MAX_RPC_BYTES } = require('../lib/privacyRpc'); +const { createPrivacyRpc, ABI, DEPOSIT_TOPIC, WITHDRAWAL_TOPIC, quantity, MAX_RPC_BYTES } = require('../lib/privacyRpc'); const { POOLS, FIELD_SIZE } = require('../lib/privacyRegistry'); const pool = POOLS[0]; const w = (n) => `0x${BigInt(n).toString(16).padStart(64, '0')}`; @@ -23,9 +23,32 @@ function fixture(overrides = {}) { test('RPC has strict transport guards and only fixed read methods and pool address', async () => { const f = fixture(); const head = await f.rpc.head(); expect((await f.rpc.state(pool, head)).nextIndex).toBe(1); - expect((await f.rpc.deposits(pool, 12000000, 12000000))[0].leaf).toBe(w(1)); + expect((await f.rpc.events(pool, 12000000, 12000000))[0].leaf).toBe(w(1)); expect(new Set(f.post.mock.calls.map(([, p]) => p.method))).toEqual(new Set(['eth_chainId', 'eth_getBlockByNumber', 'eth_getCode', 'eth_call', 'eth_getLogs'])); for (const [, , opts] of f.post.mock.calls) expect(opts).toMatchObject({ timeout: 10000, maxRedirects: 0, maxContentLength: MAX_RPC_BYTES }); + const logCall = f.post.mock.calls.find(([, p]) => p.method === 'eth_getLogs'); + expect(logCall[1].params[0].topics).toEqual([[DEPOSIT_TOPIC, WITHDRAWAL_TOPIC]]); +}); + +test('strictly decodes the canonical Withdrawal ABI and sorts both event kinds', async () => { + const f = fixture(); + const position = f.values.eth_getLogs[0]; + const withdrawal = ABI.encodeEventLog(ABI.getEvent('Withdrawal'), ['0x' + '11'.repeat(20), w(44), '0x' + '22'.repeat(20), 0]); + f.values.eth_getLogs.unshift({ ...position, ...withdrawal, logIndex: '0x2' }); + const rows = await f.rpc.events(pool, 12000000, 12000000); + expect(rows.map((row) => row.kind)).toEqual(['deposit', 'withdrawal']); + expect(rows[1].nullifierHash).toBe(w(44)); +}); + +test.each(['field', 'fee', 'padding', 'topic'])('rejects malformed withdrawal %s', async (kind) => { + const f = fixture(); + const withdrawal = ABI.encodeEventLog(ABI.getEvent('Withdrawal'), ['0x' + '11'.repeat(20), w(44), '0x' + '22'.repeat(20), 0]); + if (kind === 'field') withdrawal.data = withdrawal.data.slice(0, 66) + w(FIELD_SIZE).slice(2) + withdrawal.data.slice(130); + if (kind === 'fee') withdrawal.data = withdrawal.data.slice(0, 130) + w(BigInt(pool.denomination) + 1n).slice(2); + if (kind === 'padding') withdrawal.data = '0x01' + withdrawal.data.slice(4); + if (kind === 'topic') withdrawal.topics[1] = w(1n << 200n); + f.values.eth_getLogs = [{ ...f.values.eth_getLogs[0], ...withdrawal }]; + await expect(f.rpc.events(pool, 12000000, 12000000)).rejects.toThrow(); }); test.each(['0x01', '0x', '0x-1', '1', 1, '0x20000000000000'])('rejects malformed/unsafe RPC quantity %p', (value) => { expect(() => quantity(value)).toThrow(); }); test('rejects wrong chain, stale headers and wrong block numbers', async () => { @@ -51,6 +74,26 @@ test('rejects wrong pool parameters', async () => { const rpc = createPrivacyRpc('https://a.example', { post }); await expect(rpc.state(pool, { number: 12000000 })).rejects.toThrow('privacy_pool_mismatch'); }); +test('the request fence reaches head and the final state block check', async () => { + const headFixture = fixture(); let calls = 0; + await expect(headFixture.rpc.head({ beforeRequest: () => { if (++calls > 1) throw new Error('stopped'); } })).rejects.toThrow('stopped'); + expect(headFixture.post.mock.calls.map(([, payload]) => payload.method)).toEqual(['eth_chainId']); + + const stateFixture = fixture(); calls = 0; + await expect(stateFixture.rpc.state(pool, { number: 12000000, hash: w(99) }, { + beforeRequest: () => { if (++calls > 5) throw new Error('stopped'); }, + })).rejects.toThrow('stopped'); + expect(stateFixture.post.mock.calls.map(([, payload]) => payload.method)).toEqual([ + 'eth_getCode', 'eth_call', 'eth_call', 'eth_call', 'eth_call', + ]); +}); +test('assertBlock cannot bypass a closed request fence', async () => { + const f = fixture(); + await expect(f.rpc.assertBlock({ number: 12000000, hash: w(99) }, { + beforeRequest: () => { throw new Error('stopped'); }, + })).rejects.toThrow('stopped'); + expect(f.post).not.toHaveBeenCalled(); +}); test.each(['removed', 'address', 'field', 'range', 'data', 'index'])('rejects malicious %s log', async (kind) => { const f = fixture(); const row = f.values.eth_getLogs[0]; if (kind === 'removed') row.removed = true; @@ -59,5 +102,5 @@ test.each(['removed', 'address', 'field', 'range', 'data', 'index'])('rejects ma if (kind === 'range') row.blockNumber = '0x1'; if (kind === 'data') row.data += '00'; if (kind === 'index') row.logIndex = '0x00'; - await expect(f.rpc.deposits(pool, 12000000, 12000000)).rejects.toThrow(); + await expect(f.rpc.events(pool, 12000000, 12000000)).rejects.toThrow(); });