diff --git a/lib/__tests__/scheduler-leader.test.ts b/lib/__tests__/scheduler-leader.test.ts index 3e0943516..6b317aa86 100644 --- a/lib/__tests__/scheduler-leader.test.ts +++ b/lib/__tests__/scheduler-leader.test.ts @@ -139,6 +139,20 @@ describe('scheduler leader with a shared store', () => { expect(isSchedulerLeader()).toBe(false); }); + it('stops leading when the store lost its data', async () => { + await store.epochs.bump(INIT_EPOCH); + await consentOn(new Date(Date.now() + 5 * MIN).toISOString()); + await startSchedulerLeader(jobs); + expect(isSchedulerLeader()).toBe(true); + + // Flushed: the init epoch is gone, and the next poll notices. + resetLocalStoreForTests(); + await pollChanges(); + await pass(10_000); + expect(isSchedulerLeader()).toBe(false); + expect(log.info).toHaveBeenCalledWith('scheduler: stopped leading', expect.objectContaining({ reason: 'the shared store lost its data' })); + }); + it('stops its timers when another replica has taken the lease', async () => { const slot = new Date(Date.now() + 5 * MIN).toISOString(); await consentOn(slot); diff --git a/lib/scheduler-leader.ts b/lib/scheduler-leader.ts index 8fbd29b12..d2eccf8d2 100644 --- a/lib/scheduler-leader.ts +++ b/lib/scheduler-leader.ts @@ -1,6 +1,6 @@ import { logger } from '@/lib/logger'; import { getStore, isSharedMode, replicaId } from '@/lib/shared-store'; -import { onSharedChange } from '@/lib/shared-store/changes'; +import { onSharedChange, storeLostItsData } from '@/lib/shared-store/changes'; /** * One replica runs the background jobs (telemetry heartbeat, update check) @@ -102,11 +102,18 @@ export async function tryLead(): Promise { 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) { + if (storeLostItsData()) { + // The store was emptied under us: a lease taken there now proves + // nothing, and the jobs' stored schedule is gone. Wait for it back. holds = false; - reason = `could not reach the shared store: ${message(error)}`; + reason = 'the shared store lost its data'; + } else { + 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);