Fix-e2e-tests (#35)

* Add dead-letter job handling and related tests

- Introduced the `deadLetterJobType` option in the `JobOptions` interface to specify a destination for jobs that exhaust their retry attempts.
- Updated the `JobRecord` interface to include fields for tracking dead-letter job metadata, such as `deadLetteredAt` and `deadLetterJobId`.
- Enhanced documentation to explain the dead-letter routing feature, including examples of configuration and payload structure.
- Implemented logic in both PostgreSQL and Redis backends to create dead-letter jobs when retries are exhausted.
- Added tests to verify the correct behavior of dead-letter routing and job reprocessing.

These changes improve error handling and job management in DataQueue, providing users with a robust mechanism for dealing with failed jobs.

* Enhance testing and coverage reporting in CI workflow

- Updated the CI workflow to include coverage reporting for unit and E2E tests, utilizing Codecov for coverage uploads.
- Added new scripts for running tests with coverage in both the dataqueue and E2E applications.
- Configured Vite to support coverage reporting with the V8 provider, generating reports in multiple formats.
- Updated `.gitignore` to exclude coverage directories from version control.

These changes improve test visibility and ensure comprehensive coverage reporting across the project.

* Update Codecov action to v5 and add token for coverage uploads in CI workflow

- Upgraded the Codecov GitHub Action from v4 to v5 for improved functionality.
- Added a token for secure coverage uploads for both unit and E2E tests, enhancing security and reliability in the CI process.

These changes ensure better integration with Codecov and improve coverage reporting in the CI workflow.

* update readme

---------

Co-authored-by: Nico Prananta <311343+nicnocquee@users.noreply.github.com>

Author: nicnocquee

.github/workflows/ci.yml

@@ -65,11 +65,20 @@ jobs:
         working-directory: packages/dataqueue
         run: pnpm run lint
 
-      - name: Run test
+      - name: Run test coverage
         working-directory: packages/dataqueue
-        run: pnpm run test
+        run: pnpm run test:coverage
         timeout-minutes: 3
 
+      - name: Upload unit coverage to Codecov
+        uses: codecov/codecov-action@v5
+        with:
+          files: packages/dataqueue/coverage/lcov.info
+          token: ${{ secrets.CODECOV_TOKEN }}
+          flags: unit
+          name: unit-coverage
+          fail_ci_if_error: false
+
   e2e:
     runs-on: ubuntu-latest
     services:
@@ -122,13 +131,22 @@ jobs:
         working-directory: apps/e2e
         run: pnpm run build
 
-      - name: Run E2E tests
+      - name: Run E2E coverage
         working-directory: apps/e2e
-        run: pnpm run test:e2e
+        run: pnpm run test:e2e:coverage
         timeout-minutes: 5
         env:
           PG_DATAQUEUE_DATABASE: postgres://postgres:postgres@localhost:5432/e2e_test
 
+      - name: Upload e2e coverage to Codecov
+        uses: codecov/codecov-action@v5
+        with:
+          files: apps/e2e/coverage/lcov.info
+          token: ${{ secrets.CODECOV_TOKEN }}
+          flags: e2e
+          name: e2e-coverage
+          fail_ci_if_error: false
+
       - name: Upload Playwright report
         uses: actions/upload-artifact@v4
         if: ${{ !cancelled() }}

.gitignore

@@ -3,6 +3,8 @@ dist
 data
 .turbo
 redis-data
+coverage
+**/coverage
 
 # Generated from MDX docs by prebuild
 packages/dataqueue/ai/docs-content.json

README.md

@@ -2,6 +2,8 @@
 
 ![DataQueue](./apps/website/app/opengraph-image.png)
 
+[![codecov](https://codecov.io/gh/nicnocquee/dataqueue/graph/badge.svg?token=RXECBHHO1Y)](https://codecov.io/gh/nicnocquee/dataqueue)
+
 A lightweight, Redis or PostgreSQL job queue for Node.js/TypeScript projects. Schedule, process, and manage background jobs with ease. Perfect for web apps (Next.js, etc.) deployed to serverless platforms like Vercel, AWS Lambda, etc.
 
 ## Installation

apps/docs/public/llms-full.txt

@@ -199,6 +199,7 @@ The `JobOptions` interface defines the options for creating a new job in the que
 
 - `tags?`: _string[]_ — Tags for this job. Used for grouping, searching, or batch operations.
 - `idempotencyKey?`: _string_ — Optional idempotency key. When provided, ensures that only one job exists for a given key. If a job with the same key already exists, `addJob` returns the existing job's ID instead of creating a duplicate. See [Idempotency](/usage/add-job#idempotency) for details.
+- `deadLetterJobType?`: _string_ — Optional dead-letter destination job type. When the job exhausts retries, DataQueue creates a new pending job in this job type with an envelope payload containing source metadata, original payload, and failure context.
 
 ## Example
 
@@ -213,6 +214,7 @@ const job = {
   forceKillOnTimeout: false, // Use graceful shutdown (default)
   tags: ['welcome', 'user'], // tags for grouping/searching
   idempotencyKey: 'welcome-email-user-123', // prevent duplicate jobs
+  deadLetterJobType: 'email_dead_letter', // route exhausted failures
 };
 ```
 
@@ -335,13 +337,15 @@ interface JobOptions {
   retryDelay?: number; // Base delay between retries in seconds (default: 60)
   retryBackoff?: boolean; // Use exponential backoff (default: true)
   retryDelayMax?: number; // Max delay cap in seconds (default: none)
+  deadLetterJobType?: string; // Route exhausted failures to this job type
   group?: { id: string; tier?: string }; // Optional group for global concurrency limits
 }
 ```
 
 - `retryDelay` - Base delay between retries in seconds. When `retryBackoff` is true, this is the base for exponential backoff (`retryDelay * 2^attempts`). When false, retries use this fixed delay. Default: `60`.
 - `retryBackoff` - Whether to use exponential backoff. When true, delay doubles with each attempt and includes jitter. Default: `true`.
 - `retryDelayMax` - Maximum delay cap in seconds. Only meaningful when `retryBackoff` is true. No limit when omitted.
+- `deadLetterJobType` - Optional dead-letter destination. When retries are exhausted, a new pending job is created in this job type with an envelope payload (`originalJob`, `originalPayload`, `failure`).
 - `group` - Optional grouping metadata. Use `group.id` to enforce global per-group limits with `ProcessorOptions.groupConcurrency`. `group.tier` is reserved for future policies.
 
 #### AddJobOptions
@@ -497,10 +501,11 @@ interface EditJobOptions {
   retryDelay?: number | null;
   retryBackoff?: boolean | null;
   retryDelayMax?: number | null;
+  deadLetterJobType?: string | null;
 }
 ```
 
-All fields are optional - only provided fields will be updated. Note that `jobType` cannot be changed. Set retry fields to `null` to revert to legacy default behavior.
+All fields are optional - only provided fields will be updated. Note that `jobType` cannot be changed. Set retry fields to `null` to revert to legacy default behavior. Set `deadLetterJobType` to `null` to clear dead-letter routing for pending jobs.
 
 #### Example
 
@@ -858,7 +863,7 @@ The `JobRecord` interface represents a job stored in the queue, including its st
 - `jobType`: _string_ — The type of the job.
 - `payload`: _any_ — The job payload.
 - `status`:
-  _'pending' | 'processing' | 'completed' | 'failed' | 'cancelled'_ —
+  _'pending' | 'processing' | 'completed' | 'failed' | 'cancelled' | 'waiting'_ —
   Current job status.
 - `createdAt`: _Date_ — When the job was created.
 - `updated_at`: _Date_ — When the job was last updated.
@@ -890,6 +895,9 @@ The `JobRecord` interface represents a job stored in the queue, including its st
 - `idempotencyKey?`: _string | null_ — The idempotency key for this job, if one was provided when the job was created.
 - `progress?`: _number | null_ — Progress percentage (0–100) reported by the handler via `ctx.setProgress()`. `null` if no progress has been reported. See [Progress Tracking](/usage/progress-tracking).
 - `output?`: _unknown_ — Handler output stored via `ctx.setOutput(data)` or by returning a value from the handler. `null` if no output has been stored. See [Job Output](/usage/job-output).
+- `deadLetterJobType?`: _string | null_ — Configured dead-letter destination job type for this job.
+- `deadLetteredAt?`: _Date | null_ — Timestamp when this job was routed to a dead-letter job.
+- `deadLetterJobId?`: _number | null_ — Linked dead-letter job ID created when retries were exhausted.
 
 ## Example
 
@@ -1994,8 +2002,9 @@ Type handlers as `JobHandlers<PayloadMap>` — TypeScript enforces a handler for
 1. Creating initJobQueue per request (creates a DB pool each time)
 2. Missing handler for a job type (fails with NoHandler)
 3. Not checking signal.aborted in long handlers
-4. Forgetting reclaimStuckJobs() — crashed workers leave jobs stuck
-5. Skipping migrations (PostgreSQL requires `dataqueue-cli migrate`)
+4. Forgetting dead-letter routing for critical jobs — set `deadLetterJobType` so exhausted failures are inspectable/replayable
+5. Forgetting reclaimStuckJobs() — crashed workers leave jobs stuck
+6. Skipping migrations (PostgreSQL requires `dataqueue-cli migrate`)
 ```
 
 ---
@@ -3059,6 +3068,37 @@ A job handler can fail for many reasons, such as a bug in the code or running ou
 
 When a job fails, it is marked as `failed` and retried up to `maxAttempts` times (default: 3). You can view the error history for a job in its `errorHistory` field.
 
+## Dead-letter queues
+
+You can route permanently failed jobs to a dead-letter job type using `deadLetterJobType`.
+
+When a job exhausts retries (`attempts >= maxAttempts`), DataQueue:
+
+1. Keeps the source job as `failed`.
+2. Creates a new pending dead-letter job in `deadLetterJobType`.
+3. Stores linkage metadata on the source job (`deadLetteredAt`, `deadLetterJobId`).
+
+```ts
+await jobQueue.addJob({
+  jobType: 'email',
+  payload: { to: 'user@example.com' },
+  maxAttempts: 3,
+  deadLetterJobType: 'email_dead_letter',
+});
+```
+
+The dead-letter job payload is an envelope:
+
+```ts
+{
+  originalJob: { id, jobType, attempts, maxAttempts },
+  originalPayload: { ... }, // original job payload
+  failure: { message, reason, failedAt },
+}
+```
+
+If `deadLetterJobType` is not set, behavior is unchanged: exhausted jobs remain failed without creating a dead-letter job.
+
 ## Retry configuration
 
 You can control the retry behavior per-job using three options:

apps/e2e/e2e/maintenance.spec.ts

@@ -1,9 +1,10 @@
 import { test, expect } from '@playwright/test';
 import {
   addJob,
-  processJobs,
   cleanupOldJobs,
+  processJobs,
   reclaimStuckJobs,
+  waitForJobStatus,
 } from './helpers';
 
 test.describe('Maintenance', () => {
@@ -21,9 +22,42 @@ test.describe('Maintenance', () => {
     expect(deleted).toBeGreaterThanOrEqual(0);
   });
 
-  test('reclaim stuck jobs (does not error)', async ({ request }) => {
-    // Just verify the endpoint works without errors
+  test('reclaim stuck jobs transitions a processing job back to pending', async ({
+    request,
+  }) => {
+    const { id } = await addJob(request, {
+      jobType: 'slow-job',
+      payload: { value: 'reclaim-test', delayMs: 7000 },
+    });
+
+    const processingRun = processJobs(request, {
+      batchSize: 1,
+      concurrency: 1,
+      jobType: 'slow-job',
+    });
+
+    await waitForJobStatus(request, id, 'processing', 5000, 100);
+
     const { reclaimed } = await reclaimStuckJobs(request, 0);
-    expect(reclaimed).toBeGreaterThanOrEqual(0);
+    expect(reclaimed).toBeGreaterThanOrEqual(1);
+
+    const reclaimedJob = await waitForJobStatus(
+      request,
+      id,
+      'pending',
+      3000,
+      100,
+    );
+    expect(reclaimedJob.lockedAt).toBeNull();
+    expect(reclaimedJob.lockedBy).toBeNull();
+
+    await processingRun;
+    await processJobs(request, {
+      batchSize: 1,
+      concurrency: 1,
+      jobType: 'slow-job',
+    });
+
+    await waitForJobStatus(request, id, 'completed', 10000, 100);
   });
 });

apps/e2e/package.json

@@ -7,6 +7,7 @@
     "build": "next build",
     "start": "next start --port 3099",
     "test:e2e": "PG_DATAQUEUE_DATABASE=${PG_DATAQUEUE_DATABASE:-postgres://postgres:postgres@localhost:5432/e2e_test} dataqueue-cli migrate && playwright test",
+    "test:e2e:coverage": "rm -rf coverage && mkdir -p coverage/e2e-v8 && NODE_V8_COVERAGE=coverage/e2e-v8 PG_DATAQUEUE_DATABASE=${PG_DATAQUEUE_DATABASE:-postgres://postgres:postgres@localhost:5432/e2e_test} dataqueue-cli migrate && NODE_V8_COVERAGE=coverage/e2e-v8 playwright test && c8 report --temp-directory coverage/e2e-v8 --reporter=text --reporter=html --reporter=lcov --report-dir coverage",
     "test:e2e:ui": "playwright test --ui",
     "migrate-dataqueue": "dataqueue-cli migrate"
   },
@@ -24,6 +25,7 @@
     "@types/node": "^20",
     "@types/react": "^19",
     "@types/react-dom": "^19",
+    "c8": "^10.1.3",
     "typescript": "^5"
   }
 }

package.json

@@ -15,7 +15,9 @@
     "build": "turbo run build",
     "lint": "turbo run lint",
     "test": "turbo run test",
+    "test:coverage": "turbo run test:coverage --filter \"@nicnocquee/dataqueue\"",
     "test:e2e": "turbo run test:e2e --filter e2e",
+    "test:e2e:coverage": "turbo run test:e2e:coverage --filter e2e",
     "format": "prettier --write .",
     "check-format": "prettier --check ."
   },

packages/dataqueue/package.json

@@ -23,6 +23,7 @@
     "ci": "npm run build && npm run check-format && npm run check-exports && npm run lint && npm run test",
     "lint": "tsc",
     "test": "vitest run --reporter=verbose",
+    "test:coverage": "vitest run --reporter=verbose --coverage",
     "format": "prettier --write .",
     "check-format": "prettier --check .",
     "check-exports": "attw --pack .",
@@ -56,6 +57,7 @@
     "@arethetypeswrong/cli": "^0.18.2",
     "@types/node": "^24.0.4",
     "@types/pg": "^8.15.4",
+    "@vitest/coverage-v8": "^3.2.4",
     "ioredis": "^5.9.3",
     "node-pg-migrate": "^8.0.3",
     "pnpm": "^9.0.0",

packages/dataqueue/src/backends/redis.test.ts

@@ -544,6 +544,71 @@ describe('Redis backend integration', () => {
     expect(job?.status).toBe('pending');
   });
 
+  it('reclaims an in-flight job via supervisor and allows reprocessing', async () => {
+    let firstAttempt = true;
+    const handler = vi.fn(async () => {
+      if (firstAttempt) {
+        firstAttempt = false;
+        await new Promise<void>(() => {});
+      }
+    });
+
+    const jobId = await jobQueue.addJob({
+      jobType: 'test',
+      payload: { foo: 'redis-reclaim-live-loop' },
+    });
+
+    const firstProcessor = jobQueue.createProcessor(
+      {
+        email: vi.fn(async () => {}),
+        sms: vi.fn(async () => {}),
+        test: handler,
+      },
+      { pollInterval: 25, batchSize: 1, concurrency: 1 },
+    );
+    firstProcessor.startInBackground();
+
+    let processingJob = await jobQueue.getJob(jobId);
+    for (let i = 0; i < 50; i++) {
+      if (processingJob?.status === 'processing') {
+        break;
+      }
+      await new Promise((resolve) => setTimeout(resolve, 20));
+      processingJob = await jobQueue.getJob(jobId);
+    }
+    expect(processingJob?.status).toBe('processing');
+
+    await firstProcessor.stopAndDrain(25);
+
+    const supervisor = jobQueue.createSupervisor({
+      stuckJobsTimeoutMinutes: 0,
+      cleanupJobsDaysToKeep: 0,
+      cleanupEventsDaysToKeep: 0,
+      expireTimedOutTokens: false,
+    });
+    const maintenance = await supervisor.start();
+    expect(maintenance.reclaimedJobs).toBe(1);
+
+    const reclaimedJob = await jobQueue.getJob(jobId);
+    expect(reclaimedJob?.status).toBe('pending');
+    expect(reclaimedJob?.lockedAt).toBeNull();
+    expect(reclaimedJob?.lockedBy).toBeNull();
+
+    const secondProcessor = jobQueue.createProcessor(
+      {
+        email: vi.fn(async () => {}),
+        sms: vi.fn(async () => {}),
+        test: handler,
+      },
+      { batchSize: 1, concurrency: 1 },
+    );
+    await secondProcessor.start();
+
+    const completedJob = await jobQueue.getJob(jobId);
+    expect(completedJob?.status).toBe('completed');
+    expect(handler).toHaveBeenCalledTimes(2);
+  });
+
   it('getPool should throw for Redis backend', () => {
     expect(() => jobQueue.getPool()).toThrow(
       'getPool() is only available with the PostgreSQL backend',