Skip to content

Rehmy implementation gf/4 #460

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Wilfred007 merged 3 commits into from
Aug 16, 2026
Merged

Rehmy implementation gf/4 #460

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
2f31a02
Add migration for schedule row-level locking columns
Aug 16, 2026
0a34505
Add FOR UPDATE SKIP LOCKED to schedule claim query
Aug 16, 2026
f9286e5
Update schedule executor tests for row-locking changes
Aug 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions backend/src/db/migrations/028_schedule_row_locking.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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;
172 changes: 163 additions & 9 deletions backend/src/services/__tests__/scheduleExecutor.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ jest.mock('../../config/database.js', () => ({
__esModule: true,
default: {
connect: jest.fn(),
query: jest.fn(),
},
}));

Expand All @@ -30,7 +31,6 @@ describe('ScheduleExecutor', () => {
const mockStellarService = StellarService as jest.Mocked<typeof StellarService>;
const mockScheduleService = scheduleService as jest.Mocked<typeof scheduleService>;

const mockConnect = jest.fn();
const mockRelease = jest.fn();
const mockClientQuery = jest.fn();

Expand All @@ -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';
Expand Down Expand Up @@ -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,
Expand All @@ -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({
Expand All @@ -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,
Expand All @@ -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();
});

Expand Down Expand Up @@ -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();

Expand Down Expand Up @@ -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'))
Expand All @@ -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();

Expand All @@ -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');

Expand Down Expand Up @@ -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();
Expand All @@ -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();
});
Expand All @@ -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();
Expand Down Expand Up @@ -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();
});
});
});
Loading
Loading