fix(scheduler): cron jobs run single-runner via a Postgres advisory lock

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);
+  });
 });

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<void> {
-    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();
     }
   }
 }

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();
 

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<IdempotencyKey>,
+    @InjectDataSource() private readonly dataSource: DataSource,
   ) {}
 
   @Cron(CronExpression.EVERY_HOUR)
   async removeExpired(): Promise<void> {
-    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();
     }
   }
 }

src/escrow/escrow.service.ts

@@ -584,20 +584,14 @@ export class EscrowService {
     recipients: Array<[string, number]>,
     manager?: EntityManager,
   ): Promise<ContractInvocationResult> {
-    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,

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.

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));
   }
 }