From d9ebf5bcb0ca59d5a5c07b861ad2d098e78a7e24 Mon Sep 17 00:00:00 2001 From: feywind <57276408+feywind@users.noreply.github.com> Date: Thu, 10 Sep 2026 15:14:39 -0400 Subject: [PATCH 1/3] feat(gax): support resumable uploads Add the client-side implementation of the resumable upload protocol: - ResumableUploadDescriptor and ResumableUploadSession, plus the resumableUploadStub that generated clients wire into createApiCall - resumableSourceFromFile, a seekable source backed by a local file - CallOptions.resumableUpload carrying the transport context that generated clients pass to the stub - exports from index, fallback and descriptor, and client-libraries docs - unit and hermetic system tests covering the state machine, transient retries, recovery from state mismatches and resume from a saved URL The onProgress callback may return void; the documented usage logs progress without returning a value. --- core/packages/gax/client-libraries.md | 42 + core/packages/gax/src/clientInterface.ts | 2 + core/packages/gax/src/descriptor.ts | 1 + core/packages/gax/src/fallback.ts | 12 + core/packages/gax/src/gax.ts | 14 + core/packages/gax/src/index.ts | 13 + .../gax/src/resumableSourceFromFile.ts | 43 + core/packages/gax/src/resumableUpload.ts | 1459 +++++++++++++++++ .../gax/test/system-test/resumableUpload.ts | 286 ++++ .../packages/gax/test/unit/resumableUpload.ts | 1245 ++++++++++++++ 10 files changed, 3117 insertions(+) create mode 100644 core/packages/gax/src/resumableSourceFromFile.ts create mode 100644 core/packages/gax/src/resumableUpload.ts create mode 100644 core/packages/gax/test/system-test/resumableUpload.ts create mode 100644 core/packages/gax/test/unit/resumableUpload.ts diff --git a/core/packages/gax/client-libraries.md b/core/packages/gax/client-libraries.md index d64cea0550fc..d66025315cd4 100644 --- a/core/packages/gax/client-libraries.md +++ b/core/packages/gax/client-libraries.md @@ -149,6 +149,48 @@ in the second parameter: const [response] = await client.sampleMethod(request, options); ``` +### Resumable uploads + +Some APIs expose methods that upload large payloads through the resumable +upload protocol. For these methods, the client method no +longer returns the response directly; it returns a +[`ResumableUpload`](https://googleapis.dev/nodejs/google-gax/latest/classes/ResumableUpload.html) +helper. Call `start()` with a `NodeJS.ReadableStream` and await `finished()` +for the final response: + +```ts +const helper = await client.createResumableUpload(request); +await helper.start({ + uploadStream: dataStream, + chunkSize: 8 * 1024 * 1024, // 8MB chunks + onProgress: status => { + console.log(`Committed ${status.bytesUploaded} bytes to ${status.uploadUrl}`); + }, +}); +const response = await helper.finished(); +``` + +The session URL is available as `helper.uploadUrl` once the upload has +started. Save it if you need to resume the upload later — for example after a +process crash or network drop. To resume, pass the saved URL to `start()` on a +new helper, along with a fresh stream of the same payload: + +```ts +const helper = await client.createResumableUpload(); +await helper.start({ + uploadStream: dataStream, + resumeUrl: savedUploadUrl, +}); +const response = await helper.finished(); +``` + +The current implementation requires a seekable stream (for example, a file +stream). Errors fall into three categories: transient errors (retried with +exponential backoff), state mismatches (recovered by querying the server for +the committed byte offset), and fatal errors (propagated to the caller). +The whole session is bounded by a global deadline (10 minutes by default, +scaled up for large payloads and overridable via `globalDeadlineMs`). + ### Long-running operations Some methods are expected to run longer. They return an object of type diff --git a/core/packages/gax/src/clientInterface.ts b/core/packages/gax/src/clientInterface.ts index bf4571d62689..931caa524d51 100644 --- a/core/packages/gax/src/clientInterface.ts +++ b/core/packages/gax/src/clientInterface.ts @@ -25,6 +25,7 @@ import { PageDescriptor, StreamDescriptor, } from './descriptor'; +import {ResumableUploadDescriptor} from './resumableUpload'; import * as longrunning from './longRunningCalls/longrunning'; import * as operationProtos from '../protos/operations'; @@ -51,6 +52,7 @@ export interface Descriptors { stream: {[name: string]: StreamDescriptor}; longrunning: {[name: string]: LongrunningDescriptor}; batching?: {[name: string]: BundleDescriptor}; + resumableUpload?: {[name: string]: ResumableUploadDescriptor}; } export interface Callback< diff --git a/core/packages/gax/src/descriptor.ts b/core/packages/gax/src/descriptor.ts index 5cfad86eb48e..42a37ceab512 100644 --- a/core/packages/gax/src/descriptor.ts +++ b/core/packages/gax/src/descriptor.ts @@ -31,3 +31,4 @@ export {LongRunningDescriptor as LongrunningDescriptor} from './longRunningCalls export {PageDescriptor} from './paginationCalls/pageDescriptor'; export {StreamDescriptor} from './streamingCalls/streamDescriptor'; export {BundleDescriptor} from './bundlingCalls/bundleDescriptor'; +export {ResumableUploadDescriptor} from './resumableUpload'; diff --git a/core/packages/gax/src/fallback.ts b/core/packages/gax/src/fallback.ts index 32d122b3a67c..18daacaed539 100644 --- a/core/packages/gax/src/fallback.ts +++ b/core/packages/gax/src/fallback.ts @@ -55,6 +55,18 @@ export { PageDescriptor, StreamDescriptor, } from './descriptor'; +export { + ResumableUploadDescriptor, + ResumableUploadSession, + ResumableUploadState, + resumableUploadStub, +} from './resumableUpload'; +export type { + ResumableUploadContext, + ResumableUploadProgress, + ResumableUploadStartParams, + ResumableSource, +} from './resumableUpload'; export {StreamType} from './streamingCalls/streaming'; diff --git a/core/packages/gax/src/gax.ts b/core/packages/gax/src/gax.ts index f9a7e23da913..cfe8895d8b16 100644 --- a/core/packages/gax/src/gax.ts +++ b/core/packages/gax/src/gax.ts @@ -19,6 +19,7 @@ */ import type {Message} from 'protobufjs'; +import type {ResumableUploadContext} from './resumableUpload'; import {warn} from './warnings'; import {GoogleError} from './googleError'; import {BundleOptions} from './bundlingCalls/bundleExecutor'; @@ -172,6 +173,11 @@ export interface CallOptions { apiName?: string; retryRequestOptions?: RetryRequestOptions; enableTelemetryTracing?: boolean; + /** + * Internal context used by resumable upload methods. Populated by + * GAPIC-generated client libraries; do not set manually. + */ + resumableUpload?: ResumableUploadContext; } export class CallSettings { @@ -189,6 +195,7 @@ export class CallSettings { apiName?: string; retryRequestOptions?: RetryRequestOptions; enableTelemetryTracing?: boolean; + resumableUpload?: ResumableUploadContext; /** * @param {Object} settings - An object containing parameters of this settings. @@ -223,6 +230,8 @@ export class CallSettings { this.apiName = settings.apiName ?? undefined; this.retryRequestOptions = settings.retryRequestOptions; this.enableTelemetryTracing = settings.enableTelemetryTracing; + this.resumableUpload = + 'resumableUpload' in settings ? settings.resumableUpload : undefined; } /** @@ -247,6 +256,7 @@ export class CallSettings { let apiName = this.apiName; let retryRequestOptions = this.retryRequestOptions; let enableTelemetryTracing = this.enableTelemetryTracing; + let resumableUpload = this.resumableUpload; // If the user provides a timeout to the method, that timeout value will be used // to override the backoff settings. @@ -305,6 +315,9 @@ export class CallSettings { if ('enableTelemetryTracing' in options) { enableTelemetryTracing = options.enableTelemetryTracing; } + if ('resumableUpload' in options) { + resumableUpload = options.resumableUpload; + } return new CallSettings({ timeout, @@ -318,6 +331,7 @@ export class CallSettings { apiName, retryRequestOptions, enableTelemetryTracing, + resumableUpload, }); } } diff --git a/core/packages/gax/src/index.ts b/core/packages/gax/src/index.ts index b0d09dd3b4db..f1bc80e8c5b6 100644 --- a/core/packages/gax/src/index.ts +++ b/core/packages/gax/src/index.ts @@ -36,6 +36,19 @@ export { PageDescriptor, StreamDescriptor, } from './descriptor'; +export {ResumableUploadDescriptor} from './resumableUpload'; +export { + ResumableUploadSession, + resumableUploadStub, + ResumableUploadState, +} from './resumableUpload'; +export type { + ResumableUploadContext, + ResumableUploadProgress, + ResumableUploadStartParams, + ResumableSource, +} from './resumableUpload'; +export {resumableSourceFromFile} from './resumableSourceFromFile'; export { CallOptions, CallSettings, diff --git a/core/packages/gax/src/resumableSourceFromFile.ts b/core/packages/gax/src/resumableSourceFromFile.ts new file mode 100644 index 000000000000..0c6b8cc672c2 --- /dev/null +++ b/core/packages/gax/src/resumableSourceFromFile.ts @@ -0,0 +1,43 @@ +/** + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import {createReadStream} from 'fs'; +import {statSync} from 'fs'; + +import {ResumableSource} from './resumableUpload'; + +/** + * Creates a {@link ResumableSource} backed by a local file. + * + * This factory lives outside the fallback transport entrypoint so browser + * builds do not pull in the Node.js `fs` module. Generated clients should + * delegate to this function rather than constructing file streams directly. + */ +export function resumableSourceFromFile(filePath: string): ResumableSource { + const stat = statSync(filePath); + return { + size: stat.size, + getStream: (offset?: number) => { + const start = offset ?? 0; + if (start < 0 || start > stat.size) { + throw new RangeError( + `Invalid start offset ${start} for file of size ${stat.size}.`, + ); + } + return createReadStream(filePath, {start}); + }, + }; +} diff --git a/core/packages/gax/src/resumableUpload.ts b/core/packages/gax/src/resumableUpload.ts new file mode 100644 index 000000000000..1aff192c44b7 --- /dev/null +++ b/core/packages/gax/src/resumableUpload.ts @@ -0,0 +1,1459 @@ +/** + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +/** + * Resumable upload protocol support. + * + * This module implements the client-side state machine for the resumable + * upload protocol used by Google APIs to transfer large payloads over + * HTTP(S). The upload session is managed by a {@link ResumableUploadSession} + * object, which is returned by GAPIC-generated client methods for resumable + * upload RPCs. + * + * The protocol commands are sent through the `X-Goog-Upload-Command` header + * and the payload is transferred in discrete chunks. Transient errors + * (Category 1) are retried with exponential backoff, state mismatches + * (Category 2) trigger a recovery phase that queries the server for the + * committed byte offset, and everything else (Category 3) is fatal. + */ + +import type {AuthClient, GoogleAuth} from 'google-auth-library'; +import * as protobuf from 'protobufjs'; +import * as serializer from 'proto3-json-serializer'; + +import {APICaller} from './apiCaller'; +import {APICallback, GRPCCall, SimpleCallbackFunction} from './apitypes'; +import {OngoingCall, OngoingCallPromise} from './call'; +import {Descriptor} from './descriptor'; +import {decodeResponse} from './fallbackRest'; +import {CallSettings, createDefaultBackoffSettings, RetryOptions} from './gax'; +import {GoogleError} from './googleError'; +import {Status, rpcCodeFromHttpStatusCode} from './status'; +import {transcode} from './transcoding'; + +export const DEFAULT_CHUNK_SIZE = 8 * 1024 * 1024; +export const DEFAULT_GLOBAL_DEADLINE_MS = 10 * 60 * 1000; +export const MAX_GLOBAL_DEADLINE_MS = 24 * 60 * 60 * 1000; +export const DEFAULT_MAX_INNER_RETRIES = 5; +export const DEFAULT_PER_REQUEST_TIMEOUT_MS = 60 * 1000; +export const DEFAULT_STALL_TIMEOUT_MS = 15 * 1000; +// Assumed sustained upload throughput, in bytes per millisecond, used to +// scale the global deadline when `uploadSize` is provided (~5 MB/s). +const DEFAULT_UPLOAD_RATE_BYTES_PER_MS = 5 * 1024 * 1024; + +// Resumable upload protocol headers. +const UPLOAD_PROTOCOL_HEADER = 'x-goog-upload-protocol'; +const UPLOAD_PROTOCOL_RESUMABLE = 'resumable'; +const UPLOAD_COMMAND_HEADER = 'x-goog-upload-command'; +const UPLOAD_OFFSET_HEADER = 'x-goog-upload-offset'; +const UPLOAD_URL_HEADER = 'x-goog-upload-url'; +const UPLOAD_STATUS_HEADER = 'x-goog-upload-status'; +const UPLOAD_SIZE_RECEIVED_HEADER = 'x-goog-upload-size-received'; +const UPLOAD_CHUNK_GRANULARITY_HEADER = 'x-goog-upload-chunk-granularity'; + +// Resumable upload protocol commands. +const COMMAND_START = 'start'; +const COMMAND_UPLOAD = 'upload'; +const COMMAND_QUERY = 'query'; +const COMMAND_FINALIZE = 'finalize'; +const COMMAND_CANCEL = 'cancel'; +const COMMAND_UPLOAD_FINALIZE = 'upload, finalize'; + +// Category 1 errors are transient and can be retried without modification. +const CATEGORY_1_RETRY_CODES = new Set([408, 429, 500, 502, 503, 504]); +// Category 2 errors are state mismatches; recovery must query the server for +// the committed byte offset before retrying. +const CATEGORY_2_RETRY_CODES = new Set([400, 412, 416]); + +/** The possible states of a resumable upload session. */ +export enum ResumableUploadState { + /** The upload has not yet begun transmitting. */ + STARTING = 'STARTING', + /** The stream transfer is in progress. */ + TRANSMISSION = 'TRANSMISSION', + /** The transfer is complete, but we are waiting for confirmation. */ + FINALIZING = 'FINALIZING', + /** Recovery from an existing upload session URL. */ + RECOVERY = 'RECOVERY', +} + +/** + * A seekable source of upload bytes. + * + * The upload session only accepts a `ResumableSource`, never a bare stream, + * because recovery may need to open a new stream at the server-committed + * byte offset. Callers that only have a non-seekable stream can wrap it in a + * `ResumableSource` whose `getStream(offset)` throws when an offset other + * than 0 is requested. + */ +export interface ResumableSource { + /** + * Generates a readable stream starting at the specified byte offset. If + * `offset` is omitted or 0, the stream starts from the beginning. + */ + getStream: (offset?: number) => NodeJS.ReadableStream | ReadableStream; + /** Total byte length of the data. */ + size: number; +} + +/** Progress reported to the `onProgress` callback. */ +export interface ResumableUploadProgress { + /** The number of bytes committed by the server so far. */ + bytesUploaded: number; + /** The session URL, which can be saved and reused to resume the upload. */ + uploadUrl: string; +} + +/** + * Parameters accepted by {@link ResumableUploadSession.start}. + */ +export interface ResumableUploadStartParams { + /** + * Seekable source of the upload payload. The session may call + * `getStream(offset)` again during recovery, so callers must not pass a + * bare one-shot `Readable`. + */ + uploadSource: ResumableSource; + /** + * Desired chunk size in bytes. The effective chunk size is rounded down to + * a multiple of the server-provided chunk granularity. + */ + chunkSize?: number; + /** + * Called after each committed chunk and after recovery queries. The return + * value is reserved for future cancellation support and is not yet acted + * on. + */ + onProgress?: (status: ResumableUploadProgress) => boolean | void; + /** + * Reserved for headers that may accompany upload-phase data. Not yet + * supported; passing a value is a compile-time error. + */ + uploadHeaders?: never; + /** + * Session URL of a previous (possibly interrupted) upload session. When + * provided, the `start` command is skipped and the upload enters the + * recovery phase to determine the server-committed byte offset. + */ + resumeUrl?: string; + /** + * Total size of the payload in bytes, if known. Used to scale the global + * deadline for large payloads. + */ + uploadSize?: number; + /** + * Override for the global deadline, in milliseconds. The deadline bounds + * the entire upload session, including time spent waiting for data from + * the upload source. + */ + globalDeadlineMs?: number; + /** + * Headers that must only be sent with the initial `start` request (for + * example, `developer-token`). + */ + startHeaders?: {[name: string]: string}; + /** + * Reserved for checksum validation. Not yet supported; passing a value is + * a compile-time error. + */ + validationAlgorithm?: never; + /** + * Reserved for a user-supplied checksum. Not yet supported; passing a + * value is a compile-time error. + */ + providedChecksum?: never; + /** Per-request timeout in milliseconds. Defaults to 60000. */ + timeout?: number; + /** + * Inactivity timeout, in milliseconds, for an in-flight upload/finalize + * request. When no response arrives within this window, the request is + * aborted and the session enters the recovery phase. Defaults to 15000. + */ + stallTimeoutMs?: number; + /** + * Retry configuration for the inner (Category 1) retry loop. Pass `null` + * to disable inner retries. + */ + retry?: Partial | null; +} + +/** + * Internal context used to construct a {@link ResumableUploadSession}. This + * is populated by GAPIC-generated client methods. + */ +export interface ResumableUploadContext { + /** Authenticated client used for all HTTP requests. */ + auth: GoogleAuth | AuthClient; + /** The hostname of the API service endpoint. */ + servicePath: string; + /** The port of the API service endpoint. */ + servicePort: number; + /** The protocol (usually `https`). */ + protocol: string; + /** The protobuf method descriptor for the resumable upload RPC. */ + rpc: protobuf.Method; + /** The initial metadata request for the `start` command. */ + request: {}; + /** The upload prefix to use for the `start` command endpoint. */ + uploadPrefix?: string; + numericEnums?: boolean; + minifyJson?: boolean; +} + +interface ResumableUploadResponse { + status: number; + headers: {get(name: string): string | null}; + body: Buffer; +} + +/** Transient error that should be retried (Category 1). */ +class TransientError extends Error {} + +/** State mismatch error that triggers the recovery phase (Category 2). */ +class Category2Error extends Error { + constructor( + message: string, + public httpStatusCode?: number, + ) { + super(message); + this.code = Status.FAILED_PRECONDITION; + } + code: Status; +} + +/** Fatal error that cannot be recovered from (Category 3). */ +class Category3Error extends Error { + constructor(message: string, httpStatusCode?: number) { + super(message); + if (httpStatusCode !== undefined) { + this.code = rpcCodeFromHttpStatusCode(httpStatusCode); + } + } + code?: Status; +} + +/** Raised when an in-flight request made no progress for too long. */ +class StallError extends Error {} + +/** + * Internal signal used to restart transmission from a server-committed + * offset by opening a fresh stream from the `ResumableSource`. + */ +class RestartUploadError extends Error { + constructor(public offset: number) { + super(`Restarting the resumable upload transmission from byte ${offset}.`); + } +} + +function createGoogleError(message: string, code?: Status): GoogleError { + const err = new GoogleError(message); + if (code !== undefined) { + err.code = code; + } + return err; +} + +interface ReadChunkResult { + chunk: Buffer | null; + eof: boolean; + remainder: Buffer; +} + +interface TransmitResult { + finalized: boolean; + response: {} | null; + /** The local offset after the server-committed bytes. */ + newOffset: number; + /** Bytes the stream must skip when the server committed more than expected. */ + skipAhead: number; +} + +/** + * A no-op RPC stub used by GAPIC-generated resumable upload methods. + * Resumable uploads perform their own HTTP requests through the + * {@link ResumableUploadSession} object, so the stub passed to + * `createApiCall` is never invoked. + */ +export const resumableUploadStub = (() => { + return {cancel() {}}; +}) as unknown as GRPCCall; + +function sleep(ms: number): Promise { + return new Promise(resolve => setTimeout(resolve, ms)); +} + +function asBuffer(chunk: unknown): Buffer { + if (Buffer.isBuffer(chunk)) { + return chunk; + } + if (chunk instanceof Uint8Array) { + return Buffer.from(chunk); + } + return Buffer.from(String(chunk)); +} + +function parseHeaderInt(value: string | null): number | null { + if (value === null || value === '') { + return null; + } + const parsed = parseInt(value, 10); + return Number.isNaN(parsed) ? null : parsed; +} + +/** + * Descriptor that identifies a method as a resumable upload method and + * provides the caller that constructs the {@link ResumableUploadSession} + * object. + */ +export class ResumableUploadDescriptor implements Descriptor { + constructor(public uploadPrefix: string = '/resumable/upload') {} + + getApiCaller(): APICaller { + return new ResumableUploadApiCaller(this); + } +} + +/** + * API caller for resumable upload methods. The GAPIC method call resolves + * with a {@link ResumableUploadSession} object; the actual upload state + * machine is driven by {@link ResumableUploadSession.start}. + */ +export class ResumableUploadApiCaller implements APICaller { + constructor(private descriptor: ResumableUploadDescriptor) {} + + init(callback?: APICallback): OngoingCallPromise | OngoingCall { + if (callback) { + return new OngoingCall(callback); + } + return new OngoingCallPromise(); + } + + // The regular API call function is never invoked for resumable uploads; + // the upload session performs its own HTTP requests. + // eslint-disable-next-line @typescript-eslint/no-unused-vars + wrap(func: GRPCCall): GRPCCall { + return func; + } + + call( + // eslint-disable-next-line @typescript-eslint/no-unused-vars + apiCall: SimpleCallbackFunction, + // eslint-disable-next-line @typescript-eslint/no-unused-vars + argument: {}, + settings: CallSettings, + canceller: OngoingCallPromise, + ): void { + const context = settings.resumableUpload; + if (!context) { + canceller.callback!( + createGoogleError( + 'The resumable upload transport context was not provided. ' + + 'This is a bug in the generated client library.', + ), + ); + return; + } + const session = new ResumableUploadSession({ + ...context, + uploadPrefix: + context.uploadPrefix ?? + this.descriptor.uploadPrefix ?? + '/resumable/upload', + }); + canceller.completed = true; + canceller.callback!(null, session); + } + + fail(canceller: OngoingCallPromise, err: GoogleError): void { + canceller.callback!(err); + } + + result(canceller: OngoingCallPromise) { + return canceller.promise; + } +} + +/** + * Client-side state machine for the resumable upload protocol. + * + * A `ResumableUploadSession` object is returned by a GAPIC-generated client + * method for a resumable upload RPC. The user calls + * {@link ResumableUploadSession.start} + * with a source and upload parameters, optionally supplies a `resumeUrl` from + * a previous session, and awaits {@link ResumableUploadSession.finished} for the + * final RPC response. + */ +export class ResumableUploadSession { + private context: ResumableUploadContext; + private params: ResumableUploadStartParams | null = null; + private state_: ResumableUploadState = ResumableUploadState.STARTING; + private uploadUrl_: string | null = null; + private response_: {} | null = null; + private committedBytes_ = 0; + private startTimeMs = 0; + private globalDeadlineMs = DEFAULT_GLOBAL_DEADLINE_MS; + private effectiveChunkSize_ = DEFAULT_CHUNK_SIZE; + private activeAbortController: AbortController | null = null; + private activeStream_: NodeJS.ReadableStream | ReadableStream | null = null; + private activeIterator_: AsyncIterator | null = null; + private deadlineTimer: ReturnType | null = null; + private stallTimer: ReturnType | null = null; + private canceled_ = false; + private started_ = false; + private done_ = false; + private finishedPromise_: Promise<{}>; + private resolveFinished_!: (response: {}) => void; + private rejectFinished_!: (err: Error) => void; + + constructor(context: ResumableUploadContext) { + this.context = context; + this.finishedPromise_ = new Promise<{}>((resolve, reject) => { + this.resolveFinished_ = resolve; + this.rejectFinished_ = reject; + }); + } + + /** The session URL, available once the upload session has been started. */ + get uploadUrl(): string | null { + return this.uploadUrl_; + } + + /** The current state of the upload session. */ + get state(): ResumableUploadState { + return this.state_; + } + + /** The actual chunk size used for the session (granularity rounded). */ + get chunkSize(): number | null { + return this.started_ ? this.effectiveChunkSize_ : null; + } + + /** The number of bytes the server has committed so far. */ + get committedBytes(): number { + return this.committedBytes_; + } + + /** + * Cancels the upload session, aborts any in-flight HTTP request, releases + * the active upload source stream, and rejects the promise returned by + * {@link ResumableUploadSession.finished}. + */ + cancel(): void { + if (this.canceled_ || this.done_) { + return; + } + this.canceled_ = true; + this.clearDeadlineTimer(); + this.clearStallTimer(); + if (this.uploadUrl_) { + // Best-effort server-side cancellation of the session. + const controller = new AbortController(); + this.context.auth + .request({ + url: this.uploadUrl_, + method: 'POST', + headers: { + [UPLOAD_PROTOCOL_HEADER]: UPLOAD_PROTOCOL_RESUMABLE, + [UPLOAD_COMMAND_HEADER]: COMMAND_CANCEL, + }, + signal: controller.signal, + responseType: 'text', + timeout: this.params?.timeout ?? DEFAULT_PER_REQUEST_TIMEOUT_MS, + validateStatus: () => true, + }) + .catch(() => {}); + } + this.activeAbortController?.abort(); + // Release the active upload source stream so the transmission loop does + // not stay subscribed to it indefinitely (for example, when it is + // stalled). + this.releaseActiveStream(); + const err = createGoogleError( + 'The resumable upload was cancelled.', + Status.CANCELLED, + ); + this.rejectFinished_(err); + } + + /** + * Starts the upload session and begins transmitting the source. + * + * The returned promise resolves once the upload session has been + * established (either via the `start` command or via recovery from a + * `resumeUrl`) and the transmission loop is running. Await + * {@link ResumableUploadSession.finished} for the final response. + */ + async start(params: ResumableUploadStartParams): Promise { + if (this.started_) { + throw createGoogleError('The resumable upload has already been started.'); + } + if (this.canceled_) { + throw createGoogleError('The resumable upload was cancelled.'); + } + if (!params.uploadSource) { + throw createGoogleError( + 'uploadSource must be provided to start a resumable upload.', + ); + } + this.params = params; + this.started_ = true; + this.startTimeMs = Date.now(); + this.globalDeadlineMs = this.computeGlobalDeadlineMs(params); + this.armDeadlineTimer(); + + let sessionUrl: string; + let granularity: number | null = null; + + try { + if (params.resumeUrl) { + this.state_ = ResumableUploadState.RECOVERY; + const committedOffset = await this.queryOffset(params.resumeUrl); + this.uploadUrl_ = params.resumeUrl; + this.committedBytes_ = committedOffset; + this.reportProgress(); + sessionUrl = params.resumeUrl; + } else { + this.state_ = ResumableUploadState.STARTING; + const started = await this.sendStart(); + sessionUrl = started.uploadUrl; + granularity = started.granularity; + this.uploadUrl_ = sessionUrl; + if (started.committedOffset !== undefined) { + this.committedBytes_ = started.committedOffset; + this.reportProgress(); + } + } + } catch (err) { + // Session setup failed before transmission began; make sure awaiting + // `finished()` does not hang forever. + this.done_ = true; + this.clearDeadlineTimer(); + this.rejectFinished_(err as Error); + throw err; + } + + this.effectiveChunkSize_ = this.computeEffectiveChunkSize( + params.chunkSize, + granularity, + ); + this.state_ = ResumableUploadState.TRANSMISSION; + + // The transmission loop runs in the background; `start()` resolves once + // the session is established so the user can inspect `uploadUrl`. + void this.runTransmission(sessionUrl); + } + + /** + * Returns a promise that resolves with the final RPC response when the + * upload completes, or rejects if the upload fails or is cancelled. + */ + finished(): Promise<{}> { + return this.finishedPromise_; + } + + /** Total payload size in bytes, preferring the explicit upload size. */ + private uploadSizeForDeadline(params: ResumableUploadStartParams): number { + if (params.uploadSize !== undefined && params.uploadSize > 0) { + return params.uploadSize; + } + return params.uploadSource.size > 0 ? params.uploadSource.size : 0; + } + + private computeGlobalDeadlineMs(params: ResumableUploadStartParams): number { + if (params.globalDeadlineMs !== undefined && params.globalDeadlineMs > 0) { + return params.globalDeadlineMs; + } + let deadline = DEFAULT_GLOBAL_DEADLINE_MS; + const uploadSize = this.uploadSizeForDeadline(params); + if (uploadSize > 0) { + const scaled = Math.ceil(uploadSize / DEFAULT_UPLOAD_RATE_BYTES_PER_MS); + deadline = Math.max(deadline, scaled); + } + return Math.min(deadline, MAX_GLOBAL_DEADLINE_MS); + } + + private computeEffectiveChunkSize( + chunkSize: number | undefined, + granularity: number | null, + ): number { + const requested = chunkSize ?? DEFAULT_CHUNK_SIZE; + if (!granularity || granularity <= 0) { + return requested; + } + const effective = Math.floor(requested / granularity) * granularity; + // Non-final chunks must be a multiple of the server granularity. If the + // user's requested chunk size rounds down to zero, fall back to the + // smallest legal chunk size. + return effective > 0 ? effective : granularity; + } + + private async sendStart(): Promise<{ + uploadUrl: string; + granularity: number | null; + committedOffset?: number; + }> { + const rpc = this.context.rpc; + if (!rpc.resolvedRequestType) { + throw new Category3Error( + `Cannot start resumable upload for method ${rpc.name}: ` + + 'the resolved request type is unavailable.', + ); + } + const message = rpc.resolvedRequestType.fromObject(this.context.request); + const json = serializer.toProto3JSON(message, { + numericEnums: this.context.numericEnums ?? false, + }); + if (!json || typeof json !== 'object' || Array.isArray(json)) { + throw new Category3Error( + `Cannot serialize the request for resumable upload method ${rpc.name}.`, + ); + } + + let queryString = ''; + try { + const transcoded = transcode(json, rpc.parsedOptions); + queryString = transcoded?.queryString ?? ''; + } catch { + // The method may not have a google.api.http rule; the request body is + // still sent as proto JSON to the upload endpoint. + } + if (this.context.numericEnums) { + queryString = `${queryString ? `${queryString}&` : ''}$alt=json%3Benum-encoding=int`; + } + if (this.context.minifyJson) { + queryString = `${queryString ? `${queryString}&` : ''}$prettyPrint=0`; + } + + const uploadPrefix = this.context.uploadPrefix ?? '/resumable/upload'; + const url = `${this.getEndpointBase()}${uploadPrefix}${ + queryString ? `?${queryString}` : '' + }`; + const body = JSON.stringify(json); + const startHeaders: {[name: string]: string} = { + 'content-type': 'application/json', + ...(this.params?.startHeaders ?? {}), + }; + + const response = await this.sendCommandWithRetry( + url, + COMMAND_START, + body, + -1, + startHeaders, + ); + const uploadUrl = response.headers.get(UPLOAD_URL_HEADER); + if (!uploadUrl) { + throw new Category3Error( + 'The resumable upload start response did not include a ' + + `${UPLOAD_URL_HEADER} header.`, + ); + } + const granularity = parseHeaderInt( + response.headers.get(UPLOAD_CHUNK_GRANULARITY_HEADER), + ); + // A successful start normally includes X-Goog-Upload-Status. If it is + // missing, the session was still created: reconcile with a query and + // resume from the server-committed offset instead of failing. + if (response.headers.get(UPLOAD_STATUS_HEADER) === null) { + this.state_ = ResumableUploadState.RECOVERY; + const committedOffset = await this.queryOffset(uploadUrl); + this.state_ = ResumableUploadState.STARTING; + return {uploadUrl, granularity, committedOffset}; + } + return {uploadUrl, granularity}; + } + + /** + * Builds the protocol/host/port prefix for upload endpoints, parsing a + * `host:port` form from `servicePath` when present (matching the fallback + * transport behavior). + */ + private getEndpointBase(): string { + let servicePath = this.context.servicePath; + let servicePort = this.context.servicePort; + const match = servicePath.match(/^(.*):(\d+)$/); + if (match) { + servicePath = match[1]; + servicePort = parseInt(match[2], 10); + } + return `${this.context.protocol}://${servicePath}:${servicePort}`; + } + + private async queryOffset(sessionUrl: string): Promise { + this.state_ = ResumableUploadState.RECOVERY; + try { + const response = await this.sendCommandWithRetry( + sessionUrl, + COMMAND_QUERY, + null, + -1, + undefined, + ); + if (response.status < 200 || response.status >= 300) { + throw new Category3Error( + `Resumable upload recovery query failed: HTTP ${response.status}.`, + response.status, + ); + } + const size = parseHeaderInt( + response.headers.get(UPLOAD_SIZE_RECEIVED_HEADER), + ); + if (size === null) { + throw new Category3Error( + 'The resumable upload recovery query did not include a ' + + `${UPLOAD_SIZE_RECEIVED_HEADER} header.`, + ); + } + return size; + } finally { + this.state_ = ResumableUploadState.TRANSMISSION; + } + } + + private async runTransmission(sessionUrl: string): Promise { + try { + let buffer = Buffer.alloc(0); + let offset = this.committedBytes_; + let previousChunk: Buffer | null = null; + let response: {} | null = null; + await this.openSourceAt(offset); + + // eslint-disable-next-line no-constant-condition + while (true) { + this.assertActive(); + this.assertDeadline(); + const read = await this.readNextChunk(buffer); + buffer = read.remainder; + + if (read.chunk === null) { + // EOF with no pending bytes: either the stream was empty or the + // payload ended exactly on a chunk boundary. Send the finalize + // command on its own. + this.state_ = ResumableUploadState.FINALIZING; + response = await this.transmitFinalize(sessionUrl, null, offset); + break; + } + + const transmit = await this.transmitChunk( + sessionUrl, + read.chunk, + offset, + previousChunk, + read.eof, + ); + previousChunk = read.chunk; + offset = transmit.newOffset; + this.committedBytes_ = offset; + this.reportProgress(); + + if (transmit.skipAhead > 0) { + if (buffer.length >= transmit.skipAhead) { + buffer = buffer.subarray(transmit.skipAhead); + } else { + buffer = await this.skipBytes(transmit.skipAhead - buffer.length); + } + } + if (transmit.finalized) { + response = transmit.response; + break; + } + if (response !== null) { + break; + } + } + + if (this.state_ !== ResumableUploadState.FINALIZING) { + this.state_ = ResumableUploadState.FINALIZING; + } + if (response === null) { + throw new Category3Error( + 'The resumable upload completed without a final response.', + ); + } + this.response_ = response; + this.done_ = true; + this.clearDeadlineTimer(); + this.resolveFinished_(response); + } catch (err) { + if (err instanceof RestartUploadError) { + // The server reported an offset that is not covered by the local + // buffer. Re-open the source at the committed offset and continue. + try { + this.committedBytes_ = err.offset; + this.reportProgress(); + await this.openSourceAt(err.offset); + // Restart the transmission loop from the new offset. + await this.runTransmission(sessionUrl); + return; + } catch (restartErr) { + if (!this.canceled_) { + this.done_ = true; + this.clearDeadlineTimer(); + this.rejectFinished_(restartErr as Error); + } + return; + } + } + if (!this.canceled_) { + this.done_ = true; + this.clearDeadlineTimer(); + this.rejectFinished_(err as Error); + } + } + } + + /** + * Arms a watchdog timer that enforces the global deadline across the whole + * session, including time spent waiting for data from the upload stream + * (which the point-in-time {@link assertDeadline} checks cannot cover). + */ + private armDeadlineTimer(): void { + this.clearDeadlineTimer(); + this.deadlineTimer = setTimeout(() => { + if (this.canceled_ || this.done_) { + return; + } + // Abort any in-flight request and release the stream so the + // transmission loop terminates instead of remaining suspended. + this.activeAbortController?.abort(); + this.clearStallTimer(); + this.releaseActiveStream(); + this.done_ = true; + this.rejectFinished_(this.deadlineError()); + }, this.globalDeadlineMs); + } + + private clearDeadlineTimer(): void { + if (this.deadlineTimer) { + clearTimeout(this.deadlineTimer); + this.deadlineTimer = null; + } + } + + private clearStallTimer(): void { + if (this.stallTimer) { + clearTimeout(this.stallTimer); + this.stallTimer = null; + } + } + + /** + * Opens the current upload stream from the source at the given byte + * offset, releasing any previously active stream first. + */ + private async openSourceAt(offset: number): Promise { + const source = this.params?.uploadSource; + if (!source) { + throw createGoogleError( + 'Cannot open the upload source because start() was not called.', + ); + } + this.releaseActiveStream(); + const stream = source.getStream(offset); + this.activeStream_ = stream; + this.activeIterator_ = (stream as unknown as AsyncIterable)[ + Symbol.asyncIterator + ](); + } + + /** + * Best-effort release of the active upload stream, so a transmission + * loop blocked waiting for stream data does not stay subscribed forever. + */ + private releaseActiveStream(): void { + if (this.activeStream_) { + ( + this.activeStream_ as unknown as { + destroy?: (error?: Error) => void; + } + ).destroy?.(); + this.activeStream_ = null; + this.activeIterator_ = null; + } + } + + private deadlineError(): GoogleError { + return createGoogleError( + 'The resumable upload exceeded its global deadline of ' + + `${this.globalDeadlineMs} ms.`, + Status.DEADLINE_EXCEEDED, + ); + } + + /** + * Reads the next chunk of the requested size from the stream, aggregating + * stream data events until the chunk is full or the stream ends. + */ + private async readNextChunk(buffer: Buffer): Promise { + const iterator = this.activeIterator_; + if (!iterator) { + throw createGoogleError( + 'Cannot read from the upload source because it is not open.', + ); + } + const chunkSize = this.effectiveChunkSize_; + while (buffer.length < chunkSize) { + const next = await iterator.next(); + if (next.done) { + if (buffer.length === 0) { + return {chunk: null, eof: true, remainder: Buffer.alloc(0)}; + } + return {chunk: buffer, eof: true, remainder: Buffer.alloc(0)}; + } + buffer = Buffer.concat([buffer, asBuffer(next.value)]); + } + const chunk = buffer.subarray(0, chunkSize); + return {chunk, eof: false, remainder: buffer.subarray(chunkSize)}; + } + + /** + * Discards `bytes` bytes from the active stream, returning any leftover + * bytes from the chunk that crossed the boundary so they can be prepended + * to the transmission buffer. + */ + private async skipBytes(bytes: number): Promise { + const iterator = this.activeIterator_; + if (!iterator) { + throw createGoogleError( + 'Cannot skip the upload source because it is not open.', + ); + } + let remaining = bytes; + while (remaining > 0) { + const next = await iterator.next(); + if (next.done) { + throw new Category3Error( + `The server committed ${bytes} bytes beyond the end of the ` + + 'provided stream; the payload appears shorter than expected.', + ); + } + const chunk = asBuffer(next.value); + if (chunk.length >= remaining) { + return chunk.subarray(remaining); + } + remaining -= chunk.length; + } + return Buffer.alloc(0); + } + + /** + * Transmits a chunk, applying the outer recovery loop when the server + * reports a state mismatch (Category 2 error). + */ + private async transmitChunk( + sessionUrl: string, + chunk: Buffer, + offset: number, + previousChunk: Buffer | null, + isFinal: boolean, + ): Promise { + let currentChunk = chunk; + let currentOffset = offset; + let currentPrevious = previousChunk; + + // eslint-disable-next-line no-constant-condition + while (true) { + this.assertActive(); + this.assertDeadline(); + const command = isFinal ? COMMAND_UPLOAD_FINALIZE : COMMAND_UPLOAD; + try { + const response = await this.sendCommandWithRetry( + sessionUrl, + command, + currentChunk, + currentOffset, + undefined, + ); + if (isFinal) { + return { + finalized: true, + response: this.decodeFinalResponse(response), + newOffset: currentOffset + currentChunk.length, + skipAhead: 0, + }; + } + return { + finalized: false, + response: null, + newOffset: currentOffset + currentChunk.length, + skipAhead: 0, + }; + } catch (err) { + const recoverable = + err instanceof Category2Error || err instanceof StallError; + if (!recoverable) { + throw err; + } + + // Outer recovery: query the server for the exact committed offset, + // align the local state, and re-enter the transmission phase. + this.state_ = ResumableUploadState.RECOVERY; + const serverOffset = await this.queryOffset(sessionUrl); + this.state_ = ResumableUploadState.TRANSMISSION; + this.reportProgress(); + + if (err instanceof StallError) { + // A stalled request may have committed any number of bytes; the + // safest recovery is to re-open the source at the committed offset. + throw new RestartUploadError(serverOffset); + } + if (serverOffset === currentOffset) { + // Nothing was committed; retry the same chunk. + continue; + } + if (serverOffset > currentOffset) { + if (serverOffset === currentOffset + currentChunk.length) { + // The chunk was committed but the response was lost. + if (isFinal) { + const finalResponse = await this.transmitFinalize( + sessionUrl, + null, + serverOffset, + ); + return { + finalized: true, + response: finalResponse, + newOffset: serverOffset, + skipAhead: 0, + }; + } + return { + finalized: false, + response: null, + newOffset: serverOffset, + skipAhead: 0, + }; + } + if (serverOffset > currentOffset + currentChunk.length) { + // The server is ahead of the local state; skip the stream forward + // by the number of bytes already committed. + if (isFinal) { + // The final chunk was already committed; retrieve the response + // by sending the finalize command on its own. + const finalResponse = await this.transmitFinalize( + sessionUrl, + null, + serverOffset, + ); + return { + finalized: true, + response: finalResponse, + newOffset: serverOffset, + skipAhead: 0, + }; + } + return { + finalized: false, + response: null, + newOffset: serverOffset, + skipAhead: serverOffset - (currentOffset + currentChunk.length), + }; + } + // The server committed part of this chunk; retransmit the tail. + currentChunk = currentChunk.subarray(serverOffset - currentOffset); + currentOffset = serverOffset; + continue; + } + + // The server is behind the local state: data from the previous chunk + // must be re-transmitted from the in-memory buffer. + const bufferedStart = currentPrevious + ? currentOffset - currentPrevious.length + : currentOffset; + if (currentPrevious && serverOffset >= bufferedStart) { + const rebuilt = Buffer.concat([currentPrevious, currentChunk]); + currentChunk = rebuilt.subarray(serverOffset - bufferedStart); + currentOffset = serverOffset; + currentPrevious = null; + continue; + } + // The requested offset is not covered by the in-memory buffer. The + // source is seekable, so restart the stream at the server offset. + throw new RestartUploadError(serverOffset); + } + } + } + + /** + * Sends the `finalize` command (optionally preceded by the final chunk), + * recovering from state mismatches by re-querying the committed offset and + * re-aligning the in-memory buffer. + */ + private async transmitFinalize( + sessionUrl: string, + finalChunk: Buffer | null, + offset: number, + ): Promise<{}> { + let chunk = finalChunk; + let currentOffset = offset; + + // eslint-disable-next-line no-constant-condition + while (true) { + this.assertActive(); + this.assertDeadline(); + const command = chunk ? COMMAND_UPLOAD_FINALIZE : COMMAND_FINALIZE; + try { + const response = await this.sendCommandWithRetry( + sessionUrl, + command, + chunk, + currentOffset, + undefined, + ); + return this.decodeFinalResponse(response); + } catch (err) { + const recoverable = + err instanceof Category2Error || err instanceof StallError; + if (!recoverable) { + throw err; + } + const serverOffset = await this.queryOffset(sessionUrl); + this.reportProgress(); + if (err instanceof StallError) { + throw new RestartUploadError(serverOffset); + } + if (chunk) { + if (serverOffset === currentOffset) { + // Nothing was committed; retry the same final chunk. + continue; + } + if (serverOffset === currentOffset + chunk.length) { + // The final chunk was committed; finalize on its own. + chunk = null; + currentOffset = serverOffset; + continue; + } + if ( + serverOffset > currentOffset && + serverOffset < currentOffset + chunk.length + ) { + chunk = chunk.subarray(serverOffset - currentOffset); + currentOffset = serverOffset; + continue; + } + if (serverOffset > currentOffset + chunk.length) { + chunk = null; + currentOffset = serverOffset; + continue; + } + } else { + if (serverOffset >= currentOffset) { + currentOffset = serverOffset; + continue; + } + } + + throw new RestartUploadError(serverOffset); + } + } + } + + private decodeFinalResponse(response: ResumableUploadResponse): {} { + try { + return decodeResponse(this.context.rpc, true, response.body); + } catch (err) { + throw new Category3Error( + `Failed to decode the resumable upload final response: ${ + (err as Error).message + }`, + ); + } + } + + /** + * Sends a single protocol command, retrying transient (Category 1) errors + * with exponential backoff. + */ + private async sendCommandWithRetry( + url: string, + command: string, + body: Buffer | string | null, + offset: number, + extraHeaders: {[name: string]: string} | undefined, + ): Promise { + const retry = this.getRetrySettings(); + let attempt = 0; + let delay = retry.initialDelayMs; + + // eslint-disable-next-line no-constant-condition + while (true) { + try { + return await this.fetchCommand( + url, + command, + body, + offset, + extraHeaders, + ); + } catch (err) { + if (this.canceled_) { + throw createGoogleError( + 'The resumable upload was cancelled.', + Status.CANCELLED, + ); + } + if (err instanceof Category2Error || err instanceof Category3Error) { + throw err; + } + if (!(err instanceof TransientError)) { + throw err; + } + if (attempt >= retry.maxRetries) { + throw createGoogleError( + `Exceeded the maximum number of retries (${retry.maxRetries}) ` + + `while sending the resumable upload command "${command}".`, + Status.DEADLINE_EXCEEDED, + ); + } + this.assertDeadline(); + await sleep(delay); + delay = Math.min(delay * retry.delayMultiplier, retry.maxDelayMs); + attempt += 1; + } + } + } + + private async fetchCommand( + url: string, + command: string, + body: Buffer | string | null, + offset: number, + extraHeaders: {[name: string]: string} | undefined, + ): Promise { + const headers: {[name: string]: string} = { + [UPLOAD_PROTOCOL_HEADER]: UPLOAD_PROTOCOL_RESUMABLE, + [UPLOAD_COMMAND_HEADER]: command, + }; + const requiresOffset = + body !== null || + command === COMMAND_UPLOAD || + command === COMMAND_UPLOAD_FINALIZE || + command === COMMAND_FINALIZE; + if (requiresOffset && offset >= 0) { + headers[UPLOAD_OFFSET_HEADER] = String(offset); + } + if (command === COMMAND_START) { + headers['content-type'] = 'application/json'; + } else if (body !== null) { + headers['content-type'] = 'application/octet-stream'; + } + if (extraHeaders) { + for (const [name, value] of Object.entries(extraHeaders)) { + headers[name.toLowerCase()] = value; + } + } + + const controller = new AbortController(); + this.activeAbortController = controller; + let stalled = false; + const isTransferRequest = + command === COMMAND_UPLOAD || + command === COMMAND_UPLOAD_FINALIZE || + command === COMMAND_FINALIZE; + if (isTransferRequest) { + this.clearStallTimer(); + this.stallTimer = setTimeout(() => { + stalled = true; + controller.abort(new Error('Resumable upload stalled.')); + }, this.params?.stallTimeoutMs ?? DEFAULT_STALL_TIMEOUT_MS); + } + try { + // Mirror the fetch behavior used by the REST fallback transport: + // `responseType: 'stream'` returns a Response-like object with access + // to status, headers, and the body. `validateStatus` is disabled so + // error statuses can be classified as Category 1/2/3 by the state + // machine instead of being rejected by gaxios. + const response = (await this.context.auth.request({ + url, + method: 'POST', + headers, + body: + body === null || body === undefined + ? undefined + : typeof body === 'string' + ? body + : Buffer.from(body), + signal: controller.signal, + responseType: 'stream', + timeout: this.params?.timeout ?? DEFAULT_PER_REQUEST_TIMEOUT_MS, + validateStatus: () => true, + })) as unknown as Response; + const responseBody = Buffer.from(await response.arrayBuffer()); + const uploadResponse: ResumableUploadResponse = { + status: response.status, + headers: this.normalizeHeaders(response.headers), + body: responseBody, + }; + this.throwOnNonTransientStatus(uploadResponse, command); + return uploadResponse; + } catch (err) { + if (this.canceled_) { + throw createGoogleError( + 'The resumable upload was cancelled.', + Status.CANCELLED, + ); + } + if (stalled) { + throw new StallError( + 'The resumable upload request stalled: no progress was made within ' + + `${this.params?.stallTimeoutMs ?? DEFAULT_STALL_TIMEOUT_MS} ms.`, + ); + } + if (err instanceof Category2Error || err instanceof Category3Error) { + throw err; + } + throw new TransientError( + 'Transient failure while sending the resumable upload command ' + + `"${command}": ${(err as Error).message}`, + ); + } finally { + this.clearStallTimer(); + if (this.activeAbortController === controller) { + this.activeAbortController = null; + } + } + } + + /** + * Classifies the HTTP response status. Transient errors are thrown as + * {@link TransientError}, state mismatches as {@link Category2Error}, and + * everything else as {@link Category3Error}. + */ + private throwOnNonTransientStatus( + response: ResumableUploadResponse, + command: string, + ): void { + if (response.status >= 200 && response.status < 300) { + const uploadStatus = response.headers.get(UPLOAD_STATUS_HEADER); + if (uploadStatus === null) { + if (this.state_ === ResumableUploadState.RECOVERY) { + // During a recovery query there is no offset to reconcile, so a + // missing status header is fatal. + throw new Category3Error( + `The resumable upload ${command} response did not include a ` + + `${UPLOAD_STATUS_HEADER} header while in recovery.`, + ); + } + if (command === COMMAND_START) { + // The start handler reconciles via the returned session URL. + return; + } + // A missing status header on a starting/transmission/finalizing + // response is a recoverable state mismatch. + if ( + this.state_ === ResumableUploadState.STARTING || + this.state_ === ResumableUploadState.TRANSMISSION || + this.state_ === ResumableUploadState.FINALIZING + ) { + throw new Category2Error( + 'The resumable upload response did not include a ' + + `${UPLOAD_STATUS_HEADER} header.`, + ); + } + throw new Category3Error( + `The resumable upload ${command} response did not include a ` + + `${UPLOAD_STATUS_HEADER} header.`, + ); + } + return; + } + if (CATEGORY_2_RETRY_CODES.has(response.status)) { + throw new Category2Error( + `Resumable upload state mismatch: HTTP ${response.status}.`, + response.status, + ); + } + if (CATEGORY_1_RETRY_CODES.has(response.status)) { + throw new TransientError( + `Resumable upload transient error: HTTP ${response.status}.`, + ); + } + throw new Category3Error( + `Resumable upload failed: HTTP ${response.status}.`, + response.status, + ); + } + + private normalizeHeaders(headers: unknown): { + get(name: string): string | null; + } { + if (headers && typeof (headers as {get?: unknown}).get === 'function') { + return headers as {get(name: string): string | null}; + } + const map = new Map(); + if (headers && typeof headers === 'object') { + for (const [name, value] of Object.entries( + headers as {[name: string]: unknown}, + )) { + map.set(name.toLowerCase(), String(value)); + } + } + return { + get(name: string): string | null { + return map.get(name.toLowerCase()) ?? null; + }, + }; + } + + private getRetrySettings(): { + maxRetries: number; + initialDelayMs: number; + delayMultiplier: number; + maxDelayMs: number; + } { + const retry = this.params?.retry; + if (retry === null) { + return { + maxRetries: 0, + initialDelayMs: 100, + delayMultiplier: 1.3, + maxDelayMs: 60000, + }; + } + const backoffSettings = + retry?.backoffSettings ?? createDefaultBackoffSettings(); + return { + maxRetries: backoffSettings.maxRetries ?? DEFAULT_MAX_INNER_RETRIES, + initialDelayMs: backoffSettings.initialRetryDelayMillis ?? 100, + delayMultiplier: backoffSettings.retryDelayMultiplier ?? 1.3, + maxDelayMs: backoffSettings.maxRetryDelayMillis ?? 60000, + }; + } + + private reportProgress(): void { + if (this.params?.onProgress && this.uploadUrl_) { + this.params.onProgress({ + bytesUploaded: this.committedBytes_, + uploadUrl: this.uploadUrl_, + }); + } + } + + private assertActive(): void { + if (this.canceled_) { + throw createGoogleError( + 'The resumable upload was cancelled.', + Status.CANCELLED, + ); + } + } + + private assertDeadline(): void { + if (Date.now() - this.startTimeMs >= this.globalDeadlineMs) { + throw this.deadlineError(); + } + } +} diff --git a/core/packages/gax/test/system-test/resumableUpload.ts b/core/packages/gax/test/system-test/resumableUpload.ts new file mode 100644 index 000000000000..6ec61014824e --- /dev/null +++ b/core/packages/gax/test/system-test/resumableUpload.ts @@ -0,0 +1,286 @@ +/** + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +// Hermetic end-to-end test for the resumable upload protocol: a real HTTP +// server (no credentials required) exercises the full state machine, +// including a simulated crash and cross-helper session resume. + +/* eslint-disable @typescript-eslint/no-explicit-any */ + +import assert from 'assert'; +import * as crypto from 'crypto'; +import * as fs from 'fs'; +import * as http from 'http'; +import {AddressInfo} from 'net'; +import * as os from 'os'; +import * as path from 'path'; +import * as protobuf from 'protobufjs'; +import {Readable} from 'stream'; +import {after, before, describe, it} from 'mocha'; + +import * as gax from '../../src'; +import {ResumableUploadContext} from '../../src/resumableUpload'; + +const GRANULARITY = 1024 * 1024; +const CHUNK_SIZE = 2 * 1024 * 1024; + +const PROTO = ` +syntax = "proto3"; +package test.v1; +message UploadRequest { string name = 1; } +message UploadResponse { string status = 1; } +service UploadService { + rpc CreateUpload(UploadRequest) returns (UploadResponse); +} +`; + +class MockResumableUploadServer { + server: http.Server; + port = 0; + sessionUrl = ''; + received = Buffer.alloc(0); + commands: Array<{command: string; offset: number; bodyLength: number}> = []; + + constructor() { + this.server = http.createServer((req, res) => { + void this.handle(req, res); + }); + } + + private async handle(req: http.IncomingMessage, res: http.ServerResponse) { + const chunks: Buffer[] = []; + for await (const chunk of req) { + chunks.push(chunk as Buffer); + } + const body = Buffer.concat(chunks); + const command = String(req.headers['x-goog-upload-command'] ?? ''); + const offset = parseInt( + String(req.headers['x-goog-upload-offset'] ?? '-1'), + 10, + ); + this.commands.push({command, offset, bodyLength: body.length}); + + const send = ( + status: number, + headers: {[name: string]: string}, + responseBody = '', + ) => { + res.writeHead(status, headers); + res.end(responseBody); + }; + const active = (extra: {[name: string]: string} = {}) => ({ + 'x-goog-upload-status': 'active', + 'x-goog-upload-size-received': String(this.received.length), + ...extra, + }); + + if (req.url === '/resumable/upload' && command === 'start') { + JSON.parse(body.toString()); + return send(200, { + 'x-goog-upload-status': 'active', + 'x-goog-upload-url': this.sessionUrl, + 'x-goog-upload-chunk-granularity': String(GRANULARITY), + }); + } + if (req.url !== '/upload/session-1') { + return send(404, active(), 'not found'); + } + if (command === 'query') { + return send(200, active()); + } + if (command === 'cancel') { + return send(200, {'x-goog-upload-status': 'cancelled'}); + } + if (command === 'upload' || command === 'upload, finalize') { + if (offset !== this.received.length) { + return send(416, active(), 'offset mismatch'); + } + this.received = Buffer.concat([this.received, body]); + if (command === 'upload, finalize') { + return send( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({status: 'done'}), + ); + } + return send(200, active()); + } + if (command === 'finalize') { + if (offset !== this.received.length) { + return send(416, active(), 'offset mismatch'); + } + return send( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({status: 'done'}), + ); + } + return send(400, active(), `unknown command ${command}`); + } + + listen(): Promise { + return new Promise(resolve => { + this.server.listen(0, '127.0.0.1', () => { + this.port = (this.server.address() as AddressInfo).port; + this.sessionUrl = `http://127.0.0.1:${this.port}/upload/session-1`; + resolve(); + }); + }); + } + + close(): Promise { + return new Promise(resolve => this.server.close(() => resolve())); + } +} + +class CrashingStream extends Readable { + private data: Buffer; + private pos = 0; + private limit: number; + + constructor(data: Buffer, limit: number, start = 0) { + super(); + this.data = start >= limit ? Buffer.alloc(0) : data.subarray(start); + this.limit = Math.max(limit - start, 0); + } + + _read(): void { + if (this.pos >= this.limit) { + this.destroy(new Error('simulated process crash')); + return; + } + const end = Math.min(this.pos + 65536, this.limit); + const chunk = this.data.subarray(this.pos, end); + this.pos = end; + this.push(chunk); + } +} + +// A minimal authenticated client that forwards requests to the local mock +// server through the real fetch implementation. +const fakeAuth = { + async request(opts: { + url?: string; + method?: string; + headers?: {[name: string]: string}; + body?: string | Buffer; + signal?: AbortSignal; + }) { + return fetch(opts.url!, { + method: opts.method, + headers: opts.headers, + body: opts.body, + signal: opts.signal, + }); + }, +}; + +describe('resumable upload (system)', () => { + let server: MockResumableUploadServer; + let context: ResumableUploadContext; + let data: Buffer; + let file: string; + + before(async () => { + server = new MockResumableUploadServer(); + await server.listen(); + const root = protobuf.parse(PROTO).root; + const service = root.lookupService('test.v1.UploadService'); + service.resolveAll(); + context = { + auth: fakeAuth as any, + servicePath: `127.0.0.1:${server.port}`, + servicePort: server.port, + protocol: 'http', + rpc: service.methods.CreateUpload, + request: {name: 'test'}, + uploadPrefix: '/resumable/upload', + }; + data = crypto.randomBytes(Math.floor(5.5 * GRANULARITY)); + file = path.join(os.tmpdir(), 'gax-resumable-upload-e2e.bin'); + fs.writeFileSync(file, data); + }); + + after(async () => { + try { + fs.unlinkSync(file); + } catch { + // ignore + } + await server.close(); + }); + + it('uploads a payload over HTTP and completes the session', async () => { + const helper = new gax.ResumableUploadSession(context); + const progress: Array<{bytesUploaded: number}> = []; + await helper.start({ + uploadSource: gax.resumableSourceFromFile(file), + chunkSize: CHUNK_SIZE, + onProgress: status => { + progress.push(status); + }, + }); + const response = (await helper.finished()) as {status: string}; + + assert.strictEqual(response.status, 'done'); + assert.ok(server.received.equals(data)); + assert.deepStrictEqual( + server.commands.map(c => c.command), + ['start', 'upload', 'upload', 'upload, finalize'], + ); + assert.deepStrictEqual( + server.commands.slice(1).map(c => c.offset), + [0, 2 * GRANULARITY, 4 * GRANULARITY], + ); + assert.ok(progress.length >= 2); + }); + + it('recovers from a simulated crash using a saved session URL', async () => { + server.received = Buffer.alloc(0); + server.commands = []; + + const crashed = new gax.ResumableUploadSession(context); + const crashSource: gax.ResumableSource = { + size: data.length, + getStream: (offset = 0) => + new CrashingStream(data, 2 * CHUNK_SIZE, offset), + }; + await crashed.start({ + uploadSource: crashSource, + chunkSize: CHUNK_SIZE, + }); + await assert.rejects(crashed.finished(), /simulated process crash/); + const committed = server.received.length; + assert.strictEqual(committed, 2 * CHUNK_SIZE); + const sessionUrl = crashed.uploadUrl!; + + const resumed = new gax.ResumableUploadSession(context); + const resumedProgress: Array<{bytesUploaded: number}> = []; + await resumed.start({ + uploadSource: gax.resumableSourceFromFile(file), + resumeUrl: sessionUrl, + chunkSize: CHUNK_SIZE, + onProgress: status => { + resumedProgress.push(status); + }, + }); + const response = (await resumed.finished()) as {status: string}; + + assert.strictEqual(response.status, 'done'); + assert.ok(server.received.equals(data)); + assert.strictEqual(resumedProgress[0].bytesUploaded, committed); + }); +}); diff --git a/core/packages/gax/test/unit/resumableUpload.ts b/core/packages/gax/test/unit/resumableUpload.ts new file mode 100644 index 000000000000..627758b658db --- /dev/null +++ b/core/packages/gax/test/unit/resumableUpload.ts @@ -0,0 +1,1245 @@ +/** + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +/* eslint-disable @typescript-eslint/no-explicit-any */ + +import assert from 'assert'; +import {afterEach, describe, it} from 'mocha'; +import * as protobuf from 'protobufjs'; +import {Readable} from 'stream'; +import * as sinon from 'sinon'; + +import * as gax from '../../src'; +import {createApiCall} from '../../src/createApiCall'; +import {ResumableUploadContext} from '../../src/resumableUpload'; +import {Status} from '../../src/status'; + +const GRANULARITY = 1024 * 1024; +const SESSION_URL = 'https://example.com/upload/session-123'; + +const PROTO = ` +syntax = "proto3"; +package test.v1; +message UploadRequest { string name = 1; } +message UploadResponse { string name = 1; } +service UploadService { + rpc CreateUpload(UploadRequest) returns (UploadResponse); +} +`; + +interface MockRequestOptions { + url?: string; + method?: string; + headers?: {[name: string]: string}; + body?: string | Buffer; + signal?: AbortSignal; + responseType?: string; + timeout?: number; + validateStatus?: (status: number) => boolean; +} + +interface MockResponse { + status: number; + headers: {get(name: string): string | null}; + arrayBuffer(): Promise; +} + +function resumableUploadResponse( + status: number, + headers: {[name: string]: string}, + body = '', +): MockResponse { + const normalized: {[name: string]: string} = {}; + for (const [name, value] of Object.entries(headers)) { + normalized[name.toLowerCase()] = value; + } + return { + status, + headers: { + get(name: string): string | null { + return normalized[name.toLowerCase()] ?? null; + }, + }, + async arrayBuffer(): Promise { + return Buffer.from(body) as unknown as ArrayBuffer; + }, + }; +} + +type RequestHandler = ( + opts: MockRequestOptions, +) => MockResponse | Promise; + +function mockAuth(handler: RequestHandler) { + return { + request: sinon.stub().callsFake(async (opts: MockRequestOptions) => { + return handler(opts); + }), + }; +} + +function commandOf(opts: MockRequestOptions): string { + return opts.headers?.['x-goog-upload-command'] ?? ''; +} + +function offsetOf(opts: MockRequestOptions): number { + const value = opts.headers?.['x-goog-upload-offset']; + return value === undefined ? -1 : parseInt(value, 10); +} + +function bodyLength(opts: MockRequestOptions): number { + return opts.body === undefined ? 0 : opts.body.length; +} + +function sleep(ms: number): Promise { + return new Promise(resolve => setTimeout(resolve, ms)); +} + +interface BufferSourceFixture { + source: gax.ResumableSource; + streams: Readable[]; +} + +function bufferSource(payload: Buffer): BufferSourceFixture { + const streams: Readable[] = []; + return { + source: { + size: payload.length, + getStream: (offset?: number) => { + const start = offset ?? 0; + const stream = Readable.from([payload.subarray(start)]); + streams.push(stream); + return stream; + }, + }, + streams, + }; +} + +const root = protobuf.parse(PROTO).root; +const uploadService = root.lookupService('test.v1.UploadService'); +uploadService.resolveAll(); +const rpc = uploadService.methods.CreateUpload; + +function buildContext( + auth: any, + overrides: Partial = {}, +): ResumableUploadContext { + return { + auth, + servicePath: 'example.com', + servicePort: 443, + protocol: 'https', + rpc, + request: {name: 'test'}, + uploadPrefix: '/resumable/upload', + ...overrides, + }; +} + +describe('resumable upload', () => { + afterEach(() => { + sinon.restore(); + }); + + it('uploads a payload in chunks and resolves with the final response', async () => { + const requests: MockRequestOptions[] = []; + const auth = mockAuth(async opts => { + requests.push(opts); + const command = commandOf(opts); + if (command === 'start') { + return resumableUploadResponse(200, { + 'x-goog-upload-url': SESSION_URL, + 'x-goog-upload-status': 'active', + 'x-goog-upload-chunk-granularity': String(GRANULARITY), + }); + } + if (command === 'upload') { + return resumableUploadResponse(200, {'x-goog-upload-status': 'active'}); + } + if (command === 'upload, finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + if (command === 'finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + throw new Error(`Unexpected command: ${command}`); + }); + + const helper = new gax.ResumableUploadSession(buildContext(auth)); + const progress: Array<{bytesUploaded: number; uploadUrl: string}> = []; + const payload = Buffer.concat([ + Buffer.alloc(GRANULARITY), + Buffer.alloc(GRANULARITY), + Buffer.alloc(GRANULARITY), + ]); + const fixture = bufferSource(payload); + + await helper.start({ + uploadSource: fixture.source, + chunkSize: GRANULARITY, + onProgress: status => { + progress.push(status); + return false; + }, + }); + const response = await helper.finished(); + + assert.deepStrictEqual(response, {name: 'complete'}); + assert.strictEqual(helper.uploadUrl, SESSION_URL); + assert.deepStrictEqual( + requests.map(r => commandOf(r)), + ['start', 'upload', 'upload', 'upload', 'finalize'], + ); + assert.deepStrictEqual( + requests.slice(1).map(r => offsetOf(r)), + [0, GRANULARITY, 2 * GRANULARITY, 3 * GRANULARITY], + ); + assert.deepStrictEqual( + requests.slice(1, 4).map(r => bodyLength(r)), + [GRANULARITY, GRANULARITY, GRANULARITY], + ); + assert.strictEqual(requests[0].body, JSON.stringify({name: 'test'})); + assert.strictEqual(requests[0].headers!['x-goog-upload-offset'], undefined); + assert.deepStrictEqual(progress, [ + {bytesUploaded: GRANULARITY, uploadUrl: SESSION_URL}, + {bytesUploaded: 2 * GRANULARITY, uploadUrl: SESSION_URL}, + {bytesUploaded: 3 * GRANULARITY, uploadUrl: SESSION_URL}, + ]); + }); + + it('combines the final partial chunk with the finalize command', async () => { + const requests: MockRequestOptions[] = []; + const auth = mockAuth(async opts => { + requests.push(opts); + const command = commandOf(opts); + if (command === 'start') { + return resumableUploadResponse(200, { + 'x-goog-upload-url': SESSION_URL, + 'x-goog-upload-status': 'active', + }); + } + if (command === 'upload') { + return resumableUploadResponse(200, {'x-goog-upload-status': 'active'}); + } + if (command === 'upload, finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + if (command === 'finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + throw new Error(`Unexpected command: ${command}`); + }); + + const helper = new gax.ResumableUploadSession(buildContext(auth)); + const payload = Buffer.concat([ + Buffer.alloc(GRANULARITY), + Buffer.alloc(GRANULARITY / 2), + ]); + await helper.start({ + uploadSource: bufferSource(payload).source, + chunkSize: GRANULARITY, + }); + await helper.finished(); + + assert.deepStrictEqual( + requests.map(r => commandOf(r)), + ['start', 'upload', 'upload, finalize'], + ); + assert.deepStrictEqual( + requests.slice(1).map(r => offsetOf(r)), + [0, GRANULARITY], + ); + assert.strictEqual(bodyLength(requests[2]), GRANULARITY / 2); + }); + + it('parses host and port from the service path', async () => { + const requests: MockRequestOptions[] = []; + const auth = mockAuth(async opts => { + requests.push(opts); + const command = commandOf(opts); + if (command === 'start') { + return resumableUploadResponse(200, { + 'x-goog-upload-url': SESSION_URL, + 'x-goog-upload-status': 'active', + }); + } + if (command === 'finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + throw new Error(`Unexpected command: ${command}`); + }); + + const helper = new gax.ResumableUploadSession( + buildContext(auth, {servicePath: 'example.com:8443'}), + ); + await helper.start({ + uploadSource: bufferSource(Buffer.alloc(0)).source, + chunkSize: GRANULARITY, + }); + await helper.finished(); + + assert.strictEqual( + requests[0].url, + 'https://example.com:8443/resumable/upload', + ); + }); + + it('rounds the chunk size down to the server granularity', async () => { + const requests: MockRequestOptions[] = []; + const auth = mockAuth(async opts => { + requests.push(opts); + const command = commandOf(opts); + if (command === 'start') { + return resumableUploadResponse(200, { + 'x-goog-upload-url': SESSION_URL, + 'x-goog-upload-status': 'active', + 'x-goog-upload-chunk-granularity': String(GRANULARITY), + }); + } + if (command === 'upload') { + return resumableUploadResponse(200, {'x-goog-upload-status': 'active'}); + } + if (command === 'upload, finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + if (command === 'finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + throw new Error(`Unexpected command: ${command}`); + }); + + const helper = new gax.ResumableUploadSession(buildContext(auth)); + const payload = Buffer.concat([ + Buffer.alloc(3 * GRANULARITY), + Buffer.alloc(1), + ]); + await helper.start({ + uploadSource: bufferSource(payload).source, + chunkSize: 3.5 * GRANULARITY, + }); + await helper.finished(); + + assert.deepStrictEqual( + requests.slice(1).map(r => bodyLength(r)), + [3 * GRANULARITY, 1], + ); + }); + + it('retries transient (Category 1) errors with backoff', async () => { + let uploadAttempts = 0; + const requests: MockRequestOptions[] = []; + const auth = mockAuth(async opts => { + requests.push(opts); + const command = commandOf(opts); + if (command === 'start') { + return resumableUploadResponse(200, { + 'x-goog-upload-url': SESSION_URL, + 'x-goog-upload-status': 'active', + }); + } + if (command === 'upload') { + uploadAttempts += 1; + if (uploadAttempts === 1) { + return resumableUploadResponse(503, { + 'x-goog-upload-status': 'active', + }); + } + return resumableUploadResponse(200, {'x-goog-upload-status': 'active'}); + } + if (command === 'upload, finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + if (command === 'finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + throw new Error(`Unexpected command: ${command}`); + }); + + const helper = new gax.ResumableUploadSession(buildContext(auth)); + await helper.start({ + uploadSource: bufferSource(Buffer.alloc(GRANULARITY)).source, + chunkSize: GRANULARITY, + }); + await helper.finished(); + + assert.strictEqual(uploadAttempts, 2); + const uploads = requests.filter(r => commandOf(r) === 'upload'); + assert.ok( + Buffer.from(uploads[0].body as Buffer).equals( + Buffer.from(uploads[1].body as Buffer), + ), + ); + }); + + it('does not retry transient errors when retries are disabled', async () => { + let uploadAttempts = 0; + const auth = mockAuth(async opts => { + const command = commandOf(opts); + if (command === 'start') { + return resumableUploadResponse(200, { + 'x-goog-upload-url': SESSION_URL, + 'x-goog-upload-status': 'active', + }); + } + if (command === 'upload') { + uploadAttempts += 1; + return resumableUploadResponse(503, {'x-goog-upload-status': 'active'}); + } + throw new Error(`Unexpected command: ${command}`); + }); + + const helper = new gax.ResumableUploadSession(buildContext(auth)); + await helper.start({ + uploadSource: bufferSource(Buffer.alloc(GRANULARITY)).source, + chunkSize: GRANULARITY, + retry: null, + }); + await assert.rejects( + helper.finished(), + (err: gax.GoogleError) => err.code === Status.DEADLINE_EXCEEDED, + ); + assert.strictEqual(uploadAttempts, 1); + }); + + it('recovers from a 416 state mismatch by querying the offset', async () => { + const requests: MockRequestOptions[] = []; + let firstUpload = true; + const auth = mockAuth(async opts => { + requests.push(opts); + const command = commandOf(opts); + if (command === 'start') { + return resumableUploadResponse(200, { + 'x-goog-upload-url': SESSION_URL, + 'x-goog-upload-status': 'active', + }); + } + if (command === 'query') { + return resumableUploadResponse(200, { + 'x-goog-upload-status': 'active', + 'x-goog-upload-size-received': '0', + }); + } + if (command === 'upload') { + if (firstUpload) { + firstUpload = false; + return resumableUploadResponse(416, { + 'x-goog-upload-status': 'active', + }); + } + return resumableUploadResponse(200, {'x-goog-upload-status': 'active'}); + } + if (command === 'upload, finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + if (command === 'finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + throw new Error(`Unexpected command: ${command}`); + }); + + const helper = new gax.ResumableUploadSession(buildContext(auth)); + await helper.start({ + uploadSource: bufferSource(Buffer.alloc(GRANULARITY)).source, + chunkSize: GRANULARITY, + }); + await helper.finished(); + + assert.deepStrictEqual( + requests.map(r => commandOf(r)), + ['start', 'upload', 'query', 'upload', 'finalize'], + ); + const uploads = requests.filter(r => commandOf(r) === 'upload'); + assert.strictEqual(uploads.length, 2); + assert.ok( + Buffer.from(uploads[0].body as Buffer).equals( + Buffer.from(uploads[1].body as Buffer), + ), + ); + }); + + it('continues from the committed offset when the chunk was saved', async () => { + const requests: MockRequestOptions[] = []; + const auth = mockAuth(async opts => { + requests.push(opts); + const command = commandOf(opts); + if (command === 'start') { + return resumableUploadResponse(200, { + 'x-goog-upload-url': SESSION_URL, + 'x-goog-upload-status': 'active', + }); + } + if (command === 'query') { + return resumableUploadResponse(200, { + 'x-goog-upload-status': 'active', + 'x-goog-upload-size-received': String(GRANULARITY), + }); + } + if (command === 'upload') { + if (offsetOf(opts) === 0) { + return resumableUploadResponse(416, { + 'x-goog-upload-status': 'active', + }); + } + return resumableUploadResponse(200, {'x-goog-upload-status': 'active'}); + } + if (command === 'upload, finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + if (command === 'upload') { + return resumableUploadResponse(200, {'x-goog-upload-status': 'active'}); + } + if (command === 'finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + throw new Error(`Unexpected command: ${command}`); + }); + + const helper = new gax.ResumableUploadSession(buildContext(auth)); + const payload = Buffer.concat([ + Buffer.alloc(GRANULARITY), + Buffer.alloc(GRANULARITY), + ]); + await helper.start({ + uploadSource: bufferSource(payload).source, + chunkSize: GRANULARITY, + }); + await helper.finished(); + + const uploads = requests.filter(r => commandOf(r) === 'upload'); + assert.deepStrictEqual( + uploads.map(r => offsetOf(r)), + [0, GRANULARITY], + ); + }); + + it('treats a 200 response without X-Goog-Upload-Status as recoverable', async () => { + const requests: MockRequestOptions[] = []; + let firstUpload = true; + const auth = mockAuth(async opts => { + requests.push(opts); + const command = commandOf(opts); + if (command === 'start') { + return resumableUploadResponse(200, { + 'x-goog-upload-url': SESSION_URL, + 'x-goog-upload-status': 'active', + }); + } + if (command === 'query') { + return resumableUploadResponse(200, { + 'x-goog-upload-status': 'active', + 'x-goog-upload-size-received': '0', + }); + } + if (command === 'upload') { + if (firstUpload) { + firstUpload = false; + return resumableUploadResponse(200, {}); + } + return resumableUploadResponse(200, {'x-goog-upload-status': 'active'}); + } + if (command === 'upload, finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + if (command === 'finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + throw new Error(`Unexpected command: ${command}`); + }); + + const helper = new gax.ResumableUploadSession(buildContext(auth)); + await helper.start({ + uploadSource: bufferSource(Buffer.alloc(GRANULARITY)).source, + chunkSize: GRANULARITY, + }); + await helper.finished(); + + assert.deepStrictEqual( + requests.map(r => commandOf(r)), + ['start', 'upload', 'query', 'upload', 'finalize'], + ); + }); + + it('throws fatal (Category 3) errors without retrying', async () => { + let uploadAttempts = 0; + const auth = mockAuth(async opts => { + const command = commandOf(opts); + if (command === 'start') { + return resumableUploadResponse(200, { + 'x-goog-upload-url': SESSION_URL, + 'x-goog-upload-status': 'active', + }); + } + if (command === 'upload') { + uploadAttempts += 1; + return resumableUploadResponse(403, {'x-goog-upload-status': 'active'}); + } + throw new Error(`Unexpected command: ${command}`); + }); + + const helper = new gax.ResumableUploadSession(buildContext(auth)); + await helper.start({ + uploadSource: bufferSource(Buffer.alloc(GRANULARITY)).source, + chunkSize: GRANULARITY, + }); + await assert.rejects( + helper.finished(), + (err: gax.GoogleError) => err.code === Status.PERMISSION_DENIED, + ); + assert.strictEqual(uploadAttempts, 1); + }); + + it('resumes from a session URL by opening the source at the offset', async () => { + const requests: MockRequestOptions[] = []; + const auth = mockAuth(async opts => { + requests.push(opts); + const command = commandOf(opts); + if (command === 'start') { + return resumableUploadResponse(200, { + 'x-goog-upload-url': SESSION_URL, + 'x-goog-upload-status': 'active', + }); + } + if (command === 'query') { + return resumableUploadResponse(200, { + 'x-goog-upload-status': 'active', + 'x-goog-upload-size-received': String(2 * GRANULARITY), + }); + } + if (command === 'upload, finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + if (command === 'upload') { + return resumableUploadResponse(200, {'x-goog-upload-status': 'active'}); + } + if (command === 'finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + throw new Error(`Unexpected command: ${command}`); + }); + + const helper = new gax.ResumableUploadSession(buildContext(auth)); + const payload = Buffer.concat([ + Buffer.alloc(GRANULARITY), + Buffer.alloc(GRANULARITY), + Buffer.alloc(GRANULARITY), + ]); + await helper.start({ + uploadSource: bufferSource(payload).source, + chunkSize: GRANULARITY, + resumeUrl: SESSION_URL, + }); + const response = await helper.finished(); + + assert.deepStrictEqual(response, {name: 'complete'}); + assert.deepStrictEqual( + requests.map(r => commandOf(r)), + ['query', 'upload', 'finalize'], + ); + assert.strictEqual(offsetOf(requests[1]), 2 * GRANULARITY); + assert.strictEqual(bodyLength(requests[1]), GRANULARITY); + assert.strictEqual(offsetOf(requests[2]), 3 * GRANULARITY); + assert.strictEqual(helper.uploadUrl, SESSION_URL); + }); + + it('resumes from a committed offset that does not align with the chunk boundary', async () => { + const requests: MockRequestOptions[] = []; + const auth = mockAuth(async opts => { + requests.push(opts); + const command = commandOf(opts); + if (command === 'query') { + return resumableUploadResponse(200, { + 'x-goog-upload-status': 'active', + 'x-goog-upload-size-received': '6', + }); + } + if (command === 'upload') { + return resumableUploadResponse(200, {'x-goog-upload-status': 'active'}); + } + if (command === 'upload, finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + if (command === 'finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + throw new Error(`Unexpected command: ${command}`); + }); + + // The committed offset (6) splits the second stream chunk, so the bytes + // past the boundary must be preserved and transmitted. + const payload = Buffer.from([...Array(20).keys()]); + const helper = new gax.ResumableUploadSession(buildContext(auth)); + await helper.start({ + uploadSource: bufferSource(payload).source, + chunkSize: 8, + resumeUrl: SESSION_URL, + }); + const response = await helper.finished(); + + assert.deepStrictEqual(response, {name: 'complete'}); + assert.deepStrictEqual( + requests.map(r => commandOf(r)), + ['query', 'upload', 'upload, finalize'], + ); + const uploads = requests.filter( + r => commandOf(r) === 'upload' || commandOf(r) === 'upload, finalize', + ); + assert.deepStrictEqual( + uploads.map(r => offsetOf(r)), + [6, 14], + ); + const transmitted = Buffer.concat( + uploads.map(r => Buffer.from(r.body as Buffer)), + ); + assert.ok(transmitted.equals(payload.subarray(6))); + }); + + it('skips ahead to the server offset without dropping buffered bytes', async () => { + const requests: MockRequestOptions[] = []; + const auth = mockAuth(async opts => { + requests.push(opts); + const command = commandOf(opts); + if (command === 'start') { + return resumableUploadResponse(200, { + 'x-goog-upload-url': SESSION_URL, + 'x-goog-upload-status': 'active', + }); + } + if (command === 'query') { + return resumableUploadResponse(200, { + 'x-goog-upload-status': 'active', + 'x-goog-upload-size-received': '12', + }); + } + if (command === 'upload') { + if (offsetOf(opts) === 0) { + return resumableUploadResponse(416, { + 'x-goog-upload-status': 'active', + }); + } + return resumableUploadResponse(200, {'x-goog-upload-status': 'active'}); + } + if (command === 'upload, finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + if (command === 'finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + throw new Error(`Unexpected command: ${command}`); + }); + + // The first 12-byte stream chunk overflows the 8-byte chunk size, so the + // remainder (bytes 8-12) sits in the buffer when the server reports it + // committed 12 bytes; the skip must consume the buffer head, not the + // next stream chunk. + const payload = Buffer.from([...Array(20).keys()]); + const helper = new gax.ResumableUploadSession(buildContext(auth)); + await helper.start({ + uploadSource: bufferSource(payload).source, + chunkSize: 8, + }); + const response = await helper.finished(); + + assert.deepStrictEqual(response, {name: 'complete'}); + assert.deepStrictEqual( + requests.map(r => commandOf(r)), + ['start', 'upload', 'query', 'upload', 'finalize'], + ); + const uploads = requests.filter( + r => + (commandOf(r) === 'upload' && offsetOf(r) !== 0) || + commandOf(r) === 'finalize', + ); + assert.deepStrictEqual( + uploads.map(r => offsetOf(r)), + [12, 20], + ); + const transmitted = Buffer.concat( + uploads.map(r => + r.body === undefined ? Buffer.alloc(0) : Buffer.from(r.body as Buffer), + ), + ); + assert.ok(transmitted.equals(payload.subarray(12))); + }); + + it('aborts when the global deadline is exceeded', async () => { + const auth = mockAuth(async opts => { + const command = commandOf(opts); + if (command === 'start') { + await sleep(20); + return resumableUploadResponse(200, { + 'x-goog-upload-url': SESSION_URL, + 'x-goog-upload-status': 'active', + }); + } + throw new Error(`Unexpected command: ${command}`); + }); + + const helper = new gax.ResumableUploadSession(buildContext(auth)); + const finished = helper.finished(); + const rejection = assert.rejects( + finished, + (err: gax.GoogleError) => err.code === Status.DEADLINE_EXCEEDED, + ); + await helper.start({ + uploadSource: bufferSource(Buffer.alloc(10)).source, + globalDeadlineMs: 5, + uploadSize: 10 * 1024 * 1024 * 1024, + }); + await rejection; + }); + + it('rejects with DEADLINE_EXCEEDED when the stream stalls past the deadline', async () => { + const auth = mockAuth(async opts => { + const command = commandOf(opts); + if (command === 'start') { + return resumableUploadResponse(200, { + 'x-goog-upload-url': SESSION_URL, + 'x-goog-upload-status': 'active', + }); + } + throw new Error(`Unexpected command: ${command}`); + }); + + const helper = new gax.ResumableUploadSession(buildContext(auth)); + // The stream never yields data, so the transmission loop blocks on the + // stream read; only the session deadline can end the upload. + const stalled = new Readable({read() {}}); + const source: gax.ResumableSource = { + size: 0, + getStream: () => stalled, + }; + await helper.start({ + uploadSource: source, + globalDeadlineMs: 50, + }); + await assert.rejects( + helper.finished(), + (err: gax.GoogleError) => err.code === Status.DEADLINE_EXCEEDED, + ); + assert.ok(stalled.destroyed); + }); + + it('cancels an in-flight upload and notifies the server', async () => { + const requests: MockRequestOptions[] = []; + const auth = mockAuth(opts => { + requests.push(opts); + const command = commandOf(opts); + if (command === 'start') { + return resumableUploadResponse(200, { + 'x-goog-upload-url': SESSION_URL, + 'x-goog-upload-status': 'active', + }); + } + if (command === 'upload') { + // Never resolves until the upload is cancelled. + return new Promise(() => {}); + } + if (command === 'cancel') { + return resumableUploadResponse(200, { + 'x-goog-upload-status': 'cancelled', + }); + } + throw new Error(`Unexpected command: ${command}`); + }); + + const helper = new gax.ResumableUploadSession(buildContext(auth)); + const stream = new Readable({read() {}}); + stream.push(Buffer.alloc(GRANULARITY)); + const source: gax.ResumableSource = { + size: GRANULARITY, + getStream: () => stream, + }; + await helper.start({uploadSource: source, chunkSize: GRANULARITY}); + + helper.cancel(); + await assert.rejects( + helper.finished(), + (err: gax.GoogleError) => err.code === Status.CANCELLED, + ); + assert.ok( + requests.some(r => commandOf(r) === 'cancel'), + 'expected a cancel command to be sent', + ); + assert.ok(stream.destroyed); + }); + + it('resolves the GAPIC method call with a ResumableUploadSession object', async () => { + const auth = mockAuth(() => { + throw new Error('No request should be made before start()'); + }); + const descriptor = new gax.ResumableUploadDescriptor('/resumable/upload'); + const settings = new gax.CallSettings(); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const func = async () => ({cancel() {}}); + const apiCall = createApiCall( + // eslint-disable-next-line @typescript-eslint/no-explicit-any + Promise.resolve(func as any), + settings, + descriptor, + ); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const result = (await (apiCall( + {name: 'test'}, + {resumableUpload: buildContext(auth)}, + ) as Promise)) as any; + assert.ok(result[0] instanceof gax.ResumableUploadSession); + assert.strictEqual( + // eslint-disable-next-line @typescript-eslint/no-explicit-any + (result[0] as any).context.uploadPrefix, + '/resumable/upload', + ); + }); + + it('rejects the GAPIC method call when the transport context is missing', async () => { + const descriptor = new gax.ResumableUploadDescriptor(); + const settings = new gax.CallSettings(); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const func = async () => ({cancel() {}}); + const apiCall = createApiCall( + // eslint-disable-next-line @typescript-eslint/no-explicit-any + Promise.resolve(func as any), + settings, + descriptor, + ); + await assert.rejects( + // eslint-disable-next-line @typescript-eslint/no-explicit-any + apiCall({} as any, {} as any) as Promise, + /resumable upload transport context/, + ); + }); + + it('exposes the granularity-rounded chunk size on the session', async () => { + const auth = mockAuth(async opts => { + const command = commandOf(opts); + if (command === 'start') { + return resumableUploadResponse(200, { + 'x-goog-upload-url': SESSION_URL, + 'x-goog-upload-status': 'active', + 'x-goog-upload-chunk-granularity': String(GRANULARITY), + }); + } + if (command === 'upload') { + return resumableUploadResponse(200, {'x-goog-upload-status': 'active'}); + } + if (command === 'upload, finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + throw new Error(`Unexpected command: ${command}`); + }); + + const helper = new gax.ResumableUploadSession(buildContext(auth)); + await helper.start({ + uploadSource: bufferSource(Buffer.alloc(GRANULARITY + 1)).source, + chunkSize: 3.5 * GRANULARITY, + }); + assert.strictEqual(helper.chunkSize, 3 * GRANULARITY); + await helper.finished(); + }); + + it('reconciles a start response that is missing X-Goog-Upload-Status', async () => { + const requests: MockRequestOptions[] = []; + const auth = mockAuth(async opts => { + requests.push(opts); + const command = commandOf(opts); + if (command === 'start') { + return resumableUploadResponse(200, { + 'x-goog-upload-url': SESSION_URL, + 'x-goog-upload-chunk-granularity': String(GRANULARITY), + }); + } + if (command === 'query') { + return resumableUploadResponse(200, { + 'x-goog-upload-status': 'active', + 'x-goog-upload-size-received': '0', + }); + } + if (command === 'upload') { + return resumableUploadResponse(200, {'x-goog-upload-status': 'active'}); + } + if (command === 'upload, finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + throw new Error(`Unexpected command: ${command}`); + }); + + const helper = new gax.ResumableUploadSession(buildContext(auth)); + await helper.start({ + uploadSource: bufferSource(Buffer.alloc(GRANULARITY + 1)).source, + chunkSize: GRANULARITY, + }); + await helper.finished(); + + assert.deepStrictEqual( + requests.map(r => commandOf(r)), + ['start', 'query', 'upload', 'upload, finalize'], + ); + }); + + it('treats a missing status header during recovery as fatal', async () => { + const requests: MockRequestOptions[] = []; + const auth = mockAuth(async opts => { + requests.push(opts); + const command = commandOf(opts); + if (command === 'start') { + return resumableUploadResponse(200, { + 'x-goog-upload-url': SESSION_URL, + 'x-goog-upload-status': 'active', + }); + } + if (command === 'upload') { + return resumableUploadResponse(200, {}); + } + if (command === 'query') { + return resumableUploadResponse(200, {}); + } + throw new Error(`Unexpected command: ${command}`); + }); + + const helper = new gax.ResumableUploadSession(buildContext(auth)); + await helper.start({ + uploadSource: bufferSource(Buffer.alloc(GRANULARITY)).source, + chunkSize: GRANULARITY, + }); + await assert.rejects(helper.finished(), /header while in recovery/i); + assert.deepStrictEqual( + requests.map(r => commandOf(r)), + ['start', 'upload', 'query'], + ); + }); + + it('sends start headers only on the start request and applies timeouts everywhere', async () => { + const requests: MockRequestOptions[] = []; + const auth = mockAuth(async opts => { + requests.push(opts); + const command = commandOf(opts); + if (command === 'start') { + return resumableUploadResponse(200, { + 'x-goog-upload-url': SESSION_URL, + 'x-goog-upload-status': 'active', + }); + } + if (command === 'upload') { + return resumableUploadResponse(200, {'x-goog-upload-status': 'active'}); + } + if (command === 'upload, finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + throw new Error(`Unexpected command: ${command}`); + }); + + const helper = new gax.ResumableUploadSession(buildContext(auth)); + await helper.start({ + uploadSource: bufferSource(Buffer.alloc(GRANULARITY + 1)).source, + chunkSize: GRANULARITY, + timeout: 4321, + startHeaders: {'developer-token': 'token-123'}, + }); + await helper.finished(); + + assert.strictEqual(requests[0].headers!['developer-token'], 'token-123'); + for (const request of requests.slice(1)) { + assert.strictEqual(request.headers!['developer-token'], undefined); + assert.strictEqual(request.timeout, 4321); + } + }); + + it('recovers from a stalled upload request by querying and reopening the source', async () => { + const requests: MockRequestOptions[] = []; + let uploadAttempts = 0; + const auth = mockAuth(opts => { + requests.push(opts); + const command = commandOf(opts); + if (command === 'start') { + return resumableUploadResponse(200, { + 'x-goog-upload-url': SESSION_URL, + 'x-goog-upload-status': 'active', + }); + } + if (command === 'upload') { + uploadAttempts += 1; + if (uploadAttempts === 1) { + return new Promise((resolve, reject) => { + opts.signal?.addEventListener('abort', () => { + reject(new Error('aborted by stall detection')); + }); + // Keep `resolve` referenced so the promise stays pending until + // the stall timer aborts the request. + void resolve; + }); + } + return resumableUploadResponse(200, {'x-goog-upload-status': 'active'}); + } + if (command === 'query') { + return resumableUploadResponse(200, { + 'x-goog-upload-status': 'active', + 'x-goog-upload-size-received': '0', + }); + } + if (command === 'upload, finalize') { + return resumableUploadResponse( + 200, + {'x-goog-upload-status': 'final'}, + JSON.stringify({name: 'complete'}), + ); + } + throw new Error(`Unexpected command: ${command}`); + }); + + const payload = Buffer.alloc(GRANULARITY + 1); + const fixture = bufferSource(payload); + const helper = new gax.ResumableUploadSession(buildContext(auth)); + await helper.start({ + uploadSource: fixture.source, + chunkSize: GRANULARITY, + stallTimeoutMs: 25, + }); + await helper.finished(); + + assert.deepStrictEqual( + requests.map(r => commandOf(r)), + ['start', 'upload', 'query', 'upload', 'upload, finalize'], + ); + assert.ok(fixture.streams.length >= 2); + }); + + it('rejects resumption when the source cannot be re-opened at the offset', async () => { + const auth = mockAuth(async opts => { + const command = commandOf(opts); + if (command === 'query') { + return resumableUploadResponse(200, { + 'x-goog-upload-status': 'active', + 'x-goog-upload-size-received': String(GRANULARITY), + }); + } + throw new Error(`Unexpected command: ${command}`); + }); + + const payload = Buffer.alloc(2 * GRANULARITY); + const source: gax.ResumableSource = { + size: payload.length, + getStream: (offset = 0) => { + if (offset > 0) { + throw new Error('source cannot seek'); + } + return Readable.from([payload.subarray(offset)]); + }, + }; + const helper = new gax.ResumableUploadSession(buildContext(auth)); + await helper.start({ + uploadSource: source, + chunkSize: GRANULARITY, + resumeUrl: SESSION_URL, + }); + await assert.rejects(helper.finished(), /source cannot seek/); + }); +}); From e3a636355ee95cd3ce50b63d6cdc2ce476a20926 Mon Sep 17 00:00:00 2001 From: feywind <57276408+feywind@users.noreply.github.com> Date: Thu, 10 Sep 2026 18:36:57 -0400 Subject: [PATCH 2/3] fix(gax): satisfy Buffer typings and lint under newer TypeScript CI type-checks this package with the repository root TypeScript (^5.8.3, currently 5.9.3) rather than the 5.8.3 pinned here, which tightens the Buffer generics: - annotate the transmission buffer as Buffer instead of letting Buffer.alloc() narrow it to Buffer, which readNextChunk() and skipBytes() cannot assign to - copy forwarded bodies into Uint8Array.from() in the system test, since fetch()'s BodyInit does not accept Buffer - merge the two 'fs' imports in resumableSourceFromFile.ts Found by the monorepo linter (bin/linter.mjs) on #9287. --- core/packages/gax/src/resumableSourceFromFile.ts | 3 +-- core/packages/gax/src/resumableUpload.ts | 5 ++++- core/packages/gax/test/system-test/resumableUpload.ts | 8 +++++++- 3 files changed, 12 insertions(+), 4 deletions(-) diff --git a/core/packages/gax/src/resumableSourceFromFile.ts b/core/packages/gax/src/resumableSourceFromFile.ts index 0c6b8cc672c2..09ea8923ea94 100644 --- a/core/packages/gax/src/resumableSourceFromFile.ts +++ b/core/packages/gax/src/resumableSourceFromFile.ts @@ -14,8 +14,7 @@ * limitations under the License. */ -import {createReadStream} from 'fs'; -import {statSync} from 'fs'; +import {createReadStream, statSync} from 'fs'; import {ResumableSource} from './resumableUpload'; diff --git a/core/packages/gax/src/resumableUpload.ts b/core/packages/gax/src/resumableUpload.ts index 1aff192c44b7..c909d26c279e 100644 --- a/core/packages/gax/src/resumableUpload.ts +++ b/core/packages/gax/src/resumableUpload.ts @@ -725,7 +725,10 @@ export class ResumableUploadSession { private async runTransmission(sessionUrl: string): Promise { try { - let buffer = Buffer.alloc(0); + // Annotated rather than inferred: Buffer.alloc() narrows to + // Buffer, while the remainder returned by readNextChunk() + // and skipBytes() is the wider Buffer. + let buffer: Buffer = Buffer.alloc(0); let offset = this.committedBytes_; let previousChunk: Buffer | null = null; let response: {} | null = null; diff --git a/core/packages/gax/test/system-test/resumableUpload.ts b/core/packages/gax/test/system-test/resumableUpload.ts index 6ec61014824e..0b95eb2a5409 100644 --- a/core/packages/gax/test/system-test/resumableUpload.ts +++ b/core/packages/gax/test/system-test/resumableUpload.ts @@ -179,10 +179,16 @@ const fakeAuth = { body?: string | Buffer; signal?: AbortSignal; }) { + // fetch()'s BodyInit does not accept Buffer, so copy the + // bytes into a plain Uint8Array before forwarding them. + const body = + typeof opts.body === 'string' || opts.body === undefined + ? opts.body + : Uint8Array.from(opts.body); return fetch(opts.url!, { method: opts.method, headers: opts.headers, - body: opts.body, + body, signal: opts.signal, }); }, From 0f4ee8d86870858c03d45130856367b25a0db104 Mon Sep 17 00:00:00 2001 From: feywind <57276408+feywind@users.noreply.github.com> Date: Thu, 10 Sep 2026 18:53:49 -0400 Subject: [PATCH 3/3] fix(gax): correct resumable upload docs and dead deadline scaling - The client-libraries.md examples passed `uploadStream: dataStream`, but the session takes `uploadSource` (a `ResumableSource`) and `dataStream` was never defined, so the snippets would not compile. Use `client.getResumableSource()` in both examples, and correct the helper name to `ResumableUploadSession`. - `DEFAULT_UPLOAD_RATE_BYTES_PER_MS` is bytes per millisecond but held `5 * 1024 * 1024`, i.e. ~5 GiB/s rather than the documented ~5 MiB/s. Since `computeGlobalDeadlineMs` combines the scaled value with `Math.max` against the 10 minute default, size-based scaling could not engage below a ~3 TB payload. Express the documented rate per millisecond instead. --- core/packages/gax/client-libraries.md | 12 ++++++------ core/packages/gax/src/resumableUpload.ts | 2 +- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/core/packages/gax/client-libraries.md b/core/packages/gax/client-libraries.md index d66025315cd4..79a573d09e1a 100644 --- a/core/packages/gax/client-libraries.md +++ b/core/packages/gax/client-libraries.md @@ -154,14 +154,14 @@ const [response] = await client.sampleMethod(request, options); Some APIs expose methods that upload large payloads through the resumable upload protocol. For these methods, the client method no longer returns the response directly; it returns a -[`ResumableUpload`](https://googleapis.dev/nodejs/google-gax/latest/classes/ResumableUpload.html) -helper. Call `start()` with a `NodeJS.ReadableStream` and await `finished()` -for the final response: +[`ResumableUploadSession`](https://googleapis.dev/nodejs/google-gax/latest/classes/ResumableUploadSession.html) +helper. Call `start()` with a `ResumableSource` (see `getResumableSource()` +below) and await `finished()` for the final response: ```ts const helper = await client.createResumableUpload(request); await helper.start({ - uploadStream: dataStream, + uploadSource: client.getResumableSource(filePath), chunkSize: 8 * 1024 * 1024, // 8MB chunks onProgress: status => { console.log(`Committed ${status.bytesUploaded} bytes to ${status.uploadUrl}`); @@ -173,12 +173,12 @@ const response = await helper.finished(); The session URL is available as `helper.uploadUrl` once the upload has started. Save it if you need to resume the upload later — for example after a process crash or network drop. To resume, pass the saved URL to `start()` on a -new helper, along with a fresh stream of the same payload: +new helper, along with a new source for the same payload: ```ts const helper = await client.createResumableUpload(); await helper.start({ - uploadStream: dataStream, + uploadSource: client.getResumableSource(filePath), resumeUrl: savedUploadUrl, }); const response = await helper.finished(); diff --git a/core/packages/gax/src/resumableUpload.ts b/core/packages/gax/src/resumableUpload.ts index c909d26c279e..980285c75a8e 100644 --- a/core/packages/gax/src/resumableUpload.ts +++ b/core/packages/gax/src/resumableUpload.ts @@ -52,7 +52,7 @@ export const DEFAULT_PER_REQUEST_TIMEOUT_MS = 60 * 1000; export const DEFAULT_STALL_TIMEOUT_MS = 15 * 1000; // Assumed sustained upload throughput, in bytes per millisecond, used to // scale the global deadline when `uploadSize` is provided (~5 MB/s). -const DEFAULT_UPLOAD_RATE_BYTES_PER_MS = 5 * 1024 * 1024; +const DEFAULT_UPLOAD_RATE_BYTES_PER_MS = (5 * 1024 * 1024) / 1000; // Resumable upload protocol headers. const UPLOAD_PROTOCOL_HEADER = 'x-goog-upload-protocol';