Redis Daq job queue mach

welcome email nob, uploaded image resize, PDF report chenmoH - HTTP request Daq Qu' Qo'. job queue work jaH 'ej user nom jang. Redis primitives ghaj queue mach (heavier message broker pagh).

inline work Qatlh

naive version slow work request handler Daq Qu':

app.post('/signup', async (req, res) => {
  const user = await createUser(req.body);
  await sendWelcomeEmail(user); // blocks the response on an SMTP round trip
  res.json(user);
});

email provider slow pagh down vaj signup Hoch hang pagh fail (user create Qapla' 'ach). queue "user chenmoH" "email nob" pIm. request nom nob, worker process delivery handle (retries rap).

core primitive: blocking list pop

Redis lists LPUSH (front chel) 'ej BRPOP (back vo' pop, empty vaj wait). pair queue ghaH:

import { createClient } from 'redis';

const redis = createClient();
await redis.connect();

// Producer: enqueue a job
async function enqueue(queueName, job) {
  await redis.lPush(queueName, JSON.stringify(job));
}

// Consumer: block until a job is available, then process it
async function dequeue(queueName) {
  const result = await redis.brPop(queueName, 0); // 0 = block indefinitely
  return JSON.parse(result.element);
}

lo':

await enqueue('emails', { type: 'welcome', userId: user.id });

// worker.js — runs as a separate process
while (true) {
  const job = await dequeue('emails');
  await sendWelcomeEmail(job.userId);
}

Qu' 'ach gap tIn: worker crash qaSDI' BRPOP job remove 'ach processing Qu'pa' - job Qaw'. BRPOP destructive: job picked up record pagh.

reliable: reliable-queue pattern

Redis documentation pattern: job main queue vo' per-worker "processing" list Daq atomic move - job nuqDaq existbe' state pagh.

async function dequeueReliable(queueName, processingKey) {
  const raw = await redis.brPopLPush(queueName, processingKey, 0);
  return raw ? JSON.parse(raw) : null;
}

async function ack(processingKey, rawJob) {
  await redis.lRem(processingKey, 1, rawJob);
}

while (true) {
  const raw = await redis.brPopLPush('emails', 'emails:processing', 0);
  try {
    const job = JSON.parse(raw);
    await sendWelcomeEmail(job.userId);
    await redis.lRem('emails:processing', 1, raw);
  } catch (err) {
    console.error('job failed, leaving in processing list for recovery', err);
  }
}

worker job Dung Hegh vaj job emails:processing Daq pagh (vanish pagh). recovery process (periodic pagh worker startup) list scan, threshold vo' qan jobs main queue Daq push retry vaD.

retry backoff chel

failed job nom return pagh 'ej tight loop fail. attempt count payload Daq track, delayed re-enqueue lo':

async function handleJob(raw, processingKey) {
  const job = JSON.parse(raw);
  try {
    await sendWelcomeEmail(job.userId);
    await redis.lRem(processingKey, 1, raw);
  } catch (err) {
    await redis.lRem(processingKey, 1, raw);
    job.attempts = (job.attempts ?? 0) + 1;

    if (job.attempts >= 5) {
      await redis.lPush('emails:dead-letter', JSON.stringify(job));
      return;
    }

    const delayMs = 1000 * 2 ** job.attempts;
    setTimeout(() => enqueue('emails', job), delayMs);
  }
}

dead-letter queue legh: failed attempts law' vaj automatic retry Qo', job human inspect laH Daq jaH. pagh vaj permanently broken job (bad data, downstream service gone) retry forever, real work logs Daq drown.

scheduled / delayed jobs sorted set

LPUSH/BRPOP "nom run" neH. "future time specific run" (reminder email, subscription renewal check) sorted set Dun, scheduled timestamp score:

async function scheduleAt(queueName, job, runAtMs) {
  await redis.zAdd(`${queueName}:scheduled`, {
    score: runAtMs,
    value: JSON.stringify(job),
  });
}

poller periodic due jobs regular queue Daq move:

async function promoteDueJobs(queueName) {
  const now = Date.now();
  const due = await redis.zRangeByScore(`${queueName}:scheduled`, 0, now);

  for (const raw of due) {
    await redis.multi()
      .zRem(`${queueName}:scheduled`, raw)
      .lPush(queueName, raw)
      .exec();
  }
}

setInterval(() => promoteDueJobs('emails'), 1000);

multi() transaction qatlh: job scheduled set vo' remove 'ej active queue Daq land (process interrupt 'ach).

SoH mach Qo'

queue Dun, 'ach production job systems more poQ: job priorities, concurrency limits, dashboard stuck jobs, exactly-once-ish delivery Redis failover. maintained library lo': BullMQ Node ecosystem standard, patterns Hoch implement Redis Daq. minimal version wa'logh mach Dun - job queue nuq responsible Sov, library defaults failure modes Dun'a' evaluate.