feat(keeper): take postgres advisory lock per job to prevent replica collisions

packages/db/src/client.ts

@@ -11,3 +11,14 @@ export function createDatabase(connectionString: string = process.env.DATABASE_U
   const sql = postgres(connectionString);
   return drizzle(sql, { schema });
 }
+
+export function createRawClient(connectionString: string = process.env.DATABASE_URL ?? ""): postgres.Sql {
+  if (!connectionString) {
+    throw new Error("@ballast/db: DATABASE_URL is not set");
+  }
+  return postgres(connectionString);
+}
+
+export type RawSql = postgres.Sql;
+export { postgres };
+

services/keeper/README.md

@@ -23,6 +23,17 @@ Jobs run one at a time through a shared queue, because they share one account an
 Each job returns `{ processed, skipped, errors }`; a failing row is logged and does not stop the rest.
 Liquidation and margin steps are written to be safe to repeat after a crash.
 
+## Replica Behaviour and Advisory Locking
+
+All keeper jobs sign with a single operator account and need sequential transaction sequence numbers. To prevent duplicate submissions, race conditions, or `tx_bad_seq` errors across multiple replicas (e.g. `--scale keeper=2` or during rolling restarts):
+
+- Each tick acquires a session-level Postgres advisory lock (`pg_try_advisory_lock`) on a dedicated connection.
+- If another keeper instance holds the lock, the tick is logged and skipped.
+- When the job finishes, the lock is released. If an instance crashes or disconnects, Postgres automatically releases the advisory lock.
+- `KEEPER_LOCK_SCOPE`: controls locking granularity:
+  - `job` (default): lock key is `hashtext('keeper:' || name)`, allowing different jobs to run while locking out duplicates of the same job.
+  - `global`: lock key is `hashtext('keeper:global')`, ensuring strict single-job execution across the entire cluster.
+
 ### Storage TTL Maintenance
 
 Soroban persistent storage entries archive if not extended. `CreditLine::extend_ttl(line)` is an unauthenticated contract entry point that extends instance and persistent entry TTLs (`Line`, `Claimable`, `Liquidation`). Keepers or periodic tasks can call this for dormant lines to ensure they remain accessible without requiring restoration.

services/keeper/src/index.ts

@@ -1,9 +1,10 @@
 import cron from "node-cron";
-import { createDatabase } from "@ballast/db";
+import { createDatabase, createRawClient } from "@ballast/db";
 import { loadNetworkConfig, requireContractId } from "@ballast/network-config";
 import { checkOperatorBalance, createConsoleAlerter, createLogger } from "@ballast/observability";
 import { CreditLineClient, PledgeVaultClient, RepoDvpClient } from "@ballast/contract-clients";
 import { createOperatorSigner } from "@ballast/operator-signing";
+import { executeJobWithLock, PostgresLockProvider, type LockProvider } from "./lock.js";
 import { runRepoUnwindJob } from "./jobs/repoUnwind.js";
 import { runRepoMarginJob } from "./jobs/repoMargin.js";
 import { runCureExpiryJob } from "./jobs/cureExpiry.js";
@@ -33,34 +34,36 @@ const repoDvpClient = new RepoDvpClient({
 // Every job signs with the one keeper account, and a Stellar account's transactions must land in
 // sequence-number order — two jobs building transactions at the same moment would both take the
 // same number and one would be rejected. So jobs queue up behind each other.
+// Across multiple keeper instances or rolling restarts, a Postgres session-level advisory lock
+// prevents duplicate job execution and sequence conflicts.
 let queue: Promise<void> = Promise.resolve();
+const rawSql = createRawClient();
+const defaultLockProvider = new PostgresLockProvider(rawSql);
 
 // Set to true once shutdown is initiated; prevents new ticks from entering the queue.
 let shuttingDown = false;
 
-function runJob(name: string, fn: () => Promise<JobResult>): Promise<void> {
+export function runJob(
+  name: string,
+  fn: () => Promise<JobResult>,
+  lockProvider: LockProvider = defaultLockProvider,
+): Promise<void> {
   if (shuttingDown) {
     logger.info({ job: name }, "keeper: shutdown in progress — skipping new job");
     return Promise.resolve();
   }
-  queue = queue.then(() => execute(name, fn));
+  queue = queue.then(async () => {
+    await execute(name, fn, lockProvider);
+  });
   return queue;
 }
 
-async function execute(name: string, fn: () => Promise<JobResult>): Promise<void> {
-  logger.info({ job: name }, "keeper: job starting");
-  try {
-    const result = await fn();
-    logger.info(
-      { job: name, processed: result.processed, skipped: result.skipped, errors: result.errors.length },
-      "keeper: job finished",
-    );
-    for (const e of result.errors) {
-      logger.error({ job: name, id: e.id, message: e.message }, "keeper: row failed within job");
-    }
-  } catch (err) {
-    logger.error({ job: name, err }, "keeper: job crashed");
-  }
+export function execute(
+  name: string,
+  fn: () => Promise<JobResult>,
+  lockProvider: LockProvider = defaultLockProvider,
+): Promise<boolean> {
+  return executeJobWithLock(name, fn, lockProvider, logger);
 }
 
 // PRD §6.5 defaults: repo leg 2 and cure-window expiries every 5 minutes; interest accrual
@@ -139,7 +142,7 @@ async function shutdown(signal: string): Promise<void> {
   // 3. Close the database pool.
   logger.info("keeper: closing database pool");
   try {
-    await db.$client.end();
+    await Promise.all([db.$client.end(), rawSql.end()]);
     logger.info("keeper: database pool closed");
   } catch (err) {
     logger.error({ err }, "keeper: error closing database pool");

services/keeper/src/lock.test.ts

@@ -0,0 +1,226 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import { executeJobWithLock, getLockKey, PostgresLockProvider, type LockProvider } from "./lock.js";
+
+describe("getLockKey", () => {
+  const originalEnv = process.env.KEEPER_LOCK_SCOPE;
+
+  beforeEach(() => {
+    delete process.env.KEEPER_LOCK_SCOPE;
+  });
+
+  it("defaults to job scope", () => {
+    expect(getLockKey("cure-expiry")).toBe("keeper:cure-expiry");
+    expect(getLockKey("repo-unwind")).toBe("keeper:repo-unwind");
+  });
+
+  it("uses global key when scope is global", () => {
+    expect(getLockKey("cure-expiry", "global")).toBe("keeper:global");
+    expect(getLockKey("repo-unwind", "global")).toBe("keeper:global");
+  });
+
+  it("respects KEEPER_LOCK_SCOPE environment variable", () => {
+    process.env.KEEPER_LOCK_SCOPE = "global";
+    expect(getLockKey("cure-expiry")).toBe("keeper:global");
+
+    process.env.KEEPER_LOCK_SCOPE = "job";
+    expect(getLockKey("cure-expiry")).toBe("keeper:cure-expiry");
+
+    if (originalEnv) {
+      process.env.KEEPER_LOCK_SCOPE = originalEnv;
+    } else {
+      delete process.env.KEEPER_LOCK_SCOPE;
+    }
+  });
+});
+
+describe("PostgresLockProvider", () => {
+  it("acquires advisory lock successfully and returns unlock/release handles", async () => {
+    const unlockMock = vi.fn().mockResolvedValue([]);
+    const releaseMock = vi.fn();
+    const reservedMock = vi.fn().mockImplementation((strings: TemplateStringsArray) => {
+      const query = strings.join("");
+      if (query.includes("pg_try_advisory_lock")) {
+        return Promise.resolve([{ locked: true }]);
+      }
+      if (query.includes("pg_advisory_unlock")) {
+        return unlockMock();
+      }
+      return Promise.resolve([]);
+    });
+    (reservedMock as any).release = releaseMock;
+
+    const sqlClient = {
+      reserve: vi.fn().mockResolvedValue(reservedMock),
+    };
+
+    const provider = new PostgresLockProvider(sqlClient as any);
+    const lock = await provider.acquire("keeper:sync-collateral");
+
+    expect(lock).not.toBeNull();
+    expect(sqlClient.reserve).toHaveBeenCalledTimes(1);
+    expect(releaseMock).not.toHaveBeenCalled();
+
+    await lock!.unlock();
+    expect(unlockMock).toHaveBeenCalledTimes(1);
+
+    lock!.release();
+    expect(releaseMock).toHaveBeenCalledTimes(1);
+  });
+
+  it("returns null and releases reservation when advisory lock is already held", async () => {
+    const releaseMock = vi.fn();
+    const reservedMock = vi.fn().mockImplementation((strings: TemplateStringsArray) => {
+      const query = strings.join("");
+      if (query.includes("pg_try_advisory_lock")) {
+        return Promise.resolve([{ locked: false }]);
+      }
+      return Promise.resolve([]);
+    });
+    (reservedMock as any).release = releaseMock;
+
+    const sqlClient = {
+      reserve: vi.fn().mockResolvedValue(reservedMock),
+    };
+
+    const provider = new PostgresLockProvider(sqlClient as any);
+    const lock = await provider.acquire("keeper:sync-collateral");
+
+    expect(lock).toBeNull();
+    expect(releaseMock).toHaveBeenCalledTimes(1);
+  });
+
+  it("releases connection and propagates error when lock query throws", async () => {
+    const releaseMock = vi.fn();
+    const reservedMock = vi.fn().mockRejectedValue(new Error("connection terminated"));
+    (reservedMock as any).release = releaseMock;
+
+    const sqlClient = {
+      reserve: vi.fn().mockResolvedValue(reservedMock),
+    };
+
+    const provider = new PostgresLockProvider(sqlClient as any);
+    await expect(provider.acquire("keeper:sync-collateral")).rejects.toThrow("connection terminated");
+    expect(releaseMock).toHaveBeenCalledTimes(1);
+  });
+});
+
+describe("executeJobWithLock", () => {
+  function makeMockLogger() {
+    return {
+      info: vi.fn(),
+      error: vi.fn(),
+    };
+  }
+
+  function createInMemoryLockProvider() {
+    const heldLocks = new Set<string>();
+    return {
+      heldLocks,
+      acquire: vi.fn().mockImplementation(async (lockKey: string) => {
+        if (heldLocks.has(lockKey)) {
+          return null;
+        }
+        heldLocks.add(lockKey);
+        let released = false;
+        return {
+          unlock: vi.fn().mockImplementation(async () => {
+            heldLocks.delete(lockKey);
+          }),
+          release: vi.fn().mockImplementation(() => {
+            released = true;
+          }),
+        };
+      }),
+    };
+  }
+
+  it("executes the job when lock is acquired and unlocks on completion", async () => {
+    const provider = createInMemoryLockProvider();
+    const logger = makeMockLogger();
+    const jobFn = vi.fn().mockResolvedValue({ processed: 2, skipped: 0, errors: [] });
+
+    const executed = await executeJobWithLock("repo-unwind", jobFn, provider, logger);
+
+    expect(executed).toBe(true);
+    expect(jobFn).toHaveBeenCalledTimes(1);
+    expect(provider.heldLocks.size).toBe(0);
+    expect(logger.info).toHaveBeenCalledWith(
+      expect.objectContaining({ job: "repo-unwind", lockKey: "keeper:repo-unwind" }),
+      "keeper: job starting",
+    );
+    expect(logger.info).toHaveBeenCalledWith(
+      expect.objectContaining({ job: "repo-unwind", processed: 2, skipped: 0, errors: 0 }),
+      "keeper: job finished",
+    );
+  });
+
+  it("skips tick without executing when another replica holds the lock", async () => {
+    const provider = createInMemoryLockProvider();
+    const logger1 = makeMockLogger();
+    const logger2 = makeMockLogger();
+
+    // Instance 1 holds the lock during an ongoing job
+    let finishInstance1Job: (res: any) => void;
+    const instance1JobPromise = new Promise((resolve) => {
+      finishInstance1Job = resolve;
+    });
+
+    const instance1Run = executeJobWithLock("repo-margin", () => instance1JobPromise as any, provider, logger1);
+
+    // Yield to let instance 1 acquire the lock
+    await new Promise((r) => setTimeout(r, 10));
+
+    // Instance 2 attempts to run the same job concurrently
+    const instance2Fn = vi.fn().mockResolvedValue({ processed: 1, skipped: 0, errors: [] });
+    const instance2Executed = await executeJobWithLock("repo-margin", instance2Fn, provider, logger2);
+
+    expect(instance2Executed).toBe(false);
+    expect(instance2Fn).not.toHaveBeenCalled();
+    expect(logger2.info).toHaveBeenCalledWith(
+      expect.objectContaining({ job: "repo-margin", lockKey: "keeper:repo-margin" }),
+      "keeper: advisory lock busy; skipping tick",
+    );
+
+    // Now instance 1 completes
+    finishInstance1Job!({ processed: 5, skipped: 0, errors: [] });
+    const instance1Executed = await instance1Run;
+    expect(instance1Executed).toBe(true);
+
+    // Next tick can acquire the lock again
+    const nextTickExecuted = await executeJobWithLock("repo-margin", instance2Fn, provider, logger2);
+    expect(nextTickExecuted).toBe(true);
+    expect(instance2Fn).toHaveBeenCalledTimes(1);
+  });
+
+  it("releases lock when job function crashes", async () => {
+    const provider = createInMemoryLockProvider();
+    const logger = makeMockLogger();
+    const crashingJob = vi.fn().mockRejectedValue(new Error("unexpected crash"));
+
+    const executed = await executeJobWithLock("interest-accrual", crashingJob, provider, logger);
+
+    expect(executed).toBe(false);
+    expect(provider.heldLocks.size).toBe(0);
+    expect(logger.error).toHaveBeenCalledWith(
+      expect.objectContaining({ job: "interest-accrual" }),
+      "keeper: job crashed",
+    );
+  });
+
+  it("handles lock provider acquisition failures gracefully", async () => {
+    const failingProvider: LockProvider = {
+      acquire: vi.fn().mockRejectedValue(new Error("Postgres connection timeout")),
+    };
+    const logger = makeMockLogger();
+    const jobFn = vi.fn();
+
+    const executed = await executeJobWithLock("repo-unwind", jobFn, failingProvider, logger);
+
+    expect(executed).toBe(false);
+    expect(jobFn).not.toHaveBeenCalled();
+    expect(logger.error).toHaveBeenCalledWith(
+      expect.objectContaining({ job: "repo-unwind", lockKey: "keeper:repo-unwind" }),
+      "keeper: error acquiring advisory lock",
+    );
+  });
+});