diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 45bea0f..23138e4 100644 --- a/packages/db/src/client.ts +++ b/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 }; + diff --git a/services/keeper/README.md b/services/keeper/README.md index 6db463a..bd7c7b7 100644 --- a/services/keeper/README.md +++ b/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. diff --git a/services/keeper/src/index.ts b/services/keeper/src/index.ts index cda3684..1a4b2c7 100644 --- a/services/keeper/src/index.ts +++ b/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 = 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): Promise { +export function runJob( + name: string, + fn: () => Promise, + lockProvider: LockProvider = defaultLockProvider, +): Promise { 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): Promise { - 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, + lockProvider: LockProvider = defaultLockProvider, +): Promise { + 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 { // 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"); diff --git a/services/keeper/src/lock.test.ts b/services/keeper/src/lock.test.ts new file mode 100644 index 0000000..473b1b4 --- /dev/null +++ b/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(); + 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", + ); + }); +}); diff --git a/services/keeper/src/lock.ts b/services/keeper/src/lock.ts new file mode 100644 index 0000000..5c6bffa --- /dev/null +++ b/services/keeper/src/lock.ts @@ -0,0 +1,96 @@ +import type { RawSql } from "@ballast/db"; + +export type KeeperLockScope = "global" | "job"; + +export function getLockKey(name: string, scope?: KeeperLockScope): string { + const resolvedScope = scope ?? (process.env.KEEPER_LOCK_SCOPE as KeeperLockScope) ?? "job"; + return resolvedScope === "global" ? "keeper:global" : `keeper:${name}`; +} + +export interface LockClient { + unlock(): Promise; + release(): void | Promise; +} + +export interface LockProvider { + acquire(lockKey: string): Promise; +} + +export class PostgresLockProvider implements LockProvider { + constructor(private sqlClient: RawSql) {} + + async acquire(lockKey: string): Promise { + const reserved = await this.sqlClient.reserve(); + try { + const [res] = await reserved`SELECT pg_try_advisory_lock(hashtext(${lockKey})) as locked`; + if (!res?.locked) { + reserved.release(); + return null; + } + return { + unlock: async () => { + try { + await reserved`SELECT pg_advisory_unlock(hashtext(${lockKey}))`; + } catch { + // ignore + } + }, + release: () => { + reserved.release(); + }, + }; + } catch (err) { + reserved.release(); + throw err; + } + } +} + +export interface JobExecutionLogger { + info(obj: Record, msg?: string): void; + error(obj: Record, msg?: string): void; +} + +export async function executeJobWithLock( + name: string, + fn: () => Promise<{ processed: number; skipped: number; errors: Array<{ id?: string; message: string }> }>, + lockProvider: LockProvider, + logger: JobExecutionLogger, + scope?: KeeperLockScope, +): Promise { + const lockKey = getLockKey(name, scope); + let lockClient: LockClient | null; + try { + lockClient = await lockProvider.acquire(lockKey); + } catch (err) { + logger.error({ job: name, lockKey, err }, "keeper: error acquiring advisory lock"); + return false; + } + + if (!lockClient) { + logger.info({ job: name, lockKey }, "keeper: advisory lock busy; skipping tick"); + return false; + } + + logger.info({ job: name, lockKey }, "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"); + } + return true; + } catch (err) { + logger.error({ job: name, err }, "keeper: job crashed"); + return false; + } finally { + try { + await lockClient.unlock(); + } finally { + await lockClient.release(); + } + } +}