dataqueue-advanced

name	dataqueue-advanced
description	Advanced DataQueue patterns — job dependencies, step memoization, waits, tokens, cron, timeouts, tags, idempotency.

DataQueue Advanced Patterns

Job Dependencies

Use dependsOn on addJob or addJobs so a job stays pending until prerequisites are satisfied (PostgreSQL and Redis). Combining jobIds and tags requires both to be clear (logical AND).

Prerequisites by job id (dependsOn.jobIds)

The job runs only after every listed job has reached completed. DataQueue validates ids, rejects self-dependencies and cycles, and cancels dependents (transitively) if a prerequisite ends failed or cancelled.

• addJob: use only positive existing job ids.
• addJobs: import batchDepRef from @nicnocquee/dataqueue to point at another entry in the same batch — e.g. dependsOn: { jobIds: [batchDepRef(0)] } waits for the job at index 0.

import { batchDepRef } from '@nicnocquee/dataqueue';

const [idA, idB] = await queue.addJobs([
  { jobType: 'ingest', payload: { fileId: '1' } },
  {
    jobType: 'transform',
    payload: { fileId: '1' },
    dependsOn: { jobIds: [batchDepRef(0)] },
  },
]);

Tag drain (dependsOn.tags)

Wait until there is no other active job (pending, processing, or waiting) whose tags are a superset of every tag in dependsOn.tags. Use for “wave” or tenant barriers. If a matching job fails or is cancelled, dependent jobs waiting on those tags are cancelled (transitively).

await queue.addJob({
  jobType: 'finalize_wave',
  payload: { wave: 2 },
  tags: ['wave:2'],
  dependsOn: { tags: ['wave:1'] },
});

Persisted fields on JobRecord: dependsOnJobIds, dependsOnTags.

Step Memoization with ctx.run()

Wrap side-effectful work in ctx.run(stepName, fn). Results are cached in the database — when the handler re-runs after a wait, completed steps replay from cache without re-executing.

const handler = async (payload, signal, ctx) => {
  const data = await ctx.run('fetch-data', async () => {
    return await fetchFromAPI(payload.url);
  });

  await ctx.run('send-notification', async () => {
    await notify(data.userId, data.message);
  });
};

Rules:

• Step names must be unique within a handler.
• Step names must be stable across deployments while jobs are waiting.
• Step order must not change conditionally between re-invocations.

Time-Based Waits

waitFor (duration)

const handler = async (payload, signal, ctx) => {
  await ctx.run('step-1', async () => {
    /* ... */
  });
  await ctx.waitFor({ hours: 24 });
  await ctx.run('step-2', async () => {
    /* ... */
  });
};

Duration fields: seconds, minutes, hours, days, weeks, months, years (additive).

waitUntil (date)

await ctx.waitUntil(new Date('2025-03-01T09:00:00Z'));

How waits work internally

1. Handler throws a WaitSignal internally.
2. Job moves to 'waiting' status — worker lock is released.
3. After the wait expires, job becomes 'pending' again.
4. Handler re-runs from top; ctx.run() replays cached steps.

Waiting jobs are idle — they hold no lock, no concurrency slot, no resources.

Token-Based Waits (Human-in-the-Loop)

Create a token, send it to an external actor, and wait for them to complete it.

const handler = async (payload, signal, ctx) => {
  const token = await ctx.run('create-token', async () => {
    return await ctx.createToken({ timeout: '48h', tags: ['approval'] });
  });

  await ctx.run('notify', async () => {
    await sendSlack(`Approve: ${token.id}`);
  });

  const result = await ctx.waitForToken<{ approved: boolean }>(token.id);
  if (result.ok) {
    await ctx.run('process', async () => {
      if (result.output.approved) await approve(payload.id);
    });
  }
};

Complete tokens externally:

await queue.completeToken(tokenId, { approved: true });

Expire timed-out tokens periodically:

await queue.expireTimedOutTokens();

Cron Scheduling

const cronId = await queue.addCronJob({
  scheduleName: 'daily-report',
  cronExpression: '0 9 * * *',
  jobType: 'generate_report',
  payload: { reportId: 'daily', userId: 'system' },
  timezone: 'America/New_York',
  allowOverlap: false,
});

The processor automatically enqueues due cron jobs before each batch — no manual triggering needed.

Manage schedules:

await queue.pauseCronJob(cronId);
await queue.resumeCronJob(cronId);
await queue.editCronJob(cronId, { cronExpression: '0 */2 * * *' });
await queue.removeCronJob(cronId);
const schedules = await queue.listCronJobs('active');

Timeout Management

Proactive — ctx.prolong()

const handler = async (payload, signal, ctx) => {
  ctx.prolong(60_000); // set deadline to 60s from now
  await doHeavyWork();
  ctx.prolong(); // reset to original timeoutMs
};

Reactive — ctx.onTimeout()

const handler = async (payload, signal, ctx) => {
  let step = 0;
  ctx.onTimeout(() => {
    if (step < 3) return 30_000; // extend 30s
  });
  step = 1;
  await doStep1();
  step = 2;
  await doStep2();
  step = 3;
  await doStep3();
};

Both update locked_at in the DB, preventing premature reclamation.

Force Kill on Timeout

await queue.addJob({
  jobType: 'task',
  payload: {
    /* ... */
  },
  timeoutMs: 5000,
  forceKillOnTimeout: true,
});

Limitations of forceKillOnTimeout:

• Requires Node.js (not Bun).
• Handler must be serializable (no closures over external variables).
• prolong, onTimeout, ctx.run, waits are NOT available.

Event Hooks

Subscribe to real-time job lifecycle events. Works identically with PostgreSQL and Redis.

const queue = initJobQueue<MyPayloadMap>(config);

queue.on('job:completed', ({ jobId, jobType }) => {
  console.log(`Job ${jobId} (${jobType}) completed`);
});

queue.on('job:failed', ({ jobId, jobType, error, willRetry }) => {
  console.error(`Job ${jobId} failed: ${error.message}`);
  if (!willRetry) {
    alertOps(`Permanent failure for job ${jobId}`);
  }
});

queue.on('error', (error) => {
  Sentry.captureException(error);
});

Available events

Event	Payload
job:added	{ jobId, jobType }
job:processing	{ jobId, jobType }
job:completed	{ jobId, jobType }
job:failed	{ jobId, jobType, error, willRetry }
job:cancelled	{ jobId }
job:retried	{ jobId }
job:waiting	{ jobId, jobType }
job:progress	{ jobId, progress }
error	Error

Listener management

const listener = ({ jobId }) => console.log(jobId);
queue.on('job:completed', listener);
queue.off('job:completed', listener);
queue.once('job:added', ({ jobId }) => console.log('First job:', jobId));
queue.removeAllListeners('job:completed');
queue.removeAllListeners(); // all events

The error event fires alongside onError callbacks in ProcessorOptions and SupervisorOptions -- both mechanisms work independently.

Tags

await queue.addJob({
  jobType: 'email',
  payload: {
    /* ... */
  },
  tags: ['welcome', 'onboarding'],
});

const jobs = await queue.getJobsByTags(['welcome'], 'any');
await queue.cancelAllUpcomingJobs({
  tags: { values: ['onboarding'], mode: 'all' },
});

Tag query modes: 'exact', 'all', 'any', 'none'.

Group-Based Concurrency

Use job group.id plus processor groupConcurrency to enforce a global cap per group across all workers/instances (PostgreSQL and Redis).

await queue.addJob({
  jobType: 'email',
  payload: {
    /* ... */
  },
  group: { id: 'tenant_abc', tier: 'gold' }, // tier is optional/reserved
});

const processor = queue.createProcessor(handlers, {
  batchSize: 20,
  concurrency: 10,
  groupConcurrency: 2,
});

Ungrouped jobs are unaffected by groupConcurrency.

Idempotency

const jobId = await queue.addJob({
  jobType: 'email',
  payload: { to: '[e-mail]', subject: 'Welcome', body: '...' },
  idempotencyKey: `welcome-${userId}`,
});

If a job with the same key exists, returns the existing job ID. Key is unique across all statuses until cleanupOldJobs removes it.

Transactional Job Creation (PostgreSQL Only)

Insert a job within an existing database transaction so the job is enqueued atomically with other writes:

import { Pool } from 'pg';

const pool = new Pool({ connectionString: process.env.DATABASE_URL });

async function registerUser(email: string, name: string) {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');

    await client.query('INSERT INTO users (email, name) VALUES ($1, $2)', [
      email,
      name,
    ]);

    const queue = getJobQueue();
    await queue.addJob(
      {
        jobType: 'send_email',
        payload: { to: email, subject: 'Welcome!', body: `Hi ${name}!` },
      },
      { db: client },
    );

    await client.query('COMMIT');
  } catch (error) {
    await client.query('ROLLBACK');
    throw error;
  } finally {
    client.release();
  }
}

The db option accepts any object matching DatabaseClient { query(text, values): Promise<{ rows, rowCount }> } — works with pg.PoolClient, pg.Client, or compatible ORM query runners.

The job event ('added') is also inserted within the same transaction.

Retry Strategy

Configure how failed jobs are retried with retryDelay, retryBackoff, and retryDelayMax.

Fixed delay

await queue.addJob({
  jobType: 'email',
  payload: {
    /* ... */
  },
  maxAttempts: 5,
  retryDelay: 30, // 30 seconds between each retry
  retryBackoff: false,
});

Exponential backoff with cap

await queue.addJob({
  jobType: 'email',
  payload: {
    /* ... */
  },
  maxAttempts: 10,
  retryDelay: 5, // base: 5 seconds
  retryBackoff: true, // default — delay doubles each attempt with jitter
  retryDelayMax: 300, // never wait more than 5 minutes
});

Cron schedules with retry config

await queue.addCronJob({
  scheduleName: 'daily-sync',
  cronExpression: '0 * * * *',
  jobType: 'sync',
  payload: { source: 'api' },
  retryDelay: 60,
  retryBackoff: true,
  retryDelayMax: 600,
});

Every job enqueued by the schedule inherits the retry settings.

Dead-letter routing

Set deadLetterJobType on jobs (or cron schedules) to route exhausted failures:

await queue.addJob({
  jobType: 'email',
  payload: { to: '[e-mail]' },
  maxAttempts: 3,
  deadLetterJobType: 'email_dead_letter',
});

Dead-letter jobs receive envelope payload:

{
  originalJob: { id, jobType, attempts, maxAttempts },
  originalPayload: {...},
  failure: { message, reason, failedAt }
}

Default behavior

When no retry options are set, the legacy formula 2^attempts * 60 seconds is used. This is fully backward compatible.

Maintenance

Use createSupervisor() to automate all maintenance tasks in a long-running server:

const supervisor = queue.createSupervisor({
  intervalMs: 60_000,
  stuckJobsTimeoutMinutes: 10,
  cleanupJobsDaysToKeep: 30,
  cleanupEventsDaysToKeep: 30,
});
supervisor.startInBackground();

For serverless or one-off scripts, call supervisor.start() (runs once) or use the manual methods:

await queue.reclaimStuckJobs(10); // reclaim jobs stuck > 10 min
await queue.cleanupOldJobs(30); // delete completed jobs > 30 days
await queue.cleanupOldJobEvents(30); // delete old events > 30 days
await queue.expireTimedOutTokens(); // expire overdue tokens