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

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