fix(scheduler): add advisory-lock leader election

backend/src/services/__tests__/scheduleExecutor.test.ts

@@ -58,11 +58,11 @@ describe('ScheduleExecutor', () => {
   });
 
   describe('initialize', () => {
-    it('should set up cron job to run every minute', () => {
+    it('should probe scheduler leadership every 15 seconds', () => {
       executor.initialize();
 
       expect(mockCron.schedule).toHaveBeenCalledWith(
-        '* * * * *',
+        '*/15 * * * * *',
         expect.any(Function)
       );
     });
@@ -78,6 +78,53 @@ describe('ScheduleExecutor', () => {
 
       consoleSpy.mockRestore();
     });
+
+    it('should skip the scheduler pass when another pod holds leadership', async () => {
+      const leaderQuery = jest.fn().mockResolvedValueOnce({
+        rows: [{ acquired: false }],
+      });
+      const leaderRelease = jest.fn();
+      (mockPool.connect as jest.Mock).mockResolvedValueOnce({
+        query: leaderQuery,
+        release: leaderRelease,
+      });
+      const processSpy = jest.spyOn(executor, 'processDueSchedules').mockResolvedValue();
+
+      executor.initialize();
+      const callback = (mockCron.schedule as jest.Mock).mock.calls[0][1] as () => Promise<void>;
+      await callback();
+
+      expect(leaderQuery).toHaveBeenCalledWith(
+        expect.stringContaining('pg_try_advisory_lock'),
+        [expect.any(Number), expect.any(Number)]
+      );
+      expect(processSpy).not.toHaveBeenCalled();
+      expect(leaderRelease).toHaveBeenCalledWith(false);
+    });
+
+    it('should hold and release leadership around one scheduler pass', async () => {
+      const leaderQuery = jest.fn()
+        .mockResolvedValueOnce({ rows: [{ acquired: true }] })
+        .mockResolvedValueOnce({ rows: [{ unlocked: true }] });
+      const leaderRelease = jest.fn();
+      (mockPool.connect as jest.Mock).mockResolvedValueOnce({
+        query: leaderQuery,
+        release: leaderRelease,
+      });
+      const processSpy = jest.spyOn(executor, 'processDueSchedules').mockResolvedValue();
+
+      executor.initialize();
+      const callback = (mockCron.schedule as jest.Mock).mock.calls[0][1] as () => Promise<void>;
+      await callback();
+
+      expect(processSpy).toHaveBeenCalledTimes(1);
+      expect(leaderQuery).toHaveBeenNthCalledWith(
+        2,
+        expect.stringContaining('pg_advisory_unlock'),
+        [expect.any(Number), expect.any(Number)]
+      );
+      expect(leaderRelease).toHaveBeenCalledWith(false);
+    });
   });
 
   describe('stop', () => {
@@ -161,7 +208,7 @@ describe('ScheduleExecutor', () => {
       expect(executor.recordExecution).toHaveBeenCalledWith(1, {
         success: true,
         transactionHash: 'abc123',
-      });
+      }, expect.any(String));
       expect(mockRelease).toHaveBeenCalled();
     });
 
@@ -421,6 +468,7 @@ describe('ScheduleExecutor', () => {
 
   describe('recordExecution', () => {
     const scheduleId = 1;
+    const claimOwner = 'pod-a:11111111-1111-4111-8111-111111111111';
 
     it('should record successful execution', async () => {
       const executionResult: ExecutionResult = {
@@ -437,7 +485,7 @@ describe('ScheduleExecutor', () => {
 
       mockScheduleService.updateAfterExecution.mockResolvedValue();
 
-      await executor.recordExecution(scheduleId, executionResult);
+      await executor.recordExecution(scheduleId, executionResult, claimOwner);
 
       expect(mockClientQuery).toHaveBeenCalledWith('BEGIN');
       expect(mockClientQuery).toHaveBeenCalledWith(
@@ -454,11 +502,12 @@ describe('ScheduleExecutor', () => {
       );
       expect(mockScheduleService.updateAfterExecution).toHaveBeenCalledWith(
         scheduleId,
-        executionResult
+        executionResult,
+        claimOwner
       );
       expect(mockClientQuery).toHaveBeenCalledWith(
         expect.stringContaining('UPDATE schedules SET locked_by = NULL'),
-        [scheduleId]
+        [scheduleId, claimOwner]
       );
       expect(mockClientQuery).toHaveBeenCalledWith('COMMIT');
       expect(mockRelease).toHaveBeenCalled();
@@ -482,7 +531,7 @@ describe('ScheduleExecutor', () => {
 
       mockScheduleService.updateAfterExecution.mockResolvedValue();
 
-      await executor.recordExecution(scheduleId, executionResult);
+      await executor.recordExecution(scheduleId, executionResult, claimOwner);
 
       expect(mockClientQuery).toHaveBeenCalledWith(
         expect.stringContaining('INSERT INTO execution_history'),
@@ -498,7 +547,8 @@ describe('ScheduleExecutor', () => {
       );
       expect(mockScheduleService.updateAfterExecution).toHaveBeenCalledWith(
         scheduleId,
-        executionResult
+        executionResult,
+        claimOwner
       );
     });
 
@@ -513,7 +563,7 @@ describe('ScheduleExecutor', () => {
         .mockRejectedValueOnce(new Error('Database error')); // INSERT fails
 
       await expect(
-        executor.recordExecution(scheduleId, executionResult)
+        executor.recordExecution(scheduleId, executionResult, claimOwner)
       ).rejects.toThrow('Database error');
 
       expect(mockClientQuery).toHaveBeenCalledWith('ROLLBACK');
@@ -529,13 +579,58 @@ describe('ScheduleExecutor', () => {
       mockClientQuery.mockRejectedValueOnce(new Error('Connection error'));
 
       await expect(
-        executor.recordExecution(scheduleId, executionResult)
+        executor.recordExecution(scheduleId, executionResult, claimOwner)
       ).rejects.toThrow('Connection error');
 
       expect(mockRelease).toHaveBeenCalled();
     });
   });
 
+  describe('per-generation claim safety', () => {
+    it('releases only the matching claim owner, never a newer pass', async () => {
+      await (executor as any).releaseClaim(19, 'pod-a:old-claim');
+      expect(mockPool.query).toHaveBeenCalledWith(
+        expect.stringContaining('WHERE id = $1 AND locked_by = $2'),
+        [19, 'pod-a:old-claim']
+      );
+    });
+
+    it('renews all actively claimed rows while a long payment is still running', async () => {
+      jest.useFakeTimers();
+      let finish!: (result: ExecutionResult) => void;
+      const processing = new Promise<ExecutionResult>((resolve) => { finish = resolve; });
+      const row = {
+        id: 19, organizationId: 1, userId: 2, frequency: 'once',
+        startDate: new Date(), endDate: null, paymentConfig: { recipients: [{}] },
+        nextRunTimestamp: new Date(), lastRunTimestamp: null,
+        status: 'active', createdAt: new Date(), updatedAt: new Date(),
+      };
+      mockClientQuery
+        .mockResolvedValueOnce({ rows: [] })
+        .mockResolvedValueOnce({ rows: [row] })
+        .mockResolvedValueOnce({ rows: [] });
+      jest.spyOn(executor, 'executeSchedule').mockReturnValue(processing);
+      jest.spyOn(executor, 'recordExecution').mockResolvedValue();
+      try {
+        const job = executor.processDueSchedules();
+        // Allow BEGIN, claimed rows, and COMMIT to settle before the tick.
+        for (let i = 0; i < 5; i++) await Promise.resolve();
+        await jest.advanceTimersByTimeAsync(60_000);
+        const renewal = (mockPool.query as jest.Mock).mock.calls.find(
+          ([sql]: [string]) => sql.includes('SET locked_at = NOW()')
+        );
+        expect(renewal).toBeDefined();
+        expect(renewal![0]).toContain('WHERE locked_by = $1');
+        expect(renewal![1][1]).toEqual([19]);
+        finish({ success: true, transactionHash: 'hash-19' });
+        await job;
+        expect(jest.getTimerCount()).toBe(0);
+      } finally {
+        jest.useRealTimers();
+      }
+    });
+  });
+
   describe('concurrent pod safety', () => {
     it('should only allow one pod to claim a schedule when two run concurrently', async () => {
       const mockSchedule = {

backend/src/services/__tests__/scheduleService.test.ts

@@ -993,6 +993,37 @@ describe('ScheduleService', () => {
 
   describe('updateAfterExecution', () => {
     const scheduleId = 1;
+    it('refuses to commit the result after claim ownership has moved', async () => {
+      const owner = 'pod-a:old-generation';
+      const row = {
+        id: scheduleId, frequency: 'once', timeOfDay: '14:30',
+        startDate: new Date('2024-01-15'), timezone: 'UTC',
+        lastRunTimestamp: null,
+      };
+      mockClientQuery
+        .mockResolvedValueOnce({ rows: [] }) // BEGIN
+        .mockResolvedValueOnce({ rows: [row] }) // SELECT existing owner
+        .mockResolvedValueOnce({ rowCount: 0, rows: [] }) // stolen claim
+        .mockResolvedValueOnce({ rows: [] }); // ROLLBACK
+
+      await expect(service.updateAfterExecution(
+        scheduleId, { success: true, transactionHash: 'hash' }, owner
+      )).rejects.toThrow('Schedule claim lost before finalizing execution');
+      const select = mockClientQuery.mock.calls.find(([sql]) =>
+        typeof sql === 'string' && sql.includes('SELECT') && sql.includes('FROM schedules')
+      );
+      const update = mockClientQuery.mock.calls.find(([sql]) =>
+        typeof sql === 'string' && sql.includes('UPDATE schedules')
+      );
+      expect(select?.[0]).toContain('AND locked_by = $2');
+      expect(select?.[1]).toEqual([scheduleId, owner]);
+      expect(update?.[0]).toContain('AND locked_by = $5');
+      expect(update?.[1][4]).toBe(owner);
+      expect(mockClientQuery).toHaveBeenCalledWith('ROLLBACK');
+      expect(mockClientQuery).not.toHaveBeenCalledWith('COMMIT');
+    });
+
+
 
     describe('successful execution', () => {
       it('should mark one-time schedule as completed after successful execution', async () => {