diff --git a/README.md b/README.md index b9a5993..70f886d 100644 --- a/README.md +++ b/README.md @@ -158,16 +158,17 @@ Global behavior: | **`--password`** | Wallet unlock password. In non-interactive mode, required where the wallet is encrypted. Value can be a literal string or a path to a file containing the password. | | **`--without-tor`** | Disable Tor for non-RPC HTTP (default: Tor on for private-protocol and Pimlico-backed commands, including `transfer` / `transact-raw` / names). Or set `KOHAKU_WITHOUT_TOR=1`. Ethereum RPC stays clearnet. Review contacts with `view-network-traffic`. | | **Proving artifacts** | Railgun/Tornado keys live under `/proving-artifacts`. Pre-warm with `fetch-artifacts`. Remote base: `KOHAKU_ARTIFACTS_BASE_URL` (default `https://artifacts.0000000000.org`). Large Tor GETs: `KOHAKU_TOR_CDN_TIMEOUT_MS` (default `45000`). Debug: `KOHAKU_TOR_DEBUG=1`. | +| **Public-sync cache** | Shared **Railgun Subsquid** and **Tornado saga** HTTP pages live under `/public-sync-cache` and speed up those syncs (`balances`, `shield`, `unshield`). Prefetch with `fetch-sync-cache`. Snapshot base: `KOHAKU_SYNC_CACHE_BASE_URL` (default `https://artifacts.0000000000.org/sync-cache/v1`). Snapshot is historical; live HTTP still fills anything newer. Never evicts — at `KOHAKU_PUBLIC_SYNC_CACHE_MAX_BYTES` (default 1 GiB) new pages stop being stored instead. Privacy Pools is **not** covered (its cold sync is bundled state JSON plus `eth_getLogs`, which is never HTTP-cached). Wipe with `kohaku clear-tor-cache --public-sync`. | --- ### `create-wallet ` -Create a BIP-39 seed wallet encrypted on disk. +Create a BIP-39 seed wallet encrypted on disk. The `` argument is a single token (no spaces) and cannot be `proving-artifacts` or `public-sync-cache` (those are cache directories under `--dataDir`). **New seed:** records the current chain tip in `.stealth-start-block` so later `balances` stealth scans do not walk announcement history from before the wallet existed. Uses `--rpc-url` / `RPC_URL` when set; otherwise a public RPC for mainnet or Sepolia (`--testnet`). -**Import (`--import`):** scans used public HD indexes via RPC. Optionally pass `--stealth-start-block` to set `.stealth-start-block` (you know roughly when the seed was first used); the first `balances` run then discovers stealth payments from that floor, same as for new wallets. +**Import (`--import`):** scans used public HD indexes via RPC. Writes `.stealth-start-block` so the first `balances` stealth scan starts at the Kohaku-schema floor (mainnet `25700000`, Sepolia `11455454`) unless you pass `--stealth-start-block`. That flag can still go back as far as the ERC-5564 announcer deploy block. | Option | Description | |--------|-------------| @@ -175,7 +176,7 @@ Create a BIP-39 seed wallet encrypted on disk. | `--import` | Restore from mnemonic instead of generating a new one. | | `--long-seed` | Generate a 24-word (256-bit) mnemonic instead of the default 12-word (128-bit). Ignored with `--import`. | | `--rpc-url ` | Required with `--import` (or `RPC_URL`) to scan used addresses. Optional for new wallets when writing `.stealth-start-block`. | -| `--stealth-start-block ` | With `--import`: write `.stealth-start-block` for later `balances` stealth scans. New wallets set this automatically from the current tip. | +| `--stealth-start-block ` | With `--import`: write `.stealth-start-block` for later `balances` stealth scans. Default when omitted: mainnet `25700000`, Sepolia `11455454`. New wallets set this automatically from the current tip. | | `--mnemonic ` | Mnemonic (required with `--non-interactive --import`). | | `--password ` | Encryption password (required with `--non-interactive`). | | `--non-interactive` | No prompts; no mnemonic box on create. | @@ -189,7 +190,7 @@ Create a BIP-39 seed wallet encrypted on disk. kohaku create-wallet myWallet --testnet kohaku create-wallet myWallet24 --testnet --long-seed kohaku create-wallet restored --testnet --import --rpc-url "$RPC_URL" -kohaku create-wallet restored --testnet --import --rpc-url "$RPC_URL" --stealth-start-block 5000000 +kohaku create-wallet restored --testnet --import --rpc-url "$RPC_URL" --stealth-start-block 10000000 ``` --- @@ -310,7 +311,7 @@ Show aggregated **public** balances (ETH + default ERC-20s for the chain, plus a By default, private balances are included only for `DEFAULT_PRIVACY_PROTOCOL` (if set). Otherwise only public balances are shown, with a short warning. Pass `--include` to sync one or more protocols explicitly (required for multiple protocols at once, or for any private balance when the env is unset). -`balances` always loads **already-imported** stealth accounts into public totals. It also scans ERC-5564 announcements for new payments (same spinner progress as protocol first-sync). `--stealth-start-block` only sets the history floor; use `--skip-stealth-scan` to skip discovery for a faster run. +`balances` always loads **already-imported** stealth accounts into public totals. It also scans ERC-5564 announcements for new payments (same spinner progress as protocol first-sync). A line of the form `Stealth scan from block {start} · {latest - start} blocks` prints before that progress bar. `--stealth-start-block` sets the history floor (and can back-date below the wallet file, down to the announcer deploy block); use `--skip-stealth-scan` to skip discovery for a faster run. | Option | Description | |--------|-------------| @@ -321,7 +322,7 @@ By default, private balances are included only for `DEFAULT_PRIVACY_PROTOCOL` (i | `--verbose` | Human: per-address public breakdown + private note list for included protocols. JSON: adds `public_account_indexes_by_address` and `private_notes`. | | `--tokensList ` | Extra ERC-20 addresses (comma- or space-separated), merged with chain defaults. | | `--without-tor` | Disable Tor for privacy HTTP when syncing private protocols (default: Tor on). Covers Railgun Subsquid/PPOI, Tornado saga/artifacts, Privacy Pools ASP/fastrelay, etc. RPC stays clearnet. Or set `KOHAKU_WITHOUT_TOR=1`. | -| `--stealth-start-block ` | Floor for the ERC-5564 announcement **scan** (decimal or `0x`-hex); skips older history on first/full scan. When omitted, uses the wallet’s `.stealth-start-block` file if present (written by `create-wallet`). Not a way to skip scanning. | +| `--stealth-start-block ` | Floor for the ERC-5564 announcement **scan** (decimal or `0x`-hex); skips older history on first/full scan. Can back-date below the wallet `.stealth-start-block` (as far as the announcer deploy block). When omitted, uses that file if present, otherwise the Kohaku import default (mainnet `25700000`, Sepolia `11455454`). Not a way to skip scanning. | | `--skip-stealth-scan` | Skip announcement discovery for this run. **Already-imported** stealth accounts still appear in public balances. | | `--non-interactive` | JSON only; requires `--wallet` and `--password`. | | `--dataDir ` | Data root. | @@ -686,9 +687,9 @@ kohaku set-name-reverse-record --wallet testWallet --name alice.eth --broadcast ### `fetch-artifacts` -Download Railgun + Tornado proving artifacts into `/proving-artifacts` so later **prove / unshield** can load circuits from disk without re-fetching (and without Tor→clearnet fallback). Event sync (`balances`, first shield) does **not** need these files — Tornado cold sync is saga CDN + chain, Railgun is Subsquid. +Download Railgun + Tornado proving artifacts into `/proving-artifacts` so later **prove / unshield** can load circuits from disk without re-fetching (and without Tor→clearnet fallback). Event sync (`balances`, first shield) does **not** need these files — Tornado cold sync is saga CDN + chain, Railgun is Subsquid, Privacy Pools is bundled state JSON + `eth_getLogs` + ASP. -With no selectors, downloads the **full set** (~260 MB). Or narrow the download: +With no selectors, downloads the **full proving set** (~260 MB). Narrow with `--variant` / `--poi` / `--tornado` / keys. The public-sync snapshot is a separate concern — see [`fetch-sync-cache`](#fetch-sync-cache). | Option / args | Description | |---------------|-------------| @@ -701,7 +702,7 @@ With no selectors, downloads the **full set** (~260 MB). Or narrow the download: | `--dataDir ` | Data root (cache lives at `/proving-artifacts`). | | `--non-interactive` | JSON summary only. | -Remote base URL: env `KOHAKU_ARTIFACTS_BASE_URL` (default: `https://artifacts.0000000000.org`, same path layout as MacWha `artifacts/`). +Remote proving-artifact base URL: env `KOHAKU_ARTIFACTS_BASE_URL` (default: `https://artifacts.0000000000.org`, same path layout as MacWha `artifacts/`). **Examples:** @@ -714,6 +715,47 @@ kohaku fetch-artifacts railgun/01x03/proving_key.bin.br --- +### `fetch-sync-cache` + +Download a published snapshot of public protocol-sync HTTP pages into `/public-sync-cache`, so a fresh wallet's first `balances` / `shield` / `unshield` replays **Railgun Subsquid** and **Tornado saga** history from disk instead of paging it over Tor. Anything newer than the snapshot is still fetched live and written through, so the cache stays current as you use it. + +Privacy Pools is deliberately **not** in the snapshot: its cold sync is a bundled state JSON plus `eth_getLogs`, and RPC calls never pass through the HTTP cache. Its first sync stays slow. + +The snapshot ships as a **manifest plus ~8 MiB chunks** rather than one large archive. Each chunk is fetched one at a time, checked against the `sha256` in the manifest, and extracted on arrival, so a dropped Tor circuit costs one chunk instead of the whole transfer. Re-running skips chunks whose entries are already on disk, which makes an interrupted download resumable and a partial cache repairable. + +| Option | Description | +|--------|-------------| +| *(none)* | Fetch the manifest, then download and install every chunk not already present. | +| `--force` | Re-download chunks even when all of their entries are already cached. | +| `--pack ` | Publisher mode: pack `/public-sync-cache` into `chunk-NNN.tar.gz` + `manifest.json` in ``. | +| `--chunk-bytes ` | Target compressed bytes per chunk with `--pack` (default `8388608`). | +| `--without-tor` | Download over clearnet. Much faster, and reveals only that this IP fetched public pool data. Or set `KOHAKU_WITHOUT_TOR=1`. | +| `--dataDir ` | Data root (cache lives at `/public-sync-cache`). | +| `--non-interactive` | JSON summary only. | + +Snapshot base URL: env `KOHAKU_SYNC_CACHE_BASE_URL` (default: `https://artifacts.0000000000.org/sync-cache/v1`, a versioned prefix so a newer snapshot cannot break clients pinned to an older manifest). Per-chunk time budget: env `KOHAKU_SYNC_CACHE_CHUNK_TIMEOUT_MS` (default `300000`); each chunk gets 3 attempts. Chunk URLs are excluded from proving-artifact routing, so the 45 s `KOHAKU_TOR_CDN_TIMEOUT_MS` cap does not apply to them. + +Exits non-zero if any chunk ultimately failed. The entries that did land are still valid and usable — re-run to retry the rest. + +**Examples:** + +```bash +# Consumers +kohaku fetch-sync-cache +kohaku fetch-sync-cache --without-tor + +# Publisher: sync the protocols you want covered on each network first, then pack +RPC_URL=https://mainnet-rpc kohaku balances --wallet snap-main --include railgun,tornado --without-tor +RPC_URL=https://sepolia-rpc kohaku balances --wallet snap-sep --include railgun,tornado --without-tor +kohaku fetch-sync-cache --pack ./sync-cache-v1 +``` + +Then upload everything in `./sync-cache-v1` (chunks **and** `manifest.json`) to `artifacts.0000000000.org/sync-cache/v1/`, serving chunks as `application/gzip` and the manifest as `application/json`. To stage a snapshot before publishing, point `KOHAKU_SYNC_CACHE_BASE_URL` at any host that serves those files. + +The cache never evicts: once it reaches `KOHAKU_PUBLIC_SYNC_CACHE_MAX_BYTES` (default 1 GiB) new responses simply stop being stored, so an installed snapshot is never cannibalised to make room for fresher pages. When filling a cache you intend to publish, raise that ceiling for the *sync* runs so later pages are still captured rather than silently dropped. + +--- + ### `view-network-traffic` Browse the per-wallet network traffic log (what the CLI contacted, when, and whether the request went over Tor). Useful for reviewing anonymity risk. @@ -750,12 +792,14 @@ Delete the on-disk [tor-js](https://github.com/privacy-ethereum/tor-js) Arti cac | Option | Description | |--------|-------------| +| `--public-sync` | Also delete `/public-sync-cache` (Railgun Subsquid / Tornado saga HTTP cache). | | `--non-interactive` | Print JSON `{ cleared, path }` instead of a human message. | **Examples:** ```bash kohaku clear-tor-cache +kohaku clear-tor-cache --public-sync kohaku clear-tor-cache --non-interactive ``` @@ -796,9 +840,9 @@ Files include `public-accounts.json`, stealth storage, `rg-storage.json`, `ppv1- ## Tips - **Dry run vs broadcast:** `transfer`, `transact-raw`, `shield`, `unshield`, `init-profile`, and the name commands default to *prepare or simulate only*. Always read the printed transaction data before adding `--broadcast`. -- **Tor (all-but-RPC):** Non-RPC HTTP (Pimlico, Railgun Subsquid/PPOI, Tornado saga/artifacts, Privacy Pools ASP/fastrelay, …) goes through [tor-js](https://github.com/privacy-ethereum/tor-js) by default on `balances` (when syncing private protocols), `shield`, `unshield`, Tornado note import/export, `transfer`, `transact-raw`, and name commands. Ethereum RPC stays clearnet. Use `--without-tor` or `KOHAKU_WITHOUT_TOR=1` to skip. **Saga CDN and proving artifacts are Tor-or-fail** (no clearnet fallback; large GETs time out after `KOHAKU_TOR_CDN_TIMEOUT_MS`, default 45s). First protocol sync (saga / Subsquid / RPC catch-up) shows live progress on the spinner and does not download proving keys. Artifacts are served from `/proving-artifacts` when cached; otherwise fetched from `KOHAKU_ARTIFACTS_BASE_URL` (default: `https://artifacts.0000000000.org`) on **prove / unshield**. Pre-warm keys with `kohaku fetch-artifacts` (optionally `--without-tor` for a one-shot clearnet download). After Tor bootstrap corruption, run `kohaku clear-tor-cache`. Set `KOHAKU_TOR_DEBUG=1` for per-request Tor logs. A keyed RPC URL still identifies you to that provider regardless of Tor. Review with `view-network-traffic --wallet `. +- **Tor (all-but-RPC):** Non-RPC HTTP (Pimlico, Railgun Subsquid/PPOI, Tornado saga/artifacts, Privacy Pools ASP/fastrelay, …) goes through [tor-js](https://github.com/privacy-ethereum/tor-js) by default on `balances` (when syncing private protocols), `shield`, `unshield`, Tornado note import/export, `transfer`, `transact-raw`, and name commands. Ethereum RPC stays clearnet. Use `--without-tor` or `KOHAKU_WITHOUT_TOR=1` to skip. **Saga CDN and proving artifacts are Tor-or-fail** (no clearnet fallback; large GETs time out after `KOHAKU_TOR_CDN_TIMEOUT_MS`, default 45s). First protocol sync (saga / Subsquid / ASP / RPC catch-up) shows live progress on the spinner and does not download proving keys. Artifacts are served from `/proving-artifacts` when cached; otherwise fetched from `KOHAKU_ARTIFACTS_BASE_URL` (default: `https://artifacts.0000000000.org`) on **prove / unshield**. Pre-warm keys with `kohaku fetch-artifacts` (optionally `--without-tor` for a one-shot clearnet download). Prefetch historical Subsquid/saga pages with `kohaku fetch-sync-cache`, which pulls a chunked, `sha256`-verified snapshot one piece at a time. After Tor bootstrap corruption, run `kohaku clear-tor-cache`. Railgun Subsquid and Tornado saga HTTP pages are reused from `/public-sync-cache` (wipe with `kohaku clear-tor-cache --public-sync`); Privacy Pools ASP and RPC are always live. Set `KOHAKU_TOR_DEBUG=1` for per-request Tor logs. A keyed RPC URL still identifies you to that provider regardless of Tor. Review with `view-network-traffic --wallet `. - **Fresh addresses:** Use `next-fresh-address` before funding, and `unshield --next` when you want withdrawals to land on a new public key that was not your shield source. Use `next-fresh-address --peek` to see the next address without persisting it (e.g. when building `--tail-calls` for a later `unshield --next`). Do not pass a peeked address as Tornado `--to` together with `--tail-calls` — peeked addresses are not stored, so the CLI will refuse rather than 7702 a note-derived key. -- **Profile / stealth:** Prefer `init-profile` to publish ERC-6538 keys (and optionally a name). Unshield to stored stealth accounts with `--to s0`. Print the meta URI with `see-stealth-meta-address`. New wallets store `.stealth-start-block` at creation so first `balances` stealth scans skip pre-wallet announcer history; imports can set the same via `create-wallet --import --stealth-start-block`. +- **Profile / stealth:** Prefer `init-profile` to publish ERC-6538 keys (and optionally a name). Unshield to stored stealth accounts with `--to s0`. Print the meta URI with `see-stealth-meta-address`. New wallets store `.stealth-start-block` at creation so first `balances` stealth scans skip pre-wallet announcer history; imports default to mainnet `25700000` / Sepolia `11455454` (Kohaku-schema floor) and can still pass `--stealth-start-block` to start earlier. - **Privacy Pools note size:** Each unshield uses one note; large shields may require multiple unshields if balances are split across notes. - **Tornado notes:** Use `export-tornado-note` / `import-tornado-note` to move legacy note secrets between wallets for testing or recovery. - **Private key / seed exports:** `export-private-key`, `reveal-seed-phrase`, and `export-tornado-note` print raw secrets to stdout. Avoid terminal logs, shell history, and shared environments. diff --git a/src/commands/balances.ts b/src/commands/balances.ts index 335c2b6..5a726be 100644 --- a/src/commands/balances.ts +++ b/src/commands/balances.ts @@ -622,6 +622,7 @@ export function registerBalancesCommand(program: Command): void { verbose: !!opts.verbose, withoutTor: opts.withoutTor, stealthStartBlock, + stealthStartBlockBackdate: stealthStartBlockFlag !== undefined, skipStealthScan: !!opts.skipStealthScan, onTorStatus: (message) => { loading.start(message); @@ -629,6 +630,11 @@ export function registerBalancesCommand(program: Command): void { onSyncProgress: (message) => { loading.start(message); }, + onStealthScanStart: quiet + ? undefined + : (message) => { + log.info(message); + }, onWarning: (msg) => { if (!quiet) { log.warn(chalk.yellow(msg)); diff --git a/src/commands/clear-tor-cache.ts b/src/commands/clear-tor-cache.ts index 8aad381..6527824 100644 --- a/src/commands/clear-tor-cache.ts +++ b/src/commands/clear-tor-cache.ts @@ -3,10 +3,14 @@ import type { Command } from "commander"; import { cliOptions } from "../utils/cli-command-options.js"; import { logCliJson } from "../utils/cli-quiet.js"; +import { DEFAULT_DATA_DIR } from "../utils/rpc.js"; +import { clearPublicSyncCache } from "../utils/public-sync-cache.js"; import { clearTorJsCache } from "../utils/tor.js"; type Opts = { nonInteractive?: boolean; + publicSync?: boolean; + dataDir?: string; }; export function registerClearTorCacheCommand(program: Command): void { @@ -16,16 +20,33 @@ export function registerClearTorCacheCommand(program: Command): void { "Delete the on-disk tor-js cache (~/.local/share/tor-js); use after Tor bootstrap corruption errors" ) .option("--non-interactive", cliOptions.nonInteractiveListWallets) + .option( + "--public-sync", + "Also delete /public-sync-cache (Railgun Subsquid / Tornado saga HTTP cache)" + ) + .option("--dataDir ", cliOptions.dataDir) .action((opts: Opts) => { - const result = clearTorJsCache(); + const tor = clearTorJsCache(); + const publicSync = opts.publicSync + ? clearPublicSyncCache(opts.dataDir ?? DEFAULT_DATA_DIR) + : undefined; if (opts.nonInteractive) { - logCliJson(result); + logCliJson({ ...tor, publicSync }); return; } console.log( - result.cleared - ? chalk.green(`Cleared Tor cache: ${result.path}`) - : chalk.dim(`No Tor cache to clear (${result.path})`) + tor.cleared + ? chalk.green(`Cleared Tor cache: ${tor.path}`) + : chalk.dim(`No Tor cache to clear (${tor.path})`) ); + if (publicSync) { + console.log( + publicSync.cleared + ? chalk.green( + `Cleared public sync cache: ${publicSync.path} (${publicSync.filesRemoved} file(s))` + ) + : chalk.dim(`No public sync cache to clear (${publicSync.path})`) + ); + } }); } diff --git a/src/commands/createWallet.ts b/src/commands/createWallet.ts index 7602687..1bb7602 100644 --- a/src/commands/createWallet.ts +++ b/src/commands/createWallet.ts @@ -99,7 +99,7 @@ export function registerCreateWalletCommand(program: Command): void { .option("--testnet", "Use testnet chain ID (11155111) instead of mainnet (1)") .option( "--stealth-start-block ", - "With --import: write `.stealth-start-block` for later balances stealth scans. New wallets record the current tip automatically." + "With --import: write `.stealth-start-block` for later balances stealth scans (default: mainnet 25700000, Sepolia 11455454). New wallets record the current tip automatically." ) .option( "--long-seed", diff --git a/src/commands/export-tornado-note.ts b/src/commands/export-tornado-note.ts index c9fa1e6..203681a 100644 --- a/src/commands/export-tornado-note.ts +++ b/src/commands/export-tornado-note.ts @@ -28,6 +28,7 @@ import { import { resolveTokenMeta } from "../utils/tokens-util"; import { withTor } from "../utils/tor.js"; import { runWithSyncProgress } from "../utils/sync-progress.js"; +import { isFirstProtocolSync } from "../utils/first-sync.js"; import { assertTornadoExactPoolDenomination } from "../utils/tornado-pools.js"; import { resolveWalletDir, @@ -290,7 +291,8 @@ export function registerExportTornadoNoteCommand(program: Command): void { () => runWithSyncProgress( { - protocol: "tornado", + source: "tornado", + firstRun: isFirstProtocolSync(walletDir, "tornado"), onUpdate: quiet ? undefined : (message) => spin.start(message), }, () => diff --git a/src/commands/fetch-sync-cache.ts b/src/commands/fetch-sync-cache.ts new file mode 100644 index 0000000..f7e38ca --- /dev/null +++ b/src/commands/fetch-sync-cache.ts @@ -0,0 +1,323 @@ +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import chalk from "chalk"; +import type { Command } from "commander"; + +import { cliOptions } from "../utils/cli-command-options.js"; +import { cliErrorFromCaught } from "../utils/cli-errors.js"; +import { logCliJson } from "../utils/cli-quiet.js"; +import { + countPublicSyncCacheEntries, + DEFAULT_CHUNK_BYTES, + installPublicSyncCacheChunk, + packPublicSyncCacheChunks, + parseSyncCacheManifest, + publicSyncCacheDir, + publicSyncCacheHasAllKeys, + resolveSyncCacheBaseUrl, + resolveSyncCacheChunkTimeoutMs, + setPublicSyncCacheDataDir, + syncCacheChunkUrl, + syncCacheManifestUrl, + type SyncCacheChunk, + type SyncCacheManifest, +} from "../utils/public-sync-cache.js"; +import { DEFAULT_DATA_DIR } from "../utils/rpc.js"; +import { ensureKohakuFetchPatch, withTor } from "../utils/tor.js"; + +const MANIFEST_TIMEOUT_MS = 60_000; +const CHUNK_ATTEMPTS = 3; + +type Opts = { + pack?: string; + chunkBytes?: string; + force?: boolean; + withoutTor?: boolean; + dataDir?: string; + nonInteractive?: boolean; +}; + +type ChunkOutcome = { + name: string; + status: "installed" | "skipped" | "failed"; + installed: number; + error?: string; +}; + +function formatMib(bytes: number): string { + return `${(bytes / (1024 * 1024)).toFixed(1)} MiB`; +} + +function parseChunkBytes(raw: string | undefined): number { + if (raw === undefined) return DEFAULT_CHUNK_BYTES; + const n = Number(raw); + if (!Number.isFinite(n) || !Number.isInteger(n) || n < 1024) { + throw new Error( + `--chunk-bytes must be an integer of at least 1024 (got ${JSON.stringify(raw)})` + ); + } + return n; +} + +/** + * The chunk URLs sit outside the proving-artifact routing, so `kohakuFetch` + * applies no timeout of its own and this budget is the only bound. + */ +async function fetchWithTimeout( + url: string, + timeoutMs: number +): Promise { + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), timeoutMs); + try { + const res = await fetch(url, { signal: controller.signal }); + if (!res.ok) throw new Error(`HTTP ${res.status}`); + return Buffer.from(await res.arrayBuffer()); + } catch (e) { + if (controller.signal.aborted) { + throw new Error(`timed out after ${timeoutMs}ms`); + } + throw e; + } finally { + clearTimeout(timer); + } +} + +async function downloadManifest( + nonInteractive: boolean +): Promise { + const url = syncCacheManifestUrl(); + if (!nonInteractive) { + console.error(chalk.dim(`Manifest: ${url}`)); + } + let raw: Buffer; + try { + raw = await fetchWithTimeout(url, MANIFEST_TIMEOUT_MS); + } catch (e) { + const message = e instanceof Error ? e.message : String(e); + throw new Error( + `Sync-cache manifest fetch failed (${message}) from ${url}. Override the location with KOHAKU_SYNC_CACHE_BASE_URL.` + ); + } + return parseSyncCacheManifest(raw.toString("utf8")); +} + +async function installChunk(opts: { + chunk: SyncCacheChunk; + dataDir: string; + tmpRoot: string; + timeoutMs: number; +}): Promise { + const { chunk, dataDir, tmpRoot, timeoutMs } = opts; + const url = syncCacheChunkUrl(chunk.name); + const archivePath = join(tmpRoot, chunk.name); + let lastError = ""; + + for (let attempt = 1; attempt <= CHUNK_ATTEMPTS; attempt++) { + try { + writeFileSync(archivePath, await fetchWithTimeout(url, timeoutMs)); + const { installed } = installPublicSyncCacheChunk({ + dataDir, + archivePath, + chunk, + }); + return installed; + } catch (e) { + lastError = e instanceof Error ? e.message : String(e); + if (attempt < CHUNK_ATTEMPTS) { + await new Promise((r) => setTimeout(r, 1000 * attempt)); + } + } finally { + rmSync(archivePath, { force: true }); + } + } + throw new Error(lastError); +} + +async function downloadChunks(opts: { + manifest: SyncCacheManifest; + dataDir: string; + force: boolean; + nonInteractive: boolean; +}): Promise { + const { manifest, dataDir, force, nonInteractive } = opts; + const timeoutMs = resolveSyncCacheChunkTimeoutMs(); + const tmpRoot = mkdtempSync(join(tmpdir(), "kohaku-sync-cache-")); + const outcomes: ChunkOutcome[] = []; + const total = manifest.chunks.length; + + try { + for (let i = 0; i < total; i++) { + const chunk = manifest.chunks[i]!; + const label = `[${i + 1}/${total}] ${chunk.name}`; + + if (!force && publicSyncCacheHasAllKeys(chunk.keys, dataDir)) { + outcomes.push({ name: chunk.name, status: "skipped", installed: 0 }); + if (!nonInteractive) { + console.error(chalk.dim(`${label} skip (${chunk.keys.length} entries cached)`)); + } + continue; + } + + if (!nonInteractive) { + console.error(chalk.dim(`${label} fetch ${formatMib(chunk.bytes)}`)); + } + try { + const installed = await installChunk({ + chunk, + dataDir, + tmpRoot, + timeoutMs, + }); + outcomes.push({ name: chunk.name, status: "installed", installed }); + if (!nonInteractive) { + console.error(chalk.dim(` ok (${installed} entries)`)); + } + } catch (e) { + const error = e instanceof Error ? e.message : String(e); + outcomes.push({ name: chunk.name, status: "failed", installed: 0, error }); + if (!nonInteractive) { + console.error(chalk.red(` failed: ${error}`)); + } + } + } + } finally { + rmSync(tmpRoot, { recursive: true, force: true }); + } + return outcomes; +} + +export function registerFetchSyncCacheCommand(program: Command): void { + program + .command("fetch-sync-cache") + .description( + "Download the public-sync snapshot in verifiable chunks into /public-sync-cache (or --pack a new one)" + ) + .option( + "--pack ", + "Publisher mode: pack /public-sync-cache into chunk-NNN.tar.gz + manifest.json in " + ) + .option( + "--chunk-bytes ", + `Target compressed bytes per chunk with --pack (default: ${DEFAULT_CHUNK_BYTES})` + ) + .option( + "--force", + "Re-download chunks whose entries are already present in the local cache" + ) + .option("--without-tor", cliOptions.withoutTorSyncCacheFetch) + .option("--dataDir ", cliOptions.dataDir) + .option("--non-interactive", cliOptions.nonInteractiveListWallets) + .action(async (opts: Opts) => { + try { + const dataDir = opts.dataDir ?? DEFAULT_DATA_DIR; + setPublicSyncCacheDataDir(dataDir); + ensureKohakuFetchPatch(); + + if (opts.pack) { + const { outputDir, manifest } = packPublicSyncCacheChunks({ + dataDir, + outputDir: resolve(opts.pack), + chunkBytes: parseChunkBytes(opts.chunkBytes), + }); + const packedBytes = manifest.chunks.reduce((n, c) => n + c.bytes, 0); + if (opts.nonInteractive) { + logCliJson({ packed: true, outputDir, manifest }); + } else { + console.log( + chalk.green( + `Packed ${manifest.entries} entries into ${manifest.chunks.length} chunks (${formatMib(packedBytes)} from ${formatMib(manifest.rawBytes)} raw)` + ) + ); + console.log(chalk.dim(`Output: ${outputDir}`)); + console.log( + chalk.dim( + `Publish these files (including manifest.json) at ${resolveSyncCacheBaseUrl()}/` + ) + ); + } + return; + } + + const useTor = !opts.withoutTor; + const cacheDir = publicSyncCacheDir(dataDir); + if (!opts.nonInteractive) { + console.error( + chalk.dim( + `Public-sync cache: ${cacheDir}\nBase URL: ${resolveSyncCacheBaseUrl()}\nRoute: ${useTor ? "Tor" : "clearnet (--without-tor)"}` + ) + ); + } + + const outcomes = await withTor( + useTor, + { + onStatus: opts.nonInteractive + ? undefined + : (m) => console.error(chalk.dim(m)), + }, + async () => { + const manifest = await downloadManifest(!!opts.nonInteractive); + if (!opts.nonInteractive) { + console.error( + chalk.dim( + `Snapshot: ${manifest.entries} entries in ${manifest.chunks.length} chunks` + ) + ); + } + return downloadChunks({ + manifest, + dataDir, + force: !!opts.force, + nonInteractive: !!opts.nonInteractive, + }); + } + ); + + const installed = outcomes.filter((o) => o.status === "installed"); + const skipped = outcomes.filter((o) => o.status === "skipped"); + const failed = outcomes.filter((o) => o.status === "failed"); + const entriesInstalled = installed.reduce((n, o) => n + o.installed, 0); + const result = { + cacheDir, + baseUrl: resolveSyncCacheBaseUrl(), + via: useTor ? ("tor" as const) : ("clearnet" as const), + chunks: outcomes.length, + chunksInstalled: installed.length, + chunksSkipped: skipped.length, + chunksFailed: failed.length, + entriesInstalled, + entries: countPublicSyncCacheEntries(dataDir), + errors: failed.map((o) => ({ chunk: o.name, error: o.error ?? "" })), + }; + + if (opts.nonInteractive) { + logCliJson(result); + } else { + console.log( + chalk.green( + `Sync cache: ${entriesInstalled} entries from ${installed.length} chunks, ${skipped.length} chunks already present, ${failed.length} failed` + ) + ); + console.log( + chalk.dim(`Cache: ${cacheDir} (${result.entries} entries total)`) + ); + if (failed.length > 0) { + console.log( + chalk.yellow( + "Partial snapshot installed. Re-run to retry the failed chunks; anything still missing is fetched live during sync." + ) + ); + } + } + + if (failed.length > 0) { + process.exitCode = 1; + } + } catch (e) { + cliErrorFromCaught(e); + process.exitCode = 1; + } + }); +} diff --git a/src/commands/import-tornado-note.ts b/src/commands/import-tornado-note.ts index e44f417..deb6311 100644 --- a/src/commands/import-tornado-note.ts +++ b/src/commands/import-tornado-note.ts @@ -20,6 +20,7 @@ import { } from "../utils/rpc"; import { withTor } from "../utils/tor.js"; import { runWithSyncProgress } from "../utils/sync-progress.js"; +import { isFirstProtocolSync } from "../utils/first-sync.js"; import { resolveWalletDir, resolveWalletNameOrPrompt, @@ -164,7 +165,8 @@ export function registerImportTornadoNoteCommand(program: Command): void { () => runWithSyncProgress( { - protocol: "tornado", + source: "tornado", + firstRun: isFirstProtocolSync(walletDir, "tornado"), onUpdate: quiet ? undefined : (message) => spin.start(message), }, () => diff --git a/src/commands/shield.ts b/src/commands/shield.ts index 039e9a8..c2e2b4e 100644 --- a/src/commands/shield.ts +++ b/src/commands/shield.ts @@ -58,6 +58,7 @@ import { runWithSyncProgress, syncPluginWithProgress, } from "../utils/sync-progress.js"; +import { isFirstProtocolSync } from "../utils/first-sync.js"; import { resolveTokenMeta } from "../utils/tokens-util"; import { resolveWalletDir, @@ -594,7 +595,8 @@ export function registerShieldCommand(program: Command): void { ? await prepareProtocolShield(plugin, protocol, asset as AssetAmount) : await runWithSyncProgress( { - protocol, + source: protocol, + firstRun: isFirstProtocolSync(walletDir, protocol), onUpdate: quiet ? undefined : (message) => txSpinner.start(message), }, async () => { diff --git a/src/commands/unshield.ts b/src/commands/unshield.ts index f67c4d6..9a29c8b 100644 --- a/src/commands/unshield.ts +++ b/src/commands/unshield.ts @@ -96,7 +96,7 @@ import { runWithSyncProgress, syncPluginWithProgress, } from "../utils/sync-progress.js"; -import { primeRailgunSubsquidProgressIfNeeded } from "../utils/railgun-subsquid-progress.js"; +import { isFirstProtocolSync } from "../utils/first-sync.js"; import { resolveTokenMeta } from "../utils/tokens-util"; import { resolveTornadoTailCallsGasEstimate } from "../utils/tornado-tail-gas.js"; import { @@ -536,16 +536,12 @@ export function registerUnshieldCommand(program: Command): void { railgunBalance, } = await runWithSyncProgress( { - protocol, + source: protocol, + firstRun: isFirstProtocolSync(walletDir, protocol), onUpdate: quiet ? undefined : (message) => spin.start(message), }, async () => { - const priming = primeRailgunSubsquidProgressIfNeeded( - protocol, - chainId - ); await syncPluginWithProgress(plugin, protocol); - await priming; return maxUnshieldAmountHint(protocol, plugin, tokenMeta, chainId); } ); diff --git a/src/host/chunked-get-logs.ts b/src/host/chunked-get-logs.ts index 8b039c8..4cebf8d 100644 --- a/src/host/chunked-get-logs.ts +++ b/src/host/chunked-get-logs.ts @@ -2,7 +2,7 @@ import type { EthereumProvider } from "@kohaku-eth/provider"; import type { Filter } from "ox/Filter"; import type { RpcRequest } from "ox/RpcRequest"; -import { reportSyncProgress } from "../utils/sync-progress.js"; +import { countSyncRequest } from "../utils/sync-progress.js"; /** Max inclusive block span per `eth_getLogs` call (env / option override). */ const DEFAULT_MAX_BLOCK_SPAN = 499n; @@ -100,16 +100,6 @@ function asLogArray(result: unknown): unknown[] { throw new Error(`expected eth_getLogs result to be an array, got ${typeof result}`); } -export function estimatedGetLogsWindowCount( - fromBn: bigint, - toBn: bigint, - chunkSpan: bigint -): number { - if (fromBn > toBn || chunkSpan <= 0n) return 0; - const n = (toBn - fromBn + 1n + chunkSpan - 1n) / chunkSpan; - return n > BigInt(Number.MAX_SAFE_INTEGER) ? Number.MAX_SAFE_INTEGER : Number(n); -} - /** * Step [fromBn, toBn] in fixed inclusive windows of at most `chunkSpan` blocks. * `invoke` errors are logged then rethrown. @@ -124,14 +114,11 @@ async function fetchLogsChunked( const out: unknown[] = []; let windowFrom = fromBn; let w = 0; - const total = estimatedGetLogsWindowCount(fromBn, toBn, chunkSpan); while (windowFrom <= toBn) { const windowTo = windowFrom + chunkSpan - 1n > toBn ? toBn : windowFrom + chunkSpan - 1n; w += 1; - if (total > 0) { - reportSyncProgress({ phase: "rpc", done: w, total }); - } + countSyncRequest("rpc"); let raw: unknown; try { raw = await invoke(windowFrom, windowTo); diff --git a/src/host/makeHost.ts b/src/host/makeHost.ts index 99dbf88..546f33d 100644 --- a/src/host/makeHost.ts +++ b/src/host/makeHost.ts @@ -5,6 +5,7 @@ import { withChunkedGetLogs, withTransactionCount } from "./chunked-get-logs"; import { makeKeystore, makeRailgunKeystore } from "./keystore"; import { makeStealthAccountsStorage } from "../lib/stealth/storage"; import { makeStorage, type PluginId } from "./storage"; +import { noteProtocolStorageFreshness } from "../utils/first-sync"; import { tornadoExternalSyncForChain } from "../utils/saga-external-sync"; import { kohakuFetch } from "../utils/tor"; import type { KohakuPublicClient } from "../utils/rpc.js"; @@ -44,6 +45,9 @@ export async function makeHost(options: MakeHostOptions): Promise { externalSyncProvider: externalSyncProviderIn, } = options; + // Before anything (notably the Privacy Pools bundled snapshot) can seed storage. + noteProtocolStorageFreshness(walletDir, pluginId); + const provider = withChunkedGetLogs( withTransactionCount(kohakuViemProvider(rpc)) ); diff --git a/src/host/storage.ts b/src/host/storage.ts index 67f1ea5..a775632 100644 --- a/src/host/storage.ts +++ b/src/host/storage.ts @@ -4,7 +4,10 @@ import { loadStore, saveStore } from "../utils/aes-storage"; export type PluginId = "rg" | "ppv1" | "tc"; -function pluginStorePathForWallet(walletDir: string, pluginId: PluginId): string { +export function pluginStorePathForWallet( + walletDir: string, + pluginId: PluginId +): string { return join(walletDir, `${pluginId}-storage.json`); } diff --git a/src/index.ts b/src/index.ts index 61effd3..ee5280b 100644 --- a/src/index.ts +++ b/src/index.ts @@ -21,6 +21,7 @@ import { registerTransactRawCommand } from "./commands/transact-raw"; import { registerViewNetworkTrafficCommand } from "./commands/viewNetworkTraffic"; import { registerClearTorCacheCommand } from "./commands/clear-tor-cache"; import { registerFetchArtifactsCommand } from "./commands/fetch-artifacts"; +import { registerFetchSyncCacheCommand } from "./commands/fetch-sync-cache"; import { registerRegisterNameCommand } from "./commands/register-name"; import { registerRenewNameCommand } from "./commands/renew-name"; import { registerTransferNameCommand } from "./commands/transfer-name"; @@ -67,6 +68,7 @@ async function main(): Promise { registerViewNetworkTrafficCommand(program); registerClearTorCacheCommand(program); registerFetchArtifactsCommand(program); + registerFetchSyncCacheCommand(program); await program.parseAsync(process.argv); } diff --git a/src/lib/balances-snapshot.ts b/src/lib/balances-snapshot.ts index 8a13935..7102ddc 100644 --- a/src/lib/balances-snapshot.ts +++ b/src/lib/balances-snapshot.ts @@ -4,7 +4,7 @@ import { formatUnits, getAddress, isAddress } from "viem"; import { formatCaughtError } from "../utils/cli-errors"; import { makePublicClient, disposePublicClient } from "../utils/rpc"; import { runWithSyncProgress } from "../utils/sync-progress.js"; -import { primeRailgunSubsquidProgressIfNeeded } from "../utils/railgun-subsquid-progress.js"; +import { isFirstProtocolSync } from "../utils/first-sync.js"; import { runWithWalletTrafficLog, withTor } from "../utils/tor"; import { withProtocolRuntime } from "./protocol-runtime"; import { @@ -28,7 +28,11 @@ import { } from "./private-notes"; import { makePublicAccountsStorage } from "../utils/public-accounts"; import { deriveStealthKeypair } from "./stealth/keys.js"; -import { scanAndImportStealthAnnouncements } from "./stealth/scan.js"; +import { + formatStealthScanStartLog, + resolveStealthScanWindow, + scanAndImportStealthAnnouncements, +} from "./stealth/scan.js"; import { makeStealthAccountsStorage } from "./stealth/storage.js"; import { attachUsdValuesToRowsLists, @@ -137,21 +141,25 @@ async function loadProtocolNotes( tokenMeta: Map, onSyncProgress?: (message: string) => void ): Promise { - const notes = await runWithSyncProgress({ protocol, onUpdate: onSyncProgress }, async () => { - const priming = primeRailgunSubsquidProgressIfNeeded(protocol, chainId); - return withProtocolRuntime( - { protocol, rpcUrl, walletDir, password, mnemonic, chainId }, - async (_host, plugin) => { - await priming; - const notesFn = (plugin as AnyPlugin).notes; - if (!notesFn) { - throw new Error(`${protocol} plugin does not expose notes()`); + const notes = await runWithSyncProgress( + { + source: protocol, + firstRun: isFirstProtocolSync(walletDir, protocol), + onUpdate: onSyncProgress, + }, + async () => + withProtocolRuntime( + { protocol, rpcUrl, walletDir, password, mnemonic, chainId }, + async (_host, plugin) => { + const notesFn = (plugin as AnyPlugin).notes; + if (!notesFn) { + throw new Error(`${protocol} plugin does not expose notes()`); + } + // Preserve method `this` binding for class-based plugin implementations. + return (plugin as AnyPlugin).notes!(undefined, false); } - // Preserve method `this` binding for class-based plugin implementations. - return (plugin as AnyPlugin).notes!(undefined, false); - } - ); - }); + ) + ); return mapProtocolNotes(protocol, notes, tokenMeta); } @@ -164,16 +172,18 @@ async function loadPrivateBalancesForProtocol( chainId: bigint, onSyncProgress?: (message: string) => void ): Promise { - return runWithSyncProgress({ protocol, onUpdate: onSyncProgress }, async () => { - const priming = primeRailgunSubsquidProgressIfNeeded(protocol, chainId); - return withProtocolRuntime( - { protocol, rpcUrl, walletDir, password, mnemonic, chainId }, - async (_host, plugin) => { - await priming; - return plugin.balance(undefined); - } - ); - }); + return runWithSyncProgress( + { + source: protocol, + firstRun: isFirstProtocolSync(walletDir, protocol), + onUpdate: onSyncProgress, + }, + async () => + withProtocolRuntime( + { protocol, rpcUrl, walletDir, password, mnemonic, chainId }, + async (_host, plugin) => plugin.balance(undefined) + ) + ); } async function loadErc20Meta( @@ -221,11 +231,22 @@ export type LoadBalancesSnapshotOptions = { * When set on a first/full history pass, skips announcer history before this block. */ stealthStartBlock?: bigint; + /** + * When true, `stealthStartBlock` came from `--stealth-start-block` and may + * start below lastScannedBlock (back-date). Do not set this for the wallet + * file — that is passed as stealthStartBlock on every incremental run. + */ + stealthStartBlockBackdate?: boolean; /** * Skip ERC-5564 announcement discovery. Already-imported stealth accounts * are still included in public balances. */ skipStealthScan?: boolean; + /** + * Durable "Stealth scan from block …" line. Omit in quiet / --non-interactive + * mode so JSON stdout stays clean. + */ + onStealthScanStart?: (message: string) => void; }; export type PrivateBalancesSnapshot = { @@ -435,6 +456,8 @@ async function loadBalancesSnapshotInner( onTorStatus, onSyncProgress, skipStealthScan, + stealthStartBlockBackdate, + onStealthScanStart, } = opts; const chainIdString = chainId.toString(); @@ -485,8 +508,29 @@ async function loadBalancesSnapshotInner( if (!skipStealthScan) { try { const keypair = deriveStealthKeypair(mnemonic, chainId); + const stealthStore = makeStealthAccountsStorage( + walletDir, + password + ).getStore(); + const latest = await rpcForPublic.getBlockNumber(); + const window = resolveStealthScanWindow({ + chainId, + latest, + startFromBlock: opts.stealthStartBlock, + lastScannedBlock: stealthStore.lastScannedBlock, + fullHistoryScanned: stealthStore.fullHistoryScanned, + backdate: stealthStartBlockBackdate, + }); + const prelude = + onStealthScanStart && window.fromBlock <= window.latest + ? formatStealthScanStartLog(window.fromBlock, window.latest) + : undefined; + if (prelude && !process.stdout.isTTY) { + onStealthScanStart?.(prelude); + } await runWithSyncProgress( - { protocol: "stealth", onUpdate: onSyncProgress }, + // The scan itself reports whether this is a first pass. + { source: "stealth", onUpdate: onSyncProgress, prelude }, () => scanAndImportStealthAnnouncements({ client: rpcForPublic, @@ -495,6 +539,8 @@ async function loadBalancesSnapshotInner( keypair, chainId, startFromBlock: opts.stealthStartBlock, + backdate: stealthStartBlockBackdate, + latest, }) ); } catch (e) { diff --git a/src/lib/create-wallet.ts b/src/lib/create-wallet.ts index 0f9b5cf..1f019e9 100644 --- a/src/lib/create-wallet.ts +++ b/src/lib/create-wallet.ts @@ -17,6 +17,7 @@ import { resolveWalletDir, writeWalletType, } from "../utils/wallets-util.js"; +import { defaultStealthImportStartBlock } from "./stealth/constants.js"; import { writeStealthStartBlock } from "./stealth/start-block-file.js"; export async function findLastTouchedPublicIndex( @@ -67,7 +68,8 @@ export type CreateWalletOnDiskInput = { rpcUrl?: string; /** * Import only: persist as `.stealth-start-block` for later `balances` scans. - * New (generated) wallets record the current chain tip automatically instead. + * When omitted, writes the Kohaku-schema floor for the chain. New (generated) + * wallets record the current chain tip automatically instead. */ stealthStartBlock?: bigint; }; @@ -121,10 +123,11 @@ export async function createWalletOnDisk( let stealthStartBlockWritten: bigint | undefined; if (input.importMode) { - if (input.stealthStartBlock !== undefined) { - writeStealthStartBlock(walletDir, input.stealthStartBlock); - stealthStartBlockWritten = input.stealthStartBlock; - } + const block = + input.stealthStartBlock ?? + defaultStealthImportStartBlock(expectedChainId); + writeStealthStartBlock(walletDir, block); + stealthStartBlockWritten = block; } else { const { blockNumber } = await fetchCurrentBlockNumber({ testnet: input.testnet, diff --git a/src/lib/shield-flow.ts b/src/lib/shield-flow.ts index ff247a5..e067ffa 100644 --- a/src/lib/shield-flow.ts +++ b/src/lib/shield-flow.ts @@ -5,6 +5,7 @@ import { Mnemonic } from "derive-railgun-keys"; import { makePublicClient, disposePublicClient, type KohakuPublicClient } from "../utils/rpc"; import { withTor } from "../utils/tor"; import { runWithSyncProgress, syncPluginWithProgress } from "../utils/sync-progress.js"; +import { isFirstProtocolSync } from "../utils/first-sync.js"; import { rpcForWalletOps, withProtocolRuntime } from "./protocol-runtime.js"; import { ERC20_ABI } from "../utils/tokens-util"; import type { ResolvedTokenMeta } from "../utils/tokens-util"; @@ -538,7 +539,11 @@ export async function prepareShieldPlan(opts: { protocol === "railgun" ? await prepareProtocolShield(plugin, protocol, asset as AssetAmount) : await runWithSyncProgress( - { protocol, onUpdate: onSyncProgress }, + { + source: protocol, + firstRun: isFirstProtocolSync(walletDir, protocol), + onUpdate: onSyncProgress, + }, async () => { await syncPluginWithProgress(plugin, protocol); return prepareProtocolShield(plugin, protocol, asset as AssetAmount); diff --git a/src/lib/stealth/constants.ts b/src/lib/stealth/constants.ts index dc02883..ccedf48 100644 --- a/src/lib/stealth/constants.ts +++ b/src/lib/stealth/constants.ts @@ -53,3 +53,17 @@ export function stealthAnnouncerStartBlock(chainId: bigint): bigint { if (chainId === 11155111n) return BigInt(ERC5564_StartBlocks.SEPOLIA); return 0n; } + +/** + * Optimistic first-scan floor for Kohaku-schema stealth keys. + * + * The ERC-5564 announcer is older than this CLI's derivation scheme; imports + * that omit `--stealth-start-block` start here instead of at contract deploy. + * `balances --stealth-start-block` can still back-date down to + * {@link stealthAnnouncerStartBlock}. + */ +export function defaultStealthImportStartBlock(chainId: bigint): bigint { + if (chainId === 1n) return 25_700_000n; + if (chainId === 11155111n) return 11_455_454n; + return 0n; +} diff --git a/src/lib/stealth/scan.ts b/src/lib/stealth/scan.ts index b430d22..9c82984 100644 --- a/src/lib/stealth/scan.ts +++ b/src/lib/stealth/scan.ts @@ -5,11 +5,12 @@ import { STEALTH_SCHEME_ID } from "eth-stealth-address-resolver"; import { getAddress, parseAbiItem, type Hex, type Log } from "viem"; import { privateKeyToAccount } from "viem/accounts"; -import { resolveGetLogsMaxBlockSpan, estimatedGetLogsWindowCount } from "../../host/chunked-get-logs.js"; +import { resolveGetLogsMaxBlockSpan } from "../../host/chunked-get-logs.js"; import type { KohakuPublicClient } from "../../utils/rpc.js"; -import { reportSyncProgress } from "../../utils/sync-progress.js"; +import { countSyncRequest, noteSyncFirstRun } from "../../utils/sync-progress.js"; import { STEALTH_ANNOUNCER_ADDRESS, + defaultStealthImportStartBlock, stealthAnnouncerStartBlock, } from "./constants.js"; import type { StealthKeypair } from "./keys.js"; @@ -56,6 +57,94 @@ export function parseStealthStartBlock(raw: string): bigint { return block; } +export type StealthScanWindowInput = { + chainId: bigint; + latest: bigint; + /** + * Inclusive lower bound from `--stealth-start-block`, `.stealth-start-block`, + * or omitted (Kohaku import default). Clamped to at least the announcer + * deploy block. + */ + startFromBlock?: bigint; + lastScannedBlock?: string | null; + fullHistoryScanned?: boolean; + fullRescan?: boolean; + /** + * True when the user passed `--stealth-start-block` on this run. Allows + * starting below `lastScannedBlock+1`. The wallet file must not set this — + * `balances` already passes the file as `startFromBlock` on every run. + */ + backdate?: boolean; +}; + +export type StealthScanWindow = { + fromBlock: bigint; + latest: bigint; + startFloor: bigint; + needsFullHistory: boolean; +}; + +/** + * Inclusive first-pass floor: max(requested or Kohaku default, announcer deploy). + */ +export function resolveStealthScanFloor(opts: { + chainId: bigint; + startFromBlock?: bigint; +}): bigint { + const announcerDeploy = stealthAnnouncerStartBlock(opts.chainId); + const requested = + opts.startFromBlock !== undefined + ? opts.startFromBlock + : defaultStealthImportStartBlock(opts.chainId); + return requested > announcerDeploy ? requested : announcerDeploy; +} + +/** First/full vs incremental stealth `getLogs` window. */ +export function resolveStealthScanWindow( + opts: StealthScanWindowInput +): StealthScanWindow { + const startFloor = resolveStealthScanFloor({ + chainId: opts.chainId, + startFromBlock: opts.startFromBlock, + }); + const needsFullHistory = Boolean(opts.fullRescan || !opts.fullHistoryScanned); + const lastScanned = opts.lastScannedBlock + ? BigInt(opts.lastScannedBlock) + : null; + + let fromBlock: bigint; + if (needsFullHistory) { + // Resume mid-pass from lastScannedBlock when present; allow a higher + // --stealth-start-block to jump the floor forward past an earlier cursor. + fromBlock = startFloor; + if (lastScanned !== null) { + const resume = lastScanned + 1n; + if (resume > fromBlock) fromBlock = resume; + } + } else { + fromBlock = lastScanned !== null ? lastScanned + 1n : startFloor; + } + + if (opts.backdate && startFloor < fromBlock) { + fromBlock = startFloor; + } + + return { + fromBlock, + latest: opts.latest, + startFloor, + needsFullHistory, + }; +} + +/** Durable line printed before the Stealth progress bar. */ +export function formatStealthScanStartLog( + fromBlock: bigint, + latest: bigint +): string { + return `Stealth scan from block ${fromBlock.toString()} · ${(latest - fromBlock).toString()} blocks`; +} + /** * Scan ERC-5564 announcements for payments to this wallet's stealth keys and * persist any new stealth accounts. @@ -64,10 +153,11 @@ export function parseStealthStartBlock(raw: string): bigint { * automatically (same wallet password / AES envelope as other stores). * * History: - * - First run (or stores that never completed a full pass): from announcer - * deploy block (or `startFromBlock` if higher) → latest, in - * KOHAKU_GETLOGS_MAX_BLOCK_SPAN chunks (default 499). - * - Later runs: lastScannedBlock+1 → latest only. + * - First run (or stores that never completed a full pass): from the Kohaku + * import default (or `startFromBlock` if set), clamped to at least the + * announcer deploy block → latest, in KOHAKU_GETLOGS_MAX_BLOCK_SPAN chunks + * (default 499). + * - Later runs: lastScannedBlock+1 → latest only, unless `backdate` is set. */ export async function scanAndImportStealthAnnouncements(opts: { client: KohakuPublicClient; @@ -79,10 +169,17 @@ export async function scanAndImportStealthAnnouncements(opts: { fullRescan?: boolean; /** * Inclusive lower bound for the first/full history pass. Clamped to at least - * the announcer deploy block. Incremental scans still resume from - * lastScannedBlock+1. + * the announcer deploy block. When omitted, uses the Kohaku import default. + * Incremental scans still resume from lastScannedBlock+1 unless `backdate`. */ startFromBlock?: bigint; + /** + * When true, start from the (clamped) `startFromBlock` even if + * lastScannedBlock is already ahead. Only the CLI flag should set this. + */ + backdate?: boolean; + /** Skip a second `getBlockNumber` when the caller already resolved the window. */ + latest?: bigint; }): Promise { // Creates stealth-accounts.json on first use for pre-existing wallets. const storage = makeStealthAccountsStorage(opts.walletDir, opts.password); @@ -90,29 +187,21 @@ export async function scanAndImportStealthAnnouncements(opts: { storage.setMeta({ metaAddressURI: opts.keypair.stealthMetaAddressURI }); } - const latest = await opts.client.getBlockNumber(); - const announcerDeploy = stealthAnnouncerStartBlock(opts.chainId); - const startFloor = - opts.startFromBlock !== undefined && opts.startFromBlock > announcerDeploy - ? opts.startFromBlock - : announcerDeploy; + const latest = opts.latest ?? (await opts.client.getBlockNumber()); const store = storage.getStore(); - const needsFullHistory = - opts.fullRescan || !store.fullHistoryScanned; - - let fromBlock: bigint; - if (needsFullHistory) { - // Resume mid-pass from lastScannedBlock when present; allow --stealth-start-block - // to jump the floor forward past an earlier cursor. - fromBlock = startFloor; - if (store.lastScannedBlock) { - const resume = BigInt(store.lastScannedBlock) + 1n; - if (resume > fromBlock) fromBlock = resume; - } - } else { - const stored = store.lastScannedBlock; - fromBlock = stored ? BigInt(stored) + 1n : startFloor; - } + const window = resolveStealthScanWindow({ + chainId: opts.chainId, + latest, + startFromBlock: opts.startFromBlock, + lastScannedBlock: store.lastScannedBlock, + fullHistoryScanned: store.fullHistoryScanned, + fullRescan: opts.fullRescan, + backdate: opts.backdate, + }); + const { fromBlock, needsFullHistory } = window; + // `stealth-accounts.json` can exist without a scan (profile / account writes), + // so the full-history flag is the only reliable "first scan" signal. + noteSyncFirstRun(!store.fullHistoryScanned); if (fromBlock > latest) { if (needsFullHistory) storage.markFullHistoryScanned(); @@ -131,16 +220,11 @@ export async function scanAndImportStealthAnnouncements(opts: { let announcementsChecked = 0; let alreadyKnown = 0; const name = store.name; - let chunk = 0; - const total = estimatedGetLogsWindowCount(fromBlock, latest, maxSpan); let cursor = fromBlock; while (cursor <= latest) { const to = cursor + maxSpan - 1n > latest ? latest : cursor + maxSpan - 1n; - chunk += 1; - if (total > 0) { - reportSyncProgress({ phase: "rpc", done: chunk, total }); - } + countSyncRequest("rpc"); const logs = (await opts.client.getLogs({ address: STEALTH_ANNOUNCER_ADDRESS, diff --git a/src/lib/unshield-flow.ts b/src/lib/unshield-flow.ts index 2a3b7e5..5bf1a77 100644 --- a/src/lib/unshield-flow.ts +++ b/src/lib/unshield-flow.ts @@ -12,7 +12,7 @@ import { } from "../utils/rpc.js"; import { makePublicAccountsStorage } from "../utils/public-accounts.js"; import { runWithSyncProgress, syncPluginWithProgress } from "../utils/sync-progress.js"; -import { primeRailgunSubsquidProgressIfNeeded } from "../utils/railgun-subsquid-progress.js"; +import { isFirstProtocolSync } from "../utils/first-sync.js"; import { withTor } from "../utils/tor.js"; import type { ResolvedTokenMeta } from "../utils/tokens-util.js"; import { @@ -355,12 +355,18 @@ async function runUnshieldWithPlugin( (opts.protocol === "privacy-pools" || opts.protocol === "tornado") && typeof maybeSync.sync === "function" ) { - await syncPluginWithProgress(maybeSync, opts.protocol, opts.onStatus); + await syncPluginWithProgress(maybeSync, opts.protocol, { + firstRun: isFirstProtocolSync(opts.walletDir, opts.protocol), + onUpdate: opts.onStatus, + }); } else if (opts.protocol === "railgun") { await runWithSyncProgress( - { protocol: opts.protocol, onUpdate: opts.onStatus }, + { + source: opts.protocol, + firstRun: isFirstProtocolSync(opts.walletDir, opts.protocol), + onUpdate: opts.onStatus, + }, async () => { - await primeRailgunSubsquidProgressIfNeeded(opts.protocol, opts.chainId); // Railgun syncs on balance(); warm it so prepareUnshield is not silent. await ( plugin as { balance: (assets: unknown) => Promise } diff --git a/src/utils/cli-command-options.ts b/src/utils/cli-command-options.ts index 1faf72b..39beaa1 100644 --- a/src/utils/cli-command-options.ts +++ b/src/utils/cli-command-options.ts @@ -21,6 +21,8 @@ export const cliOptions = { "Disable Tor for non-RPC HTTP (default: Tor on). RPC stays clearnet. Or set KOHAKU_WITHOUT_TOR=1", withoutTorArtifactsFetch: "Download proving artifacts over clearnet (default: Tor). Reveals that this IP fetched kohaku circuits; later shield/unshield stay Tor-only from the local cache. Or set KOHAKU_WITHOUT_TOR=1", + withoutTorSyncCacheFetch: + "Download snapshot chunks over clearnet (default: Tor). Much faster, and reveals only that this IP fetched public Railgun/Tornado pool data. Or set KOHAKU_WITHOUT_TOR=1", stealthStartBlock: - "Start ERC-5564 announcement scan at this block (decimal or 0x-hex); skips older history on first/full scan. Does not skip the scan itself (use --skip-stealth-scan). balances also reads `.stealth-start-block` from the wallet when this flag is omitted", + "Start ERC-5564 announcement scan at this block (decimal or 0x-hex); skips older history on first/full scan. Can back-date below the wallet `.stealth-start-block` (down to the announcer deploy block). When omitted, uses that file, or the Kohaku import default. Does not skip the scan itself (use --skip-stealth-scan)", } as const; diff --git a/src/utils/cli-quiet.ts b/src/utils/cli-quiet.ts index 80f0800..c513f95 100644 --- a/src/utils/cli-quiet.ts +++ b/src/utils/cli-quiet.ts @@ -1,4 +1,9 @@ import { jsonStringifyWithBigInt } from "./json-bigint"; +import { + clearTerminalOwner, + setTerminalOwner, + type TerminalOwner, +} from "./progress-terminal.js"; /** * With `--non-interactive`, skip clack spinners and similar UI so stdout stays @@ -23,31 +28,55 @@ export type QuietSpinner = { export function manageSpinner( spin: QuietSpinner, quiet: boolean -): QuietSpinner & { readonly active: boolean } { +): QuietSpinner & { readonly active: boolean } & TerminalOwner { let active = false; - return { + let lastMessage: string | undefined; + + const managed = { get active() { return active; }, start(msg?: string) { if (quiet) return; + lastMessage = msg ?? lastMessage; if (active) { spin.message?.(msg); return; } spin.start(msg); active = true; + setTerminalOwner(managed); }, message(msg?: string) { if (quiet || !active) return; + lastMessage = msg ?? lastMessage; spin.message?.(msg); }, stop(msg?: string, code?: number) { if (quiet || !active) return; spin.stop(msg, code); active = false; + clearTerminalOwner(managed); + }, + /** Yields the terminal line; see `progress-terminal`. */ + suspend(): string | null { + if (quiet || !active) return null; + spin.stop(""); + active = false; + // clack's stop() leaves a bare "◇" line behind; drop it so whoever takes + // the line starts clean. + if (process.stdout.isTTY) { + process.stdout.write("\u001b[1A\u001b[2K\u001b[999D"); + } + return lastMessage ?? ""; + }, + resume(msg: string | null): void { + if (quiet) return; + managed.start(msg ?? lastMessage); }, }; + + return managed; } /** diff --git a/src/utils/first-sync.ts b/src/utils/first-sync.ts new file mode 100644 index 0000000..cbeb224 --- /dev/null +++ b/src/utils/first-sync.ts @@ -0,0 +1,50 @@ +import { existsSync } from "node:fs"; + +import { pluginStorePathForWallet, type PluginId } from "../host/storage.js"; +import { pluginIdForProtocol, type SupportedProtocol } from "./plugins.js"; + +/** + * `${walletDir}|${pluginId}` -> storage file was missing when the Host was built. + * + * Privacy Pools seeds a bundled state snapshot into `ppv1-storage.json` while the + * plugin is constructed, i.e. before any sync runs, so a plain file check made at + * sync time would report an incremental sync on a genuinely first one. Recording + * the answer at Host creation (the one step that always precedes plugin + * construction) keeps callers free to ask whenever it suits them. + */ +const observed = new Map(); + +function key(walletDir: string, pluginId: PluginId): string { + return `${walletDir}|${pluginId}`; +} + +function storeIsMissing(walletDir: string, pluginId: PluginId): boolean { + return !existsSync(pluginStorePathForWallet(walletDir, pluginId)); +} + +/** Record whether this wallet has protocol state yet. Called from `makeHost`. */ +export function noteProtocolStorageFreshness( + walletDir: string, + pluginId: PluginId +): void { + const k = key(walletDir, pluginId); + if (observed.has(k)) return; + observed.set(k, storeIsMissing(walletDir, pluginId)); +} + +/** True when this wallet has no persisted state for `protocol` yet. */ +export function isFirstProtocolSync( + walletDir: string, + protocol: SupportedProtocol +): boolean { + const pluginId = pluginIdForProtocol(protocol); + return ( + observed.get(key(walletDir, pluginId)) ?? + storeIsMissing(walletDir, pluginId) + ); +} + +/** Test hook: drop memoized observations. */ +export function resetFirstSyncObservations(): void { + observed.clear(); +} diff --git a/src/utils/network-traffic-log.ts b/src/utils/network-traffic-log.ts index 2c93894..fd12fdc 100644 --- a/src/utils/network-traffic-log.ts +++ b/src/utils/network-traffic-log.ts @@ -33,6 +33,8 @@ export type TrafficClearnetReason = | "rpc" /** Served from ~/.kohaku-cli/proving-artifacts (no network). */ | "local-artifact" + /** Served from /public-sync-cache (Subsquid / saga). */ + | "public-sync-cache" /** @deprecated Legacy log rows only — clearnet artifact fallback removed. */ | "artifact-fallback" /** @deprecated Legacy log rows only — clearnet saga fallback removed. */ @@ -262,7 +264,8 @@ export function clearNetworkTrafficLog(walletDir: string): boolean { export function isLocalEntry(entry: NetworkTrafficEntry): boolean { if ( entry.clearnetReason === "loopback" || - entry.clearnetReason === "local-artifact" + entry.clearnetReason === "local-artifact" || + entry.clearnetReason === "public-sync-cache" ) { return true; } diff --git a/src/utils/plugins.ts b/src/utils/plugins.ts index 9f145b3..b6d63d7 100644 --- a/src/utils/plugins.ts +++ b/src/utils/plugins.ts @@ -108,12 +108,9 @@ export function resolveProtocolOption( return { ok: false, error: "missing" }; } -/** - * Parses `--include railgun,tornado`. Returns `null` when omitted - * (caller may apply env / empty defaults). - */ -export function parseIncludeProtocols( - raw: string | undefined +function parseProtocolList( + raw: string | undefined, + flag: "--include" | "--protocol" ): SupportedProtocol[] | null { if (!raw?.trim()) return null; @@ -124,7 +121,7 @@ export function parseIncludeProtocols( for (const part of parts) { if (!isSupportedProtocol(part)) { throw new Error( - `Invalid protocol in --include: ${part}. Use: ${SUPPORTED_PROTOCOLS_HELP}` + `Invalid protocol in ${flag}: ${part}. Use: ${SUPPORTED_PROTOCOLS_HELP}` ); } if (seen.has(part)) continue; @@ -134,13 +131,23 @@ export function parseIncludeProtocols( if (out.length === 0) { throw new Error( - `--include requires at least one protocol. Use: ${SUPPORTED_PROTOCOLS_HELP}` + `${flag} requires at least one protocol. Use: ${SUPPORTED_PROTOCOLS_HELP}` ); } return out; } +/** + * Parses `--include railgun,tornado`. Returns `null` when omitted + * (caller may apply env / empty defaults). + */ +export function parseIncludeProtocols( + raw: string | undefined +): SupportedProtocol[] | null { + return parseProtocolList(raw, "--include"); +} + /** * Resolves which private protocols `balances` should sync: * - explicit `--include` → those protocols diff --git a/src/utils/progress-terminal.ts b/src/utils/progress-terminal.ts new file mode 100644 index 0000000..7b45fdf --- /dev/null +++ b/src/utils/progress-terminal.ts @@ -0,0 +1,38 @@ +/** + * Tracks which spinner currently owns the terminal line. + * + * A clack spinner erases everything below its cursor on every frame, so a + * second animator cannot share the screen with it. Sync progress needs to take + * the line over for the duration of a sync (see `sync-progress-renderer`), and + * this registry lets it do so without threading spinner objects through the + * library APIs that sit between a command and its sync. + * + * One process draws to one terminal, so a single slot is enough. + */ +export type TerminalOwner = { + /** Stops drawing and returns the message to restore with, or `null` if idle. */ + suspend(): string | null; + resume(message: string | null): void; +}; + +let active: TerminalOwner | null = null; + +export function setTerminalOwner(owner: TerminalOwner): void { + active = owner; +} + +export function clearTerminalOwner(owner: TerminalOwner): void { + if (active === owner) active = null; +} + +/** + * Suspends the current owner so the caller can draw. Returns a restore function, + * or `null` when nothing held the line. + */ +export function takeTerminal(): (() => void) | null { + const owner = active; + if (!owner) return null; + const restore = owner.suspend(); + if (restore === null) return null; + return () => owner.resume(restore); +} diff --git a/src/utils/proving-artifacts.ts b/src/utils/proving-artifacts.ts index ccb1653..9811d33 100644 --- a/src/utils/proving-artifacts.ts +++ b/src/utils/proving-artifacts.ts @@ -171,6 +171,19 @@ export function parseFetchArtifactSelection(opts: { return [...out].sort(); } +/** + * Public-sync snapshot chunks are published under the artifacts base but are + * not proving artifacts. Routing them through the artifact path would apply the + * Tor CDN hard timeout, duplicate every chunk into `proving-artifacts`, and pin + * later runs to the first snapshot ever fetched. + */ +function isSyncCacheRelativeKey(relativeKey: string): boolean { + return ( + relativeKey.startsWith("sync-cache/") || + relativeKey === "public-sync-cache.tar.gz" + ); +} + /** * Map a plugin/WASM request URL to a cache-relative key, or null if not an * artifact we manage. @@ -182,14 +195,16 @@ export function artifactRelativeKeyFromUrl(urlStr: string): string | null { if (href.startsWith(MACWHA_ARTIFACTS_PREFIX)) { const rest = href.slice(MACWHA_ARTIFACTS_PREFIX.length).replace(/^\/+/, ""); - return rest || null; + if (!rest || isSyncCacheRelativeKey(rest)) return null; + return rest; } const base = resolveArtifactsBaseUrl(); const basePrefix = `${base}/`; if (href.startsWith(basePrefix)) { const rest = href.slice(basePrefix.length).replace(/^\/+/, ""); - return rest || null; + if (!rest || isSyncCacheRelativeKey(rest)) return null; + return rest; } if (href === TORNADO_CIRCUIT_URL || href.endsWith("/tornado.json")) { diff --git a/src/utils/public-sync-cache.ts b/src/utils/public-sync-cache.ts new file mode 100644 index 0000000..e391ec5 --- /dev/null +++ b/src/utils/public-sync-cache.ts @@ -0,0 +1,622 @@ +import { spawnSync } from "node:child_process"; +import { createHash } from "node:crypto"; +import { + copyFileSync, + existsSync, + mkdirSync, + mkdtempSync, + readdirSync, + readFileSync, + rmSync, + statSync, + writeFileSync, +} from "node:fs"; +import { homedir, tmpdir } from "node:os"; +import { basename, join } from "node:path"; +import { gzipSync } from "node:zlib"; + +import { categorizeUrl } from "./network-traffic-log.js"; + +const DEFAULT_MAX_BYTES = 1024 * 1024 * 1024; +const DEFAULT_MAX_ENTRY_BYTES = 50 * 1024 * 1024; +const CACHE_KEY_FILE_RE = /^[a-f0-9]{64}\.(bin|json)$/; + +function defaultDataDir(): string { + return join(homedir(), ".kohaku-cli"); +} + +let activeDataDir: string = defaultDataDir(); + +export function publicSyncCacheDir(dataDir: string = activeDataDir): string { + return join(dataDir, "public-sync-cache"); +} + +export function getPublicSyncCacheDataDir(): string { + return activeDataDir; +} + +export function setPublicSyncCacheDataDir(dataDir: string): void { + activeDataDir = dataDir; +} + +export function isPublicSyncCacheableUrl(url: string): boolean { + const category = categorizeUrl(url); + return category === "subsquid" || category === "saga"; +} + +export function isPublicSyncCacheableRequest( + url: string, + method: string +): boolean { + const m = method.toUpperCase(); + if (m !== "GET" && m !== "HEAD" && m !== "POST") return false; + return isPublicSyncCacheableUrl(url); +} + +export function publicSyncCacheKey( + method: string, + url: string, + body: Uint8Array | Buffer | string = "" +): string { + const h = createHash("sha256"); + h.update(method.toUpperCase()); + h.update("\n"); + h.update(url); + h.update("\n"); + if (typeof body === "string") h.update(body); + else h.update(body); + return h.digest("hex"); +} + +type CacheMeta = { + url: string; + method: string; + contentType: string | null; + storedAt: string; + bytes: number; +}; + +function metaPath(dir: string, key: string): string { + return join(dir, `${key}.json`); +} + +function bodyPath(dir: string, key: string): string { + return join(dir, `${key}.bin`); +} + +export function publicSyncCacheHasEntries( + dataDir: string = activeDataDir +): boolean { + const dir = publicSyncCacheDir(dataDir); + if (!existsSync(dir)) return false; + return readdirSync(dir).some((name) => name.endsWith(".bin")); +} + +export function clearPublicSyncCache( + dataDir: string = activeDataDir +): { cleared: boolean; path: string; filesRemoved: number } { + const dir = publicSyncCacheDir(dataDir); + if (!existsSync(dir)) { + return { cleared: false, path: dir, filesRemoved: 0 }; + } + const files = readdirSync(dir); + rmSync(dir, { recursive: true, force: true }); + return { cleared: files.length > 0, path: dir, filesRemoved: files.length }; +} + +function maxCacheBytes(): number { + const raw = process.env.KOHAKU_PUBLIC_SYNC_CACHE_MAX_BYTES?.trim(); + if (raw && /^\d+$/.test(raw)) return Number(raw); + return DEFAULT_MAX_BYTES; +} + +function cacheBytesOnDisk(dir: string): number { + if (!existsSync(dir)) return 0; + let total = 0; + for (const name of readdirSync(dir)) { + if (!CACHE_KEY_FILE_RE.test(name)) continue; + try { + total += statSync(join(dir, name)).size; + } catch { + // ignore unreadable entries + } + } + return total; +} + +/** + * The cache never evicts: once it is full, new responses simply are not stored. + * Entries already on disk (notably an installed snapshot) are therefore never + * dropped to make room for fresher pages, which would silently punch holes in a + * snapshot the user intends to keep or republish. + */ +function hasRoomFor(dir: string, key: string, incomingBytes: number): boolean { + const max = maxCacheBytes(); + let total = cacheBytesOnDisk(dir); + // Refreshing an existing key replaces its bytes rather than adding to them. + for (const path of [bodyPath(dir, key), metaPath(dir, key)]) { + try { + if (existsSync(path)) total -= statSync(path).size; + } catch { + // ignore unreadable entries + } + } + return total + incomingBytes <= max; +} + +export function readPublicSyncCache(opts: { + method: string; + url: string; + body?: Uint8Array | Buffer | string; + dataDir?: string; +}): { body: Buffer; contentType: string | null } | null { + const dir = publicSyncCacheDir(opts.dataDir ?? activeDataDir); + const key = publicSyncCacheKey(opts.method, opts.url, opts.body ?? ""); + const bin = bodyPath(dir, key); + const metaFile = metaPath(dir, key); + if (!existsSync(bin) || !existsSync(metaFile)) return null; + try { + const meta = JSON.parse(readFileSync(metaFile, "utf8")) as CacheMeta; + const body = readFileSync(bin); + return { body, contentType: meta.contentType ?? null }; + } catch { + return null; + } +} + +/** Persist a network response only when it is HTTP 200. */ +export function cachePublicSyncNetworkResponse(opts: { + method: string; + url: string; + body?: Uint8Array | Buffer | string; + status: number; + responseBody: Uint8Array | Buffer; + contentType?: string | null; + dataDir?: string; +}): void { + if (opts.status !== 200) return; + writePublicSyncCache({ + method: opts.method, + url: opts.url, + body: opts.body, + responseBody: opts.responseBody, + contentType: opts.contentType, + dataDir: opts.dataDir, + }); +} + +export function writePublicSyncCache(opts: { + method: string; + url: string; + body?: Uint8Array | Buffer | string; + responseBody: Uint8Array | Buffer; + contentType?: string | null; + dataDir?: string; +}): void { + const responseBody = Buffer.from(opts.responseBody); + if (responseBody.byteLength > DEFAULT_MAX_ENTRY_BYTES) return; + + const dir = publicSyncCacheDir(opts.dataDir ?? activeDataDir); + const key = publicSyncCacheKey(opts.method, opts.url, opts.body ?? ""); + const meta: CacheMeta = { + url: opts.url, + method: opts.method.toUpperCase(), + contentType: opts.contentType ?? null, + storedAt: new Date().toISOString(), + bytes: responseBody.byteLength, + }; + const metaBody = JSON.stringify(meta); + + mkdirSync(dir, { recursive: true }); + if ( + !hasRoomFor(dir, key, responseBody.byteLength + Buffer.byteLength(metaBody)) + ) { + return; + } + + writeFileSync(bodyPath(dir, key), responseBody); + writeFileSync(metaPath(dir, key), metaBody); +} + +export function buildCachedPublicSyncResponse( + requestUrl: string, + body: Buffer, + contentType: string | null +): Response { + const copy = Uint8Array.from(body); + const headers = new Headers({ + "content-length": String(copy.byteLength), + "x-kohaku-public-cache": "hit", + }); + if (contentType) headers.set("content-type", contentType); + const res = new Response(copy, { status: 200, statusText: "OK", headers }); + try { + Object.defineProperty(res, "url", { + value: requestUrl, + configurable: true, + enumerable: true, + }); + } catch { + // ignore + } + return res; +} + +/** + * Snapshot chunks live under a versioned prefix so publishing a newer set does + * not break clients pinned to an older manifest. Override with + * `KOHAKU_SYNC_CACHE_BASE_URL`. + */ +export const DEFAULT_SYNC_CACHE_BASE_URL = + "https://artifacts.0000000000.org/sync-cache/v1"; + +/** Chunks are ~8 MiB, but Tor throughput varies wildly; be generous. */ +const DEFAULT_CHUNK_TIMEOUT_MS = 300_000; + +/** + * Above the largest observed single compressed entry (~6.6 MiB), so no entry is + * forced into an oversized chunk of its own. + */ +export const DEFAULT_CHUNK_BYTES = 8 * 1024 * 1024; + +export const SYNC_CACHE_MANIFEST_NAME = "manifest.json"; +const CHUNK_NAME_RE = /^chunk-\d{3,}\.tar\.gz$/; +const SHA256_RE = /^[a-f0-9]{64}$/; +const CACHE_KEY_RE = /^[a-f0-9]{64}$/; + +export function resolveSyncCacheBaseUrl(): string { + const fromEnv = process.env.KOHAKU_SYNC_CACHE_BASE_URL?.trim(); + const url = fromEnv || DEFAULT_SYNC_CACHE_BASE_URL; + return url.replace(/\/+$/, ""); +} + +export function syncCacheManifestUrl(): string { + return `${resolveSyncCacheBaseUrl()}/${SYNC_CACHE_MANIFEST_NAME}`; +} + +export function syncCacheChunkUrl(chunkName: string): string { + if (!CHUNK_NAME_RE.test(chunkName)) { + throw new Error(`Refusing unsafe chunk name ${JSON.stringify(chunkName)}`); + } + return `${resolveSyncCacheBaseUrl()}/${chunkName}`; +} + +export function resolveSyncCacheChunkTimeoutMs(): number { + const raw = process.env.KOHAKU_SYNC_CACHE_CHUNK_TIMEOUT_MS?.trim(); + if (!raw) return DEFAULT_CHUNK_TIMEOUT_MS; + const n = Number(raw); + if (!Number.isFinite(n) || n < 1_000) return DEFAULT_CHUNK_TIMEOUT_MS; + return Math.floor(n); +} + +function cacheEntryKeys(dir: string): string[] { + if (!existsSync(dir)) return []; + const keys = new Set(); + for (const name of readdirSync(dir)) { + if (!name.endsWith(".bin")) continue; + const key = name.slice(0, -".bin".length); + if (!CACHE_KEY_FILE_RE.test(`${key}.bin`)) continue; + if (existsSync(metaPath(dir, key))) keys.add(key); + } + return [...keys]; +} + +export function countPublicSyncCacheEntries( + dataDir: string = activeDataDir +): number { + return cacheEntryKeys(publicSyncCacheDir(dataDir)).length; +} + +function runTar(args: string[]): void { + const result = spawnSync("tar", args, { encoding: "utf8" }); + if (result.error) { + throw new Error( + `tar failed to start (${result.error.message}). Install tar to pack/install public-sync snapshots.` + ); + } + if (result.status !== 0) { + throw new Error( + `tar ${args[0]} failed: ${(result.stderr || result.stdout || "").trim() || `exit ${result.status}`}` + ); + } +} + +function assertSafeTarListing(archivePath: string): void { + const result = spawnSync("tar", ["-tzf", archivePath], { encoding: "utf8" }); + if (result.error) { + throw new Error( + `tar failed to start (${result.error.message}). Install tar to install public-sync snapshots.` + ); + } + if (result.status !== 0) { + throw new Error( + `tar list failed: ${(result.stderr || result.stdout || "").trim() || `exit ${result.status}`}` + ); + } + for (const line of (result.stdout ?? "").split("\n")) { + const entry = line.trim(); + if (!entry) continue; + if (entry.startsWith("/") || entry.includes("..")) { + throw new Error( + `Refusing public-sync snapshot with unsafe path ${JSON.stringify(entry)}` + ); + } + } +} + +function walkFiles(dir: string, acc: string[] = []): string[] { + if (!existsSync(dir)) return acc; + for (const name of readdirSync(dir, { withFileTypes: true })) { + const full = join(dir, name.name); + if (name.isDirectory()) walkFiles(full, acc); + else if (name.isFile()) acc.push(full); + } + return acc; +} + +export type SyncCacheChunk = { + name: string; + bytes: number; + sha256: string; + keys: string[]; +}; + +export type SyncCacheManifest = { + version: 1; + createdAt: string; + entries: number; + rawBytes: number; + chunks: SyncCacheChunk[]; +}; + +function sha256File(path: string): string { + return createHash("sha256").update(readFileSync(path)).digest("hex"); +} + +/** + * Group keys into bins under `chunkBytes` of compressed output. gzip's window is + * only 32 KiB, so a member's standalone compressed size is a good predictor of + * its contribution to the chunk. First-fit-decreasing keeps the bins full; an + * entry larger than the target gets a chunk to itself. + */ +function binPackKeys( + cacheDir: string, + keys: string[], + chunkBytes: number +): string[][] { + const sized = keys + .map((key) => ({ + key, + size: + gzipSync(readFileSync(bodyPath(cacheDir, key))).byteLength + + statSync(metaPath(cacheDir, key)).size, + })) + .sort((a, b) => b.size - a.size); + + const bins: Array<{ keys: string[]; size: number }> = []; + for (const { key, size } of sized) { + const bin = bins.find((b) => b.size + size <= chunkBytes); + if (bin) { + bin.keys.push(key); + bin.size += size; + } else { + bins.push({ keys: [key], size }); + } + } + return bins.map((b) => b.keys.sort()); +} + +/** + * Pack `/public-sync-cache` into independently verifiable `.tar.gz` + * chunks plus a manifest, so a consumer can fetch, check, and extract them one + * at a time instead of transferring one large archive. + */ +export function packPublicSyncCacheChunks(opts: { + dataDir?: string; + outputDir: string; + chunkBytes?: number; +}): { outputDir: string; manifest: SyncCacheManifest } { + const dataDir = opts.dataDir ?? activeDataDir; + const chunkBytes = opts.chunkBytes ?? DEFAULT_CHUNK_BYTES; + if (chunkBytes < 1024) { + throw new Error(`chunkBytes must be at least 1024 (got ${chunkBytes})`); + } + const cacheDir = publicSyncCacheDir(dataDir); + const selected = cacheEntryKeys(cacheDir); + if (selected.length === 0) { + throw new Error(`No public-sync-cache entries in ${cacheDir}`); + } + + mkdirSync(opts.outputDir, { recursive: true }); + const groups = binPackKeys(cacheDir, selected, chunkBytes); + const chunks: SyncCacheChunk[] = []; + let rawBytes = 0; + + for (let i = 0; i < groups.length; i++) { + const keys = groups[i]!; + const name = `chunk-${String(i).padStart(3, "0")}.tar.gz`; + const chunkPath = join(opts.outputDir, name); + const staging = mkdtempSync(join(tmpdir(), "kohaku-sync-cache-pack-")); + try { + for (const key of keys) { + copyFileSync(bodyPath(cacheDir, key), join(staging, `${key}.bin`)); + copyFileSync(metaPath(cacheDir, key), join(staging, `${key}.json`)); + rawBytes += + statSync(bodyPath(cacheDir, key)).size + + statSync(metaPath(cacheDir, key)).size; + } + runTar(["-czf", chunkPath, "-C", staging, "."]); + } finally { + rmSync(staging, { recursive: true, force: true }); + } + chunks.push({ + name, + bytes: statSync(chunkPath).size, + sha256: sha256File(chunkPath), + keys, + }); + } + + const manifest: SyncCacheManifest = { + version: 1, + createdAt: new Date().toISOString(), + entries: selected.length, + rawBytes, + chunks, + }; + writeFileSync( + join(opts.outputDir, SYNC_CACHE_MANIFEST_NAME), + `${JSON.stringify(manifest, null, 2)}\n` + ); + return { outputDir: opts.outputDir, manifest }; +} + +/** Validate an untrusted manifest before acting on any of its chunk names. */ +export function parseSyncCacheManifest(raw: string): SyncCacheManifest { + let parsed: unknown; + try { + parsed = JSON.parse(raw); + } catch (e) { + throw new Error( + `Sync-cache manifest is not valid JSON: ${e instanceof Error ? e.message : String(e)}` + ); + } + if (!parsed || typeof parsed !== "object") { + throw new Error("Sync-cache manifest must be a JSON object"); + } + const m = parsed as Record; + if (m.version !== 1) { + throw new Error( + `Unsupported sync-cache manifest version ${JSON.stringify(m.version)} (expected 1). Upgrade kohaku-cli.` + ); + } + if (!Array.isArray(m.chunks) || m.chunks.length === 0) { + throw new Error("Sync-cache manifest has no chunks"); + } + + const seen = new Set(); + const chunks: SyncCacheChunk[] = m.chunks.map((entry, i) => { + if (!entry || typeof entry !== "object") { + throw new Error(`Sync-cache manifest chunk ${i} is not an object`); + } + const c = entry as Record; + const name = c.name; + if (typeof name !== "string" || !CHUNK_NAME_RE.test(name)) { + throw new Error( + `Refusing unsafe chunk name ${JSON.stringify(name)} in sync-cache manifest` + ); + } + if (seen.has(name)) { + throw new Error(`Duplicate chunk ${name} in sync-cache manifest`); + } + seen.add(name); + if (typeof c.sha256 !== "string" || !SHA256_RE.test(c.sha256)) { + throw new Error(`Chunk ${name} has no valid sha256 in sync-cache manifest`); + } + if (typeof c.bytes !== "number" || !Number.isInteger(c.bytes) || c.bytes <= 0) { + throw new Error(`Chunk ${name} has no valid byte size in sync-cache manifest`); + } + if (!Array.isArray(c.keys) || c.keys.length === 0) { + throw new Error(`Chunk ${name} lists no cache keys in sync-cache manifest`); + } + const keys = c.keys.map((k) => { + if (typeof k !== "string" || !CACHE_KEY_RE.test(k)) { + throw new Error( + `Chunk ${name} lists invalid cache key ${JSON.stringify(k)}` + ); + } + return k; + }); + return { name, bytes: c.bytes, sha256: c.sha256, keys }; + }); + + return { + version: 1, + createdAt: typeof m.createdAt === "string" ? m.createdAt : "", + entries: + typeof m.entries === "number" && Number.isInteger(m.entries) + ? m.entries + : chunks.reduce((n, c) => n + c.keys.length, 0), + rawBytes: typeof m.rawBytes === "number" ? m.rawBytes : 0, + chunks, + }; +} + +/** True when every key already has both an on-disk body and metadata file. */ +export function publicSyncCacheHasAllKeys( + keys: string[], + dataDir: string = activeDataDir +): boolean { + const dir = publicSyncCacheDir(dataDir); + return keys.every( + (key) => existsSync(bodyPath(dir, key)) && existsSync(metaPath(dir, key)) + ); +} + +/** + * Verify a downloaded chunk against its manifest entry, then merge it into the + * cache. Rejects the chunk rather than installing partial data on mismatch. + */ +export function installPublicSyncCacheChunk(opts: { + dataDir?: string; + archivePath: string; + chunk: SyncCacheChunk; +}): { installed: number; cacheDir: string } { + const { archivePath, chunk } = opts; + if (!existsSync(archivePath)) { + throw new Error(`Chunk archive not found: ${archivePath}`); + } + const actualBytes = statSync(archivePath).size; + if (actualBytes !== chunk.bytes) { + throw new Error( + `Chunk ${chunk.name} size mismatch: expected ${chunk.bytes} bytes, got ${actualBytes}` + ); + } + const actualSha = sha256File(archivePath); + if (actualSha !== chunk.sha256) { + throw new Error( + `Chunk ${chunk.name} sha256 mismatch: expected ${chunk.sha256}, got ${actualSha}` + ); + } + return installPublicSyncCacheArchive({ + dataDir: opts.dataDir, + archivePath, + }); +} + +/** + * Extract a snapshot tarball into `/public-sync-cache` (merge; + * existing keys are overwritten). + */ +export function installPublicSyncCacheArchive(opts: { + dataDir?: string; + archivePath: string; +}): { installed: number; cacheDir: string } { + const dataDir = opts.dataDir ?? activeDataDir; + const cacheDir = publicSyncCacheDir(dataDir); + if (!existsSync(opts.archivePath)) { + throw new Error(`Snapshot archive not found: ${opts.archivePath}`); + } + assertSafeTarListing(opts.archivePath); + + const staging = mkdtempSync(join(tmpdir(), "kohaku-public-sync-unpack-")); + try { + runTar(["-xzf", opts.archivePath, "-C", staging]); + mkdirSync(cacheDir, { recursive: true }); + let installed = 0; + for (const file of walkFiles(staging)) { + const name = basename(file); + if (!CACHE_KEY_FILE_RE.test(name)) continue; + copyFileSync(file, join(cacheDir, name)); + if (name.endsWith(".bin")) installed += 1; + } + if (installed === 0) { + throw new Error( + `No public-sync-cache entries found in ${opts.archivePath}` + ); + } + return { installed, cacheDir }; + } finally { + rmSync(staging, { recursive: true, force: true }); + } +} diff --git a/src/utils/railgun-subsquid-progress.ts b/src/utils/railgun-subsquid-progress.ts deleted file mode 100644 index ead5a7a..0000000 --- a/src/utils/railgun-subsquid-progress.ts +++ /dev/null @@ -1,105 +0,0 @@ -import type { SupportedProtocol } from "./plugins.js"; -import { reportSyncProgress } from "./sync-progress.js"; -import { kohakuFetch } from "./tor.js"; - -/** - * `SubsquidSyncer::new` in railgun-rs pages GraphQL at this size, then - * fetches one empty terminating page per stream (`fetch_paged`). - */ -export const RAILGUN_SUBSQUID_PAGE_SIZE = 20_000; - -/** - * UtxoSyncer and TxidSyncer each call `latest_block()` (BlockNumberQuery) - * before paging commitments / nullifiers / transactions. - */ -const LATEST_BLOCK_QUERIES = 2; - -/** - * Fallback URLs from railgun-rs `ChainConfig` (WASM `chainConfig()` is not - * safe to call before `ensureInitialized`, which happens only when the - * plugin is constructed). - */ -const SUBSQUID_ENDPOINTS: Record = { - "1": "https://rail-squid.squids.live/squid-railgun-ethereum-v2/v/v1/graphql", - "11155111": - "https://rail-squid.squids.live/squid-railgun-eth-sepolia-v2/v/v1/graphql", -}; - -const COUNT_QUERY = `query RailgunSyncEstimate { - commitmentsConnection(orderBy: id_ASC) { totalCount } - nullifiersConnection(orderBy: id_ASC) { totalCount } - transactionsConnection(orderBy: id_ASC) { totalCount } -}`; - -function pagesForCount(count: number): number { - if (!Number.isFinite(count) || count <= 0) return 1; - return Math.ceil(count / RAILGUN_SUBSQUID_PAGE_SIZE) + 1; -} - -type CountResponse = { - errors?: { message?: string }[]; - data?: { - commitmentsConnection?: { totalCount?: number }; - nullifiersConnection?: { totalCount?: number }; - transactionsConnection?: { totalCount?: number }; - }; -}; - -function asCount(value: unknown): number | undefined { - return typeof value === "number" && Number.isFinite(value) ? value : undefined; -} - -/** - * One GraphQL count query so the Railgun spinner can show a stable - * `done/total` (not n/n+1). Cold-sync estimate: this probe + 2 latest-block - * queries + paged commitments, nullifiers, and transactions. Incremental - * sync may finish early. Failures leave progress as a request count only. - */ -export async function primeRailgunSubsquidProgress( - chainId: bigint -): Promise { - const endpoint = SUBSQUID_ENDPOINTS[chainId.toString()]; - if (!endpoint) return; - - const res = await kohakuFetch(endpoint, { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify({ query: COUNT_QUERY }), - }); - if (!res.ok) return; - - const json = (await res.json()) as CountResponse; - if (json.errors?.length || !json.data) return; - - const commitments = asCount(json.data.commitmentsConnection?.totalCount); - const nullifiers = asCount(json.data.nullifiersConnection?.totalCount); - const transactions = asCount(json.data.transactionsConnection?.totalCount); - if ( - commitments == null || - nullifiers == null || - transactions == null - ) { - return; - } - - // Probe HTTP is already counted. WASM still does latest-block + 3 streams. - const total = - 1 + - LATEST_BLOCK_QUERIES + - pagesForCount(commitments) + - pagesForCount(nullifiers) + - pagesForCount(transactions); - reportSyncProgress({ phase: "subsquid", total }); -} - -export async function primeRailgunSubsquidProgressIfNeeded( - protocol: SupportedProtocol, - chainId: bigint -): Promise { - if (protocol !== "railgun") return; - try { - await primeRailgunSubsquidProgress(chainId); - } catch { - // Leave the spinner on request-count + elapsed (no fake n/n+1 bar). - } -} diff --git a/src/utils/saga-external-sync.ts b/src/utils/saga-external-sync.ts index 20070b1..61d5817 100644 --- a/src/utils/saga-external-sync.ts +++ b/src/utils/saga-external-sync.ts @@ -9,8 +9,6 @@ import type { Network, } from "@kohaku-eth/plugins"; -import { reportSyncProgress } from "./sync-progress.js"; - type Hex = `0x${string}`; type SagaChunk = { @@ -59,19 +57,6 @@ function collectSegments(entry: SagaProtocolEntry): SagaChunk[] { ); } -/** Unique gzip files for Tornado Cash entries only (index also lists PP / Railgun). */ -function uniqueTornadoSagaFiles(index: SagaIndex, chainId?: bigint): Set { - const files = new Set(); - for (const [key, entry] of Object.entries(index.availableProtocols)) { - if (!key.startsWith("tornado-cash-")) continue; - if (chainId != null && hexToBigInt(entry.chainId) !== chainId) continue; - for (const seg of collectSegments(entry)) { - if (seg.file) files.add(seg.file); - } - } - return files; -} - async function loadIndex( network: Network, baseUrl: string @@ -131,50 +116,22 @@ async function readGunzipLines( * Host {@link ExternalSyncProvider} backed by the saga CDN. * Tornado Cash reads bulk historical logs from here during cold sync. * - * Progress `total` is 1 (`index.json`) plus unique `tornado-cash-*` gzip - * files on this chain. Incremental sync may finish before 100% (hot-head - * only). The same index also lists Privacy Pools / Railgun; those files - * are not counted. + * Every fetch here is counted as a saga request by the Tor fetch wrapper, so + * this provider does no progress bookkeeping of its own. */ export function createSagaExternalSyncProvider(opts: { baseUrl: string; network: Network; - chainId?: bigint; }): ExternalSyncProvider & { firstCoveredBlock(params: ExternalSyncPoolId): Promise; } { let indexPromise: Promise | null = null; - const plannedFiles = new Set(); - - const reportPlannedTotal = () => { - reportSyncProgress({ - phase: "saga", - total: Math.max(1 + plannedFiles.size, 1), - }); - }; const getIndex = () => { - indexPromise ??= loadIndex(opts.network, opts.baseUrl).then((index) => { - for (const file of uniqueTornadoSagaFiles(index, opts.chainId)) { - plannedFiles.add(file); - } - // index.json is already counted as 1 saga HTTP request. - reportSyncProgress({ - phase: "saga", - done: 1, - total: Math.max(1 + plannedFiles.size, 1), - }); - return index; - }); + indexPromise ??= loadIndex(opts.network, opts.baseUrl); return indexPromise; }; - const notePlannedFiles = (files: Iterable) => { - const before = plannedFiles.size; - for (const file of files) plannedFiles.add(file); - if (plannedFiles.size !== before) reportPlannedTotal(); - }; - const getEntry = async (params: ExternalSyncPoolId) => { const entry = findProtocolEntry(await getIndex(), params); if (!entry) { @@ -224,9 +181,6 @@ export function createSagaExternalSyncProvider(opts: { return segFrom <= toBlock && segLast >= fromBlock; }); - notePlannedFiles(segments.map((seg) => seg.file)); - reportPlannedTotal(); - const out: ExternalRawEvent[] = []; for (const seg of segments) { const url = `${opts.baseUrl}/${seg.file}`; @@ -266,6 +220,5 @@ export function tornadoExternalSyncForChain( return createSagaExternalSyncProvider({ baseUrl: sagaBaseUrlForChain(chainId), network, - chainId, }); } diff --git a/src/utils/sync-progress-renderer.ts b/src/utils/sync-progress-renderer.ts new file mode 100644 index 0000000..9a7148e --- /dev/null +++ b/src/utils/sync-progress-renderer.ts @@ -0,0 +1,150 @@ +import { writeSync } from "node:fs"; +import { Worker } from "node:worker_threads"; + +import { takeTerminal } from "./progress-terminal.js"; + +/** Cursor to column 0 + erase below, the same reset clack uses. */ +const CLEAR_LINE = "\u001b[999D\u001b[J"; +const HIDE_CURSOR = "\u001b[?25l"; + +/** + * Draws the sync status line from a worker thread. + * + * Protocol syncs spend most of their wall time inside synchronous WASM: a real + * Railgun first sync blocks for ~27s building UTXO trees, then ~65s on txid + * trees and POI. Nothing on the main thread can animate through that, so the + * spinner and the elapsed clock freeze and the command looks hung. + * + * A worker has its own event loop, so it keeps drawing. It must write to the fd + * directly: a worker's `process.stdout` is piped through the *parent* thread, so + * those writes would queue up behind the very block we are drawing through. + */ +const WORKER_SOURCE = ` +const { parentPort, workerData } = require('node:worker_threads'); +const { writeSync } = require('node:fs'); + +const FRAMES = ['\u25D2', '\u25D0', '\u25D3', '\u25D1']; +const CLEAR = ${JSON.stringify(CLEAR_LINE)}; +const STALE_MS = 2000; + +let prefix = workerData.prefix; +const startedAt = workerData.startedAt; +let lastUpdate = Date.now(); +let frame = 0; + +function elapsed() { + const secs = Math.max(0, Math.round((Date.now() - startedAt) / 1000)); + if (secs < 60) return secs + 's'; + const mins = Math.floor(secs / 60); + return mins + 'm ' + String(secs % 60).padStart(2, '0') + 's'; +} + +function write(text) { + try { + writeSync(1, text); + } catch { + // A closed or busy fd must never take down a sync. + } +} + +function render() { + // No update for a while means the main thread is busy, not that we are stuck + // waiting on the network the phase label names. + const busy = Date.now() - lastUpdate > STALE_MS ? ' \u00b7 still working' : ''; + write( + CLEAR + + FRAMES[frame++ % FRAMES.length] + + ' ' + prefix + ' \u00b7 ' + elapsed() + busy + ); +} + +const timer = setInterval(render, 120); +render(); + +parentPort.on('message', (msg) => { + if (msg && msg.type === 'state') { + prefix = msg.prefix; + lastUpdate = Date.now(); + } +}); +`; + +export type ProgressRenderer = { + update(prefix: string): void; + stop(): void; +}; + +function disabled(): boolean { + const raw = process.env.KOHAKU_NO_WORKER_PROGRESS?.trim().toLowerCase(); + return raw === "1" || raw === "true" || raw === "yes"; +} + +/** + * Takes over the terminal line and animates it from a worker thread. + * + * Returns `null` when worker rendering is unavailable or nothing owns the line, + * in which case the caller falls back to emitting plain message strings. + */ +export function startProgressRenderer( + prefix: string, + startedAt: number, + prelude?: string +): ProgressRenderer | null { + if (disabled() || !process.stdout.isTTY) return null; + + // A clack spinner erases everything below its cursor every frame, so it has + // to stand down before the worker can draw. + const restoreTerminal = takeTerminal(); + if (!restoreTerminal) return null; + + if (prelude) { + try { + writeSync(1, prelude.endsWith("\n") ? prelude : `${prelude}\n`); + } catch { + // A closed fd must never take down a sync. + } + } + + let worker: Worker; + try { + worker = new Worker(WORKER_SOURCE, { + eval: true, + workerData: { prefix, startedAt }, + stdout: true, + stderr: true, + }); + } catch { + restoreTerminal(); + return null; + } + // Never hold the process open or surface a render failure as a sync failure. + worker.unref(); + worker.on("error", () => {}); + + writeSync(1, HIDE_CURSOR); + let stopped = false; + + return { + update(next: string) { + if (stopped) return; + try { + worker.postMessage({ type: "state", prefix: next }); + } catch { + // Worker already gone; the line just stops updating. + } + }, + stop() { + if (stopped) return; + stopped = true; + // Terminate before clearing so the worker cannot redraw over the line + // between our clear and the resumed spinner's first frame. + void worker.terminate(); + try { + writeSync(1, CLEAR_LINE); + } catch { + // ignore + } + restoreTerminal(); + }, + }; +} diff --git a/src/utils/sync-progress.ts b/src/utils/sync-progress.ts index b93c125..ac660f0 100644 --- a/src/utils/sync-progress.ts +++ b/src/utils/sync-progress.ts @@ -1,52 +1,55 @@ +import { writeSync } from "node:fs"; import { AsyncLocalStorage } from "node:async_hooks"; import type { SupportedProtocol } from "./plugins.js"; +import { + startProgressRenderer, + type ProgressRenderer, +} from "./sync-progress-renderer.js"; /** Privacy-protocol sync, or ERC-5564 stealth announcement scan. */ export type SyncProgressSource = SupportedProtocol | "stealth"; -export type SyncProgressPhase = "saga" | "rpc" | "subsquid" | "asp" | "sync"; - -export type SyncProgressUpdate = { - phase?: SyncProgressPhase; - done?: number; - total?: number; - detail?: string; -}; +/** Where the current sync work is happening. */ +export type SyncProgressPhase = "rpc" | "subsquid" | "saga" | "asp"; type SyncProgressStore = { - protocol: SyncProgressSource; + source: SyncProgressSource; + firstRun: boolean; started: number; onUpdate?: (message: string) => void; - phase: SyncProgressPhase; - done?: number; - total?: number; - detail?: string; - httpCounts: Partial>; + phase?: SyncProgressPhase; + counts: Partial>; lastEmit: number; lastMessage: string; emitTimer?: ReturnType; + /** Owns the status line when set; it renders the elapsed clock itself. */ + renderer?: ProgressRenderer; + lastPrefix?: string; }; const als = new AsyncLocalStorage(); -const PROTOCOL_LABEL: Record = { +const SOURCE_LABEL: Record = { railgun: "Railgun", "privacy-pools": "Privacy Pools", tornado: "Tornado Cash", stealth: "Stealth", }; -const PHASE_NOUN: Record = { - saga: "saga req", +const PHASE_LABEL: Record = { rpc: "RPC logs", - subsquid: "Subsquid", + subsquid: "Subsquid over Tor", + saga: "saga CDN", asp: "ASP", - sync: "sync", }; const COALESCE_MS = 250; -const TICK_MS = 5_000; +/** + * Redraw at least once a second so the elapsed counter visibly advances during + * long stretches with no progress events (cached pages feeding CPU-bound WASM). + */ +const TICK_MS = 1_000; const SLOW_HINT_MS = 8_000; function formatElapsed(ms: number): string { @@ -57,58 +60,32 @@ function formatElapsed(ms: number): string { return `${m}m ${s.toString().padStart(2, "0")}s`; } -function asciiBar(done: number, total: number, width = 14): string { - const pct = Math.min(1, Math.max(0, done / total)); - const filled = Math.round(pct * width); - if (filled >= width) return `[${"=".repeat(width)}]`; - if (filled <= 0) return `[${" ".repeat(width)}]`; - return `[${"=".repeat(filled - 1)}>${" ".repeat(width - filled)}]`; -} - -function slowHint(store: SyncProgressStore, hasBar: boolean): string { - if (hasBar) return ""; - if (Date.now() - store.started < SLOW_HINT_MS) return ""; - return " · first run can take several minutes"; +/** Everything before the elapsed time, which the renderer appends itself. */ +function formatPrefix(store: SyncProgressStore): string { + const name = SOURCE_LABEL[store.source]; + const verb = store.source === "stealth" ? "scan" : "sync"; + const label = store.firstRun ? `${name} first ${verb}` : `${name} ${verb}`; + const phase = store.phase ? PHASE_LABEL[store.phase] : "starting"; + const count = store.phase ? store.counts[store.phase] : undefined; + return `${label} · ${phase}${count ? ` · ${count} req` : ""}`; } function formatMessage(store: SyncProgressStore): string { - const elapsed = formatElapsed(Date.now() - store.started); - const name = PROTOCOL_LABEL[store.protocol]; - const hasBar = - store.total != null && store.total > 0 && store.done != null; - const slow = slowHint(store, hasBar); - - if (hasBar) { - const bar = asciiBar(store.done!, store.total!); - const verb = store.protocol === "stealth" ? "scan" : "first sync"; - const noun = - store.protocol === "stealth" - ? "announcement chunks" - : store.phase === "subsquid" - ? "Subsquid req" - : PHASE_NOUN[store.phase]; - return `${name} ${verb} ${bar} ${store.done}/${store.total} ${noun} ${elapsed}${slow}`; - } - - const count = store.httpCounts[store.phase as "saga" | "subsquid" | "asp"]; - const req = count ? ` · ${count} req` : ""; - - if (store.phase === "subsquid") { - return `${name} sync · Subsquid over Tor${req} · ${elapsed}${slow}`; - } - if (store.phase === "asp") { - return `${name} sync · ASP${req} · ${elapsed}${slow}`; - } - if (store.phase === "saga") { - return `${name} sync · saga CDN${req} · ${elapsed}${slow}`; - } - if (store.detail) { - return `${name} ${store.protocol === "stealth" ? "scan" : "sync"} · ${store.detail} · ${elapsed}${slow}`; - } - return `${name} ${store.protocol === "stealth" ? "scan" : "sync"} · ${elapsed}${slow}`; + const slow = + store.firstRun && Date.now() - store.started >= SLOW_HINT_MS + ? " · first run can take several minutes" + : ""; + return `${formatPrefix(store)} · ${formatElapsed(Date.now() - store.started)}${slow}`; } function flush(store: SyncProgressStore, force = false): void { + if (store.renderer) { + const prefix = formatPrefix(store); + if (prefix === store.lastPrefix) return; + store.lastPrefix = prefix; + store.renderer.update(prefix); + return; + } if (!store.onUpdate) return; const now = Date.now(); if (!force && now - store.lastEmit < COALESCE_MS) { @@ -131,53 +108,49 @@ function flush(store: SyncProgressStore, force = false): void { store.onUpdate(msg); } -/** Report determinate sync progress (saga requests, RPC log windows). */ -export function reportSyncProgress(update: SyncProgressUpdate): void { +/** Switch the reported phase without counting a request. */ +export function reportSyncPhase(phase: SyncProgressPhase): void { const store = als.getStore(); if (!store) return; - if (update.phase) store.phase = update.phase; - if (update.done != null) store.done = update.done; - if (update.total != null) store.total = update.total; - if (update.detail != null) store.detail = update.detail; + store.phase = phase; flush(store); } /** - * Count categorized HTTP (Subsquid / ASP / saga). After a primed `total`, - * `done` advances and `n` stays fixed (no n/n+1). Without a prime, show a - * request count only. Does not replace an RPC catch-up bar. + * Count one request against `phase` and make it the reported phase. The count + * is a liveness signal only: there is no denominator, so it never stalls at a + * wrong total or counts backwards. */ -export function reportSyncHttp(category: "saga" | "subsquid" | "asp"): void { +export function countSyncRequest(phase: SyncProgressPhase): void { const store = als.getStore(); if (!store) return; - store.httpCounts[category] = (store.httpCounts[category] ?? 0) + 1; - - if ( - store.phase === "rpc" && - store.total != null && - store.total > 0 - ) { - flush(store); - return; - } - - if (category === "subsquid" || category === "saga") { - store.phase = category; - store.done = store.httpCounts[category] ?? 0; - flush(store); - return; - } + store.counts[phase] = (store.counts[phase] ?? 0) + 1; + store.phase = phase; + flush(store); +} - store.phase = category; - store.done = undefined; - store.total = undefined; +/** + * Refine the first-run label from inside the scope, for sources that only learn + * it after opening their own storage (the stealth scan). + */ +export function noteSyncFirstRun(firstRun: boolean): void { + const store = als.getStore(); + if (!store) return; + store.firstRun = firstRun; flush(store); } export async function runWithSyncProgress( opts: { - protocol: SyncProgressSource; + source: SyncProgressSource; + firstRun?: boolean; onUpdate?: (message: string) => void; + /** + * Durable line written after taking the terminal and before the progress + * bar. TTY-only (the worker renderer); callers should print it themselves + * when stdout is not a TTY. + */ + prelude?: string; }, fn: () => Promise ): Promise { @@ -187,18 +160,41 @@ export async function runWithSyncProgress( } const store: SyncProgressStore = { - protocol: opts.protocol, + source: opts.source, + firstRun: opts.firstRun ?? false, started: Date.now(), onUpdate: opts.onUpdate, - phase: "sync", - httpCounts: {}, + counts: {}, lastEmit: 0, lastMessage: "", }; - const tick = opts.onUpdate - ? setInterval(() => flush(store, true), TICK_MS) - : undefined; + // A worker keeps drawing through the long synchronous WASM stretches that + // make up most of a sync; the main-thread tick cannot. `onUpdate` being unset + // means quiet mode, where nothing may touch the terminal. + if (opts.onUpdate) { + const prefix = formatPrefix(store); + store.renderer = + startProgressRenderer(prefix, store.started, opts.prelude) ?? undefined; + if (store.renderer) store.lastPrefix = prefix; + else if (opts.prelude && process.stdout.isTTY) { + // Worker renderer unavailable (e.g. KOHAKU_NO_WORKER_PROGRESS): still + // persist the range line after the spinner has the line. + try { + writeSync( + 1, + opts.prelude.endsWith("\n") ? opts.prelude : `${opts.prelude}\n` + ); + } catch { + // ignore + } + } + } + + const tick = + opts.onUpdate && !store.renderer + ? setInterval(() => flush(store, true), TICK_MS) + : undefined; try { return await als.run(store, async () => { @@ -208,6 +204,12 @@ export async function runWithSyncProgress( } finally { if (tick) clearInterval(tick); if (store.emitTimer) clearTimeout(store.emitTimer); + if (store.renderer) { + store.renderer.stop(); + store.renderer = undefined; + // Hand the final line back to the caller's spinner. + opts.onUpdate?.(formatMessage(store)); + } } } @@ -218,11 +220,15 @@ export async function runWithSyncProgress( export async function syncPluginWithProgress( plugin: { sync?: () => Promise }, protocol: SupportedProtocol, - onUpdate?: (message: string) => void + opts: { + firstRun?: boolean; + onUpdate?: (message: string) => void; + } = {} ): Promise { if (protocol !== "privacy-pools" && protocol !== "tornado") return; if (typeof plugin.sync !== "function") return; - await runWithSyncProgress({ protocol, onUpdate }, () => - plugin.sync!.call(plugin) + await runWithSyncProgress( + { source: protocol, firstRun: opts.firstRun, onUpdate: opts.onUpdate }, + () => plugin.sync!.call(plugin) ); } diff --git a/src/utils/tor.ts b/src/utils/tor.ts index 95fd218..6e89903 100644 --- a/src/utils/tor.ts +++ b/src/utils/tor.ts @@ -14,6 +14,8 @@ * fetched via Tor from `KOHAKU_ARTIFACTS_BASE_URL` (default: * https://artifacts.0000000000.org). No clearnet fallback. * Saga CDN (Tornado event sync): Tor-or-fail (no clearnet fallback). + * Subsquid GraphQL and saga CDN responses are also stored under + * `/public-sync-cache` so later wallets reuse public Railgun/Tornado pages. * Artifacts are for prove/unshield, not pool-event sync; pre-warm with * `kohaku fetch-artifacts` (optionally `--without-tor`). * @@ -37,7 +39,7 @@ import { type TrafficClearnetReason, } from "./network-traffic-log.js"; import { formatCaughtError, withTorBootstrapHint } from "./cli-errors.js"; -import { reportSyncHttp } from "./sync-progress.js"; +import { countSyncRequest } from "./sync-progress.js"; import { artifactRelativeKeyFromUrl, buildLocalArtifactResponse, @@ -47,6 +49,13 @@ import { setArtifactsDataDir, writeCachedArtifact, } from "./proving-artifacts.js"; +import { + buildCachedPublicSyncResponse, + cachePublicSyncNetworkResponse, + isPublicSyncCacheableRequest, + readPublicSyncCache, + setPublicSyncCacheDataDir, +} from "./public-sync-cache.js"; /** tor-js Arti directory cache (`createAutoStorage('tor-js')`). */ export function torJsCacheDir(): string { @@ -270,7 +279,7 @@ async function loggedFetch( const started = Date.now(); const category = meta.category ?? categorizeUrl(meta.url); if (category === "subsquid" || category === "asp" || category === "saga") { - reportSyncHttp(category); + countSyncRequest(category); } try { const res = await doFetch(); @@ -412,6 +421,57 @@ function artifactFailHint(): string { * no Tor session) from KOHAKU_ARTIFACTS_BASE_URL — never a Tor→clearnet retry. * Saga CDN: Tor-or-fail with hard timeout. */ + +async function requestBodyBytesForCache( + input: RequestInfo | URL, + init?: RequestInit +): Promise { + const body = init?.body; + if (typeof body === "string") return Buffer.from(body); + if (body instanceof Uint8Array) return Buffer.from(body); + if (typeof ArrayBuffer !== "undefined" && body instanceof ArrayBuffer) { + return Buffer.from(body); + } + if (ArrayBuffer.isView(body)) { + return Buffer.from(body.buffer, body.byteOffset, body.byteLength); + } + if ( + typeof Request !== "undefined" && + input instanceof Request && + init?.body === undefined + ) { + const m = input.method.toUpperCase(); + if (m !== "GET" && m !== "HEAD") { + return Buffer.from(await input.clone().arrayBuffer()); + } + } + return Buffer.alloc(0); +} + +async function storePublicSyncResponseIfOk( + url: string, + method: string, + reqBody: Buffer, + cacheable: boolean, + res: Response +): Promise { + if (!cacheable || res.status !== 200) return res; + const buf = Buffer.from(await res.arrayBuffer()); + cachePublicSyncNetworkResponse({ + method, + url, + body: reqBody, + status: 200, + responseBody: buf, + contentType: res.headers.get("content-type"), + }); + return buildCachedPublicSyncResponse( + url, + buf, + res.headers.get("content-type") + ); +} + export async function kohakuFetch( input: RequestInfo | URL, init?: RequestInit @@ -513,6 +573,30 @@ export async function kohakuFetch( } } + const cacheable = isPublicSyncCacheableRequest(url, method); + const reqBody = cacheable + ? await requestBodyBytesForCache(input, init) + : Buffer.alloc(0); + if (cacheable) { + const cached = readPublicSyncCache({ method, url, body: reqBody }); + if (cached) { + return loggedFetch( + input, + init, + { + via: "clearnet", + clearnetReason: "public-sync-cache", + url, + }, + async () => + buildCachedPublicSyncResponse(url, cached.body, cached.contentType) + ); + } + } + + const wrapStore = (inner: () => Promise) => async () => + storePublicSyncResponseIfOk(url, method, reqBody, cacheable, await inner()); + const decision = shouldUseClearnet(url); if (decision.clearnet) { return loggedFetch( @@ -523,7 +607,7 @@ export async function kohakuFetch( clearnetReason: decision.reason, url, }, - () => clearnetFetch(input, init) + wrapStore(() => clearnetFetch(input, init)) ); } @@ -541,8 +625,9 @@ export async function kohakuFetch( url, requestBytes, }, - async () => + wrapStore(async () => withRequestUrl(await activeSession!.client.fetch(url, torInit), url) + ) ); } @@ -560,8 +645,9 @@ export async function kohakuFetch( url, requestBytes, }, - async () => + wrapStore(async () => withRequestUrl(await activeSession!.client.fetch(url, torInit), url) + ) ), timeoutMs ); @@ -728,8 +814,9 @@ export function runWithWalletTrafficLog( fn: () => Promise ): Promise { ensureKohakuFetchPatch(); - // `/` — keep proving-artifact cache under the same dataDir. + // `/` — keep proving-artifact and public-sync caches under the same dataDir. setArtifactsDataDir(dirname(walletDir)); + setPublicSyncCacheDataDir(dirname(walletDir)); return runWithTrafficLogWallet(walletDir, fn); } diff --git a/src/utils/wallets-util.ts b/src/utils/wallets-util.ts index 46650c4..8a02720 100644 --- a/src/utils/wallets-util.ts +++ b/src/utils/wallets-util.ts @@ -14,17 +14,43 @@ export function parseRequiredWalletName(wallet: string | undefined): string | nu return trimmed ? trimmed : null; } +/** Sibling dirs under `` that must never be used as a wallet name. */ +export const RESERVED_WALLET_DIR_NAMES = [ + "proving-artifacts", + "public-sync-cache", +] as const; + +function isReservedWalletDirName(name: string): boolean { + const lower = name.toLowerCase(); + return (RESERVED_WALLET_DIR_NAMES as readonly string[]).includes(lower); +} + export function walletNameToDirSegment(name: string): string { const trimmed = name.trim(); if (!trimmed) { throw new Error("Wallet name cannot be empty"); } + if (/\s/.test(trimmed)) { + throw new Error( + "Wallet name cannot contain spaces. The first argument is the wallet name, not the seed phrase — pass the mnemonic with --mnemonic or the import prompt." + ); + } + if (isReservedWalletDirName(trimmed)) { + throw new Error( + `Wallet name ${JSON.stringify(trimmed)} is reserved (${RESERVED_WALLET_DIR_NAMES.join(", ")})` + ); + } const safe = trimmed.replace(/[^a-zA-Z0-9._-]+/g, "_").replace(/^_+|_+$/g, ""); if (!safe) { throw new Error( "Wallet name must contain at least one letter, digit, dot, hyphen, or underscore" ); } + if (isReservedWalletDirName(safe)) { + throw new Error( + `Wallet name ${JSON.stringify(trimmed)} is reserved (${RESERVED_WALLET_DIR_NAMES.join(", ")})` + ); + } return safe; } @@ -69,6 +95,8 @@ export function expectedChainIdStringFromWalletDir(walletDir: string): string { export async function resolveWalletPassword(opts: { flagPassword?: string | undefined; nonInteractive?: boolean | undefined; + /** Prompt text when asking interactively (default: "Wallet password:"). */ + promptMessage?: string | undefined; validate?: ((password: string) => void | Promise) | undefined; }): Promise { const fromFlag = opts.flagPassword?.trim(); @@ -104,7 +132,7 @@ export async function resolveWalletPassword(opts: { } for (;;) { const pw = await password({ - message: "Wallet password:", + message: opts.promptMessage?.trim() || "Wallet password:", mask: "*", }); if (pw?.trim()) { diff --git a/tests/first-sync.test.ts b/tests/first-sync.test.ts new file mode 100644 index 0000000..59f75fb --- /dev/null +++ b/tests/first-sync.test.ts @@ -0,0 +1,68 @@ +import assert from "node:assert/strict"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, it } from "node:test"; + +import { + isFirstProtocolSync, + noteProtocolStorageFreshness, + resetFirstSyncObservations, +} from "../src/utils/first-sync.js"; + +const dirs: string[] = []; + +function walletDir(): string { + const dir = mkdtempSync(join(tmpdir(), "kohaku-first-sync-")); + dirs.push(dir); + return dir; +} + +afterEach(() => { + resetFirstSyncObservations(); + for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true, force: true }); +}); + +describe("isFirstProtocolSync", () => { + it("is true when the protocol has no storage file", () => { + const dir = walletDir(); + assert.equal(isFirstProtocolSync(dir, "railgun"), true); + assert.equal(isFirstProtocolSync(dir, "tornado"), true); + assert.equal(isFirstProtocolSync(dir, "privacy-pools"), true); + }); + + it("is false once the protocol has storage", () => { + const dir = walletDir(); + writeFileSync(join(dir, "rg-storage.json"), "{}"); + assert.equal(isFirstProtocolSync(dir, "railgun"), false); + // Storage is per protocol, so the others are untouched. + assert.equal(isFirstProtocolSync(dir, "tornado"), true); + }); + + it("stays true when storage was seeded after the Host was built", () => { + const dir = walletDir(); + // Privacy Pools writes a bundled snapshot while the plugin is constructed, + // i.e. before any sync runs. + noteProtocolStorageFreshness(dir, "ppv1"); + writeFileSync(join(dir, "ppv1-storage.json"), "{}"); + assert.equal(isFirstProtocolSync(dir, "privacy-pools"), true); + }); + + it("keeps the first observation for a wallet", () => { + const dir = walletDir(); + writeFileSync(join(dir, "tc-storage.json"), "{}"); + noteProtocolStorageFreshness(dir, "tc"); + noteProtocolStorageFreshness(dir, "tc"); + assert.equal(isFirstProtocolSync(dir, "tornado"), false); + }); + + it("tracks wallets independently", () => { + const withState = walletDir(); + const fresh = walletDir(); + writeFileSync(join(withState, "rg-storage.json"), "{}"); + noteProtocolStorageFreshness(withState, "rg"); + noteProtocolStorageFreshness(fresh, "rg"); + assert.equal(isFirstProtocolSync(withState, "railgun"), false); + assert.equal(isFirstProtocolSync(fresh, "railgun"), true); + }); +}); diff --git a/tests/progress-terminal.test.ts b/tests/progress-terminal.test.ts new file mode 100644 index 0000000..c11fd81 --- /dev/null +++ b/tests/progress-terminal.test.ts @@ -0,0 +1,95 @@ +import assert from "node:assert/strict"; +import { describe, it } from "node:test"; + +import { manageSpinner, type QuietSpinner } from "../src/utils/cli-quiet.js"; +import { + clearTerminalOwner, + takeTerminal, +} from "../src/utils/progress-terminal.js"; +import { startProgressRenderer } from "../src/utils/sync-progress-renderer.js"; + +type Call = { op: string; msg?: string }; + +function fakeSpinner(): { spin: QuietSpinner; calls: Call[] } { + const calls: Call[] = []; + return { + calls, + spin: { + start: (msg?: string) => calls.push({ op: "start", msg }), + stop: (msg?: string) => calls.push({ op: "stop", msg }), + message: (msg?: string) => calls.push({ op: "message", msg }), + }, + }; +} + +describe("terminal ownership", () => { + it("suspends the active spinner and restores its last message", () => { + const { spin, calls } = fakeSpinner(); + const managed = manageSpinner(spin, false); + managed.start("Loading balances..."); + managed.message?.("Railgun sync · 3 req"); + + const restore = takeTerminal(); + assert.ok(restore, "an active spinner should hand over the line"); + assert.equal(managed.active, false, "spinner must stop drawing"); + assert.deepEqual(calls.at(-1), { op: "stop", msg: "" }); + + restore(); + assert.equal(managed.active, true); + assert.deepEqual( + calls.at(-1), + { op: "start", msg: "Railgun sync · 3 req" }, + "restores the message it was last showing" + ); + + managed.stop("done"); + }); + + it("reports no owner once the spinner has stopped", () => { + const { spin } = fakeSpinner(); + const managed = manageSpinner(spin, false); + managed.start("working"); + managed.stop("done"); + assert.equal(takeTerminal(), null); + }); + + it("never takes the terminal in quiet mode", () => { + const { spin, calls } = fakeSpinner(); + const managed = manageSpinner(spin, true); + managed.start("Loading balances..."); + assert.equal(takeTerminal(), null); + assert.deepEqual(calls, [], "quiet mode must not touch the terminal"); + }); +}); + +describe("worker progress renderer", () => { + it("declines to render when nothing owns the terminal line", () => { + const { spin } = fakeSpinner(); + const managed = manageSpinner(spin, false); + managed.start("held"); + clearTerminalOwner(managed); + + assert.equal(startProgressRenderer("Railgun first sync", Date.now()), null); + managed.stop("done"); + }); + + it("declines to render when explicitly disabled", () => { + const { spin } = fakeSpinner(); + const managed = manageSpinner(spin, false); + managed.start("held"); + + const prev = process.env.KOHAKU_NO_WORKER_PROGRESS; + process.env.KOHAKU_NO_WORKER_PROGRESS = "1"; + try { + assert.equal( + startProgressRenderer("Railgun first sync", Date.now()), + null + ); + assert.equal(managed.active, true, "the spinner keeps the line"); + } finally { + if (prev === undefined) delete process.env.KOHAKU_NO_WORKER_PROGRESS; + else process.env.KOHAKU_NO_WORKER_PROGRESS = prev; + managed.stop("done"); + } + }); +}); diff --git a/tests/proving-artifacts.test.ts b/tests/proving-artifacts.test.ts index 1751124..3116013 100644 --- a/tests/proving-artifacts.test.ts +++ b/tests/proving-artifacts.test.ts @@ -102,4 +102,26 @@ describe("artifactRelativeKeyFromUrl", () => { assert.equal(artifactRelativeKeyFromUrl("https://example.com/foo"), null); assert.equal(artifactRelativeKeyFromUrl("not a url"), null); }); + + it("excludes sync-cache snapshot URLs sharing the artifacts base", () => { + delete process.env.KOHAKU_ARTIFACTS_BASE_URL; + // Otherwise these would inherit the artifact Tor timeout, get duplicated + // into proving-artifacts, and pin later runs to the first snapshot fetched. + for (const key of [ + "sync-cache/v1/manifest.json", + "sync-cache/v1/chunk-000.tar.gz", + "public-sync-cache.tar.gz", + ]) { + assert.equal( + artifactRelativeKeyFromUrl(`${DEFAULT_ARTIFACTS_BASE_URL}/${key}`), + null, + `expected ${key} to be excluded` + ); + assert.equal( + artifactRelativeKeyFromUrl(`${MACWHA}/${key}`), + null, + `expected MacWha ${key} to be excluded` + ); + } + }); }); diff --git a/tests/public-sync-cache.test.ts b/tests/public-sync-cache.test.ts new file mode 100644 index 0000000..a9ddf9f --- /dev/null +++ b/tests/public-sync-cache.test.ts @@ -0,0 +1,600 @@ +import assert from "node:assert/strict"; +import { spawnSync } from "node:child_process"; +import { randomBytes } from "node:crypto"; +import { + existsSync, + mkdtempSync, + readdirSync, + readFileSync, + rmSync, + writeFileSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, it } from "node:test"; +import { gzipSync } from "node:zlib"; + +import { + cachePublicSyncNetworkResponse, + countPublicSyncCacheEntries, + DEFAULT_SYNC_CACHE_BASE_URL, + installPublicSyncCacheArchive, + installPublicSyncCacheChunk, + isPublicSyncCacheableRequest, + packPublicSyncCacheChunks, + parseSyncCacheManifest, + publicSyncCacheHasAllKeys, + publicSyncCacheHasEntries, + readPublicSyncCache, + resolveSyncCacheBaseUrl, + setPublicSyncCacheDataDir, + SYNC_CACHE_MANIFEST_NAME, + writePublicSyncCache, + type SyncCacheManifest, +} from "../src/utils/public-sync-cache.js"; + +const MAINNET_SQUID = + "https://rail-squid.squids.live/squid-railgun-ethereum-v2/v/v1/graphql"; +const SEPOLIA_SQUID = + "https://rail-squid.squids.live/squid-railgun-eth-sepolia-v2/v/v1/graphql"; +const SAGA_URL = "https://saga.gordosoluciones.xyz/events.json"; + +const dirs: string[] = []; +const ORIG_BASE_URL = process.env.KOHAKU_SYNC_CACHE_BASE_URL; +const ORIG_MAX_BYTES = process.env.KOHAKU_PUBLIC_SYNC_CACHE_MAX_BYTES; + +function tmp(): string { + const dir = mkdtempSync(join(tmpdir(), "kohaku-public-sync-")); + dirs.push(dir); + return dir; +} + +function gzipUstar(name: string, content: Buffer): Buffer { + const header = Buffer.alloc(512); + Buffer.from(name).copy(header, 0, 0, Math.min(name.length, 99)); + header.write("0000644\0", 100, 8, "utf8"); + header.write("0000000\0", 108, 8, "utf8"); + header.write("0000000\0", 116, 8, "utf8"); + header.write( + `${content.length.toString(8).padStart(11, "0")}\0`, + 124, + 12, + "utf8" + ); + header.write("00000000000\0", 136, 12, "utf8"); + header.write(" ", 148, 8, "utf8"); + header[156] = 0x30; + header.write("ustar\0", 257, 6, "utf8"); + header.write("00", 263, 2, "utf8"); + let sum = 0; + for (const byte of header) sum += byte; + header.write(`${sum.toString(8).padStart(6, "0")}\0 `, 148, 8, "utf8"); + const pad = + content.length % 512 === 0 ? 0 : 512 - (content.length % 512); + return gzipSync( + Buffer.concat([header, content, Buffer.alloc(pad), Buffer.alloc(1024)]) + ); +} + +afterEach(() => { + while (dirs.length) { + const dir = dirs.pop()!; + rmSync(dir, { recursive: true, force: true }); + } + if (ORIG_BASE_URL === undefined) { + delete process.env.KOHAKU_SYNC_CACHE_BASE_URL; + } else { + process.env.KOHAKU_SYNC_CACHE_BASE_URL = ORIG_BASE_URL; + } + if (ORIG_MAX_BYTES === undefined) { + delete process.env.KOHAKU_PUBLIC_SYNC_CACHE_MAX_BYTES; + } else { + process.env.KOHAKU_PUBLIC_SYNC_CACHE_MAX_BYTES = ORIG_MAX_BYTES; + } +}); + +function readManifest(outputDir: string): SyncCacheManifest { + return parseSyncCacheManifest( + readFileSync(join(outputDir, SYNC_CACHE_MANIFEST_NAME), "utf8") + ); +} + +/** Install every chunk of a packed snapshot into `dataDir`. */ +function installAllChunks( + outputDir: string, + manifest: SyncCacheManifest, + dataDir: string +): number { + let installed = 0; + for (const chunk of manifest.chunks) { + installed += installPublicSyncCacheChunk({ + dataDir, + archivePath: join(outputDir, chunk.name), + chunk, + }).installed; + } + return installed; +} + +describe("isPublicSyncCacheableRequest", () => { + it("allows GET/POST Subsquid and saga URLs only", () => { + assert.equal( + isPublicSyncCacheableRequest( + "https://rail-squid.squids.live/graphql", + "POST" + ), + true + ); + assert.equal( + isPublicSyncCacheableRequest( + "https://saga.gordosoluciones.xyz/events.json", + "GET" + ), + true + ); + assert.equal( + isPublicSyncCacheableRequest("https://api.0xbow.io/pools", "GET"), + false + ); + assert.equal( + isPublicSyncCacheableRequest("https://dw.0xbow.io/state", "POST"), + false + ); + assert.equal( + isPublicSyncCacheableRequest("https://example.com/graphql", "POST"), + false + ); + assert.equal( + isPublicSyncCacheableRequest("https://fastrelay.xyz/relayer", "POST"), + false + ); + assert.equal( + isPublicSyncCacheableRequest( + "https://rail-squid.squids.live/graphql", + "PUT" + ), + false + ); + }); +}); + +describe("public sync HTTP cache", () => { + it("hits after a 200 write and misses unknown keys", () => { + const dataDir = tmp(); + setPublicSyncCacheDataDir(dataDir); + const url = "https://rail-squid.squids.live/graphql"; + const body = '{"query":"{ ping }"}'; + assert.equal( + readPublicSyncCache({ method: "POST", url, body }), + null + ); + writePublicSyncCache({ + method: "POST", + url, + body, + responseBody: Buffer.from('{"data":{"ping":true}}'), + contentType: "application/json", + }); + const hit = readPublicSyncCache({ method: "POST", url, body }); + assert.ok(hit); + assert.equal(hit.body.toString(), '{"data":{"ping":true}}'); + assert.equal(hit.contentType, "application/json"); + assert.equal(publicSyncCacheHasEntries(dataDir), true); + }); + + it("stops writing at the size ceiling without evicting existing entries", () => { + const dataDir = tmp(); + setPublicSyncCacheDataDir(dataDir); + process.env.KOHAKU_PUBLIC_SYNC_CACHE_MAX_BYTES = "4096"; + + const first = '{"page":0}'; + writePublicSyncCache({ + method: "POST", + url: MAINNET_SQUID, + body: first, + responseBody: randomBytes(3072), + contentType: "application/json", + }); + const stored = readPublicSyncCache({ + method: "POST", + url: MAINNET_SQUID, + body: first, + }); + assert.ok(stored, "first entry should fit under the ceiling"); + + // Now full: later pages are skipped rather than evicting the first. + for (let i = 1; i < 4; i++) { + const body = `{"page":${i}}`; + writePublicSyncCache({ + method: "POST", + url: MAINNET_SQUID, + body, + responseBody: randomBytes(3072), + }); + assert.equal( + readPublicSyncCache({ method: "POST", url: MAINNET_SQUID, body }), + null, + `page ${i} should not have been cached` + ); + } + + const survivor = readPublicSyncCache({ + method: "POST", + url: MAINNET_SQUID, + body: first, + }); + assert.ok(survivor, "existing entry must never be evicted"); + assert.equal(survivor.body.byteLength, 3072); + assert.equal(countPublicSyncCacheEntries(dataDir), 1); + }); + + it("refreshes an existing key even when the cache is full", () => { + const dataDir = tmp(); + setPublicSyncCacheDataDir(dataDir); + process.env.KOHAKU_PUBLIC_SYNC_CACHE_MAX_BYTES = "4096"; + const body = '{"page":0}'; + writePublicSyncCache({ + method: "POST", + url: MAINNET_SQUID, + body, + responseBody: Buffer.alloc(3072, 1), + }); + // Same key, so its bytes are replaced rather than added to the total. + writePublicSyncCache({ + method: "POST", + url: MAINNET_SQUID, + body, + responseBody: Buffer.alloc(3072, 2), + }); + const hit = readPublicSyncCache({ + method: "POST", + url: MAINNET_SQUID, + body, + }); + assert.ok(hit); + assert.equal(hit.body[0], 2); + assert.equal(countPublicSyncCacheEntries(dataDir), 1); + }); + + it("does not cache non-200 responses", () => { + const dataDir = tmp(); + setPublicSyncCacheDataDir(dataDir); + const url = "https://saga.gordosoluciones.xyz/state.json"; + cachePublicSyncNetworkResponse({ + method: "GET", + url, + status: 500, + responseBody: Buffer.from("nope"), + contentType: "text/plain", + }); + assert.equal(readPublicSyncCache({ method: "GET", url }), null); + cachePublicSyncNetworkResponse({ + method: "GET", + url, + status: 200, + responseBody: Buffer.from("ok"), + contentType: "text/plain", + }); + const hit = readPublicSyncCache({ method: "GET", url }); + assert.ok(hit); + assert.equal(hit.body.toString(), "ok"); + }); +}); + +describe("resolveSyncCacheBaseUrl", () => { + it("uses the default base unless KOHAKU_SYNC_CACHE_BASE_URL is set", () => { + delete process.env.KOHAKU_SYNC_CACHE_BASE_URL; + assert.equal(resolveSyncCacheBaseUrl(), DEFAULT_SYNC_CACHE_BASE_URL); + process.env.KOHAKU_SYNC_CACHE_BASE_URL = "https://example.test/snap/v2/"; + assert.equal(resolveSyncCacheBaseUrl(), "https://example.test/snap/v2"); + }); +}); + +describe("pack/install public-sync chunks", () => { + function seedThreeEntries(dataDir: string): void { + setPublicSyncCacheDataDir(dataDir); + writePublicSyncCache({ + method: "POST", + url: MAINNET_SQUID, + body: '{"query":"{ commitments }"}', + responseBody: Buffer.from('{"data":{"commitments":[1]}}'), + contentType: "application/json", + }); + writePublicSyncCache({ + method: "POST", + url: SEPOLIA_SQUID, + body: "{}", + responseBody: Buffer.from("sepolia-page"), + }); + writePublicSyncCache({ + method: "GET", + url: SAGA_URL, + responseBody: Buffer.from('{"chunks":[]}'), + contentType: "application/json", + }); + } + + it("round-trips mixed-network cache entries through chunks", () => { + const src = tmp(); + const dest = tmp(); + seedThreeEntries(src); + + const outputDir = tmp(); + const { manifest } = packPublicSyncCacheChunks({ + dataDir: src, + outputDir, + }); + assert.equal(manifest.version, 1); + assert.equal(manifest.entries, 3); + assert.ok(manifest.rawBytes > 0); + assert.equal(existsSync(join(outputDir, SYNC_CACHE_MANIFEST_NAME)), true); + for (const chunk of manifest.chunks) { + assert.equal(existsSync(join(outputDir, chunk.name)), true); + assert.match(chunk.name, /^chunk-\d{3}\.tar\.gz$/); + assert.match(chunk.sha256, /^[a-f0-9]{64}$/); + } + const allKeys = manifest.chunks.flatMap((c) => c.keys); + assert.equal(allKeys.length, 3); + assert.equal(new Set(allKeys).size, 3); + assert.deepEqual(readManifest(outputDir), manifest); + + assert.equal(installAllChunks(outputDir, manifest, dest), 3); + setPublicSyncCacheDataDir(dest); + const mainnet = readPublicSyncCache({ + method: "POST", + url: MAINNET_SQUID, + body: '{"query":"{ commitments }"}', + }); + assert.ok(mainnet); + assert.equal(mainnet.body.toString(), '{"data":{"commitments":[1]}}'); + assert.equal(mainnet.contentType, "application/json"); + const sepolia = readPublicSyncCache({ + method: "POST", + url: SEPOLIA_SQUID, + body: "{}", + }); + assert.ok(sepolia); + assert.equal(sepolia.body.toString(), "sepolia-page"); + const saga = readPublicSyncCache({ method: "GET", url: SAGA_URL }); + assert.ok(saga); + assert.equal(saga.body.toString(), '{"chunks":[]}'); + }); + + it("splits into multiple chunks and isolates oversized entries", () => { + const src = tmp(); + setPublicSyncCacheDataDir(src); + // Incompressible bodies, so each entry alone exceeds the 2 KiB target. + for (let i = 0; i < 4; i++) { + writePublicSyncCache({ + method: "POST", + url: MAINNET_SQUID, + body: `{"page":${i}}`, + responseBody: randomBytes(4096), + }); + } + const outputDir = tmp(); + const { manifest } = packPublicSyncCacheChunks({ + dataDir: src, + outputDir, + chunkBytes: 2048, + }); + assert.equal(manifest.entries, 4); + assert.equal(manifest.chunks.length, 4); + for (const chunk of manifest.chunks) { + assert.equal(chunk.keys.length, 1); + } + + const dest = tmp(); + assert.equal(installAllChunks(outputDir, manifest, dest), 4); + }); + + it("packs small entries together into one chunk", () => { + const src = tmp(); + seedThreeEntries(src); + const outputDir = tmp(); + const { manifest } = packPublicSyncCacheChunks({ + dataDir: src, + outputDir, + }); + assert.equal(manifest.chunks.length, 1); + assert.equal(manifest.chunks[0]!.keys.length, 3); + }); + + it("rejects a chunk whose sha256 or size does not match the manifest", () => { + const src = tmp(); + seedThreeEntries(src); + const outputDir = tmp(); + const { manifest } = packPublicSyncCacheChunks({ + dataDir: src, + outputDir, + }); + const chunk = manifest.chunks[0]!; + const archivePath = join(outputDir, chunk.name); + + assert.throws( + () => + installPublicSyncCacheChunk({ + dataDir: tmp(), + archivePath, + chunk: { ...chunk, sha256: "b".repeat(64) }, + }), + /sha256 mismatch/ + ); + assert.throws( + () => + installPublicSyncCacheChunk({ + dataDir: tmp(), + archivePath, + chunk: { ...chunk, bytes: chunk.bytes + 1 }, + }), + /size mismatch/ + ); + }); + + it("reports whether all keys of a chunk are already cached", () => { + const src = tmp(); + const dest = tmp(); + seedThreeEntries(src); + const outputDir = tmp(); + const { manifest } = packPublicSyncCacheChunks({ + dataDir: src, + outputDir, + }); + const keys = manifest.chunks.flatMap((c) => c.keys); + assert.equal(publicSyncCacheHasAllKeys(keys, dest), false); + installAllChunks(outputDir, manifest, dest); + assert.equal(publicSyncCacheHasAllKeys(keys, dest), true); + assert.equal(publicSyncCacheHasAllKeys([...keys, "c".repeat(64)], dest), false); + }); + + it("installs only 64-hex cache files and ignores junk", () => { + const dest = tmp(); + const staging = tmp(); + const key = "a".repeat(64); + writeFileSync(join(staging, `${key}.bin`), "body"); + writeFileSync( + join(staging, `${key}.json`), + JSON.stringify({ + url: MAINNET_SQUID, + method: "POST", + contentType: "application/json", + storedAt: new Date().toISOString(), + bytes: 4, + }) + ); + writeFileSync(join(staging, "README.txt"), "not a cache entry"); + writeFileSync(join(staging, "notes.bin"), "nope"); + const archivePath = join(tmp(), "mixed.tar.gz"); + const tar = spawnSync("tar", ["-czf", archivePath, "-C", staging, "."], { + encoding: "utf8", + }); + assert.equal(tar.status, 0, tar.stderr); + + const installed = installPublicSyncCacheArchive({ + dataDir: dest, + archivePath, + }); + assert.equal(installed.installed, 1); + const names = readdirSync(installed.cacheDir).sort(); + assert.deepEqual(names, [`${key}.bin`, `${key}.json`]); + assert.equal(existsSync(join(installed.cacheDir, "README.txt")), false); + }); + + it("refuses archive members with .. or absolute paths", () => { + const dest = tmp(); + const archivePath = join(tmp(), "unsafe.tar.gz"); + writeFileSync( + archivePath, + gzipUstar("../" + "b".repeat(64) + ".bin", Buffer.from("x")) + ); + assert.throws( + () => installPublicSyncCacheArchive({ dataDir: dest, archivePath }), + /unsafe path/ + ); + }); + + it("throws when packing an empty cache", () => { + const dataDir = tmp(); + assert.throws( + () => + packPublicSyncCacheChunks({ + dataDir, + outputDir: join(dataDir, "out"), + }), + /No public-sync-cache entries/ + ); + }); +}); + +describe("parseSyncCacheManifest", () => { + const chunk = { + name: "chunk-000.tar.gz", + bytes: 100, + sha256: "a".repeat(64), + keys: ["b".repeat(64)], + }; + + function manifest(overrides: Record = {}): string { + return JSON.stringify({ version: 1, chunks: [chunk], ...overrides }); + } + + it("accepts a well-formed manifest and defaults soft fields", () => { + const parsed = parseSyncCacheManifest(manifest()); + assert.equal(parsed.version, 1); + assert.equal(parsed.entries, 1); + assert.equal(parsed.rawBytes, 0); + assert.deepEqual(parsed.chunks, [chunk]); + }); + + it("rejects malformed JSON and non-objects", () => { + assert.throws(() => parseSyncCacheManifest("{nope"), /not valid JSON/); + assert.throws(() => parseSyncCacheManifest("null"), /must be a JSON object/); + }); + + it("rejects a manifest with no chunks", () => { + assert.throws( + () => parseSyncCacheManifest(manifest({ chunks: [] })), + /no chunks/ + ); + assert.throws( + () => parseSyncCacheManifest(manifest({ chunks: "nope" })), + /no chunks/ + ); + }); + + it("rejects an unsupported version", () => { + assert.throws( + () => parseSyncCacheManifest(manifest({ version: 2 })), + /Unsupported sync-cache manifest version/ + ); + }); + + it("rejects chunk names that could escape the output directory", () => { + for (const name of [ + "../chunk-000.tar.gz", + "/abs/chunk-000.tar.gz", + "nested/chunk-000.tar.gz", + "chunk-000.tar.gz.sh", + "chunk-0.tar.gz", + ]) { + assert.throws( + () => parseSyncCacheManifest(manifest({ chunks: [{ ...chunk, name }] })), + /unsafe chunk name/, + `expected ${name} to be rejected` + ); + } + }); + + it("rejects missing or malformed integrity fields", () => { + assert.throws( + () => + parseSyncCacheManifest( + manifest({ chunks: [{ ...chunk, sha256: "nope" }] }) + ), + /no valid sha256/ + ); + assert.throws( + () => + parseSyncCacheManifest(manifest({ chunks: [{ ...chunk, bytes: 0 }] })), + /no valid byte size/ + ); + assert.throws( + () => parseSyncCacheManifest(manifest({ chunks: [{ ...chunk, keys: [] }] })), + /lists no cache keys/ + ); + assert.throws( + () => + parseSyncCacheManifest( + manifest({ chunks: [{ ...chunk, keys: ["../etc/passwd"] }] }) + ), + /invalid cache key/ + ); + }); + + it("rejects duplicate chunk names", () => { + assert.throws( + () => parseSyncCacheManifest(manifest({ chunks: [chunk, chunk] })), + /Duplicate chunk/ + ); + }); +}); diff --git a/tests/pure-helpers.test.ts b/tests/pure-helpers.test.ts index bc1b247..784cb2c 100644 --- a/tests/pure-helpers.test.ts +++ b/tests/pure-helpers.test.ts @@ -3,7 +3,6 @@ import { describe, it } from "node:test"; import { formatUsdCents } from "../src/lib/usd-values.js"; import { jsonStringifyWithBigInt } from "../src/utils/json-bigint.js"; -import { estimatedGetLogsWindowCount } from "../src/host/chunked-get-logs.js"; import { formatLegacyTornadoNote } from "../src/commands/export-tornado-note.js"; import { normalizeTornadoNoteInput } from "../src/commands/import-tornado-note.js"; @@ -23,15 +22,6 @@ describe("jsonStringifyWithBigInt", () => { }); }); -describe("estimatedGetLogsWindowCount", () => { - it("counts inclusive windows of at most chunkSpan blocks", () => { - assert.equal(estimatedGetLogsWindowCount(0n, 498n, 499n), 1); - assert.equal(estimatedGetLogsWindowCount(0n, 499n, 499n), 2); - assert.equal(estimatedGetLogsWindowCount(100n, 99n, 499n), 0); - assert.equal(estimatedGetLogsWindowCount(0n, 10n, 0n), 0); - }); -}); - describe("normalizeTornadoNoteInput", () => { it("accepts classic notes and prefixes the short form", () => { const classic = diff --git a/tests/stealth-scan-window.test.ts b/tests/stealth-scan-window.test.ts new file mode 100644 index 0000000..de5d7fc --- /dev/null +++ b/tests/stealth-scan-window.test.ts @@ -0,0 +1,157 @@ +import assert from "node:assert/strict"; +import { describe, it } from "node:test"; + +import { + defaultStealthImportStartBlock, + stealthAnnouncerStartBlock, +} from "../src/lib/stealth/constants.js"; +import { + formatStealthScanStartLog, + resolveStealthScanFloor, + resolveStealthScanWindow, +} from "../src/lib/stealth/scan.js"; + +const MAINNET = 1n; +const SEPOLIA = 11155111n; +const LATEST = 30_000_000n; + +describe("defaultStealthImportStartBlock", () => { + it("uses the Kohaku-schema floors", () => { + assert.equal(defaultStealthImportStartBlock(MAINNET), 25_700_000n); + assert.equal(defaultStealthImportStartBlock(SEPOLIA), 11_455_454n); + }); +}); + +describe("resolveStealthScanFloor", () => { + it("uses the Kohaku import default when start is omitted, not announcer deploy", () => { + assert.equal(resolveStealthScanFloor({ chainId: MAINNET }), 25_700_000n); + assert.equal(resolveStealthScanFloor({ chainId: SEPOLIA }), 11_455_454n); + assert.ok( + resolveStealthScanFloor({ chainId: MAINNET }) > + stealthAnnouncerStartBlock(MAINNET) + ); + assert.ok( + resolveStealthScanFloor({ chainId: SEPOLIA }) > + stealthAnnouncerStartBlock(SEPOLIA) + ); + }); + + it("honors an explicit block below the default but at/above deploy", () => { + const deploy = stealthAnnouncerStartBlock(MAINNET); + assert.equal( + resolveStealthScanFloor({ chainId: MAINNET, startFromBlock: deploy }), + deploy + ); + assert.equal( + resolveStealthScanFloor({ + chainId: MAINNET, + startFromBlock: deploy + 1n, + }), + deploy + 1n + ); + }); + + it("clamps explicit 0 to the announcer deploy block", () => { + assert.equal( + resolveStealthScanFloor({ chainId: MAINNET, startFromBlock: 0n }), + stealthAnnouncerStartBlock(MAINNET) + ); + assert.equal( + resolveStealthScanFloor({ chainId: SEPOLIA, startFromBlock: 0n }), + stealthAnnouncerStartBlock(SEPOLIA) + ); + }); +}); + +describe("resolveStealthScanWindow", () => { + it("starts a first pass at the Kohaku default", () => { + const w = resolveStealthScanWindow({ + chainId: MAINNET, + latest: LATEST, + }); + assert.equal(w.fromBlock, 25_700_000n); + assert.equal(w.startFloor, 25_700_000n); + assert.equal(w.needsFullHistory, true); + }); + + it("resumes a mid-pass forward past an earlier floor", () => { + const w = resolveStealthScanWindow({ + chainId: MAINNET, + latest: LATEST, + startFromBlock: 25_700_000n, + lastScannedBlock: "25800000", + fullHistoryScanned: false, + }); + assert.equal(w.fromBlock, 25_800_001n); + }); + + it("resumes incrementally from lastScannedBlock+1 without backdate", () => { + const w = resolveStealthScanWindow({ + chainId: MAINNET, + latest: LATEST, + startFromBlock: 25_700_000n, + lastScannedBlock: "28000000", + fullHistoryScanned: true, + }); + assert.equal(w.fromBlock, 28_000_001n); + assert.equal(w.needsFullHistory, false); + }); + + it("does not treat the wallet-file floor as a back-date", () => { + const w = resolveStealthScanWindow({ + chainId: MAINNET, + latest: LATEST, + startFromBlock: 25_700_000n, + lastScannedBlock: "28000000", + fullHistoryScanned: true, + backdate: false, + }); + assert.equal(w.fromBlock, 28_000_001n); + }); + + it("back-dates from the CLI flag below lastScannedBlock", () => { + const deploy = stealthAnnouncerStartBlock(MAINNET); + const w = resolveStealthScanWindow({ + chainId: MAINNET, + latest: LATEST, + startFromBlock: deploy, + lastScannedBlock: "28000000", + fullHistoryScanned: true, + backdate: true, + }); + assert.equal(w.fromBlock, deploy); + }); + + it("does not back-date when the flag is at or above the resume cursor", () => { + const w = resolveStealthScanWindow({ + chainId: MAINNET, + latest: LATEST, + startFromBlock: 29_000_000n, + lastScannedBlock: "28000000", + fullHistoryScanned: false, + backdate: true, + }); + assert.equal(w.fromBlock, 29_000_000n); + }); + + it("ignores a higher flag on incremental runs", () => { + const w = resolveStealthScanWindow({ + chainId: MAINNET, + latest: LATEST, + startFromBlock: 29_000_000n, + lastScannedBlock: "28000000", + fullHistoryScanned: true, + backdate: true, + }); + assert.equal(w.fromBlock, 28_000_001n); + }); +}); + +describe("formatStealthScanStartLog", () => { + it("prints start block and latest minus start", () => { + assert.equal( + formatStealthScanStartLog(25_700_000n, 25_800_000n), + "Stealth scan from block 25700000 · 100000 blocks" + ); + }); +}); diff --git a/tests/sync-progress.test.ts b/tests/sync-progress.test.ts new file mode 100644 index 0000000..976ec7d --- /dev/null +++ b/tests/sync-progress.test.ts @@ -0,0 +1,181 @@ +import assert from "node:assert/strict"; +import { describe, it } from "node:test"; + +import { + countSyncRequest, + noteSyncFirstRun, + reportSyncPhase, + runWithSyncProgress, +} from "../src/utils/sync-progress.js"; + +/** Progress emits are coalesced, so settle before inspecting messages. */ +const settle = () => new Promise((r) => setTimeout(r, 300)); + +async function capture( + opts: Parameters[0], + body: () => void +): Promise { + const messages: string[] = []; + await runWithSyncProgress({ ...opts, onUpdate: (m) => messages.push(m) }, async () => { + body(); + await settle(); + }); + return messages; +} + +describe("sync progress phases", () => { + it("reports the most recent phase, without latching on an earlier one", async () => { + const messages = await capture({ source: "railgun" }, () => { + for (let i = 0; i < 3; i++) countSyncRequest("rpc"); + countSyncRequest("subsquid"); + }); + const last = messages.at(-1)!; + assert.match(last, /Subsquid over Tor/, messages.join("\n")); + assert.doesNotMatch(last, /RPC logs/); + }); + + it("keeps a separate count per phase and never resurrects another phase's", async () => { + const messages = await capture({ source: "privacy-pools" }, () => { + for (let i = 0; i < 7; i++) countSyncRequest("rpc"); + for (let i = 0; i < 2; i++) countSyncRequest("asp"); + }); + assert.match(messages.at(-1)!, /ASP · 2 req/, messages.join("\n")); + + const backToRpc = await capture({ source: "privacy-pools" }, () => { + countSyncRequest("rpc"); + countSyncRequest("saga"); + countSyncRequest("rpc"); + }); + assert.match(backToRpc.at(-1)!, /RPC logs · 2 req/, backToRpc.join("\n")); + }); + + it("counts cumulatively across separate getLogs ranges", async () => { + const messages = await capture({ source: "privacy-pools" }, () => { + for (let i = 0; i < 11; i++) countSyncRequest("rpc"); + for (let i = 0; i < 11; i++) countSyncRequest("rpc"); + }); + assert.match(messages.at(-1)!, /RPC logs · 22 req/, messages.join("\n")); + }); + + it("switches phase without a count via reportSyncPhase", async () => { + const messages = await capture({ source: "tornado" }, () => { + reportSyncPhase("saga"); + }); + const last = messages.at(-1)!; + assert.match(last, /saga CDN/, messages.join("\n")); + assert.doesNotMatch(last, /req/); + }); + + it("reports 'starting' before any phase is known", async () => { + const messages = await capture({ source: "railgun" }, () => {}); + assert.match(messages[0]!, /^Railgun sync · starting · \d+s$/, messages.join("\n")); + }); +}); + +describe("sync progress labels", () => { + it("marks a first run and leaves incremental runs unmarked", async () => { + const first = await capture({ source: "railgun", firstRun: true }, () => { + countSyncRequest("subsquid"); + }); + assert.match(first.at(-1)!, /^Railgun first sync · /, first.join("\n")); + + const incremental = await capture({ source: "railgun" }, () => { + countSyncRequest("subsquid"); + }); + assert.match(incremental.at(-1)!, /^Railgun sync · /, incremental.join("\n")); + }); + + it("calls the stealth source a scan", async () => { + const first = await capture({ source: "stealth", firstRun: true }, () => { + countSyncRequest("rpc"); + }); + assert.match(first.at(-1)!, /^Stealth first scan · /, first.join("\n")); + + const incremental = await capture({ source: "stealth" }, () => { + countSyncRequest("rpc"); + }); + assert.match(incremental.at(-1)!, /^Stealth scan · /, incremental.join("\n")); + }); + + it("lets an inner scope refine the first-run label", async () => { + const messages = await capture({ source: "stealth" }, () => { + noteSyncFirstRun(true); + countSyncRequest("rpc"); + }); + assert.match(messages.at(-1)!, /^Stealth first scan · /, messages.join("\n")); + }); + + it("names each protocol", async () => { + for (const [source, label] of [ + ["railgun", "Railgun"], + ["privacy-pools", "Privacy Pools"], + ["tornado", "Tornado Cash"], + ] as const) { + const messages = await capture({ source }, () => countSyncRequest("rpc")); + assert.match(messages.at(-1)!, new RegExp(`^${label} sync · `)); + } + }); +}); + +describe("sync progress emission", () => { + it("does not emit a prelude through onUpdate", async () => { + const messages = await capture( + { + source: "stealth", + prelude: "Stealth scan from block 25700000 · 100000 blocks", + }, + () => {} + ); + assert.ok( + messages.every((m) => !m.includes("Stealth scan from block")), + messages.join("\n") + ); + }); + + it("emits nothing when no onUpdate is provided", async () => { + let emitted = 0; + await runWithSyncProgress({ source: "railgun" }, async () => { + countSyncRequest("rpc"); + await settle(); + }); + // Nested scopes reuse the outer store, so a nested onUpdate is ignored too. + await runWithSyncProgress({ source: "railgun" }, async () => { + await runWithSyncProgress( + { source: "tornado", onUpdate: () => (emitted += 1) }, + async () => { + countSyncRequest("rpc"); + await settle(); + } + ); + }); + assert.equal(emitted, 0); + }); + + it("formats elapsed time past a minute as m ss", async () => { + const messages: string[] = []; + const realNow = Date.now; + let now = realNow(); + Date.now = () => now; + try { + await runWithSyncProgress( + { source: "railgun", onUpdate: (m) => messages.push(m) }, + async () => { + now += 59_000; + countSyncRequest("rpc"); + now += 1_000; + countSyncRequest("rpc"); + } + ); + } finally { + Date.now = realNow; + } + assert.ok( + messages.some((m) => m.endsWith("59s")), + messages.join("\n") + ); + assert.ok( + messages.some((m) => m.includes("1m 00s")), + messages.join("\n") + ); + }); +}); diff --git a/tests/wallets-util-path.test.ts b/tests/wallets-util-path.test.ts index f3ba8b6..ac32e20 100644 --- a/tests/wallets-util-path.test.ts +++ b/tests/wallets-util-path.test.ts @@ -18,11 +18,39 @@ describe("parseRequiredWalletName", () => { describe("walletNameToDirSegment", () => { it("replaces unsafe characters without allowing path traversal", () => { assert.equal(walletNameToDirSegment("alice"), "alice"); - assert.equal(walletNameToDirSegment("My Wallet"), "My_Wallet"); assert.equal(walletNameToDirSegment("../etc/passwd"), ".._etc_passwd"); assert.equal(walletNameToDirSegment("/tmp/foo"), "tmp_foo"); }); + it("rejects names with spaces", () => { + assert.throws( + () => walletNameToDirSegment("My Wallet"), + /cannot contain spaces/ + ); + assert.throws( + () => + walletNameToDirSegment( + "abandon ability able about above absent absorb abstract" + ), + /cannot contain spaces/ + ); + }); + + it("rejects reserved data-dir names", () => { + assert.throws( + () => walletNameToDirSegment("proving-artifacts"), + /reserved/ + ); + assert.throws( + () => walletNameToDirSegment("Public-Sync-Cache"), + /reserved/ + ); + assert.throws( + () => walletNameToDirSegment("proving-artifacts/"), + /reserved/ + ); + }); + it("rejects empty or punctuation-only names", () => { assert.throws(() => walletNameToDirSegment(""), /cannot be empty/); assert.throws( @@ -38,8 +66,7 @@ describe("walletNameToDirSegment", () => { describe("resolveWalletDir", () => { it("joins the sanitized segment onto the data dir", () => { - const dir = resolveWalletDir("/data", "My Wallet"); - assert.ok(dir.endsWith("My_Wallet")); - assert.equal(dir.includes("My Wallet"), false); + const dir = resolveWalletDir("/data", "alice"); + assert.ok(dir.endsWith("alice")); }); });