diff --git a/lib/__tests__/scheduler-leader.test.ts b/lib/__tests__/scheduler-leader.test.ts index 6b41f059b..8cf22aa34 100644 --- a/lib/__tests__/scheduler-leader.test.ts +++ b/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); diff --git a/lib/telemetry/sender.ts b/lib/telemetry/sender.ts index fa05b55f8..cb77372cb 100644 --- a/lib/telemetry/sender.ts +++ b/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 { 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 { async function releaseSlot(claim: Extract): Promise { 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(() => {}); } diff --git a/lib/telemetry/types.ts b/lib/telemetry/types.ts index 12c0ba472..0e8fccfe2 100644 --- a/lib/telemetry/types.ts +++ b/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; }