Run stream events and callbacks in Node's AsyncLocalStorage contexts

Stream events and callbacks ran with no store. Node's streams run each step in the context of the code that triggers it: a step deferred with process.nextTick (resume, emitReadable, endReadable, afterWrite, the destroy and finish steps) in the context that scheduled it, and other events in the context of the push(), write(), end(), destroy() or pipe() call that emits them. Native-backed streams run their events in the context of their handle. The runtime now captures the active context where it defers a stream step and enters it around the step's dispatch: - the stream tick queue keeps each tick's scheduling context; - the runtime's own process.nextTick jobs (zlib completions, pipeline and finished steps, readline) run in the context they were queued in, as user ticks already did; - fs read and write stream requests keep the context of the call that started them, so a stream's open, reads, writes and close follow its creation; - zlib streams keep their creation context and run each engine step's completion in it. In programs that use AsyncLocalStorage, stream.finished and stream.pipeline callbacks are bound to the context of their call, like the other completion callbacks. Programs that do not use AsyncLocalStorage compile to the same IR. child_process and http2 events are unchanged. Corpus 7383 and 7384.
2026-10-27 06:43:19 +08:00 · 2026-10-22 17:05:08 -05:00
parent 14098794d0
commit 1d8d19e634
14 changed files with 525 additions and 28 deletions
@@ -3,5 +3,8 @@
  Events of `net`, `tls`, `http` and `https` sockets, servers, requests and responses run with the stores Node runs them with.
  A socket's or client request's events see the context of the call that created it, and a server's events and its requests' and responses' events see the context of its `listen()` call.
  Socket `write` and `end` callbacks, `'finish'` and `server.close(callback)` see the context of their call.
-  Events and callbacks of other streams run with no store, where Node runs them in the context of the code that triggers them.
+  Stream events and callbacks run with the stores of the code that triggers them, as in Node.
+  Stream steps that Node defers with `process.nextTick` see the context that scheduled them, and `stream.finished` and `stream.pipeline` callbacks see the context of their call.
+  `fs` read and write streams and `zlib` streams run their events and callbacks in the context that created them.
+  Events of `child_process` children and of `http2` sessions and streams do not yet run in the contexts Node gives them.
  `AsyncResource` has no lowering yet.
@@ -1392,7 +1392,9 @@
          "tests/corpus/7111-als-bind-snapshot-options.ts",
          "tests/corpus/7112-als-bind-snapshot-dynamic.mjs",
          "tests/corpus/7119-als-net-event-context.mjs",
-          "tests/corpus/7380-als-http-event-context.mjs"
+          "tests/corpus/7380-als-http-event-context.mjs",
+          "tests/corpus/7383-als-stream-events.ts",
+          "tests/corpus/7384-als-stream-events.mjs"
        ]
      },
      "anchorSource": "chapter"
@@ -1447,7 +1449,9 @@
          "tests/corpus/7111-als-bind-snapshot-options.ts",
          "tests/corpus/7112-als-bind-snapshot-dynamic.mjs",
          "tests/corpus/7119-als-net-event-context.mjs",
-          "tests/corpus/7380-als-http-event-context.mjs"
+          "tests/corpus/7380-als-http-event-context.mjs",
+          "tests/corpus/7383-als-stream-events.ts",
+          "tests/corpus/7384-als-stream-events.mjs"
        ]
      },
      "anchorSource": "exact"
@@ -1525,7 +1529,9 @@
          "tests/corpus/7111-als-bind-snapshot-options.ts",
          "tests/corpus/7112-als-bind-snapshot-dynamic.mjs",
          "tests/corpus/7119-als-net-event-context.mjs",
-          "tests/corpus/7380-als-http-event-context.mjs"
+          "tests/corpus/7380-als-http-event-context.mjs",
+          "tests/corpus/7383-als-stream-events.ts",
+          "tests/corpus/7384-als-stream-events.mjs"
        ]
      },
      "anchorSource": "exact"
@@ -1557,7 +1563,9 @@
          "tests/corpus/7111-als-bind-snapshot-options.ts",
          "tests/corpus/7112-als-bind-snapshot-dynamic.mjs",
          "tests/corpus/7119-als-net-event-context.mjs",
-          "tests/corpus/7380-als-http-event-context.mjs"
+          "tests/corpus/7380-als-http-event-context.mjs",
+          "tests/corpus/7383-als-stream-events.ts",
+          "tests/corpus/7384-als-stream-events.mjs"
        ]
      },
      "anchorSource": "exact"
@@ -1589,7 +1597,9 @@
          "tests/corpus/7111-als-bind-snapshot-options.ts",
          "tests/corpus/7112-als-bind-snapshot-dynamic.mjs",
          "tests/corpus/7119-als-net-event-context.mjs",
-          "tests/corpus/7380-als-http-event-context.mjs"
+          "tests/corpus/7380-als-http-event-context.mjs",
+          "tests/corpus/7383-als-stream-events.ts",
+          "tests/corpus/7384-als-stream-events.mjs"
        ]
      },
      "anchorSource": "exact"
@@ -1621,7 +1631,9 @@
          "tests/corpus/7111-als-bind-snapshot-options.ts",
          "tests/corpus/7112-als-bind-snapshot-dynamic.mjs",
          "tests/corpus/7119-als-net-event-context.mjs",
-          "tests/corpus/7380-als-http-event-context.mjs"
+          "tests/corpus/7380-als-http-event-context.mjs",
+          "tests/corpus/7383-als-stream-events.ts",
+          "tests/corpus/7384-als-stream-events.mjs"
        ]
      },
      "anchorSource": "exact"
@@ -1653,7 +1665,9 @@
          "tests/corpus/7111-als-bind-snapshot-options.ts",
          "tests/corpus/7112-als-bind-snapshot-dynamic.mjs",
          "tests/corpus/7119-als-net-event-context.mjs",
-          "tests/corpus/7380-als-http-event-context.mjs"
+          "tests/corpus/7380-als-http-event-context.mjs",
+          "tests/corpus/7383-als-stream-events.ts",
+          "tests/corpus/7384-als-stream-events.mjs"
        ]
      },
      "anchorSource": "exact"
@@ -1962,6 +1976,8 @@
          "tests/corpus/7112-als-bind-snapshot-dynamic.mjs",
          "tests/corpus/7119-als-net-event-context.mjs",
          "tests/corpus/7380-als-http-event-context.mjs",
+          "tests/corpus/7383-als-stream-events.ts",
+          "tests/corpus/7384-als-stream-events.mjs",
          "tests/harness/node-test.test.ts"
        ]
      },
@@ -2529,7 +2545,9 @@
          "tests/corpus/7111-als-bind-snapshot-options.ts",
          "tests/corpus/7112-als-bind-snapshot-dynamic.mjs",
          "tests/corpus/7119-als-net-event-context.mjs",
-          "tests/corpus/7380-als-http-event-context.mjs"
+          "tests/corpus/7380-als-http-event-context.mjs",
+          "tests/corpus/7383-als-stream-events.ts",
+          "tests/corpus/7384-als-stream-events.mjs"
        ]
      },
      "anchorSource": "exact"
@@ -429,7 +429,7 @@
    { "symbols": ["crypto.hkdf", "crypto.hkdfSync", "crypto.scrypt", "crypto.scryptSync"], "status": "partial", "evidence": ["tests/corpus/node24-crypto-kdf.ts", "tests/corpus/node24-crypto-kdf-errors.ts", "tests/corpus/node24-crypto-kdf-options.ts", "tests/corpus/5438-crypto-kdf-digests.ts"] },
    { "symbols": ["EventEmitter", "emitter.on", "emitter.addListener", "emitter.once", "emitter.prependListener", "emitter.prependOnceListener", "emitter.off", "emitter.removeListener", "emitter.removeAllListeners", "emitter.emit", "emitter.listenerCount", "emitter.listeners", "emitter.rawListeners", "emitter.eventNames", "emitter.setMaxListeners", "emitter.getMaxListeners", "events.once"], "status": "partial", "evidence": ["tests/corpus/1677-emitter-listeners.ts", "tests/corpus/2620-ee-override-once-order.ts"] },
    { "symbols": ["events.getEventListeners", "events.getMaxListeners", "events.listenerCount", "events.setMaxListeners", "events.defaultMaxListeners"], "status": "partial", "evidence": ["tests/corpus/5885-events-max-listener-helpers.ts", "tests/corpus/5846-events-esm-default-max.ts", "tests/corpus/5847-events-cjs-default-max.cjs"] },
-    { "symbols": ["AsyncLocalStorage", "asyncLocalStorage.run", "asyncLocalStorage.exit", "asyncLocalStorage.getStore", "asyncLocalStorage.enterWith", "asyncLocalStorage.disable", "asyncLocalStorage.name", "AsyncLocalStorage.bind", "AsyncLocalStorage.snapshot"], "status": "partial", "evidence": ["tests/corpus/2213-async-local-storage.cjs", "tests/corpus/5815-als-callback-context.ts", "tests/corpus/7110-als-completion-callbacks.ts", "tests/corpus/7111-als-bind-snapshot-options.ts", "tests/corpus/7112-als-bind-snapshot-dynamic.mjs", "tests/corpus/7119-als-net-event-context.mjs", "tests/corpus/7380-als-http-event-context.mjs"] },
+    { "symbols": ["AsyncLocalStorage", "asyncLocalStorage.run", "asyncLocalStorage.exit", "asyncLocalStorage.getStore", "asyncLocalStorage.enterWith", "asyncLocalStorage.disable", "asyncLocalStorage.name", "AsyncLocalStorage.bind", "AsyncLocalStorage.snapshot"], "status": "partial", "evidence": ["tests/corpus/2213-async-local-storage.cjs", "tests/corpus/5815-als-callback-context.ts", "tests/corpus/7110-als-completion-callbacks.ts", "tests/corpus/7111-als-bind-snapshot-options.ts", "tests/corpus/7112-als-bind-snapshot-dynamic.mjs", "tests/corpus/7119-als-net-event-context.mjs", "tests/corpus/7380-als-http-event-context.mjs", "tests/corpus/7383-als-stream-events.ts", "tests/corpus/7384-als-stream-events.mjs"] },
    { "symbols": ["StringDecoder", "stringDecoder.write", "stringDecoder.end"], "status": "partial", "evidence": ["tests/corpus/1662-string-decoder-encodings.ts", "tests/corpus/node24-string-decoder.ts", "tests/corpus/5882-string-decoder-runtime-encoding.ts"] },
    { "symbols": ["setImmediate", "setInterval", "setTimeout", "clearImmediate", "clearInterval", "clearTimeout", "timersPromises.setTimeout", "timersPromises.setImmediate", "timersPromises.setInterval", "immediate.hasRef", "immediate.ref", "immediate.unref", "timeout.hasRef", "timeout.ref", "timeout.unref", "timeout.refresh"], "status": "partial", "evidence": ["tests/corpus/1804-timers-module.ts", "tests/corpus/1805-timer-callback-args.ts", "tests/corpus/2093-timers-promises.ts", "tests/corpus/2852-timers-promises-interval.ts", "tests/corpus/5830-timers-promises-signal.ts"] },
    { "symbols": ["timersPromises.scheduler.wait", "timersPromises.scheduler.yield"], "status": "partial", "evidence": ["tests/corpus/5839-timers-promises-scheduler.ts", "tests/corpus/5889-timers-scheduler-signal.ts"] },
@@ -1,7 +1,8 @@
 /* Completion callbacks in programs that use AsyncLocalStorage. Node runs
 * the callback of an operation a callback-style API starts (fs.readFile,
 * crypto.randomBytes, zlib.gzip, dns.lookup, server.listen, net.connect,
- * http.get, ...) with the stores active when the call started it. The runtime invokes these callbacks from
+ * http.get, stream.finished, stream.pipeline, ...) with the stores active
+ * when the call started it. The runtime invokes these callbacks from
 * the event loop, where the context is empty, so in programs whose
 * scheduling carries contexts (RuntimeFeatures asyncContext) each such
 * argument is bound to the active context at the call: als.bindCallback
@@ -50,6 +51,16 @@ const COMPLETION_CALLBACKS: ReadonlyMap<IrLibFn, CallbackPosition> = new Map<
  ["net.listenPathCb", 2],
  ["process.stderrWriteBytesCb", 2],
  ["process.stdoutWriteBytesCb", 2],
+  // stream.finished and stream.pipeline callbacks, which Node binds to
+  // the context of the call (eos' and pipeline's AsyncResource.bind).
+  ["stream.finished", 1],
+  ["stream.finishedDyn", 1],
+  ["stream.finishedOpts", "last"],
+  ["stream.finishedOptsSignal", "last"],
+  ["stream.pipeline", "last"],
+  ["stream.pipelineDyn", "last"],
+  ["stream.pipelineStages", "last"],
+  ["stream.pipelineStagesDyn", "last"],
  ["zlib.cbOpts", 3],
  ["zlib.cbOptsDyn", 3],
  ["zlib.deflateCb", 1],
@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7383-als-stream-events.ts"
+ ],
+ "diags": []
+}
@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7384-als-stream-events.mjs"
+ ],
+ "diags": []
+}
@@ -1235,10 +1235,14 @@ static SCR_TL ScrNtick *scr_nt_tail = NULL;
 * it, because 'exit' listeners run AFTER the loop's teardown and may
 * enqueue ticks that must never run (Node) yet must not leak. */

+static ScrClosure *scr_als_bind(ScrClosure *cb /*moves*/);
+
+/* A tick runs with the AsyncLocalStorage context active when it was
+ * queued, as Node's process.nextTick does. */
 void scr_next_tick(ScrClosure *cb /*moves*/) {
  ScrNtick *t = calloc(1, sizeof *t);
  if (!t) scr_oom();
-  t->cb = cb;
+  t->cb = scr_als_bind(cb);
  if (scr_nt_tail) scr_nt_tail->next = t;
  else scr_nt_head = t;
  scr_nt_tail = t;
@@ -4022,7 +4026,7 @@ double scr_set_timeout_handle_als(ScrClosure *cb, double ms) {
  return scr_set_timeout_handle(scr_als_bind(cb), ms);
 }
 double scr_set_immediate_als(ScrClosure *cb) { return scr_set_immediate(scr_als_bind(cb)); }
-void scr_next_tick_als(ScrClosure *cb) { scr_next_tick(scr_als_bind(cb)); }
+void scr_next_tick_als(ScrClosure *cb) { scr_next_tick(cb); } /* scr_next_tick binds */
 void scr_queue_microtask_als(ScrClosure *cb) { scr_queue_microtask(scr_als_bind(cb)); }

 /* A typed callback bound to the context active when it was bound: the
@@ -79,6 +79,9 @@ typedef struct ScrFssOp {
  int64_t result;     /* answer or -errno */
  int retries;        /* WRITE: zero-byte retries */
  ScrError *carry;    /* FSYNC/CLOSE: the error to pass on (+1) */
+  /* The AsyncLocalStorage context the step was started in (+1): its
+   * continuation runs in it, as Node runs an fs request's callback. */
+  ScrAlsCtx *ctx;
 } ScrFssOp;

 static void scr_fss_work(void *p) {
@@ -101,6 +104,7 @@ static void scr_fss_op_destroy(void *p) {
  free(op->buf);
  if (op->chunk) scr_bytes_release(op->chunk);
  if (op->carry) scr_error_release(op->carry);
+  scr_als_ctx_release(op->ctx);
  free(op);
 }

@@ -111,13 +115,20 @@ static ScrFssOp *scr_fss_op_new(ScrStream *s, ScrFssOpKind kind, int fd) {
  op->kind = kind;
  op->fd = fd;
  op->pos = -1;
+  op->ctx = scr_als_capture();
  return op;
 }

 static void scr_fss_after(void *p);

+static void scr_fss_after_in_context(void *p) {
+  ScrAlsCtx *prev = scr_als_dispatch_enter(((ScrFssOp *)p)->ctx);
+  scr_fss_after(p);
+  scr_als_dispatch_leave(prev);
+}
+
 static void scr_fss_submit(ScrFssOp *op) {
-  scr_work_submit(op, &scr_fss_work, &scr_fss_after, &scr_fss_op_destroy);
+  scr_work_submit(op, &scr_fss_work, &scr_fss_after_in_context, &scr_fss_op_destroy);
 }

 /* ── close (Node's close/_close) ──────────────────────────────────────── */
@@ -515,6 +515,10 @@ typedef struct ScrStreamTick {
  ScrError *err;   /* owned or NULL (SCR_ST_ERROR) */
  ScrClosure *cb;  /* owned or NULL (SCR_ST_AFTER_WRITE / SCR_ST_WCB_ERR / SCR_ST_WRITE_ERR) */
  bool hold;       /* SCR_ST_AFTER_WRITE: counted in w.after_holds */
+  /* The AsyncLocalStorage context active when the tick was scheduled (+1,
+   * NULL outside a store): Node schedules these steps with
+   * process.nextTick, so their events and callbacks run in it. */
+  ScrAlsCtx *ctx;
  struct ScrStreamTick *next;
 } ScrStreamTick;

@@ -531,6 +535,7 @@ static void scr_st_tick(ScrStream *s, ScrStreamTickOp op, ScrError *err /*moves*
  t->op = op;
  t->err = err;
  t->cb = cb;
+  t->ctx = scr_als_capture();
  if (scr_st_tail) scr_st_tail->next = t;
  else scr_st_head = t;
  scr_st_tail = t;
@@ -6861,6 +6866,21 @@ static void scr_stream_run_tick(ScrStreamTick *t) {

 static void scr_stream_dispatch(void);

+/* A tick runs in the context it was scheduled in. */
+static void scr_stream_run_tick_in_context(ScrStreamTick *t) {
+  ScrAlsCtx *prev = scr_als_dispatch_enter(t->ctx);
+  scr_stream_run_tick(t);
+  scr_als_dispatch_leave(prev);
+}
+
+static void scr_stream_tick_free(ScrStreamTick *t) {
+  scr_stream_release(t->s);
+  if (t->err) scr_error_release(t->err);
+  if (t->cb) scr_closure_release(t->cb);
+  scr_als_ctx_release(t->ctx);
+  free(t);
+}
+
 /* One tick-marker's dispatch: the queue's FIFO head (markers and entries
 * are enqueued 1:1 in the same order). After an uncaught throw the
 * remaining entries drop through the station's exc branch below, keeping
@@ -6870,11 +6890,8 @@ static void scr_stream_dispatch_one(void) {
  if (t == NULL || scr_exc_pending()) return;
  scr_st_head = t->next;
  if (scr_st_head == NULL) scr_st_tail = NULL;
-  scr_stream_run_tick(t);
-  scr_stream_release(t->s);
-  if (t->err) scr_error_release(t->err);
-  if (t->cb) scr_closure_release(t->cb);
-  free(t);
+  scr_stream_run_tick_in_context(t);
+  scr_stream_tick_free(t);
  if (scr_exc_pending()) scr_stream_dispatch(); /* its exc branch drops the rest */
 }

@@ -6883,11 +6900,8 @@ static void scr_stream_dispatch(void) {
    ScrStreamTick *t = scr_st_head;
    scr_st_head = t->next;
    if (scr_st_head == NULL) scr_st_tail = NULL;
-    scr_stream_run_tick(t);
-    scr_stream_release(t->s);
-    if (t->err) scr_error_release(t->err);
-    if (t->cb) scr_closure_release(t->cb);
-    free(t);
+    scr_stream_run_tick_in_context(t);
+    scr_stream_tick_free(t);
  }
  /* An uncaught throw ends the process (the loop returns to main): the
   * undelivered ticks are DROPPED — release their references here so the
@@ -6896,10 +6910,7 @@ static void scr_stream_dispatch(void) {
    while (scr_st_head != NULL) {
      ScrStreamTick *t = scr_st_head;
      scr_st_head = t->next;
-      scr_stream_release(t->s);
-      if (t->err) scr_error_release(t->err);
-      if (t->cb) scr_closure_release(t->cb);
-      free(t);
+      scr_stream_tick_free(t);
    }
    scr_st_tail = NULL;
  }
@@ -51,6 +51,9 @@ typedef struct ScrZlibStream {
  bool done_pending;   /* Transform's kCallback: completion waits for _read */
  double rlen_before;  /* readableLength when the chunk's transform began */
  ScrBytes *flush_bufs[6]; /* zero-length sentinels, one per flush kind */
+  /* The AsyncLocalStorage context of the stream's creation (+1, NULL
+   * outside a store): Node runs the handle's step callbacks in it. */
+  ScrAlsCtx *als;
 } ScrZlibStream;

 static void scr_zlib_stream_oom(void) {
@@ -81,6 +84,7 @@ static void scr_zs_release_v(void *p) {
  for (size_t i = 0; i < 6; i++) {
    if (zs->flush_bufs[i]) scr_bytes_release(zs->flush_bufs[i]);
  }
+  scr_als_ctx_release(zs->als);
  free(zs);
 }

@@ -242,6 +246,7 @@ static void scr_zs_step_after(void *payload) {
  ScrZlibStep *op = payload;
  ScrZlibStream *zs = op->zs;
  ScrStream *s = op->s;
+  ScrAlsCtx *prev = scr_als_dispatch_enter(zs->als);
  zs->write_in_progress = false;
  ScrZlibError ze;
  if (scr_zlib_engine_error(zs->eng, &ze)) {
@@ -252,6 +257,7 @@ static void scr_zs_step_after(void *payload) {
    scr_zs_process(zs, s);
  }
  if (zs->pending_close) scr_zs_close_engine(zs);
+  scr_als_dispatch_leave(prev);
 }

 static void scr_zs_step_destroy(void *payload) {
@@ -463,6 +469,7 @@ ScrStream *scr_zlib_stream_new(double mode_d, ScrDyn *opts) {
  ScrZlibStream *zs = calloc(1, sizeof *zs);
  if (!zs) scr_zlib_stream_oom();
  zs->rc = 1;
+  zs->als = scr_als_capture();
  zs->mode = mode;
  zs->eng = scr_zlib_engine_from_options(mode, &zo);
  scr_zlib_options_drop(&zo);
@@ -0,0 +1,204 @@
+// AsyncLocalStorage contexts of stream events and callbacks. A stream
+// step that Node defers with process.nextTick runs in the context of the
+// code that scheduled it; other events run in the context of the call that
+// emits them: push(), write(), end(), destroy() or pipe(). stream.finished
+// and stream.pipeline callbacks run in the context of their call. fs read
+// and write streams and zlib streams run their events in the context that
+// created them, as Node's handles do.
+import { AsyncLocalStorage } from "node:async_hooks";
+import * as fs from "node:fs";
+import * as os from "node:os";
+import * as path from "node:path";
+import { finished, pipeline, Readable, Transform, Writable } from "node:stream";
+import { pipeline as pipelinePromise } from "node:stream/promises";
+import * as zlib from "node:zlib";
+
+const store = new AsyncLocalStorage<string>();
+const lines: string[] = [];
+const log = (what: string): void => {
+  lines.push(`${what}: ${store.getStore() ?? "none"}`);
+};
+const step = (fn: (done: () => void) => void): Promise<void> =>
+  new Promise<void>((resolve) => fn(resolve));
+
+// pipe(): the destination's writes and 'finish' follow the pipe() call.
+await step((done) => {
+  const source = new Readable({ read(): void {} });
+  const sink = new Writable({
+    write(chunk: Buffer, encoding: BufferEncoding, cb: (error?: Error | null) => void): void {
+      log("pipe write");
+      cb();
+    },
+  });
+  sink.on("finish", () => {
+    log("pipe finish");
+    done();
+  });
+  store.run("pipe", () => source.pipe(sink));
+  store.run("push", () => {
+    source.push("a");
+    source.push(null);
+  });
+});
+
+// A Transform's steps follow write() and end(); 'end' follows end().
+await step((done) => {
+  const upper = new Transform({
+    transform(chunk: Buffer, encoding: BufferEncoding, cb: (error?: Error | null, data?: string) => void): void {
+      log("transform");
+      cb(null, chunk.toString().toUpperCase());
+    },
+    flush(cb: (error?: Error | null, data?: string) => void): void {
+      log("flush");
+      cb(null, "!");
+    },
+  });
+  upper.on("data", () => log("transform data"));
+  upper.on("end", () => {
+    log("transform end");
+    done();
+  });
+  store.run("write", () => upper.write("a"));
+  store.run("end", () => upper.end());
+});
+
+// A write that completes later: its callback, _final, 'prefinish', the
+// end() callback, 'finish' and 'close' follow the write() call.
+await step((done) => {
+  const slow = new Writable({
+    write(chunk: Buffer, encoding: BufferEncoding, cb: (error?: Error | null) => void): void {
+      log("slow write");
+      setImmediate(cb);
+    },
+    final(cb: (error?: Error | null) => void): void {
+      log("final");
+      cb();
+    },
+  });
+  slow.on("prefinish", () => log("prefinish"));
+  slow.on("finish", () => log("finish"));
+  slow.on("close", () => {
+    log("close");
+    done();
+  });
+  store.run("slow-write", () => slow.write("a", () => log("write callback")));
+  store.run("slow-end", () => slow.end(() => log("end callback")));
+});
+
+// 'readable' and 'end' follow the push() that triggers them.
+await step((done) => {
+  const source = new Readable({ read(): void {} });
+  source.on("readable", () => {
+    log("readable");
+    while (source.read() !== null) {}
+  });
+  source.on("end", () => {
+    log("readable end");
+    done();
+  });
+  store.run("push-x", () => source.push("x"));
+  setTimeout(() => store.run("push-eof", () => source.push(null)), 5);
+});
+
+// destroy(): 'error' and 'close' follow the destroy() call.
+await step((done) => {
+  const source = new Readable({ read(): void {} });
+  source.on("error", () => log("destroy error"));
+  source.on("close", () => {
+    log("destroy close");
+    done();
+  });
+  store.run("destroy", () => source.destroy(new Error("stopped")));
+});
+
+// finished() and pipeline() callbacks follow their call, also when another
+// context ends or destroys the stream.
+await step((done) => {
+  const source = new Readable({ read(): void {} });
+  store.run("finished", () =>
+    finished(source, () => {
+      log("finished callback");
+      done();
+    }),
+  );
+  source.resume();
+  store.run("finished-eof", () => source.push(null));
+});
+await step((done) => {
+  const source = new Readable({ read(): void {} });
+  const sink = new Writable({
+    write(chunk: Buffer, encoding: BufferEncoding, cb: (error?: Error | null) => void): void {
+      cb();
+    },
+  });
+  store.run("pipeline", () =>
+    pipeline(source, sink, (err: NodeJS.ErrnoException | null) => {
+      log(`pipeline callback ${err ? err.message : "ok"}`);
+      done();
+    }),
+  );
+  setTimeout(() => store.run("pipeline-destroy", () => source.destroy(new Error("boom"))), 5);
+});
+
+// stream/promises pipeline: the stages follow the call; after the await
+// the caller's context is back.
+await store.run("promise-pipeline", () =>
+  pipelinePromise(
+    Readable.from(["a"]),
+    new Writable({
+      write(chunk: Buffer, encoding: BufferEncoding, cb: (error?: Error | null) => void): void {
+        log("promise pipeline write");
+        cb();
+      },
+    }),
+  ),
+);
+log("after the promise pipeline");
+
+// fs streams: the events follow the context that created the stream.
+const file = path.join(os.tmpdir(), `scriptc-7383-${process.pid}.txt`);
+await step((done) => {
+  const out = store.run("fs-write-create", () => fs.createWriteStream(file));
+  out.on("finish", () => log("fs finish"));
+  out.on("close", () => {
+    log("fs write close");
+    done();
+  });
+  store.run("fs-write", () => out.write("hello", () => log("fs write callback")));
+  store.run("fs-end", () => out.end());
+});
+await step((done) => {
+  const input = store.run("fs-read-create", () => fs.createReadStream(file));
+  store.run("fs-listen", () => {
+    input.on("open", () => log("fs open"));
+    input.on("data", () => log("fs data"));
+    input.on("end", () => log("fs end"));
+    input.on("close", () => {
+      log("fs read close");
+      done();
+    });
+  });
+});
+fs.unlinkSync(file);
+
+// zlib streams: the events follow the context that created the stream.
+await step((done) => {
+  const gzip = store.run("gzip-create", () => zlib.createGzip());
+  const gunzip = store.run("gunzip-create", () => zlib.createGunzip());
+  const parts: Buffer[] = [];
+  const sink = new Writable({
+    write(chunk: Buffer, encoding: BufferEncoding, cb: (error?: Error | null) => void): void {
+      log("gunzip output");
+      parts.push(chunk);
+      cb();
+    },
+  });
+  store.run("zlib-pipeline", () =>
+    pipeline(Readable.from(["hello ", "world"]), gzip, gunzip, sink, () => {
+      log(`zlib pipeline callback ${Buffer.concat(parts).toString()}`);
+      done();
+    }),
+  );
+});
+
+for (const line of lines) console.log(line);
@@ -0,0 +1,214 @@
+// AsyncLocalStorage contexts of stream events and callbacks, from
+// JavaScript. Stream steps that Node defers with process.nextTick run in
+// the context that scheduled them: resume() after a 'data' listener,
+// uncork() from a tick, a write after end(), a push() from a user tick.
+// Other events follow the call that emits them. stream.finished callbacks
+// run in the context of their call, with stream/promises' finished()
+// resolving in the context that awaits it. readline lines follow the
+// createInterface() call that resumes the input, and zlib streams run
+// their events and flush() callbacks in the context that created them.
+import { AsyncLocalStorage } from "node:async_hooks";
+import readline from "node:readline";
+import { Duplex, finished, PassThrough, Readable, Writable } from "node:stream";
+import { finished as finishedPromise } from "node:stream/promises";
+import zlib from "node:zlib";
+
+const store = new AsyncLocalStorage();
+const lines = [];
+const log = (what) => lines.push(`${what}: ${store.getStore() ?? "none"}`);
+const step = (fn) => new Promise((resolve) => fn(resolve));
+
+// pipe() and unpipe() events, and writes through a PassThrough.
+await step((done) => {
+  const through = new PassThrough();
+  const sink = new Writable({ write(chunk, encoding, cb) { log("pipe write"); cb(); } });
+  sink.on("pipe", () => log("pipe event"));
+  sink.on("unpipe", () => log("unpipe event"));
+  store.run("pipe", () => through.pipe(sink));
+  store.run("through-write", () => through.write("q"));
+  setTimeout(() => {
+    store.run("unpipe", () => through.unpipe(sink));
+    done();
+  }, 5);
+});
+
+// cork() and uncork() from a tick: the writes follow the tick's context.
+await step((done) => {
+  const sink = new Writable({ write(chunk, encoding, cb) { log("corked write"); cb(); } });
+  sink.on("finish", () => {
+    log("corked finish");
+    done();
+  });
+  store.run("cork", () => {
+    sink.cork();
+    sink.write("a");
+    sink.write("b");
+  });
+  store.run("uncork", () => process.nextTick(() => sink.uncork()));
+  setTimeout(() => store.run("corked-end", () => sink.end()), 5);
+});
+
+// A write after end(): its callback and 'error' follow the write() call.
+await step((done) => {
+  const sink = new Writable({ write(chunk, encoding, cb) { cb(); } });
+  sink.on("error", () => log("write after end error"));
+  sink.end();
+  store.run("late-write", () =>
+    sink.write("x", () => {
+      log("late write callback");
+      done();
+    }),
+  );
+});
+
+// A failing _write: the callback, 'error' and 'close' follow write().
+await step((done) => {
+  const sink = new Writable({ write(chunk, encoding, cb) { cb(new Error("bad")); } });
+  sink.on("error", () => log("failed write error"));
+  sink.on("close", () => {
+    log("failed write close");
+    done();
+  });
+  store.run("failed-write", () => sink.write("x", () => log("failed write callback")));
+});
+
+// A Duplex that pushes from _write: 'data' and 'end' follow the resume()
+// tick, which the 'data' listener scheduled with no store.
+await step((done) => {
+  const duplex = new Duplex({
+    read() {},
+    write(chunk, encoding, cb) {
+      log("duplex write");
+      this.push(chunk);
+      cb();
+    },
+  });
+  duplex.on("data", () => {
+    log("duplex data");
+    duplex.push(null);
+  });
+  duplex.on("end", () => {
+    log("duplex end");
+    done();
+  });
+  store.run("duplex-write", () => duplex.write("z"));
+});
+
+// A push() from a user tick, a 'data' listener registered in a store with
+// setEncoding(), and PassThrough's 'end' and 'close' after end().
+await step((done) => {
+  const source = new Readable({ read() {} });
+  source.on("data", () => log("tick data"));
+  source.on("end", () => {
+    log("tick end");
+    done();
+  });
+  store.run("user-tick", () =>
+    process.nextTick(() => {
+      log("user tick");
+      source.push("x");
+    }),
+  );
+  setTimeout(() => store.run("tick-eof", () => source.push(null)), 5);
+});
+await step((done) => {
+  const source = new Readable({ read() {} });
+  source.setEncoding("utf8");
+  store.run("data-listener", () => source.on("data", () => log("text data")));
+  source.on("end", () => {
+    log("text end");
+    done();
+  });
+  source.push("hi");
+  source.push(null);
+});
+await step((done) => {
+  const through = new PassThrough();
+  through.on("end", () => log("through end"));
+  through.on("close", () => {
+    log("through close");
+    done();
+  });
+  through.resume();
+  store.run("through-end", () => through.end("last"));
+});
+
+// finished() on a Writable, and stream/promises' finished().
+await step((done) => {
+  const sink = new Writable({ write(chunk, encoding, cb) { setTimeout(cb, 1); } });
+  store.run("finished-writable", () =>
+    finished(sink, () => {
+      log("finished callback");
+      done();
+    }),
+  );
+  sink.on("finish", () => log("writable finish"));
+  store.run("writable-end", () => {
+    sink.write("a");
+    sink.end();
+  });
+});
+await step((done) => {
+  const source = new Readable({ read() {} });
+  source.resume();
+  store.run("finished-promise", () => finishedPromise(source)).then(() => {
+    log("finished promise");
+    done();
+  });
+  setTimeout(() => store.run("promise-eof", () => source.push(null)), 5);
+});
+
+// Readable.from() over an async generator: the reads follow the resume()
+// tick, scheduled with no store.
+await step((done) => {
+  async function* letters() {
+    log("generator start");
+    yield "a";
+    log("generator end");
+  }
+  const source = store.run("from", () => Readable.from(letters()));
+  source.on("data", () => log("from data"));
+  source.on("end", () => {
+    log("from end");
+    done();
+  });
+});
+
+// readline over a Readable: createInterface() resumes the input.
+await step((done) => {
+  const input = Readable.from(["a\nb\n"]);
+  const rl = store.run("readline", () => readline.createInterface({ input }));
+  rl.on("line", (line) => log(`line ${line}`));
+  rl.on("close", () => {
+    log("readline close");
+    done();
+  });
+});
+
+// zlib streams: events and flush() callbacks follow the creation context.
+await step((done) => {
+  const gzip = store.run("gzip-create", () => zlib.createGzip());
+  store.run("gzip-listen", () => {
+    gzip.on("data", () => log("gzip data"));
+    gzip.on("end", () => {
+      log("gzip end");
+      done();
+    });
+  });
+  store.run("gzip-write", () => {
+    gzip.write("hello");
+    gzip.end();
+  });
+});
+await step((done) => {
+  const deflate = store.run("deflate-create", () => zlib.createDeflate());
+  store.run("deflate-flush", () => deflate.flush(() => log("deflate flush callback")));
+  deflate.on("data", () => {});
+  deflate.on("end", () => {
+    log("deflate end");
+    done();
+  });
+  store.run("deflate-end", () => deflate.end("x"));
+});
+
+for (const line of lines) console.log(line);
@@ -0,0 +1 @@
+{ "plain": 4500, "san": 9000 }
@@ -0,0 +1 @@
+{ "plain": 4500, "san": 9000 }