fix: send one heartbeat when the scheduler lead moves after a stall or a crash #1066

lib/__tests__/scheduler-leader.test.ts

@@ -245,6 +245,40 @@ describe('scheduler leader with a shared store', () => {
     expect(sent).toHaveBeenCalledTimes(1);
   });
 
+  it('sends once when the old leader stalled past its slot and the new one picked a later slot', async () => {
+    const oldSlot = new Date(Date.now() - 1000).toISOString();
+    await consentOn(oldSlot);
+    // The new leader found the slot passed and scheduled an hour out.
+    const newSlot = new Date(Date.now() + 60 * MIN).toISOString();
+    // The stalled leader wakes up and sends for its slot first.
+    expect((await sendOnce({ reason: 'scheduled', slot: oldSlot })).ok).toBe(true);
+    vi.advanceTimersByTime(61 * MIN);
+    expect((await sendOnce({ reason: 'scheduled', slot: newSlot })).ok).toBe(false);
+    expect(sent).toHaveBeenCalledTimes(1);
+  });
+
+  it('sends once when a leader died after sending but before storing the next slot', async () => {
+    const slot = new Date(Date.now() - 1000).toISOString();
+    await consentOn(slot);
+    expect((await sendOnce({ reason: 'scheduled', slot })).ok).toBe(true);
+    // nextScheduledAt still names the slot just sent; the next leader
+    // finds it passed and fires an hour later.
+    vi.advanceTimersByTime(60 * MIN);
+    expect((await sendOnce({ reason: 'scheduled', slot: new Date().toISOString() })).ok).toBe(false);
+    expect(sent).toHaveBeenCalledTimes(1);
+  });
+
+  it('still sends the next scheduled heartbeat after a manual "send now"', async () => {
+    const slot = new Date(Date.now() - 1000).toISOString();
+    await consentOn(slot);
+    expect((await sendOnce({ reason: 'scheduled', slot })).ok).toBe(true);
+    vi.advanceTimersByTime(20 * 60 * MIN);
+    expect((await sendOnce({ reason: 'admin-manual' })).ok).toBe(true);
+    vi.advanceTimersByTime(2 * 60 * MIN);
+    expect((await sendOnce({ reason: 'scheduled', slot: new Date().toISOString() })).ok).toBe(true);
+    expect(sent).toHaveBeenCalledTimes(3);
+  });
+
   it('gives the slot back when the send fails', async () => {
     const slot = new Date(Date.now() - 1000).toISOString();
     await consentOn(slot);
@@ -288,6 +322,16 @@ describe('scheduler leader with a shared store', () => {
 });
 
 describe('scheduler leader on a single instance', () => {
+  it('sends a scheduled heartbeat an hour after the last one, as before', async () => {
+    setStoreForTests(null);
+    const slot = new Date(Date.now() - 1000).toISOString();
+    await consentOn(slot);
+    expect((await sendOnce({ reason: 'scheduled', slot })).ok).toBe(true);
+    vi.advanceTimersByTime(60 * MIN);
+    expect((await sendOnce({ reason: 'scheduled', slot: new Date().toISOString() })).ok).toBe(true);
+    expect((await loadState()).lastScheduledSentAt).toBeUndefined();
+  });
+
   it('leads at once and restarts the heartbeat when consent changes in this process', async () => {
     setStoreForTests(null);
     await startSchedulerLeader(jobs);

lib/telemetry/sender.ts

@@ -1,4 +1,5 @@
 import { logger } from '@/lib/logger';
+import { isSharedMode } from '@/lib/shared-store';
 import { announceChange } from '@/lib/shared-store/changes';
 import {
   effectiveConsent, endpointEnabled, loadState, loadStateVersioned, saveStateIfUnchanged, updateState,
@@ -12,6 +13,12 @@ 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;
+/**
+ * With a shared store, a scheduled heartbeat this recent means the day's is
+ * done. A new leader computes its own slot when the old one stalled past its
+ * slot or died before storing the next one, so the slot alone can't tell.
+ */
+const SCHEDULED_GAP_MS = 20 * 60 * 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
@@ -34,7 +41,9 @@ function jitteredDelay(base: number): number {
   return Math.max(60_000, base + j);
 }
 
-type Claim = { claimed: true; previous: string | null; at: string } | { claimed: false };
+type Claim =
+  | { claimed: true; previous: string | null; previousScheduled: string | null | undefined; at: string; shared: boolean }
+  | { claimed: false };
 
 /**
  * Take the scheduled send for `slot` (its nextScheduledAt): set lastSentAt
@@ -44,14 +53,20 @@ type Claim = { claimed: true; previous: string | null; at: string } | { claimed:
  */
 async function claimSlot(slot: string): Promise<Claim> {
   const slotMs = new Date(slot).getTime();
+  const shared = isSharedMode();
   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 };
+    if (shared && state.lastScheduledSentAt) {
+      const scheduled = new Date(state.lastScheduledSentAt).getTime();
+      if (!Number.isNaN(scheduled) && Date.now() - scheduled < SCHEDULED_GAP_MS) return { claimed: false };
+    }
     const at = new Date().toISOString();
-    if (await saveStateIfUnchanged({ ...state, lastSentAt: at }, version)) {
-      return { claimed: true, previous: state.lastSentAt, at };
+    const next = shared ? { ...state, lastSentAt: at, lastScheduledSentAt: at } : { ...state, lastSentAt: at };
+    if (await saveStateIfUnchanged(next, version)) {
+      return { claimed: true, previous: state.lastSentAt, previousScheduled: state.lastScheduledSentAt, at, shared };
     }
   }
   return { claimed: false };
@@ -61,6 +76,7 @@ async function claimSlot(slot: string): Promise<Claim> {
 async function releaseSlot(claim: Extract<Claim, { claimed: true }>): Promise<void> {
   await updateState((state) => {
     if (state.lastSentAt === claim.at) state.lastSentAt = claim.previous;
+    if (claim.shared && state.lastScheduledSentAt === claim.at) state.lastScheduledSentAt = claim.previousScheduled ?? null;
   }).catch(() => {});
 }
 

lib/telemetry/types.ts

@@ -41,6 +41,11 @@ export interface TelemetryStateFile {
   endpoint: string;
   consentedAt: string | null;
   lastSentAt: string | null;
+  /**
+   * With a shared store only: when the last scheduled heartbeat went out.
+   * Unlike lastSentAt, the admin's "send now" doesn't set it.
+   */
+  lastScheduledSentAt?: string | null;
   nextScheduledAt: string | null;
 }