diff --git a/src/lib/bns-index-core.js b/src/lib/bns-index-core.js index 16be928..fbe7530 100644 --- a/src/lib/bns-index-core.js +++ b/src/lib/bns-index-core.js @@ -108,8 +108,10 @@ function mergeServers(preferred, rest) { // resolver the resolver-web.js module (already imported) // WebSocket WebSocket class for electrum (swap with setWebSocket, e.g. Tor) // snapshotDir where the two local copies are written (null: never write) -// warmPaths extra read-only snapshots to warm from, in order, after -// the local copies (Ariadne's file for Theseus, a bundled one) +// snapshotBase their file name without .json (default bns-name-snapshot) +// warmPaths extra read-only snapshots to warm from (Ariadne's copies +// for Theseus, a bundled one); with the local copies, the +// one holding the most evidence wins // electrumCacheFile discovered electrum pool (must be as protected as the // snapshot: a writable pool file chooses the evidence) // publishedUrls override for NETWORK.nameListUrl / nameListMirrorUrl @@ -124,7 +126,8 @@ export function createIndexer(opts) { let WS = opts.WebSocket; const network = R.NETWORK?.id || "chipnet"; const beacon = R.BEACON_SCRIPTHASH; - const paths = opts.snapshotDir ? snapshotPaths(opts.snapshotDir) : null; + const base = opts.snapshotBase || "bns-name-snapshot"; + const paths = opts.snapshotDir ? snapshotPaths(opts.snapshotDir, base) : null; const publishedUrls = (opts.publishedUrls || [R.NETWORK?.nameListUrl, R.NETWORK?.nameListMirrorUrl]).filter(Boolean); const st = { @@ -175,7 +178,7 @@ export function createIndexer(opts) { st.index = idx; st.tlds = typeof R.getBcnrTlds === "function" ? R.getBcnrTlds(idx) : []; if (snap && paths && reason !== "warm") { - try { writeSnapshotCopies(opts.snapshotDir, { ...snap, savedAt: new Date().toISOString() }); } + try { writeSnapshotCopies(opts.snapshotDir, { ...snap, savedAt: new Date().toISOString() }, base); } catch (e) { log("snapshot write failed:", e.message); } } if (!changed.length && reason !== "warm") { ev.emit("fresh", { gen: st.gen, builtAt: st.builtAt }); return false; } @@ -189,9 +192,14 @@ export function createIndexer(opts) { const candidates = []; if (paths) candidates.push(paths.current, paths.previous); for (const p of opts.warmPaths || []) if (p) candidates.push(p); + // Every readable candidate, richest first (the earlier path on a tie). + const found = []; for (const p of candidates) { const snap = readSnapshotFile(p); - if (!snap || (beacon && snap.beacon && snap.beacon !== beacon)) continue; + if (snap && !(beacon && snap.beacon && snap.beacon !== beacon)) found.push({ p, snap }); + } + found.sort((a, b) => b.snap.history.length - a.snap.history.length); + for (const { p, snap } of found) { try { const t0 = Date.now(); const idx = R.buildIndexFromSnapshot({ snapshot: snap }); diff --git a/src/lib/bns-source-chain.js b/src/lib/bns-source-chain.js index 9d2544f..da3b377 100644 --- a/src/lib/bns-source-chain.js +++ b/src/lib/bns-source-chain.js @@ -16,7 +16,28 @@ import fs from "node:fs"; import path from "node:path"; import { EventEmitter } from "node:events"; -import { readSnapshotFile, diffIndex } from "./bns-index-core.js"; + +// No imports between the shared modules: Theseus ships each as a single .mjs +// in its resources, where a relative import would not resolve. These two +// are the same as in bns-index-core.js. +function readSnapshotFile(p) { + try { + if (!p || !fs.existsSync(p)) return null; + const parsed = JSON.parse(fs.readFileSync(p, "utf8").replace(/^\uFEFF/, "")); + if (!parsed || Array.isArray(parsed) || !Array.isArray(parsed.history) || typeof parsed.txs !== "object") return null; + return parsed; + } catch { return null; } +} +function diffIndex(prev, next) { + const changed = []; + if (!prev) return [...next.keys()]; + for (const [k, e] of next) { + const o = prev.get(k); + if (!o || JSON.stringify(o) !== JSON.stringify(e)) changed.push(k); + } + for (const k of prev.keys()) if (!next.has(k)) changed.push(k); + return changed; +} const PROVISIONAL_TTL_MS = 30_000; const NEGATIVE_TTL_MS = 30_000; @@ -35,14 +56,16 @@ function readIndexFile(p) { // opts: // resolver resolver-web.js module // connectPipe async () => client (bns-pipe.js connectIndexerPipe), or null -// localPaths the indexer's copies (current, previous), in order +// localPaths raw snapshots to read: the indexer's copies (current, +// previous), then any others; the one with the most +// evidence wins // statePath the indexer's small state file ({builtAt, ...}), optional // ownCopyPath this reader's last-known-good copy (index format) -// publicSnapshotUrls, publicIndexerUrl +// publicSnapshotUrls, publicIndexerUrl [] / null switch a source off // staleMs local data older than this counts as stale (default 5 min) // log // Events: "change" {changed, source}, "mismatch" {name, provisional, verified}, -// "pipe" {connected, error?} +// "pipe" {connected, error?}, "fresh" {builtAt}, "paused" boolean export function createSourceChain(opts) { const R = opts.resolver; const log = opts.log || (() => {}); @@ -50,12 +73,12 @@ export function createSourceChain(opts) { const staleMs = opts.staleMs ?? 5 * 60_000; const publicRefreshMs = opts.publicRefreshMs ?? 60 * 60_000; const publicSnapshotUrls = (opts.publicSnapshotUrls || [R.NETWORK?.nameListUrl, R.NETWORK?.nameListMirrorUrl]).filter(Boolean); - const publicIndexerUrl = opts.publicIndexerUrl ?? "https://navigate.st"; + const publicIndexerUrl = opts.publicIndexerUrl === undefined ? "https://navigate.st" : opts.publicIndexerUrl; const st = { index: new Map(), source: null, builtAt: 0, gen: 0, tlds: [], historyLen: 0, fileMtimes: {}, pipe: null, pipeError: null, pipeSince: 0, lastPush: 0, - lastPublic: 0, lastOwnWrite: 0, + lastPublic: 0, lastOwnWrite: 0, paused: false, }; const provisional = new Map(); // key -> { at, entry|null } @@ -90,18 +113,23 @@ export function createSourceChain(opts) { try { return opts.statePath ? JSON.parse(fs.readFileSync(opts.statePath, "utf8")) : null; } catch { return null; } } function loadLocal(reason = "start") { + // The copy with the most evidence; the earlier path on a tie. + let best = null; for (const p of opts.localPaths || []) { const snap = readSnapshotFile(p); - if (!snap) continue; + if (!snap || (R.BEACON_SCRIPTHASH && snap.beacon && snap.beacon !== R.BEACON_SCRIPTHASH)) continue; + try { st.fileMtimes[p] = fs.statSync(p).mtimeMs; } catch {} + if (!best || snap.history.length > best.snap.history.length) best = { p, snap }; + } + if (best) { + const { p, snap } = best; try { - const mtime = fs.statSync(p).mtimeMs; - st.fileMtimes[p] = mtime; if (snap.history.length < st.historyLen && st.source !== null && reason !== "start") return false; const idx = R.buildIndexFromSnapshot({ snapshot: snap }); const state = readState(); - const saved = Date.parse(snap.savedAt || "") || mtime; + const saved = Date.parse(snap.savedAt || "") || st.fileMtimes[p] || 0; const builtAt = Math.max(saved, Number(state?.builtAt) || 0); - adopt(idx, `local:${path.basename(p)}`, { builtAt, historyLen: snap.history.length, gen: Number(state?.gen) || st.gen }); + adopt(idx, `local:${p}`, { builtAt, historyLen: snap.history.length, gen: Number(state?.gen) || st.gen }); return true; } catch (e) { log(`source chain: ${p} unusable: ${e.message}`); } } @@ -143,6 +171,7 @@ export function createSourceChain(opts) { if (hello.network && R.NETWORK?.id && hello.network !== R.NETWORK.id) throw new Error(`indexer is on ${hello.network}`); const snap = await c.request("snapshot", {}, { timeoutMs: 60_000 }); st.pipe = c; st.pipeError = null; st.pipeSince = st.lastPush = Date.now(); backoff = 5_000; + setPaused(!!hello.paused); adopt(new Map(snap.names), "pipe", { builtAt: snap.builtAt, gen: snap.gen, tlds: snap.tlds }); writeOwnCopy(); c.on("push", onPush); @@ -173,7 +202,8 @@ export function createSourceChain(opts) { } async function onPush(m) { st.lastPush = Date.now(); - if (m.event === "fresh") { st.builtAt = m.builtAt || st.builtAt; return; } + if (m.event === "fresh") { st.builtAt = m.builtAt || st.builtAt; ev.emit("fresh", { builtAt: st.builtAt }); return; } + if (m.event === "paused") { setPaused(!!m.paused); return; } if (m.event !== "change") return; if (m.gen <= st.gen) return; if (m.entries && m.gen === st.gen + 1) { @@ -190,6 +220,18 @@ export function createSourceChain(opts) { writeOwnCopy(); } + function setPaused(on) { + if (on === st.paused) return; + st.paused = on; + ev.emit("paused", on); + } + // One delta poll in the live indexer, on demand (it refuses while paused). + async function requestPoll() { + const c = st.pipe; + if (!c) return false; + try { const r = await c.request("poll", {}, { timeoutMs: 50_000 }); return !!r.polled; } catch { return false; } + } + // ---- 4: public snapshots ---- async function refreshPublic() { st.lastPublic = Date.now(); @@ -247,7 +289,7 @@ export function createSourceChain(opts) { async function resolve(key, { verifiedOnly = false } = {}) { const hit = lookup(key); if (hit && (!hit.provisional || !verifiedOnly)) return hit; - if (verifiedOnly || !stale()) return null; + if (verifiedOnly || !stale() || !publicIndexerUrl) return null; const entry = await fetchProvisional(key); return entry ? { entry, provisional: true, source: "public-indexer" } : null; } @@ -258,18 +300,18 @@ export function createSourceChain(opts) { connect(); tick = setInterval(() => { if (!st.pipe && localChanged()) loadLocal("file-changed"); - if (!st.pipe && Date.now() - st.builtAt > publicRefreshMs && Date.now() - st.lastPublic > publicRefreshMs) refreshPublic(); + if (!st.pipe && publicSnapshotUrls.length && Date.now() - st.builtAt > publicRefreshMs && Date.now() - st.lastPublic > publicRefreshMs) refreshPublic(); }, 10_000); // A silent pipe is a wedged indexer: the indexer pushes "fresh" after // every 30 s poll, so minutes of silence means ask, and drop it if mute. beat = setInterval(async () => { const c = st.pipe; if (!c || Date.now() - st.lastPush < 3 * 60_000) return; - try { const h = await c.request("hello", {}, { timeoutMs: 10_000 }); st.lastPush = Date.now(); if (h.gen > st.gen) onPush({ event: "change", gen: h.gen, changed: [] }); } + try { const h = await c.request("hello", {}, { timeoutMs: 10_000 }); st.lastPush = Date.now(); setPaused(!!h.paused); if (h.gen > st.gen) onPush({ event: "change", gen: h.gen, changed: [] }); } catch { c.close(); } }, 60_000); // No data at all yet: the public snapshot now, in the background. - if (!st.index.size) setTimeout(() => { if (!st.pipe) refreshPublic(); }, opts.connectPipe ? 3_000 : 0); + if (!st.index.size && publicSnapshotUrls.length) setTimeout(() => { if (!st.pipe) refreshPublic(); }, opts.connectPipe ? 3_000 : 0); tick.unref?.(); beat.unref?.(); } function stop() { @@ -280,12 +322,16 @@ export function createSourceChain(opts) { // defineProperties, not assign: assign would freeze the getters. return Object.defineProperties(ev, Object.getOwnPropertyDescriptors({ - start, stop, lookup, resolve, refreshPublic, loadLocal, + start, stop, lookup, resolve, refreshPublic, loadLocal, requestPoll, + get builtAt() { return st.builtAt; }, + get paused() { return st.paused; }, + get pipeConnected() { return !!st.pipe; }, + get source() { return st.source; }, get index() { return st.index; }, get tlds() { return st.tlds; }, status() { return { - source: st.source, size: st.index.size, builtAt: st.builtAt, gen: st.gen, stale: stale(), + source: st.source, size: st.index.size, builtAt: st.builtAt, gen: st.gen, stale: stale(), paused: st.paused, pipe: { connected: !!st.pipe, since: st.pipe ? st.pipeSince : null, error: st.pipeError, verdict: st.pipe?.verdict || null }, localPaths: opts.localPaths || [], ownCopyPath: opts.ownCopyPath || null, provisional: [...provisional.values()].filter((p) => p.entry).length,