feat: run the telemetry heartbeat and update check on one instance of the deployment #1066

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

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

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<void>;
+  stop(): Promise<void>;
+}
+
+interface LeaderGlobals {
+  jobs: ScheduledJob[];
+  /** Jobs whose start went through while leading; the others are retried on each renewal. */
+  started: Set<string>;
+  leading: boolean;
+  interval?: NodeJS.Timeout;
+  trying?: Promise<boolean>;
+  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<void> {
+  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<void> {
+  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<void> {
+  g.leading = true;
+  logger.info('scheduler: leading', { replica: replicaId() });
+  await startJobs(g);
+}
+
+async function stepDown(g: LeaderGlobals, reason: string): Promise<void> {
+  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<boolean> {
+  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<void> {
+  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<void> {
+  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<void> {
+  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<unknown> {
+  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];
+}

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<Claim> {
+  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<Claim, { claimed: true }>): Promise<void> {
+  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<void> {
   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<void> {
-  await sendOnce({ reason: 'scheduled' });
+async function tick(slot: string): Promise<void> {
+  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<void> {
   const { consent } = await effectiveConsent();
   if (consent !== 'on') {
@@ -119,7 +179,8 @@ export async function stopScheduler(): Promise<void> {
 }
 
 // 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<void> {
-  await stopScheduler();
-  await startScheduler();
+  await announceChange('telemetry');
 }

lib/version-check/sender.ts

@@ -72,15 +72,22 @@ async function scheduleNext(delayMs: number): Promise<void> {
   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<void> {
-  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<void> {
   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<void> {
   if (disabledByEnv()) {
     logger.info('version-check: scheduler not started (disabled by env)');