From 3c7e39d977a5adeb187e7211e18e4f8125ff31cc Mon Sep 17 00:00:00 2001 From: ohmayr Date: Tue, 20 Oct 2026 21:48:15 +0000 Subject: [PATCH 1/4] feat(bigquery): resume interrupted Arrow ReadRows streams Resume Storage Read API ReadRows streams at offset + rowsReceived when interrupted by transient gRPC errors (DEADLINE_EXCEEDED, UNAVAILABLE, or INTERNAL with RST_STREAM), up to MAX_STREAM_RETRIES consecutive attempts, waiting 100 ms before the first reopen and doubling the wait for each consecutive retry. Surface non-retryable errors or exhausted retries on the stream even after earlier batches have been emitted. Abandoned gRPC streams (on reconnect or early cancel) keep a no-op 'error' listener, because google-gax emits CANCELLED asynchronously after cancel() and an unhandled 'error' event would crash the process. --- handwritten/bigquery/src/arrow.ts | 127 ++++++++++++++--- handwritten/bigquery/test/arrow.ts | 222 ++++++++++++++++++++++++++++- 2 files changed, 329 insertions(+), 20 deletions(-) diff --git a/handwritten/bigquery/src/arrow.ts b/handwritten/bigquery/src/arrow.ts index 7dec8fee610..63ca70e971a 100644 --- a/handwritten/bigquery/src/arrow.ts +++ b/handwritten/bigquery/src/arrow.ts @@ -15,6 +15,7 @@ */ import {ResourceStream} from '@google-cloud/paginator'; +import {setTimeout as delay} from 'timers/promises'; import type {RecordBatch, Schema} from 'apache-arrow'; import type {BigQuery, JobsQueryResponse, Query} from './bigquery'; import type {Job} from './job'; @@ -132,6 +133,17 @@ interface ArrowStreamContext { const readClients = new WeakMap(); +// gRPC status codes safe to resume from a row offset on ReadRows. +const GRPC_STATUS_DEADLINE_EXCEEDED = 4; +const GRPC_STATUS_INTERNAL = 13; +const GRPC_STATUS_UNAVAILABLE = 14; + +// How many times in a row a failed ReadRows call is reopened before the +// error is passed to the caller. Any rows read reset the count. +const MAX_STREAM_RETRIES = 3; +// Delay before the first reopen, doubled for each consecutive retry. +const RETRY_BASE_DELAY_MS = 100; + /** * Returns the first project ID from `candidates`, falling back to the * `BigQuery` client's project and ignoring the uninitialized `'{{projectId}}'` @@ -431,14 +443,47 @@ async function readDestinationSchema( return session?.arrowSchema?.serializedSchema || undefined; } +/** + * Returns `true` if a `ReadRows` gRPC stream error is transient and safe to + * resume from the next unread row offset. + * + * @private + */ +function isRetryableStreamError(err: Error & {code?: number}): boolean { + if ( + err.code === GRPC_STATUS_DEADLINE_EXCEEDED || + err.code === GRPC_STATUS_UNAVAILABLE + ) { + return true; + } + // Long-lived HTTP/2 streams closed by the server with RST_STREAM surface as + // INTERNAL (code 13) in gRPC and are safe to reconnect at the next offset. + return ( + err.code === GRPC_STATUS_INTERNAL && err.message.includes('RST_STREAM') + ); +} + +/** + * Progress through a Storage Read API stream, kept across reconnects. + * + * @private + */ +interface ReadRowsProgress { + // The next unread row. + offset: number; + emittedBatch: boolean; + // The last Arrow schema seen, needed to decode later record batches. + serializedSchema?: Uint8Array; +} + /** * Streams Arrow `RecordBatch`es from a Storage Read API stream starting at - * `offset`, pausing `ReadRows` while the consumer is behind. + * `offset`. If `ReadRows` fails with a transient error, reopens it at the next + * unread row, waiting longer before each consecutive retry. * - * Resolves when `ReadRows` ends or the consumer destroys the stream, with - * whether any batch was emitted and the last Arrow schema seen, and rejects - * on a `ReadRows` or decode error. In every case, the listeners added here - * are removed, and `ReadRows` is cancelled unless it already finished. + * Resolves when the stream ends or the consumer destroys it, with whether any + * batch was emitted and the last Arrow schema seen. Rejects with the error + * once it isn't retryable or the retries run out. * * @private */ @@ -450,20 +495,63 @@ async function streamReadRows( initialSchema?: Uint8Array | string; }, ): Promise<{emittedBatch: boolean; serializedSchema?: Uint8Array}> { - const {bq, arrow, storage, stream} = ctx; const {readStream, offset, initialSchema} = options; + const progress: ReadRowsProgress = { + offset, + emittedBatch: false, + serializedSchema: initialSchema ? toUint8Array(initialSchema) : undefined, + }; + + let retries = 0; + while (!ctx.stream.destroyed) { + const attemptOffset = progress.offset; + try { + await readRowsOnce(ctx, readStream, progress); + break; + } catch (err) { + if (progress.offset > attemptOffset) { + retries = 0; + } + if ( + !isRetryableStreamError(err as Error & {code?: number}) || + retries >= MAX_STREAM_RETRIES + ) { + throw err; + } + retries++; + // Wait before reopening so a brief network blip doesn't use up every + // retry at once. + await delay(RETRY_BASE_DELAY_MS * 2 ** (retries - 1)); + } + } + const {emittedBatch, serializedSchema} = progress; + return {emittedBatch, serializedSchema}; +} + +/** + * Opens one `ReadRows` call at `progress.offset` and pushes its Arrow + * `RecordBatch`es, pausing `ReadRows` while the consumer is behind. Updates + * `progress` for each batch, so a retry resumes at the next unread row. + * + * Resolves when `ReadRows` ends or the consumer destroys the stream, and + * rejects on a `ReadRows` or decode error. In every case, the listeners added + * here are removed, and `ReadRows` is cancelled unless it already finished. + * + * @private + */ +function readRowsOnce( + ctx: ArrowStreamContext, + readStream: string, + progress: ReadRowsProgress, +): Promise { + const {bq, arrow, storage, stream} = ctx; return new Promise((resolve, reject) => { const grpcStream = getStorageReadClient(bq, storage).readRows({ readStream, - offset, + offset: progress.offset, }); - let serializedSchema: Uint8Array | undefined = initialSchema - ? toUint8Array(initialSchema) - : undefined; - let emittedBatch = false; - let settled = false; // Detaches every listener added here and settles the promise. Cancels // only this ReadRows call, keeping the shared BigQueryReadClient open. @@ -489,7 +577,7 @@ async function streamReadRows( if (opts.err) { reject(opts.err); } else { - resolve({emittedBatch, serializedSchema}); + resolve(); } }; @@ -502,7 +590,9 @@ async function streamReadRows( // and may also emit responses without record batch bytes (e.g., // heartbeats). if (resp.arrowSchema?.serializedSchema) { - serializedSchema = toUint8Array(resp.arrowSchema.serializedSchema); + progress.serializedSchema = toUint8Array( + resp.arrowSchema.serializedSchema, + ); } const rawBatch = resp.arrowRecordBatch?.serializedRecordBatch; const hasBatchBytes = @@ -514,13 +604,18 @@ async function streamReadRows( return; } try { - const batches = decodeArrowPayload(arrow, serializedSchema, rawBatch); + const batches = decodeArrowPayload( + arrow, + progress.serializedSchema, + rawBatch, + ); for (const batch of batches) { if (stream.destroyed) { settle({cancel: true}); return; } - emittedBatch = true; + progress.offset += batch.numRows; + progress.emittedBatch = true; if (!stream.push(batch)) { grpcStream.pause(); } diff --git a/handwritten/bigquery/test/arrow.ts b/handwritten/bigquery/test/arrow.ts index 0069967a658..352e8c1b1c2 100644 --- a/handwritten/bigquery/test/arrow.ts +++ b/handwritten/bigquery/test/arrow.ts @@ -182,20 +182,47 @@ function stubJobGetMetadata( } /** - * Returns a fake ReadRows stream that emits `responses`, then ends, on the - * next tick. + * Returns a fake ReadRows stream that emits `responses` on the next tick, + * then ends, or fails with `err` once the consumer has read them. */ -function fakeReadRowsStream(responses: object[] = []): EventEmitter { +function fakeReadRowsStream( + responses: object[] = [], + err?: Error, +): EventEmitter { const grpcStream = new EventEmitter(); process.nextTick(() => { for (const resp of responses) { grpcStream.emit('data', resp); } - grpcStream.emit('end'); + if (err) { + setImmediate(() => grpcStream.emit('error', err)); + } else { + grpcStream.emit('end'); + } }); return grpcStream; } +/** Returns an error with a gRPC status `code`, like google-gax emits. */ +function grpcError(message: string, code: number): Error { + return Object.assign(new Error(message), {code}); +} + +/** Returns a ReadRowsResponse carrying a serialized Arrow table as bytes. */ +function toReadRowsResponse(payload: ReturnType) { + return { + arrowSchema: { + serializedSchema: Buffer.from(payload.serializedSchema, 'base64'), + }, + arrowRecordBatch: { + serializedRecordBatch: Buffer.from( + payload.serializedRecordBatch, + 'base64', + ), + }, + }; +} + /** A ReadRowsResponse carrying `SAMPLE_PAYLOAD` as bytes. */ const SAMPLE_READ_ROWS_RESPONSE = { arrowSchema: { @@ -1979,4 +2006,191 @@ describe('Arrow Support', () => { assert.strictEqual(readSessions, 0); }); }); + + describe('BigQuery#createQueryArrowStream (stream resumption)', () => { + const jobReference = { + projectId: 'test-project', + jobId: 'job-resume-1', + location: 'US', + }; + const readStream = + 'projects/test-project/locations/US/jobs/job-resume-1/streams/_default'; + let bq: BigQuery; + // The ReadRows streams opened, in order. + let grpcStreams: EventEmitter[]; + + beforeEach(() => { + bq = new BigQuery({projectId: 'test-project'}); + grpcStreams = []; + }); + + const track = (grpcStream: EventEmitter) => { + grpcStreams.push(grpcStream); + return grpcStream; + }; + + // Stubs jobs.query to return `inline` as its first page out of + // `totalRows`, so the remaining rows are read with ReadRows. + const stubFirstPage = ( + totalRows: number, + inline?: ReturnType, + inlineRows = 0, + ) => + stubJobsQuery(bq, { + res: { + jobComplete: true, + jobReference, + pageRowCount: String(inlineRows), + totalRows: String(totalRows), + arrowSchema: {serializedSchema: inline?.serializedSchema}, + arrowRecordBatch: { + serializedRecordBatch: inline?.serializedRecordBatch, + }, + }, + }); + + it('should resume ReadRows at offset + rowsReceived on UNAVAILABLE, DEADLINE_EXCEEDED, and RST_STREAM errors without gaps or duplicates', async () => { + stubFirstPage(4, serializeArrowTable(makeSampleTable([1], ['alice'])), 1); + const page2 = toReadRowsResponse( + serializeArrowTable(makeSampleTable([2, 3], ['bob', 'carol'])), + ); + const page3 = toReadRowsResponse( + serializeArrowTable(makeSampleTable([4], ['dave'])), + ); + // Each ReadRows call's responses and the error it fails with, if any. + const attempts: Array<[object[], Error?]> = [ + [[page2], grpcError('Service unavailable', 14)], + [[], grpcError('Deadline exceeded', 4)], + [[], grpcError('Received RST_STREAM with code 2', 13)], + // Without a schema, so the batch decodes with the one sent earlier. + [[{arrowRecordBatch: page3.arrowRecordBatch}]], + ]; + const requests: ReadRowsRequest[] = []; + stubStorageReadClient(bq, { + readRows: req => { + const [responses, err] = attempts[requests.push(req) - 1]; + return track(fakeReadRowsStream(responses, err)); + }, + }); + + const batches: arrow.RecordBatch[] = await bq + .createQueryArrowStream('SELECT id, name FROM tbl') + .toArray(); + + // Verify ReadRows started after the inline row, and every reopen + // resumed after the 2 rows read before the first error. + assert.deepStrictEqual(requests, [ + {readStream, offset: 1}, + {readStream, offset: 3}, + {readStream, offset: 3}, + {readStream, offset: 3}, + ]); + // Verify all rows arrived in order without gaps or duplicates. + const table = new arrow.Table(batches); + assert.deepStrictEqual( + Array.from(table.getChild('id')!.toArray()), + [1, 2, 3, 4], + ); + assert.deepStrictEqual(table.getChild('name')!.toArray(), [ + 'alice', + 'bob', + 'carol', + 'dave', + ]); + // Verify each attempt, failed or not, kept none of our listeners. + grpcStreams.forEach(assertReadRowsListenersRemoved); + }); + + it('should surface non-retryable ReadRows errors after yielding earlier batches', async () => { + stubFirstPage(3, serializeArrowTable(makeSampleTable([1], ['alice'])), 1); + stubStorageReadClient(bq, { + readRows: () => + track( + fakeReadRowsStream( + [ + toReadRowsResponse( + serializeArrowTable(makeSampleTable([2], ['bob'])), + ), + ], + // INTERNAL without RST_STREAM isn't safe to resume. + grpcError('Corrupted stream state', 13), + ), + ), + }); + + const batches: arrow.RecordBatch[] = []; + await assert.rejects(async () => { + for await (const batch of bq.createQueryArrowStream( + 'SELECT id, name FROM tbl', + )) { + batches.push(batch); + } + }, /Corrupted stream state/); + + // Verify the inline and streamed batches arrived, and ReadRows wasn't + // reopened. + assert.strictEqual(batches.length, 2); + assert.strictEqual(grpcStreams.length, 1); + assertReadRowsListenersRemoved(grpcStreams[0]); + }); + + it('should stop retrying and surface the error when consecutive retryable errors exceed MAX_STREAM_RETRIES', async () => { + stubFirstPage(2); + stubStorageReadClient(bq, { + readRows: () => + track( + fakeReadRowsStream([], grpcError('Persistent UNAVAILABLE', 14)), + ), + }); + + await assert.rejects( + bq.createQueryArrowStream('SELECT id, name FROM tbl').toArray(), + /Persistent UNAVAILABLE/, + ); + + // Verify the first call and 3 retries each kept none of our listeners. + assert.strictEqual(grpcStreams.length, 4); + grpcStreams.forEach(assertReadRowsListenersRemoved); + }); + + it('should surface a synchronous readRows error when reopening ReadRows', async () => { + stubFirstPage(1); + let readRowsCalls = 0; + stubStorageReadClient(bq, { + readRows: () => { + if (++readRowsCalls > 1) { + throw new Error('readRows failed'); + } + return fakeReadRowsStream([], grpcError('Service unavailable', 14)); + }, + }); + + await assert.rejects( + bq.createQueryArrowStream('SELECT id, name FROM tbl').toArray(), + /readRows failed/, + ); + assert.strictEqual(readRowsCalls, 2); + }); + + it('should not reopen ReadRows if the stream is destroyed while waiting to retry', async () => { + stubFirstPage(1); + let readRowsCalls = 0; + stubStorageReadClient(bq, { + readRows: () => { + readRowsCalls++; + return fakeReadRowsStream([], grpcError('Service unavailable', 14)); + }, + }); + + const stream = bq.createQueryArrowStream('SELECT id, name FROM tbl'); + stream.resume(); + // Destroy the stream after the first error, within the first retry + // delay (100 ms), then wait past it. + await new Promise(resolve => setTimeout(resolve, 20)); + stream.destroy(); + await new Promise(resolve => setTimeout(resolve, 250)); + + assert.strictEqual(readRowsCalls, 1); + }); + }); }); From b703117ba3bfcbcddcf84e4e5bd48214b0203a5f Mon Sep 17 00:00:00 2001 From: ohmayr Date: Wed, 21 Oct 2026 00:58:50 +0000 Subject: [PATCH 2/4] fix(bigquery): stop retrying DEADLINE_EXCEEDED; stub Arrow retry waits DEADLINE_EXCEEDED on ReadRows means the caller's deadline ran out, so pass it to the caller instead of reopening the stream, matching the design doc and the Python reader. Route the retry backoff through retryBackoff.wait so unit tests replace it instead of sleeping on real timers, and check the 100/200/400 ms sequence. Mark readRowsOnce async. --- handwritten/bigquery/src/arrow.ts | 21 ++++++++---- handwritten/bigquery/test/arrow.ts | 51 ++++++++++++++++++++++++++---- 2 files changed, 58 insertions(+), 14 deletions(-) diff --git a/handwritten/bigquery/src/arrow.ts b/handwritten/bigquery/src/arrow.ts index 63ca70e971a..86e0f60d4df 100644 --- a/handwritten/bigquery/src/arrow.ts +++ b/handwritten/bigquery/src/arrow.ts @@ -134,7 +134,6 @@ interface ArrowStreamContext { const readClients = new WeakMap(); // gRPC status codes safe to resume from a row offset on ReadRows. -const GRPC_STATUS_DEADLINE_EXCEEDED = 4; const GRPC_STATUS_INTERNAL = 13; const GRPC_STATUS_UNAVAILABLE = 14; @@ -443,6 +442,16 @@ async function readDestinationSchema( return session?.arrowSchema?.serializedSchema || undefined; } +/** + * Waits between `ReadRows` retries. Exported so tests can replace the wait + * without real timers. + * + * @private + */ +export const retryBackoff = { + wait: (ms: number): Promise => delay(ms), +}; + /** * Returns `true` if a `ReadRows` gRPC stream error is transient and safe to * resume from the next unread row offset. @@ -450,10 +459,8 @@ async function readDestinationSchema( * @private */ function isRetryableStreamError(err: Error & {code?: number}): boolean { - if ( - err.code === GRPC_STATUS_DEADLINE_EXCEEDED || - err.code === GRPC_STATUS_UNAVAILABLE - ) { + // DEADLINE_EXCEEDED isn't retried: it means the caller's deadline ran out. + if (err.code === GRPC_STATUS_UNAVAILABLE) { return true; } // Long-lived HTTP/2 streams closed by the server with RST_STREAM surface as @@ -521,7 +528,7 @@ async function streamReadRows( retries++; // Wait before reopening so a brief network blip doesn't use up every // retry at once. - await delay(RETRY_BASE_DELAY_MS * 2 ** (retries - 1)); + await retryBackoff.wait(RETRY_BASE_DELAY_MS * 2 ** (retries - 1)); } } const {emittedBatch, serializedSchema} = progress; @@ -539,7 +546,7 @@ async function streamReadRows( * * @private */ -function readRowsOnce( +async function readRowsOnce( ctx: ArrowStreamContext, readStream: string, progress: ReadRowsProgress, diff --git a/handwritten/bigquery/test/arrow.ts b/handwritten/bigquery/test/arrow.ts index 352e8c1b1c2..f99104546dc 100644 --- a/handwritten/bigquery/test/arrow.ts +++ b/handwritten/bigquery/test/arrow.ts @@ -30,6 +30,7 @@ import { checkArrowDependencies, CreateReadSessionRequest, getStorageReadClient, + retryBackoff, StorageModule, } from '../src/arrow'; import bigquery from '../src/types'; @@ -2018,10 +2019,22 @@ describe('Arrow Support', () => { let bq: BigQuery; // The ReadRows streams opened, in order. let grpcStreams: EventEmitter[]; + // The retry backoffs requested, in ms. Replaces the real wait, so these + // tests don't use timers. + let waits: number[]; + const realWait = retryBackoff.wait; beforeEach(() => { bq = new BigQuery({projectId: 'test-project'}); grpcStreams = []; + waits = []; + retryBackoff.wait = async ms => { + waits.push(ms); + }; + }); + + after(() => { + retryBackoff.wait = realWait; }); const track = (grpcStream: EventEmitter) => { @@ -2049,7 +2062,7 @@ describe('Arrow Support', () => { }, }); - it('should resume ReadRows at offset + rowsReceived on UNAVAILABLE, DEADLINE_EXCEEDED, and RST_STREAM errors without gaps or duplicates', async () => { + it('should resume ReadRows at offset + rowsReceived on UNAVAILABLE and RST_STREAM errors without gaps or duplicates', async () => { stubFirstPage(4, serializeArrowTable(makeSampleTable([1], ['alice'])), 1); const page2 = toReadRowsResponse( serializeArrowTable(makeSampleTable([2, 3], ['bob', 'carol'])), @@ -2060,7 +2073,7 @@ describe('Arrow Support', () => { // Each ReadRows call's responses and the error it fails with, if any. const attempts: Array<[object[], Error?]> = [ [[page2], grpcError('Service unavailable', 14)], - [[], grpcError('Deadline exceeded', 4)], + [[], grpcError('Connection reset', 14)], [[], grpcError('Received RST_STREAM with code 2', 13)], // Without a schema, so the batch decodes with the one sent earlier. [[{arrowRecordBatch: page3.arrowRecordBatch}]], @@ -2099,6 +2112,25 @@ describe('Arrow Support', () => { ]); // Verify each attempt, failed or not, kept none of our listeners. grpcStreams.forEach(assertReadRowsListenersRemoved); + // Verify the backoff doubled for each consecutive retry. + assert.deepStrictEqual(waits, [100, 200, 400]); + }); + + it('should not retry DEADLINE_EXCEEDED, which means the caller deadline ran out', async () => { + stubFirstPage(2); + stubStorageReadClient(bq, { + readRows: () => + track(fakeReadRowsStream([], grpcError('Deadline exceeded', 4))), + }); + + await assert.rejects( + bq.createQueryArrowStream('SELECT id, name FROM tbl').toArray(), + /Deadline exceeded/, + ); + // Verify ReadRows wasn't reopened. + assert.strictEqual(grpcStreams.length, 1); + assert.deepStrictEqual(waits, []); + assertReadRowsListenersRemoved(grpcStreams[0]); }); it('should surface non-retryable ReadRows errors after yielding earlier batches', async () => { @@ -2151,6 +2183,7 @@ describe('Arrow Support', () => { // Verify the first call and 3 retries each kept none of our listeners. assert.strictEqual(grpcStreams.length, 4); grpcStreams.forEach(assertReadRowsListenersRemoved); + assert.deepStrictEqual(waits, [100, 200, 400]); }); it('should surface a synchronous readRows error when reopening ReadRows', async () => { @@ -2183,13 +2216,17 @@ describe('Arrow Support', () => { }); const stream = bq.createQueryArrowStream('SELECT id, name FROM tbl'); + // The consumer gives up while the stream waits to retry. + retryBackoff.wait = async ms => { + waits.push(ms); + stream.destroy(); + }; + const closed = new Promise(resolve => stream.once('close', resolve)); stream.resume(); - // Destroy the stream after the first error, within the first retry - // delay (100 ms), then wait past it. - await new Promise(resolve => setTimeout(resolve, 20)); - stream.destroy(); - await new Promise(resolve => setTimeout(resolve, 250)); + await closed; + // Verify the stream waited once and didn't reopen ReadRows. + assert.deepStrictEqual(waits, [100]); assert.strictEqual(readRowsCalls, 1); }); }); From ef7ca58b16e48d6bb1362afb8d74825823a29ee4 Mon Sep 17 00:00:00 2001 From: ohmayr Date: Wed, 21 Oct 2026 07:28:03 +0000 Subject: [PATCH 3/4] fix(bigquery): reconnect ReadRows immediately without client-side delay Reopen ReadRows at the next unread row offset without an added client-side timer on mid-stream UNAVAILABLE and RST_STREAM drops, matching ReadStream in @google-cloud/bigquery-storage and Python's ReadRowsStream. Initial connection failures when opening or reopening a ReadRows stream are already retried with exponential backoff by google-gax (retry-request). --- handwritten/bigquery/src/arrow.ts | 18 +------------ handwritten/bigquery/test/arrow.ts | 42 ------------------------------ 2 files changed, 1 insertion(+), 59 deletions(-) diff --git a/handwritten/bigquery/src/arrow.ts b/handwritten/bigquery/src/arrow.ts index 86e0f60d4df..32c35b87fe6 100644 --- a/handwritten/bigquery/src/arrow.ts +++ b/handwritten/bigquery/src/arrow.ts @@ -15,7 +15,6 @@ */ import {ResourceStream} from '@google-cloud/paginator'; -import {setTimeout as delay} from 'timers/promises'; import type {RecordBatch, Schema} from 'apache-arrow'; import type {BigQuery, JobsQueryResponse, Query} from './bigquery'; import type {Job} from './job'; @@ -140,8 +139,6 @@ const GRPC_STATUS_UNAVAILABLE = 14; // How many times in a row a failed ReadRows call is reopened before the // error is passed to the caller. Any rows read reset the count. const MAX_STREAM_RETRIES = 3; -// Delay before the first reopen, doubled for each consecutive retry. -const RETRY_BASE_DELAY_MS = 100; /** * Returns the first project ID from `candidates`, falling back to the @@ -442,16 +439,6 @@ async function readDestinationSchema( return session?.arrowSchema?.serializedSchema || undefined; } -/** - * Waits between `ReadRows` retries. Exported so tests can replace the wait - * without real timers. - * - * @private - */ -export const retryBackoff = { - wait: (ms: number): Promise => delay(ms), -}; - /** * Returns `true` if a `ReadRows` gRPC stream error is transient and safe to * resume from the next unread row offset. @@ -486,7 +473,7 @@ interface ReadRowsProgress { /** * Streams Arrow `RecordBatch`es from a Storage Read API stream starting at * `offset`. If `ReadRows` fails with a transient error, reopens it at the next - * unread row, waiting longer before each consecutive retry. + * unread row. * * Resolves when the stream ends or the consumer destroys it, with whether any * batch was emitted and the last Arrow schema seen. Rejects with the error @@ -526,9 +513,6 @@ async function streamReadRows( throw err; } retries++; - // Wait before reopening so a brief network blip doesn't use up every - // retry at once. - await retryBackoff.wait(RETRY_BASE_DELAY_MS * 2 ** (retries - 1)); } } const {emittedBatch, serializedSchema} = progress; diff --git a/handwritten/bigquery/test/arrow.ts b/handwritten/bigquery/test/arrow.ts index f99104546dc..81afb8f7f75 100644 --- a/handwritten/bigquery/test/arrow.ts +++ b/handwritten/bigquery/test/arrow.ts @@ -30,7 +30,6 @@ import { checkArrowDependencies, CreateReadSessionRequest, getStorageReadClient, - retryBackoff, StorageModule, } from '../src/arrow'; import bigquery from '../src/types'; @@ -2019,22 +2018,10 @@ describe('Arrow Support', () => { let bq: BigQuery; // The ReadRows streams opened, in order. let grpcStreams: EventEmitter[]; - // The retry backoffs requested, in ms. Replaces the real wait, so these - // tests don't use timers. - let waits: number[]; - const realWait = retryBackoff.wait; beforeEach(() => { bq = new BigQuery({projectId: 'test-project'}); grpcStreams = []; - waits = []; - retryBackoff.wait = async ms => { - waits.push(ms); - }; - }); - - after(() => { - retryBackoff.wait = realWait; }); const track = (grpcStream: EventEmitter) => { @@ -2112,8 +2099,6 @@ describe('Arrow Support', () => { ]); // Verify each attempt, failed or not, kept none of our listeners. grpcStreams.forEach(assertReadRowsListenersRemoved); - // Verify the backoff doubled for each consecutive retry. - assert.deepStrictEqual(waits, [100, 200, 400]); }); it('should not retry DEADLINE_EXCEEDED, which means the caller deadline ran out', async () => { @@ -2129,7 +2114,6 @@ describe('Arrow Support', () => { ); // Verify ReadRows wasn't reopened. assert.strictEqual(grpcStreams.length, 1); - assert.deepStrictEqual(waits, []); assertReadRowsListenersRemoved(grpcStreams[0]); }); @@ -2183,7 +2167,6 @@ describe('Arrow Support', () => { // Verify the first call and 3 retries each kept none of our listeners. assert.strictEqual(grpcStreams.length, 4); grpcStreams.forEach(assertReadRowsListenersRemoved); - assert.deepStrictEqual(waits, [100, 200, 400]); }); it('should surface a synchronous readRows error when reopening ReadRows', async () => { @@ -2204,30 +2187,5 @@ describe('Arrow Support', () => { ); assert.strictEqual(readRowsCalls, 2); }); - - it('should not reopen ReadRows if the stream is destroyed while waiting to retry', async () => { - stubFirstPage(1); - let readRowsCalls = 0; - stubStorageReadClient(bq, { - readRows: () => { - readRowsCalls++; - return fakeReadRowsStream([], grpcError('Service unavailable', 14)); - }, - }); - - const stream = bq.createQueryArrowStream('SELECT id, name FROM tbl'); - // The consumer gives up while the stream waits to retry. - retryBackoff.wait = async ms => { - waits.push(ms); - stream.destroy(); - }; - const closed = new Promise(resolve => stream.once('close', resolve)); - stream.resume(); - await closed; - - // Verify the stream waited once and didn't reopen ReadRows. - assert.deepStrictEqual(waits, [100]); - assert.strictEqual(readRowsCalls, 1); - }); }); }); From 90f8ea1dfb1f21eb67cd5a7c77efad38cad8c1fd Mon Sep 17 00:00:00 2001 From: ohmayr Date: Wed, 21 Oct 2026 22:37:39 +0000 Subject: [PATCH 4/4] fix(bigquery): guard err.message and destroyed destination streams Use optional chaining on err.message in isRetryableStreamError and skip decoding the 0-row schema batch in streamDestinationTable when the stream is already destroyed. --- handwritten/bigquery/src/arrow.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/handwritten/bigquery/src/arrow.ts b/handwritten/bigquery/src/arrow.ts index 32c35b87fe6..c925854d981 100644 --- a/handwritten/bigquery/src/arrow.ts +++ b/handwritten/bigquery/src/arrow.ts @@ -453,7 +453,8 @@ function isRetryableStreamError(err: Error & {code?: number}): boolean { // Long-lived HTTP/2 streams closed by the server with RST_STREAM surface as // INTERNAL (code 13) in gRPC and are safe to reconnect at the next offset. return ( - err.code === GRPC_STATUS_INTERNAL && err.message.includes('RST_STREAM') + err.code === GRPC_STATUS_INTERNAL && + Boolean(err.message?.includes('RST_STREAM')) ); } @@ -665,7 +666,7 @@ async function streamDestinationTable( initialSchema: sessionSchema, }); // A stream that ends without rows still yields a 0-row RecordBatch. - if (!emittedBatch) { + if (!emittedBatch && !stream.destroyed) { pushBatches(stream, decodeArrowPayload(arrow, serializedSchema)); } }