Async Queue Temporal Expert

---
name: async-queue-temporal-expert
description: "Async queues & durable sagas: BullMQ v5, Trigger.dev, and Temporal.io / Antrean async & workflow tahan-gagal: BullMQ, Trigger.dev, Temporal."
author: "Roedy Rustam"
version: "4.8.0"

---

# Async Queue & Durable Workflow Expert (2026 Unified Edition)

[English](#english) | [Bahasa Indonesia](#bahasa-indonesia)

---

<a name="english"></a>
## 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`
- **Cross-Domain Synergy & Subagent Swarm Integration**: `chatbot-messaging-expert`, `durable-agent-execution-expert`

### 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
| Tier | Engine | Best For |
|------|--------|----------|
| **Tier 1: Redis Task Queues** | BullMQ v5 | High-throughput worker jobs, priority queues, rate limiting, DLQ |
| **Tier 2: Serverless Durable Tasks** | Trigger.dev v3 / Inngest | Step-checkpointed tasks, automatic resume across crashes, zero infra |
| **Tier 3: Distributed Sagas** | Temporal.io | Multi-service orchestration, compensating rollbacks, long-running workflows |

### 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)

```typescript
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

```typescript
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

```typescript
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.

---

### Tier 4: BullMQ `FlowProducer` Parent-Child Fan-Out/Fan-In DAG (TypeScript)
Coordinate parallel child chunk workers (e.g., document OCR pages or multi-tenant invoice batches) and trigger the parent aggregation job only after all children succeed:

```typescript
// src/queues/invoice-batch-flow.ts
import { FlowProducer } from 'bullmq';
import Redis from 'ioredis';

const flowRedis = new Redis(process.env.REDIS_URL!, { maxRetriesPerRequest: null });
export const invoiceFlowProducer = new FlowProducer({ connection: flowRedis });

export async function dispatchMonthlyInvoiceFlow(
  batchId: string,
  tenantIds: readonly string[]
): Promise<{ parentJobId: string | undefined; childCount: number }> {
  const flowTree = await invoiceFlowProducer.add({
    name: 'finalize-invoice-batch',
    queueName: 'invoice-batch-aggregator',
    data: { batchId, totalTenants: tenantIds.length },
    opts: {
      jobId: `batch_parent_${batchId}`,
      failParentOnFailure: true,
    },
    children: tenantIds.map((tenantId) => ({
      name: 'generate-tenant-invoice',
      queueName: 'invoice-tenant-worker',
      data: { batchId, tenantId },
      opts: {
        jobId: `invoice_${batchId}_${tenantId}`,
        attempts: 4,
        backoff: { type: 'exponential', delay: 2_000 },
      },
    })),
  });

  return {
    parentJobId: flowTree.job.id,
    childCount: tenantIds.length,
  };
}
```

### Tier 5: Long-Running Temporal Activity Heartbeat & Cooperative Cancellation (TypeScript)
Long-running activities (video transcoding, large dataset exports) must emit periodic heartbeats with checkpoint progress so Temporal detects stalled workers quickly and resumes from the last saved cursor:

```typescript
// src/temporal/activities/export-dataset.ts
import { Context } from '@temporalio/activity';

export interface ExportCheckpoint {
  readonly lastProcessedCursor: string | null;
  readonly rowsExported: number;
}

export async function exportTenantDatasetActivity(
  tenantId: string,
  fetchPage: (cursor: string | null) => Promise<{ rows: unknown[]; nextCursor: string | null }>
): Promise<{ totalRows: number }> {
  const ctx = Context.current();
  const priorCheckpoint = ctx.info.heartbeatDetails as ExportCheckpoint | undefined;

  let cursor: string | null = priorCheckpoint?.lastProcessedCursor ?? null;
  let rowsExported = priorCheckpoint?.rowsExported ?? 0;

  while (true) {
    if (ctx.cancellationSignal.aborted) {
      throw new Error(`ACTIVITY_CANCELLED: Export for tenant ${tenantId} aborted cleanly at row ${rowsExported}.`);
    }

    const page = await fetchPage(cursor);
    rowsExported += page.rows.length;
    cursor = page.nextCursor;

    ctx.heartbeat({
      lastProcessedCursor: cursor,
      rowsExported,
    } satisfies ExportCheckpoint);

    if (!cursor) break;
  }

  return { totalRows: rowsExported };
}
```

### Hard Guardrails (Declarative Invariants)
- **Mandatory `maxRetriesPerRequest: null` on BullMQ Redis Connections**: BullMQ blocking commands (`BRPOPLPUSH` / `BZPOPMIN`) crash if `ioredis` uses the default `maxRetriesPerRequest: 20`.
- **Strict Temporal Workflow Determinism**: Files imported by `@temporalio/workflow` must never call `Date.now()`, `Math.random()`, `process.env`, `fetch()`, or filesystem/DB I/O directly; isolate all non-deterministic side effects inside Activities.
- **Deterministic Business `jobId` Deduplication**: Always pass a deterministic `jobId` derived from the domain idempotency key (`jobId: \`notif_${idempotencyKey}\``) when enqueueing jobs.
- **Pointer-Only Queue Payloads**: Never push multi-megabyte buffers or raw PDFs into Redis/Temporal job payloads; upload blobs to S3/R2 and pass only the object key URI.

---

### 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
| Dimension | Sovereign Mandate & Deterministic Contract |
| :--- | :--- |
| **Frontier Model Cascade** | Medium / High (`Gemini 4 Argon`, `GPT-5.6 Sol Medium`, `Claude Sonnet 5.5 High`) |
| **Availability Fallback** | `Gemini 4 Argon/Pro` -> `Gemini 3.8 Flash`; `Claude Opus 5.5` -> `Claude Opus 4.6` (Zero unhandled missing-model errors) |
| **KV-Cache & Token Protocol** | Pin static instructions & schemas at prompt prefix (`kv-cache-prefix-optimizer`); pass surgical `CONTEXT_MAP.md` slices (<250 tokens/call) during Swarm Fan-Out |
| **SaaT CLI & MCP Dispatch** | CLI: `vibes skill exec async-queue-temporal-expert <audit\|eval\|verify\|generate\|blueprint>` / MCP: `vibes_skill_dispatch`, `vibes_eval_sql`, `vibes_verify_invariants`, `vibes_verify_gate` |
| **NFR & Anti-Slop Invariants** | P99 Query Latency < 15ms, 100% Row Level Security (RLS) coverage on multi-tenant tables, explicit B-Tree indexes on all Foreign Keys; 100% complete zero-placeholder output verified via `node scripts/check-anti-slop.js --strict` |

---


---

<a name="bahasa-indonesia"></a>
## 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`
- **Sinergi Lintas Domain & Integrasi Swarm Subagent**: `n8n-automation-expert`, `chatbot-messaging-expert`, `durable-agent-execution-expert`

### 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
| Tier | Engine | Cocok Untuk |
|------|--------|-------------|
| **Tier 1: Antrean Redis** | BullMQ v5 | Job worker throughput tinggi, antrean prioritas, rate limiting, DLQ |
| **Tier 2: Task Serverless** | Trigger.dev v3 / Inngest | Task dengan checkpoint, resume otomatis, tanpa perlu mengelola infra server |
| **Tier 3: Saga Terdistribusi** | Temporal.io | Orkestrasi multi-layanan, rollback kompensasi, workflow berdurasi panjang |

### 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
```typescript
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
```typescript
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
```typescript
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
| Dimensi | Mandat Berdaulat & Kontrak Deterministik |
| :--- | :--- |
| **Kaskade Model Frontier** | Medium / High (`Gemini 4 Argon`, `GPT-5.6 Sol Medium`, `Claude Sonnet 5.5 High`) |
| **Invarian Fallback Ketersediaan** | `Gemini 4 Argon/Pro` -> `Gemini 3.8 Flash`; `Claude Opus 5.5` -> `Claude Opus 4.6` (Bebas gagal saat model belum tersedia/kuota penuh) |
| **Protokol Hemat Token & KV-Cache** | Kunci instruksi & skema statis di prefix awal (`kv-cache-prefix-optimizer`); kirim potongan `CONTEXT_MAP.md` (<250 token/panggilan) saat Fan-Out |
| **Eksekusi SaaT CLI & MCP** | CLI: `vibes skill exec async-queue-temporal-expert <audit\|eval\|verify\|generate\|blueprint>` / MCP: `vibes_skill_dispatch`, `vibes_eval_sql`, `vibes_verify_invariants`, `vibes_verify_gate` |
| **Invarian NFR & Anti-Slop** | Latensi Query P99 < 15ms, cakupan Row Level Security (RLS) 100% pada tabel multi-tenant, indeks B-Tree eksplisit pada seluruh Foreign Key; kode 100% tuntas tanpa placeholder diverifikasi melalui `node scripts/check-anti-slop.js --strict` |