From 287c50fe1a7c486bd5e72477bbc19b7a1703e607 Mon Sep 17 00:00:00 2001 From: jason5ng32 Date: Mon, 5 Oct 2026 02:21:58 +0800 Subject: [PATCH] Feat(datasets): one updater engine for every offline dataset MaxMind and CAIDA each had their own updater with its own lock, state file, publish step, boot download and timer. One engine (common/dataset-updater.js) now owns all of that, and each dataset is a row in common/datasets.js that holds only its own logic: find the newest remote version, fetch or distill it, validate, reload. - Locks: proper-lockfile, a heartbeat lock shared by every process, so a crashed holder's lock goes stale on its own; a booting server waits for a live holder and loads what it published. - State: one file per dataset, written with write-file-atomic; the old .maxmind-update-state.json / .caida-update-state.json are read once, so an upgrade doesn't re-download. - Schedule: croner at a fixed time (DATASET_UPDATE_CRON, default 04:30 local) instead of 24 h from process start, plus a catch-up run at boot for a check a downtime skipped. DATASET_AUTO_UPDATE (default on) replaces MAXMIND_AUTO_UPDATE / CAIDA_AUTO_UPDATE, which still apply when it is unset. - Reload: the engine watches every dataset's files, so a publish by another process or a hand-placed .mmdb is picked up; MaxMind's own watcher goes. MaxMind's City and ASN editions now publish together under one version. Co-Authored-By: Claude Opus 5.5 --- .env.example | 10 +- .gitignore | 4 + api/AGENTS.md | 18 +- backend-server.js | 34 +- common/as-org-db.js | 6 +- common/as-rel-db.js | 4 +- common/caida-updater.js | 453 --------------- common/dataset-updater.js | 360 ++++++++++++ common/datasets.js | 263 +++++++++ common/maxmind-service.js | 35 +- common/maxmind-updater.js | 532 ------------------ common/peeringdb-db.js | 4 +- common/sentry-cron.js | 4 +- package.json | 5 +- pnpm-lock.yaml | 49 ++ tests/dataset-updater.test.js | 275 +++++++++ ...caida-updater.test.js => datasets.test.js} | 159 ++++-- 17 files changed, 1122 insertions(+), 1093 deletions(-) delete mode 100644 common/caida-updater.js create mode 100644 common/dataset-updater.js create mode 100644 common/datasets.js delete mode 100644 common/maxmind-updater.js create mode 100644 tests/dataset-updater.test.js rename tests/{caida-updater.test.js => datasets.test.js} (53%) diff --git a/.env.example b/.env.example index 2e2efed0..726c6e60 100644 --- a/.env.example +++ b/.env.example @@ -17,12 +17,14 @@ RIPESTAT_SOURCE_APP="" # dashboard, not its name. CLOUDFLARE_ACCOUNT_ID="" CLOUDFLARE_KV_NAMESPACE_ID="" -# MaxMind +# MaxMind — GeoLite2 is downloaded only when both are set MAXMIND_ACCOUNT_ID="" MAXMIND_LICENSE_KEY="" -MAXMIND_AUTO_UPDATE="false" -# CAIDA -CAIDA_AUTO_UPDATE="false" +# Offline datasets (MaxMind, CAIDA, PeeringDB): missing ones download at boot; +# all are re-checked daily at DATASET_UPDATE_CRON (server-local time) unless +# DATASET_AUTO_UPDATE is "false" +DATASET_AUTO_UPDATE="true" +DATASET_UPDATE_CRON="30 4 * * *" # SITE VITE_SITE_URL="https://ipcheck.ing" # DOCS SITE — origin of the GitBook docs site powering the in-app diff --git a/.gitignore b/.gitignore index 15165f03..b56f2d3a 100644 --- a/.gitignore +++ b/.gitignore @@ -55,6 +55,10 @@ common/peeringdb-db/.caida-update-state.json common/peeringdb-db/.caida-update.lock common/peeringdb-db/*.bak common/peeringdb-db/*.next +# Dataset updater (common/dataset-updater.js): state (+ its atomic-write +# temp files) and the lock directory, in every dataset directory. +common/*-db/.dataset-state.json* +common/*-db/.dataset.lock .learnings/ docs/ .plan/ diff --git a/api/AGENTS.md b/api/AGENTS.md index 05861753..2f37ccfc 100644 --- a/api/AGENTS.md +++ b/api/AGENTS.md @@ -42,10 +42,20 @@ behind `dns-resolver` (gated by `tests/dns-resolvers-data.test.js`). no prefix or relationship counts — the ASN Profile has both. - **`/api/whois` `?q=`:** IPs (RDAP, whoiser fallback), domains (whoiser, RDAP fallback), `AS` (RDAP autnum); all carry a `__raw` block. -- **Offline datasets** (`common/caida-updater.js` rows): fetched when missing at - boot, refreshed daily with `CAIDA_AUTO_UPDATE=true`; PeeringDB only with a - Cloudflare key. Its 116 MB dump is stream-distilled (never parsed whole) to a - ~6 MB per-ASN index without contact data (`common/peeringdb-distill.js`). +- **Offline datasets** (MaxMind, CAIDA, PeeringDB) are rows in + `common/datasets.js`, run by one engine, + `common/dataset-updater.js` (lock, atomic state, publish, boot download, + daily schedule). A row holds only its own logic — find the newest remote + version, fetch / distill it, validate, reload; a new dataset is a new row, + never a new updater. Fetched when missing at boot; refreshed daily at + `DATASET_UPDATE_CRON` (default 04:30 local) unless `DATASET_AUTO_UPDATE=false` + (a row's pre-engine flag, e.g. `CAIDA_AUTO_UPDATE`, still applies when that + is unset), with a catch-up run at boot for a check a downtime skipped. Files + changed on disk by anyone else (another process, a hand-placed `.mmdb`) are + reloaded by the engine's watcher. MaxMind only with its credentials (City + ASN publish together); + PeeringDB only with a Cloudflare key; its 116 MB dump is + stream-distilled (never parsed whole) to a ~6 MB per-ASN index without + contact data (`common/peeringdb-distill.js`). ## Conventions diff --git a/backend-server.js b/backend-server.js index 5e5315d4..a6a94949 100644 --- a/backend-server.js +++ b/backend-server.js @@ -46,9 +46,9 @@ import { isCompleteProfile } from './common/asn-profile.js'; import validateConfigs from './api/configs.js'; import getUserinfo from './api/get-user-info.js'; import updateUserAchievement from './api/update-user-achievement.js'; -import { reloadMaxMindDatabases, startMaxMindFileWatcher } from './common/maxmind-service.js'; -import { startMaxMindAutoUpdate, bootstrapMaxMindIfMissing } from './common/maxmind-updater.js'; -import { startCaidaAutoUpdate, bootstrapCaidaIfMissing } from './common/caida-updater.js'; +import { reloadMaxMindDatabases, isMaxMindReady } from './common/maxmind-service.js'; +import { bootstrapDatasets, startDatasetScheduler, watchDatasets } from './common/dataset-updater.js'; +import { datasets } from './common/datasets.js'; import { bootstrapServiceStatus, startServiceStatusPolling } from './common/service-status-store.js'; import { initUpstreamUserAgent } from './common/upstream-ua.js'; @@ -338,22 +338,24 @@ if (process.env.SENTRY_DSN_BACKEND) { } -// Bootstrap every offline dataset (MaxMind, CAIDA incl. the PeeringDB -// mirror) before accepting traffic so we never serve mid-download. Each step -// is non-fatal: a failure leaves the dependent API in a degraded state -// (MaxMind → 503; CAIDA → empty graph, RIPEstat fallback or no peering -// section) but doesn't block the listener. +// Bootstrap every offline dataset (common/datasets.js) before accepting +// traffic so we never serve mid-download. Each step is non-fatal: a failure +// leaves the dependent API in a degraded state (MaxMind → 503; CAIDA → empty +// graph, RIPEstat fallback or no peering section) but doesn't block the +// listener. async function bootBackend() { - await bootstrapMaxMindIfMissing({ reload: reloadMaxMindDatabases }); - await reloadMaxMindDatabases('startup').catch(() => { - logger.error('❌ MaxMind API will return 503 until databases are loaded successfully'); - }); - await bootstrapCaidaIfMissing(); + await bootstrapDatasets(datasets); + // A database already on disk was not loaded by the bootstrap. + if (!isMaxMindReady()) { + await reloadMaxMindDatabases('startup').catch(() => { + logger.error('❌ MaxMind API will return 503 until databases are available: set MAXMIND_ACCOUNT_ID + ' + + 'MAXMIND_LICENSE_KEY, or drop GeoLite2-City.mmdb + GeoLite2-ASN.mmdb into common/maxmind-db/'); + }); + } await bootstrapServiceStatus(); - startMaxMindFileWatcher(); - startMaxMindAutoUpdate({ reload: reloadMaxMindDatabases }); - startCaidaAutoUpdate(); + watchDatasets(datasets); + startDatasetScheduler(datasets); startServiceStatusPolling(); app.listen(backEndPort, () => { diff --git a/common/as-org-db.js b/common/as-org-db.js index 67b6b5ae..ee6dd9ad 100644 --- a/common/as-org-db.js +++ b/common/as-org-db.js @@ -1,5 +1,5 @@ -// Local CAIDA as2org lookup. Snapshot file is auto-downloaded by -// common/caida-updater.js; this module just parses and serves. +// Local CAIDA as2org lookup. Snapshot file is auto-downloaded by the +// dataset updater (common/datasets.js); this module just parses and serves. // // We use CAIDA's pipe-delimited TXT (~12MB) rather than the equivalent // JSONL (~28MB): identical content but split('|') beats JSON.parse per @@ -81,7 +81,7 @@ function loadDatabase() { loadDatabase(); -/** Reload the snapshot after caida-updater publishes a fresh file. */ +/** Reload the snapshot after the dataset updater publishes a fresh file. */ export function reloadAsOrgDatabase(reason = 'reload') { logger.info(`🔄 Reloading CAIDA as2org snapshot (${reason})`); loadDatabase(); diff --git a/common/as-rel-db.js b/common/as-rel-db.js index c0b64ae3..3d793751 100644 --- a/common/as-rel-db.js +++ b/common/as-rel-db.js @@ -1,5 +1,5 @@ // Local CAIDA AS Relationships lookup. Snapshot is downloaded -// by common/caida-updater.js; this module parses and serves. +// by the dataset updater (common/datasets.js); this module parses and serves. // // CAIDA as-rel2 row format: `|||` // rel = -1 → p2c (a is provider of b) ← providers + customers indexes @@ -123,7 +123,7 @@ function loadDatabase() { loadDatabase(); -/** Reload the snapshot after caida-updater publishes a fresh file. */ +/** Reload the snapshot after the dataset updater publishes a fresh file. */ export function reloadAsRelDatabase(reason = 'reload') { logger.info(`🔄 Reloading CAIDA as-rel snapshot (${reason})`); loadDatabase(); diff --git a/common/caida-updater.js b/common/caida-updater.js deleted file mode 100644 index c32dcb42..00000000 --- a/common/caida-updater.js +++ /dev/null @@ -1,453 +0,0 @@ -// Vendor-scoped CAIDA auto-updater. Each dataset is a row in `datasets` -// below; the orchestration (lock / state / atomic publish / validation / -// reload) is shared. Mirrors common/maxmind-updater.js's editions-array -// pattern. -// -// Registered datasets: -// - as2org (AS → org-name mapping) common/as-org-db -// - as-rel (AS relationships, p2c) common/as-rel-db -// - peeringdb (PeeringDB dump → per-ASN index) common/peeringdb-db -// -// All share CAIDA_AUTO_UPDATE=true to gate the periodic scheduler (off by -// default); bootstrap always runs so a fresh checkout works. A row may add -// `enabled()` (skipped by both when false — peeringdb needs the Cloudflare -// key, as the ASN Profile is hidden without it), leave out `decompress` -// (the download is used as is) and add `distill(rawPath, tempDir, { signal })` -// (turns the download into the published file; the raw file is deleted). - -import fs from 'fs'; -import fsp from 'fs/promises'; -import os from 'os'; -import path from 'path'; -import { Readable } from 'stream'; -import { pipeline } from 'stream/promises'; -import logger from './logger.js'; -import { fetchUpstream } from './fetch-with-timeout.js'; -import { withCronMonitor } from './sentry-cron.js'; -import { createDecompressor } from './decompress.js'; -import { AS_ORG_DB_DIR, AS_ORG_FILE, reloadAsOrgDatabase } from './as-org-db.js'; -import { AS_REL_DB_DIR, AS_REL_FILE, reloadAsRelDatabase } from './as-rel-db.js'; -import { PEERINGDB_DB_DIR, PEERINGDB_FILE, readPeeringdbIndex, reloadPeeringdbDatabase } from './peeringdb-db.js'; -import { distillPeeringdbDump } from './peeringdb-distill.js'; -import { hasRadarApiKey } from './cf-radar.js'; - -const UPDATE_INTERVAL_MS = 24 * 60 * 60 * 1000; -const INITIAL_UPDATE_DELAY_MS = 60 * 1000; -const LOCK_STALE_MS = 2 * 60 * 60 * 1000; -const BOOTSTRAP_TIMEOUT_MS = 2 * 60 * 1000; -const STATE_FILE = '.caida-update-state.json'; -const LOCK_FILE = '.caida-update.lock'; - -// ---------- PeeringDB row helpers ---------- -// Arrow consts, so declared ahead of the `datasets` registry that holds them. - -// Healthy index has ~33k networks; <20k indicates truncation or schema change. -const validatePeeringdb = async (stagedPath) => { - const MIN_VALID = 20000; - const { nets } = readPeeringdbIndex(stagedPath); - if (nets.size < MIN_VALID) { - throw new Error(`Staged PeeringDB index has only ${nets.size} networks; refusing to publish`); - } -}; - -// CAIDA mirrors the full PeeringDB dump daily (a day late) under YYYY/MM/. -// List the current UTC month, then the previous one for the days before the -// month's first dump lands; the newest filename wins. The listing is small, -// so it takes fetchUpstream's timeout and User-Agent (the caller's signal -// still aborts it); the dump download itself does not. Exported for tests. -export const findPeeringdbDump = async ({ signal, now = new Date() } = {}) => { - const base = 'https://publicdata.caida.org/datasets/peeringdb/'; - for (const back of [0, 1]) { - const month = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth() - back, 1)); - const dir = `${base}${month.getUTCFullYear()}/${String(month.getUTCMonth() + 1).padStart(2, '0')}/`; - const res = await fetchUpstream(dir, { signal }); - if (res.status === 404) continue; - if (!res.ok) throw new Error(`Directory listing failed: HTTP ${res.status}`); - const html = await res.text(); - const names = [...html.matchAll(/\bpeeringdb_2_dump_\d{4}_\d{2}_\d{2}\.json\b/g)].map(m => m[0]); - if (names.length === 0) continue; - names.sort((a, b) => b.localeCompare(a)); - return { url: dir + names[0], identifier: names[0] }; - } - throw new Error('No PeeringDB dump in the current or previous month listing'); -}; - -// A row without `enabled` is always on. Exported for tests. -export const isDatasetEnabled = (dataset) => !dataset.enabled || Boolean(dataset.enabled()); - -// Exported for tests (rows are re-pointed at a temp dir there). -export const datasets = [ - { - id: 'as2org', - dbDir: AS_ORG_DB_DIR, - canonicalFile: AS_ORG_FILE, - archiveExt: '.txt.gz', - decompress: 'gzip', - // Stable 'latest' symlink server-side → HEAD + Last-Modified is enough. - findRemote: async ({ signal } = {}) => { - const url = 'https://publicdata.caida.org/datasets/as-organizations/latest.as-org2info.txt.gz'; - const res = await fetch(url, { method: 'HEAD', redirect: 'follow', signal }); - if (!res.ok) throw new Error(`HEAD failed: HTTP ${res.status}`); - return { url, identifier: res.headers.get('last-modified') }; - }, - validate: validateAsOrg, - reload: reloadAsOrgDatabase, - }, - { - id: 'as-rel', - dbDir: AS_REL_DB_DIR, - canonicalFile: AS_REL_FILE, - archiveExt: '.txt.bz2', - decompress: 'bzip2', - // No 'latest' symlink — scrape directory listing, lex-sort YYYYMMDD - // filenames (= chrono), take the newest. - findRemote: async ({ signal } = {}) => { - const dir = 'https://publicdata.caida.org/datasets/as-relationships/serial-2/'; - const res = await fetch(dir, { signal }); - if (!res.ok) throw new Error(`Directory listing failed: HTTP ${res.status}`); - const html = await res.text(); - const matches = [...html.matchAll(/\b(\d{8})\.as-rel2\.txt\.bz2\b/g)]; - if (matches.length === 0) throw new Error('No as-rel2.txt.bz2 found in listing'); - matches.sort((a, b) => b[1].localeCompare(a[1])); - const filename = matches[0][0]; - return { url: dir + filename, identifier: filename }; - }, - validate: validateAsRel, - reload: reloadAsRelDatabase, - }, - { - id: 'peeringdb', - dbDir: PEERINGDB_DB_DIR, - canonicalFile: PEERINGDB_FILE, - archiveExt: '.json', - enabled: hasRadarApiKey, - findRemote: findPeeringdbDump, - // ~116 MB raw dump → ~6 MB index; only the index is published. - distill: async (rawPath, tempDir, { signal } = {}) => { - const outPath = path.join(tempDir, PEERINGDB_FILE); - const counts = await distillPeeringdbDump(rawPath, outPath, { signal }); - logger.info({ dataset: 'peeringdb', ...counts }, 'PeeringDB dump distilled'); - return outPath; - }, - validate: validatePeeringdb, - reload: reloadPeeringdbDatabase, - }, -]; - -let schedulerStarted = false; -const updateInProgress = new Set(); - -// ---------- Public API ---------- - -/** - * Download any missing CAIDA snapshots at boot. Always runs regardless of - * CAIDA_AUTO_UPDATE — a missing snapshot at boot means the operator wants - * one. Each dataset is independent; one failing doesn't abort others. - * Never throws. - */ -export async function bootstrapCaidaIfMissing() { - for (const dataset of datasets.filter(isDatasetEnabled)) { - try { - await bootstrapDataset(dataset); - } catch (error) { - logger.warn({ err: error, dataset: dataset.id }, '⚠️ CAIDA bootstrap failed'); - } - } -} - -/** Start the periodic updater. Opt-in via CAIDA_AUTO_UPDATE=true. */ -export function startCaidaAutoUpdate() { - if (schedulerStarted) return; - schedulerStarted = true; - if (!isAutoUpdateEnabled()) return; - - // withCronMonitor reports ok/error check-ins to Sentry Crons (no-op - // without a backend DSN). Note runAllUpdates degrades per-dataset - // internally, so the monitor mainly catches a dead/hung scheduler. - const run = () => { - withCronMonitor('caida-auto-update', runAllUpdates, { - schedule: { type: 'interval', value: UPDATE_INTERVAL_MS / (60 * 60 * 1000), unit: 'hour' }, - checkinMargin: 60, - maxRuntime: 30, - }).catch(error => { - logger.error({ err: error }, 'CAIDA auto update tick failed'); - }); - }; - setTimeout(run, INITIAL_UPDATE_DELAY_MS).unref?.(); - setInterval(run, UPDATE_INTERVAL_MS).unref?.(); - - const nextRunAt = new Date(Date.now() + INITIAL_UPDATE_DELAY_MS); - logger.info(`🗓️ CAIDA auto update plan: next check at ${nextRunAt.toLocaleString('en-US', { hour12: false })}, then every 24 hours (${datasets.filter(isDatasetEnabled).length} datasets)`); -} - -// ---------- Orchestration ---------- - -async function runAllUpdates() { - for (const dataset of datasets.filter(isDatasetEnabled)) { - try { - await updateDataset(dataset); - } catch (error) { - logger.error({ err: error, dataset: dataset.id }, 'CAIDA update failed'); - } - } -} - -// Exported for tests (with a temp-dir dataset row); not part of the API. -export async function bootstrapDataset(dataset) { - await fsp.mkdir(dataset.dbDir, { recursive: true }); - await clearOrphanedLock(dataset); - - if (snapshotExists(dataset)) return { status: 'present' }; - - const timeoutMinutes = BOOTSTRAP_TIMEOUT_MS / 60000; - logger.warn(`📥 CAIDA ${dataset.id} snapshot missing; attempting initial download (timeout ${timeoutMinutes} min)...`); - - const controller = new AbortController(); - const timer = setTimeout(() => controller.abort(new Error('bootstrap timed out')), BOOTSTRAP_TIMEOUT_MS); - timer.unref?.(); - - try { - const result = await updateDataset(dataset, { - signal: controller.signal, - reloadReason: 'bootstrap', - }); - if (result.updated) { - logger.warn(`✅ CAIDA ${dataset.id} snapshot downloaded and ready`); - return { status: 'downloaded' }; - } - logger.warn(`⚠️ CAIDA ${dataset.id} bootstrap did not publish (${result.reason}).`); - return { status: 'no-op', reason: result.reason }; - } catch (error) { - const reason = controller.signal.aborted - ? `download did not complete within ${timeoutMinutes} min` - : error.message; - logger.warn(`⚠️ CAIDA ${dataset.id} initial download failed: ${reason}`); - return { status: 'failed', error }; - } finally { - clearTimeout(timer); - } -} - -// Exported for tests, like bootstrapDataset. -export async function updateDataset(dataset, { signal, reloadReason = 'auto update' } = {}) { - if (updateInProgress.has(dataset.id)) { - return { updated: false, reason: 'already-running' }; - } - updateInProgress.add(dataset.id); - - await fsp.mkdir(dataset.dbDir, { recursive: true }); - - const lock = await acquireUpdateLock(dataset); - if (!lock) { - updateInProgress.delete(dataset.id); - return { updated: false, reason: 'locked' }; - } - - const tempDir = await fsp.mkdtemp(path.join(os.tmpdir(), `myip-caida-${dataset.id}-`)); - - try { - const result = await downloadAndPublish(dataset, tempDir, { signal }); - if (result.updated && dataset.reload) { - dataset.reload(reloadReason); - } - return result; - } finally { - updateInProgress.delete(dataset.id); - await fsp.rm(tempDir, { recursive: true, force: true }); - await lock.release(); - } -} - -async function downloadAndPublish(dataset, tempDir, { signal } = {}) { - const state = await readUpdateState(dataset); - const remote = await dataset.findRemote({ signal }); - const targetPath = path.join(dataset.dbDir, dataset.canonicalFile); - - if (fs.existsSync(targetPath) && state.identifier && state.identifier === remote.identifier) { - return { updated: false, reason: 'not-modified' }; - } - - let stagedPath = await downloadAndDecompress(dataset, remote.url, tempDir, { signal }); - if (dataset.distill) { - const rawPath = stagedPath; - try { - stagedPath = await dataset.distill(rawPath, tempDir, { signal }); - } catch (error) { - throw new Error(`${dataset.id} distill phase failed: ${error.message}`); - } finally { - await fsp.rm(rawPath, { force: true }); - } - } - await dataset.validate(stagedPath); - await publishFile(stagedPath, targetPath); - - await writeUpdateState(dataset, { - identifier: remote.identifier, - updatedAt: new Date().toISOString(), - }); - - logger.info({ dataset: dataset.id, identifier: remote.identifier }, 'CAIDA snapshot updated'); - return { updated: true, identifier: remote.identifier }; -} - -// Stream archive to disk, then decompress to staged .txt (a row without -// `decompress` stages the download itself). Split so errors clearly -// identify which phase failed (network vs decompression). -async function downloadAndDecompress(dataset, url, tempDir, { signal } = {}) { - const archivePath = path.join(tempDir, `archive${dataset.archiveExt}`); - const stagedPath = path.join(tempDir, dataset.canonicalFile); - - const response = await fetch(url, { redirect: 'follow', signal }); - if (!response.ok || !response.body) { - throw new Error(`Failed to download ${dataset.id}: HTTP ${response.status}`); - } - try { - await pipeline(Readable.fromWeb(response.body), fs.createWriteStream(archivePath), { signal }); - } catch (error) { - throw new Error(`${dataset.id} download phase failed: ${error.message}`); - } - if (!dataset.decompress) return archivePath; - - try { - await pipeline( - fs.createReadStream(archivePath), - createDecompressor(dataset.decompress), - fs.createWriteStream(stagedPath), - ); - } catch (error) { - throw new Error(`${dataset.id} ${dataset.decompress} decompression phase failed: ${error.message}`); - } - - return stagedPath; -} - -// ---------- Validators (per-dataset, refuse partial / corrupt downloads) ---------- - -// Healthy as2org resolves ~110k ASNs; <50k indicates truncation or schema change. -async function validateAsOrg(stagedPath) { - const MIN_VALID = 50000; - const text = await fsp.readFile(stagedPath, 'utf8'); - const orgs = new Map(); - const pending = []; - for (const rawLine of text.split('\n')) { - const line = rawLine.trim(); - if (!line || line.startsWith('#')) continue; - const parts = line.split('|'); - if (parts.length === 5 && parts[0] && parts[2]) { - orgs.set(parts[0], parts[2]); - } else if (parts.length === 6 && Number(parts[0]) && parts[3]) { - pending.push(parts[3]); - } - } - let resolved = 0; - for (const orgId of pending) if (orgs.has(orgId)) resolved++; - if (resolved < MIN_VALID) { - throw new Error(`Staged as2org has only ${resolved} resolvable ASNs; refusing to publish`); - } -} - -// Healthy as-rel2 has ~400k p2c rows; <100k indicates truncation. -async function validateAsRel(stagedPath) { - const MIN_VALID = 100000; - const text = await fsp.readFile(stagedPath, 'utf8'); - let p2cRows = 0; - for (const rawLine of text.split('\n')) { - const line = rawLine.trim(); - if (!line || line.startsWith('#')) continue; - const parts = line.split('|'); - if (parts.length >= 3 && parts[2] === '-1') p2cRows++; - } - if (p2cRows < MIN_VALID) { - throw new Error(`Staged as-rel has only ${p2cRows} p2c rows; refusing to publish`); - } -} - -// ---------- Generic helpers ---------- - -async function publishFile(stagedPath, targetPath) { - const nextPath = `${targetPath}.next`; - const backupPath = `${targetPath}.bak`; - await fsp.copyFile(stagedPath, nextPath); - await fsp.copyFile(targetPath, backupPath).catch(error => { - if (error.code !== 'ENOENT') throw error; - }); - try { - await fsp.rename(nextPath, targetPath); - } catch (error) { - await fsp.copyFile(backupPath, targetPath).catch(() => {}); - throw error; - } finally { - await Promise.all([ - fsp.rm(nextPath, { force: true }), - fsp.rm(backupPath, { force: true }), - ]); - } -} - -function isAutoUpdateEnabled() { - return process.env.CAIDA_AUTO_UPDATE === 'true'; -} - -// Any non-dot file with the canonical file's extension (.txt for the -// CAIDA rows, so manual downloads under another name still count). -function snapshotExists(dataset) { - if (!fs.existsSync(dataset.dbDir)) return false; - const ext = path.extname(dataset.canonicalFile); - return fs.readdirSync(dataset.dbDir).some(f => f.endsWith(ext) && !f.startsWith('.')); -} - -// At boot, any pre-existing lock is necessarily from a previous crashed run -// (we just started, no in-process updater can hold it). Clear it so a dead -// Ctrl+C doesn't block restart for LOCK_STALE_MS. -async function clearOrphanedLock(dataset) { - const lockPath = path.join(dataset.dbDir, LOCK_FILE); - if (fs.existsSync(lockPath)) { - await fsp.rm(lockPath, { force: true }); - logger.warn(`🧹 Cleared orphaned CAIDA ${dataset.id} lock from previous boot`); - } -} - -async function readUpdateState(dataset) { - try { - const content = await fsp.readFile(path.join(dataset.dbDir, STATE_FILE), 'utf8'); - return JSON.parse(content); - } catch (error) { - if (error.code === 'ENOENT') return {}; - throw error; - } -} - -async function writeUpdateState(dataset, state) { - await fsp.writeFile( - path.join(dataset.dbDir, STATE_FILE), - `${JSON.stringify(state, null, 2)}\n`, - 'utf8', - ); -} - -async function acquireUpdateLock(dataset) { - const lockPath = path.join(dataset.dbDir, LOCK_FILE); - try { - const handle = await fsp.open(lockPath, 'wx'); - await handle.writeFile(JSON.stringify({ - pid: process.pid, - dataset: dataset.id, - startedAt: new Date().toISOString(), - })); - return { - release: async () => { - await handle.close(); - await fsp.rm(lockPath, { force: true }); - }, - }; - } catch (error) { - if (error.code !== 'EEXIST') throw error; - const stat = await fsp.stat(lockPath).catch(() => null); - if (stat && Date.now() - stat.mtimeMs > LOCK_STALE_MS) { - await fsp.rm(lockPath, { force: true }); - return acquireUpdateLock(dataset); - } - logger.info(`CAIDA ${dataset.id} update skipped: another process is updating`); - return null; - } -} diff --git a/common/dataset-updater.js b/common/dataset-updater.js new file mode 100644 index 00000000..7f1e965b --- /dev/null +++ b/common/dataset-updater.js @@ -0,0 +1,360 @@ +// One engine for every offline snapshot the backend serves — the rows in +// common/datasets.js. A row supplies the domain part: where the newest +// remote version is, how to fetch and distill it into the files it +// publishes, how to validate them, how to reload them. The engine owns the +// rest: +// - a cross-process lock per dataset directory (proper-lockfile: the lock +// carries a heartbeat, so a crashed holder's lock goes stale on its own); +// - a state file per dataset, written atomically (write-file-atomic); +// - publishing a row's files together, rolled back if any rename fails; +// - the boot download of a missing snapshot, under a timeout; +// - the daily schedule (croner), at a fixed wall-clock time, plus a +// catch-up run at boot for datasets a downtime kept from their check; +// - reloading files published by anyone else — another process, or an +// operator dropping files in by hand (watchDatasets). +// +// Row shape: +// id short name, used in logs +// dir directory holding the published files, state and lock +// files published file names, relative to `dir` +// enabled() optional; false skips the row everywhere (a missing key +// or credential), as if it weren't registered +// findRemote({ signal }) → { identifier, … } — identifier names the remote +// version; the whole object is handed on to fetch +// fetch({ remote, tempDir, signal }) → { [file]: stagedPath } for every +// name in `files` +// validate(staged) throws to refuse a truncated or corrupt download +// reload(reason) swaps the in-memory copy for the published files +// legacyState optional { file, toState(json) → { identifier, updatedAt } }: +// the state file an earlier updater wrote, read once so +// an upgrade doesn't re-download +// legacyAutoUpdateEnv optional env var name that used to gate this row's +// schedule (see isAutoUpdateEnabled) +// +// State file (`.dataset-state.json`): { identifier, updatedAt, checkedAt } — +// the published version, when it was published, when the remote was last +// asked about it. + +import fs from 'fs'; +import fsp from 'fs/promises'; +import os from 'os'; +import path from 'path'; +import { Readable } from 'stream'; +import { pipeline } from 'stream/promises'; +import { setTimeout as sleep } from 'timers/promises'; +import lockfile from 'proper-lockfile'; +import writeFileAtomic from 'write-file-atomic'; +import { Cron } from 'croner'; +import logger from './logger.js'; +import { withCronMonitor } from './sentry-cron.js'; +import { createDecompressor } from './decompress.js'; + +export const STATE_FILE = '.dataset-state.json'; +export const LOCK_FILE = '.dataset.lock'; + +// A holder refreshes its lock every minute; one untouched for ten is dead. +const LOCK_OPTIONS = { stale: 10 * 60 * 1000, update: 60 * 1000 }; +const LOCK_RETRY_MS = 2000; +const BOOTSTRAP_TIMEOUT_MS = 5 * 60 * 1000; +export const DEFAULT_UPDATE_CRON = '30 4 * * *'; +// A dataset not checked for this long missed a daily run (the server was +// down at the scheduled time): check it shortly after boot. +const CATCH_UP_AFTER_MS = 25 * 60 * 60 * 1000; +const CATCH_UP_DELAY_MS = 60 * 1000; + +// ---------- download helpers for rows ---------- + +/** Stream `url` into `destPath`. Long downloads: no fetchUpstream timeout, the caller's signal bounds it. */ +export const downloadToFile = async (url, destPath, { signal, headers } = {}) => { + const response = await fetch(url, { redirect: 'follow', signal, headers }); + if (!response.ok || !response.body) throw new Error(`download failed: HTTP ${response.status}`); + await pipeline(Readable.fromWeb(response.body), fs.createWriteStream(destPath), { signal }); +}; + +/** Decompress `srcPath` ('gzip' | 'bzip2') into `destPath`. */ +export const decompressFile = async (srcPath, destPath, format) => { + await pipeline(fs.createReadStream(srcPath), createDecompressor(format), fs.createWriteStream(destPath)); +}; + +// ---------- row state ---------- + +export const isRowEnabled = (row) => !row.enabled || Boolean(row.enabled()); + +/** Whether every file the row publishes is on disk. */ +export const isPresent = (row) => row.files.every((file) => fs.existsSync(path.join(row.dir, file))); + +/** The row's state, falling back to an earlier updater's state file; {} when neither exists. */ +export const readState = async (row) => { + const read = async (file) => JSON.parse(await fsp.readFile(path.join(row.dir, file), 'utf8')); + try { + return await read(STATE_FILE); + } catch (error) { + if (error.code !== 'ENOENT') throw error; + } + if (!row.legacyState) return {}; + try { + return row.legacyState.toState(await read(row.legacyState.file)) || {}; + } catch { + return {}; + } +}; + +const writeState = (row, state) => + writeFileAtomic(path.join(row.dir, STATE_FILE), `${JSON.stringify(state, null, 2)}\n`); + +// The published files as this process last loaded them (mtime + size per +// file), so the watcher can tell someone else's publish from its own. +const loadedFingerprints = new Map(); +const fingerprint = (row) => row.files.map((file) => { + try { + const stat = fs.statSync(path.join(row.dir, file)); + return `${stat.mtimeMs}:${stat.size}`; + } catch { + return '-'; + } +}).join('|'); +const reloadRow = async (row, reason) => { + loadedFingerprints.set(row.id, fingerprint(row)); + await row.reload(reason); +}; + +// ---------- lock ---------- + +// Resolves the release function, or null when another process holds the +// lock and `wait` is off. With `wait`, polls until the lock frees or `signal` +// aborts (which rejects). +const acquireLock = async (row, { wait = false, signal } = {}) => { + for (;;) { + try { + return await lockfile.lock(row.dir, { + ...LOCK_OPTIONS, + lockfilePath: path.join(row.dir, LOCK_FILE), + // Losing the heartbeat mid-update must not kill the server; + // the update finishes and the next one re-locks. + onCompromised: (error) => logger.warn({ err: error, dataset: row.id }, 'dataset lock compromised'), + }); + } catch (error) { + if (error.code !== 'ELOCKED') throw error; + if (!wait) return null; + await sleep(LOCK_RETRY_MS, undefined, { signal }); + } + } +}; + +// ---------- publish ---------- + +// Copy every staged file next to its target, then rename them all into +// place; a failed rename restores the previous files. +const publish = async (row, staged) => { + const plan = row.files.map((file) => { + const target = path.join(row.dir, file); + return { source: staged[file], target, next: `${target}.next`, backup: `${target}.bak` }; + }); + try { + for (const step of plan) { + await fsp.copyFile(step.source, step.next); + await fsp.copyFile(step.target, step.backup).catch((error) => { + if (error.code !== 'ENOENT') throw error; + }); + } + try { + for (const step of plan) await fsp.rename(step.next, step.target); + } catch (error) { + for (const step of plan) await fsp.copyFile(step.backup, step.target).catch(() => {}); + throw error; + } + } finally { + for (const step of plan) { + await fsp.rm(step.next, { force: true }); + await fsp.rm(step.backup, { force: true }); + } + } +}; + +// ---------- update ---------- + +/** + * One locked update of `row`: ask the remote, and when it has a version the + * state doesn't (or a file is missing) fetch, validate, publish, record and + * reload. Resolves { updated: true, identifier } or { updated: false, reason: + * 'not-modified' | 'locked' }; throws when a step fails (nothing published). + * `wait` waits out another holder of the lock instead of answering 'locked'. + */ +export const updateDataset = async (row, { signal, reason = 'auto update', wait = false } = {}) => { + await fsp.mkdir(row.dir, { recursive: true }); + const release = await acquireLock(row, { wait, signal }); + if (!release) return { updated: false, reason: 'locked' }; + const tempDir = await fsp.mkdtemp(path.join(os.tmpdir(), `myip-dataset-${row.id}-`)); + try { + const state = await readState(row); + const remote = await row.findRemote({ signal }); + const now = new Date().toISOString(); + if (isPresent(row) && state.identifier === remote.identifier) { + await writeState(row, { ...state, checkedAt: now }); + return { updated: false, reason: 'not-modified' }; + } + const staged = await row.fetch({ remote, tempDir, signal }); + const missing = row.files.filter((file) => !staged[file]); + if (missing.length) throw new Error(`${row.id} fetch produced no ${missing.join(', ')}`); + await row.validate(staged); + await publish(row, staged); + await writeState(row, { identifier: remote.identifier, updatedAt: now, checkedAt: now }); + logger.info({ dataset: row.id, identifier: remote.identifier }, 'dataset updated'); + await reloadRow(row, reason); + return { updated: true, identifier: remote.identifier }; + } finally { + await fsp.rm(tempDir, { recursive: true, force: true }); + await release().catch(() => {}); + } +}; + +/** + * Boot: download `row` if a file it publishes is missing, under a timeout. + * Another process (another backend, a manual run) holding the lock is waited + * out, and what it published is loaded. Never throws; resolves { status: + * 'disabled' | 'present' | 'downloaded' | 'loaded' | 'no-op' | 'failed' }. + */ +export const bootstrapDataset = async (row, { timeoutMs = BOOTSTRAP_TIMEOUT_MS } = {}) => { + if (!isRowEnabled(row)) return { status: 'disabled' }; + if (isPresent(row)) return { status: 'present' }; + + const minutes = Math.round(timeoutMs / 60000); + logger.warn(`📥 Dataset ${row.id} missing; downloading (timeout ${minutes} min)...`); + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(new Error('bootstrap timed out')), timeoutMs); + timer.unref?.(); + try { + const result = await updateDataset(row, { signal: controller.signal, reason: 'bootstrap', wait: true }); + if (result.updated) { + logger.warn(`✅ Dataset ${row.id} downloaded and ready`); + return { status: 'downloaded' }; + } + // The remote matched what another process published while we waited. + if (isPresent(row)) { + await reloadRow(row, 'bootstrap'); + logger.warn(`✅ Dataset ${row.id} published by another process; loaded`); + return { status: 'loaded' }; + } + logger.warn(`⚠️ Dataset ${row.id} bootstrap did not publish (${result.reason})`); + return { status: 'no-op' }; + } catch (error) { + const why = controller.signal.aborted ? `did not complete within ${minutes} min` : error.message; + logger.warn({ err: error }, `⚠️ Dataset ${row.id} initial download failed: ${why}`); + return { status: 'failed', error }; + } finally { + clearTimeout(timer); + } +}; + +/** Bootstrap every row in parallel; never throws. */ +export const bootstrapDatasets = (rows) => Promise.all(rows.map((row) => bootstrapDataset(row))); + +// ---------- schedule ---------- + +/** + * Whether the schedule updates `row`. DATASET_AUTO_UPDATE decides when set + * ('false' = off); otherwise the row's pre-engine flag when set, so an + * existing .env keeps its choice; otherwise on. + */ +export const isAutoUpdateEnabled = (row, env = process.env) => { + const flag = env.DATASET_AUTO_UPDATE; + if (flag) return flag !== 'false'; + const legacy = row.legacyAutoUpdateEnv && env[row.legacyAutoUpdateEnv]; + if (legacy) return legacy === 'true'; + return true; +}; + +/** Update each row in turn; one failing doesn't stop the rest. */ +export const updateDatasets = async (rows) => { + for (const row of rows) { + try { + await updateDataset(row); + } catch (error) { + logger.error({ err: error, dataset: row.id }, 'dataset update failed'); + } + } +}; + +/** + * The rows of `rows` a downtime kept from their daily check: last checked + * (or published, for a state from before checks were recorded) longer ago + * than a day and a bit, or never. Exported for tests. + */ +export const rowsMissingACheck = async (rows, now = Date.now()) => { + const stale = []; + for (const row of rows) { + const { checkedAt, updatedAt } = await readState(row).catch(() => ({})); + const last = Date.parse(checkedAt || updatedAt); + if (!(now - last < CATCH_UP_AFTER_MS)) stale.push(row); + } + return stale; +}; + +let scheduler = null; + +/** + * Run the auto-updating rows daily at DATASET_UPDATE_CRON (server-local time, + * default 04:30). `protect` skips a tick while the previous run is still + * going. Sentry Crons sees the run as one 'dataset-update' check-in. + */ +export const startDatasetScheduler = (rows) => { + if (scheduler) return scheduler; + const scheduled = rows.filter((row) => isRowEnabled(row) && isAutoUpdateEnabled(row)); + if (!scheduled.length) { + logger.info('🗓️ Dataset auto update: off'); + return null; + } + const pattern = process.env.DATASET_UPDATE_CRON || DEFAULT_UPDATE_CRON; + const timezone = Intl.DateTimeFormat().resolvedOptions().timeZone; + scheduler = new Cron(pattern, { protect: true, unref: true }, () => withCronMonitor( + 'dataset-update', + () => updateDatasets(scheduled), + { schedule: { type: 'crontab', value: pattern }, timezone, checkinMargin: 60, maxRuntime: 60 }, + ).catch((error) => logger.error({ err: error }, 'dataset update run failed'))); + const next = scheduler.nextRun()?.toLocaleString('en-US', { hour12: false }); + logger.info(`🗓️ Dataset auto update: ${scheduled.map((row) => row.id).join(', ')} at cron ${pattern} (next ${next})`); + + rowsMissingACheck(scheduled).then((stale) => { + if (!stale.length) return; + logger.info(`🗓️ Dataset catch-up in ${CATCH_UP_DELAY_MS / 1000}s: ${stale.map((row) => row.id).join(', ')}`); + setTimeout(() => updateDatasets(stale), CATCH_UP_DELAY_MS).unref?.(); + }); + return scheduler; +}; + +// ---------- watch ---------- + +/** + * Reload a row when its published files change under it — another process + * published, or someone dropped files in by hand — but not after this + * process's own publish (already reloaded). Polls every `intervalMs`; events + * within `settleMs` coalesce (a publish renames several files). Watches every + * row, enabled or not: hand-placed files need no credentials. Returns a + * stop function. Options are injectable for tests. + */ +export const watchDatasets = (rows, { intervalMs = 5000, settleMs = 1000 } = {}) => { + const stops = []; + for (const row of rows) { + if (!loadedFingerprints.has(row.id)) loadedFingerprints.set(row.id, fingerprint(row)); + let timer = null; + const check = () => { + clearTimeout(timer); + timer = setTimeout(() => { + const current = fingerprint(row); + if (current === loadedFingerprints.get(row.id) || !isPresent(row)) return; + logger.info({ dataset: row.id }, 'dataset files changed on disk; reloading'); + reloadRow(row, 'file change').catch((error) => { + logger.warn({ err: error, dataset: row.id }, 'dataset reload after a file change failed'); + }); + }, settleMs); + timer.unref?.(); + }; + for (const file of row.files) { + const target = path.join(row.dir, file); + fs.watchFile(target, { interval: intervalMs, persistent: false }, check); + stops.push(() => fs.unwatchFile(target, check)); + } + stops.push(() => clearTimeout(timer)); + } + return () => stops.forEach((stop) => stop()); +}; diff --git a/common/datasets.js b/common/datasets.js new file mode 100644 index 00000000..df0dee3a --- /dev/null +++ b/common/datasets.js @@ -0,0 +1,263 @@ +// The offline datasets the backend serves, as rows for the engine in +// common/dataset-updater.js (row shape documented there). Each row holds +// only the dataset's own logic: where its newest remote version is, how to +// turn the download into the published files, a validator that refuses a +// truncated or corrupt download, and the reload of its in-memory copy. +// +// maxmind GeoLite2 City + ASN, published common/maxmind-service.js +// together; only with MaxMind credentials +// as2org CAIDA AS → org name common/as-org-db.js +// as-rel CAIDA AS relationships common/as-rel-db.js +// peeringdb PeeringDB dump (CAIDA mirror), common/peeringdb-db.js +// distilled to a per-ASN index; only with the Cloudflare key, +// as the ASN Profile is hidden without it +// +// Sources: https://dev.maxmind.com/geoip/geolite2-free-geolocation-data · +// https://publicdata.caida.org/datasets/ (CAIDA AUA) · https://www.peeringdb.com + +import fsp from 'fs/promises'; +import path from 'path'; +import * as tar from 'tar'; +import maxmind from 'maxmind'; +import logger from './logger.js'; +import { fetchUpstream } from './fetch-with-timeout.js'; +import { downloadToFile, decompressFile } from './dataset-updater.js'; +import { AS_ORG_DB_DIR, AS_ORG_FILE, reloadAsOrgDatabase } from './as-org-db.js'; +import { AS_REL_DB_DIR, AS_REL_FILE, reloadAsRelDatabase } from './as-rel-db.js'; +import { PEERINGDB_DB_DIR, PEERINGDB_FILE, readPeeringdbIndex, reloadPeeringdbDatabase } from './peeringdb-db.js'; +import { distillPeeringdbDump } from './peeringdb-distill.js'; +import { hasRadarApiKey } from './cf-radar.js'; +import { MAXMIND_DB_DIR, MAXMIND_CITY_DB, MAXMIND_ASN_DB, reloadMaxMindDatabases } from './maxmind-service.js'; + +// The pre-engine CAIDA updater's state: same { identifier, updatedAt } shape. +const CAIDA_LEGACY_STATE = { file: '.caida-update-state.json', toState: (json) => json }; + +// Download an archive and decompress it to the one file a row publishes. +const fetchCompressed = (file, format) => async ({ remote, tempDir, signal }) => { + const archive = path.join(tempDir, 'archive'); + const staged = path.join(tempDir, file); + await downloadToFile(remote.url, archive, { signal }); + await decompressFile(archive, staged, format); + return { [file]: staged }; +}; + +// ---------- validators ---------- + +// Healthy as2org resolves ~110k ASNs; <50k indicates truncation or schema change. +const validateAsOrg = async ({ [AS_ORG_FILE]: stagedPath }) => { + const MIN_VALID = 50000; + const text = await fsp.readFile(stagedPath, 'utf8'); + const orgs = new Map(); + const pending = []; + for (const rawLine of text.split('\n')) { + const line = rawLine.trim(); + if (!line || line.startsWith('#')) continue; + const parts = line.split('|'); + if (parts.length === 5 && parts[0] && parts[2]) { + orgs.set(parts[0], parts[2]); + } else if (parts.length === 6 && Number(parts[0]) && parts[3]) { + pending.push(parts[3]); + } + } + let resolved = 0; + for (const orgId of pending) if (orgs.has(orgId)) resolved++; + if (resolved < MIN_VALID) { + throw new Error(`Staged as2org has only ${resolved} resolvable ASNs; refusing to publish`); + } +}; + +// Healthy as-rel2 has ~400k p2c rows; <100k indicates truncation. +const validateAsRel = async ({ [AS_REL_FILE]: stagedPath }) => { + const MIN_VALID = 100000; + const text = await fsp.readFile(stagedPath, 'utf8'); + let p2cRows = 0; + for (const rawLine of text.split('\n')) { + const line = rawLine.trim(); + if (!line || line.startsWith('#')) continue; + const parts = line.split('|'); + if (parts.length >= 3 && parts[2] === '-1') p2cRows++; + } + if (p2cRows < MIN_VALID) { + throw new Error(`Staged as-rel has only ${p2cRows} p2c rows; refusing to publish`); + } +}; + +// Healthy index has ~33k networks; <20k indicates truncation or schema change. +const validatePeeringdb = async ({ [PEERINGDB_FILE]: stagedPath }) => { + const MIN_VALID = 20000; + const { nets } = readPeeringdbIndex(stagedPath); + if (nets.size < MIN_VALID) { + throw new Error(`Staged PeeringDB index has only ${nets.size} networks; refusing to publish`); + } +}; + +// ---------- remote lookups ---------- + +// CAIDA mirrors the full PeeringDB dump daily (a day late) under YYYY/MM/. +// List the current UTC month, then the previous one for the days before the +// month's first dump lands; the newest filename wins. Exported for tests. +export const findPeeringdbDump = async ({ signal, now = new Date() } = {}) => { + const base = 'https://publicdata.caida.org/datasets/peeringdb/'; + for (const back of [0, 1]) { + const month = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth() - back, 1)); + const dir = `${base}${month.getUTCFullYear()}/${String(month.getUTCMonth() + 1).padStart(2, '0')}/`; + const res = await fetchUpstream(dir, { signal }); + if (res.status === 404) continue; + if (!res.ok) throw new Error(`Directory listing failed: HTTP ${res.status}`); + const html = await res.text(); + const names = [...html.matchAll(/\bpeeringdb_2_dump_\d{4}_\d{2}_\d{2}\.json\b/g)].map((m) => m[0]); + if (names.length === 0) continue; + names.sort((a, b) => b.localeCompare(a)); + return { url: dir + names[0], identifier: names[0] }; + } + throw new Error('No PeeringDB dump in the current or previous month listing'); +}; + +// ---------- MaxMind ---------- + +// The two editions, published together: one version identifier covers both. +const MAXMIND_EDITIONS = [ + { editionId: 'GeoLite2-City', file: MAXMIND_CITY_DB }, + { editionId: 'GeoLite2-ASN', file: MAXMIND_ASN_DB }, +]; + +export const hasMaxMindCredentials = () => + Boolean(process.env.MAXMIND_ACCOUNT_ID && process.env.MAXMIND_LICENSE_KEY); + +const maxmindUrl = (editionId) => `https://download.maxmind.com/geoip/databases/${editionId}/download?suffix=tar.gz`; +const maxmindAuth = () => ({ + Authorization: `Basic ${Buffer.from(`${process.env.MAXMIND_ACCOUNT_ID}:${process.env.MAXMIND_LICENSE_KEY}`).toString('base64')}`, +}); + +// Last-Modified of each edition, joined: either edition changing is a new +// version. Exported for tests (and the legacy-state mapping below). +export const maxmindIdentifier = (lastModifieds) => lastModifieds.map((value) => value || '?').join(' | '); + +// The pre-engine MaxMind updater kept { [editionId]: { lastModified, updatedAt } }. +const MAXMIND_LEGACY_STATE = { + file: '.maxmind-update-state.json', + toState: (json) => ({ + identifier: maxmindIdentifier(MAXMIND_EDITIONS.map(({ editionId }) => json[editionId]?.lastModified)), + updatedAt: json['GeoLite2-City']?.updatedAt, + }), +}; + +// The archive nests the .mmdb in a dated directory; find it by name. +const findFile = async (dir, name) => { + for (const entry of await fsp.readdir(dir, { withFileTypes: true })) { + const entryPath = path.join(dir, entry.name); + if (entry.isFile() && entry.name === name) return entryPath; + if (entry.isDirectory()) { + const found = await findFile(entryPath, name); + if (found) return found; + } + } + return null; +}; + +// ---------- rows ---------- + +// Exported for tests (rows are re-pointed at a temp dir there). +export const datasets = [ + { + id: 'maxmind', + dir: MAXMIND_DB_DIR, + files: MAXMIND_EDITIONS.map(({ file }) => file), + enabled: hasMaxMindCredentials, + legacyState: MAXMIND_LEGACY_STATE, + legacyAutoUpdateEnv: 'MAXMIND_AUTO_UPDATE', + findRemote: async ({ signal } = {}) => { + const lastModifieds = []; + for (const { editionId } of MAXMIND_EDITIONS) { + const res = await fetchUpstream(maxmindUrl(editionId), { method: 'HEAD', headers: maxmindAuth(), signal }); + if (!res.ok) throw new Error(`Failed to check ${editionId}: HTTP ${res.status}`); + lastModifieds.push(res.headers.get('last-modified')); + } + return { identifier: maxmindIdentifier(lastModifieds) }; + }, + fetch: async ({ tempDir, signal }) => { + const staged = {}; + for (const { editionId, file } of MAXMIND_EDITIONS) { + const archive = path.join(tempDir, `${editionId}.tar.gz`); + const extractDir = path.join(tempDir, editionId); + await fsp.mkdir(extractDir); + await downloadToFile(maxmindUrl(editionId), archive, { signal, headers: maxmindAuth() }); + await tar.x({ file: archive, cwd: extractDir }); + staged[file] = await findFile(extractDir, file); + if (!staged[file]) throw new Error(`${editionId} archive did not contain ${file}`); + } + return staged; + }, + // Both must open as MaxMind databases. + validate: async (staged) => { + await Promise.all(MAXMIND_EDITIONS.map(({ file }) => maxmind.open(staged[file]))); + }, + reload: reloadMaxMindDatabases, + }, + { + id: 'as2org', + dir: AS_ORG_DB_DIR, + files: [AS_ORG_FILE], + legacyState: CAIDA_LEGACY_STATE, + legacyAutoUpdateEnv: 'CAIDA_AUTO_UPDATE', + // Stable 'latest' symlink server-side → HEAD + Last-Modified is enough. + findRemote: async ({ signal } = {}) => { + const url = 'https://publicdata.caida.org/datasets/as-organizations/latest.as-org2info.txt.gz'; + const res = await fetchUpstream(url, { method: 'HEAD', signal }); + if (!res.ok) throw new Error(`HEAD failed: HTTP ${res.status}`); + return { url, identifier: res.headers.get('last-modified') }; + }, + fetch: fetchCompressed(AS_ORG_FILE, 'gzip'), + validate: validateAsOrg, + reload: reloadAsOrgDatabase, + }, + { + id: 'as-rel', + dir: AS_REL_DB_DIR, + files: [AS_REL_FILE], + legacyState: CAIDA_LEGACY_STATE, + legacyAutoUpdateEnv: 'CAIDA_AUTO_UPDATE', + // No 'latest' symlink — scrape the directory listing; YYYYMMDD + // filenames sort chronologically, the newest wins. + findRemote: async ({ signal } = {}) => { + const dir = 'https://publicdata.caida.org/datasets/as-relationships/serial-2/'; + const res = await fetchUpstream(dir, { signal }); + if (!res.ok) throw new Error(`Directory listing failed: HTTP ${res.status}`); + const html = await res.text(); + const matches = [...html.matchAll(/\b(\d{8})\.as-rel2\.txt\.bz2\b/g)]; + if (matches.length === 0) throw new Error('No as-rel2.txt.bz2 found in listing'); + matches.sort((a, b) => b[1].localeCompare(a[1])); + const filename = matches[0][0]; + return { url: dir + filename, identifier: filename }; + }, + fetch: fetchCompressed(AS_REL_FILE, 'bzip2'), + validate: validateAsRel, + reload: reloadAsRelDatabase, + }, + { + id: 'peeringdb', + dir: PEERINGDB_DB_DIR, + files: [PEERINGDB_FILE], + enabled: hasRadarApiKey, + legacyState: CAIDA_LEGACY_STATE, + legacyAutoUpdateEnv: 'CAIDA_AUTO_UPDATE', + findRemote: findPeeringdbDump, + // ~116 MB raw dump → ~6 MB index; only the index is published. + fetch: async ({ remote, tempDir, signal }) => { + const raw = path.join(tempDir, 'dump.json'); + const staged = path.join(tempDir, PEERINGDB_FILE); + await downloadToFile(remote.url, raw, { signal }); + try { + const counts = await distillPeeringdbDump(raw, staged, { signal }); + logger.info({ dataset: 'peeringdb', ...counts }, 'PeeringDB dump distilled'); + } catch (error) { + throw new Error(`peeringdb distill failed: ${error.message}`); + } finally { + await fsp.rm(raw, { force: true }); + } + return { [PEERINGDB_FILE]: staged }; + }, + validate: validatePeeringdb, + reload: reloadPeeringdbDatabase, + }, +]; diff --git a/common/maxmind-service.js b/common/maxmind-service.js index f3f52fe0..f76453fa 100644 --- a/common/maxmind-service.js +++ b/common/maxmind-service.js @@ -1,4 +1,3 @@ -import fs from 'fs'; import path from 'path'; import { fileURLToPath } from 'url'; import maxmind from 'maxmind'; @@ -33,11 +32,9 @@ export const normalizeLang = (tag) => { let cityLookup = null; let asnLookup = null; let reloadPromise = null; -let watchersStarted = false; -let reloadDebounceTimer = null; /** - * Return the canonical MaxMind database paths used by the backend and updater. + * Return the canonical MaxMind database paths. */ export function getMaxMindDbPaths() { return { @@ -92,36 +89,6 @@ export async function reloadMaxMindDatabases(reason = 'manual') { return reloadPromise; } -/** - * Watch database files and reload readers when another process publishes new files. - */ -export function startMaxMindFileWatcher() { - if (watchersStarted) { - return; - } - - watchersStarted = true; - - // Debounce file events so City and ASN replacements are handled as one reload. - const scheduleReload = () => { - clearTimeout(reloadDebounceTimer); - reloadDebounceTimer = setTimeout(() => { - reloadMaxMindDatabases('file change').catch(() => { - // Keep the existing readers when a newly written database is invalid. - }); - }, 1000); - reloadDebounceTimer.unref?.(); - }; - - for (const filePath of [cityDbPath, asnDbPath]) { - fs.watchFile(filePath, { interval: 5000, persistent: false }, (current, previous) => { - if (current.mtimeMs !== previous.mtimeMs || current.size !== previous.size) { - scheduleReload(); - } - }); - } -} - /** * Bare ISO country code of an IP (null when unknown) plus the prefix length * of the MaxMind network holding it, so a caller can walk a range one network diff --git a/common/maxmind-updater.js b/common/maxmind-updater.js deleted file mode 100644 index 7e7c93d1..00000000 --- a/common/maxmind-updater.js +++ /dev/null @@ -1,532 +0,0 @@ -import fs from 'fs'; -import fsp from 'fs/promises'; -import os from 'os'; -import path from 'path'; -import { Readable } from 'stream'; -import { pipeline } from 'stream/promises'; -import * as tar from 'tar'; -import maxmind from 'maxmind'; -import logger from './logger.js'; -import { withCronMonitor } from './sentry-cron.js'; -import { - getMaxMindDbPaths, - MAXMIND_ASN_DB, - MAXMIND_CITY_DB, -} from './maxmind-service.js'; - -const UPDATE_INTERVAL_MS = 24 * 60 * 60 * 1000; // 24 hours -const INITIAL_UPDATE_DELAY_MS = 60 * 1000; -const LOCK_STALE_MS = 2 * 60 * 60 * 1000; -// Cap the blocking "download databases at boot if missing" path. Long enough -// for a reasonable initial download over slow residential links, short enough -// that a misconfigured credential or a blocked outbound connection doesn't -// hold up the whole server indefinitely. We only abort the download — the -// server still starts after this timeout, just without MaxMind ready. -const BOOTSTRAP_TIMEOUT_MS = 5 * 60 * 1000; -const STATE_FILE = '.maxmind-update-state.json'; -const LOCK_FILE = '.maxmind-update.lock'; -const MAXMIND_UPDATE_ENV_KEYS = [ - 'MAXMIND_ACCOUNT_ID', - 'MAXMIND_LICENSE_KEY', - 'MAXMIND_AUTO_UPDATE', -]; - -const editions = [ - { - editionId: 'GeoLite2-City', - fileName: MAXMIND_CITY_DB, - }, - { - editionId: 'GeoLite2-ASN', - fileName: MAXMIND_ASN_DB, - }, -]; - -let schedulerStarted = false; -let updateInProgress = false; - -/** - * Start the periodic MaxMind updater when credentials and opt-in env vars are present. - */ -export function startMaxMindAutoUpdate({ reload } = {}) { - if (schedulerStarted) { - return; - } - - schedulerStarted = true; - - if (!hasAnyUpdateEnvironment()) { - return; - } - - if (!isAutoUpdateEnabled()) { - logger.info('MaxMind auto update plan: disabled'); - return; - } - - if (!hasDownloadCredentials()) { - logger.info('MaxMind auto update skipped: MAXMIND_ACCOUNT_ID or MAXMIND_LICENSE_KEY is missing'); - return; - } - - // Run one update cycle and keep the scheduler alive if that cycle fails. - // withCronMonitor reports ok/error check-ins to Sentry Crons (no-op - // without a backend DSN) so a silently dead scheduler gets noticed. - const run = () => { - withCronMonitor('maxmind-auto-update', () => updateMaxMindDatabases({ reload }), { - schedule: { type: 'interval', value: UPDATE_INTERVAL_MS / (60 * 60 * 1000), unit: 'hour' }, - checkinMargin: 60, // minutes of lateness tolerated before "missed" - maxRuntime: 30, // minutes before an in-progress run counts as failed - }).catch(error => { - logger.error({ err: error }, 'MaxMind auto update failed'); - }); - }; - - const initialTimer = setTimeout(run, INITIAL_UPDATE_DELAY_MS); - initialTimer.unref?.(); - - const intervalTimer = setInterval(run, UPDATE_INTERVAL_MS); - intervalTimer.unref?.(); - - logUpdatePlan(); -} - -/** - * One-shot "download the MaxMind databases at boot if they are missing" path. - * - * The decision tree matches what we want the operator to experience on a - * cold checkout: - * - Both mmdb files already on disk → do nothing, let reload handle it. - * - Files missing AND no MAXMIND_ACCOUNT_ID / MAXMIND_LICENSE_KEY configured - * → print a clear "how to fix this" warning and return; server will start - * anyway but the MaxMind API will 503 until DBs are provided. - * - Files missing AND credentials present → run a single download cycle, - * capped at BOOTSTRAP_TIMEOUT_MS. On failure (timeout, auth error, network - * block), log a warning and return; the server still starts. - * - * MAXMIND_AUTO_UPDATE is intentionally NOT consulted here — that flag only - * gates the periodic scheduler. If the operator put valid credentials in .env, - * the intent is obviously "I want MaxMind working", and making them set a - * second flag just to get the first download would be confusing. - * - * Never throws: the caller ({@link ./../backend-server.js}) treats this as - * advisory and always proceeds to `app.listen`. - */ -export async function bootstrapMaxMindIfMissing({ reload } = {}) { - const { cityDbPath, asnDbPath } = getMaxMindDbPaths(); - if (fs.existsSync(cityDbPath) && fs.existsSync(asnDbPath)) { - return { status: 'present' }; - } - - if (!hasDownloadCredentials()) { - logger.warn( - '⚠️ MaxMind databases are missing and MAXMIND_ACCOUNT_ID / MAXMIND_LICENSE_KEY are not configured.\n' + - ' Set the credentials in .env and restart, or drop GeoLite2-City.mmdb + GeoLite2-ASN.mmdb\n' + - ' into common/maxmind-db/. Starting server anyway; MaxMind API will return 503 until\n' + - ' databases are available.' - ); - return { status: 'missing-credentials' }; - } - - const timeoutMinutes = BOOTSTRAP_TIMEOUT_MS / 60000; - logger.info(`📥 MaxMind databases missing; attempting initial download (timeout ${timeoutMinutes} min)...`); - - const controller = new AbortController(); - const timer = setTimeout(() => controller.abort(new Error('bootstrap timed out')), BOOTSTRAP_TIMEOUT_MS); - timer.unref?.(); - - try { - await updateMaxMindDatabases({ - reload, - signal: controller.signal, - reloadReason: 'bootstrap', - }); - return { status: 'downloaded' }; - } catch (error) { - // Disambiguate "we hit the 5-minute cap" from "the download itself - // failed" so operators know whether to look at network / firewall vs - // credentials / MaxMind account status. - const reason = controller.signal.aborted - ? `download did not complete within ${timeoutMinutes} min (check connectivity to download.maxmind.com)` - : error.message; - logger.warn( - `⚠️ MaxMind initial download failed: ${reason}\n` + - ' Starting server anyway; MaxMind API will return 503 until databases are available.' - ); - return { status: 'failed', error }; - } finally { - clearTimeout(timer); - } -} - -/** - * Run one locked update cycle and reload readers only after a successful publish. - * - * `signal` is an optional AbortSignal that is threaded into every HTTP fetch - * and stream pipeline, so callers (notably the bootstrap path at startup) can - * cap the whole cycle with a single timeout and have in-flight network I/O - * actually stop instead of quietly dangling in the background. - * - * `reloadReason` lets the caller tag the reload log line ("auto update" for - * the scheduler, "bootstrap" for the startup download, etc.). - */ -export async function updateMaxMindDatabases({ reload, signal, reloadReason = 'auto update' } = {}) { - if (updateInProgress) { - return { updated: false, reason: 'already-running' }; - } - - updateInProgress = true; - - const { dbDir } = getMaxMindDbPaths(); - await fsp.mkdir(dbDir, { recursive: true }); - - const lock = await acquireUpdateLock(dbDir); - if (!lock) { - updateInProgress = false; - return { updated: false, reason: 'locked' }; - } - - const tempDir = await fsp.mkdtemp(path.join(os.tmpdir(), 'myip-maxmind-')); - - try { - const result = await downloadAndReplaceDatabases(dbDir, tempDir, { signal }); - - if (result.updated && reload) { - await reload(reloadReason); - } - - return result; - } finally { - updateInProgress = false; - await fsp.rm(tempDir, { recursive: true, force: true }); - await lock.release(); - } -} - -/** - * Check the explicit opt-in flag for automatic MaxMind updates. - */ -function isAutoUpdateEnabled() { - return process.env.MAXMIND_AUTO_UPDATE === 'true'; -} - -/** - * Check whether both MaxMind download credential variables are configured. - */ -function hasDownloadCredentials() { - return Boolean(process.env.MAXMIND_ACCOUNT_ID && process.env.MAXMIND_LICENSE_KEY); -} - -/** - * Check whether the deployment has any MaxMind update-related environment config. - */ -function hasAnyUpdateEnvironment() { - return MAXMIND_UPDATE_ENV_KEYS.some(key => Object.hasOwn(process.env, key)); -} - -/** - * Print the next MaxMind update check and repeat interval. - */ -function logUpdatePlan() { - const nextRunAt = new Date(Date.now() + INITIAL_UPDATE_DELAY_MS); - logger.info(`🗓️ MaxMind auto update plan: next check at ${formatScheduleTime(nextRunAt)}, then every ${formatDuration(UPDATE_INTERVAL_MS)}`); -} - -/** - * Format scheduler timestamps for startup logs. - */ -function formatScheduleTime(date) { - return date.toLocaleString('en-US', { hour12: false }); -} - -/** - * Format millisecond durations into compact hour or minute text. - */ -function formatDuration(milliseconds) { - const hours = milliseconds / (60 * 60 * 1000); - if (Number.isInteger(hours)) { - return `${hours} hour${hours === 1 ? '' : 's'}`; - } - - const minutes = milliseconds / (60 * 1000); - if (Number.isInteger(minutes)) { - return `${minutes} minute${minutes === 1 ? '' : 's'}`; - } - - return `${milliseconds} ms`; -} - -/** - * Download all required newer editions, validate them, then publish them together. - */ -async function downloadAndReplaceDatabases(dbDir, tempDir, { signal } = {}) { - const state = await readUpdateState(dbDir); - const plannedUpdates = []; - - for (const edition of editions) { - const remoteInfo = await getRemoteInfo(edition, { signal }); - const targetPath = path.join(dbDir, edition.fileName); - - if (!shouldDownloadDatabase(targetPath, state, edition.editionId, remoteInfo.lastModified)) { - continue; - } - - const downloadedPath = await downloadAndExtractDatabase(edition, tempDir, { signal }); - plannedUpdates.push({ - ...edition, - lastModified: remoteInfo.lastModified, - sourcePath: downloadedPath, - targetPath, - }); - } - - if (plannedUpdates.length === 0) { - return { updated: false, reason: 'not-modified' }; - } - - await validatePlannedDatabases(dbDir, plannedUpdates); - - await publishDatabases(plannedUpdates); - - for (const update of plannedUpdates) { - state[update.editionId] = { - lastModified: update.lastModified, - updatedAt: new Date().toISOString(), - }; - } - - await writeUpdateState(dbDir, state); - logger.info({ editions: plannedUpdates.map(update => update.editionId) }, 'MaxMind databases updated'); - - return { - updated: true, - editions: plannedUpdates.map(update => update.editionId), - }; -} - -/** - * Fetch remote metadata for a MaxMind edition without downloading the archive. - */ -async function getRemoteInfo(edition, { signal } = {}) { - const response = await fetch(getDownloadUrl(edition.editionId), { - method: 'HEAD', - headers: getAuthHeaders(), - redirect: 'follow', - signal, - }); - - if (!response.ok) { - throw new Error(`Failed to check ${edition.editionId}: HTTP ${response.status}`); - } - - return { - lastModified: response.headers.get('last-modified'), - }; -} - -/** - * Decide whether the local database should be refreshed from MaxMind. - */ -function shouldDownloadDatabase(targetPath, state, editionId, lastModified) { - if (!fs.existsSync(targetPath)) { - return true; - } - - if (!lastModified) { - return true; - } - - const lastRemoteUpdate = Date.parse(lastModified); - if (Number.isNaN(lastRemoteUpdate)) { - return true; - } - - if (!state[editionId]?.lastModified) { - return true; - } - - const recordedUpdate = Date.parse(state[editionId].lastModified); - return Number.isNaN(recordedUpdate) || lastRemoteUpdate > recordedUpdate; -} - -/** - * Download a MaxMind archive, extract it, and return the matching .mmdb file path. - */ -async function downloadAndExtractDatabase(edition, tempDir, { signal } = {}) { - const editionDir = path.join(tempDir, edition.editionId); - const archivePath = path.join(tempDir, `${edition.editionId}.tar.gz`); - - await fsp.mkdir(editionDir, { recursive: true }); - - const response = await fetch(getDownloadUrl(edition.editionId), { - headers: getAuthHeaders(), - redirect: 'follow', - signal, - }); - - if (!response.ok || !response.body) { - throw new Error(`Failed to download ${edition.editionId}: HTTP ${response.status}`); - } - - await pipeline(Readable.fromWeb(response.body), fs.createWriteStream(archivePath), { signal }); - await tar.x({ file: archivePath, cwd: editionDir }); - - const mmdbPath = await findExtractedDatabase(editionDir, edition.fileName); - if (!mmdbPath) { - throw new Error(`Downloaded ${edition.editionId} archive did not contain ${edition.fileName}`); - } - - return mmdbPath; -} - -/** - * Open the staged database set to ensure both City and ASN files are valid. - */ -async function validatePlannedDatabases(dbDir, plannedUpdates) { - const stagedPaths = new Map(plannedUpdates.map(update => [update.fileName, update.sourcePath])); - - const cityPath = stagedPaths.get(MAXMIND_CITY_DB) || path.join(dbDir, MAXMIND_CITY_DB); - const asnPath = stagedPaths.get(MAXMIND_ASN_DB) || path.join(dbDir, MAXMIND_ASN_DB); - - await Promise.all([ - maxmind.open(cityPath), - maxmind.open(asnPath), - ]); -} - -/** - * Atomically publish prepared databases while keeping backups for rollback. - */ -async function publishDatabases(plannedUpdates) { - const preparedUpdates = plannedUpdates.map(update => ({ - ...update, - nextPath: `${update.targetPath}.next`, - backupPath: `${update.targetPath}.bak`, - })); - - for (const update of preparedUpdates) { - await fsp.copyFile(update.sourcePath, update.nextPath); - await fsp.copyFile(update.targetPath, update.backupPath).catch(error => { - if (error.code !== 'ENOENT') { - throw error; - } - }); - } - - try { - for (const update of preparedUpdates) { - await fsp.rename(update.nextPath, update.targetPath); - } - } catch (error) { - for (const update of preparedUpdates) { - await fsp.copyFile(update.backupPath, update.targetPath).catch(() => {}); - } - throw error; - } finally { - for (const update of preparedUpdates) { - await Promise.all([ - fsp.rm(update.nextPath, { force: true }), - fsp.rm(update.backupPath, { force: true }), - ]); - } - } -} - -/** - * Recursively find the expected .mmdb file inside an extracted MaxMind archive. - */ -async function findExtractedDatabase(dir, fileName) { - const entries = await fsp.readdir(dir, { withFileTypes: true }); - - for (const entry of entries) { - const entryPath = path.join(dir, entry.name); - - if (entry.isFile() && entry.name === fileName) { - return entryPath; - } - - if (entry.isDirectory()) { - const found = await findExtractedDatabase(entryPath, fileName); - if (found) { - return found; - } - } - } - - return null; -} - -/** - * Build the MaxMind permalink download URL for an edition archive. - */ -function getDownloadUrl(editionId) { - return `https://download.maxmind.com/geoip/databases/${editionId}/download?suffix=tar.gz`; -} - -/** - * Build the Basic Auth header required by MaxMind downloads. - */ -function getAuthHeaders() { - const auth = Buffer.from(`${process.env.MAXMIND_ACCOUNT_ID}:${process.env.MAXMIND_LICENSE_KEY}`).toString('base64'); - return { - Authorization: `Basic ${auth}`, - }; -} - -/** - * Read the updater state file that records previously downloaded remote versions. - */ -async function readUpdateState(dbDir) { - try { - const content = await fsp.readFile(path.join(dbDir, STATE_FILE), 'utf8'); - return JSON.parse(content); - } catch (error) { - if (error.code === 'ENOENT') { - return {}; - } - throw error; - } -} - -/** - * Persist the updater state after a successful database publish. - */ -async function writeUpdateState(dbDir, state) { - await fsp.writeFile(path.join(dbDir, STATE_FILE), `${JSON.stringify(state, null, 2)}\n`, 'utf8'); -} - -/** - * Acquire a filesystem lock so multiple pm2 processes do not update concurrently. - */ -async function acquireUpdateLock(dbDir) { - const lockPath = path.join(dbDir, LOCK_FILE); - - try { - const handle = await fsp.open(lockPath, 'wx'); - await handle.writeFile(JSON.stringify({ - pid: process.pid, - startedAt: new Date().toISOString(), - })); - - return { - // Release the inter-process lock after this update cycle finishes. - release: async () => { - await handle.close(); - await fsp.rm(lockPath, { force: true }); - }, - }; - } catch (error) { - if (error.code !== 'EEXIST') { - throw error; - } - - const stat = await fsp.stat(lockPath).catch(() => null); - if (stat && Date.now() - stat.mtimeMs > LOCK_STALE_MS) { - await fsp.rm(lockPath, { force: true }); - return acquireUpdateLock(dbDir); - } - - logger.info('MaxMind update skipped: another process is updating databases'); - return null; - } -} diff --git a/common/peeringdb-db.js b/common/peeringdb-db.js index db7f9d18..44bb7cbd 100644 --- a/common/peeringdb-db.js +++ b/common/peeringdb-db.js @@ -1,6 +1,6 @@ // Local PeeringDB lookup: a network's self-reported peering profile (types, // website, scope, traffic, ratio, policy, IRR as-set), the exchanges it is -// on and the facilities it is in. common/caida-updater.js downloads CAIDA's +// on and the facilities it is in. common/datasets.js downloads CAIDA's // daily mirror of the full PeeringDB dump and distills it // (common/peeringdb-distill.js → buildPeeringdbIndex below); only the // compact per-ASN index ever lands in this directory. Contact data (poc, @@ -234,7 +234,7 @@ const loadDatabase = () => { loadDatabase(); -/** Reload the index after caida-updater publishes a fresh one. */ +/** Reload the index after the dataset updater publishes a fresh one. */ export const reloadPeeringdbDatabase = (reason = 'reload') => { logger.info(`🔄 Reloading PeeringDB index (${reason})`); loadDatabase(); diff --git a/common/sentry-cron.js b/common/sentry-cron.js index 7f6f83d5..bd52d3d8 100644 --- a/common/sentry-cron.js +++ b/common/sentry-cron.js @@ -1,5 +1,5 @@ -// Sentry Cron check-in wrapper for the backend's periodic jobs (MaxMind / -// CAIDA / service-status). Same gating as sentry-instrument.js: without +// Sentry Cron check-in wrapper for the backend's periodic jobs (the dataset +// updater, service-status). Same gating as sentry-instrument.js: without // SENTRY_DSN_BACKEND this is a pure passthrough and @sentry/node is never // loaded, so job modules can call it unconditionally. // diff --git a/package.json b/package.json index 6a51f6ed..4cf5a89c 100644 --- a/package.json +++ b/package.json @@ -42,6 +42,7 @@ "class-variance-authority": "^0.7.1", "clsx": "^2.1.1", "concurrently": "^10.0.5", + "croner": "^10.0.1", "dagre": "^0.8.5", "dotenv": "^17.4.2", "express": "^5.2.1", @@ -55,6 +56,7 @@ "pino": "^10.3.1", "pino-http": "^11.0.0", "pino-pretty": "^13.1.3", + "proper-lockfile": "^4.1.2", "reka-ui": "^2.10.4", "tailwind-merge": "^3.7.0", "tar": "^7.5.22", @@ -69,7 +71,8 @@ "vue-router": "^5.3.1", "vue-sonner": "^2.0.9", "whoiser": "^1.18.0", - "world-atlas": "^2.0.2" + "world-atlas": "^2.0.2", + "write-file-atomic": "^8.0.0" }, "devDependencies": { "@sentry/vite-plugin": "^5.4.0", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 26fdb3c5..71cfba8b 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -217,6 +217,9 @@ importers: concurrently: specifier: ^10.0.5 version: 10.0.5 + croner: + specifier: ^10.0.1 + version: 10.0.1 dagre: specifier: ^0.8.5 version: 0.8.5 @@ -256,6 +259,9 @@ importers: pino-pretty: specifier: ^13.1.3 version: 13.1.3 + proper-lockfile: + specifier: ^4.1.2 + version: 4.1.2 reka-ui: specifier: ^2.10.4 version: 2.10.4(vue@3.5.42) @@ -301,6 +307,9 @@ importers: world-atlas: specifier: ^2.0.2 version: 2.0.2 + write-file-atomic: + specifier: ^8.0.0 + version: 8.0.0 devDependencies: '@sentry/vite-plugin': specifier: ^5.4.0 @@ -1695,6 +1704,10 @@ packages: core-js@3.49.0: resolution: {integrity: sha512-es1U2+YTtzpwkxVLwAFdSpaIMyQaq0PBgm3YD1W3Qpsn1NAmO3KSgZfu+oGSWVu6NvLHoHCV/aYcsE5wiB7ALg==} + croner@10.0.1: + resolution: {integrity: sha512-ixNtAJndqh173VQ4KodSdJEI6nuioBWI0V1ITNKhZZsO0pEMoDxz539T4FTTbSZ/xIOSuDnzxLVRqBVSvPNE2g==} + engines: {node: '>=18.0'} + csstype@3.2.3: resolution: {integrity: sha512-z1HGKcYy2xA8AGQfwrn0PAy+PB7X/GSj3UVJW9qKyn43xWa+gl5nXmU4qqLMRzWVLFC8KusUX8T/0kCiOYpAIQ==} @@ -2485,6 +2498,9 @@ packages: resolution: {integrity: sha512-7PiHtLll5LdnKIMw100I+8xJXR5gW2QwWYkT6iJva0bXitZKa/XMrSbdmg3r2Xnaidz9Qumd0VPaMrZlF9V9sA==} engines: {node: '>=0.4.0'} + proper-lockfile@4.1.2: + resolution: {integrity: sha512-TjNPblN4BwAWMXU8s9AEz4JmQxnD1NNL7bNOY/AKUzyamc379FWASUhc/K1pL2noVb+XmZKLL68cjzLsiOAMaA==} + protobufjs@7.6.5: resolution: {integrity: sha512-/FPD0nUc9jH6rfFjji9IBqOz4pcSE3CsT1m7Ep6Mdb0LxSUMj8hgl6GomOvZzpNpAqqGaXA0P3VSrZLFzIhQrw==} engines: {node: '>=12.0.0'} @@ -2557,6 +2573,10 @@ packages: resolution: {integrity: sha512-QT7FVMXfWOYFbeRBF6nu+I6tr2Tf3u0q8RIEjNob/heKY/nh7drD/k7eeMFmSQgnTtCzLDcCu/XEnpW2wk4xCQ==} engines: {node: '>=9.3.0 || >=8.10.0 <9.0.0'} + retry@0.12.0: + resolution: {integrity: sha512-9LkiTwjUh6rT555DtE9rTX+BKByPfrMzEAtnlEtdEwr3Nkffwiihqe2bWADg+OQRjt9gl6ICdmB/ZFDCGAtSow==} + engines: {node: '>= 4'} + rolldown@1.2.5: resolution: {integrity: sha512-VD2IE5PUG4Oj8zz2VGykiYd5wbnjdIiSsNQb8Qu5B+noEp+A78mu2iVvpp27g8es14Tk9rofNs5Tku9iQCS4fA==} engines: {node: ^20.19.0 || >=22.12.0} @@ -2630,6 +2650,13 @@ packages: resolution: {integrity: sha512-ZX99e6tRweoUXqR+VBrslhda51Nh5MTQwou5tnUDgbtyM0dBgmhEDtWGP/xbKn6hqfPRHujUNwz5fy/wbbhnpw==} engines: {node: '>= 0.4'} + signal-exit@3.0.7: + resolution: {integrity: sha512-wnD2ZE+l+SPC/uoS0vXeE9L1+0wuaMqKlfz9AMUo38JsyLSBWSFcHR1Rri62LZc12vLr1gb3jl7iwQhgwpAbGQ==} + + signal-exit@4.1.0: + resolution: {integrity: sha512-bzyZ1e88w9O1iNJbKnOlvYTrWPDl46O1bG0D3XInv+9tkPrxrN8jUUTiFlDkkmKWgn1M6CfIA13SuGqOa9Korw==} + engines: {node: '>=14'} + simple-update-notifier@2.0.0: resolution: {integrity: sha512-a2B9Y0KlNXl9u/vsW6sTIu9vGEpfKu2wRV6l1H3XEas/0gUIzGzBoP/IouTcUQbm9JWZLH3COxyn03TYlFax6w==} engines: {node: '>=10'} @@ -2982,6 +3009,10 @@ packages: wrappy@1.0.2: resolution: {integrity: sha512-l4Sp/DRseor9wL6EvV2+TuQn63dMkPjZ/sp9XkghTEbV9KlPS1xUsZ3u7/IQO4wxtcFB4bgpQPRcR3QCvezPcQ==} + write-file-atomic@8.0.0: + resolution: {integrity: sha512-dYwyZredl67GyLLIHJnRM3h2PcOmN5SkcgC7eM5DPDEOEl6dLFqVrMg3F1Ea32usj4VSVZtd2H4MtKTNOf6nPg==} + engines: {node: ^22.22.2 || ^24.15.0 || >=26.0.0} + ws@8.21.0: resolution: {integrity: sha512-Vsp28b7DRcimFQvrqu2Wek3z1iYxDCWqHYB8Qsnk/S4RfaCQzPGPyBNuVjJV3cd6UiKtUtp6sNM77gWvzcCH+g==} engines: {node: '>=10.0.0'} @@ -4512,6 +4543,8 @@ snapshots: core-js@3.49.0: {} + croner@10.0.1: {} + csstype@3.2.3: {} d3-array@3.2.4: @@ -5265,6 +5298,12 @@ snapshots: progress@2.0.3: {} + proper-lockfile@4.1.2: + dependencies: + graceful-fs: 4.2.11 + retry: 0.12.0 + signal-exit: 3.0.7 + protobufjs@7.6.5: dependencies: '@protobufjs/aspromise': 1.1.2 @@ -5349,6 +5388,8 @@ snapshots: transitivePeerDependencies: - supports-color + retry@0.12.0: {} + rolldown@1.2.5: dependencies: '@oxc-project/types': 0.146.0 @@ -5477,6 +5518,10 @@ snapshots: side-channel-map: 1.0.1 side-channel-weakmap: 1.0.2 + signal-exit@3.0.7: {} + + signal-exit@4.1.0: {} + simple-update-notifier@2.0.0: dependencies: semver: 7.6.3 @@ -5767,6 +5812,10 @@ snapshots: wrappy@1.0.2: {} + write-file-atomic@8.0.0: + dependencies: + signal-exit: 4.1.0 + ws@8.21.0: {} y18n@5.0.8: {} diff --git a/tests/dataset-updater.test.js b/tests/dataset-updater.test.js new file mode 100644 index 00000000..69415d60 --- /dev/null +++ b/tests/dataset-updater.test.js @@ -0,0 +1,275 @@ +// Tests for common/dataset-updater.js — the engine, over synthetic rows in a +// temp dir: publish + state + reload, not-modified, refusals, the earlier +// updaters' state, the cross-process lock, the boot download, the schedule's +// env resolution and catch-up, and the watcher. No network: rows fetch from +// local strings. + +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { after, afterEach, describe, it } from 'node:test'; +import lockfile from 'proper-lockfile'; + +import { + STATE_FILE, LOCK_FILE, updateDataset, bootstrapDataset, readState, isPresent, + isAutoUpdateEnabled, startDatasetScheduler, DEFAULT_UPDATE_CRON, rowsMissingACheck, watchDatasets, +} from '../common/dataset-updater.js'; + +const root = fs.mkdtempSync(path.join(os.tmpdir(), 'myip-dataset-test-')); +after(() => fs.rmSync(root, { recursive: true, force: true })); + +const savedEnv = { auto: process.env.DATASET_AUTO_UPDATE, cron: process.env.DATASET_UPDATE_CRON }; +afterEach(() => { + for (const [name, value] of [['DATASET_AUTO_UPDATE', savedEnv.auto], ['DATASET_UPDATE_CRON', savedEnv.cron]]) { + if (value === undefined) delete process.env[name]; + else process.env[name] = value; + } +}); + +// A two-file row whose remote is `version`; records fetches and reloads. +const makeRow = (overrides = {}) => { + const log = { fetches: 0, reloads: [] }; + const row = { + id: 'test', + dir: fs.mkdtempSync(path.join(root, 'row-')), + files: ['a.txt', 'b.txt'], + version: 'v1', + findRemote: async () => ({ identifier: row.version }), + fetch: async ({ remote, tempDir }) => { + log.fetches++; + const staged = {}; + for (const file of row.files) { + staged[file] = path.join(tempDir, file); + fs.writeFileSync(staged[file], `${file} ${remote.identifier}\n`); + } + return staged; + }, + validate: async () => {}, + reload: async (reason) => { log.reloads.push(reason); }, + ...overrides, + }; + return { row, log }; +}; +const read = (row, file) => fs.readFileSync(path.join(row.dir, file), 'utf8'); +const state = (row) => JSON.parse(fs.readFileSync(path.join(row.dir, STATE_FILE), 'utf8')); + +describe('updateDataset', () => { + it('publishes every file, records the version and reloads', async () => { + const { row, log } = makeRow(); + assert.deepEqual(await updateDataset(row), { updated: true, identifier: 'v1' }); + assert.equal(read(row, 'a.txt'), 'a.txt v1\n'); + assert.equal(read(row, 'b.txt'), 'b.txt v1\n'); + assert.equal(state(row).identifier, 'v1'); + assert.deepEqual(log.reloads, ['auto update']); + // Only the published files, the state and no leftovers (lock released). + assert.deepEqual(fs.readdirSync(row.dir).sort(), [STATE_FILE, 'a.txt', 'b.txt']); + }); + + it('an unchanged remote only records the check', async () => { + const { row, log } = makeRow(); + await updateDataset(row); + const first = state(row); + assert.deepEqual(await updateDataset(row), { updated: false, reason: 'not-modified' }); + assert.equal(log.fetches, 1); + assert.equal(state(row).updatedAt, first.updatedAt); + assert.ok(state(row).checkedAt >= first.checkedAt); + }); + + it('a new remote version, or a missing file, is fetched again', async () => { + const { row, log } = makeRow(); + await updateDataset(row); + row.version = 'v2'; + assert.equal((await updateDataset(row)).updated, true); + assert.equal(read(row, 'b.txt'), 'b.txt v2\n'); + fs.rmSync(path.join(row.dir, 'a.txt')); + assert.equal((await updateDataset(row)).updated, true); + assert.equal(log.fetches, 3); + }); + + it('publishes nothing when validation refuses or a file is not produced', async () => { + const refused = makeRow({ validate: async () => { throw new Error('too small'); } }); + await assert.rejects(updateDataset(refused.row), /too small/); + assert.deepEqual(fs.readdirSync(refused.row.dir), []); + assert.deepEqual(refused.log.reloads, []); + + const { row } = makeRow(); + row.fetch = async ({ tempDir }) => { + const a = path.join(tempDir, 'a.txt'); + fs.writeFileSync(a, 'a'); + return { 'a.txt': a }; + }; + await assert.rejects(updateDataset(row), /fetch produced no b\.txt/); + assert.deepEqual(fs.readdirSync(row.dir), []); + }); + + it('reads an earlier updater\'s state, so an upgrade does not re-download', async () => { + const { row, log } = makeRow({ + legacyState: { file: '.old-state.json', toState: (json) => ({ identifier: json.version }) }, + }); + for (const file of row.files) fs.writeFileSync(path.join(row.dir, file), 'old'); + fs.writeFileSync(path.join(row.dir, '.old-state.json'), JSON.stringify({ version: 'v1' })); + assert.equal((await readState(row)).identifier, 'v1'); + assert.deepEqual(await updateDataset(row), { updated: false, reason: 'not-modified' }); + assert.equal(log.fetches, 0); + assert.equal(state(row).identifier, 'v1', 'carried into the new state file'); + }); + + it('answers locked while another holder has the lock', async () => { + const { row, log } = makeRow(); + const release = await lockfile.lock(row.dir, { lockfilePath: path.join(row.dir, LOCK_FILE) }); + try { + assert.deepEqual(await updateDataset(row), { updated: false, reason: 'locked' }); + assert.equal(log.fetches, 0); + } finally { + await release(); + } + assert.equal((await updateDataset(row)).updated, true); + }); +}); + +describe('bootstrapDataset', () => { + it('skips a disabled row and a present one', async () => { + const disabled = makeRow({ enabled: () => false }); + assert.equal((await bootstrapDataset(disabled.row)).status, 'disabled'); + const { row, log } = makeRow(); + await updateDataset(row); + assert.equal((await bootstrapDataset(row)).status, 'present'); + assert.equal(log.fetches, 1); + }); + + it('downloads a missing dataset and reloads it as a bootstrap', async () => { + const { row, log } = makeRow(); + assert.equal((await bootstrapDataset(row)).status, 'downloaded'); + assert.ok(isPresent(row)); + assert.deepEqual(log.reloads, ['bootstrap']); + }); + + it('waits out another holder of the lock, then loads what it published', async () => { + const { row, log } = makeRow(); + const release = await lockfile.lock(row.dir, { lockfilePath: path.join(row.dir, LOCK_FILE) }); + // The other process publishes v1 and its state, then lets go. + setTimeout(async () => { + for (const file of row.files) fs.writeFileSync(path.join(row.dir, file), `${file} v1\n`); + fs.writeFileSync(path.join(row.dir, STATE_FILE), JSON.stringify({ identifier: 'v1' })); + await release(); + }, 100); + assert.equal((await bootstrapDataset(row)).status, 'loaded'); + assert.equal(log.fetches, 0, 'no second download'); + assert.deepEqual(log.reloads, ['bootstrap']); + }); + + it('never throws: a failing or slow download is reported as failed', async () => { + const broken = makeRow({ findRemote: async () => { throw new Error('listing 503'); } }); + const failed = await bootstrapDataset(broken.row); + assert.equal(failed.status, 'failed'); + assert.match(failed.error.message, /listing 503/); + + const slow = makeRow({ + findRemote: ({ signal }) => new Promise((resolve, reject) => { + signal.addEventListener('abort', () => reject(signal.reason)); + }), + }); + assert.equal((await bootstrapDataset(slow.row, { timeoutMs: 50 })).status, 'failed'); + }); +}); + +describe('isAutoUpdateEnabled', () => { + const row = { legacyAutoUpdateEnv: 'OLD_FLAG' }; + + it('is on by default', () => { + assert.equal(isAutoUpdateEnabled(row, {}), true); + assert.equal(isAutoUpdateEnabled({}, {}), true); + }); + + it('DATASET_AUTO_UPDATE decides when set, over the legacy flag', () => { + assert.equal(isAutoUpdateEnabled(row, { DATASET_AUTO_UPDATE: 'false', OLD_FLAG: 'true' }), false); + assert.equal(isAutoUpdateEnabled(row, { DATASET_AUTO_UPDATE: 'true', OLD_FLAG: 'false' }), true); + }); + + it('otherwise an existing .env\'s legacy flag keeps its choice', () => { + assert.equal(isAutoUpdateEnabled(row, { OLD_FLAG: 'false' }), false); + assert.equal(isAutoUpdateEnabled(row, { OLD_FLAG: 'true' }), true); + }); +}); + +describe('startDatasetScheduler', () => { + it('schedules nothing when no row auto-updates', () => { + process.env.DATASET_AUTO_UPDATE = 'false'; + assert.equal(startDatasetScheduler([makeRow().row]), null); + }); + + it('runs at DATASET_UPDATE_CRON, defaulting to 04:30', () => { + assert.equal(DEFAULT_UPDATE_CRON, '30 4 * * *'); + process.env.DATASET_AUTO_UPDATE = 'true'; + process.env.DATASET_UPDATE_CRON = '15 5 * * *'; + const job = startDatasetScheduler([makeRow().row]); + try { + assert.equal(job.getPattern(), '15 5 * * *'); + const next = job.nextRun(); + assert.equal(next.getHours(), 5); + assert.equal(next.getMinutes(), 15); + } finally { + job.stop(); + } + }); +}); + +describe('rowsMissingACheck', () => { + const HOUR = 60 * 60 * 1000; + const now = Date.parse('2026-10-05T12:00:00Z'); + const withState = (stateBody) => { + const { row } = makeRow(); + if (stateBody) fs.writeFileSync(path.join(row.dir, STATE_FILE), JSON.stringify(stateBody)); + return row; + }; + + it('picks rows not checked for over a day, or never', async () => { + const fresh = withState({ checkedAt: new Date(now - 3 * HOUR).toISOString() }); + const missed = withState({ checkedAt: new Date(now - 30 * HOUR).toISOString() }); + const never = withState(null); + // A pre-engine state has no checkedAt: its publish time stands in. + const legacy = withState({ updatedAt: new Date(now - 2 * HOUR).toISOString() }); + const stale = await rowsMissingACheck([fresh, missed, never, legacy], now); + assert.deepEqual(stale, [missed, never]); + }); +}); + +describe('watchDatasets', () => { + const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); + const fast = { intervalMs: 20, settleMs: 30 }; + + it('reloads when someone else changes the files, not after its own publish', async () => { + const { row, log } = makeRow(); + await updateDataset(row); + const stop = watchDatasets([row], fast); + try { + row.version = 'v2'; + await updateDataset(row); // this process: reloads once, by itself + await sleep(150); + assert.deepEqual(log.reloads, ['auto update', 'auto update']); + + // Another process (or a hand) replaces the files. + for (const file of row.files) fs.writeFileSync(path.join(row.dir, file), `${file} by hand, longer\n`); + await sleep(150); + assert.deepEqual(log.reloads, ['auto update', 'auto update', 'file change']); + } finally { + stop(); + } + }); + + it('picks up files dropped into an empty dataset, once all are there', async () => { + const { row, log } = makeRow({ enabled: () => false }); // no credentials: hand-placed only + const stop = watchDatasets([row], fast); + try { + fs.writeFileSync(path.join(row.dir, 'a.txt'), 'a'); + await sleep(150); + assert.deepEqual(log.reloads, [], 'not while a file is still missing'); + fs.writeFileSync(path.join(row.dir, 'b.txt'), 'b'); + await sleep(150); + assert.deepEqual(log.reloads, ['file change']); + } finally { + stop(); + } + }); +}); diff --git a/tests/caida-updater.test.js b/tests/datasets.test.js similarity index 53% rename from tests/caida-updater.test.js rename to tests/datasets.test.js index 29579cf0..9df3b5ff 100644 --- a/tests/caida-updater.test.js +++ b/tests/datasets.test.js @@ -1,8 +1,10 @@ -// Tests for common/caida-updater.js — the PeeringDB row and the two -// capabilities it adds (no decompression; a distill step), the Cloudflare-key -// gate, and that a decompressing row still publishes as before. Every row is -// re-pointed at a temp dir and `fetch` is stubbed: no network, nothing -// written into the repo's dataset directories. +// Tests for common/datasets.js — the dataset rows run through the engine +// (common/dataset-updater.js): the MaxMind row (two editions published +// together; the credential gate), the PeeringDB row (no decompression; a +// distill step; the Cloudflare-key gate), a decompressing CAIDA row, and the +// rows reading the pre-engine updaters' state. Every row is re-pointed at a temp dir +// and `fetch` is stubbed: no network, nothing written into the repo's +// dataset directories. import assert from 'node:assert/strict'; import fs from 'node:fs'; @@ -11,23 +13,25 @@ import path from 'node:path'; import zlib from 'node:zlib'; import { after, afterEach, describe, it } from 'node:test'; -import { - datasets, updateDataset, bootstrapDataset, findPeeringdbDump, isDatasetEnabled, -} from '../common/caida-updater.js'; +import * as tar from 'tar'; + +import { datasets, findPeeringdbDump, maxmindIdentifier, hasMaxMindCredentials } from '../common/datasets.js'; +import { updateDataset, isRowEnabled } from '../common/dataset-updater.js'; import { PEERINGDB_FILE, readPeeringdbIndex, expandNet } from '../common/peeringdb-db.js'; import { setUpstreamUserAgent } from '../common/fetch-with-timeout.js'; const realFetch = globalThis.fetch; -const savedEnv = { key: process.env.CLOUDFLARE_API_KEY, legacy: process.env.CLOUDFLARE_API }; +const ENV_KEYS = ['CLOUDFLARE_API_KEY', 'CLOUDFLARE_API', 'MAXMIND_ACCOUNT_ID', 'MAXMIND_LICENSE_KEY']; +const savedEnv = Object.fromEntries(ENV_KEYS.map((name) => [name, process.env[name]])); afterEach(() => { globalThis.fetch = realFetch; - for (const [name, value] of [['CLOUDFLARE_API_KEY', savedEnv.key], ['CLOUDFLARE_API', savedEnv.legacy]]) { - if (value === undefined) delete process.env[name]; - else process.env[name] = value; + for (const name of ENV_KEYS) { + if (savedEnv[name] === undefined) delete process.env[name]; + else process.env[name] = savedEnv[name]; } }); -const root = fs.mkdtempSync(path.join(os.tmpdir(), 'myip-caida-test-')); +const root = fs.mkdtempSync(path.join(os.tmpdir(), 'myip-datasets-test-')); after(() => fs.rmSync(root, { recursive: true, force: true })); const row = (id) => datasets.find((dataset) => dataset.id === id); @@ -37,7 +41,7 @@ const tempRow = (id, overrides = {}) => { const reloads = []; const dataset = { ...row(id), - dbDir: fs.mkdtempSync(path.join(root, `${id}-`)), + dir: fs.mkdtempSync(path.join(root, `${id}-`)), reload: (reason) => reloads.push(reason), ...overrides, }; @@ -129,11 +133,11 @@ describe('peeringdb row', () => { it('is gated on the Cloudflare key; the CAIDA rows are not', () => { delete process.env.CLOUDFLARE_API_KEY; delete process.env.CLOUDFLARE_API; - assert.equal(isDatasetEnabled(row('peeringdb')), false); - assert.equal(isDatasetEnabled(row('as2org')), true); - assert.equal(isDatasetEnabled(row('as-rel')), true); + assert.equal(isRowEnabled(row('peeringdb')), false); + assert.equal(isRowEnabled(row('as2org')), true); + assert.equal(isRowEnabled(row('as-rel')), true); process.env.CLOUDFLARE_API_KEY = 'k'; - assert.equal(isDatasetEnabled(row('peeringdb')), true); + assert.equal(isRowEnabled(row('peeringdb')), true); }); it('downloads without decompressing, distills, publishes only the index', async () => { @@ -146,11 +150,11 @@ describe('peeringdb row', () => { assert.deepEqual(result, { updated: true, identifier: 'peeringdb_2_dump_2026_10_02.json' }); assert.deepEqual(reloads, ['auto update']); - const files = fs.readdirSync(dataset.dbDir).sort(); - assert.deepEqual(files, ['.caida-update-state.json', PEERINGDB_FILE]); - const published = fs.readFileSync(path.join(dataset.dbDir, PEERINGDB_FILE), 'utf8'); + const files = fs.readdirSync(dataset.dir).sort(); + assert.deepEqual(files, ['.dataset-state.json', PEERINGDB_FILE]); + const published = fs.readFileSync(path.join(dataset.dir, PEERINGDB_FILE), 'utf8'); assert.equal(published.includes('@'), false, 'no contact data'); - const record = expandNet(readPeeringdbIndex(path.join(dataset.dbDir, PEERINGDB_FILE)), 64500); + const record = expandNet(readPeeringdbIndex(path.join(dataset.dir, PEERINGDB_FILE)), 64500); assert.deepEqual(record.ixs, [{ name: 'AMS-IX', city: 'Amsterdam', country: 'NL', speed: 100000, rsPeer: true }]); // Same dump again → nothing to do. @@ -163,7 +167,7 @@ describe('peeringdb row', () => { findRemote: (opts) => findPeeringdbDump({ ...opts, now: new Date('2026-10-20T00:00:00Z') }), }); await assert.rejects(updateDataset(dataset), /only 1 networks/); - assert.deepEqual(fs.readdirSync(dataset.dbDir), []); + assert.deepEqual(fs.readdirSync(dataset.dir), []); assert.deepEqual(reloads, []); }); @@ -172,25 +176,12 @@ describe('peeringdb row', () => { const { dataset } = tempRow('peeringdb', { findRemote: (opts) => findPeeringdbDump({ ...opts, now: new Date('2026-10-20T00:00:00Z') }), }); - await assert.rejects(updateDataset(dataset), /peeringdb distill phase failed: truncated/); - }); - - it('bootstrap skips a present index but not a directory holding only state files', async () => { - const requested = stubFetch({}); - const { dataset } = tempRow('peeringdb'); - fs.writeFileSync(path.join(dataset.dbDir, '.caida-update-state.json'), '{}'); - assert.equal((await bootstrapDataset(dataset)).status, 'failed'); - assert.ok(requested.length > 0, 'a state file alone is not a snapshot'); - - requested.length = 0; - fs.writeFileSync(path.join(dataset.dbDir, PEERINGDB_FILE), '{}'); - assert.equal((await bootstrapDataset(dataset)).status, 'present'); - assert.deepEqual(requested, []); + await assert.rejects(updateDataset(dataset), /peeringdb distill failed: truncated/); }); }); -describe('decompressing rows', () => { - it('still download, decompress and publish the text snapshot', async () => { +describe('CAIDA rows', () => { + it('a decompressing row downloads, decompresses and publishes the text snapshot', async () => { const text = '# as2org fixture\n1|20260101|Example|US|ARIN\n'; stubFetch({ 'https://example.invalid/latest.txt.gz': [200, zlib.gzipSync(text)] }); const { dataset, reloads } = tempRow('as2org', { @@ -198,7 +189,95 @@ describe('decompressing rows', () => { validate: async () => {}, }); assert.deepEqual(await updateDataset(dataset), { updated: true, identifier: 'v1' }); - assert.equal(fs.readFileSync(path.join(dataset.dbDir, dataset.canonicalFile), 'utf8'), text); + assert.equal(fs.readFileSync(path.join(dataset.dir, dataset.files[0]), 'utf8'), text); assert.deepEqual(reloads, ['auto update']); }); + + it('read the pre-engine CAIDA state, so an upgrade does not re-download', async () => { + const requested = stubFetch({}); + const { dataset } = tempRow('as-rel', { + findRemote: async () => ({ url: 'https://example.invalid/x.bz2', identifier: '20260901.as-rel2.txt.bz2' }), + }); + fs.writeFileSync(path.join(dataset.dir, dataset.files[0]), 'snapshot'); + fs.writeFileSync(path.join(dataset.dir, '.caida-update-state.json'), + JSON.stringify({ identifier: '20260901.as-rel2.txt.bz2', updatedAt: '2026-09-06T00:32:40.420Z' })); + assert.deepEqual(await updateDataset(dataset), { updated: false, reason: 'not-modified' }); + assert.deepEqual(requested, []); + }); + + it('keep the pre-engine CAIDA_AUTO_UPDATE as their legacy schedule flag', () => { + for (const id of ['as2org', 'as-rel', 'peeringdb']) { + assert.equal(row(id).legacyAutoUpdateEnv, 'CAIDA_AUTO_UPDATE', id); + } + }); +}); + +describe('maxmind row', () => { + const MM = 'https://download.maxmind.com/geoip/databases/'; + + // A tar.gz holding `/`, as MaxMind ships it. + const archiveOf = async (file, body) => { + const dir = fs.mkdtempSync(path.join(root, 'mm-src-')); + fs.mkdirSync(path.join(dir, 'GeoLite2_20261003')); + fs.writeFileSync(path.join(dir, 'GeoLite2_20261003', file), body); + const out = path.join(dir, 'archive.tar.gz'); + await tar.c({ gzip: true, cwd: dir, file: out }, ['GeoLite2_20261003']); + return fs.readFileSync(out); + }; + + // Answers HEAD with each edition's Last-Modified, GET with its archive; + // records the requests. + const stubMaxMind = (archives, lastModified = { City: 'Fri, 02 Oct 2026 16:46:10 GMT', ASN: 'Sun, 04 Oct 2026 08:30:32 GMT' }) => { + const requested = []; + globalThis.fetch = async (url, init = {}) => { + const edition = String(url).includes('GeoLite2-City') ? 'City' : 'ASN'; + requested.push(`${init.method || 'GET'} ${edition} ${init.headers?.Authorization ? 'auth' : 'anon'}`); + if (init.method === 'HEAD') return new Response(null, { headers: { 'last-modified': lastModified[edition] } }); + return new Response(archives[edition]); + }; + return requested; + }; + + it('is gated on both credentials', () => { + delete process.env.MAXMIND_ACCOUNT_ID; + process.env.MAXMIND_LICENSE_KEY = 'key'; + assert.equal(hasMaxMindCredentials(), false); + assert.equal(isRowEnabled(row('maxmind')), false); + process.env.MAXMIND_ACCOUNT_ID = 'id'; + assert.equal(isRowEnabled(row('maxmind')), true); + }); + + it('downloads both editions with auth, extracts and publishes them together', async () => { + process.env.MAXMIND_ACCOUNT_ID = 'id'; + process.env.MAXMIND_LICENSE_KEY = 'key'; + const requested = stubMaxMind({ + City: await archiveOf('GeoLite2-City.mmdb', 'city-db'), + ASN: await archiveOf('GeoLite2-ASN.mmdb', 'asn-db'), + }); + const { dataset, reloads } = tempRow('maxmind', { validate: async () => {} }); // real mmdb check needs real files + const result = await updateDataset(dataset); + assert.equal(result.identifier, maxmindIdentifier(['Fri, 02 Oct 2026 16:46:10 GMT', 'Sun, 04 Oct 2026 08:30:32 GMT'])); + assert.equal(fs.readFileSync(path.join(dataset.dir, 'GeoLite2-City.mmdb'), 'utf8'), 'city-db'); + assert.equal(fs.readFileSync(path.join(dataset.dir, 'GeoLite2-ASN.mmdb'), 'utf8'), 'asn-db'); + assert.deepEqual(reloads, ['auto update']); + assert.ok(requested.every((entry) => entry.endsWith('auth')), 'every request authenticated'); + }); + + it('reads the pre-engine MaxMind state, so an upgrade does not re-download', async () => { + process.env.MAXMIND_ACCOUNT_ID = 'id'; + process.env.MAXMIND_LICENSE_KEY = 'key'; + const requested = stubMaxMind({}); + const { dataset } = tempRow('maxmind'); + for (const file of dataset.files) fs.writeFileSync(path.join(dataset.dir, file), 'db'); + fs.writeFileSync(path.join(dataset.dir, '.maxmind-update-state.json'), JSON.stringify({ + 'GeoLite2-City': { lastModified: 'Fri, 02 Oct 2026 16:46:10 GMT', updatedAt: '2026-10-03T03:55:12.165Z' }, + 'GeoLite2-ASN': { lastModified: 'Sun, 04 Oct 2026 08:30:32 GMT', updatedAt: '2026-10-04T16:27:03.986Z' }, + })); + assert.deepEqual(await updateDataset(dataset), { updated: false, reason: 'not-modified' }); + assert.deepEqual(requested, ['HEAD City auth', 'HEAD ASN auth'], 'only the version checks'); + }); + + it('keeps MAXMIND_AUTO_UPDATE as its legacy schedule flag', () => { + assert.equal(row('maxmind').legacyAutoUpdateEnv, 'MAXMIND_AUTO_UPDATE'); + }); });