Workflow-002: Batch Processing

Fresh

Pattern: Batch trigger with concurrency control Use Case: ETL pipelines, data enrichment, bulk API calls, large dataset processing

Overview

Process large datasets by splitting work into parallel chunks with controlled concurrency. Trigger.dev handles scheduling, retries, and state — you just write the processing logic.

Batch Processing Flow

Pattern 1: Simple Batch (Under 1,000 Items)

// trigger/batch-processor.ts
import { task } from "@trigger.dev/sdk";
import { processItem } from "./process-item";

export const batchProcessor = task({
  id: "batch-processor",
  run: async (payload: { items: Array<{ id: string; data: unknown }> }) => {
    // Trigger all items in parallel, wait for all results
    const results = await processItem.batchTriggerAndWait(
      payload.items.map(item => ({ payload: item }))
    );

    const succeeded = results.runs.filter(r => r.ok).map(r => r.output);
    const failed = results.runs.filter(r => !r.ok);

    console.log(`Processed: ${succeeded.length}/${payload.items.length}`);
    if (failed.length > 0) {
      console.error(`Failed: ${failed.length} items`);
    }

    return { processed: succeeded.length, failed: failed.length };
  },
});

Pattern 2: Large Dataset with Streaming (Memory Efficient)

// trigger/large-batch-processor.ts
import { task } from "@trigger.dev/sdk";
import { processItem } from "./process-item";
import { db } from "./db";

export const largeBatchProcessor = task({
  id: "large-batch-processor",
  run: async (payload: { query: string }) => {
    // Stream items from DB — no need to load all into memory
    async function* generateItems() {
      const cursor = db.query(payload.query).cursor();
      for await (const row of cursor) {
        yield {
          payload: {
            id: row.id,
            data: row.data
          }
        };
      }
    }

    // batchTrigger accepts AsyncGenerator directly
    const batchHandle = await processItem.batchTrigger(generateItems());

    return { batchId: batchHandle.batchId };
  },
});

Pattern 3: Chunked Processing (Very Large Datasets)

// trigger/chunked-processor.ts
import { task } from "@trigger.dev/sdk";
import { processChunk } from "./process-chunk";

const CHUNK_SIZE = 500;

export const chunkedProcessor = task({
  id: "chunked-processor",
  run: async (payload: { totalCount: number }) => {
    const chunks = [];
    for (let offset = 0; offset < payload.totalCount; offset += CHUNK_SIZE) {
      chunks.push({ offset, limit: CHUNK_SIZE });
    }

    // Process chunks in parallel
    const chunkResults = await processChunk.batchTriggerAndWait(
      chunks.map(chunk => ({ payload: chunk }))
    );

    let totalProcessed = 0;
    for (const result of chunkResults.runs) {
      if (result.ok) {
        totalProcessed += result.output.count;
      }
    }

    return { totalProcessed };
  },
});

// Chunk processor handles a slice of data
export const processChunk = task({
  id: "process-chunk",
  run: async (payload: { offset: number; limit: number }) => {
    const rows = await db.query(
      `SELECT * FROM items LIMIT $1 OFFSET $2`,
      [payload.limit, payload.offset]
    );

    // Process each row in this chunk
    const results = await Promise.all(rows.map(row => processRow(row)));

    return { count: results.length };
  },
});

Controlling Concurrency

Global Concurrency Limit on Worker Task

export const processItem = task({
  id: "process-item",
  queue: {
    concurrencyLimit: 10,  // Max 10 running at once globally
  },
  run: async (payload) => {
    // Process one item
  },
});

Per-User Concurrency (Multi-Tenant)

// Limit concurrency per customer
await processItem.batchTrigger(
  items.map(item => ({ payload: item })),
  {
    queue: { name: `customer-${customerId}`, concurrencyLimit: 5 },
    concurrencyKey: customerId,
  }
);

Error Handling in Batches

const results = await processItem.batchTriggerAndWait(items);

// Strategy 1: Fail entire batch on any failure
const failedRuns = results.runs.filter(r => !r.ok);
if (failedRuns.length > 0) {
  throw new Error(`${failedRuns.length} items failed — will retry entire batch`);
}

// Strategy 2: Collect partial results, log failures
const outputs = [];
for (const run of results.runs) {
  if (run.ok) {
    outputs.push(run.output);
  } else {
    console.error(`Run ${run.id} failed:`, run.error);
    // Continue without throwing — partial success is OK
  }
}

// Strategy 3: Retry only failed items
const failedItems = results.runs
  .filter(r => !r.ok)
  .map((r, i) => items[i]);  // Get original payloads

if (failedItems.length > 0) {
  await processItem.batchTrigger(failedItems);
}

Rate Limit Handling

import { BatchTriggerError } from "@trigger.dev/sdk";

async function triggerWithRetry(items, maxRetries = 3) {
  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      return await processItem.batchTrigger(items);
    } catch (error) {
      if (error instanceof BatchTriggerError && error.isRateLimited) {
        const waitMs = error.retryAfterMs ?? 10000;
        console.log(`Rate limited. Waiting ${waitMs}ms`);
        await new Promise(r => setTimeout(r, waitMs));
        continue;
      }
      throw error;
    }
  }
  throw new Error("Max retries exceeded");
}

Practical ETL Example

export const etlTask = task({
  id: "etl-pipeline",
  maxDuration: 3600,  // 1 hour max
  run: async (payload: { sourceTable: string; destTable: string }) => {
    // 1. Extract
    const count = await db.count(payload.sourceTable);
    console.log(`Extracting ${count} records`);

    // 2. Transform + Load in chunks
    const CHUNK = 250;
    const chunks = Math.ceil(count / CHUNK);

    const chunkTasks = Array.from({ length: chunks }, (_, i) => ({
      payload: {
        sourceTable: payload.sourceTable,
        destTable: payload.destTable,
        offset: i * CHUNK,
        limit: CHUNK
      }
    }));

    const results = await transformLoadChunk.batchTriggerAndWait(chunkTasks);

    const totalLoaded = results.runs
      .filter(r => r.ok)
      .reduce((sum, r) => sum + r.output.loaded, 0);

    return {
      extracted: count,
      loaded: totalLoaded,
      chunks: chunks
    };
  },
});

See Also

• Workflow-001: AI Agent Workflow
• SOP-002: Writing Tasks
• Quick Reference: Commands