diff --git a/packages/runtime/src/scr_async.c b/packages/runtime/src/scr_async.c index 321256087..cf1be7a33 100644 --- a/packages/runtime/src/scr_async.c +++ b/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; } } diff --git a/packages/runtime/src/scr_runtime.h b/packages/runtime/src/scr_runtime.h index 1745bc672..93e308708 100644 --- a/packages/runtime/src/scr_runtime.h +++ b/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 diff --git a/packages/runtime/src/scr_stream.c b/packages/runtime/src/scr_stream.c index 53b19435e..0d2c5fdbd 100644 --- a/packages/runtime/src/scr_stream.c +++ b/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; diff --git a/tests/corpus/2606-stream-ticks-are-nextticks.ts b/tests/corpus/2606-stream-ticks-are-nextticks.ts new file mode 100644 index 000000000..156528e09 --- /dev/null +++ b/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"));