Skip to content

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

Merged
greatest0fallt1me merged 1 commit into
Predictify-org:mainfrom
innocentharuna:solver/issue-983-45-a0
Oct 19, 2026
Merged

greatest0fallt1me merged 1 commit into
Predictify-org:mainfrom
innocentharuna:solver/issue-983-45-a0

Conversation

@innocentharuna

Copy link
Copy Markdown
Contributor

Summary

I fixed #983 in the code, but nothing has been run. I couldn't run anything here, so the new tests, lint and tsc are all unverified until CI runs.

Each in-scope item is implemented, but I took a few liberties:

  • Heartbeat: heartbeat is an optional field on the handler input, not a second argument. defaultMarketWatcherHandler already uses its second argument for the database. The handler also takes an optional batch size so the tests can use 3 batches without 1,500 watchers.
  • Poison jobs: recoverExpiredLeases(now, maxAttempts) now marks a job failed once its attempts are used up, so a job that keeps crashing workers can't loop forever.
  • Conflict retry: on a live lease conflict, the delayed re-add uses id <jobKey>:recover:<attempt>:<leaseUntilMs>, so it can't collide with the sweeper's :recover:<attempt> id.
  • Sweeper lock: the Redis lock is never released and just expires after one sweep interval, so only one replica sweeps per interval.
  • Entrypoint: startMarketWatcher is exported from src/workers/marketWatcher.ts so the drain order can be tested without real signals.

What I changed:

  • Service (src/services/marketWatcherJobService.ts):
    • Added renewLease and listDueRetryable to the repo, plus an exported LeaseLostError.
    • heartbeat() is passed to the handler. Losing the lease makes the job return skipped, with no succeeded commit.
    • process() now re-adds the job delayed until lease_until + 1s when a claim fails on a live lease.
  • Sweeper (src/jobs/marketWatcherLeaseSweeper.ts):
    • It takes a Redis lock, calls recoverExpiredLeases and picks up due retryable jobs.
    • It skips any job that still has a live BullMQ job, and re-enqueues the rest with jobId: "<jobKey>:recover:<attempt>".
    • It logs each recovery with its jobKey.
  • Entrypoint and scripts: src/workers/marketWatcher.ts starts the worker and the sweeper and drains on SIGTERM/SIGINT. I added the market-watcher and market-watcher:dev scripts to package.json.
  • Config and metrics: MARKET_WATCHER_SWEEP_INTERVAL_MS (default 30000) and the counters market_watcher_jobs_recovered_total and market_watcher_lease_renewals_total{outcome}.
  • Docs: a market-watcher section in docs/job-queue.md.

The PR still needs the sequence diagram or timeline for the crash-recovery case, and the test output and coverage summary. I didn't write them.

{"summary": "Added lease recovery for orphaned market-watcher jobs, lease renewal during fan-out, a delayed re-add on live-lease conflict, and a runnable worker entrypoint. New `src/jobs/marketWatcherLeaseSweeper.ts` (Redis SET NX PX lock, recoverExpiredLeases, re-enqueue of due retryable jobs with no live BullMQ job under `<jobKey>:recover:<attempt>`). The repo gained `renewLease` and `listDueRetryable`, and `recoverExpiredLeases` now fails jobs that hit maxAttempts. The handler gets `heartbeat()` and throws `LeaseLostError`, which the coordinator turns into `skipped`. A claim that fails on a live lease now schedules a delayed re-add at lease_until+1s. Added `src/workers/marketWatcher.ts` (worker + sweeper, SIGTERM/SIGINT drain), npm scripts `market-watcher` and `market-watcher:dev`, the `MARKET_WATCHER_SWEEP_INTERVAL_MS` env var, the metrics `market_watcher_jobs_recovered_total` and `market_watcher_lease_renewals_total{outcome}`, and docs in `docs/job-queue.md`.", "test_plan": "Nothing was run locally. CI should run `npm test -- tests/marketWatcherJobService.test.ts tests/marketWatcherWorker.test.ts tests/marketWatcherLeaseSweeper.test.ts`, then `npm run lint`, `npx tsc -p tsconfig.json --noEmit` and `npm test` with coverage. Tests cover: crash then sweeper recovery then success (row 1, distinct `:recover:` id although the original job id is completed, row 2); delayed re-add on live-lease conflict (3); 3-batch fan-out with heartbeat renewal (4); renewal failure stops before the next batch and returns skipped with no succeeded co

## Changed files
- `docs/job-queue.md`
- `package.json`
- `src/config/env-schema.ts`
- `src/metrics/registry.ts`
- `src/services/marketWatcherJobService.ts`
- `tests/marketWatcherJobService.test.ts`
- `tests/marketWatcherWorker.test.ts`
- `src/jobs/marketWatcherLeaseSweeper.ts`
- `src/workers/marketWatcher.ts`
- `tests/marketWatcherLeaseSweeper.test.ts`

## Test plan
Repository CI.

This PR was created as a draft by the issue solver bot. It will remain a
draft until repository CI passes.

Closes #983

@drips-wave

drips-wave Bot commented Oct 18, 2026

Copy link
Copy Markdown

@innocentharuna Great news! 🎉 Based on an automated assessment of this PR, the linked Wave issue(s) no longer count against your application limits.

You can now already apply to more issues while waiting for a review of this PR. Keep up the great work! 🚀

Learn more about application limits

@innocentharuna
innocentharuna marked this pull request as ready for review October 18, 2026 23:01
@greatest0fallt1me
greatest0fallt1me merged commit b7e6c13 into Predictify-org:main Oct 19, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

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

3 participants