fix: stop running the background jobs while the shared store has lost its data #1066

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

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