diff --git a/backend/src/services/__tests__/scheduleExecutor.test.ts b/backend/src/services/__tests__/scheduleExecutor.test.ts index 75c4229e..0c668928 100644 --- a/backend/src/services/__tests__/scheduleExecutor.test.ts +++ b/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; + 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; + 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((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 = { diff --git a/backend/src/services/__tests__/scheduleService.test.ts b/backend/src/services/__tests__/scheduleService.test.ts index e5b30ba9..c4cea79d 100644 --- a/backend/src/services/__tests__/scheduleService.test.ts +++ b/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 () => { diff --git a/backend/src/services/scheduleExecutor.ts b/backend/src/services/scheduleExecutor.ts index f34c242a..5a91942f 100644 --- a/backend/src/services/scheduleExecutor.ts +++ b/backend/src/services/scheduleExecutor.ts @@ -6,31 +6,81 @@ import { scheduleService } from './scheduleService.js'; import type { Schedule, ExecutionResult, PaymentRecipient } from '../types/schedule.js'; import { Operation, Asset, Memo, Keypair } from '@stellar/stellar-sdk'; import os from 'node:os'; +import { randomUUID } from 'node:crypto'; +import type { PoolClient } from 'pg'; + +const LEADER_ELECTION_INTERVAL = '*/15 * * * * *'; +// Two int32 advisory-lock keys: ASCII-ish PAYD / SCHD namespaces. +const SCHEDULER_LOCK_NAMESPACE = 0x50415944; +const SCHEDULER_LOCK_KEY = 0x53434844; +const CLAIM_HEARTBEAT_MS = 60_000; export class ScheduleExecutor { private cronJob: ScheduledTask | null = null; private readonly podId: string; + private runInProgress = false; constructor() { this.podId = `${os.hostname()}-${process.pid}`; } /** - * Initialize the cron job to run every minute - * Sets up node-cron job with error handling and logging + * Probe scheduler leadership every 15 seconds. + * + * The advisory lock is session-scoped, so PostgreSQL releases it automatically + * if the leader pod dies or loses its database connection. Keeping the lock on + * a dedicated client for the full scheduler pass guarantees that at most one + * pod enters processDueSchedules at a time. */ initialize(): void { - // Cron expression: run every minute - this.cronJob = cron.schedule('* * * * *', async () => { + this.cronJob = cron.schedule(LEADER_ELECTION_INTERVAL, async () => { + if (this.runInProgress) { + return; + } + + this.runInProgress = true; + let leaderClient: PoolClient | null = null; + let hasLeadership = false; + let destroyLeaderConnection = false; + try { - console.log('[ScheduleExecutor] Running scheduled task check...'); + leaderClient = await pool.connect(); + const election = await leaderClient.query<{ acquired: boolean }>( + 'SELECT pg_try_advisory_lock($1, $2) AS acquired', + [SCHEDULER_LOCK_NAMESPACE, SCHEDULER_LOCK_KEY] + ); + + hasLeadership = election.rows[0]?.acquired === true; + if (!hasLeadership) { + return; + } + await this.processDueSchedules(); } catch (error) { - console.error('[ScheduleExecutor] Error in cron job execution:', error); + console.error('[ScheduleExecutor] Error in leader scheduler execution:', error); + } finally { + if (leaderClient) { + if (hasLeadership) { + try { + const unlock = await leaderClient.query<{ unlocked: boolean }>( + 'SELECT pg_advisory_unlock($1, $2) AS unlocked', + [SCHEDULER_LOCK_NAMESPACE, SCHEDULER_LOCK_KEY] + ); + destroyLeaderConnection = unlock.rows[0]?.unlocked !== true; + } catch (error) { + destroyLeaderConnection = true; + console.error('[ScheduleExecutor] Failed to release scheduler leadership:', error); + } + } + + leaderClient.release(destroyLeaderConnection); + } + + this.runInProgress = false; } }); - console.log('[ScheduleExecutor] Cron job initialized - running every minute'); + console.log('[ScheduleExecutor] Cron job initialized - checking leadership every 15 seconds'); } /** @@ -52,7 +102,11 @@ export class ScheduleExecutor { // Reclaim rows from crashed pods before attempting our own claim await this.releaseStaleClaims(); + // Reuse of a hostname+PID is not reuse of row ownership. A unique token + // fences late completion or cleanup after another scheduler pass claims it. + const claimOwner = `${this.podId.slice(0, 80)}:${randomUUID()}`; const client = await pool.connect(); + let heartbeat: ReturnType | null = null; try { await client.query('BEGIN'); @@ -87,12 +141,24 @@ export class ScheduleExecutor { updated_at as "updatedAt" `; - const result = await client.query(claimQuery, [this.podId]); + const result = await client.query(claimQuery, [claimOwner]); await client.query('COMMIT'); const claimedSchedules = result.rows; if (claimedSchedules.length > 0) { + // Processing is serial. Refresh ALL claimed rows, including rows that + // are queued but not yet running; renewal cannot touch a newer owner. + const ids = claimedSchedules.map((row: { id: number }) => row.id); + heartbeat = setInterval(() => { + void pool.query( + `UPDATE schedules SET locked_at = NOW() + WHERE locked_by = $1 AND id = ANY($2::integer[])`, + [claimOwner, ids] + ).catch((error: unknown) => { + console.error('[ScheduleExecutor] Row-claim renewal failed:', error); + }); + }, CLAIM_HEARTBEAT_MS); console.log(`[ScheduleExecutor] Claimed ${claimedSchedules.length} due schedule(s)`); } @@ -117,7 +183,7 @@ export class ScheduleExecutor { const executionResult = await this.executeSchedule(schedule); - await this.recordExecution(schedule.id, executionResult); + await this.recordExecution(schedule.id, executionResult, claimOwner); if (executionResult.success) { successCount++; @@ -143,7 +209,7 @@ export class ScheduleExecutor { message: error instanceof Error ? error.message : 'System error in executor', details: error as any, }, - }); + }, claimOwner); } catch (recordError) { console.error( `[ScheduleExecutor] Failed to record execution error for schedule ID ${scheduleRow.id}:`, @@ -152,7 +218,7 @@ export class ScheduleExecutor { } } finally { // Always release the claim so the row is available for the next cycle - await this.releaseClaim(scheduleRow.id); + await this.releaseClaim(scheduleRow.id, claimOwner); } } @@ -165,6 +231,7 @@ export class ScheduleExecutor { await client.query('ROLLBACK').catch(() => {}); throw error; } finally { + if (heartbeat) clearInterval(heartbeat); client.release(); } } @@ -172,11 +239,11 @@ export class ScheduleExecutor { /** * Release the row-level claim after execution (success or failure). */ - private async releaseClaim(scheduleId: number): Promise { + private async releaseClaim(scheduleId: number, claimOwner: string): Promise { try { await pool.query( - 'UPDATE schedules SET locked_by = NULL, locked_at = NULL WHERE id = $1', - [scheduleId] + 'UPDATE schedules SET locked_by = NULL, locked_at = NULL WHERE id = $1 AND locked_by = $2', + [scheduleId, claimOwner] ); } catch (error) { console.error(`[ScheduleExecutor] Failed to release claim for schedule ID ${scheduleId}:`, error); @@ -298,7 +365,7 @@ export class ScheduleExecutor { * @param scheduleId - The schedule ID * @param result - The execution result */ - async recordExecution(scheduleId: number, result: ExecutionResult): Promise { + async recordExecution(scheduleId: number, result: ExecutionResult, claimOwner: string): Promise { const client = await pool.connect(); try { await client.query('BEGIN'); @@ -334,12 +401,12 @@ export class ScheduleExecutor { await client.query(insertQuery, insertValues); // Update schedule state using ScheduleService - await scheduleService.updateAfterExecution(scheduleId, result); + await scheduleService.updateAfterExecution(scheduleId, result, claimOwner); // Clear the lock now that execution is recorded await client.query( - 'UPDATE schedules SET locked_by = NULL, locked_at = NULL WHERE id = $1', - [scheduleId] + 'UPDATE schedules SET locked_by = NULL, locked_at = NULL WHERE id = $1 AND locked_by = $2', + [scheduleId, claimOwner] ); await client.query('COMMIT'); diff --git a/backend/src/services/scheduleService.ts b/backend/src/services/scheduleService.ts index 5072bd34..803ae2ae 100644 --- a/backend/src/services/scheduleService.ts +++ b/backend/src/services/scheduleService.ts @@ -361,6 +361,7 @@ export class ScheduleService { async updateAfterExecution( scheduleId: number, executionResult: ExecutionResult, + claimOwner?: string, ): Promise { const client = await pool.connect(); try { @@ -377,9 +378,12 @@ export class ScheduleService { last_run_timestamp as "lastRunTimestamp" FROM schedules WHERE id = $1 + ${claimOwner ? 'AND locked_by = $2' : ''} `; - const selectResult = await client.query(selectQuery, [scheduleId]); + const selectResult = await client.query( + selectQuery, claimOwner ? [scheduleId, claimOwner] : [scheduleId] + ); if (selectResult.rows.length === 0) { throw new Error(`Schedule with ID ${scheduleId} not found`); @@ -422,14 +426,21 @@ export class ScheduleService { next_run_timestamp = COALESCE($3, next_run_timestamp), updated_at = CURRENT_TIMESTAMP WHERE id = $4 + ${claimOwner ? 'AND locked_by = $5' : ''} `; - await client.query(updateQuery, [ + const updated = await client.query(updateQuery, [ executionTime, newStatus, nextRunTimestamp, scheduleId, + ...(claimOwner ? [claimOwner] : []), ]); + // If a dead pod or a newer scheduler pass took over the row, do not + // advance its next-run status or commit stale result metadata. + if (claimOwner && updated.rowCount !== 1) { + throw new Error('Schedule claim lost before finalizing execution'); + } await client.query('COMMIT'); } catch (error) {