From eb0005f3efa65a19ff03af3c9df3d56554a47a1d Mon Sep 17 00:00:00 2001 From: Linus Rath <139418639+rathlinus@users.noreply.github.com> Date: Sat, 31 Oct 2026 23:10:30 +0100 Subject: [PATCH] feat: share telemetry and update-check state across instances, with one instance id and login count #1066 --- app/api/admin/telemetry/route.ts | 33 ++-- ...lemetry-version-check-shared-store.test.ts | 119 +++++++++++++ lib/telemetry/index.ts | 2 +- lib/telemetry/login-tracker.ts | 80 +++------ lib/telemetry/sender.ts | 11 +- lib/telemetry/state.ts | 166 +++++++++++++----- lib/version-check/index.ts | 2 +- lib/version-check/sender.ts | 32 ++-- lib/version-check/state.ts | 90 ++++++++-- lib/version-check/types.ts | 5 + 10 files changed, 393 insertions(+), 147 deletions(-) create mode 100644 lib/__tests__/telemetry-version-check-shared-store.test.ts diff --git a/app/api/admin/telemetry/route.ts b/app/api/admin/telemetry/route.ts index d0bb0f8c1..aac9d0263 100644 --- a/app/api/admin/telemetry/route.ts +++ b/app/api/admin/telemetry/route.ts @@ -4,8 +4,7 @@ import { auditLog } from '@/lib/admin/audit'; import { logger } from '@/lib/logger'; import { effectiveConsent, - loadState, - saveState, + updateState, buildPayload, sendOnce, reschedule, @@ -13,6 +12,7 @@ import { getLoginCounts, resolveEndpointAllowed, } from '@/lib/telemetry'; +import { storedDataUnavailableResponse } from '@/lib/stored-data-errors'; /** * GET /api/admin/telemetry @@ -48,6 +48,8 @@ export async function GET(request: NextRequest) { logger.error('telemetry GET error', { error: err instanceof Error ? err.message : 'unknown', }); + const unavailable = storedDataUnavailableResponse(err); + if (unavailable) return unavailable; return NextResponse.json({ error: 'Internal server error' }, { status: 500 }); } } @@ -90,13 +92,15 @@ export async function POST(request: NextRequest) { if (body.consent !== 'on' && body.consent !== 'off') { return NextResponse.json({ error: 'consent must be "on" or "off"' }, { status: 400 }); } - const state = await loadState(); - const before = state.consent; - state.consent = body.consent; - if (body.consent === 'on' && !state.consentedAt) { - state.consentedAt = new Date().toISOString(); - } - await saveState(state); + const consent = body.consent; + const now = new Date().toISOString(); + let before: string | undefined; + // Against the stored state, so a schedule written meanwhile stays. + await updateState((state) => { + before = state.consent; + state.consent = consent; + if (consent === 'on' && !state.consentedAt) state.consentedAt = now; + }); await reschedule(); await auditLog('telemetry.set-consent', { from: before, to: body.consent }, ip); return NextResponse.json({ ok: true }); @@ -113,11 +117,12 @@ export async function POST(request: NextRequest) { return NextResponse.json({ error: check.reason }, { status: 400 }); } } - const state = await loadState(); - const before = state.endpoint; + let before: string | undefined; // Empty means no collector: heartbeats stop until a URL is set again. - state.endpoint = trimmed; - await saveState(state); + const state = await updateState((current) => { + before = current.endpoint; + current.endpoint = trimmed; + }); await reschedule(); await auditLog('telemetry.set-endpoint', { from: before, to: state.endpoint }, ip); return NextResponse.json({ ok: true, endpoint: state.endpoint }); @@ -138,6 +143,8 @@ export async function POST(request: NextRequest) { logger.error('telemetry POST error', { error: err instanceof Error ? err.message : 'unknown', }); + const unavailable = storedDataUnavailableResponse(err); + if (unavailable) return unavailable; return NextResponse.json({ error: 'Internal server error' }, { status: 500 }); } } diff --git a/lib/__tests__/telemetry-version-check-shared-store.test.ts b/lib/__tests__/telemetry-version-check-shared-store.test.ts new file mode 100644 index 000000000..895a26d72 --- /dev/null +++ b/lib/__tests__/telemetry-version-check-shared-store.test.ts @@ -0,0 +1,119 @@ +// @vitest-environment node +import { mkdtempSync, readdirSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { setStoreForTests } from '@/lib/shared-store'; +import { resetLocalStoreForTests } from '@/lib/shared-store/local'; +import type { UpdateStatus } from '@/lib/version-check/types'; +import { separateDiskStore, type SeparateDiskStore } from './helpers/separate-disk-store'; + +// Telemetry and the version check with a shared store (#1066): one instance +// id, one login list and one state for the deployment. + +vi.mock('@/lib/admin/third-party-mode', () => ({ isNoThirdPartyTrafficOn: async () => false })); + +let disks: string; +let store: SeparateDiskStore; + +function onReplica(name: 'a' | 'b') { + vi.stubEnv('TELEMETRY_DATA_DIR', path.join(disks, name, 'telemetry')); + vi.stubEnv('VERSION_CHECK_DATA_DIR', path.join(disks, name, 'version-check')); +} + +beforeEach(() => { + disks = mkdtempSync(path.join(tmpdir(), 'bw-telemetry-replicas-')); + store = separateDiskStore(); + setStoreForTests(store); + vi.stubEnv('BULWARK_TELEMETRY', ''); + onReplica('a'); +}); + +afterEach(() => { + setStoreForTests(null); + resetLocalStoreForTests(); + vi.unstubAllEnvs(); + vi.unstubAllGlobals(); + store.cleanup(); + rmSync(disks, { recursive: true, force: true }); +}); + +describe('telemetry with a shared store', () => { + it('gives every replica one instance id, even when they ask at once', async () => { + const { getInstanceId } = await import('@/lib/telemetry/state'); + const [first, second] = await Promise.all([getInstanceId(), getInstanceId()]); + expect(second).toBe(first); + onReplica('b'); + expect(await getInstanceId()).toBe(first); + }); + + it('counts logins on every replica together and reports the same numbers everywhere', async () => { + // A fresh module copy per replica: nothing carried over in memory. + const tracker = async () => { + vi.resetModules(); + return import('@/lib/telemetry/login-tracker'); + }; + const a = await tracker(); + await Promise.all([ + a.recordLogin('alice@example.org', 'https://mail.example.org'), + a.recordLogin('bob@example.org', 'https://mail.example.org'), + ]); + onReplica('b'); + const b = await tracker(); + await b.recordLogin('carol@example.org', 'https://mail.example.org'); + expect(await b.getLoginCounts()).toEqual({ total: 3, active7d: 3 }); + expect(await a.getLoginCounts()).toEqual({ total: 3, active7d: 3 }); + }); + + it('shows on replica B the consent set on replica A', async () => { + const { updateState, effectiveConsent } = await import('@/lib/telemetry/state'); + await updateState((state) => { state.consent = 'on'; }); + onReplica('b'); + expect((await effectiveConsent()).consent).toBe('on'); + }); +}); + +function status(current: string, latest: string): UpdateStatus { + return { + schema: 1, current, latest, updateAvailable: current !== latest, severity: current !== latest ? 'normal' : 'none', + url: null, advisory: null, checkedAt: new Date().toISOString(), + }; +} + +describe('version check with a shared store', () => { + it('keeps each running version\'s status during a rolling update', async () => { + vi.stubEnv('BULWARK_UPDATE_CHECK_URL', 'https://version.example/'); + const answer = (current: string) => vi.stubGlobal('fetch', vi.fn(async () => new Response(JSON.stringify(status(current, '1.21.0'))))); + const { checkOnce, loadState, freshStatus } = await import('@/lib/version-check'); + + vi.stubEnv('NEXT_PUBLIC_APP_VERSION', '1.20.0'); + answer('1.20.0'); + expect((await checkOnce()).ok).toBe(true); + + onReplica('b'); + vi.stubEnv('NEXT_PUBLIC_APP_VERSION', '1.21.0'); + answer('1.21.0'); + expect((await checkOnce()).ok).toBe(true); + expect(freshStatus(await loadState())).toMatchObject({ current: '1.21.0', updateAvailable: false }); + + // A replica still on the old version keeps showing its banner. + onReplica('a'); + vi.stubEnv('NEXT_PUBLIC_APP_VERSION', '1.20.0'); + expect(freshStatus(await loadState())).toMatchObject({ current: '1.20.0', updateAvailable: true }); + }); +}); + +describe('telemetry and version check on a single instance', () => { + it('keep today\'s files', async () => { + setStoreForTests(null); + const { getInstanceId, loadState } = await import('@/lib/telemetry/state'); + const { recordLogin } = await import('@/lib/telemetry/login-tracker'); + const versionCheck = await import('@/lib/version-check'); + await getInstanceId(); + await loadState(); + await recordLogin('alice@example.org', 'https://mail.example.org'); + await versionCheck.saveState({ ...(await versionCheck.loadState()), lastCheckedAt: 'x' }); + expect(readdirSync(path.join(disks, 'a', 'telemetry')).sort()).toEqual(['.telemetry-id', 'logins.json', 'state.json']); + expect(readdirSync(path.join(disks, 'a', 'version-check'))).toEqual(['state.json']); + }); +}); diff --git a/lib/telemetry/index.ts b/lib/telemetry/index.ts index 1717c1bf7..2cc956d4c 100644 --- a/lib/telemetry/index.ts +++ b/lib/telemetry/index.ts @@ -1,7 +1,7 @@ export { startScheduler, stopScheduler, reschedule, sendOnce } from './sender'; export { buildPayload, markProcessStart } from './payload'; export { - loadState, saveState, getInstanceId, effectiveConsent, + loadState, saveState, updateState, getInstanceId, effectiveConsent, } from './state'; export { recordLogin, getLoginCounts } from './login-tracker'; export { diff --git a/lib/telemetry/login-tracker.ts b/lib/telemetry/login-tracker.ts index 3534ca365..96f776e81 100644 --- a/lib/telemetry/login-tracker.ts +++ b/lib/telemetry/login-tracker.ts @@ -1,9 +1,6 @@ -import { readFile, writeFile, mkdir, rename } from 'node:fs/promises'; -import { existsSync } from 'node:fs'; -import path from 'node:path'; import { createHmac } from 'node:crypto'; import { logger } from '@/lib/logger'; -import { getInstanceId } from './state'; +import { getInstanceId, telemetryDocs } from './state'; // We never store usernames or server URLs in the clear. Each login is // recorded as HMAC-SHA256(username + '@' + serverUrl, instance_id), so the @@ -22,39 +19,17 @@ interface LoginsFile { const SEVEN_DAYS_MS = 7 * 24 * 60 * 60 * 1000; const RETENTION_MS = 90 * 24 * 60 * 60 * 1000; -let cache: LoginsFile | null = null; +/** `logins.json` in the telemetry collection (#1066): one list for every replica. */ +const LOGINS_DOC = 'logins.json'; -function getDir(): string { - return process.env.TELEMETRY_DATA_DIR || path.join(process.cwd(), 'data', 'telemetry'); -} - -function loginsPath(): string { - return path.join(getDir(), 'logins.json'); -} - -async function ensureDir(): Promise { - const dir = getDir(); - if (!existsSync(dir)) await mkdir(dir, { recursive: true }); -} - -async function loadFile(): Promise { - if (cache) return cache; +function parse(data: Buffer | null): LoginsFile { + if (!data) return { records: [] }; try { - const raw = await readFile(loginsPath(), 'utf8'); - const parsed = JSON.parse(raw) as Partial; - cache = Array.isArray(parsed?.records) ? { records: parsed.records as LoginRecord[] } : { records: [] }; + const parsed = JSON.parse(data.toString('utf8')) as Partial; + return Array.isArray(parsed?.records) ? { records: parsed.records as LoginRecord[] } : { records: [] }; } catch { - cache = { records: [] }; + return { records: [] }; } - return cache; -} - -async function saveFile(file: LoginsFile): Promise { - await ensureDir(); - cache = file; - const tmp = loginsPath() + '.tmp'; - await writeFile(tmp, JSON.stringify(file), 'utf8'); - await rename(tmp, loginsPath()); } function normalizeServer(serverUrl: string): string { @@ -71,29 +46,31 @@ async function hashIdentity(username: string, serverUrl: string): Promise { if (!username || !serverUrl) return; try { const id = await hashIdentity(username, serverUrl); - const file = await loadFile(); const now = new Date().toISOString(); - const cutoff = Date.now() - RETENTION_MS; - const next: LoginRecord[] = []; - let updated = false; - for (const rec of file.records) { - const ts = new Date(rec.lastLoginAt).getTime(); - if (Number.isNaN(ts) || ts < cutoff) continue; - if (rec.id === id) { - next.push({ id, lastLoginAt: now }); - updated = true; - } else { - next.push(rec); + await telemetryDocs().update(LOGINS_DOC, (current) => { + const cutoff = Date.now() - RETENTION_MS; + const next: LoginRecord[] = []; + let updated = false; + for (const rec of parse(current).records) { + const ts = new Date(rec.lastLoginAt).getTime(); + if (Number.isNaN(ts) || ts < cutoff) continue; + if (rec.id === id) { + next.push({ id, lastLoginAt: now }); + updated = true; + } else { + next.push(rec); + } } - } - if (!updated) next.push({ id, lastLoginAt: now }); - await saveFile({ records: next }); + if (!updated) next.push({ id, lastLoginAt: now }); + return JSON.stringify({ records: next }); + }); } catch (err) { logger.debug?.('telemetry: recordLogin failed', { error: err instanceof Error ? err.message : String(err), @@ -103,11 +80,12 @@ export async function recordLogin(username: string, serverUrl: string): Promise< /** * Total distinct accounts seen in the 90-day retention window, plus those - * with a login in the last 7 days. + * with a login in the last 7 days. Read from the store each time, so every + * replica reports the same numbers. */ export async function getLoginCounts(): Promise<{ total: number; active7d: number }> { try { - const file = await loadFile(); + const file = parse((await telemetryDocs().get(LOGINS_DOC))?.data ?? null); const cutoff = Date.now() - RETENTION_MS; const sevenAgo = Date.now() - SEVEN_DAYS_MS; let total = 0; diff --git a/lib/telemetry/sender.ts b/lib/telemetry/sender.ts index f719f1ccc..f47b5a374 100644 --- a/lib/telemetry/sender.ts +++ b/lib/telemetry/sender.ts @@ -1,5 +1,5 @@ import { logger } from '@/lib/logger'; -import { effectiveConsent, endpointEnabled, loadState, saveState } from './state'; +import { effectiveConsent, endpointEnabled, loadState, updateState } from './state'; import { buildPayload } from './payload'; import { fetchTelemetryTarget, resolveEndpointAllowed } from './endpoint-guard'; import { DEFAULT_ENDPOINT } from './types'; @@ -49,9 +49,8 @@ export async function sendOnce(opts?: { reason?: string }): Promise<{ }); const ok = res.ok; if (ok) { - const next = await loadState(); - next.lastSentAt = new Date().toISOString(); - await saveState(next); + const at = new Date().toISOString(); + await updateState((next) => { next.lastSentAt = at; }); } logger.info('telemetry: heartbeat', { ok, status: res.status, reason: opts?.reason ?? 'scheduled', @@ -67,9 +66,7 @@ export async function sendOnce(opts?: { reason?: string }): Promise<{ async function scheduleNext(delayMs: number): Promise { if (currentTimer) clearTimeout(currentTimer); const at = new Date(Date.now() + delayMs).toISOString(); - const state = await loadState(); - state.nextScheduledAt = at; - await saveState(state); + await updateState((state) => { state.nextScheduledAt = at; }); currentTimer = setTimeout(() => { void tick(); }, delayMs); // Don't keep the process alive just for this. currentTimer.unref?.(); diff --git a/lib/telemetry/state.ts b/lib/telemetry/state.ts index cae5ccd96..cde8620b6 100644 --- a/lib/telemetry/state.ts +++ b/lib/telemetry/state.ts @@ -1,8 +1,8 @@ -import { readFile, writeFile, mkdir, rename } from 'node:fs/promises'; -import { existsSync } from 'node:fs'; import path from 'node:path'; import { randomUUID } from 'node:crypto'; import { logger } from '@/lib/logger'; +import { getStore, type CollectionOptions } from '@/lib/shared-store'; +import { isStoreUnavailable } from '@/lib/stored-data-errors'; import type { TelemetryStateFile, ConsentState } from './types'; import { DEFAULT_ENDPOINT } from './types'; import { isNoThirdPartyTrafficOn } from '@/lib/admin/third-party-mode'; @@ -12,8 +12,26 @@ export function getDir(): string { path.join(process.cwd(), 'data', 'telemetry'); } -function statePath(): string { return path.join(getDir(), 'state.json'); } -function idPath(): string { return path.join(getDir(), '.telemetry-id'); } +/** + * Telemetry's documents (#1066): `state.json`, `logins.json` and the + * instance id `.telemetry-id`. Local mode: today's files in the telemetry + * dir. Shared mode: one copy for the whole deployment, so every replica + * reports as the same instance with the same counts. + */ +export const TELEMETRY_COLLECTION: CollectionOptions = { + local: { + dir: () => getDir(), + idOf: (relative) => (['state.json', 'logins.json', '.telemetry-id'].includes(relative) ? relative : null), + }, +}; + +export const STATE_DOC = 'state.json'; +const ID_DOC = '.telemetry-id'; +const UUID = /^[0-9a-f-]{36}$/i; + +export function telemetryDocs() { + return getStore().collection('telemetry', TELEMETRY_COLLECTION); +} function envOverride(): ConsentState | null { const v = (process.env.BULWARK_TELEMETRY ?? '').toLowerCase(); @@ -26,21 +44,29 @@ function envOverride(): ConsentState | null { return null; } -export async function ensureDir(): Promise { - if (!existsSync(getDir())) await mkdir(getDir(), { recursive: true }); -} - +/** + * The instance id, created once: of several replicas (or module copies) + * asking at once, every one gets the id the first one stored. + */ export async function getInstanceId(): Promise { - await ensureDir(); - try { - const id = (await readFile(idPath(), 'utf8')).trim(); - if (/^[0-9a-f-]{36}$/i.test(id)) return id; - } catch { /* generate fresh */ } - const fresh = randomUUID(); - const tmp = idPath() + '.tmp'; - await writeFile(tmp, fresh, 'utf8'); - await rename(tmp, idPath()); - return fresh; + const docs = telemetryDocs(); + for (let attempt = 0; attempt < 3; attempt++) { + let current; + try { + current = await docs.get(ID_DOC); + } catch (error) { + if (isStoreUnavailable(error)) throw error; + current = null; // unreadable: make a fresh one, as before + } + const id = current?.data.toString('utf8').trim(); + if (id && UUID.test(id)) return id; + const fresh = randomUUID(); + const result = await docs.put(ID_DOC, fresh, { ifVersion: current ? current.version : null }); + if (result.ok) return fresh; + const won = result.current?.data.toString('utf8').trim(); + if (won && UUID.test(won)) return won; + } + throw new Error('telemetry: could not store an instance id'); } // Default consent is 'off' - telemetry is opt-in. Admins can enable it during @@ -56,36 +82,92 @@ const DEFAULTS: TelemetryStateFile = { nextScheduledAt: null, }; -export async function loadState(): Promise { - await ensureDir(); +function serialize(state: TelemetryStateFile): string { + return JSON.stringify(state, null, 2); +} + +function parse(data: Buffer): TelemetryStateFile { + return { ...DEFAULTS, ...(JSON.parse(data.toString('utf8')) as Partial) }; +} + +function readFailed(err: unknown): TelemetryStateFile { + // Unreadable is not missing: answer with the defaults (so nothing is + // sent on a guess) but leave the file alone, or a passing read error + // would switch the admin's consent off for good. + logger.warn('telemetry: state read failed', { + error: err instanceof Error ? err.message : String(err), + }); + return { ...DEFAULTS }; +} + +/** + * The state with its stored version (null when it couldn't be read), for a + * compare-and-set. Throws StoreUnavailableError when the shared store is down. + */ +export async function loadStateVersioned(): Promise<{ state: TelemetryStateFile; version: string | null }> { + const docs = telemetryDocs(); + let current; try { - const raw = await readFile(statePath(), 'utf8'); - const parsed = JSON.parse(raw) as Partial; - return { ...DEFAULTS, ...parsed }; + current = await docs.get(STATE_DOC); } catch (err) { - if ((err as NodeJS.ErrnoException).code !== 'ENOENT') { - // Unreadable is not missing: answer with the defaults (so nothing is - // sent on a guess) but leave the file alone, or a passing read error - // would switch the admin's consent off for good. - logger.warn('telemetry: state read failed', { - error: err instanceof Error ? err.message : String(err), - }); - return { ...DEFAULTS }; - } - // First-ever load on a fresh install: persist the default-off state so the - // instance id and (lack of) consent are stable across restarts. The admin - // can opt in later via the UI or the BULWARK_TELEMETRY env var. - const fresh: TelemetryStateFile = { ...DEFAULTS }; - await saveState(fresh); - return fresh; + if (isStoreUnavailable(err)) throw err; + return { state: readFailed(err), version: null }; } + if (current) { + try { + return { state: parse(current.data), version: current.version }; + } catch (err) { + return { state: readFailed(err), version: null }; + } + } + // First-ever load on a fresh install: persist the default-off state so the + // instance id and (lack of) consent are stable across restarts. The admin + // can opt in later via the UI or the BULWARK_TELEMETRY env var. + const fresh: TelemetryStateFile = { ...DEFAULTS }; + const written = await docs.put(STATE_DOC, serialize(fresh), { ifVersion: null }); + if (written.ok) return { state: fresh, version: written.version }; + // Another replica created it first. + return written.current + ? { state: parse(written.current.data), version: written.current.version } + : { state: fresh, version: null }; +} + +export async function loadState(): Promise { + return (await loadStateVersioned()).state; } export async function saveState(state: TelemetryStateFile): Promise { - await ensureDir(); - const tmp = statePath() + '.tmp'; - await writeFile(tmp, JSON.stringify(state, null, 2), 'utf8'); - await rename(tmp, statePath()); + await telemetryDocs().put(STATE_DOC, serialize(state)); +} + +/** + * Change some fields of the stored state without undoing a write that + * landed meanwhile (another replica scheduling, the admin changing consent). + * `change` may run more than once. Returns the state written. + */ +export async function updateState(change: (state: TelemetryStateFile) => void): Promise { + let next: TelemetryStateFile = { ...DEFAULTS }; + await telemetryDocs().update(STATE_DOC, (current) => { + let state: TelemetryStateFile; + try { + state = current ? parse(current) : { ...DEFAULTS }; + } catch { + state = { ...DEFAULTS }; // a corrupt file: this write replaces it + } + change(state); + next = state; + return serialize(state); + }); + return next; +} + +/** + * Write `state` only if the stored state is still at `version` (from + * loadStateVersioned). False when something else wrote first. + */ +export async function saveStateIfUnchanged(state: TelemetryStateFile, version: string | null): Promise { + if (version === null) return false; + return (await telemetryDocs().put(STATE_DOC, serialize(state), { ifVersion: version })).ok; } // Effective consent: the no-third-party mode wins over everything (#903), diff --git a/lib/version-check/index.ts b/lib/version-check/index.ts index efc5e213c..86d5d9a16 100644 --- a/lib/version-check/index.ts +++ b/lib/version-check/index.ts @@ -1,6 +1,6 @@ export { startScheduler, stopScheduler, checkOnce } from './sender'; export { - loadState, saveState, effectiveEndpoint, disabledByEnv, + loadState, saveState, updateState, effectiveEndpoint, disabledByEnv, getCurrentVersion, isStatusStale, freshStatus, } from './state'; export { fetchStatus, parseStatus } from './fetcher'; diff --git a/lib/version-check/sender.ts b/lib/version-check/sender.ts index ec05be376..34009c652 100644 --- a/lib/version-check/sender.ts +++ b/lib/version-check/sender.ts @@ -1,7 +1,9 @@ import { logger } from '@/lib/logger'; import { - disabledByEnv, effectiveEndpoint, getCurrentVersion, isStatusStale, loadState, saveState, + disabledByEnv, effectiveEndpoint, freshStatus, getCurrentVersion, isStatusStale, loadState, saveState, + updateState, withVersionStatus, } from './state'; +import { isSharedMode } from '@/lib/shared-store'; import { fetchStatus } from './fetcher'; import type { UpdateStatus } from './types'; import { isNoThirdPartyTrafficOn } from '@/lib/admin/third-party-mode'; @@ -39,13 +41,15 @@ export async function checkOnce(opts?: { reason?: string }): Promise<{ const now = new Date().toISOString(); const result = await fetchStatus(endpoint, current); - const next = await loadState(); - next.lastCheckedAt = now; - if (result.ok) { - next.lastSuccessAt = now; - next.status = result.status; - } - await saveState(next); + const shared = isSharedMode(); + await updateState((next) => { + next.lastCheckedAt = now; + if (result.ok) { + next.lastSuccessAt = now; + next.status = result.status; + if (shared) next.statuses = withVersionStatus(next.statuses, result.status); + } + }); logger.info('version-check: ran', { ok: result.ok, @@ -60,9 +64,7 @@ export async function checkOnce(opts?: { reason?: string }): Promise<{ async function scheduleNext(delayMs: number): Promise { if (currentTimer) clearTimeout(currentTimer); const at = new Date(Date.now() + delayMs).toISOString(); - const state = await loadState(); - state.nextScheduledAt = at; - await saveState(state); + await updateState((state) => { state.nextScheduledAt = at; }); currentTimer = setTimeout(() => { void tick(); }, delayMs); // Don't keep the process alive just for this. currentTimer.unref?.(); @@ -92,8 +94,12 @@ export async function startScheduler(): Promise { // version we no longer run, and its schedule would keep that stale answer // (e.g. "update available" for the version we just upgraded from) on screen // for up to an hour. Drop it and check promptly instead (#913). - const stale = isStatusStale(state.status); - if (stale) { + // With a shared store, replicas on another version may still be serving + // that status: keep it, and only check promptly when there is none for + // this version yet. + const shared = isSharedMode(); + const stale = shared ? !freshStatus(state) && state.status !== null : isStatusStale(state.status); + if (stale && !shared) { logger.info('version-check: discarding status from a previous version', { cached: state.status?.current ?? null, running: getCurrentVersion(), diff --git a/lib/version-check/state.ts b/lib/version-check/state.ts index 1797cb3e9..0bd5c19b6 100644 --- a/lib/version-check/state.ts +++ b/lib/version-check/state.ts @@ -1,7 +1,7 @@ -import { readFile, writeFile, mkdir, rename } from 'node:fs/promises'; -import { existsSync } from 'node:fs'; import path from 'node:path'; import { logger } from '@/lib/logger'; +import { getStore, type CollectionOptions } from '@/lib/shared-store'; +import { isStoreUnavailable } from '@/lib/stored-data-errors'; import type { UpdateStatus, VersionCheckStateFile } from './types'; import { DEFAULT_VERSION_ENDPOINT } from './types'; @@ -10,7 +10,23 @@ function getDir(): string { path.join(process.cwd(), 'data', 'version-check'); } -function statePath(): string { return path.join(getDir(), 'state.json'); } +/** + * `state.json` (#1066). Local mode: today's file in the version-check dir. + * Shared mode: one state for the deployment, written by the replica that + * runs the scheduled check and read by all. + */ +export const VERSION_CHECK_COLLECTION: CollectionOptions = { + local: { + dir: () => getDir(), + idOf: (relative) => (relative === 'state.json' ? relative : null), + }, +}; + +export const STATE_DOC = 'state.json'; + +function docs() { + return getStore().collection('version-check', VERSION_CHECK_COLLECTION); +} const DEFAULTS: VersionCheckStateFile = { endpoint: DEFAULT_VERSION_ENDPOINT, @@ -20,31 +36,44 @@ const DEFAULTS: VersionCheckStateFile = { status: null, }; -export async function ensureDir(): Promise { - if (!existsSync(getDir())) await mkdir(getDir(), { recursive: true }); +function parse(data: Buffer): VersionCheckStateFile { + return { ...DEFAULTS, ...(JSON.parse(data.toString('utf8')) as Partial) }; } +function serialize(state: VersionCheckStateFile): string { + return JSON.stringify(state, null, 2); +} + +/** The stored state, or the defaults. Throws StoreUnavailableError when the shared store is down. */ export async function loadState(): Promise { - await ensureDir(); try { - const raw = await readFile(statePath(), 'utf8'); - const parsed = JSON.parse(raw) as Partial; - return { ...DEFAULTS, ...parsed }; + const current = await docs().get(STATE_DOC); + return current ? parse(current.data) : { ...DEFAULTS }; } catch (err) { - if ((err as NodeJS.ErrnoException).code !== 'ENOENT') { - logger.warn('version-check: state read failed', { - error: err instanceof Error ? err.message : String(err), - }); - } + if (isStoreUnavailable(err)) throw err; + logger.warn('version-check: state read failed', { + error: err instanceof Error ? err.message : String(err), + }); return { ...DEFAULTS }; } } export async function saveState(state: VersionCheckStateFile): Promise { - await ensureDir(); - const tmp = statePath() + '.tmp'; - await writeFile(tmp, JSON.stringify(state, null, 2), 'utf8'); - await rename(tmp, statePath()); + await docs().put(STATE_DOC, serialize(state)); +} + +/** Change some fields against what is stored now. `change` may run more than once. */ +export async function updateState(change: (state: VersionCheckStateFile) => void): Promise { + await docs().update(STATE_DOC, (current) => { + let state: VersionCheckStateFile; + try { + state = current ? parse(current) : { ...DEFAULTS }; + } catch { + state = { ...DEFAULTS }; + } + change(state); + return serialize(state); + }); } export function disabledByEnv(): boolean { @@ -78,6 +107,29 @@ export function isStatusStale( return status.current !== current; } +/** How many running versions keep their own status (a rolling update runs two). */ +const KEPT_VERSIONS = 4; + +/** + * The status for the version this replica runs. With a shared store, + * replicas on two versions during a rolling update share one state; each + * reads the status checked for its own version from `statuses`, so the + * banner doesn't come and go as the other version's checks land. + */ export function freshStatus(state: VersionCheckStateFile): UpdateStatus | null { - return isStatusStale(state.status) ? null : state.status; + if (!isStatusStale(state.status)) return state.status; + const own = state.statuses?.[getCurrentVersion()]; + return own && !isStatusStale(own) ? own : null; +} + +/** `statuses` with `status` added for its version, keeping the newest few. */ +export function withVersionStatus( + statuses: Record | undefined, + status: UpdateStatus, +): Record { + const next = { ...(statuses ?? {}), [status.current]: status }; + const newest = Object.entries(next) + .sort(([, a], [, b]) => b.checkedAt.localeCompare(a.checkedAt)) + .slice(0, KEPT_VERSIONS); + return Object.fromEntries(newest); } diff --git a/lib/version-check/types.ts b/lib/version-check/types.ts index b624626cc..bf70e69d8 100644 --- a/lib/version-check/types.ts +++ b/lib/version-check/types.ts @@ -20,6 +20,11 @@ export interface VersionCheckStateFile { lastSuccessAt: string | null; nextScheduledAt: string | null; status: UpdateStatus | null; + /** + * With a shared store only: the last status per running version, so + * replicas on two versions during a rolling update each keep their own. + */ + statuses?: Record; } export const DEFAULT_VERSION_ENDPOINT = 'https://version.telemetry.bulwarkmail.org/';