From 2f31a02f784da5ce12c0e41b0f6e74a88f1f564b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9Crhemy-arc=E2=80=9D?= <“rhemaadzer@gmail.com”> Date: Sun, 16 Aug 2026 07:50:00 +0100 Subject: [PATCH 1/3] Add migration for schedule row-level locking columns Adds locked_by and locked_at columns to the schedules table to support claim-based concurrency control. Includes an index for efficient lookup of claimable rows (active + due + unlocked). --- .../src/db/migrations/028_schedule_row_locking.sql | 12 ++++++++++++ 1 file changed, 12 insertions(+) create mode 100644 backend/src/db/migrations/028_schedule_row_locking.sql 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; From 0a34505152a886877c158de13ae7afecd753b334 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9Crhemy-arc=E2=80=9D?= <“rhemaadzer@gmail.com”> Date: Sun, 16 Aug 2026 09:30:00 +0100 Subject: [PATCH 2/3] Add FOR UPDATE SKIP LOCKED to schedule claim query Replace the plain SELECT in processDueSchedules() with an atomic UPDATE ... FROM (SELECT ... FOR UPDATE SKIP LOCKED) pattern that marks each claimed row with the pod's hostname+pid. This prevents two pods from executing the same schedule simultaneously. Also adds releaseClaim() to clear locks after execution and releaseStaleClaims() to reclaim rows from crashed pods (5min timeout). --- backend/src/services/scheduleExecutor.ts | 106 ++++++++++++++++++----- 1 file changed, 83 insertions(+), 23 deletions(-) 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'); From f9286e5832c7094233223c122d29162bf2eb47aa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9Crhemy-arc=E2=80=9D?= <“rhemaadzer@gmail.com”> Date: Sun, 16 Aug 2026 11:15:00 +0100 Subject: [PATCH 3/3] Update schedule executor tests for row-locking changes Update existing processDueSchedules tests to match the new claim query (BEGIN/UPDATE FOR UPDATE SKIP LOCKED/COMMIT pattern) and add lock-clear assertions to recordExecution tests. --- .../__tests__/scheduleExecutor.test.ts | 172 +++++++++++++++++- 1 file changed, 163 insertions(+), 9 deletions(-) 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(); + }); + }); });