Skip to content

perf: optimize stream pipeline by eliminating events-intercept #9221

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
alkatrivedi wants to merge 3 commits into
base: main
Choose a base branch
from
+13 −22
Open

perf: optimize stream pipeline by eliminating events-intercept #9221

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5f1c3e7
perf: optimize stream pipeline by eliminating events-intercept
alkatrivedi Aug 27, 2026
6f9caca
gemini review comments
alkatrivedi Aug 28, 2026
a4c89d2
refactor
alkatrivedi Aug 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 0 additions & 3 deletions handwritten/spanner/package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,13 +73,11 @@
"big.js": "^7.0.0",
"checkpoint-stream": "^0.1.2",
"duplexify": "^4.1.3",
"events-intercept": "^2.0.0",
"extend": "^3.0.2",
"google-auth-library": "^10.0.0-rc.1",
"google-gax": "5.0.6",
"grpc-gcp": "^1.0.1",
"lodash.snakecase": "^4.1.1",
"merge-stream": "^2.0.0",
"p-queue": "^6.0.2",
"protobufjs": "^7.4.0",
"retry-request": "^8.0.0",
Expand All @@ -98,7 +96,6 @@
"@types/extend": "^3.0.4",
"@types/is": "^0.0.25",
"@types/lodash.snakecase": "^4.1.9",
"@types/merge-stream": "^2.0.0",
"@types/mocha": "^10.0.10",
"@types/mv": "^2.1.4",
"@types/ncp": "^2.0.8",
Expand Down
32 changes: 13 additions & 19 deletions handwritten/spanner/src/partial-result-stream.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,18 +16,15 @@

import {GrpcService} from './common-grpc/service';
import * as checkpointStream from 'checkpoint-stream';

@olavloite olavloite Aug 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This actually also imports events-intercept as a transitive dependency. So the removal of the import below does not actually remove it entirely from this file. And checkpointStream also monkey-patches the emit method. So while this PR gets rid of some of the monkey-patching of the query stream, it does not get rid of all of it.

Test:

it('should not have events-intercept monkey-patching the stream pipeline', () => {
  const stream = checkpointStream.obj();
  // @ts-ignore
  const hasIntercept = typeof stream.intercept === 'function';
  assert.strictEqual(
    hasIntercept,
    false,
    'events-intercept is still monkey-patching stream pipeline via checkpoint-stream',
  );
});

We could instead implement our own specific 'checkpointStream' without monkey-patching and remove the entire dependency on checkpointStream here:

class CheckpointStream extends Transform {
  private queue: google.spanner.v1.PartialResultSet[] = [];
  private maxQueued: number;
  private isCheckpointFn: (chunk: google.spanner.v1.PartialResultSet) => boolean;

  constructor(options: {
    maxQueued?: number;
    isCheckpointFn: (chunk: google.spanner.v1.PartialResultSet) => boolean;
  }) {
    super({objectMode: true});
    this.maxQueued = options.maxQueued ?? 10;
    this.isCheckpointFn = options.isCheckpointFn;
  }

  _transform(
    chunk: google.spanner.v1.PartialResultSet,
    enc: string,
    callback: () => void,
  ): void {
    this.queue.push(chunk);
    const isCheckpoint = this.isCheckpointFn(chunk);
    if (isCheckpoint) {
      this.emit('checkpoint', chunk);
      this._flushQueue();
    } else if (this.queue.length > this.maxQueued) {
      this._flushQueue();
    }
    callback();
  }

  private _flushQueue(): void {
    while (this.queue.length > 0) {
      this.push(this.queue.shift());
    }
  }

  reset(): void {
    this.queue = [];
  }

  _flush(callback: () => void): void {
    this._flushQueue();
    callback();
  }
}

import * as eventsIntercept from 'events-intercept';
import mergeStream = require('merge-stream');
import {common as p} from 'protobufjs';
import {Readable, Transform} from 'stream';
import {PassThrough, Readable, Transform} from 'stream';
import * as streamEvents from 'stream-events';
import {grpc, CallOptions} from 'google-gax';
import {DeadlineError, isRetryableInternalError} from './transaction-runner';

import {codec, JSONOptions, Json, Field, Value} from './codec';
import {protos} from '@google-cloud/spanner-api';
import google = protos.google;
import * as stream from 'stream';
import {isDefined, isEmpty, isString} from './helper';

const originalDecode = codec.decode;
Expand Down Expand Up @@ -253,7 +250,7 @@
this._options.columnsMetadata,
name,
)
? (this._options.columnsMetadata as any)[name]

Check warning on line 253 in handwritten/spanner/src/partial-result-stream.ts

View workflow job for this annotation

GitHub Actions / lint 

Unexpected any. Specify a different type
: undefined;
if (codec.decode !== originalDecode) {
return val =>
Expand Down Expand Up @@ -578,17 +575,18 @@
const maxQueued = 10;
let lastResumeToken: ResumeToken;
let lastRequestStream: Readable;
let errorListener: (err: grpc.ServiceError) => void;
const startTime = Date.now();
const timeout = options?.gaxOptions?.timeout ?? Infinity;

// mergeStream allows multiple streams to be connected into one. This is good;
// requestsStream allows multiple streams to be connected into one. This is good;
// if we need to retry a request and pipe more data to the user's stream.
// We also add an additional stream that can be used to flush any remaining
// items in the checkpoint stream that have been received, and that did not
// contain a resume token.
const requestsStream = mergeStream();
const flushStream = new stream.PassThrough({objectMode: true});
requestsStream.add(flushStream);
const requestsStream = new PassThrough({objectMode: true});
const flushStream = new PassThrough({objectMode: true});
flushStream.pipe(requestsStream);
const partialRSStream = new PartialResultStream(options);
const userStream = streamEvents(partialRSStream);
// We keep track of the number of PartialResultSets that did not include a
Expand Down Expand Up @@ -617,7 +615,6 @@
// then push `null` to end the stream.
flushStream.push({resumeToken: '_'});
flushStream.push(null);
requestsStream.end();
});
};
const makeRequest = (): void => {
Expand All @@ -626,7 +623,11 @@
}
lastRequestStream = requestFn(lastResumeToken);
lastRequestStream.on('end', endListener);
requestsStream.add(lastRequestStream);
errorListener = (err: grpc.ServiceError) => {
setImmediate(() => retry(err));

@olavloite olavloite Aug 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Deferring the retry here with setImmediate(..) means that other events could be handled before the retry. If for example lastRequestStream emits 'end' before the retry, the retry will fail. It could also cause multiple errors to trigger multiple retries.

Verification test cases:

it('should successfully retry when the failed stream emits an error followed by end', done => {
  const fakeCheckpointStream = through.obj();
  // eslint-disable-next-line @typescript-eslint/no-explicit-any
  (fakeCheckpointStream as any).reset = () => {};
  sandbox.stub(checkpointStream, 'obj').returns(fakeCheckpointStream);

  const firstStream = through.obj();
  const secondStream = through.obj();

  const requestFnStub = sandbox.stub();

  // First request fails with UNAVAILABLE and immediately ends
  requestFnStub.onCall(0).callsFake(() => {
    setImmediate(() => {
      firstStream.emit('error', {
        code: grpc.status.UNAVAILABLE,
        message: 'Unavailable',
      } as grpc.ServiceError);
      firstStream.end();
    });
    return firstStream;
  });

  // Retried request succeeds and delivers data
  requestFnStub.onCall(1).callsFake(() => {
    setImmediate(() => {
      secondStream.push(RESULT_WITH_TOKEN);
      fakeCheckpointStream.emit('checkpoint', RESULT_WITH_TOKEN);
      secondStream.end();
    });
    return secondStream;
  });

  const receivedRows: Row[] = [];
  partialResultStream(requestFnStub)
    .on('data', row => receivedRows.push(row))
    .on('error', done)
    .on('end', () => {
      try {
        assert.strictEqual(requestFnStub.callCount, 2, 'Should have retried once');
        assert.strictEqual(receivedRows.length, 1, 'Should receive data from retried stream');
        done();
      } catch (e) {
        done(e);
      }
    });
});

it('should only spawn a single retry when multiple errors are emitted in rapid succession', done => {
  const fakeCheckpointStream = through.obj();
  // eslint-disable-next-line @typescript-eslint/no-explicit-any
  (fakeCheckpointStream as any).reset = () => {};
  sandbox.stub(checkpointStream, 'obj').returns(fakeCheckpointStream);

  const firstStream = through.obj();
  const secondStream = through.obj();

  const requestFnStub = sandbox.stub();

  // First request emits two error events synchronously
  requestFnStub.onCall(0).callsFake(() => {
    setImmediate(() => {
      const err = {
        code: grpc.status.UNAVAILABLE,
        message: 'Unavailable',
      } as grpc.ServiceError;
      firstStream.emit('error', err);
      firstStream.emit('error', err);
    });
    return firstStream;
  });

  // Second request succeeds
  requestFnStub.onCall(1).callsFake(() => {
    setImmediate(() => {
      secondStream.push(RESULT_WITH_TOKEN);
      fakeCheckpointStream.emit('checkpoint', RESULT_WITH_TOKEN);
      secondStream.end();
    });
    return secondStream;
  });

  partialResultStream(requestFnStub)
    .on('error', done)
    .pipe(
      concat(rows => {
        try {
          // Exactly 1 initial request + 1 retry request = 2 calls total
          assert.strictEqual(requestFnStub.callCount, 2, 'Should only trigger one retry request');
          assert.strictEqual(rows.length, 1);
          done();
        } catch (e) {
          done(e);
        }
      }),
    );
});

Suggested fix:

--- a/handwritten/spanner/src/partial-result-stream.ts
+++ b/handwritten/spanner/src/partial-result-stream.ts
@@ -618,17 +618,27 @@ export function partialResultStream(
       flushStream.push(null);
     });
   };
+
+  const destroyRequestStream = (): void => {
+    if (lastRequestStream) {
+      lastRequestStream.removeListener('end', endListener);
+      lastRequestStream.removeAllListeners('error');
+      lastRequestStream.on('error', () => {});
+      lastRequestStream.unpipe(requestsStream);
+      lastRequestStream.destroy();
+    }
+  };
+
   const makeRequest = (): void => {
     if (isDefined(lastResumeToken) && lastResumeToken.length > 0) {
       partialRSStream._resetPendingValues();
     }
     lastRequestStream = requestFn(lastResumeToken);
     lastRequestStream.on('end', endListener);
     errorListener = (err: grpc.ServiceError) => {
+      destroyRequestStream();
       setImmediate(() => retry(err));
     };
     lastRequestStream.on('error', errorListener);
     lastRequestStream.pipe(requestsStream, {end: false});
   };

};
lastRequestStream.on('error', errorListener);
lastRequestStream.pipe(requestsStream, {end: false});
};

const retry = (err: grpc.ServiceError): void => {
Expand Down Expand Up @@ -659,6 +660,8 @@

if (lastRequestStream) {
lastRequestStream.removeListener('end', endListener);
lastRequestStream.removeAllListeners('error');
lastRequestStream.on('error', () => {}); // Prevent unhandled exception crash
lastRequestStream.destroy();
}
Comment on lines 661 to 666

@olavloite olavloite Aug 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This cleanup code is skipped if the early return above for non-retriable errors is used (the one on line 657). That causes the lastRequestStream open and with all event listeners attached.

Verification test:

it('should destroy the request stream and detach listeners on non-retryable errors', done => {
  const fakeCheckpointStream = through.obj();
  sandbox.stub(checkpointStream, 'obj').returns(fakeCheckpointStream);

  const fakeStream = through.obj();
  const destroySpy = sandbox.spy(fakeStream, 'destroy');

  const requestFnStub = sandbox.stub().callsFake(() => {
    setImmediate(() => {
      fakeStream.emit('error', {
        code: grpc.status.INVALID_ARGUMENT,
        message: 'Invalid query argument.',
      } as grpc.ServiceError);
    });
    return fakeStream;
  });

  partialResultStream(requestFnStub)
    .on('data', () => {})
    .on('error', err => {
      try {
        assert.strictEqual(err.code, grpc.status.INVALID_ARGUMENT);
        assert.strictEqual(destroySpy.called, true, 'Request stream should be destroyed on non-retryable error');
        assert.strictEqual(fakeStream.listenerCount('end'), 0, 'endListener should be removed');
        assert.strictEqual(fakeStream.listenerCount('error'), 0, 'errorListener should be removed');
        done();
      } catch (e) {
        done(e);
      }
    });
});

Suggested fix:

--- a/handwritten/spanner/src/partial-result-stream.ts
+++ b/handwritten/spanner/src/partial-result-stream.ts
@@ -633,6 +633,7 @@ export function partialResultStream(
   };
 
   const retry = (err: grpc.ServiceError): void => {
+    destroyRequestStream();
     const elapsed = Date.now() - startTime;
     if (elapsed >= timeout) {
       // The timeout has reached so this will flush any rows the
@@ -659,13 +660,6 @@ export function partialResultStream(
       return;
     }
 
-    if (lastRequestStream) {
-      lastRequestStream.removeListener('end', endListener);
-      lastRequestStream.removeAllListeners('error');
-      lastRequestStream.on('error', () => {}); // Prevent unhandled exception crash
-      lastRequestStream.destroy();
-    }
     // Delay the retry until all the values that are already in the stream
     // pipeline have been handled. This ensures that the checkpoint stream is

// Delay the retry until all the values that are already in the stream
Expand All @@ -674,15 +677,6 @@
};

userStream.once('reading', makeRequest);
eventsIntercept.patch(requestsStream);

@olavloite olavloite Aug 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If the application calls stream.destroy(), then that is not propagated into 'our' stream. We should add a listener on the userStream for close and make sure that we clean up then the user stream is closed.

Verification test case:

it('should destroy the underlying request stream when the user destroys the returned stream', done => {
  const fakeStream = through.obj();
  const destroySpy = sandbox.spy(fakeStream, 'destroy');

  const requestFnStub = sandbox.stub().returns(fakeStream);

  const stream = partialResultStream(requestFnStub);

  // Read first row and immediately destroy stream
  stream.on('data', () => {
    stream.destroy();
  });

  stream.on('close', () => {
    setImmediate(() => {
      try {
        assert.strictEqual(
          destroySpy.called,
          true,
          'Underlying request stream must be destroyed when user cancels the stream',
        );
        done();
      } catch (e) {
        done(e);
      }
    });
  });

  fakeStream.push(RESULT_WITH_TOKEN);
});

Suggested fix:

--- a/handwritten/spanner/src/partial-result-stream.ts
+++ b/handwritten/spanner/src/partial-result-stream.ts
@@ -677,6 +677,12 @@ export function partialResultStream(
   };
 
   userStream.once('reading', makeRequest);
+  userStream.once('close', () => {
+    destroyRequestStream();
+    requestsStream.destroy();
+    flushStream.destroy();
+    batchAndSplitOnTokenStream.destroy();
+  });
 
   return (
     requestsStream


// need types for events-intercept
// eslint-disable-next-line @typescript-eslint/no-explicit-any
(requestsStream as any).intercept('error', err =>
// Retry __after__ all pending data has been processed to ensure that the
// checkpoint stream is reset at the correct position.
setImmediate(() => retry(err)),
);

return (
requestsStream
Expand Down
Loading