From 2b8e0003f713fadc305ba190932a2056eb9ce728 Mon Sep 17 00:00:00 2001 From: Yggdrasil Date: Wed, 14 Oct 2026 02:53:57 -0400 Subject: [PATCH 1/3] fix(scheduler): cron jobs run single-runner via a Postgres advisory lock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Plain @Cron fired on EVERY replica — with N replicas the bounty-expiry and idempotency-cleanup sweeps ran N times an hour, racing the same rows. Wrap each in pg_try_advisory_lock so only the lock-holding replica sweeps. Resolves #586 --- src/bounties/bounty-expiry.scheduler.ts | 39 +++++++++++++++++-- .../idempotency-cleanup.service.spec.ts | 13 ++++++- .../idempotency-cleanup.service.ts | 38 ++++++++++++++---- 3 files changed, 78 insertions(+), 12 deletions(-) diff --git a/src/bounties/bounty-expiry.scheduler.ts b/src/bounties/bounty-expiry.scheduler.ts index 99363e12..901e6df5 100644 --- a/src/bounties/bounty-expiry.scheduler.ts +++ b/src/bounties/bounty-expiry.scheduler.ts @@ -1,7 +1,13 @@ import { Injectable, Logger } from '@nestjs/common'; import { Cron, CronExpression } from '@nestjs/schedule'; +import { InjectDataSource } from '@nestjs/typeorm'; +import { DataSource } from 'typeorm'; import { BountiesService } from './bounties.service'; +// Postgres advisory-lock key for the bounty-expiry sweep — an arbitrary int +// identifying this job. Any other replica trying the same key finds it held. +const BOUNTY_EXPIRY_LOCK_KEY = 0x59676701; + /** * Runs `BountiesService.expireOverdue()` on a schedule. * @@ -18,13 +24,38 @@ import { BountiesService } from './bounties.service'; export class BountyExpiryScheduler { private readonly logger = new Logger(BountyExpiryScheduler.name); - constructor(private readonly bountiesService: BountiesService) {} + constructor( + private readonly bountiesService: BountiesService, + @InjectDataSource() private readonly dataSource: DataSource, + ) {} @Cron(CronExpression.EVERY_HOUR) async sweepOverdueBounties(): Promise { - const expired = await this.bountiesService.expireOverdue(); - if (expired > 0) { - this.logger.log(`Expired ${expired} overdue bounty(ies)`); + // A plain @Cron fires on EVERY replica — with N replicas the sweep ran N + // times an hour, racing the same rows (#586). A Postgres advisory lock + // makes it single-runner: pg_try_advisory_lock returns false on every + // replica except the one holding the lock, which then sweeps alone. + const qr = this.dataSource.createQueryRunner(); + await qr.connect(); + try { + const [{ locked }] = await qr.query( + 'SELECT pg_try_advisory_lock($1) AS locked', + [BOUNTY_EXPIRY_LOCK_KEY], + ); + if (!locked) return; // another replica holds the sweep + + try { + const expired = await this.bountiesService.expireOverdue(); + if (expired > 0) { + this.logger.log(`Expired ${expired} overdue bounty(ies)`); + } + } finally { + await qr.query('SELECT pg_advisory_unlock($1)', [ + BOUNTY_EXPIRY_LOCK_KEY, + ]); + } + } finally { + await qr.release(); } } } diff --git a/src/common/idempotency/idempotency-cleanup.service.spec.ts b/src/common/idempotency/idempotency-cleanup.service.spec.ts index 0a818824..1dc3da3e 100644 --- a/src/common/idempotency/idempotency-cleanup.service.spec.ts +++ b/src/common/idempotency/idempotency-cleanup.service.spec.ts @@ -1,6 +1,6 @@ import { Test, TestingModule } from '@nestjs/testing'; import { getRepositoryToken } from '@nestjs/typeorm'; -import { LessThan } from 'typeorm'; +import { DataSource, LessThan } from 'typeorm'; import { IdempotencyKey } from '../entities/idempotency-key.entity'; import { IdempotencyCleanupService } from './idempotency-cleanup.service'; @@ -21,10 +21,21 @@ describe('IdempotencyCleanupService', () => { .mockResolvedValue({ affected: 0 }), }; + // The DataSource mock answers the advisory-lock query with locked:true + // so the single-runner guard passes and the sweep runs (#586). + const dataSource = { + createQueryRunner: jest.fn().mockReturnValue({ + connect: jest.fn(), + query: jest.fn().mockResolvedValue([{ locked: true }]), + release: jest.fn(), + }), + }; + const module: TestingModule = await Test.createTestingModule({ providers: [ IdempotencyCleanupService, { provide: getRepositoryToken(IdempotencyKey), useValue: repo }, + { provide: DataSource, useValue: dataSource }, ], }).compile(); diff --git a/src/common/idempotency/idempotency-cleanup.service.ts b/src/common/idempotency/idempotency-cleanup.service.ts index 73c08bc7..4ce2527b 100644 --- a/src/common/idempotency/idempotency-cleanup.service.ts +++ b/src/common/idempotency/idempotency-cleanup.service.ts @@ -1,9 +1,11 @@ import { Injectable, Logger } from '@nestjs/common'; import { Cron, CronExpression } from '@nestjs/schedule'; -import { InjectRepository } from '@nestjs/typeorm'; -import { LessThan, Repository } from 'typeorm'; +import { InjectDataSource, InjectRepository } from '@nestjs/typeorm'; +import { DataSource, LessThan, Repository } from 'typeorm'; import { IdempotencyKey } from '../entities/idempotency-key.entity'; +const IDEMPOTENCY_CLEANUP_LOCK_KEY = 0x59676702; + /** * Sweeps expired idempotency_keys rows (#16). Retention itself is set per * row at insert time via IDEMPOTENCY_KEY_TTL_MS @@ -18,15 +20,37 @@ export class IdempotencyCleanupService { constructor( @InjectRepository(IdempotencyKey) private readonly repo: Repository, + @InjectDataSource() private readonly dataSource: DataSource, ) {} @Cron(CronExpression.EVERY_HOUR) async removeExpired(): Promise { - const result = await this.repo.delete({ - expiresAt: LessThan(new Date()), - }); - if (result.affected) { - this.logger.log(`Removed ${result.affected} expired idempotency key(s)`); + // Single-runner across replicas — a plain @Cron fired on every replica + // (#586); the Postgres advisory lock means only the holder sweeps. + const qr = this.dataSource.createQueryRunner(); + await qr.connect(); + try { + const [{ locked }] = await qr.query( + 'SELECT pg_try_advisory_lock($1) AS locked', + [IDEMPOTENCY_CLEANUP_LOCK_KEY], + ); + if (!locked) return; + try { + const result = await this.repo.delete({ + expiresAt: LessThan(new Date()), + }); + if (result.affected) { + this.logger.log( + `Removed ${result.affected} expired idempotency key(s)`, + ); + } + } finally { + await qr.query('SELECT pg_advisory_unlock($1)', [ + IDEMPOTENCY_CLEANUP_LOCK_KEY, + ]); + } + } finally { + await qr.release(); } } } From bd9db5142c6217ee4a85a56de3291a908ef2894a Mon Sep 17 00:00:00 2001 From: yggdrasilworld Date: Sun, 18 Oct 2026 21:50:26 -0400 Subject: [PATCH 2/3] fix: repair merge glue that left the branch uncompilable The shared base carried half-written merge artifacts: an orphaned invokeOnLockedEscrow block in escrow.service.ts, duplicated returns plus a dead parsePagination in sponsors.controller.ts, and missing describe/it closers in github-webhooks.service.spec.ts. Aligned each with the intended form so the branch typechecks and its specs load. --- src/escrow/escrow.service.ts | 10 ++-------- src/github/github-webhooks.service.spec.ts | 3 +++ src/sponsors/sponsors.controller.ts | 21 --------------------- 3 files changed, 5 insertions(+), 29 deletions(-) diff --git a/src/escrow/escrow.service.ts b/src/escrow/escrow.service.ts index 6d2dd289..c252993f 100644 --- a/src/escrow/escrow.service.ts +++ b/src/escrow/escrow.service.ts @@ -584,20 +584,14 @@ export class EscrowService { recipients: Array<[string, number]>, manager?: EntityManager, ): Promise { - return this.invokeOnLockedEscrow(escrow, operation, () => - this.soroban.invoke( - 'release', - // `release(issue_id: u64, recipients)` — u64-typed on-chain (#301). - [u64(this.onChainKeyFor(escrow)), recipients], - this.contractOpts(escrow), - ), return this.invokeOnLockedEscrow( escrow, operation, () => this.soroban.invoke( 'release', - [this.onChainKeyFor(escrow), recipients], + // `release(issue_id: u64, recipients)` — u64-typed on-chain (#301). + [u64(this.onChainKeyFor(escrow)), recipients], this.contractOpts(escrow), ), manager, diff --git a/src/github/github-webhooks.service.spec.ts b/src/github/github-webhooks.service.spec.ts index 45bb6fe8..a35ac3be 100644 --- a/src/github/github-webhooks.service.spec.ts +++ b/src/github/github-webhooks.service.spec.ts @@ -869,6 +869,9 @@ describe('GithubWebhooksService', () => { expect(bountiesService.markInReview).not.toHaveBeenCalled(); expect(bountiesService.markMergedAndRelease).not.toHaveBeenCalled(); + }); + }); + // #308: a redelivered X-GitHub-Delivery used to hit the unique constraint // on webhook_events.deliveryId and escape handleEvent as a 500, so every // redelivery of that event failed forever. diff --git a/src/sponsors/sponsors.controller.ts b/src/sponsors/sponsors.controller.ts index b7a48cca..c097bf1e 100644 --- a/src/sponsors/sponsors.controller.ts +++ b/src/sponsors/sponsors.controller.ts @@ -26,25 +26,6 @@ function validatePage(limit?: number, offset?: number) { throw new BadRequestException('offset must be >= 0'); } return { limit, offset }; -/** - * Parse optional limit/offset query params, rejecting NaN, negatives, - * non-integers, and limits above MAX_PAGE_LIMIT with a 400 (#280). - */ -function parsePagination(limit?: string, offset?: string) { - const parse = (name: string, raw: string | undefined, min: number, max: number) => { - if (raw === undefined || raw === '') return undefined; - const n = Number(raw); - if (!Number.isInteger(n) || n < min || n > max) { - throw new BadRequestException( - `${name} must be an integer between ${min} and ${max}`, - ); - } - return n; - }; - return { - limit: parse('limit', limit, 1, MAX_PAGE_LIMIT), - offset: parse('offset', offset, 0, Number.MAX_SAFE_INTEGER), - }; } @ApiTags('sponsors') @@ -73,7 +54,6 @@ export class SponsorsController { ) { this.assertOwnsSponsor(user, id); return this.sponsorsService.dashboard(id, validatePage(limit, offset)); - return this.sponsorsService.dashboard(id, parsePagination(limit, offset)); } @ApiBearerAuth() @@ -91,6 +71,5 @@ export class SponsorsController { id, validatePage(limit, offset), ); - return this.sponsorsService.milestoneProgress(id, parsePagination(limit, offset)); } } From 77196ccd2d660d6ba6154d907b13a80f294c07cc Mon Sep 17 00:00:00 2001 From: Devin Date: Sun, 18 Oct 2026 22:09:49 -0400 Subject: [PATCH 3/3] repair stale specs to match shipped signatures/fixtures --- src/bounties/bounty-expiry.scheduler.spec.ts | 39 ++++++++++++++++++++ 1 file changed, 39 insertions(+) diff --git a/src/bounties/bounty-expiry.scheduler.spec.ts b/src/bounties/bounty-expiry.scheduler.spec.ts index 58f80d1a..e475e0c6 100644 --- a/src/bounties/bounty-expiry.scheduler.spec.ts +++ b/src/bounties/bounty-expiry.scheduler.spec.ts @@ -1,18 +1,38 @@ import { Test, TestingModule } from '@nestjs/testing'; +import { DataSource } from 'typeorm'; import { BountiesService } from './bounties.service'; import { BountyExpiryScheduler } from './bounty-expiry.scheduler'; describe('BountyExpiryScheduler', () => { let scheduler: BountyExpiryScheduler; let bountiesService: { expireOverdue: jest.Mock }; + let queryRunner: { + connect: jest.Mock; + query: jest.Mock; + release: jest.Mock; + }; + let dataSource: { createQueryRunner: jest.Mock }; beforeEach(async () => { bountiesService = { expireOverdue: jest.fn() }; + queryRunner = { + connect: jest.fn().mockResolvedValue(undefined), + // First query acquires the advisory lock, second releases it. + query: jest + .fn() + .mockResolvedValueOnce([{ locked: true }]) + .mockResolvedValue([{ unlocked: true }]), + release: jest.fn().mockResolvedValue(undefined), + }; + dataSource = { + createQueryRunner: jest.fn().mockReturnValue(queryRunner), + }; const module: TestingModule = await Test.createTestingModule({ providers: [ BountyExpiryScheduler, { provide: BountiesService, useValue: bountiesService }, + { provide: DataSource, useValue: dataSource }, ], }).compile(); @@ -46,4 +66,23 @@ describe('BountyExpiryScheduler', () => { expect(loggerSpy).not.toHaveBeenCalled(); }); + + it('skips the sweep when another replica holds the advisory lock (#586)', async () => { + queryRunner.query.mockReset().mockResolvedValue([{ locked: false }]); + + await scheduler.sweepOverdueBounties(); + + expect(bountiesService.expireOverdue).not.toHaveBeenCalled(); + expect(queryRunner.release).toHaveBeenCalledTimes(1); + }); + + it('releases the advisory lock even when the sweep throws', async () => { + bountiesService.expireOverdue.mockRejectedValue(new Error('db gone')); + + await expect(scheduler.sweepOverdueBounties()).rejects.toThrow('db gone'); + + const calls = queryRunner.query.mock.calls.map((c) => String(c[0])); + expect(calls.some((q) => q.includes('pg_advisory_unlock'))).toBe(true); + expect(queryRunner.release).toHaveBeenCalledTimes(1); + }); });