Async Queue Temporal Expert

Async Queue & Durable Workflow Expert (2026 Unified Edition)

English | Bahasa Indonesia

English

Orchestration & Integration

• Upstream Triggers & APIs: api-design-expert, js-backend-expert, database-orm-expert
• Billing & Payment Sagas: saas-billing, payment-gateway-expert, doku-payment-gateway
• Scheduled & Recurring: cron-scheduler-expert, event-driven-architect, n8n-automation-expert, pdf-document-generation-expert
• Resilience & Observability: error-resilience-expert, data-telemetry-expert, logging-error-tracking-expert
• Lifecycle & Governance: zero-to-prod-orchestrator, session-memory-manager

Purpose & Overview

Unified production-grade guide for background execution pipelines, async job queues, and distributed state machines. Covers the full spectrum from simple Redis-backed task queues to complex multi-service sagas with compensating rollbacks.

3-Tier Execution Model

Core Capabilities

1. Idempotency & Deduplication: Deterministic jobId keys prevent double billing or duplicate emails.
2. Dead Letter Queues (DLQ): Auto-relocate permanently failing jobs for audit and alerting.
3. Exponential Backoff with Jitter: Prevents thundering herds on upstream services.
4. Tenant Priority Queues: VIP/enterprise tiers get lower BullMQ priority numbers (higher throughput).
5. Durable State Machines: Workflows survive restarts, deployments, and network partitions.
6. Saga Compensations: Multi-step transactions with automated reverse-order rollbacks.

Tier 1: BullMQ v5 — Redis Task Queues (TypeScript)

typescriptCopy code
import { Queue, Worker, Job } from 'bullmq';
import Redis from 'ioredis';

const redisConnection = new Redis(process.env.REDIS_URL!, {
  maxRetriesPerRequest: null, // Required by BullMQ
});

export interface NotificationPayload {
  tenantId: string;
  userId: string;
  type: 'email' | 'webhook';
  payload: Record<string, unknown>;
  idempotencyKey: string;
}

// Main Queue
export const notificationQueue = new Queue<NotificationPayload>('notifications', {
  connection: redisConnection,
  defaultJobOptions: {
    attempts: 5,
    backoff: { type: 'exponential', delay: 1500 },
    removeOnComplete: { age: 86400, count: 5000 },
    removeOnFail: false, // Preserved for DLQ audit
  },
});

// Dead Letter Queue
export const notificationDLQ = new Queue('notifications-dlq', {
  connection: redisConnection,
});

// Enqueue with deduplication & priority
export async function enqueueNotification(data: NotificationPayload, isVip = false) {
  return await notificationQueue.add('send_notification', data, {
    jobId: `notif_${data.idempotencyKey}`, // Deterministic dedup key
    priority: isVip ? 1 : 10,
  });
}

// Worker with concurrency & rate limiting
export const notificationWorker = new Worker<NotificationPayload>(
  'notifications',
  async (job: Job<NotificationPayload>) => {
    if (job.data.type === 'email') await deliverEmail(job.data);
  },
  {
    connection: redisConnection,
    concurrency: 20,
    limiter: { max: 100, duration: 1000 },
  }
);

// DLQ forwarding on retry exhaustion
notificationWorker.on('failed', async (job, error) => {
  if (job && job.attemptsMade >= (job.opts.attempts || 5)) {
    await notificationDLQ.add('failed_notification', {
      originalJobId: job.id, failedReason: error.message,
      data: job.data, exhaustedAt: new Date().toISOString(),
    });
  }
});

Tier 2: Trigger.dev v3 — Serverless Durable Tasks

typescriptCopy code
import { task } from '@trigger.dev/sdk/v3';

export const generateReport = task({
  id: 'generate-enterprise-report',
  retry: { maxAttempts: 4, minTimeoutInMs: 2000, factor: 2, randomize: true },
  run: async (payload: { tenantId: string; month: string }, { ctx }) => {
    // Each step is a durable checkpoint — survives crashes
    const data = await ctx.run('fetch-telemetry', async () => {
      return await fetchTelemetryFromWarehouse(payload.tenantId, payload.month);
    });
    const pdfUrl = await ctx.run('render-pdf', async () => {
      return await generateReportPdf(data);
    });
    await ctx.run('dispatch-webhook', async () => {
      return await sendWebhookNotification(payload.tenantId, pdfUrl);
    });
    return { success: true, pdfUrl };
  },
});

Tier 3: Temporal.io — Distributed Saga with Compensations

typescriptCopy code
import { proxyActivities, ApplicationFailure } from '@temporalio/workflow';
import type * as activities from './activities';

const { chargeCustomer, provisionLicense, sendWelcomeEmail, refundCustomer, revokeLicense } =
  proxyActivities<typeof activities>({
    startToCloseTimeout: '1 minute',
    retry: {
      initialInterval: '1s', backoffCoefficient: 2, maximumAttempts: 5,
      nonRetryableErrorTypes: ['InvalidCardError', 'AccountSuspendedError'],
    },
  });

export async function subscriptionSagaWorkflow(input: {
  customerId: string; planId: string; amountCents: number;
}) {
  const compensations: Array<() => Promise<void>> = [];
  try {
    const charge = await chargeCustomer(input.customerId, input.amountCents);
    compensations.unshift(() => refundCustomer(charge.chargeId));

    const license = await provisionLicense(input.customerId, input.planId);
    compensations.unshift(() => revokeLicense(license.licenseId));

    await sendWelcomeEmail(input.customerId, license.licenseKey);
    return { status: 'COMPLETED' };
  } catch (error) {
    for (const compensate of compensations) {
      try { await compensate(); } catch (e) { console.error('Compensation failed:', e); }
    }
    throw ApplicationFailure.create({
      message: `Saga rolled back: ${(error as Error).message}`, nonRetryable: true,
    });
  }
}

Temporal Determinism Rule: Never use Math.random(), Date.now(), or direct DB calls inside workflow files. Run them inside activities.

Implementation Checklist

• Configure maxRetriesPerRequest: null on ioredis for BullMQ v5.
• Use deterministic jobId from business logic (order_${orderId}) for deduplication.
• Forward permanently dead jobs to DLQ via worker.on('failed') listener.
• Implement rate limiting via worker limiter to protect third-party APIs.
• Enforce deterministic code inside Temporal workflows (activities for side-effects).
• Store large payloads in S3/R2; pass only IDs through queues.
• Implement Saga rollback handlers for multi-step distributed payments.

Orchestration & Integration

• Integrates with: js-backend-expert, cron-scheduler-expert, error-resilience-expert, saas-billing, doku-payment-gateway, data-telemetry-expert.

Sovereign Execution, Token Efficiency & Invariants

Bahasa Indonesia

Integrasi Orkestrasi

• Pemicu Hulu & API: api-design-expert, js-backend-expert, database-orm-expert
• Saga Billing & Pembayaran: saas-billing, payment-gateway-expert, doku-payment-gateway
• Terjadwal & Berulang: cron-scheduler-expert, event-driven-architect
• Ketahanan & Observabilitas: error-resilience-expert, data-telemetry-expert, logging-error-tracking-expert
• Siklus Hidup & Tata Kelola: zero-to-prod-orchestrator, session-memory-manager

Tujuan & Gambaran Umum

Panduan terpadu tingkat produksi untuk pipeline eksekusi latar belakang, antrean job asinkron, dan state machine terdistribusi. Mencakup spektrum penuh dari antrean Redis sederhana hingga saga multi-layanan dengan rollback kompensasi.

Model Eksekusi 3-Tier

Kemampuan Utama

1. Idempotensi & Deduplikasi: Kunci jobId deterministik mencegah duplikasi penagihan atau email ganda.
2. Dead Letter Queue (DLQ): Pemindahan otomatis job yang gagal berulang kali untuk proses audit dan peringatan darurat.
3. Backoff Eksponensial + Jitter: Mencegah beban serentak (thundering herd) pada server hilir atau API pihak ketiga.
4. Prioritas Tenant: Pengguna VIP/enterprise dialokasikan prioritas lebih tinggi (priority: 1 di BullMQ).
5. State Machine Tahan-Gagal: Alur kerja tetap berlanjut meskipun server restart, deploy ulang, atau terjadi gangguan jaringan.
6. Kompensasi Saga: Transaksi bertahap dengan pembatalan otomatis secara berurutan terbalik jika salah satu langkah gagal.

Implementasi Pola Kunci (TypeScript)

1. Tier 1: Antrean BullMQ v5 Berbasis Redis & DLQ

typescriptCopy code
import { Queue, Worker, Job } from 'bullmq';
import Redis from 'ioredis';

const koneksiRedis = new Redis(process.env.REDIS_URL!, {
  maxRetriesPerRequest: null, // Wajib untuk BullMQ v5
});

export interface NotifikasiPayload {
  tenantId: string;
  userId: string;
  jenis: 'email' | 'webhook';
  konten: Record<string, unknown>;
  kunciIdempotensi: string;
}

export const antreanNotifikasi = new Queue<NotifikasiPayload>('notifikasi', {
  connection: koneksiRedis,
  defaultJobOptions: {
    attempts: 5,
    backoff: { type: 'exponential', delay: 1500 },
    removeOnComplete: { age: 86400, count: 5000 },
    removeOnFail: false, // Pertahankan untuk audit DLQ
  },
});

export const antreanDLQ = new Queue('notifikasi-dlq', { connection: koneksiRedis });

// Masukkan job dengan deduplikasi deterministik
export async function jadwalkanNotifikasi(data: NotifikasiPayload, isVip = false) {
  return await antreanNotifikasi.add('kirim_notifikasi', data, {
    jobId: `notif_${data.kunciIdempotensi}`,
    priority: isVip ? 1 : 10,
  });
}

// Worker dengan konkurensi dan pembatas laju
export const workerNotifikasi = new Worker<NotifikasiPayload>(
  'notifikasi',
  async (job: Job<NotifikasiPayload>) => {
    if (job.data.jenis === 'email') await kirimEmail(job.data);
  },
  {
    connection: koneksiRedis,
    concurrency: 20,
    limiter: { max: 100, duration: 1000 },
  }
);

// Teruskan ke DLQ setelah batas retry habis
workerNotifikasi.on('failed', async (job, error) => {
  if (job && job.attemptsMade >= (job.opts.attempts || 5)) {
    await antreanDLQ.add('notifikasi_gagal_permanen', {
      originalJobId: job.id,
      alasanGagal: error.message,
      data: job.data,
      waktuGagal: new Date().toISOString(),
    });
  }
});

2. Tier 2: Trigger.dev v3 Serverless Durable Tasks

typescriptCopy code
import { task } from '@trigger.dev/sdk/v3';

export const tugasLaporanBulanan = task({
  id: 'generate-laporan-bulanan',
  retry: { maxAttempts: 4, minTimeoutInMs: 2000, factor: 2, randomize: true },
  run: async (payload: { tenantId: string; bulan: string }, { ctx }) => {
    const data = await ctx.run('ambil-telemetri', async () => {
      return await ambilDataGudang(payload.tenantId, payload.bulan);
    });
    const urlPdf = await ctx.run('render-pdf', async () => {
      return await renderDokumenPdf(data);
    });
    await ctx.run('kirim-webhook', async () => {
      return await kirimWebhookNotifikasi(payload.tenantId, urlPdf);
    });
    return { status: 'sukses', urlPdf };
  },
});

3. Tier 3: Temporal.io Distributed Saga & Kompensasi Rollback

typescriptCopy code
import { proxyActivities, ApplicationFailure } from '@temporalio/workflow';
import type * as activities from './activities';

const { debitRekening, alokasikanLisensi, kirimEmailSelamatDatang, kembalikanDana, batalkanLisensi } =
  proxyActivities<typeof activities>({
    startToCloseTimeout: '1 minute',
    retry: { initialInterval: '1s', backoffCoefficient: 2, maximumAttempts: 5 },
  });

export async function sagaLanggananWorkflow(input: {
  pelangganId: string; paketId: string; totalBayar: number;
}) {
  const daftarKompensasi: Array<() => Promise<void>> = [];
  try {
    const transaksi = await debitRekening(input.pelangganId, input.totalBayar);
    daftarKompensasi.unshift(() => kembalikanDana(transaksi.transaksiId));

    const lisensi = await alokasikanLisensi(input.pelangganId, input.paketId);
    daftarKompensasi.unshift(() => batalkanLisensi(lisensi.lisensiId));

    await kirimEmailSelamatDatang(input.pelangganId, lisensi.kodeLisensi);
    return { status: 'SELESAI' };
  } catch (error) {
    // Eksekusi kompensasi rollback dalam urutan terbalik
    for (const kompensasi of daftarKompensasi) {
      try { await kompensasi(); } catch (e) { console.error('Kompensasi gagal:', e); }
    }
    throw ApplicationFailure.create({
      message: `Saga berhasil di-rollback: ${(error as Error).message}`, nonRetryable: true,
    });
  }
}

Matriks Eksekusi Berdaulat, Efisiensi Token & Invarian