src: cache readable controller pull-reaction functions

The per-pull hot path allocated two V8 Functions on every CallPullIfNeeded
(the fulfil/reject reactions for the user pull() promise), via Function::New
per chunk. Cache them per-controller (Data == the controller wrapper),
created on first pull and reused thereafter; reset in ClearAlgorithms to
break the controller<->wrapper cycle on terminal states.

Effect (vs JS baseline, 30 runs): readable-read byob -42% -> -5%,
normal -77% -> -49%, async-iterator -81% -> -64%, pipe-to -81% -> -75%.
WPT streams (1404) + 39 webstream parallel tests green.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: mcollina

src/streams/readable_stream.cc

@@ -184,7 +184,8 @@ void ReactRejected(const FunctionCallbackInfo<Value>& args) {
 }
 
 // Attaches fulfill/reject reactions that re-enter C++. The reaction functions
-// carry the controller wrapper as their Data.
+// carry the controller wrapper as their Data. Used by cold paths (start);
+// the per-pull hot path uses ThenReactCached below.
 void ThenReact(Environment* env,
                Local<Promise> promise,
                Local<Object> controller_obj,
@@ -200,12 +201,38 @@ void ThenReact(Environment* env,
   USE(promise->Then(context, ff, rj));
 }
 
+// Like ThenReact, but reuses the controller's cached reaction functions
+// (creating them on first use), so the per-pull path allocates no V8 Function.
+void ThenReactCached(Environment* env,
+                     Local<Promise> promise,
+                     Local<Object> controller_obj,
+                     v8::FunctionCallback on_fulfilled,
+                     v8::Global<Function>* ff_slot,
+                     v8::Global<Function>* rj_slot) {
+  Isolate* isolate = env->isolate();
+  Local<Context> context = env->context();
+  Local<Function> ff = ff_slot->Get(isolate);
+  if (ff.IsEmpty()) {
+    if (!Function::New(context, on_fulfilled, controller_obj).ToLocal(&ff))
+      return;
+    ff_slot->Reset(isolate, ff);
+  }
+  Local<Function> rj = rj_slot->Get(isolate);
+  if (rj.IsEmpty()) {
+    if (!Function::New(context, ReactRejected, controller_obj).ToLocal(&rj))
+      return;
+    rj_slot->Reset(isolate, rj);
+  }
+  USE(promise->Then(context, ff, rj));
+}
+
 // A function that returns undefined; used both to map a promise's fulfilment
 // value to undefined and as a no-op rejection handler.
 void Noop(const FunctionCallbackInfo<Value>& args) {}
 
 // Marks a promise as handled to avoid spurious unhandled-rejection warnings on
 // internal promises, mirroring `PromisePrototypeThen(p, undefined, () => {})`.
+// Cold path (close/error/release), so the per-call Function is acceptable.
 void MarkHandled(Environment* env, Local<Promise> promise) {
   Local<Context> context = env->context();
   Local<Function> noop;
@@ -478,13 +505,19 @@ void ReadableStreamDefaultController::CallPullIfNeeded() {
     return;
   // The pull algorithm is wrapped as an async function, so it returns a Promise.
   if (!result->IsPromise()) return;
-  ThenReact(env, result.As<Promise>(), controller_obj, ReactPullFulfilled);
+  ThenReactCached(env, result.As<Promise>(), controller_obj, ReactPullFulfilled,
+                  &on_pull_fulfilled_, &on_rejected_);
 }
 
 void ReadableStreamDefaultController::ClearAlgorithms() {
   pull_algorithm_.Reset();
   cancel_algorithm_.Reset();
   size_algorithm_.Reset();
+  // Break the controller<->wrapper cycle created by the cached reaction
+  // functions (their Data is this controller's wrapper). No further pulls occur
+  // after the algorithms are cleared, so they will not be recreated.
+  on_pull_fulfilled_.Reset();
+  on_rejected_.Reset();
 }
 
 void ReadableStreamDefaultController::CloseInternal() {
@@ -1450,12 +1483,16 @@ void ReadableByteStreamController::CallPullIfNeeded() {
   if (!pull->Call(context, Undefined(isolate), 1, argv).ToLocal(&result))
     return;
   if (!result->IsPromise()) return;
-  ThenReact(env, result.As<Promise>(), controller_obj, ReactPullFulfilled);
+  ThenReactCached(env, result.As<Promise>(), controller_obj, ReactPullFulfilled,
+                  &on_pull_fulfilled_, &on_rejected_);
 }
 
 void ReadableByteStreamController::ClearAlgorithms() {
   pull_algorithm_.Reset();
   cancel_algorithm_.Reset();
+  // Break the controller<->wrapper cycle from the cached reaction functions.
+  on_pull_fulfilled_.Reset();
+  on_rejected_.Reset();
 }
 
 void ReadableByteStreamController::InvalidateBYOBRequest() {

src/streams/readable_stream.h

@@ -126,6 +126,12 @@ class ReadableStreamDefaultController final : public StreamBaseObject {
   v8::Global<v8::Function> pull_algorithm_;
   v8::Global<v8::Function> cancel_algorithm_;
   v8::Global<v8::Function> size_algorithm_;
+
+  // Cached promise-reaction functions for the pull hot path (Data == this
+  // controller's wrapper). Created once on first pull and reused for every
+  // subsequent pull, so the per-pull path allocates no V8 Function.
+  v8::Global<v8::Function> on_pull_fulfilled_;
+  v8::Global<v8::Function> on_rejected_;
 };
 
 // ReadableStreamDefaultReader — holds the queue of pending read requests as
@@ -350,6 +356,11 @@ class ReadableByteStreamController final : public StreamBaseObject {
 
   v8::Global<v8::Function> pull_algorithm_;
   v8::Global<v8::Function> cancel_algorithm_;
+
+  // Cached promise-reaction functions for the pull hot path (Data == this
+  // controller's wrapper). Created once on first pull and reused thereafter.
+  v8::Global<v8::Function> on_pull_fulfilled_;
+  v8::Global<v8::Function> on_rejected_;
 };
 
 // ReadableStreamBYOBRequest — a thin view onto the controller's first pending