From a0af628a6a66d1bdc87fac4100824a4bd876b84d Mon Sep 17 00:00:00 2001 From: Linus Rath <139418639+rathlinus@users.noreply.github.com> Date: Sat, 31 Oct 2026 23:25:12 +0100 Subject: [PATCH] feat: run the telemetry heartbeat and update check on one instance of the deployment #1066 --- instrumentation.node.ts | 15 +- lib/__tests__/scheduler-leader.test.ts | 236 +++++++++++++++++++++++++ lib/scheduler-leader.ts | 176 ++++++++++++++++++ lib/telemetry/sender.ts | 83 +++++++-- lib/version-check/sender.ts | 16 +- 5 files changed, 505 insertions(+), 21 deletions(-) create mode 100644 lib/__tests__/scheduler-leader.test.ts create mode 100644 lib/scheduler-leader.ts diff --git a/instrumentation.node.ts b/instrumentation.node.ts index f73112096..3d5d5d4b5 100644 --- a/instrumentation.node.ts +++ b/instrumentation.node.ts @@ -80,16 +80,19 @@ migrateLegacyAdminLayout() // or with BULWARK_TELEMETRY=on; BULWARK_TELEMETRY=off or an empty // collector URL keeps it from sending. // See https://bulwarkmail.org/docs/legal/privacy/telemetry - const { startScheduler, markProcessStart } = await import("./lib/telemetry"); + const { startScheduler, stopScheduler, markProcessStart } = await import("./lib/telemetry"); markProcessStart(); - await startScheduler(); - }) - .then(async () => { // Hourly check against version.telemetry.bulwarkmail.org. Disable with // BULWARK_UPDATE_CHECK=off or override the endpoint with // BULWARK_UPDATE_CHECK_URL. - const { startScheduler } = await import("./lib/version-check"); - await startScheduler(); + const versionCheck = await import("./lib/version-check"); + // Both run on one replica of the deployment, the scheduler leader + // (#1066); a single instance always leads. + const { startSchedulerLeader } = await import("./lib/scheduler-leader"); + await startSchedulerLeader([ + { name: "telemetry", start: startScheduler, stop: stopScheduler }, + { name: "version-check", start: versionCheck.startScheduler, stop: versionCheck.stopScheduler }, + ]); }) .catch((err) => { console.warn("Admin dashboard init skipped:", err instanceof Error ? err.message : err); diff --git a/lib/__tests__/scheduler-leader.test.ts b/lib/__tests__/scheduler-leader.test.ts new file mode 100644 index 000000000..89641891b --- /dev/null +++ b/lib/__tests__/scheduler-leader.test.ts @@ -0,0 +1,236 @@ +// @vitest-environment node +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { setTimeout as realSleep } from 'node:timers/promises'; +import { NextRequest } from 'next/server'; +import { replicaId, setStoreForTests } from '@/lib/shared-store'; +import { resetLocalStoreForTests } from '@/lib/shared-store/local'; +import { INIT_EPOCH, pollChanges, resetChangesForTests } from '@/lib/shared-store/changes'; +import { separateDiskStore, type SeparateDiskStore } from './helpers/separate-disk-store'; + +// The background jobs with several replicas (#1066): only the holder of the +// scheduler-leader lease runs them, another replica takes over when the +// holder's lease runs out and keeps the stored schedule, and a scheduled +// heartbeat is claimed with a compare-and-set so two leaders can't both send. + +const log = vi.hoisted(() => ({ info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() })); +vi.mock('@/lib/logger', () => ({ logger: log })); +vi.mock('@/lib/admin/third-party-mode', () => ({ isNoThirdPartyTrafficOn: async () => false })); +vi.mock('@/lib/telemetry/payload', () => ({ buildPayload: async () => ({ schema: '1' }), markProcessStart: () => {} })); +const sent = vi.hoisted(() => vi.fn(async (_url: string, _init?: unknown) => new Response('ok'))); +vi.mock('@/lib/telemetry/endpoint-guard', () => ({ + resolveEndpointAllowed: async () => ({ ok: true }), + fetchTelemetryTarget: (url: string, init?: unknown) => sent(url, init), +})); +vi.mock('@/lib/admin/session', () => ({ + requireAdminAuth: async () => ({ session: { sub: 'admin' } }), + getClientIP: () => '127.0.0.1', +})); +vi.mock('@/lib/admin/audit', () => ({ auditLog: async () => {} })); + +import { + LEADER_LEASE, isSchedulerLeader, resetSchedulerLeaderForTests, schedulerLeaderSettled, startSchedulerLeader, stopSchedulerLeader, +} from '@/lib/scheduler-leader'; +import { loadState, saveState, sendOnce, startScheduler, stopScheduler, updateState } from '@/lib/telemetry'; +import { POST as telemetryRoute } from '@/app/api/admin/telemetry/route'; + +const MIN = 60_000; +const COLLECTOR = 'https://collector.example.org/v1/heartbeat'; +const jobs = [{ name: 'telemetry', start: startScheduler, stop: stopScheduler }]; + +let dir: string; +let store: SeparateDiskStore; + +beforeEach(() => { + dir = mkdtempSync(path.join(tmpdir(), 'bw-leader-')); + vi.stubEnv('TELEMETRY_DATA_DIR', path.join(dir, 'telemetry')); + vi.stubEnv('BULWARK_TELEMETRY', ''); + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'setInterval', 'clearInterval', 'Date'] }); + vi.setSystemTime(new Date('2026-10-31T12:00:00.000Z')); + store = separateDiskStore(); + setStoreForTests(store); + sent.mockClear(); + log.info.mockClear(); +}); + +afterEach(async () => { + await stopSchedulerLeader(); + resetSchedulerLeaderForTests(); + await stopScheduler(); + resetChangesForTests(); + vi.useRealTimers(); + setStoreForTests(null); + resetLocalStoreForTests(); + vi.unstubAllEnvs(); + store.cleanup(); + rmSync(dir, { recursive: true, force: true }); +}); + +async function consentOn(nextScheduledAt: string | null) { + await saveState({ ...(await loadState()), consent: 'on', endpoint: COLLECTOR, nextScheduledAt }); +} + +/** Let time pass; the leader's tries and job starts do real file I/O, so wait for them too. */ +async function pass(ms: number) { + for (let passed = 0; passed < ms; passed += 10_000) { + await vi.advanceTimersByTimeAsync(Math.min(10_000, ms - passed)); + await schedulerLeaderSettled(); + } +} + +/** Another replica holding (and renewing) the lease while `ms` pass here. */ +async function otherReplicaLeads(ms: number) { + for (let passed = 0; passed < ms; passed += 10_000) { + expect(await store.leases.acquire(LEADER_LEASE, 'replica-a', 30_000)).toBe(true); + await pass(10_000); + } +} + +/** Let a heartbeat a timer set off finish its real file I/O, so "not sent" means it. */ +async function ioSettles() { + await realSleep(200); +} + +/** The heartbeat the timer set off does real I/O before it sends. */ +async function sends(times: number) { + await vi.waitFor(() => expect(sent).toHaveBeenCalledTimes(times)); +} + +describe('scheduler leader with a shared store', () => { + it('only the holder sends; when its lease runs out another replica takes over and keeps the schedule', async () => { + const slot = new Date(Date.now() + 20 * MIN).toISOString(); + await consentOn(slot); + await store.leases.acquire(LEADER_LEASE, 'replica-a', 30_000); + + await startSchedulerLeader(jobs); + expect(isSchedulerLeader()).toBe(false); + await otherReplicaLeads(5 * MIN); + await ioSettles(); + expect(isSchedulerLeader()).toBe(false); + + // replica-a is gone: its lease expires after 30 s, the next try takes it. + await pass(40_000); + expect(isSchedulerLeader()).toBe(true); + expect(log.info).toHaveBeenCalledWith('scheduler: leading', expect.objectContaining({ replica: expect.any(String) })); + expect((await loadState()).nextScheduledAt).toBe(slot); + expect(sent).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(new Date(slot).getTime() - Date.now() + 1000); + await sends(1); + }); + + it('stops its timers when it can no longer reach the store', async () => { + const slot = new Date(Date.now() + 5 * MIN).toISOString(); + await consentOn(slot); + await startSchedulerLeader(jobs); + expect(isSchedulerLeader()).toBe(true); + + store.down = true; + await pass(10_000); + expect(isSchedulerLeader()).toBe(false); + expect(log.info).toHaveBeenCalledWith('scheduler: stopped leading', expect.objectContaining({ reason: expect.stringContaining('shared store') })); + // Past the slot with the store still down: no heartbeat from here. + await pass(6 * MIN); + await ioSettles(); + expect(sent).not.toHaveBeenCalled(); + expect(isSchedulerLeader()).toBe(false); + }); + + it('stops its timers when another replica has taken the lease', async () => { + const slot = new Date(Date.now() + 5 * MIN).toISOString(); + await consentOn(slot); + await startSchedulerLeader(jobs); + expect(isSchedulerLeader()).toBe(true); + + // This replica's lease is gone (it stalled past the TTL) and replica-a took it. + await store.leases.release(LEADER_LEASE, replicaId()); + await otherReplicaLeads(6 * MIN); + await ioSettles(); + expect(isSchedulerLeader()).toBe(false); + expect(sent).not.toHaveBeenCalled(); + }); + + it('sends once when two leaders fire for the same slot, and not again after it', async () => { + const slot = new Date(Date.now() - 1000).toISOString(); + await consentOn(slot); + + const results = await Promise.all([ + sendOnce({ reason: 'scheduled', slot }), + sendOnce({ reason: 'scheduled', slot }), + ]); + expect(results.filter((r) => r.ok)).toHaveLength(1); + expect(sent).toHaveBeenCalledTimes(1); + + // The paused leader wakes up an hour later, still holding that slot. + vi.advanceTimersByTime(60 * MIN); + expect((await sendOnce({ reason: 'scheduled', slot })).ok).toBe(false); + expect(sent).toHaveBeenCalledTimes(1); + }); + + it('gives the slot back when the send fails', async () => { + const slot = new Date(Date.now() - 1000).toISOString(); + await consentOn(slot); + sent.mockResolvedValueOnce(new Response('no', { status: 500 })); + expect((await sendOnce({ reason: 'scheduled', slot })).ok).toBe(false); + expect((await loadState()).lastSentAt).toBeNull(); + expect((await sendOnce({ reason: 'scheduled', slot })).ok).toBe(true); + }); + + it('sends on "send now" from a replica that does not lead, even right after a heartbeat', async () => { + await consentOn(null); + await store.leases.acquire(LEADER_LEASE, 'replica-a', 30_000); + await startSchedulerLeader(jobs); + expect(isSchedulerLeader()).toBe(false); + await updateState((state) => { state.lastSentAt = new Date().toISOString(); }); + + const res = await telemetryRoute(new NextRequest('http://localhost/api/admin/telemetry', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ action: 'send-now' }), + })); + expect(res.status).toBe(200); + expect(sent).toHaveBeenCalledTimes(1); + }); + + it('starts the heartbeat on the leader when consent is turned on through another replica', async () => { + await store.epochs.bump(INIT_EPOCH); + await startSchedulerLeader(jobs); + expect(isSchedulerLeader()).toBe(true); + expect((await loadState()).nextScheduledAt).toBeNull(); + + // The other replica's admin route: write, then bump the epoch. + await updateState((state) => { state.consent = 'on'; state.endpoint = COLLECTOR; }); + await store.epochs.bump('telemetry'); + await pollChanges(); + + expect((await loadState()).nextScheduledAt).not.toBeNull(); + await vi.advanceTimersByTimeAsync(61 * MIN); + await sends(1); + }); +}); + +describe('scheduler leader on a single instance', () => { + it('leads at once and restarts the heartbeat when consent changes in this process', async () => { + setStoreForTests(null); + await startSchedulerLeader(jobs); + expect(isSchedulerLeader()).toBe(true); + expect(log.info).toHaveBeenCalledWith('scheduler: leading', expect.anything()); + + const res = await telemetryRoute(new NextRequest('http://localhost/api/admin/telemetry', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ action: 'set-consent', consent: 'on' }), + })); + expect(res.status).toBe(200); + expect(vi.getTimerCount()).toBe(1); + + await telemetryRoute(new NextRequest('http://localhost/api/admin/telemetry', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ action: 'set-consent', consent: 'off' }), + })); + expect(vi.getTimerCount()).toBe(0); + }); +}); diff --git a/lib/scheduler-leader.ts b/lib/scheduler-leader.ts new file mode 100644 index 000000000..ebbd10f56 --- /dev/null +++ b/lib/scheduler-leader.ts @@ -0,0 +1,176 @@ +import { logger } from '@/lib/logger'; +import { getStore, isSharedMode, replicaId } from '@/lib/shared-store'; +import { onSharedChange } from '@/lib/shared-store/changes'; + +/** + * One replica runs the background jobs (telemetry heartbeat, update check) + * for the whole deployment (#1066). It holds the `scheduler-leader` lease, + * renewing it every 10 s; every other replica tries to take it every 10 s, + * so when the leader dies or loses the store another one takes over within + * the 30 s lease plus one try. A leader that can't renew stops its jobs at + * once. The jobs keep their schedule in the shared state, so a new leader + * continues it instead of starting over. + * + * Local mode: the one process is always the leader. + * + * Everything lives on globalThis: instrumentation starts it, and the + * `telemetry` change listener may be run from a route's module copy. + */ + +export const LEADER_LEASE = 'scheduler-leader'; +export const LEADER_TTL_MS = 30_000; +export const LEADER_TRY_MS = 10_000; + +export interface ScheduledJob { + name: string; + /** Start the job's timer from its stored schedule. Must be idempotent. */ + start(): Promise; + stop(): Promise; +} + +interface LeaderGlobals { + jobs: ScheduledJob[]; + /** Jobs whose start went through while leading; the others are retried on each renewal. */ + started: Set; + leading: boolean; + interval?: NodeJS.Timeout; + trying?: Promise; + unlisten?: () => void; +} + +const KEY = Symbol.for('bulwark.schedulerLeader'); + +function globals(): LeaderGlobals { + const g = globalThis as typeof globalThis & { [KEY]?: LeaderGlobals }; + return (g[KEY] ??= { jobs: [], started: new Set(), leading: false }); +} + +function message(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +/** True while this process runs the background jobs. */ +export function isSchedulerLeader(): boolean { + return globals().leading; +} + +async function startJobs(g: LeaderGlobals): Promise { + for (const job of g.jobs) { + if (g.started.has(job.name)) continue; + try { + await job.start(); + if (g.leading) g.started.add(job.name); + } catch (error) { + logger.warn('scheduler: could not start a job; retrying', { job: job.name, error: message(error) }); + } + } +} + +async function stopJobs(g: LeaderGlobals): Promise { + g.started.clear(); + for (const job of g.jobs) { + try { + await job.stop(); + } catch (error) { + logger.warn('scheduler: could not stop a job', { job: job.name, error: message(error) }); + } + } +} + +async function becomeLeader(g: LeaderGlobals): Promise { + g.leading = true; + logger.info('scheduler: leading', { replica: replicaId() }); + await startJobs(g); +} + +async function stepDown(g: LeaderGlobals, reason: string): Promise { + g.leading = false; + logger.info('scheduler: stopped leading', { replica: replicaId(), reason }); + await stopJobs(g); +} + +/** + * One try: take or renew the lease, start the jobs on taking it, stop them + * on losing it (or on not reaching the store). Returns whether this replica + * leads afterwards. + */ +export async function tryLead(): Promise { + const g = globals(); + if (g.trying) return g.trying; + g.trying = (async () => { + let holds: boolean; + let reason = 'another replica holds the lease'; + try { + holds = await getStore().leases.acquire(LEADER_LEASE, replicaId(), LEADER_TTL_MS); + } catch (error) { + holds = false; + reason = `could not reach the shared store: ${message(error)}`; + } + if (holds && !g.leading) await becomeLeader(g); + else if (holds) await startJobs(g); + else if (g.leading) await stepDown(g, reason); + return g.leading; + })().finally(() => { + g.trying = undefined; + }); + return g.trying; +} + +/** Restart one job on the leader (its settings changed), keeping its stored schedule. */ +async function restartJob(name: string): Promise { + const g = globals(); + if (!g.leading) return; + const job = g.jobs.find((j) => j.name === name); + if (!job) return; + g.started.delete(name); + await job.stop(); + await job.start(); + if (g.leading) g.started.add(name); +} + +/** + * Start leading or trying to (instrumentation, once per process). In shared + * mode every replica calls it; only the lease holder runs `jobs`. + */ +export async function startSchedulerLeader(jobs: ScheduledJob[]): Promise { + const g = globals(); + g.jobs = jobs; + // Consent or the collector changed on some replica: the leader picks it up. + g.unlisten ??= onSharedChange('telemetry', () => restartJob('telemetry')); + if (!isSharedMode()) { + if (!g.leading) await becomeLeader(g); + return; + } + if (!g.interval) { + g.interval = setInterval(() => void tryLead(), LEADER_TRY_MS); + g.interval.unref(); + } + await tryLead(); +} + +/** Stop the jobs and give the lease up (shutdown, tests). */ +export async function stopSchedulerLeader(): Promise { + const g = globals(); + if (g.interval) clearInterval(g.interval); + g.interval = undefined; + g.unlisten?.(); + g.unlisten = undefined; + if (g.leading) { + if (isSharedMode()) await getStore().leases.release(LEADER_LEASE, replicaId()).catch(() => {}); + await stepDown(g, 'stopped'); + } +} + +/** Resolves when the current try (and the job starts it set off) is done (tests). */ +export function schedulerLeaderSettled(): Promise { + return globals().trying ?? Promise.resolve(); +} + +/** Forget everything (tests). */ +export function resetSchedulerLeaderForTests(): void { + const g = globals(); + if (g.interval) clearInterval(g.interval); + g.unlisten?.(); + const holder = globalThis as typeof globalThis & { [KEY]?: LeaderGlobals }; + delete holder[KEY]; +} diff --git a/lib/telemetry/sender.ts b/lib/telemetry/sender.ts index a06386565..fe7bcd5ec 100644 --- a/lib/telemetry/sender.ts +++ b/lib/telemetry/sender.ts @@ -1,5 +1,8 @@ import { logger } from '@/lib/logger'; -import { effectiveConsent, endpointEnabled, loadState, updateState } from './state'; +import { announceChange } from '@/lib/shared-store/changes'; +import { + effectiveConsent, endpointEnabled, loadState, loadStateVersioned, saveStateIfUnchanged, updateState, +} from './state'; import { buildPayload } from './payload'; import { fetchTelemetryTarget, resolveEndpointAllowed } from './endpoint-guard'; import { DEFAULT_ENDPOINT } from './types'; @@ -7,6 +10,8 @@ import { DEFAULT_ENDPOINT } from './types'; const DAY_MS = 24 * 60 * 60 * 1000; const JITTER_MS = 2 * 60 * 60 * 1000; // ± 2 hours const FIRST_DELAY_MS = 60 * 60 * 1000; // 1 hour after consent +/** A send this long before a scheduled slot already counts for that slot. */ +const SLOT_SLACK_MS = 60 * 1000; // On globalThis: Next loads this module once per bundle (instrumentation, // each route), and a reschedule from a route must clear the timer the @@ -22,7 +27,42 @@ function jitteredDelay(base: number): number { return Math.max(60_000, base + j); } -export async function sendOnce(opts?: { reason?: string }): Promise<{ +type Claim = { claimed: true; previous: string | null; at: string } | { claimed: false }; + +/** + * Take the scheduled send for `slot` (its nextScheduledAt): set lastSentAt + * with a compare-and-set on the state, unless a send at or after the slot is + * already recorded. Of two leaders firing for one slot (a handover after a + * long pause), only one gets it. + */ +async function claimSlot(slot: string): Promise { + const slotMs = new Date(slot).getTime(); + for (let attempt = 0; attempt < 5; attempt++) { + const { state, version } = await loadStateVersioned(); + if (version === null) return { claimed: false }; + const last = state.lastSentAt ? new Date(state.lastSentAt).getTime() : NaN; + if (!Number.isNaN(last) && last >= slotMs - SLOT_SLACK_MS) return { claimed: false }; + const at = new Date().toISOString(); + if (await saveStateIfUnchanged({ ...state, lastSentAt: at }, version)) { + return { claimed: true, previous: state.lastSentAt, at }; + } + } + return { claimed: false }; +} + +/** Give a claimed slot back after a failed send, unless another send landed meanwhile. */ +async function releaseSlot(claim: Extract): Promise { + await updateState((state) => { + if (state.lastSentAt === claim.at) state.lastSentAt = claim.previous; + }).catch(() => {}); +} + +/** + * Send one heartbeat. A scheduled send passes its `slot` and is skipped + * when another replica already sent for it; the admin's "send now" passes + * none and always sends. + */ +export async function sendOnce(opts?: { reason?: string; slot?: string }): Promise<{ ok: boolean; status?: number; error?: string; @@ -44,6 +84,15 @@ export async function sendOnce(opts?: { reason?: string }): Promise<{ return { ok: false, error: `endpoint blocked: ${guard.reason}` }; } + let claim: Claim | null = null; + if (opts?.slot) { + claim = await claimSlot(opts.slot); + if (!claim.claimed) { + logger.info('telemetry: heartbeat skipped, already sent for this slot', { slot: opts.slot }); + return { ok: false, error: 'already sent' }; + } + } + const payload = await buildPayload(); try { // Never follows redirects: the endpoint guard only vetted the URL we @@ -55,10 +104,11 @@ export async function sendOnce(opts?: { reason?: string }): Promise<{ signal: AbortSignal.timeout(5000), }); const ok = res.ok; - if (ok) { + if (ok && !claim) { const at = new Date().toISOString(); await updateState((next) => { next.lastSentAt = at; }); } + if (!ok && claim?.claimed) await releaseSlot(claim); logger.info('telemetry: heartbeat', { ok, status: res.status, reason: opts?.reason ?? 'scheduled', }); @@ -66,6 +116,7 @@ export async function sendOnce(opts?: { reason?: string }): Promise<{ } catch (err) { const msg = err instanceof Error ? err.message : String(err); logger.warn('telemetry: heartbeat failed', { error: msg }); + if (claim?.claimed) await releaseSlot(claim); return { ok: false, error: msg }; } } @@ -74,19 +125,28 @@ async function scheduleNext(delayMs: number): Promise { const t = timer(); if (t.current) clearTimeout(t.current); const at = new Date(Date.now() + delayMs).toISOString(); - await updateState((state) => { state.nextScheduledAt = at; }); - if (t.current) clearTimeout(t.current); - t.current = setTimeout(() => { void tick(); }, delayMs); + // The timer first: a state write that fails (store down) must not end + // the chain of heartbeats. + t.current = setTimeout(() => { void tick(at); }, delayMs); // Don't keep the process alive just for this. t.current.unref?.(); + try { + await updateState((state) => { state.nextScheduledAt = at; }); + } catch (err) { + logger.warn('telemetry: could not store the schedule', { error: err instanceof Error ? err.message : String(err) }); + } } -async function tick(): Promise { - await sendOnce({ reason: 'scheduled' }); +async function tick(slot: string): Promise { + try { + await sendOnce({ reason: 'scheduled', slot }); + } catch (err) { + logger.warn('telemetry: heartbeat failed', { error: err instanceof Error ? err.message : String(err) }); + } await scheduleNext(jitteredDelay(DAY_MS)); } -// Called from instrumentation. Idempotent. +// Run by the scheduler leader (lib/scheduler-leader.ts). Idempotent. export async function startScheduler(): Promise { const { consent } = await effectiveConsent(); if (consent !== 'on') { @@ -119,7 +179,8 @@ export async function stopScheduler(): Promise { } // Called when consent flips on/off or the collector URL changes via the UI. +// The replica that runs the jobs (any replica's change reaches it) restarts +// the timer from the stored state; in local mode that is this process, at once. export async function reschedule(): Promise { - await stopScheduler(); - await startScheduler(); + await announceChange('telemetry'); } diff --git a/lib/version-check/sender.ts b/lib/version-check/sender.ts index 7904dad93..324aced79 100644 --- a/lib/version-check/sender.ts +++ b/lib/version-check/sender.ts @@ -72,15 +72,22 @@ async function scheduleNext(delayMs: number): Promise { const t = timer(); if (t.current) clearTimeout(t.current); const at = new Date(Date.now() + delayMs).toISOString(); - await updateState((state) => { state.nextScheduledAt = at; }); - if (t.current) clearTimeout(t.current); + // The timer first: a state write that fails (store down) must not end + // the chain of checks. t.current = setTimeout(() => { void tick(); }, delayMs); // Don't keep the process alive just for this. t.current.unref?.(); + try { + await updateState((state) => { state.nextScheduledAt = at; }); + } catch (err) { + logger.warn('version-check: could not store the schedule', { error: err instanceof Error ? err.message : String(err) }); + } } async function tick(): Promise { - const result = await checkOnce({ reason: 'scheduled' }); + const result = await checkOnce({ reason: 'scheduled' }).catch((err: unknown) => ({ + ok: false as const, error: err instanceof Error ? err.message : String(err), + })); // The scheduler keeps running while third-party traffic is off, so turning // the switch off at runtime resumes the checks; checkOnce skips meanwhile. const skipped = result.error === NO_THIRD_PARTY_ERROR; @@ -88,7 +95,8 @@ async function tick(): Promise { await scheduleNext(delay); } -// Idempotent - safe to call from instrumentation hot-reload in dev. +// Run by the scheduler leader (lib/scheduler-leader.ts). Idempotent - safe +// to call from instrumentation hot-reload in dev. export async function startScheduler(): Promise { if (disabledByEnv()) { logger.info('version-check: scheduler not started (disabled by env)');