Nodejs Async Patterns

Node.js Async Patterns

When to Use

Use this skill when the user asks about:

• async/await syntax, patterns, and anti-patterns in Node.js -- including sequential vs. parallel execution, top-level await, and async IIFE patterns
• Promise composition -- Promise.all, Promise.allSettled, Promise.race, Promise.any, chaining, and custom combinators
• The Node.js event loop model -- phases (timers, pending callbacks, idle/prepare, poll, check, close), microtask queues, process.nextTick vs queueMicrotask, and tick starvation
• Node.js Streams -- Readable, Writable, Duplex, Transform streams; backpressure; pipeline; stream composition; async iteration over streams
• Worker threads -- worker_threads module, SharedArrayBuffer, Atomics, MessageChannel, thread pool sizing
• AbortController and AbortSignal -- cooperative cancellation, signal propagation, integration with fetch, fs, and third-party async APIs
• Async iterators and generators -- Symbol.asyncIterator, for await...of, async generator functions, composing async iterables
• Concurrency coordination -- queues, semaphores, rate limiting, batching, fan-out/fan-in patterns
• AsyncLocalStorage and async context propagation

Do NOT use this skill when the user asks about:

• General JavaScript language features not specific to Node.js async -- use javascript-idioms instead (covers closures, prototypes, destructuring, optional chaining)
• Handling async errors, unhandledRejection, domain modules, or structured error propagation patterns -- use nodejs-error-handling instead
• CPU profiling, flame graphs, memory leak detection, garbage collection tuning, or benchmarking async code -- use nodejs-performance instead
• Frontend browser async patterns (Web Workers, Service Workers, browser fetch) -- this skill covers Node.js server-side only
• Database query patterns, ORM async usage, or connection pool management -- use the relevant database skill
• HTTP server frameworks (Express, Fastify, Koa) async middleware specifics -- use the relevant framework skill

Process

1. Classify the Async Problem

Before writing any code, determine which category the user's problem falls into. Misclassifying leads to choosing the wrong primitive entirely.

• I/O-bound, sequential -- one async operation depends on the result of a previous one. The correct tool is async/await with await in sequence. Common mistake: wrapping multiple independent operations in sequential await when they could be parallel.
• I/O-bound, parallel -- multiple async operations are independent and can proceed simultaneously. Use Promise.all for fail-fast behavior or Promise.allSettled when partial failure is acceptable.
• CPU-bound -- any synchronous computation exceeding ~1ms that would block the event loop. Use worker_threads. Do NOT use child_process.fork for CPU tasks unless you need process isolation or legacy CommonJS module compatibility.
• Streaming data -- data arrives or must be produced incrementally (large file reads, network responses, ETL pipelines). Use Node.js Streams or async generators, NOT buffering everything into memory.
• Cancellable operations -- the user needs to abort an in-flight async operation (HTTP request timeout, user-initiated cancel, race between abort and completion). Use AbortController/AbortSignal.
• Context propagation -- you need request-scoped data (request ID, user identity, trace context) to flow through async call chains without threading it through every function argument. Use AsyncLocalStorage.
• Rate-limited or concurrency-capped -- you have N items to process but can only run K at a time. Use a semaphore pattern or a library like p-limit (concurrency cap) or p-throttle (time-rate cap).

2. Apply the Event Loop Mental Model

Every async pattern decision in Node.js requires understanding the event loop's phase ordering. Get this wrong and you introduce subtle timing bugs.

• The event loop processes phases in this order: timers (setTimeout/setInterval callbacks) → pending I/O callbacks (deferred I/O errors) → idle/prepare (internal) → poll (new I/O events) → check (setImmediate callbacks) → close callbacks (socket close events).
• Microtask queues run after every phase completes -- Promise resolution callbacks and queueMicrotask callbacks run here. process.nextTick callbacks run in a separate queue that drains before microtasks.
• Priority ordering (highest to lowest): process.nextTick → Promise microtasks → setImmediate → setTimeout(fn, 0).
• Use process.nextTick sparingly -- it can starve the event loop if called recursively. Prefer queueMicrotask for Promise-adjacent scheduling.
• Use setImmediate to yield control back to the event loop to allow I/O callbacks to fire before your next chunk of synchronous work.
• CPU-bound synchronous work blocking longer than 10ms will cause measurable latency on concurrent requests. The threshold for "acceptable" synchronous blocking is typically ≤1ms on a heavily loaded server.

3. Design Promise Composition

Choose the right combinator based on the failure semantics needed.

• Promise.all(promises) -- runs all concurrently, resolves when all resolve, rejects immediately when any reject (fail-fast). Use for parallel independent operations where you need all results and any failure is fatal.
• Promise.allSettled(promises) -- runs all concurrently, always resolves with an array of {status, value} or {status, reason} objects. Use when partial success is acceptable and you need to inspect every outcome.
• Promise.race(promises) -- resolves or rejects as soon as the first promise settles. Classic use: implementing timeouts by racing a real operation against a setTimeout-based rejection.
• Promise.any(promises) -- resolves as soon as any promise resolves (ignores rejections), rejects with AggregateError only if all reject. Use for "first successful response wins" patterns like hedged requests.
• Sequential with reduce -- when you need to chain an array of async operations sequentially: array.reduce(async (prev, item) => { await prev; return processItem(item); }, Promise.resolve()). Use sparingly -- it defeats the benefit of concurrency.
• Concurrency-capped fan-out -- when processing N items with max K concurrent operations, chunk with p-limit or implement a semaphore using a Promise counter pattern. Rule of thumb: cap concurrent outbound HTTP requests at 10--50 depending on the target server's limits; cap database queries at your connection pool size (typically 5--20).

4. Implement async/await Correctly

async/await is syntactic sugar over Promises, but the sugar introduces its own failure modes.

• Avoid the sequential trap -- await a(); await b(); when a and b are independent takes time(a) + time(b). Use const [ra, rb] = await Promise.all([a(), b()]); to take max(time(a), time(b)).
• Avoid async functions inside forEach -- array.forEach(async (item) => { ... }) does not await the callbacks; the loop exits before any async work completes. Use for...of with await, or Promise.all(array.map(async (item) => ...)) for parallel execution.
• Avoid new Promise wrapping an async function -- the Promise constructor executor is synchronous and wrapping async code inside it hides the inner async errors. Use async functions directly.
• Return vs. return await -- inside a try/catch, write return await promise not return promise. Without await, the function returns before the promise settles, so any rejection escapes the try/catch block. Outside a try/catch, bare return promise is fine and saves a microtask tick.
• Top-level await -- available in ES modules (.mjs or "type": "module" in package.json). Blocks module initialization, which is the intended behavior for connecting to databases before serving requests. Do NOT use it inside shared utility modules that many files import -- it serializes module loading.
• Async IIFE pattern -- for CommonJS files that can't use top-level await: (async () => { ... })().catch(console.error). Always attach a .catch handler to prevent unhandled rejection.

5. Implement Streams and Async Iteration

Streams are the correct answer whenever data size is unbounded or latency-to-first-byte matters more than total throughput.

• Use stream.pipeline (or its stream/promises variant) instead of .pipe() for production code. pipeline automatically handles error cleanup and stream destruction. .pipe() does not propagate errors and leaves streams open on failure.
• Backpressure -- Writable streams signal backpressure via the return value of .write() returning false. When false, stop writing and wait for the drain event. Failing to respect backpressure causes unbounded memory growth. pipeline handles this automatically.
• Transform stream pattern -- subclass Transform and implement _transform(chunk, encoding, callback). Call callback(null, transformedData) to push data downstream. Call callback(error) to signal an error. Keep transforms stateless where possible; accumulate state only when absolutely necessary (e.g., decompression, protocol framing).
• Async iteration over streams -- for await (const chunk of readable) works on any Readable stream in Node.js 12+. This is the idiomatic way to consume streams in modern code. Remember to destroy the stream on early exit: use break inside a try/finally block with stream.destroy() in the finally clause.
• Object mode streams -- setting objectMode: true allows streams to carry arbitrary JavaScript objects, not just Buffer and strings. Use for in-process data pipelines. Do NOT use object-mode streams for network I/O -- serialization must happen at the boundary.
• Readable.from -- Readable.from(asyncIterable) converts any async iterable (including async generator functions) into a Readable stream. This bridges the async generator and stream worlds.
• highWaterMark tuning -- default is 16KB for byte streams, 16 objects for object mode. For large file processing, increase to 64KB--1MB. For high-throughput object pipelines, tune based on the average object size and desired latency.

6. Implement Cancellation with AbortController

Cancellation is cooperative -- the callee must check for cancellation; the caller signals it.

• Create a signal: const controller = new AbortController(); const { signal } = controller;. Pass signal into cancelable operations. Call controller.abort() to cancel.
• Integrate with native Node.js APIs: fs.readFile, fs.writeFile, http.request (via fetch), setTimeout (via timers/promises), and EventEmitter.once all accept an { signal } option. When the signal fires, they reject with AbortError (error.name === 'AbortError').
• Implement a timeout shorthand: AbortSignal.timeout(5000) (Node.js 17.3+) creates a signal that automatically aborts after 5000ms. Combine multiple signals with AbortSignal.any([sig1, sig2]) (Node.js 20.3+).
• Propagate signals through your own async functions: Accept signal as a parameter and check signal.throwIfAborted() at each async boundary. Listen for signal.addEventListener('abort', cleanup) to release resources (close sockets, cancel timers) when abort fires.
• Race pattern for timeout: For older Node.js versions without AbortSignal.timeout: Promise.race([operation(signal), new Promise((_, reject) => setTimeout(() => { controller.abort(); reject(new Error('Timeout')); }, 5000))]).
• Distinguish cancellation from errors: Always check error.name === 'AbortError' before treating a rejection as an unexpected error. Cancellation is not an error -- it is a normal control flow outcome. Log it differently and do not trigger alerts.

7. Implement Worker Threads

Use worker threads when profiling confirms the event loop is blocked by CPU-bound work, not for I/O.

• Determine if worker threads are needed -- profile first. Only introduce workers if a specific computation exceeds ~10ms consistently. Common legitimate uses: image processing, cryptographic operations (beyond what crypto offloads natively), JSON parsing of very large payloads (>500KB), or custom compression.
• Communication pattern -- use worker.postMessage(data) and worker.on('message', handler). For structured data, postMessage uses the structured clone algorithm (no functions, no prototypes beyond built-ins). Transfer ownership of ArrayBuffer with the transfer list to avoid copying: worker.postMessage(buffer, [buffer]).
• SharedArrayBuffer + Atomics -- use for high-frequency data sharing between the main thread and workers (e.g., a shared ring buffer). Requires --experimental-shared-memory flag (Node.js <14) or Cross-Origin-Isolation headers in some environments. Use Atomics.wait for worker synchronization; never use it on the main thread (it blocks).
• Worker pool pattern -- maintain a pool of N workers where N ≈ os.cpus().length - 1 (leave one CPU for the event loop). Implement a task queue in the main thread; dispatch tasks to idle workers. Track in-flight work with a Map<workerId, {resolve, reject}>.
• workerData for initialization -- pass static configuration to the worker at startup via workerData in the Worker constructor options. This is cloned once at creation; do not pass large mutable state here.
• Error handling -- listen for worker.on('error', handler) and worker.on('exit', code => { if (code !== 0) ... }). A worker that crashes does not crash the main process, but you must detect and replace it.

8. Use AsyncLocalStorage for Context Propagation

Request-scoped context should flow through async call chains without manual threading.

• Create a store per request -- instantiate AsyncLocalStorage once at module level, then call storage.run({ requestId, userId, traceId }, asyncHandler) at the beginning of each request. All async operations called within that callback inherit the store automatically.
• Access context anywhere -- call storage.getStore() inside any async function in the call chain. Returns undefined if called outside a run() context; always guard with a null check.
• Performance cost -- AsyncLocalStorage adds ~10--20% overhead to async context switches in high-throughput scenarios (>100k req/s). For most services this is negligible. Benchmark with your actual workload before optimizing.
• Combining with Promises -- AsyncLocalStorage correctly propagates through Promise.all, async/await, and setTimeout. It does NOT propagate across worker_threads message boundaries -- you must manually serialize and deserialize context when crossing thread boundaries.
• Common use cases -- distributed tracing (propagate trace ID to all log statements and downstream calls), per-request caching, audit logging, multi-tenant data isolation.

Output Format

When responding to a user's async pattern question, produce output in this structure:

## Async Pattern Analysis

### Problem Classification
- **Category:** [I/O-bound sequential | I/O-bound parallel | CPU-bound | Streaming | Cancellable | Context propagation | Rate-limited]
- **Node.js Versions Supported:** [minimum version for the recommended approach]
- **Key Constraint:** [the primary constraint driving the pattern choice]

### Pattern Decision Matrix

| Pattern | Use When | Avoid When | Overhead |
|---------|----------|------------|----------|
| async/await sequential | operations depend on prior results | operations are independent | minimal |
| Promise.all | parallel independent I/O, fail-fast needed | partial failure is acceptable | minimal |
| Promise.allSettled | parallel I/O, partial failure acceptable | all must succeed | minimal |
| Stream pipeline | unbounded data, backpressure needed | data fits in memory, latency irrelevant | low |
| Worker threads | CPU >10ms blocks event loop | I/O-bound work | medium (thread overhead) |
| AbortController | timeout or user-initiated cancel needed | fire-and-forget operations | minimal |
| AsyncLocalStorage | request context propagation | single-request linear code | low (~10-20% async overhead) |
| Concurrency limiter | N items, max K concurrent | single item, no rate limit | minimal |

### Recommended Implementation

[Pattern name and rationale in 2-3 sentences]

**Key decision:** [Why this pattern over the alternatives]

### Code Implementation

```javascript
// [Label the code with what it demonstrates]
// Node.js [version]+ required

[Complete, production-ready code example with error handling]

Pitfalls to Avoid

• [Specific anti-pattern relevant to this solution]
• [Specific anti-pattern relevant to this solution]
• [Specific anti-pattern relevant to this solution]

Testing Approach

• [How to test this specific pattern]
• [What edge cases to test]

---

## Rules

1. **NEVER use `Promise.all` when partial failure is acceptable.** `Promise.all` rejects on the first failure, canceling the wait for all other promises. If some failures are tolerable (e.g., enriching a response from multiple optional data sources), use `Promise.allSettled` and filter the results.

2. **NEVER use `array.forEach` with async callbacks.** `forEach` ignores the returned Promise and exits the loop synchronously before any async work completes. Use `for...of` for sequential execution or `Promise.all(array.map(...))` for parallel execution.

3. **NEVER block the event loop with synchronous work exceeding 1ms in a server context.** JSON parsing of large payloads, synchronous file reads (`fs.readFileSync`), and CPU-intensive crypto operations block ALL concurrent requests. Profile first; offload to worker threads if confirmed.

4. **ALWAYS use `stream.pipeline` from `stream/promises` instead of `.pipe()` in production code.** `.pipe()` does not propagate errors and does not destroy the source stream when the destination errors, causing memory leaks and resource exhaustion.

5. **ALWAYS write `return await promise` inside `try/catch` blocks.** Without `await`, the function returns before the promise settles, and rejections escape the `try/catch`. This is one of the most common and subtle bugs in async/await code.

6. **NEVER check `error.message` to detect `AbortError`.** The message string is not guaranteed. Always check `error.name === 'AbortError'` or `error instanceof DOMException` (in environments where `DOMException` is available). In Node.js 18+, also check `error.code === 'ABORT_ERR'` for Node-native AbortErrors.

7. **ALWAYS cap concurrency when processing arrays of I/O operations.** `Promise.all(largeArray.map(fetchFromDB))` with 10,000 items opens 10,000 simultaneous connections. Use `p-limit` with a concurrency of 5--50 (tuned to your connection pool size and target server limits) or implement a semaphore pattern.

8. **NEVER use `process.nextTick` in application code for control flow.** It was designed for internal Node.js library use. Recursive `nextTick` calls starve the event loop, preventing I/O callbacks from running. Use `setImmediate` or `queueMicrotask` for yielding in application code.

9. **ALWAYS pass `AbortSignal` through your async call stack.** Cancellation must be propagated to every async operation in the chain. Catching an `AbortError` and swallowing it without re-throwing or re-aborting breaks cancellation semantics -- the upstream caller's abort becomes ineffective.

10. **NEVER create Worker threads on a per-request basis.** Thread creation and teardown is expensive (10--100ms). Workers must be pooled. If you find yourself constructing `new Worker(...)` inside a request handler, refactor to use a pre-initialized thread pool.

11. **ALWAYS handle the `drain` event when manually writing to a Writable stream.** If `writable.write(chunk)` returns `false`, stop writing and resume only on `drain`. Ignoring this causes unbounded memory growth as Node.js buffers all the undelivered chunks.

12. **NEVER assume `async/await` adds parallelism.** `await` suspends the current async function at that point -- it does NOT run operations in parallel. Parallelism requires either `Promise.all` or separate invocations that are not awaited sequentially.

---

## Edge Cases

### Legacy Callback-Based Code Integration

When integrating with Node.js core APIs or third-party libraries that use the `(err, result)` callback style, use `util.promisify`. For APIs with non-standard callback signatures (e.g., callbacks with multiple success arguments, or `(result, err)` reversed order), create a manual wrapper with `new Promise`. Never use `util.promisify` on callback APIs that emit multiple values over time (like `EventEmitter`) -- that pattern requires stream or async iterator conversion, not promisification.

For `EventEmitter` patterns, use `events.once(emitter, 'eventName')` to await a single event, or `events.on(emitter, 'eventName')` to create an async iterable over all events. Always set a timeout or hook into the emitter's error event to avoid hanging forever if the expected event never fires.

### Unintended Promise Floating (Fire and Forget)

When you intentionally fire and forget (e.g., logging to an external service asynchronously without blocking the response), you MUST attach an error handler. `asyncOperation().catch(err => logger.error('background task failed', err))`. Without the `.catch`, a rejection becomes an unhandled rejection warning (Node.js 15+ throws by default, crashing the process). Document clearly in code comments that the fire-and-forget is intentional and that the error handler is the safety net.

### Async Generator Cleanup and Early Exit

When consuming an async generator with `for await...of` and breaking early, the generator's `finally` block executes -- but only if the generator object is properly returned. When the `for...of` loop breaks, JavaScript calls `generator.return()`, which triggers cleanup. However, if you convert the generator to an array (`Array.from`) or pass it to `Promise.all`, early exit cleanup does not happen automatically. In these cases, wrap the generator in a `try/finally` and manually call `generator.return()` on early exit.

For async generators wrapped as streams via `Readable.from`, stream destruction (`stream.destroy()`) correctly propagates back and triggers the generator's `finally` block.

### Combining AbortSignal with Promise.all

`Promise.all` is not abort-aware by default. If one operation aborts, `Promise.all` rejects, but the other in-flight operations continue running until they naturally complete or fail. To truly cancel all operations in a `Promise.all` group on first failure or external abort, create a single `AbortController`, pass its signal to all operations, and call `controller.abort()` in a `.catch()` handler on the `Promise.all`. This implements "cancel all on first error" semantics similar to structured concurrency.

```javascript
const controller = new AbortController();
const { signal } = controller;

try {
  await Promise.all(
    operations.map(op => op(signal))
  );
} catch (err) {
  controller.abort(); // cancel remaining operations
  throw err;
}

Worker Thread Module Loading and Warm-Up

Worker threads run in a fresh V8 context and must re-load and re-JIT all required modules. For workers that import large dependencies (e.g., image processing libraries, ML inference runtimes), the cold start can be 200--2000ms. Mitigate by:

• Pre-creating the worker pool at application startup, not at first use
• Sending a warm-up message that forces JIT compilation of hot code paths
• Using workerData to pass initialization config rather than sending it as the first task message

If your Node.js version supports it (18.7+), use --experimental-vm-modules and the Worker eval option to share already-parsed module code.

Tick Starvation from Recursive Microtasks

A recursive async pattern that schedules new microtasks before completing (e.g., async function loop() { await processItem(); return loop(); }) can starve the event loop of I/O callbacks because the microtask queue drains completely before each event loop phase. The event loop never reaches the poll phase to pick up new I/O events. Fix by yielding to the event loop periodically with await new Promise(resolve => setImmediate(resolve)) every N iterations, where N is tuned to keep I/O latency acceptable (typically every 100--1000 iterations).

AsyncLocalStorage Across Worker Thread Boundaries

AsyncLocalStorage context does NOT automatically propagate into worker_threads. Worker threads have their own independent async context. If you need request context (trace ID, request ID) inside a worker, serialize the relevant fields from storage.getStore() and pass them via workerData or postMessage. Inside the worker, reconstruct the context manually by storing the values in module-level variables or creating a new AsyncLocalStorage store with the received data.

Example

Input from user: "I need to process 50,000 user records from a database, enrich each one by calling an external REST API, and write the results to a file. The API has a rate limit of 100 requests per second and allows up to 20 concurrent connections. How do I structure this in Node.js?"

Async Pattern Analysis

Problem Classification

• Category: I/O-bound parallel with concurrency cap + streaming output
• Node.js Versions Supported: 18+ (for stream/promises, AbortSignal.timeout, async iteration)
• Key Constraint: API rate limit (100 req/s) and connection limit (20 concurrent) prevent naive Promise.all; data volume (50k records) prevents full in-memory buffering

Pattern Decision Matrix

Recommended Implementation

Use a streaming pipeline to read database records without buffering all 50k in memory, process each record through a Transform stream that applies concurrency and rate limiting, and write results to a file using stream.pipeline. The combination of a concurrency limiter (max 20 concurrent API calls) with a time-rate throttle (max 100/s) matches the API's constraints exactly.

Key decision: Streaming pipeline over Promise.all because 50,000 records cannot be loaded into memory simultaneously (typical record + enriched result ≈ 2--10KB each = 100--500MB). Streaming keeps memory bounded to highWaterMark × record size ≈ a few MB.

Code Implementation

// Streaming enrichment pipeline with concurrency + rate limiting
// Node.js 18+ required
// Dependencies: p-limit, p-throttle (npm install p-limit p-throttle)

import { pipeline, Transform } from 'node:stream';
import { pipeline as asyncPipeline } from 'node:stream/promises';
import { createWriteStream } from 'node:fs';
import pLimit from 'p-limit';
import pThrottle from 'p-throttle';

const CONCURRENCY_LIMIT = 20;   // max simultaneous open API connections
const RATE_LIMIT = 100;          // max API calls per second
const RATE_INTERVAL = 1000;      // 1 second window in ms

// Compose both limits: a call must acquire a concurrency slot AND
// pass the throttle before firing.
const limit = pLimit(CONCURRENCY_LIMIT);

const throttledFetch = pThrottle({
  limit: RATE_LIMIT,
  interval: RATE_INTERVAL,
})(async (userId, signal) => {
  const response = await fetch(
    `https://api.example.com/enrich/${userId}`,
    { signal }
  );
  if (!response.ok) {
    throw new Error(`API error ${response.status} for user ${userId}`);
  }
  return response.json();
});

// Wrap with concurrency limiter so both constraints apply
async function enrichUser(record, signal) {
  return limit(() => throttledFetch(record.id, signal));
}

// Transform stream: receives user record objects, outputs enriched JSON lines
class EnrichmentTransform extends Transform {
  constructor(signal) {
    super({ objectMode: true, highWaterMark: 50 }); // buffer max 50 objects
    this.signal = signal;
    this.pending = new Set();
  }

  _transform(record, _encoding, callback) {
    // Check for cancellation at each record boundary
    if (this.signal.aborted) {
      callback(new Error('Pipeline aborted'));
      return;
    }

    const enrichPromise = enrichUser(record, this.signal)
      .then(enriched => {
        const result = { ...record, ...enriched, enrichedAt: new Date().toISOString() };
        // Push as newline-delimited JSON
        if (!this.push(JSON.stringify(result) + '\n')) {
          // Backpressure: downstream is full, but Transform handles this via objectMode
        }
      })
      .catch(err => {
        // Decide: fail the whole pipeline, or log and skip?
        // Here: log and skip to maximize partial success
        console.error(`Failed to enrich user ${record.id}:`, err.message);
        this.push(JSON.stringify({ ...record, enrichmentError: err.message }) + '\n');
      })
      .finally(() => {
        this.pending.delete(enrichPromise);
      });

    this.pending.add(enrichPromise);

    // Do not await -- allow up to highWaterMark records to be in-flight
    // Signal ready for next record immediately (backpressure is via p-limit)
    callback();
  }

  _flush(callback) {
    // Wait for all in-flight enrichments to complete before closing
    Promise.all(this.pending)
      .then(() => callback())
      .catch(callback);
  }
}

// Database cursor as an async iterable (pseudo-code for your DB driver)
// Replace with your actual driver's streaming/cursor API:
// - pg: client.query(new Cursor('SELECT ...'))
// - MySQL2: connection.query('SELECT ...').stream()
// - MongoDB: collection.find({}).stream()
async function* databaseCursor(db) {
  const cursor = db.collection('users').find({}).batchSize(500);
  try {
    for await (const doc of cursor) {
      yield doc;
    }
  } finally {
    await cursor.close();
  }
}

// Main pipeline
async function runEnrichmentPipeline(db) {
  // 5-minute overall timeout for the entire pipeline
  const signal = AbortSignal.timeout(5 * 60 * 1000);

  const sourceStream = Readable.from(databaseCursor(db), {
    objectMode: true,
    highWaterMark: 100, // read ahead 100 records from DB cursor
  });

  const enrichTransform = new EnrichmentTransform(signal);

  const outputStream = createWriteStream('./enriched-users.ndjson', {
    flags: 'w',
    encoding: 'utf8',
  });

  console.time('enrichment-pipeline');

  try {
    await asyncPipeline(
      sourceStream,
      enrichTransform,
      outputStream
    );
    console.timeEnd('enrichment-pipeline');
    console.log('Pipeline complete');
  } catch (err) {
    if (err.name === 'AbortError' || err.name === 'TimeoutError') {
      console.error('Pipeline timed out after 5 minutes');
    } else {
      console.error('Pipeline failed:', err);
    }
    throw err;
  }
}

Pitfalls to Avoid

• Do NOT use Promise.all(allRecords.map(enrichUser)) -- this fires 50,000 API requests simultaneously, saturates the connection pool, likely triggers API bans, and buffers all results in memory.
• Do NOT use sequential for await without the concurrency limiter -- it processes 1 record at a time, achieving roughly 1 req/s instead of the allowed 100 req/s, making the job ~100x slower than necessary.
• Do NOT call callback() inside _transform after callback(error) -- calling the Transform callback twice corrupts the stream state. Use early return after callback(error).
• Do NOT forget _flush -- without it, in-flight enrichments that haven't pushed their results yet are dropped when the source stream ends.
• Do NOT ignore backpressure signals from the output file write stream -- asyncPipeline handles this, but if you replace it with manual .pipe() calls, you must handle drain events.

Testing Approach

• Unit test enrichUser with a mocked fetch that records call timing. Assert that no more than 20 calls are in-flight simultaneously (use a counter incremented on call start, decremented on completion).
• Test rate limiting by recording timestamps of all mock API calls and asserting no 1-second window contains more than 100 calls.
• Test the timeout by injecting an AbortSignal that is pre-aborted (AbortSignal.abort()) and asserting the pipeline rejects with an AbortError rather than hanging.
• Test _flush completion by using a slow mock API (delayed with setTimeout) and asserting the output file contains results for all records that were started before the source stream ended.
• Test partial failure handling by having the mock API reject for specific user IDs and asserting those records appear in the output with enrichmentError set rather than crashing the pipeline.
• Integration test against a real or Docker-based instance of your database and a local mock API server (nock or msw in server mode) to validate end-to-end throughput approaches 100 req/s.