fix: Recover orphaned market-watcher jobs and actually start the watcher worker

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 `<jobKey>:recover:<attempt>:<leaseUntil>`.
+- **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
+  `<jobKey>:recover:<attempt>`, 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

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",

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),

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<boolean>;
+}
+
+export interface SweeperQueue {
+  add(
+    name: string,
+    data: MarketWatcherJobData,
+    options?: Record<string, unknown>,
+  ): Promise<unknown>;
+  getJob(id: string): Promise<{ getState(): Promise<string> } | 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<Redis, "set">): 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<unknown> = 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<SweepResult> {
+    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<void> {
+    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<boolean> {
+    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<string, unknown>,
+    };
+  }
+}

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],