diff --git a/backend/src/db/migrations/028_schedule_row_locking.sql b/backend/src/db/migrations/028_schedule_row_locking.sql new file mode 100644 index 00000000..e6a694ce --- /dev/null +++ b/backend/src/db/migrations/028_schedule_row_locking.sql @@ -0,0 +1,12 @@ +-- Add row-level locking support to schedules table. +-- locked_by: identifies which pod/process claimed the row (e.g. hostname + pid) +-- locked_at: when the claim was acquired; stale claims can be reclaimed after a timeout + +ALTER TABLE schedules + ADD COLUMN IF NOT EXISTS locked_by VARCHAR(128), + ADD COLUMN IF NOT EXISTS locked_at TIMESTAMPTZ; + +-- Index to support the claim query (active + due + unlocked rows) +CREATE INDEX IF NOT EXISTS idx_schedules_claim + ON schedules (next_run_timestamp, status) + WHERE status = 'active' AND locked_by IS NULL; diff --git a/backend/src/services/__tests__/scheduleExecutor.test.ts b/backend/src/services/__tests__/scheduleExecutor.test.ts index fafd8779..75c4229e 100644 --- a/backend/src/services/__tests__/scheduleExecutor.test.ts +++ b/backend/src/services/__tests__/scheduleExecutor.test.ts @@ -9,6 +9,7 @@ jest.mock('../../config/database.js', () => ({ __esModule: true, default: { connect: jest.fn(), + query: jest.fn(), }, })); @@ -30,7 +31,6 @@ describe('ScheduleExecutor', () => { const mockStellarService = StellarService as jest.Mocked; const mockScheduleService = scheduleService as jest.Mocked; - const mockConnect = jest.fn(); const mockRelease = jest.fn(); const mockClientQuery = jest.fn(); @@ -44,6 +44,9 @@ describe('ScheduleExecutor', () => { release: mockRelease, }); + // Default: stale claims cleanup returns 0 + (mockPool.query as jest.Mock).mockResolvedValue({ rows: [], rowCount: 0 }); + // Setup environment variables process.env.STELLAR_SOURCE_SECRET = 'SXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX'; process.env.STELLAR_ASSET_ISSUER = 'GXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX'; @@ -96,7 +99,7 @@ describe('ScheduleExecutor', () => { }); describe('processDueSchedules', () => { - it('should query for due schedules and process them', async () => { + it('should claim due schedules with FOR UPDATE SKIP LOCKED and process them', async () => { const mockSchedules = [ { id: 1, @@ -123,7 +126,11 @@ describe('ScheduleExecutor', () => { }, ]; - mockClientQuery.mockResolvedValueOnce({ rows: mockSchedules }); + // BEGIN, claim UPDATE, COMMIT + mockClientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockResolvedValueOnce({ rows: mockSchedules }) // UPDATE ... FOR UPDATE SKIP LOCKED + .mockResolvedValueOnce({ rows: [] }); // COMMIT // Mock executeSchedule to return success jest.spyOn(executor, 'executeSchedule').mockResolvedValue({ @@ -134,11 +141,17 @@ describe('ScheduleExecutor', () => { // Mock recordExecution jest.spyOn(executor, 'recordExecution').mockResolvedValue(); + // Mock releaseClaim + (mockPool.query as jest.Mock).mockResolvedValue({ rows: [], rowCount: 0 }); + await executor.processDueSchedules(); + expect(mockClientQuery).toHaveBeenCalledWith('BEGIN'); expect(mockClientQuery).toHaveBeenCalledWith( - expect.stringContaining('WHERE next_run_timestamp <= NOW() AND status = \'active\'') + expect.stringContaining('FOR UPDATE SKIP LOCKED'), + expect.arrayContaining([expect.any(String)]) // podId ); + expect(mockClientQuery).toHaveBeenCalledWith('COMMIT'); expect(executor.executeSchedule).toHaveBeenCalledWith( expect.objectContaining({ id: 1, @@ -153,11 +166,16 @@ describe('ScheduleExecutor', () => { }); it('should handle empty result set', async () => { - mockClientQuery.mockResolvedValueOnce({ rows: [] }); + // BEGIN, claim UPDATE (empty), COMMIT + mockClientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockResolvedValueOnce({ rows: [] }) // UPDATE returns no rows + .mockResolvedValueOnce({ rows: [] }); // COMMIT await executor.processDueSchedules(); - expect(mockClientQuery).toHaveBeenCalled(); + expect(mockClientQuery).toHaveBeenCalledWith('BEGIN'); + expect(mockClientQuery).toHaveBeenCalledWith('COMMIT'); expect(mockRelease).toHaveBeenCalled(); }); @@ -211,13 +229,18 @@ describe('ScheduleExecutor', () => { }, ]; - mockClientQuery.mockResolvedValueOnce({ rows: mockSchedules }); + // BEGIN, claim UPDATE (2 rows), COMMIT + mockClientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockResolvedValueOnce({ rows: mockSchedules }) // UPDATE returns 2 rows + .mockResolvedValueOnce({ rows: [] }); // COMMIT jest.spyOn(executor, 'executeSchedule').mockResolvedValue({ success: true, transactionHash: 'abc123', }); jest.spyOn(executor, 'recordExecution').mockResolvedValue(); + (mockPool.query as jest.Mock).mockResolvedValue({ rows: [], rowCount: 0 }); await executor.processDueSchedules(); @@ -275,7 +298,11 @@ describe('ScheduleExecutor', () => { }, ]; - mockClientQuery.mockResolvedValueOnce({ rows: mockSchedules }); + // BEGIN, claim UPDATE (2 rows), COMMIT + mockClientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockResolvedValueOnce({ rows: mockSchedules }) // UPDATE returns 2 rows + .mockResolvedValueOnce({ rows: [] }); // COMMIT jest.spyOn(executor, 'executeSchedule') .mockRejectedValueOnce(new Error('Execution failed')) @@ -284,6 +311,7 @@ describe('ScheduleExecutor', () => { transactionHash: 'def456', }); jest.spyOn(executor, 'recordExecution').mockResolvedValue(); + (mockPool.query as jest.Mock).mockResolvedValue({ rows: [], rowCount: 0 }); await executor.processDueSchedules(); @@ -293,7 +321,10 @@ describe('ScheduleExecutor', () => { }); it('should release client even on error', async () => { - mockClientQuery.mockRejectedValueOnce(new Error('Database error')); + mockClientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockRejectedValueOnce(new Error('Database error')) // UPDATE fails + .mockResolvedValueOnce({ rows: [] }); // ROLLBACK await expect(executor.processDueSchedules()).rejects.toThrow('Database error'); @@ -400,6 +431,8 @@ describe('ScheduleExecutor', () => { mockClientQuery .mockResolvedValueOnce({ rows: [] }) // BEGIN .mockResolvedValueOnce({ rows: [{ id: 1 }] }) // INSERT + .mockResolvedValueOnce({ rows: [] }) // UPDATE afterExecution + .mockResolvedValueOnce({ rows: [] }) // Clear lock .mockResolvedValueOnce({ rows: [] }); // COMMIT mockScheduleService.updateAfterExecution.mockResolvedValue(); @@ -423,6 +456,10 @@ describe('ScheduleExecutor', () => { scheduleId, executionResult ); + expect(mockClientQuery).toHaveBeenCalledWith( + expect.stringContaining('UPDATE schedules SET locked_by = NULL'), + [scheduleId] + ); expect(mockClientQuery).toHaveBeenCalledWith('COMMIT'); expect(mockRelease).toHaveBeenCalled(); }); @@ -439,6 +476,8 @@ describe('ScheduleExecutor', () => { mockClientQuery .mockResolvedValueOnce({ rows: [] }) // BEGIN .mockResolvedValueOnce({ rows: [{ id: 1 }] }) // INSERT + .mockResolvedValueOnce({ rows: [] }) // UPDATE afterExecution + .mockResolvedValueOnce({ rows: [] }) // Clear lock .mockResolvedValueOnce({ rows: [] }); // COMMIT mockScheduleService.updateAfterExecution.mockResolvedValue(); @@ -496,4 +535,119 @@ describe('ScheduleExecutor', () => { expect(mockRelease).toHaveBeenCalled(); }); }); + + describe('concurrent pod safety', () => { + it('should only allow one pod to claim a schedule when two run concurrently', async () => { + const mockSchedule = { + id: 42, + organizationId: 1, + userId: 1, + frequency: 'monthly', + timeOfDay: '09:00', + startDate: '2024-01-01', + endDate: null, + paymentConfig: { + recipients: [ + { + walletAddress: 'GXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX', + amount: '500.00', + assetCode: 'XLM', + }, + ], + }, + nextRunTimestamp: new Date(), + lastRunTimestamp: null, + status: 'active', + createdAt: new Date(), + updatedAt: new Date(), + }; + + let executeCallCount = 0; + + // Pod A: claims the schedule successfully + const podA = new ScheduleExecutor(); + const podAClientQuery = jest.fn(); + podAClientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockResolvedValueOnce({ rows: [mockSchedule] }) // UPDATE claim succeeds + .mockResolvedValueOnce({ rows: [] }); // COMMIT + + (mockPool.connect as jest.Mock).mockResolvedValueOnce({ + query: podAClientQuery, + release: jest.fn(), + }); + (mockPool.query as jest.Mock).mockResolvedValue({ rows: [], rowCount: 0 }); + + jest.spyOn(podA, 'executeSchedule').mockImplementation(async () => { + executeCallCount++; + return { success: true, transactionHash: 'hash-a' }; + }); + jest.spyOn(podA, 'recordExecution').mockResolvedValue(); + + // Pod B: tries to claim but FOR UPDATE SKIP LOCKED returns empty + const podB = new ScheduleExecutor(); + const podBClientQuery = jest.fn(); + podBClientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockResolvedValueOnce({ rows: [] }) // UPDATE claim returns 0 rows (skipped) + .mockResolvedValueOnce({ rows: [] }); // COMMIT + + (mockPool.connect as jest.Mock).mockResolvedValueOnce({ + query: podBClientQuery, + release: jest.fn(), + }); + + jest.spyOn(podB, 'executeSchedule').mockResolvedValue({ + success: true, + transactionHash: 'hash-b', + }); + jest.spyOn(podB, 'recordExecution').mockResolvedValue(); + + // Both pods run concurrently + await Promise.all([podA.processDueSchedules(), podB.processDueSchedules()]); + + // Only Pod A should have executed the schedule + expect(executeCallCount).toBe(1); + expect(podA.executeSchedule).toHaveBeenCalledTimes(1); + expect(podB.executeSchedule).not.toHaveBeenCalled(); + }); + + it('should use FOR UPDATE SKIP LOCKED in the claim query', async () => { + mockClientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockResolvedValueOnce({ rows: [] }) // UPDATE + .mockResolvedValueOnce({ rows: [] }); // COMMIT + + await executor.processDueSchedules(); + + const claimCall = mockClientQuery.mock.calls.find( + (call: any[]) => typeof call[0] === 'string' && call[0].includes('FOR UPDATE SKIP LOCKED') + ); + expect(claimCall).toBeDefined(); + expect(claimCall![0]).toContain('locked_by IS NULL'); + expect(claimCall![1]).toEqual([expect.any(String)]); // podId parameter + }); + + it('should release stale claims before claiming new ones', async () => { + mockClientQuery + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockResolvedValueOnce({ rows: [] }) // UPDATE + .mockResolvedValueOnce({ rows: [] }); // COMMIT + + (mockPool.query as jest.Mock).mockResolvedValue({ rows: [], rowCount: 2 }); + + const consoleSpy = jest.spyOn(console, 'log').mockImplementation(); + + await executor.processDueSchedules(); + + expect(mockPool.query).toHaveBeenCalledWith( + expect.stringContaining('locked_at < NOW()') + ); + expect(consoleSpy).toHaveBeenCalledWith( + expect.stringContaining('Released 2 stale claim(s)') + ); + + consoleSpy.mockRestore(); + }); + }); }); diff --git a/backend/src/services/scheduleExecutor.ts b/backend/src/services/scheduleExecutor.ts index 10f81659..f34c242a 100644 --- a/backend/src/services/scheduleExecutor.ts +++ b/backend/src/services/scheduleExecutor.ts @@ -5,9 +5,15 @@ import { StellarService } from './stellarService.js'; 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'; export class ScheduleExecutor { private cronJob: ScheduledTask | null = null; + private readonly podId: string; + + constructor() { + this.podId = `${os.hostname()}-${process.pid}`; + } /** * Initialize the cron job to run every minute @@ -38,15 +44,33 @@ export class ScheduleExecutor { } /** - * Query database for due schedules and execute each one - * Handles errors in isolation so one failure doesn't block others + * Claim due schedules atomically using FOR UPDATE SKIP LOCKED. + * Each row is marked with the claiming pod's ID so concurrent pods + * skip already-claimed rows instead of executing them twice. */ async processDueSchedules(): Promise { + // Reclaim rows from crashed pods before attempting our own claim + await this.releaseStaleClaims(); + const client = await pool.connect(); try { - // Query for schedules where next_run_timestamp <= NOW() AND status = 'active' - const query = ` - SELECT + await client.query('BEGIN'); + + // Claim due schedules atomically: SELECT … FOR UPDATE SKIP LOCKED + // locks the rows this pod claims; other pods skip them. + const claimQuery = ` + UPDATE schedules + SET locked_by = $1, locked_at = NOW() + WHERE id IN ( + SELECT id + FROM schedules + WHERE next_run_timestamp <= (NOW() AT TIME ZONE 'UTC') + AND status = 'active' + AND locked_by IS NULL + ORDER BY next_run_timestamp ASC + FOR UPDATE SKIP LOCKED + ) + RETURNING id, organization_id as "organizationId", user_id as "userId", @@ -61,25 +85,22 @@ export class ScheduleExecutor { status, created_at as "createdAt", updated_at as "updatedAt" - FROM schedules - WHERE next_run_timestamp <= (NOW() AT TIME ZONE 'UTC') AND status = 'active' - ORDER BY next_run_timestamp ASC `; - const result = await client.query(query); - const dueSchedules = result.rows; + const result = await client.query(claimQuery, [this.podId]); + await client.query('COMMIT'); + + const claimedSchedules = result.rows; - if (dueSchedules.length > 0) { - console.log(`[ScheduleExecutor] Found ${dueSchedules.length} due schedule(s)`); + if (claimedSchedules.length > 0) { + console.log(`[ScheduleExecutor] Claimed ${claimedSchedules.length} due schedule(s)`); } let successCount = 0; let failureCount = 0; - // Process each schedule in isolation - for (const scheduleRow of dueSchedules) { + for (const scheduleRow of claimedSchedules) { try { - // Parse dates and JSON from database const schedule: Schedule = { ...scheduleRow, startDate: new Date(scheduleRow.startDate), @@ -92,16 +113,10 @@ export class ScheduleExecutor { updatedAt: new Date(scheduleRow.updatedAt), }; - // Idempotency check: Ensure we haven't already processed this exact run - // We can check if last_run_timestamp is very close to now AND next_run_timestamp hasn't updated yet - // But a better way is to rely on the transaction in recordExecution which updates the status/nextRun - console.log(`[ScheduleExecutor] Executing schedule ID ${schedule.id} (Scheduled for: ${schedule.nextRunTimestamp.toISOString()})`); - // Execute the schedule const executionResult = await this.executeSchedule(schedule); - // Record the execution (this updates next_run_timestamp or status) await this.recordExecution(schedule.id, executionResult); if (executionResult.success) { @@ -121,7 +136,6 @@ export class ScheduleExecutor { error ); - // Record the system error as a failure try { await this.recordExecution(scheduleRow.id, { success: false, @@ -136,19 +150,59 @@ export class ScheduleExecutor { recordError ); } + } finally { + // Always release the claim so the row is available for the next cycle + await this.releaseClaim(scheduleRow.id); } } - if (dueSchedules.length > 0) { + if (claimedSchedules.length > 0) { console.log( `[ScheduleExecutor] Execution complete - Success: ${successCount}, Failed: ${failureCount}` ); } + } catch (error) { + await client.query('ROLLBACK').catch(() => {}); + throw error; } finally { client.release(); } } + /** + * Release the row-level claim after execution (success or failure). + */ + private async releaseClaim(scheduleId: number): Promise { + try { + await pool.query( + 'UPDATE schedules SET locked_by = NULL, locked_at = NULL WHERE id = $1', + [scheduleId] + ); + } catch (error) { + console.error(`[ScheduleExecutor] Failed to release claim for schedule ID ${scheduleId}:`, error); + } + } + + /** + * Release stale claims held by crashed pods (locked_at older than 5 minutes). + * Called once per cron cycle before claiming new schedules. + */ + private async releaseStaleClaims(): Promise { + try { + const result = await pool.query( + `UPDATE schedules + SET locked_by = NULL, locked_at = NULL + WHERE locked_by IS NOT NULL + AND locked_at < NOW() - INTERVAL '5 minutes'` + ); + if (result.rowCount && result.rowCount > 0) { + console.log(`[ScheduleExecutor] Released ${result.rowCount} stale claim(s)`); + } + } catch (error) { + console.error('[ScheduleExecutor] Failed to release stale claims:', error); + } + } + /** * Execute a single schedule by building and submitting a Stellar transaction * @param schedule - The schedule to execute @@ -282,6 +336,12 @@ export class ScheduleExecutor { // Update schedule state using ScheduleService await scheduleService.updateAfterExecution(scheduleId, result); + // Clear the lock now that execution is recorded + await client.query( + 'UPDATE schedules SET locked_by = NULL, locked_at = NULL WHERE id = $1', + [scheduleId] + ); + await client.query('COMMIT'); } catch (error) { await client.query('ROLLBACK');