fix(streams,transform): Node tick parity — tee/pipe cadence + awaits trapped in the object-spread IIFE (Next.js Flight byte-parity)

crates/perry-stdlib/src/streams.rs

@@ -697,9 +697,11 @@ pub(super) unsafe fn maybe_pull(stream_id: usize) {
     // `pull_cb` of its own; the source's enqueue fans the chunk out to both
     // branches. `force` so a highWaterMark-0 flight producer actually pulls.
     if let Some(source) = tee_source_of(stream_id) {
-        // Tick parity: the pull travels a microtask cycle (buffered chunks +
-        // close discovery included); a live producer is driven from that job.
-        tee::tee_schedule_pull(source);
+        // Tick parity: a demand-initiated pull travels TWO microtask cycles
+        // (Node's sourceReader.read() resolution + .then(fanout) reaction);
+        // buffered chunks + close discovery included. A live producer is
+        // driven from the pull job.
+        tee::tee_schedule_pull_demand(source);
         return;
     }
     maybe_pull_inner(stream_id, false);

crates/perry-stdlib/src/streams/pipe.rs

@@ -395,7 +395,25 @@ unsafe fn pipe_write_then_continue(
     // and let a tee sibling's reader outrun the pipe — Next.js cold-start
     // head reorder). Chain on the write promise only while it is pending.
     if perry_runtime::promise::js_promise_state(write_promise) == 1 {
-        schedule_pipe_step(readable_id, writable_id, promise, locks, prevent_close);
+        // Tick parity (streamsuite teepipe/teepipe2 wcc/waa): when the
+        // readable's queue is EMPTY, Node's pump has its next read parked
+        // within the write-completion reaction, so the next delivery (a tee
+        // fan-out) resolves it directly and the write lands one tick after
+        // the sibling's read. Deferring the park through a queued step made
+        // that write a tick late. Buffered chunks keep the queued step —
+        // popping synchronously would bunch writes and break the 1/tick
+        // write cadence.
+        let park_now = {
+            let g = READABLE_STREAMS.lock().unwrap();
+            g.get(&readable_id)
+                .map(|s| s.chunks.is_empty() && s.state == ReadableState::Readable)
+                .unwrap_or(false)
+        };
+        if park_now {
+            wait_for_next_read(readable_id, writable_id, promise, locks, prevent_close);
+        } else {
+            schedule_pipe_step(readable_id, writable_id, promise, locks, prevent_close);
+        }
         return;
     }
     let fulfilled = pipe_closure(

crates/perry-stdlib/src/streams/tee.rs

@@ -25,6 +25,13 @@ lazy_static::lazy_static! {
     static ref TEE_BRANCH_SOURCE: Mutex<HashMap<usize, usize>> = Mutex::new(HashMap::new());
     /// Tee sources with a pull job already queued (one job per source at a time).
     static ref TEE_PULLING: Mutex<HashSet<usize>> = Mutex::new(HashSet::new());
+    /// Tee sources whose pull pipeline has delivered at least once. The
+    /// COLD-START demand pull pays Node's full two-hop pipeline latency
+    /// (`sourceReader.read()` resolution + `.then(fanout)` reaction); once
+    /// the pipeline is warm, mid-stream re-parks resolve on the calibrated
+    /// one-hop cadence (Node's pipelined stages overlap, so only the first
+    /// delivery exposes the latency).
+    static ref TEE_STARTED: Mutex<HashSet<usize>> = Mutex::new(HashSet::new());
 }
 
 /// Sentinel `reader_handle` stamped on a tee'd source so it can't be read
@@ -155,6 +162,7 @@ fn tee_unlink(source: usize, a: usize, b: usize) {
         bs.remove(&a);
         bs.remove(&b);
     }
+    TEE_STARTED.lock().unwrap().remove(&source);
     // #6602: the unlinked source is terminal (close flow: Closed and drained
     // by pull discovery; error flow: stamped Errored above) — retire its id.
     // Its `TEE_LOCK_SENTINEL` reader_handle matches no READERS entry.
@@ -195,6 +203,12 @@ pub(super) fn evict_ids(batch: &[usize]) {
             g.remove(id);
         }
     }
+    {
+        let mut g = TEE_STARTED.lock().unwrap();
+        for id in batch {
+            g.remove(id);
+        }
+    }
 }
 
 /// Enqueue on a tee'd SOURCE: the chunk stays in the source queue (spec size
@@ -262,6 +276,44 @@ pub(super) unsafe fn tee_schedule_pull(source: usize) {
     perry_runtime::builtins::js_queue_microtask(job as i64);
 }
 
+/// Demand-initiated pull entry — a branch read parked against an empty branch
+/// queue (`maybe_pull` routing). On a COLD pipeline Node pays TWO microtask
+/// hops before that read resolves: the `sourceReader.read()` promise
+/// resolution plus the `.then(fanout)` reaction (streamsuite first-delivery
+/// cadence: node's first chunk lands after t2, Perry's landed after t1 — the
+/// one-hop-short residual behind the Next.js RSC Flight row-reorder). Once
+/// the pipeline has delivered, Node's stages overlap and a mid-stream re-park
+/// resolves on the one-hop cadence — so only the cold-start entry pays the
+/// extra hop. CHAINED cycles and producer-side arrivals keep their existing
+/// calibrated cadence throughout.
+pub(super) unsafe fn tee_schedule_pull_demand(source: usize) {
+    if TEE_STARTED.lock().unwrap().contains(&source) {
+        tee_schedule_pull(source);
+        return;
+    }
+    if !TEE_PULLING.lock().unwrap().insert(source) {
+        return;
+    }
+    let f = tee_demand_hop as *const u8;
+    perry_runtime::closure::js_register_closure_arity(f, 0);
+    let job = perry_runtime::closure::js_closure_alloc(f, 1);
+    perry_runtime::closure::js_closure_set_capture_ptr(job, 0, source as i64);
+    perry_runtime::builtins::js_queue_microtask(job as i64);
+}
+
+/// The extra demand-entry hop: hand off to the real pull job one microtask
+/// later. `TEE_PULLING` stays held across the hop (single-threaded microtask
+/// dispatch — the remove+insert below has no interleaving window), so
+/// coalescing against enqueue/close reroutes keeps working.
+extern "C" fn tee_demand_hop(closure: *const ClosureHeader) -> f64 {
+    unsafe {
+        let source = perry_runtime::closure::js_closure_get_capture_ptr(closure, 0) as usize;
+        TEE_PULLING.lock().unwrap().remove(&source);
+        tee_schedule_pull(source);
+    }
+    f64::from_bits(0x7FFC_0000_0000_0001) // TAG_UNDEFINED
+}
+
 /// The extra tick a byte-stream tee's CHAINED pull pays before the next
 /// cycle (see the chain decision in `tee_pull_microtask`).
 extern "C" fn tee_byte_chain_hop(closure: *const ClosureHeader) -> f64 {
@@ -312,6 +364,9 @@ extern "C" fn tee_pull_microtask(closure: *const ClosureHeader) -> f64 {
         };
         match chunk {
             Some(bits) => {
+                // Pipeline is warm from the first delivery on — later
+                // demand entries skip the cold-start hop.
+                TEE_STARTED.lock().unwrap().insert(source);
                 // Spec order: branch-a sees the chunk before branch-b,
                 // regardless of which branch's read triggered the pull.
                 // Byte tees clone the chunk for branch-b (CloneAsUint8Array)
@@ -336,13 +391,21 @@ extern "C" fn tee_pull_microtask(closure: *const ClosureHeader) -> f64 {
                     tee_branch_demand(a, b) || source_backlog
                 };
                 if more {
-                    if is_byte {
+                    let close_only = {
+                        let g = READABLE_STREAMS.lock().unwrap();
+                        g.get(&source)
+                            .map(|s| s.chunks.is_empty() && s.state == ReadableState::Closed)
+                            .unwrap_or(true)
+                    };
+                    if is_byte && !close_only {
                         // Byte-stream tee (bytetee.js): Node's CHAINED pulls
                         // pay one extra tick per cycle (the byte path's
-                        // clone/view hop); the first demand pull does not.
-                        // Each extra hop cedes a task-generation to racing
-                        // promise cascades (the Next.js module-require chain
-                        // must win that race).
+                        // clone/view hop); the first demand pull does not,
+                        // and neither does the CLOSE-discovery cycle after
+                        // the last chunk (node's done lands one tick after
+                        // the final delivery pair). Each extra hop cedes a
+                        // task-generation to racing promise cascades (the
+                        // Next.js module-require chain must win that race).
                         let f = tee_byte_chain_hop as *const u8;
                         perry_runtime::closure::js_register_closure_arity(f, 0);
                         let job = perry_runtime::closure::js_closure_alloc(f, 1);

crates/perry-transform/src/async_to_generator.rs

@@ -764,6 +764,13 @@ fn hoist_awaits_in_expr_full(expr: &mut Expr, next_id: &mut LocalId, hoisted: &m
         hoist_awaits_in_expr_full(expr, next_id, hoisted);
         return;
     }
+    // `{ a: f(), v: await p, ...src }` — the object-with-spread lowering
+    // wraps the build in a synthetic IIFE, trapping the await inside a
+    // NON-async closure the pre-pass would otherwise skip. Inline it so
+    // the awaits reach the enclosing async context (see the fn comment).
+    if inline_obj_iife_with_await(expr, next_id, hoisted) {
+        return;
+    }
     // Recurse into children first (innermost-first hoisting).
     perry_hir::walker::walk_expr_children_mut(expr, &mut |child| {
         hoist_awaits_in_expr_full(child, next_id, hoisted);
@@ -906,13 +913,94 @@ fn hoist_awaits_avoiding_top_level(
         hoist_awaits_avoiding_top_level(expr, next_id, hoisted);
         return;
     }
+    // Top-level awaited obj-IIFE, e.g. `return { v: await p, ...src };` —
+    // see the matching arm in `hoist_awaits_in_expr_full`.
+    if inline_obj_iife_with_await(expr, next_id, hoisted) {
+        return;
+    }
     // Outer is NOT an await. Children may contain awaits which ARE
     // nested — fully hoist them.
     perry_hir::walker::walk_expr_children_mut(expr, &mut |child| {
         hoist_awaits_in_expr_full(child, next_id, hoisted);
     });
 }
 
+/// `{ a: f(), v: await p, ...src }` lowers (lower/expr_object.rs) to a
+/// synthetic single-param IIFE (`__perry_obj_iife`) that builds the object —
+/// which traps a property-value `await` inside a NON-async closure. The
+/// pre-pass (correctly) never descends into closures, so that await stayed
+/// raw and codegen's fallback BLOCKED the frame on the re-entrant microtask
+/// pump — draining unrelated tasks mid-expression (the Next.js Flight
+/// row-reorder: next-intl's provider wrapper has exactly this shape). The
+/// IIFE runs exactly once at its own sequence point and HIR closure bodies
+/// reference enclosing locals by their original ids, so inline it: bind the
+/// seed object to the param's id, replay the body statements in evaluation
+/// order (hoisting their awaits into the enclosing async context), and
+/// replace the call with a read of the object local. Non-await obj-IIFEs
+/// keep the closure form untouched.
+fn inline_obj_iife_with_await(
+    expr: &mut Expr,
+    next_id: &mut LocalId,
+    hoisted: &mut Vec<Stmt>,
+) -> bool {
+    {
+        let Expr::Call { callee, args, .. } = &*expr else {
+            return false;
+        };
+        let Expr::Closure {
+            params,
+            body,
+            is_async,
+            is_generator,
+            ..
+        } = callee.as_ref()
+        else {
+            return false;
+        };
+        if *is_async
+            || *is_generator
+            || params.len() != 1
+            || args.len() != 1
+            || params[0].name != "__perry_obj_iife"
+            || !body_contains_await(body)
+        {
+            return false;
+        }
+    }
+    let Expr::Call { callee, args, .. } = expr else {
+        unreachable!()
+    };
+    let Expr::Closure { params, body, .. } = callee.as_mut() else {
+        unreachable!()
+    };
+    let obj_id = params[0].id;
+    let seed = args.remove(0);
+    hoisted.push(Stmt::Let {
+        id: obj_id,
+        name: "__perry_obj_iife".to_string(),
+        ty: Type::Any,
+        mutable: false,
+        init: Some(seed),
+    });
+    let stmts = std::mem::take(body);
+    for mut stmt in stmts {
+        match &mut stmt {
+            // The synthesized body ends with `return __perry_obj_iife` —
+            // any return terminates the replay.
+            Stmt::Return(_) => break,
+            Stmt::Expr(e) => {
+                hoist_awaits_in_expr_full(e, next_id, hoisted);
+                hoisted.push(stmt);
+            }
+            // Lowered obj-IIFE bodies are flat Expr/Return statements today;
+            // forward anything else untouched.
+            _ => hoisted.push(stmt),
+        }
+    }
+    *expr = Expr::LocalGet(obj_id);
+    true
+}
+
 /// Lift a sequence (comma) expression's non-final operands into statements
 /// pushed onto `hoisted`, leaving `expr` as the final operand (issue #5925).
 ///
@@ -957,7 +1045,14 @@ fn expr_contains_await(expr: &Expr) -> bool {
     if matches!(expr, Expr::Await(_)) {
         return true;
     }
-    if matches!(expr, Expr::Closure { .. }) {
+    if let Expr::Closure { params, body, .. } = expr {
+        // The synthetic object-with-spread IIFE executes inline at its own
+        // sequence point — an await inside it IS an await of the enclosing
+        // function (`inline_obj_iife_with_await` surfaces it during the
+        // hoist). Every other closure owns its awaits.
+        if params.len() == 1 && params[0].name == "__perry_obj_iife" {
+            return body_contains_await(body);
+        }
         return false;
     }
     let mut found = false;
@@ -1684,7 +1779,17 @@ fn expr_contains_await_shallow(expr: &Expr, found: &mut bool) -> bool {
         *found = true;
         return true;
     }
-    if matches!(expr, Expr::Closure { .. }) {
+    if let Expr::Closure { params, body, .. } = expr {
+        // See `expr_contains_await`: the synthetic object-with-spread IIFE's
+        // awaits belong to the enclosing function — without this, an async
+        // closure whose ONLY awaits sit in an obj-IIFE never enters the
+        // collect work set, stays un-rewritten, and its raw awaits BLOCK the
+        // frame on the busy-wait pump (next-intl's provider wrapper in the
+        // Next.js Flight row-reorder).
+        if params.len() == 1 && params[0].name == "__perry_obj_iife" && body_contains_await(body) {
+            *found = true;
+            return true;
+        }
         return false;
     }
     perry_hir::walker::walk_expr_children(expr, &mut |e| {

test-files/test_gap_async_objspread_await_suspension.ts

@@ -0,0 +1,78 @@
+// An async function whose awaits sit inside an object literal WITH a spread
+// must still suspend at the await and return a pending promise immediately —
+// like V8 — instead of blocking the frame on the microtask pump.
+//
+// `{ v: await p, ...src }` lowers to a synthetic IIFE (`__perry_obj_iife`);
+// two transform gaps left such functions un-CPS-rewritten: the hoist pre-pass
+// never descended into the IIFE, and — for NESTED async closures (the
+// turbopack factory shape) — the collect scan's closure-stop meant a function
+// whose ONLY awaits sat in the IIFE never entered the work set at all. The
+// resulting mid-frame drain reordered every promise race against the object
+// build (the Next.js Flight row swap: next-intl's provider wrapper is exactly
+// this shape).
+//
+// Detector: a properly-suspending call lets "after-call" log BEFORE the
+// resolver tick; a busy-waiting call pumps the queue inside itself, flipping
+// the order.
+
+function scenario(name: string, makeCall: (p: Promise<string>) => any) {
+  return new Promise<void>((done) => {
+    let r: (v: string) => void;
+    const pending = new Promise<string>((res) => (r = res));
+    const log: string[] = [];
+    Promise.resolve().then(() => {
+      log.push("tick1");
+      r!("V");
+    });
+    const ret = makeCall(pending);
+    log.push("after-call");
+    Promise.resolve(ret).then((v) => {
+      log.push("resolved " + JSON.stringify(v));
+      console.log(name + ": " + log.join(" | "));
+      done();
+    });
+  });
+}
+
+async function f1(p: Promise<string>) {
+  return await p;
+}
+
+function jsx(t: string, props: Record<string, unknown>) {
+  return { t, props };
+}
+
+const src = { x: 1 };
+
+// Top-level async fn, spread after/before the await.
+async function spreadAfter(p: Promise<string>) {
+  return { v: await p, ...src };
+}
+async function spreadBefore(p: Promise<string>) {
+  return { ...src, v: await p };
+}
+
+// The production shape: NESTED async fn (factory arrow), conditional awaits
+// in an object argument with a rest-spread that shadows the fn name.
+const reg: { m?: (o: any) => Promise<any> } = {};
+((a: typeof reg) => {
+  async function m({ formats, locale, p, ...m }: any) {
+    return jsx("g", {
+      formats: void 0 === formats ? await f1(p) : formats,
+      locale: locale ?? (await f1(p)),
+      ...m,
+    });
+  }
+  a.m = m;
+})(reg);
+
+async function main() {
+  await scenario("top-level-spread-after", (p) => spreadAfter(p));
+  await scenario("top-level-spread-before", (p) => spreadBefore(p));
+  await scenario("nested-factory-conditional-awaits", (p) =>
+    reg.m!({ p, extra: 2 }),
+  );
+  console.log("DONE");
+}
+
+main();