feat: share telemetry and update-check state across instances, with one instance id and login count #1066

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 });
   }
 }

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']);
+  });
+});

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 {

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<void> {
-  const dir = getDir();
-  if (!existsSync(dir)) await mkdir(dir, { recursive: true });
-}
-
-async function loadFile(): Promise<LoginsFile> {
-  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<LoginsFile>;
-    cache = Array.isArray(parsed?.records) ? { records: parsed.records as LoginRecord[] } : { records: [] };
+    const parsed = JSON.parse(data.toString('utf8')) as Partial<LoginsFile>;
+    return Array.isArray(parsed?.records) ? { records: parsed.records as LoginRecord[] } : { records: [] };
   } catch {
-    cache = { records: [] };
+    return { records: [] };
   }
-  return cache;
-}
-
-async function saveFile(file: LoginsFile): Promise<void> {
-  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<string
  * Record a successful login. Best-effort; never throws. Updates the
  * existing record's timestamp if the same identity has logged in before,
  * otherwise appends a new record. Records older than the retention window
- * are pruned on every write.
+ * are pruned on every write. Read and written in one step, so logins on
+ * several replicas at once all count.
  */
 export async function recordLogin(username: string, serverUrl: string): Promise<void> {
   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;

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<void> {
   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?.();

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<void> {
-  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<string> {
-  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<TelemetryStateFile> {
-  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<TelemetryStateFile>) };
+}
+
+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<TelemetryStateFile>;
-    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<TelemetryStateFile> {
+  return (await loadStateVersioned()).state;
 }
 
 export async function saveState(state: TelemetryStateFile): Promise<void> {
-  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<TelemetryStateFile> {
+  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<boolean> {
+  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),

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';

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<void> {
   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<void> {
   // 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(),

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<void> {
-  if (!existsSync(getDir())) await mkdir(getDir(), { recursive: true });
+function parse(data: Buffer): VersionCheckStateFile {
+  return { ...DEFAULTS, ...(JSON.parse(data.toString('utf8')) as Partial<VersionCheckStateFile>) };
 }
 
+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<VersionCheckStateFile> {
-  await ensureDir();
   try {
-    const raw = await readFile(statePath(), 'utf8');
-    const parsed = JSON.parse(raw) as Partial<VersionCheckStateFile>;
-    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<void> {
-  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<void> {
+  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<string, UpdateStatus> | undefined,
+  status: UpdateStatus,
+): Record<string, UpdateStatus> {
+  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);
 }

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<string, UpdateStatus>;
 }
 
 export const DEFAULT_VERSION_ENDPOINT = 'https://version.telemetry.bulwarkmail.org/';