Stream ticks ride the user nextTick queue: FIFO with process.nextTick, Node's order

- Every deferred stream emission posts a raw marker on the user tick queue (scr_next_tick_raw) and each marker dispatches exactly one stream tick, so resume_/emitReadable_/endReadableNT interleave with user nextTicks in enqueue order — in Node they are the same queue
- Cures the setEncoding-with-buffered-bytes shape: a nextTick registered after on('data') now observes the flowed chunks, and stream emissions run before microtasks at each checkpoint instead of after
- The loop station stays as the drain for anything a marker never reached and as the uncaught-throw cleanup of undelivered ticks
- Corpus 2606-stream-ticks-are-nextticks pins the interleavings (buffered decode, split multi-byte char, tick/data/tick ordering, end-after-tick) differentially on both backends

packages/runtime/src/scr_async.c

@@ -591,7 +591,8 @@ ScrArr *scr_active_resources(void) {
  * left queued on the uncaught paths) never run — the teardown releases
  * them, like Node dropping the queue at exit. */
 typedef struct ScrNtick {
-  ScrClosure *cb; /* owned */
+  ScrClosure *cb;    /* owned; NULL for a raw C-hook entry */
+  void (*raw)(void); /* the hook when cb == NULL (stream tick markers) */
   struct ScrNtick *next;
 } ScrNtick;
 
@@ -612,11 +613,25 @@ void scr_next_tick(ScrClosure *cb /*moves*/) {
   scr_nt_tail = t;
 }
 
+/* A RAW C-hook tick: the stream unit enqueues one marker per deferred
+ * stream emission so those emissions interleave with user nextTicks in
+ * true FIFO order — in Node they ARE nextTicks (resume_, emitReadable_,
+ * endReadableNT, ...). The hook dispatches exactly one stream tick;
+ * teardown just drops markers (the stream queue owns its entries). */
+void scr_next_tick_raw(void (*fn)(void)) {
+  ScrNtick *t = calloc(1, sizeof *t);
+  if (!t) scr_oom();
+  t->raw = fn;
+  if (scr_nt_tail) scr_nt_tail->next = t;
+  else scr_nt_head = t;
+  scr_nt_tail = t;
+}
+
 void scr_nticks_teardown(void) {
   while (scr_nt_head != NULL) {
     ScrNtick *t = scr_nt_head;
     scr_nt_head = t->next;
-    scr_closure_release(t->cb);
+    if (t->cb) scr_closure_release(t->cb);
     free(t);
   }
   scr_nt_tail = NULL;
@@ -1677,9 +1692,14 @@ void scr_loop_run(void) {
         scr_nt_head = t->next;
         if (scr_nt_head == NULL) scr_nt_tail = NULL;
         ScrClosure *cb = t->cb;
+        void (*raw)(void) = t->raw;
         free(t);
-        ((void (*)(ScrClosure *))cb->fn)(cb);
-        scr_closure_release(cb);
+        if (cb) {
+          ((void (*)(ScrClosure *))cb->fn)(cb);
+          scr_closure_release(cb);
+        } else {
+          raw(); /* one stream tick, FIFO with the user ticks around it */
+        }
         if (scr_exc_pending()) return;
       }
     }

packages/runtime/src/scr_runtime.h

@@ -3491,6 +3491,11 @@ bool scr_immediate_has_ref(double handle);
  * teardown (they must never run yet must not leak). cb ownership moves
  * in. */
 void scr_next_tick(ScrClosure *cb);
+/* A raw C-hook entry on the SAME queue: the stream unit enqueues one
+ * marker per deferred stream emission, so stream ticks and user
+ * nextTicks run in true FIFO order (in Node they are the same queue).
+ * Teardown drops markers without running them. */
+void scr_next_tick_raw(void (*fn)(void));
 void scr_nticks_teardown(void);
 /* ── the events unit (scr_events.c — OPTIONAL, link-gated) ────────────
  * Process signal/exit events and the piped-stdin surface. The unit links

packages/runtime/src/scr_stream.c

@@ -13,11 +13,12 @@
  * nothing.
  *
  * EVENT TIMING. Node schedules most stream emissions on process.nextTick;
- * here those deferrals ride a FIFO tick queue drained at the top of every
- * loop turn (scr_loop_set_stream — the station before events/net/timers,
- * the closest to nextTick; ticks scheduled from user code on the main
- * stack run when the loop starts, i.e. after all synchronous code — the
- * nextTick shape). The implemented orderings follow lib/internal/streams:
+ * here each deferral enqueues on the stream tick FIFO AND posts a raw
+ * marker on the USER nextTick queue (scr_next_tick_raw), so stream
+ * emissions and user nextTicks run in true FIFO enqueue order — Node's,
+ * where they are the same queue. The scr_loop_set_stream station remains
+ * as the drain of anything a marker never reached (and the uncaught-throw
+ * cleanup). The implemented orderings follow lib/internal/streams:
  *   - on('data') starts flowing on a TICK (resume_): synchronous code
  *     after the registration runs before the first 'data'.
  *   - push() while flowing with an empty buffer emits 'data'
@@ -350,6 +351,8 @@ typedef struct ScrStreamTick {
 static ScrStreamTick *scr_st_head = NULL;
 static ScrStreamTick *scr_st_tail = NULL;
 
+static void scr_stream_dispatch_one(void);
+
 static void scr_st_tick(ScrStream *s, ScrStreamTickOp op, ScrError *err /*moves*/,
                         ScrClosure *cb /*moves*/) {
   ScrStreamTick *t = calloc(1, sizeof *t);
@@ -361,6 +364,13 @@ static void scr_st_tick(ScrStream *s, ScrStreamTickOp op, ScrError *err /*moves*
   if (scr_st_tail) scr_st_tail->next = t;
   else scr_st_head = t;
   scr_st_tail = t;
+  /* One marker per tick on the USER nextTick queue: stream emissions are
+   * process.nextTicks in Node (resume_, emitReadable_, endReadableNT,
+   * afterWrite, ...), so they must interleave with user nextTicks in
+   * enqueue order — a `push(); on('data'); process.nextTick(assert)`
+   * sequence sees its data before the assert runs. The station dispatch
+   * below stays as the drain of anything a marker never reached. */
+  scr_next_tick_raw(&scr_stream_dispatch_one);
 }
 
 static bool scr_stream_ticks_pending(void) { return scr_st_head != NULL; }
@@ -2909,6 +2919,25 @@ static void scr_stream_run_tick(ScrStreamTick *t) {
   }
 }
 
+static void scr_stream_dispatch(void);
+
+/* 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
+ * the RC audit clean. */
+static void scr_stream_dispatch_one(void) {
+  ScrStreamTick *t = scr_st_head;
+  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);
+  if (scr_exc_pending()) scr_stream_dispatch(); /* its exc branch drops the rest */
+}
+
 static void scr_stream_dispatch(void) {
   while (scr_st_head != NULL && !scr_exc_pending() && !scr_loop_has_ready()) {
     ScrStreamTick *t = scr_st_head;

tests/corpus/2606-stream-ticks-are-nextticks.ts

@@ -0,0 +1,49 @@
+// Stream emissions are process.nextTicks in Node (resume_,
+// emitReadable_, endReadableNT ride the same FIFO as user nextTicks), so
+// a user callback queued AFTER on('data') observes the flowed chunks —
+// the setEncoding-with-buffered-bytes shape — and 'readable'/'end'
+// interleave with nextTicks in enqueue order. Node is the oracle.
+import { Readable } from "node:stream";
+
+// setEncoding() with bytes already buffered: the re-decoded content flows
+// before a nextTick registered after the 'data' subscription.
+const r = new Readable({ read() {} });
+r.push(Buffer.from("a"));
+r.push(Buffer.from("b"));
+r.setEncoding("utf8");
+const chunks: string[] = [];
+r.on("data", (c: string) => chunks.push(c));
+process.nextTick(() => {
+  console.log("nextTick-sees", JSON.stringify(chunks));
+});
+
+// A chunked multi-byte character re-assembles across the decode.
+const r2 = new Readable({ read() {} });
+r2.push(Buffer.from([0xf0]));
+r2.push(Buffer.from([0x9f]));
+r2.setEncoding("utf8");
+r2.push(Buffer.from([0x8e]));
+r2.push(Buffer.from([0x89]));
+const chunks2: string[] = [];
+r2.on("data", (c: string) => chunks2.push(c));
+process.nextTick(() => {
+  console.log("split-char", JSON.stringify(chunks2));
+});
+
+// FIFO interleave: ticks queued before the flow kick run before it; ones
+// queued after run after the data delivery.
+const r3 = new Readable({ read() {} });
+r3.push(Buffer.from("x"));
+r3.setEncoding("utf8");
+process.nextTick(() => console.log("tick-before-subscribe"));
+r3.on("data", (c: string) => console.log("data", JSON.stringify(c)));
+process.nextTick(() => console.log("tick-after-subscribe"));
+
+// 'end' rides the tick queue too: push(null) then a nextTick queued after
+// the subscription still precedes 'end' (end waits for the drain + tick).
+const r4 = new Readable({ read() {} });
+r4.push(Buffer.from("z"));
+r4.push(null);
+r4.on("data", (c: Buffer) => console.log("r4-data", c.toString("utf8")));
+r4.on("end", () => console.log("r4-end"));
+process.nextTick(() => console.log("r4-tick"));