stream: ensure that stateful transforms preserve this

Signed-off-by: James M Snell <jasnell@gmail.com>
Assisted-by: Opencode
PR-URL: #65658
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>

Author: jasnell

lib/internal/streams/iter/pull.js

@@ -505,8 +505,9 @@ function* withFlushSync(source) {
   yield null;
 }
 
-function* applyStatefulSyncTransform(source, transform) {
-  const output = transform(withFlushSync(source));
+function* applyStatefulSyncTransform(source, transform, receiver) {
+  const output = FunctionPrototypeCall(
+    transform, receiver, withFlushSync(source));
   for (const item of output) {
     if (item === null) continue;
     const batch = [];
@@ -537,7 +538,8 @@ function* createSyncPipeline(source, transforms) {
         current = applyFusedStatelessSyncTransforms(current, statelessRun);
         statelessRun = [];
       }
-      current = applyStatefulSyncTransform(current, transform.transform);
+      current = applyStatefulSyncTransform(
+        current, transform.transform, transform);
     } else {
       ArrayPrototypePush(statelessRun, transform);
     }
@@ -648,8 +650,10 @@ async function* withFlushAsync(source) {
   yield null;
 }
 
-async function* applyStatefulAsyncTransform(source, transform, options) {
-  const output = transform(withFlushAsync(source), options);
+async function* applyStatefulAsyncTransform(
+  source, transform, receiver, options) {
+  const output = FunctionPrototypeCall(
+    transform, receiver, withFlushAsync(source), options);
   for await (const item of output) {
     if (item === null) continue;
     // Fast path: item is already a Uint8Array[] batch (e.g. compression transforms)
@@ -681,8 +685,10 @@ async function* applyStatefulAsyncTransform(source, transform, options) {
  * skips isUint8ArrayBatch validation (transform guarantees valid output).
  * @yields {Uint8Array[]}
  */
-async function* applyValidatedStatefulAsyncTransform(source, transform, options) {
-  const output = transform(source, options);
+async function* applyValidatedStatefulAsyncTransform(
+  source, transform, receiver, options) {
+  const output = FunctionPrototypeCall(
+    transform, receiver, source, options);
   for await (const batch of output) {
     if (batch.length > 0) {
       yield batch;
@@ -750,10 +756,10 @@ async function* createAsyncPipeline(source, transforms, signal) {
       const opts = { __proto__: null, signal: transformSignal };
       if (transform[kValidatedTransform]) {
         current = applyValidatedStatefulAsyncTransform(
-          current, transform.transform, opts);
+          current, transform.transform, transform, opts);
       } else {
         current = applyStatefulAsyncTransform(
-          current, transform.transform, opts);
+          current, transform.transform, transform, opts);
       }
     } else {
       ArrayPrototypePush(statelessRun, transform);

test/parallel/test-stream-iter-pull-async.js

@@ -49,6 +49,19 @@ async function testPullStatefulTransform() {
   assert.strictEqual(data, 'data-ASYNC-END');
 }
 
+async function testPullStatefulTransformReceiver() {
+  const descriptor = {};
+  descriptor.transform = common.mustCall(
+    async function*(source) {
+      assert.strictEqual(this, descriptor);
+      for await (const chunks of source) {
+        yield chunks;
+      }
+    });
+
+  assert.strictEqual(await text(pull(from('receiver'), descriptor)), 'receiver');
+}
+
 async function testPullWithAbortSignal() {
   async function* gen() {
     yield [new Uint8Array([1])];
@@ -511,6 +524,7 @@ async function testTransformOptionsNotShared() {
     testPullIdentity(),
     testPullStatelessTransform(),
     testPullStatefulTransform(),
+    testPullStatefulTransformReceiver(),
     testPullWithAbortSignal(),
     testPullNormalizesSourceAtCallTime(),
     testPullPreAbortOrdering(),

test/parallel/test-stream-iter-pull-sync.js

@@ -74,6 +74,20 @@ function testPullSyncStatefulTransform() {
   assert.strictEqual(data, 'data-END');
 }
 
+function testPullSyncStatefulTransformReceiver() {
+  const descriptor = {};
+  descriptor.transform = common.mustCall(
+    function*(source) {
+      assert.strictEqual(this, descriptor);
+      yield* source;
+    });
+
+  assert.strictEqual(
+    new TextDecoder().decode(bytesSync(pullSync(fromSync('receiver'), descriptor))),
+    'receiver',
+  );
+}
+
 function testPullSyncChainedTransforms() {
   const addExcl = (chunks) => {
     if (chunks === null) return null;
@@ -210,6 +224,7 @@ Promise.all([
   testPullSyncNormalizesSourceAtCallTime(),
   testPullSyncStatelessTransform(),
   testPullSyncStatefulTransform(),
+  testPullSyncStatefulTransformReceiver(),
   testPullSyncChainedTransforms(),
   testPullSyncSourceError(),
   testPullSyncEmptySource(),

test/parallel/test-stream-iter-transform-roundtrip.js

@@ -42,6 +42,18 @@ async function testGzipRoundTrip() {
   assert.strictEqual(result, input);
 }
 
+async function testValidatedTransformReceiver() {
+  const descriptor = compressGzip();
+  const transform = descriptor.transform;
+  descriptor.transform = common.mustCall(function(source, options) {
+    assert.strictEqual(this, descriptor);
+    return Reflect.apply(transform, this, [source, options]);
+  });
+
+  const result = await bytes(pull(from('receiver'), descriptor));
+  assert.ok(result.byteLength > 0);
+}
+
 async function testGzipLargeData() {
   // 100KB of repeated text - exercises multi-chunk path
   const input = 'gzip large data test. '.repeat(5000);
@@ -250,6 +262,7 @@ async function testGzipWithLevel() {
 (async () => {
   // Gzip
   await testGzipRoundTrip();
+  await testValidatedTransformReceiver();
   await testGzipLargeData();
   await testGzipActuallyCompresses();