From 6e4f7e185db462c532e1db7afde0b8ed84fa7614 Mon Sep 17 00:00:00 2001 From: issue-solver-bot Date: Sun, 18 Oct 2026 23:22:10 +0100 Subject: [PATCH] fix: resolve issue #983 --- docs/job-queue.md | 28 +++ package.json | 4 +- src/config/env-schema.ts | 3 +- src/jobs/marketWatcherLeaseSweeper.ts | 167 +++++++++++++++++ src/metrics/registry.ts | 15 +- src/services/marketWatcherJobService.ts | 127 ++++++++++++- src/workers/marketWatcher.ts | 74 ++++++++ tests/marketWatcherJobService.test.ts | 234 +++++++++++++++++++++++- tests/marketWatcherLeaseSweeper.test.ts | 187 +++++++++++++++++++ tests/marketWatcherWorker.test.ts | 47 ++++- 10 files changed, 873 insertions(+), 13 deletions(-) create mode 100644 src/jobs/marketWatcherLeaseSweeper.ts create mode 100644 src/workers/marketWatcher.ts create mode 100644 tests/marketWatcherLeaseSweeper.test.ts diff --git a/docs/job-queue.md b/docs/job-queue.md index e6eb3a89..5507d2ef 100644 --- a/docs/job-queue.md +++ b/docs/job-queue.md @@ -131,6 +131,34 @@ await marketResolverWorker.handleEvent({ }); ``` +### Market watcher (leases, recovery, sweeper) + +Notification jobs live in Postgres (`market_watcher_jobs`) and are delivered +through BullMQ. A worker claims a job with a **lease** (`DEFAULT_LEASE_MS`, 5 min) +and every state change checks the lease token, so a stale holder can never commit. + +- **Renewal.** The handler receives `heartbeat()` and calls it between batches. + If renewal fails the handler throws `LeaseLostError`; the coordinator returns + `skipped` and nothing is committed. +- **Conflict retry.** If a claim fails while another lease is live, the job is + re-added delayed until `lease_until + 1s` with id `:recover::`. +- **Sweeper** (`src/jobs/marketWatcherLeaseSweeper.ts`). Every + `MARKET_WATCHER_SWEEP_INTERVAL_MS` (default 30000) one replica (Redis + `SET NX PX` lock) calls `recoverExpiredLeases`, which moves expired `running` + jobs to `retryable` (or `failed` once `DEFAULT_MAX_ATTEMPTS` is reached), then + re-enqueues due `retryable` jobs with no live BullMQ job under id + `:recover:`, since the original deterministic id is blocked. +- **Metrics.** `market_watcher_jobs_recovered_total`, + `market_watcher_lease_renewals_total{outcome=renewed|lost}`; each recovery is + logged with its `jobKey`. + +Run (worker plus sweeper, drains on SIGTERM/SIGINT): + +```bash +npm run market-watcher:dev # ts-node-dev +npm run build && npm run market-watcher +``` + ## Startup & shutdown — `src/index.ts` Workers are started after the database connection is established and stopped diff --git a/package.json b/package.json index bb8aa9ee..19fe61d9 100644 --- a/package.json +++ b/package.json @@ -13,7 +13,9 @@ "prestart": "npm run check-env", "start": "node dist/server.js", "indexer": "node dist/workers/indexer.js", - "indexer:dev": "ts-node-dev --respawn --transpile-only src/workers/indexer.ts", + "market-watcher": "node dist/workers/marketWatcher.js", + "market-watcher:dev": "ts-node-dev --respawn --transpile-only src/workers/marketWatcher.ts", + "indexer:dev":"ts-node-dev --respawn --transpile-only src/workers/indexer.ts", "lint": "eslint \"src/**/*.ts\"", "test": "jest", "test:unit": "jest --testPathIgnorePatterns=tests/e2e", diff --git a/src/config/env-schema.ts b/src/config/env-schema.ts index 433ead78..f62e0933 100644 --- a/src/config/env-schema.ts +++ b/src/config/env-schema.ts @@ -29,7 +29,8 @@ const baseSchema = z.object({ // ── Indexer tunables ────────────────────────────────────── INDEXER_POLL_INTERVAL_MS: z.coerce.number().int().positive().default(5000), - INDEXER_START_LEDGER: z.coerce.number().int().nonnegative().default(0), + MARKET_WATCHER_SWEEP_INTERVAL_MS: z.coerce.number().int().positive().default(30000), + INDEXER_START_LEDGER:z.coerce.number().int().nonnegative().default(0), INDEXER_REWIND_LEDGERS: z.coerce.number().int().nonnegative().default(100), INDEXER_BACKFILL_CHUNK_SIZE: z.coerce.number().int().positive().default(500), INDEXER_GAP_SCAN_INTERVAL_MS: z.coerce.number().int().positive().default(60000), diff --git a/src/jobs/marketWatcherLeaseSweeper.ts b/src/jobs/marketWatcherLeaseSweeper.ts new file mode 100644 index 00000000..7b663d53 --- /dev/null +++ b/src/jobs/marketWatcherLeaseSweeper.ts @@ -0,0 +1,167 @@ +import { randomUUID } from "node:crypto"; +import type { Redis } from "ioredis"; +import { logger } from "../config/logger"; +import { marketWatcherJobsRecoveredTotal } from "../metrics/registry"; +import { + DEFAULT_MAX_ATTEMPTS, + type MarketWatcherJobData, + type MarketWatcherJobRepo, +} from "../services/marketWatcherJobService"; +import type { MarketWatcherJob } from "../db/schema"; + +export const SWEEPER_LOCK_KEY = "market-watcher:sweeper:lock"; +const LIVE_STATES = new Set(["waiting", "delayed", "active", "prioritized", "waiting-children"]); + +export interface SweeperLock { + /** Returns true when this caller now holds the lock for ttlMs. */ + acquire(key: string, ttlMs: number): Promise; +} + +export interface SweeperQueue { + add( + name: string, + data: MarketWatcherJobData, + options?: Record, + ): Promise; + getJob(id: string): Promise<{ getState(): Promise } | null | undefined>; +} + +export interface MarketWatcherLeaseSweeperOptions { + queue: SweeperQueue; + lock: SweeperLock; + intervalMs: number; + maxAttempts?: number; + batchLimit?: number; + now?: () => Date; +} + +export interface SweepResult { + locked: boolean; + recovered: number; + enqueued: number; +} + +/** Redis `SET key token NX PX ttl` lock so only one replica sweeps per interval. */ +export function createRedisSweeperLock(redis: Pick): SweeperLock { + return { + async acquire(key, ttlMs) { + const res = await redis.set(key, randomUUID(), "PX", ttlMs, "NX"); + return res === "OK"; + }, + }; +} + +/** + * Recovers market-watcher jobs whose worker died: expired leases go back to + * `retryable` (or `failed` when attempts are exhausted) and every due retryable + * job without a live BullMQ job is re-enqueued under a fresh `:recover:` id. + */ +export class MarketWatcherLeaseSweeper { + private readonly queue: SweeperQueue; + private readonly lock: SweeperLock; + private readonly intervalMs: number; + private readonly maxAttempts: number; + private readonly batchLimit: number; + private readonly now: () => Date; + private timer: NodeJS.Timeout | null = null; + private inFlight: Promise = Promise.resolve(); + private running = false; + + constructor( + private readonly repository: MarketWatcherJobRepo, + options: MarketWatcherLeaseSweeperOptions, + ) { + this.queue = options.queue; + this.lock = options.lock; + this.intervalMs = options.intervalMs; + this.maxAttempts = options.maxAttempts ?? DEFAULT_MAX_ATTEMPTS; + this.batchLimit = options.batchLimit ?? 100; + this.now = options.now ?? (() => new Date()); + } + + async sweepOnce(): Promise { + const acquired = await this.lock.acquire(SWEEPER_LOCK_KEY, this.intervalMs); + if (!acquired) { + return { locked: false, recovered: 0, enqueued: 0 }; + } + + const now = this.now(); + const expired = await this.repository.recoverExpiredLeases(now, this.maxAttempts); + const recoveredRetryable = expired.filter((j) => j.status === "retryable"); + for (const job of expired) { + logger.warn( + { jobId: job.id, jobKey: job.jobKey, status: job.status, attempt: job.attempt }, + "market_watcher_job_lease_recovered", + ); + } + marketWatcherJobsRecoveredTotal.inc(expired.length); + + const due = await this.repository.listDueRetryable(now, this.batchLimit); + const seen = new Set(recoveredRetryable.map((j) => j.id)); + const candidates = [...recoveredRetryable, ...due.filter((j) => !seen.has(j.id))]; + + let enqueued = 0; + for (const job of candidates) { + if (await this.hasLiveQueueJob(job)) continue; + await this.queue.add("notify-watchers", this.toJobData(job), { + jobId: `${job.jobKey}:recover:${job.attempt}`, + removeOnComplete: false, + removeOnFail: false, + }); + enqueued++; + } + + return { locked: true, recovered: expired.length, enqueued }; + } + + start(): void { + if (this.running) return; + this.running = true; + this.schedule(); + } + + async stop(): Promise { + this.running = false; + if (this.timer) { + clearTimeout(this.timer); + this.timer = null; + } + await this.inFlight; + } + + private schedule(): void { + this.timer = setTimeout(() => { + this.inFlight = this.sweepOnce() + .catch((err) => logger.error({ err }, "market_watcher_sweep_failed")) + .finally(() => { + if (this.running) this.schedule(); + }); + }, this.intervalMs); + } + + private async hasLiveQueueJob(job: MarketWatcherJob): Promise { + const ids = [ + job.jobKey, + `${job.jobKey}:retry:${job.attempt}`, + `${job.jobKey}:recover:${job.attempt}`, + ]; + for (const id of ids) { + const queued = await this.queue.getJob(id); + if (queued && LIVE_STATES.has(await queued.getState())) return true; + } + return false; + } + + private toJobData(job: MarketWatcherJob): MarketWatcherJobData { + const prefix = `${job.marketId}:${job.eventType}:`; + const eventRef = job.jobKey.startsWith(prefix) ? job.jobKey.slice(prefix.length) : job.jobKey; + return { + jobId: job.id, + marketId: job.marketId, + eventType: job.eventType, + eventRef, + jobKey: job.jobKey, + payload: (job.payload ?? {}) as Record, + }; + } +} diff --git a/src/metrics/registry.ts b/src/metrics/registry.ts index f2df6f3d..39762889 100644 --- a/src/metrics/registry.ts +++ b/src/metrics/registry.ts @@ -98,7 +98,20 @@ export const marketWatcherLeaseConflictsTotal = new Counter({ registers: [register], }); -export const marketWatcherNotificationsTotal = new Counter({ +export const marketWatcherJobsRecoveredTotal = new Counter({ + name: "market_watcher_jobs_recovered_total", + help: "Market watcher jobs recovered from expired leases or stuck retryable state", + registers: [register], +}); + +export const marketWatcherLeaseRenewalsTotal = new Counter({ + name: "market_watcher_lease_renewals_total", + help: "Market watcher lease renewal attempts by outcome", + labelNames: ["outcome"] as const, + registers: [register], +}); + +export const marketWatcherNotificationsTotal =new Counter({ name: "market_watcher_notifications_total", help: "Total number of market watcher notifications generated and dispatched", registers: [register], diff --git a/src/services/marketWatcherJobService.ts b/src/services/marketWatcherJobService.ts index 87cd7133..dadcbffd 100644 --- a/src/services/marketWatcherJobService.ts +++ b/src/services/marketWatcherJobService.ts @@ -7,7 +7,7 @@ */ import { randomUUID } from "node:crypto"; -import { and, eq, isNull, lt, lte, or, sql } from "drizzle-orm"; +import { and, eq, gte, isNull, lt, lte, or, sql } from "drizzle-orm"; import type { Db } from "../db"; import { db as defaultDb } from "../db"; import { @@ -20,6 +20,7 @@ import { marketWatcherJobRetriesTotal, marketWatcherJobRunsTotal, marketWatcherLeaseConflictsTotal, + marketWatcherLeaseRenewalsTotal, marketWatcherNotificationsTotal, } from "../metrics/registry"; import { logger } from "../config/logger"; @@ -52,6 +53,19 @@ export interface MarketWatcherJobResult { export interface MarketWatcherJobHandlerInput extends MarketWatcherJobData { attempt: number; + /** + * Extends the job lease. Rejects with LeaseLostError when the lease was lost, + * in which case the handler must stop before doing further work. + */ + heartbeat?: () => Promise; +} + +/** Thrown by heartbeat() when the lease could not be renewed. */ +export class LeaseLostError extends Error { + constructor(message = "market watcher job lease lost") { + super(message); + this.name = "LeaseLostError"; + } } export type MarketWatcherNotificationHandler = ( @@ -85,7 +99,18 @@ export interface MarketWatcherJobRepo { maxAttempts: number; nextAttemptAt: Date; }): Promise<{ status: "retryable" | "failed"; attempt: number } | null>; - recoverExpiredLeases(now: Date): Promise; + renewLease( + jobId: string, + leaseToken: string, + now: Date, + leaseMs: number, + ): Promise; + /** + * Moves expired `running` jobs to `retryable`, or to `failed` once their + * attempts are exhausted. Returns the affected rows. + */ + recoverExpiredLeases(now: Date, maxAttempts?: number): Promise; + listDueRetryable(now: Date, limit: number): Promise; getJob(jobId: string): Promise; } @@ -262,11 +287,53 @@ export class DrizzleMarketWatcherJobRepo implements MarketWatcherJobRepo { return { status, attempt: job.attempt }; } - async recoverExpiredLeases(now: Date): Promise { + async renewLease( + jobId: string, + leaseToken: string, + now: Date, + leaseMs: number, + ): Promise { + const result = await this.database + .update(marketWatcherJobs) + .set({ leaseUntil: new Date(now.getTime() + leaseMs), updatedAt: now }) + .where( + and( + eq(marketWatcherJobs.id, jobId), + eq(marketWatcherJobs.leaseToken, leaseToken), + eq(marketWatcherJobs.status, "running"), + gte(marketWatcherJobs.leaseUntil, now), + ), + ) + .returning({ id: marketWatcherJobs.id }); + return result.length === 1; + } + + async listDueRetryable(now: Date, limit: number): Promise { + return this.database + .select() + .from(marketWatcherJobs) + .where( + and( + eq(marketWatcherJobs.status, "retryable"), + or( + isNull(marketWatcherJobs.nextAttemptAt), + lte(marketWatcherJobs.nextAttemptAt, now), + ), + ), + ) + .limit(limit); + } + + async recoverExpiredLeases( + now: Date, + maxAttempts: number = DEFAULT_MAX_ATTEMPTS, + ): Promise { return this.database .update(marketWatcherJobs) .set({ - status: "retryable", + status: sql`CASE WHEN ${marketWatcherJobs.attempt} >= ${maxAttempts} THEN 'failed' ELSE 'retryable' END`, + lastError: "lease expired before completion", + completedAt: sql`CASE WHEN ${marketWatcherJobs.attempt} >= ${maxAttempts} THEN ${now} ELSE NULL END`, leaseToken: null, leaseUntil: null, nextAttemptAt: now, @@ -415,10 +482,24 @@ export class MarketWatcherJobCoordinator { ); if (!claimed) { + await this.rescheduleOnLiveLease(job); logger.info({ jobId: job.jobId, jobKey: job.jobKey }, "market_watcher_job_lease_skipped"); return "skipped"; } + const heartbeat = async (): Promise => { + const renewed = await this.repository.renewLease( + claimed.id, + leaseToken, + this.now(), + this.leaseMs, + ); + marketWatcherLeaseRenewalsTotal.inc({ outcome: renewed ? "renewed" : "lost" }); + if (!renewed) { + throw new LeaseLostError(); + } + }; + try { const result = await handler({ jobId: claimed.id, @@ -428,6 +509,7 @@ export class MarketWatcherJobCoordinator { jobKey: claimed.jobKey, payload: job.payload, attempt: claimed.attempt, + heartbeat, }); if (!Number.isInteger(result.watchersNotified) || result.watchersNotified < 0) { @@ -451,6 +533,13 @@ export class MarketWatcherJobCoordinator { return "succeeded"; } catch (error) { + if (error instanceof LeaseLostError) { + logger.warn( + { jobId: job.jobId, jobKey: job.jobKey, leaseToken }, + "market_watcher_job_lease_lost", + ); + return "skipped"; + } const message = error instanceof Error ? error.message : "unknown market watcher notification failure"; const retryAt = new Date( @@ -492,6 +581,26 @@ export class MarketWatcherJobCoordinator { return failed.status; } } + + /** + * When a claim fails because another worker's lease is still live, schedule a + * delayed re-add just after the lease ends so a crashed holder cannot lose the job. + */ + private async rescheduleOnLiveLease(job: MarketWatcherJobData): Promise { + const stored = await this.repository.getJob(job.jobId); + if (!stored || stored.status !== "running" || !stored.leaseUntil) return; + + const nowMs = this.now().getTime(); + const leaseUntilMs = stored.leaseUntil.getTime(); + if (leaseUntilMs < nowMs) return; + + await this.queue.add("notify-watchers", job, { + jobId: `${job.jobKey}:recover:${stored.attempt}:${leaseUntilMs}`, + delay: leaseUntilMs + 1000 - nowMs, + removeOnComplete: false, + removeOnFail: false, + }); + } } /** @@ -501,6 +610,7 @@ export class MarketWatcherJobCoordinator { export async function defaultMarketWatcherHandler( input: MarketWatcherJobHandlerInput, database: Db = defaultDb, + batchSize = 500, ): Promise { const watchers = await database .select({ userId: marketWatchers.userId }) @@ -522,9 +632,12 @@ export async function defaultMarketWatcherHandler( data: input.payload ?? {}, })); - const BATCH_SIZE = 500; - for (let i = 0; i < rows.length; i += BATCH_SIZE) { - const chunk = rows.slice(i, i + BATCH_SIZE); + for (let i = 0; i < rows.length; i += batchSize) { + // Renew between batches; throws LeaseLostError so no further batch is written. + if (i > 0 && input.heartbeat) { + await input.heartbeat(); + } + const chunk = rows.slice(i, i + batchSize); await database.insert(notifications).values(chunk); } diff --git a/src/workers/marketWatcher.ts b/src/workers/marketWatcher.ts new file mode 100644 index 00000000..9f9bfc23 --- /dev/null +++ b/src/workers/marketWatcher.ts @@ -0,0 +1,74 @@ +import { env } from "../config/env"; +import { logger } from "../config/logger"; +import { pool } from "../db/client"; +import { + MarketWatcherLeaseSweeper, + createRedisSweeperLock, +} from "../jobs/marketWatcherLeaseSweeper"; +import { DrizzleMarketWatcherJobRepo } from "../services/marketWatcherJobService"; +import { marketWatcherQueue, redisConnection } from "../queue"; +import { MarketWatcherWorker } from "./marketWatcherWorker"; + +export interface MarketWatcherProcess { + shutdown(): Promise; +} + +/** + * Starts the market-watcher worker and the lease sweeper. `shutdown()` stops the + * sweeper, then closes the BullMQ worker (which waits for in-flight jobs). + */ +export function startMarketWatcher( + worker: Pick, + sweeper: Pick, + cleanup: () => Promise = async () => undefined, +): MarketWatcherProcess { + worker.start(); + sweeper.start(); + let shutdownPromise: Promise | null = null; + return { + shutdown() { + shutdownPromise ??= (async () => { + await sweeper.stop(); + await worker.stop(); + await cleanup(); + })(); + return shutdownPromise; + }, + }; +} + +async function main(): Promise { + const repo = new DrizzleMarketWatcherJobRepo(); + const worker = new MarketWatcherWorker(repo); + const sweeper = new MarketWatcherLeaseSweeper(repo, { + queue: marketWatcherQueue, + lock: createRedisSweeperLock(redisConnection), + intervalMs: env.MARKET_WATCHER_SWEEP_INTERVAL_MS, + }); + + const proc = startMarketWatcher(worker, sweeper, async () => { + await marketWatcherQueue.close(); + await pool.end(); + redisConnection.disconnect(); + }); + logger.info({ sweepIntervalMs: env.MARKET_WATCHER_SWEEP_INTERVAL_MS }, "market watcher process started"); + + await new Promise((resolve) => { + const onSignal = (signal: NodeJS.Signals): void => { + logger.info({ signal }, "market watcher shutdown requested; draining"); + void proc.shutdown().finally(resolve); + }; + process.once("SIGTERM", onSignal); + process.once("SIGINT", onSignal); + }); + logger.info({}, "market watcher process stopped"); +} + +if (require.main === module) { + main() + .then(() => process.exit(0)) + .catch((err) => { + logger.error({ err }, "market watcher process crashed"); + process.exit(1); + }); +} diff --git a/tests/marketWatcherJobService.test.ts b/tests/marketWatcherJobService.test.ts index 1793ac31..fe8af0c9 100644 --- a/tests/marketWatcherJobService.test.ts +++ b/tests/marketWatcherJobService.test.ts @@ -17,7 +17,9 @@ import { MarketWatcherJobCoordinator, enqueueMarketWatcherJob, defaultMarketWatcherHandler, + LeaseLostError, } from "../src/services/marketWatcherJobService"; +import { MarketWatcherLeaseSweeper } from "../src/jobs/marketWatcherLeaseSweeper"; import type { MarketWatcherJob } from "../src/db/schema"; class InMemoryMarketWatcherJobRepo implements MarketWatcherJobRepo { @@ -129,11 +131,42 @@ class InMemoryMarketWatcherJobRepo implements MarketWatcherJobRepo { return { status: job.status, attempt: job.attempt }; } - async recoverExpiredLeases(now: Date): Promise { + async renewLease( + jobId: string, + leaseToken: string, + now: Date, + leaseMs: number, + ): Promise { + const job = this.jobs.get(jobId); + if ( + !job || + job.status !== "running" || + job.leaseToken !== leaseToken || + !job.leaseUntil || + job.leaseUntil < now + ) { + return false; + } + job.leaseUntil = new Date(now.getTime() + leaseMs); + return true; + } + + async listDueRetryable(now: Date, limit: number): Promise { + return Array.from(this.jobs.values()) + .filter((j) => j.status === "retryable" && (!j.nextAttemptAt || j.nextAttemptAt <= now)) + .slice(0, limit) + .map((j) => ({ ...j })); + } + + async recoverExpiredLeases( + now: Date, + maxAttempts: number = DEFAULT_MAX_ATTEMPTS, + ): Promise { const recovered: MarketWatcherJob[] = []; for (const job of this.jobs.values()) { if (job.status === "running" && job.leaseUntil && job.leaseUntil < now) { - job.status = "retryable"; + job.status = job.attempt >= maxAttempts ? "failed" : "retryable"; + job.completedAt = job.status === "failed" ? now : null; job.leaseToken = null; job.leaseUntil = null; job.nextAttemptAt = now; @@ -503,4 +536,201 @@ describe("marketWatcherJobService", () => { expect(recovered[0].leaseToken).toBeNull(); }); }); + + describe("recovery, renewal and conflict retry", () => { + const T0 = new Date("2026-08-29T10:00:00.000Z"); + const dataFor = (job: MarketWatcherJob): MarketWatcherJobData => ({ + jobId: job.id, + marketId: job.marketId, + eventType: job.eventType, + eventRef: "evt-1", + jobKey: job.jobKey, + }); + const newJob = (repo: InMemoryMarketWatcherJobRepo) => + repo.createOrGetJob({ + marketId: "mkt-1", + jobKey: "mkt-1:market.resolved:evt-1", + eventType: "market.resolved", + }); + + it("recovers a crashed job via the sweeper, re-enqueues with a :recover: id, and succeeds", async () => { + const repo = new InMemoryMarketWatcherJobRepo(); + const job = await newJob(repo); + // Worker crashed: claimed, lease 1 minute, never completed. + await repo.claimJob(job.id, "dead-worker", T0, 60_000); + + const added: Array<{ data: MarketWatcherJobData; opts: Record }> = []; + const queue = { + add: jest.fn(async (_n: string, data: MarketWatcherJobData, opts: Record) => { + added.push({ data, opts }); + }), + // The original deterministic id exists but is completed. + getJob: jest.fn(async () => ({ getState: async () => "completed" })), + }; + let now = new Date(T0.getTime() + 120_000); + const sweeper = new MarketWatcherLeaseSweeper(repo, { + queue, + lock: { acquire: async () => true }, + intervalMs: 1000, + now: () => now, + }); + + const result = await sweeper.sweepOnce(); + expect(result).toEqual({ locked: true, recovered: 1, enqueued: 1 }); + expect((await repo.getJob(job.id))?.status).toBe("retryable"); + expect(added[0].opts.jobId).toBe(`${job.jobKey}:recover:1`); + + const coordinator = new MarketWatcherJobCoordinator(repo, { now: () => now, queue }); + const handler = jest.fn().mockResolvedValue({ watchersNotified: 2 }); + now = new Date(now.getTime() + 1000); + expect(await coordinator.process(added[0].data, handler)).toBe("succeeded"); + expect((await repo.getJob(job.id))?.status).toBe("succeeded"); + }); + + it("fails a poison job after maxAttempts including recovered attempts", async () => { + const repo = new InMemoryMarketWatcherJobRepo(); + const job = await newJob(repo); + const queue = { + add: jest.fn().mockResolvedValue({}), + getJob: jest.fn().mockResolvedValue(null), + }; + let now = T0; + const sweeper = new MarketWatcherLeaseSweeper(repo, { + queue, + lock: { acquire: async () => true }, + intervalMs: 1000, + maxAttempts: 2, + now: () => now, + }); + + for (let i = 0; i < 2; i++) { + await repo.claimJob(job.id, `dead-${i}`, now, 1000); + now = new Date(now.getTime() + 5000); + await sweeper.sweepOnce(); + } + + const final = await repo.getJob(job.id); + expect(final?.status).toBe("failed"); + expect(final?.attempt).toBe(2); + expect(queue.add).toHaveBeenCalledTimes(1); + }); + + it("schedules a delayed recover re-add at lease_until + 1s on a live lease conflict", async () => { + const repo = new InMemoryMarketWatcherJobRepo(); + const job = await newJob(repo); + await repo.claimJob(job.id, "holder", T0, 60_000); + const queue = { add: jest.fn().mockResolvedValue({}) }; + const now = new Date(T0.getTime() + 10_000); + const coordinator = new MarketWatcherJobCoordinator(repo, { now: () => now, queue }); + const handler = jest.fn(); + + expect(await coordinator.process(dataFor(job), handler)).toBe("skipped"); + expect(handler).not.toHaveBeenCalled(); + expect(queue.add).toHaveBeenCalledTimes(1); + const [, , opts] = queue.add.mock.calls[0]; + expect(opts.delay).toBe(51_000); + expect(String(opts.jobId)).toContain(`${job.jobKey}:recover:1:`); + }); + + it("does not reschedule when the conflicting job is not running", async () => { + const repo = new InMemoryMarketWatcherJobRepo(); + const job = await newJob(repo); + await repo.claimJob(job.id, "holder", T0, 60_000); + await repo.markSucceeded(job.id, "holder", 1, T0); + const queue = { add: jest.fn() }; + const coordinator = new MarketWatcherJobCoordinator(repo, { now: () => T0, queue }); + + expect(await coordinator.process(dataFor(job), jest.fn())).toBe("skipped"); + expect(queue.add).not.toHaveBeenCalled(); + }); + + it("renews the lease between batches of a long fan-out and succeeds once", async () => { + const repo = new InMemoryMarketWatcherJobRepo(); + const job = await newJob(repo); + let now = T0; + const coordinator = new MarketWatcherJobCoordinator(repo, { + now: () => now, + leaseMs: 1000, + queue: { add: jest.fn() }, + }); + const batches = jest.fn(); + const handler = async (input: { heartbeat?: () => Promise }) => { + for (let i = 0; i < 3; i++) { + if (i > 0) await input.heartbeat?.(); + batches(); + now = new Date(now.getTime() + 800); // each batch nearly exhausts the lease + } + return { watchersNotified: 3 }; + }; + + expect(await coordinator.process(dataFor(job), handler)).toBe("succeeded"); + expect(batches).toHaveBeenCalledTimes(3); + expect((await repo.getJob(job.id))?.status).toBe("succeeded"); + }); + + it("stops before the next batch and returns skipped when renewal fails", async () => { + const repo = new InMemoryMarketWatcherJobRepo(); + const job = await newJob(repo); + let now = T0; + const coordinator = new MarketWatcherJobCoordinator(repo, { + now: () => now, + leaseMs: 1000, + queue: { add: jest.fn() }, + }); + const batches = jest.fn(); + const handler = async (input: { heartbeat?: () => Promise }) => { + for (let i = 0; i < 3; i++) { + if (i > 0) await input.heartbeat?.(); + batches(); + now = new Date(now.getTime() + 2000); // lease expires during the batch + } + return { watchersNotified: 3 }; + }; + + expect(await coordinator.process(dataFor(job), handler)).toBe("skipped"); + expect(batches).toHaveBeenCalledTimes(1); + expect((await repo.getJob(job.id))?.status).toBe("running"); + }); + + it("defaultMarketWatcherHandler heartbeats between batches and stops on LeaseLostError", async () => { + const watchers = [{ userId: "u1" }, { userId: "u2" }, { userId: "u3" }]; + const values = jest.fn().mockResolvedValue(undefined); + const database = { + select: () => ({ from: () => ({ where: async () => watchers }) }), + insert: () => ({ values }), + } as never; + const base = { + jobId: "j", + marketId: "m", + eventType: "e", + eventRef: "r", + jobKey: "m:e:r", + attempt: 1, + }; + + const heartbeat = jest.fn().mockResolvedValue(undefined); + const ok = await defaultMarketWatcherHandler({ ...base, heartbeat }, database, 1); + expect(ok.watchersNotified).toBe(3); + expect(values).toHaveBeenCalledTimes(3); + expect(heartbeat).toHaveBeenCalledTimes(2); + + values.mockClear(); + const lost = jest.fn().mockRejectedValue(new LeaseLostError()); + await expect( + defaultMarketWatcherHandler({ ...base, heartbeat: lost }, database, 1), + ).rejects.toBeInstanceOf(LeaseLostError); + expect(values).toHaveBeenCalledTimes(1); + }); + + it("returns zero without heartbeat when there are no watchers", async () => { + const database = { + select: () => ({ from: () => ({ where: async () => [] }) }), + } as never; + const res = await defaultMarketWatcherHandler( + { jobId: "j", marketId: "m", eventType: "e", eventRef: "r", jobKey: "k", attempt: 1 }, + database, + ); + expect(res.watchersNotified).toBe(0); + }); + }); }); diff --git a/tests/marketWatcherLeaseSweeper.test.ts b/tests/marketWatcherLeaseSweeper.test.ts new file mode 100644 index 00000000..23b4a1a4 --- /dev/null +++ b/tests/marketWatcherLeaseSweeper.test.ts @@ -0,0 +1,187 @@ +import { + MarketWatcherLeaseSweeper, + SWEEPER_LOCK_KEY, + createRedisSweeperLock, + type SweeperLock, +} from "../src/jobs/marketWatcherLeaseSweeper"; +import type { MarketWatcherJobRepo } from "../src/services/marketWatcherJobService"; +import type { MarketWatcherJob } from "../src/db/schema"; + +const NOW = new Date("2026-08-29T10:00:00.000Z"); + +function makeJob(overrides: Partial = {}): MarketWatcherJob { + return { + id: "job-1", + marketId: "mkt-1", + jobKey: "mkt-1:market.resolved:evt-1", + eventType: "market.resolved", + status: "retryable", + attempt: 1, + leaseToken: null, + leaseUntil: null, + startedAt: null, + completedAt: null, + nextAttemptAt: NOW, + watchersNotified: 0, + payload: { a: 1 }, + lastError: null, + createdAt: NOW, + updatedAt: NOW, + ...overrides, + }; +} + +function makeRepo(recovered: MarketWatcherJob[] = [], due: MarketWatcherJob[] = []) { + return { + recoverExpiredLeases: jest.fn().mockResolvedValue(recovered), + listDueRetryable: jest.fn().mockResolvedValue(due), + } as unknown as MarketWatcherJobRepo & { + recoverExpiredLeases: jest.Mock; + listDueRetryable: jest.Mock; + }; +} + +function makeQueue(states: Record = {}) { + return { + add: jest.fn().mockResolvedValue({}), + getJob: jest.fn(async (id: string) => + states[id] ? { getState: async () => states[id] } : null, + ), + }; +} + +describe("MarketWatcherLeaseSweeper", () => { + it("recovers expired leases and re-enqueues with a :recover: id", async () => { + const job = makeJob({ attempt: 2 }); + const repo = makeRepo([job], [job]); + const queue = makeQueue({ [job.jobKey]: "completed" }); + const sweeper = new MarketWatcherLeaseSweeper(repo, { + queue, + lock: { acquire: async () => true }, + intervalMs: 1000, + now: () => NOW, + }); + + const res = await sweeper.sweepOnce(); + + expect(res).toEqual({ locked: true, recovered: 1, enqueued: 1 }); + expect(repo.recoverExpiredLeases).toHaveBeenCalledWith(NOW, 3); + expect(queue.add).toHaveBeenCalledTimes(1); + expect(queue.add).toHaveBeenCalledWith( + "notify-watchers", + expect.objectContaining({ jobId: "job-1", eventRef: "evt-1", payload: { a: 1 } }), + expect.objectContaining({ jobId: `${job.jobKey}:recover:2` }), + ); + }); + + it("re-enqueues due retryable jobs but skips ones with a live queue job", async () => { + const live = makeJob({ id: "live", jobKey: "k-live", marketId: "m", eventType: "e" }); + const dead = makeJob({ id: "dead", jobKey: "k-dead", marketId: "m", eventType: "e" }); + const repo = makeRepo([], [live, dead]); + const queue = makeQueue({ "k-live:retry:1": "delayed" }); + const sweeper = new MarketWatcherLeaseSweeper(repo, { + queue, + lock: { acquire: async () => true }, + intervalMs: 1000, + now: () => NOW, + }); + + const res = await sweeper.sweepOnce(); + + expect(res.enqueued).toBe(1); + expect(queue.add).toHaveBeenCalledWith( + "notify-watchers", + expect.objectContaining({ jobId: "dead", eventRef: "k-dead" }), + expect.objectContaining({ jobId: "k-dead:recover:1" }), + ); + }); + + it("does not enqueue jobs that became failed on recovery", async () => { + const failed = makeJob({ status: "failed", attempt: 3 }); + const repo = makeRepo([failed], []); + const queue = makeQueue(); + const sweeper = new MarketWatcherLeaseSweeper(repo, { + queue, + lock: { acquire: async () => true }, + intervalMs: 1000, + now: () => NOW, + }); + + expect(await sweeper.sweepOnce()).toEqual({ locked: true, recovered: 1, enqueued: 0 }); + expect(queue.add).not.toHaveBeenCalled(); + }); + + it("lets only one of two replicas sweep per tick", async () => { + const held = new Set(); + const lock: SweeperLock = { + async acquire(key) { + if (held.has(key)) return false; + held.add(key); + return true; + }, + }; + const job = makeJob(); + const repoA = makeRepo([job], []); + const repoB = makeRepo([job], []); + const queue = makeQueue(); + const mk = (repo: MarketWatcherJobRepo) => + new MarketWatcherLeaseSweeper(repo, { queue, lock, intervalMs: 1000, now: () => NOW }); + + const [a, b] = await Promise.all([mk(repoA).sweepOnce(), mk(repoB).sweepOnce()]); + + expect([a.locked, b.locked].sort()).toEqual([false, true]); + expect(queue.add).toHaveBeenCalledTimes(1); + expect(repoA.recoverExpiredLeases.mock.calls.length + repoB.recoverExpiredLeases.mock.calls.length).toBe(1); + }); + + it("runs on a timer until stopped", async () => { + jest.useFakeTimers(); + try { + const repo = makeRepo(); + const sweeper = new MarketWatcherLeaseSweeper(repo, { + queue: makeQueue(), + lock: { acquire: async () => true }, + intervalMs: 1000, + }); + sweeper.start(); + sweeper.start(); + await jest.advanceTimersByTimeAsync(2100); + expect(repo.recoverExpiredLeases.mock.calls.length).toBe(2); + await sweeper.stop(); + await jest.advanceTimersByTimeAsync(5000); + expect(repo.recoverExpiredLeases.mock.calls.length).toBe(2); + } finally { + jest.useRealTimers(); + } + }); + + it("keeps ticking when a sweep throws", async () => { + jest.useFakeTimers(); + try { + const repo = makeRepo(); + repo.recoverExpiredLeases.mockRejectedValueOnce(new Error("db down")); + const sweeper = new MarketWatcherLeaseSweeper(repo, { + queue: makeQueue(), + lock: { acquire: async () => true }, + intervalMs: 1000, + }); + sweeper.start(); + await jest.advanceTimersByTimeAsync(2100); + expect(repo.recoverExpiredLeases.mock.calls.length).toBe(2); + await sweeper.stop(); + } finally { + jest.useRealTimers(); + } + }); +}); + +describe("createRedisSweeperLock", () => { + it("uses SET NX PX and reports acquisition", async () => { + const set = jest.fn().mockResolvedValueOnce("OK").mockResolvedValueOnce(null); + const lock = createRedisSweeperLock({ set } as never); + + expect(await lock.acquire(SWEEPER_LOCK_KEY, 500)).toBe(true); + expect(await lock.acquire(SWEEPER_LOCK_KEY, 500)).toBe(false); + expect(set).toHaveBeenCalledWith(SWEEPER_LOCK_KEY, expect.any(String), "PX", 500, "NX"); + }); +}); diff --git a/tests/marketWatcherWorker.test.ts b/tests/marketWatcherWorker.test.ts index d7af39c9..7fc9f4ad 100644 --- a/tests/marketWatcherWorker.test.ts +++ b/tests/marketWatcherWorker.test.ts @@ -8,11 +8,12 @@ jest.mock("bullmq", () => ({ jest.mock("../src/queue", () => ({ redisConnection: {}, + marketWatcherQueue: {}, marketWatcherQueueName: "market-watcher-jobs", })); - import { Worker } from "bullmq"; import { MarketWatcherWorker } from "../src/workers/marketWatcherWorker"; +import { startMarketWatcher } from "../src/workers/marketWatcher"; import { MarketWatcherJobRepo } from "../src/services/marketWatcherJobService"; describe("MarketWatcherWorker", () => { @@ -36,3 +37,47 @@ describe("MarketWatcherWorker", () => { expect(instance.close).toHaveBeenCalledTimes(1); }); }); + +describe("startMarketWatcher entrypoint", () => { + it("starts worker and sweeper, and drains in order on shutdown (idempotent)", async () => { + const order: string[] = []; + let finishJob!: () => void; + const inFlight = new Promise((r) => (finishJob = r)); + const worker = { + start: jest.fn(() => order.push("worker.start")), + stop: jest.fn(async () => { + await inFlight; // worker.close waits for the in-flight job + order.push("worker.stop"); + }), + }; + const sweeper = { + start: jest.fn(() => order.push("sweeper.start")), + stop: jest.fn(async () => void order.push("sweeper.stop")), + }; + const cleanup = jest.fn(async () => void order.push("cleanup")); + + const proc = startMarketWatcher(worker, sweeper, cleanup); + const first = proc.shutdown(); + const second = proc.shutdown(); + order.push("signal-handled"); + finishJob(); + await Promise.all([first, second]); + + expect(order).toEqual([ + "worker.start", + "sweeper.start", + "signal-handled", + "sweeper.stop", + "worker.stop", + "cleanup", + ]); + expect(worker.stop).toHaveBeenCalledTimes(1); + }); + + it("works with the default no-op cleanup", async () => { + const worker = { start: jest.fn(), stop: jest.fn().mockResolvedValue(undefined) }; + const sweeper = { start: jest.fn(), stop: jest.fn().mockResolvedValue(undefined) }; + await startMarketWatcher(worker, sweeper).shutdown(); + expect(worker.stop).toHaveBeenCalled(); + }); +});