Task Management & DLQ

You can programmatically manage tasks, investigate failures, and handle Dead Letter Queues (DLQ) using the exported management API.

These functions allow you to build custom admin UIs or automated recovery workflows.

TIP

Dashboard Available While this page describes the programmatic API, Mongodash also provides a Visual Dashboard (GUI) which wraps these methods in a user-friendly interface. The dashboard allows you to view task lists, filter by status/error, retry failed tasks, and trigger cron jobs without writing any code.

Listing Tasks

Use getReactiveTasks to inspect the queue. You can filter by task name, status, error message, or properties of the source document.

import { getReactiveTasks } from 'mongodash';

// list currently failed tasks
const failedTasks = await getReactiveTasks({ 
    task: 'send-welcome-email', 
    status: 'failed' 
});

// list with pagination
const page1 = await getReactiveTasks(
    { task: 'send-welcome-email' }, 
    { limit: 50, skip: 0, sort: { nextRunAt: -1 } }
);

// Advanced: Helper to find task by the SOURCE document id
// This is powerful: "Find the task associated with Order #123"
const orderTasks = await getReactiveTasks({
    task: 'sync-order',
    sourceDocFilter: { _id: 'order-123' }
});

IMPORTANT

getReactiveTasks and countReactiveTasks only accept a source-id filter (sourceDocFilter: { _id: ... }). A complex source-document filter (e.g. { isVip: true }) throws here — only retryReactiveTasks resolves complex source filters (it scans the source collection in batches). To list/count by a source-document property, first resolve the ids yourself, then pass them as sourceDocFilter: { _id: [...] }.

Counting Tasks

Use countReactiveTasks for metrics or UI badges.

import { countReactiveTasks } from 'mongodash';

const dlqSize = await countReactiveTasks({ 
    task: 'send-welcome-email', 
    status: 'failed' 
});

Retrying Tasks

Use retryReactiveTasks to manually re-trigger tasks. This is useful for DLQ recovery after fixing a bug.

This operation is concurrency-safe. If a task is currently processing, it will be marked to re-run immediately after the current execution finishes (processing_dirty), ensuring no race conditions.

import { retryReactiveTasks } from 'mongodash';

// Retry ALL failed tasks for a specific job
const result = await retryReactiveTasks({ 
    task: 'send-welcome-email', 
    status: 'failed' 
});
console.log(`Retried ${result.modifiedCount} tasks.`);

// Retry specific task by Source Document ID
await retryReactiveTasks({
    task: 'sync-order',
    sourceDocFilter: { _id: 'order-123' }
});

// Bulk Retry: Retry all tasks for "VIP" orders
// This efficiently finds matching tasks and schedules them for execution.
await retryReactiveTasks({
    task: 'sync-order',
    sourceDocFilter: { isVip: true }
});