Cancellation and cleanup

Stopping an execution

Acquiring a StreamIterator with iterate(source, { signal }) gives you explicit control over how an execution stops and how its teardown is observed:

Method / Property	Behavior
iterator.cancel(reason)	Requests an immediate stop and aborts stage AbortSignals. Any pending next() rejects right away without waiting for cleanup.
iterator.return()	Requests a stop, aborts stage AbortSignals, and waits for all upstream cleanup and finalizers to complete before resolving.
iterator.throw(error)	Requests failure and rejects immediately without waiting for cleanup (does not forward .throw() into upstream generators).
iterator.closed	Promise that always resolves with a CloseReport once teardown finishes.

Only one next() call may be pending on a StreamIterator at a time; calling next() concurrently rejects with TypeError. When an execution fails or is cancelled, next() rejects once with the primary reason, and all subsequent next() calls resolve to { done: true, value: undefined }.

return() resolves to { done: true, value: undefined } on normal completion. It rejects with the primary reason when the execution has failed, including a failure caused only by cleanup. After cancellation it resolves to { done: true, value: undefined }, even if cleanup fails; inspect suppressedErrors in the resolved closed report for those errors. An existing cancellation or failure takes precedence over a later throw(error). Await closed after throw() to wait for teardown.

import { defer, finalize, iterate } from "mogami-stream";

export async function readOne(signal: AbortSignal) {
  const source = defer(({ signal }) =>
    (async function* () {
      const response = await fetch("https://example.com/data", { signal });
      if (!response.ok) {
        await response.body?.cancel();
        throw new Error(`HTTP ${response.status}`);
      }
      yield await response.text();
    })(),
  ).pipe(finalize(() => console.log("cleanup complete")));

  const iterator = iterate(source, { signal });
  try {
    return await iterator.next();
  } finally {
    try {
      await iterator.return();
    } finally {
      const report = await iterator.closed;
      if (report.kind !== "completed") console.error(report.reason);
      for (const error of report.suppressedErrors) console.error(error);
    }
  }
}

Cooperative cancellation via AbortSignal

JavaScript async iterators cannot forcibly preempt an arbitrary await. When cancel(), return(), take(n), or an external AbortSignal stops a stage, mogami-stream aborts the signal passed into defer and operator callbacks (map, filter, forEach, etc.).

Always pass that signal into cancellable APIs (fetch, fs.readFile, database queries, timers). If an upstream read or callback ignores signal and never settles, return(), closed, and terminal consumers will continue waiting for that operation to finish before completing teardown.

Finalizer order and take(n) teardown

finalize(effect) registers a synchronous or asynchronous teardown callback on a stage:

• The callback runs once per execution if the stage was actually started (i.e., received at least one next() read).
• It runs after all upstream child stages have closed and all in-flight operations in the stage have settled, and its returned promise is awaited.
• When multiple finalize stages are nested in a pipeline, they execute from inner (upstream) stages outward (downstream).

take(n) begins closing its upstream stage as soon as the nth item arrives—before delivering that nth item to downstream—while keeping downstream callback signals active so downstream operators can still process the final item.

Owning resources in a generator

Use defer to create a fresh generator per execution and keep acquisition and release together with try/finally.

import { defer, take } from "mogami-stream";

interface Cursor<T> {
  rows(options: { signal: AbortSignal }): AsyncIterable<T>;
  close(): Promise<void>;
}

export async function readFirstRows<T>(
  openCursor: (options: { signal: AbortSignal }) => Promise<Cursor<T>>,
  signal: AbortSignal,
) {
  const pages = defer(({ signal }) =>
    (async function* () {
      const cursor = await openCursor({ signal });
      try {
        yield* cursor.rows({ signal });
      } finally {
        await cursor.close();
      }
    })(),
  );

  const runA = await pages.pipe(take(10)).toArray({ signal });
  const runB = await pages.pipe(take(10)).toArray({ signal });
  return { runA, runB };
}

Each execution opens a cursor on its first read and closes it on completion, failure, or early termination. Pass signal to row reads too. If acquisition fails partway through, openCursor must release any partial resources.

Owning resources during asynchronous setup

defer factories and catchError handlers receive a ResourceContext with signal and acquire(acquire, release). Use acquire when setup opens a cursor, session, file, or another resource before returning its source:

import { catchError, defer } from "mogami-stream";

interface Cursor {
  read(signal: AbortSignal): Promise<IteratorResult<string>>;
  close(): Promise<void>;
}

type OpenCursor = (signal: AbortSignal) => Promise<Cursor>;

// The source borrows the cursor. Only the registered release callback closes it.
async function* rows(cursor: Cursor, signal: AbortSignal) {
  while (true) {
    const result = await cursor.read(signal);
    if (result.done) return;
    yield result.value;
  }
}

export function readWithFallback(
  openCursor: OpenCursor,
  openBackup: OpenCursor,
) {
  return defer(async ({ signal, acquire }) => {
    const cursor = await acquire(openCursor, (cursor) => cursor.close());
    return rows(cursor, signal);
  }).pipe(
    catchError(async (_error, { signal, acquire }) => {
      const backup = await acquire(openBackup, (cursor) => cursor.close());
      return rows(backup, signal);
    }),
  );
}

Acquisition and release timing

Call and await context.acquire while the factory or handler is running. The acquisition callback receives the stage's signal. If the stage has already stopped when acquisition would start, the callback is skipped.

If acquisition succeeds, the stage registers release(resource) before checking cancellation, so a resource arriving after cancellation is still released. In that case the acquire promise rejects with the cancellation reason and the source is not opened.

The stage calls each registered release exactly once, after its child sources close and its in-flight work settles, on completion, failure, or early termination. Releases run in reverse registration order and are awaited by closed and terminal consumers. A blocked acquisition or release therefore keeps teardown pending. The context cannot start new acquisitions after the factory or handler settles.

Acquisition and release errors

Acquisition failure fails the owning stage; an upstream defer failure can be handled by catchError. The acquisition callback remains responsible for any partially allocated resources when it throws.

Release failures follow the cleanup error precedence rules.

Ownership and double release

Ownership applies to each successful acquire call. The returned source borrows that resource; it must not release it again in its iterator's return() or finally. The example's rows generator closes independently, and the registered callback closes the cursor afterward. Do not register the same resource twice.

To adapt an already-open iterator, register its return() as the release callback and expose its next() through a borrowing source instead of returning that same iterator directly.

Ordinary returned sources retain their existing contract: the library owns an iterator only after opening the source. A plain source returned after cancellation is neither inspected nor opened nor closed, because obtaining its iterator could start unrelated work. Resources allocated directly by a factory or handler remain that callback's responsibility until ownership is explicitly registered.

Inspecting CloseReport

iterator.closed returns the termination state after cleanup. See Error handling for the CloseReport structure and error precedence.

Recovery boundaries with catchError

See catchError in Error handling for switching to a fallback source and the boundaries of error recovery.

Collecting a result with its report

See the terminal result and report recipe to collect results with secondary errors, or to observe a timeout before cleanup completes.

StreamIterator · CloseReport · finalize · catchError