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 036e922b61d4a4cd33aa40bfcb2df5430f542d22 Mon Sep 17 00:00:00 2001 From: feywind <57276408+feywind@users.noreply.github.com> Date: Thu, 10 Sep 2026 15:14:44 -0400 Subject: [PATCH 2/3] test(gax): add showcase harness for resumable uploads Adds an end-to-end harness for the gapic-showcase ResumableUploadService: a checked-in generated client, a sample that uploads a local file through a resumable session, and a run.sh that downloads the showcase server, compiles the client against the local google-gax checkout and runs the sample. Also documents the harness in test/README.md and ignores the protos it generates. --- core/packages/gax/.gitignore | 1 + core/packages/gax/test/README.md | 13 + .../test/showcase-resumable-upload/README.md | 28 ++ .../client/package.json | 28 ++ .../showcase/v1beta1/resumable_upload.proto | 47 ++ .../client/src/index.ts | 25 + .../client/src/v1beta1/index.ts | 19 + .../resumable_upload_service_client.ts | 436 ++++++++++++++++++ ...esumable_upload_service_client_config.json | 30 ++ .../resumable_upload_service_proto_list.json | 3 + .../client/tsconfig.json | 18 + .../gax/test/showcase-resumable-upload/run.sh | 113 +++++ .../test/showcase-resumable-upload/sample.js | 79 ++++ 13 files changed, 840 insertions(+) create mode 100644 core/packages/gax/test/showcase-resumable-upload/README.md create mode 100644 core/packages/gax/test/showcase-resumable-upload/client/package.json create mode 100644 core/packages/gax/test/showcase-resumable-upload/client/protos/google/showcase/v1beta1/resumable_upload.proto create mode 100644 core/packages/gax/test/showcase-resumable-upload/client/src/index.ts create mode 100644 core/packages/gax/test/showcase-resumable-upload/client/src/v1beta1/index.ts create mode 100644 core/packages/gax/test/showcase-resumable-upload/client/src/v1beta1/resumable_upload_service_client.ts create mode 100644 core/packages/gax/test/showcase-resumable-upload/client/src/v1beta1/resumable_upload_service_client_config.json create mode 100644 core/packages/gax/test/showcase-resumable-upload/client/src/v1beta1/resumable_upload_service_proto_list.json create mode 100644 core/packages/gax/test/showcase-resumable-upload/client/tsconfig.json create mode 100755 core/packages/gax/test/showcase-resumable-upload/run.sh create mode 100644 core/packages/gax/test/showcase-resumable-upload/sample.js diff --git a/core/packages/gax/.gitignore b/core/packages/gax/.gitignore index 14c3dc31fdc0..4d28f76e1bdc 100644 --- a/core/packages/gax/.gitignore +++ b/core/packages/gax/.gitignore @@ -23,3 +23,4 @@ dist/ *.tgz **/*.tgz test/showcase-echo-client/protos/protos.* +test/showcase-resumable-upload/client/protos/protos.* diff --git a/core/packages/gax/test/README.md b/core/packages/gax/test/README.md index ecd8e18fcce2..17154672f5db 100644 --- a/core/packages/gax/test/README.md +++ b/core/packages/gax/test/README.md @@ -67,6 +67,19 @@ The following steps will regenerate new Echo and Sequence clients from the lates 1. Once you have verified that nothing is broken, commit these changes and make a PR from the `regenerate-showcase-client` branch in your fork to the main gax-nodejs repo. +## [showcase-resumable-upload](./showcase-resumable-upload/) +### About +A small end-to-end harness for the resumable upload API. It downloads the +[gapic-showcase](https://github.com/googleapis/gapic-showcase) server, compiles +the checked-in generated client for the showcase `ResumableUploadService`, and runs +[`sample.js`](./showcase-resumable-upload/sample.js) against it: + +```sh +./test/showcase-resumable-upload/run.sh +``` + +See the [harness README](./showcase-resumable-upload/README.md) for details. + #### Update the showcase server See [showcase server maintenance info](#maintenance-2) for more details. diff --git a/core/packages/gax/test/showcase-resumable-upload/README.md b/core/packages/gax/test/showcase-resumable-upload/README.md new file mode 100644 index 000000000000..b2bcad05fb8a --- /dev/null +++ b/core/packages/gax/test/showcase-resumable-upload/README.md @@ -0,0 +1,28 @@ +# Showcase resumable upload harness + +This directory contains a small end-to-end example of the resumable upload +support added in the `scotty-1` work: + +* `client/` — a generated `ResumableUploadServiceClient` for the real + [gapic-showcase](https://github.com/googleapis/gapic-showcase) + `ResumableUploadService`, produced by the generator in this repo with + `--resumable_upload_methods=ResumableUploadService.UploadMedia`. +* `sample.js` — example code using `client.uploadMedia()`, + `client.getResumableSource()`, and `session.start()`. +* `run.sh` — downloads/starts a gapic-showcase server, builds the local + google-gax checkout and the generated client, then runs `sample.js`. + +Run it from anywhere: + +```sh +core/packages/gax/test/showcase-resumable-upload/run.sh +``` + +It assumes the monorepo dependencies have been installed (so `google-gax` and +`gapic-tools` can be compiled locally) and requires network access the first +time it downloads the showcase server binary. + +The default showcase version is `0.43.1`, the first release line that includes +the resumable upload service middleware. Set `SHOWCASE_VERSION`, `SHOWCASE_BIN` +(to reuse an already-downloaded binary), `SHOWCASE_PORT`, or `UPLOAD_FILE` to +override pieces of the run. diff --git a/core/packages/gax/test/showcase-resumable-upload/client/package.json b/core/packages/gax/test/showcase-resumable-upload/client/package.json new file mode 100644 index 000000000000..41484e192bf8 --- /dev/null +++ b/core/packages/gax/test/showcase-resumable-upload/client/package.json @@ -0,0 +1,28 @@ +{ + "name": "showcase-resumable-upload-client", + "version": "0.1.0", + "description": "Generated client for the gapic-showcase ResumableUploadService, used by the showcase resumable upload test harness.", + "license": "Apache-2.0", + "author": "Google LLC", + "main": "build/src/index.js", + "files": [ + "build/src", + "build/protos" + ], + "scripts": { + "compile": "tsc -p . && cp -r protos build/ && minifyProtoJson", + "compile-protos": "compileProtos src", + "prepare": "npm run compile-protos && npm run compile" + }, + "dependencies": { + "google-gax": "^6.0.0" + }, + "devDependencies": { + "@types/node": "^22.18.12", + "gapic-tools": "^2.0.0", + "typescript": "^5.8.3" + }, + "engines": { + "node": ">=18" + } +} diff --git a/core/packages/gax/test/showcase-resumable-upload/client/protos/google/showcase/v1beta1/resumable_upload.proto b/core/packages/gax/test/showcase-resumable-upload/client/protos/google/showcase/v1beta1/resumable_upload.proto new file mode 100644 index 000000000000..0f8c81a31545 --- /dev/null +++ b/core/packages/gax/test/showcase-resumable-upload/client/protos/google/showcase/v1beta1/resumable_upload.proto @@ -0,0 +1,47 @@ +// 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 +// +// https://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. + +syntax = "proto3"; + +package google.showcase.v1beta1; + +import "google/api/annotations.proto"; +import "google/api/client.proto"; + +option go_package = "github.com/googleapis/gapic-showcase/server/genproto"; +option java_package = "com.google.showcase.v1beta1"; +option java_multiple_files = true; +option ruby_package = "Google::Showcase::V1beta1"; + +// A service showcasing universal resumable upload protocol support. +service ResumableUploadService { + option (google.api.default_host) = "localhost:7469"; + + // A method with media_upload annotation enabled. + rpc UploadMedia(UploadMediaRequest) returns (UploadMediaResponse) { + option (google.api.http) = { + post: "/v1beta1/files:upload" + body: "*" + }; + } +} + +message UploadMediaRequest { + string name = 1; +} + +message UploadMediaResponse { + string name = 1; + int64 size = 2; +} diff --git a/core/packages/gax/test/showcase-resumable-upload/client/src/index.ts b/core/packages/gax/test/showcase-resumable-upload/client/src/index.ts new file mode 100644 index 000000000000..ea57097f09f5 --- /dev/null +++ b/core/packages/gax/test/showcase-resumable-upload/client/src/index.ts @@ -0,0 +1,25 @@ +// 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 +// +// https://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. +// +// ** This file is automatically generated by gapic-generator-typescript. ** +// ** https://github.com/googleapis/gapic-generator-typescript ** +// ** All changes to this file may be overwritten. ** + +import * as v1beta1 from './v1beta1'; +const ResumableUploadServiceClient = v1beta1.ResumableUploadServiceClient; +type ResumableUploadServiceClient = v1beta1.ResumableUploadServiceClient; +export { v1beta1, ResumableUploadServiceClient }; +export default { v1beta1, ResumableUploadServiceClient }; +import * as protos from '../protos/protos'; +export { protos }; diff --git a/core/packages/gax/test/showcase-resumable-upload/client/src/v1beta1/index.ts b/core/packages/gax/test/showcase-resumable-upload/client/src/v1beta1/index.ts new file mode 100644 index 000000000000..f092ac6ac16e --- /dev/null +++ b/core/packages/gax/test/showcase-resumable-upload/client/src/v1beta1/index.ts @@ -0,0 +1,19 @@ +// 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 +// +// https://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. +// +// ** This file is automatically generated by gapic-generator-typescript. ** +// ** https://github.com/googleapis/gapic-generator-typescript ** +// ** All changes to this file may be overwritten. ** + +export { ResumableUploadServiceClient } from './resumable_upload_service_client'; diff --git a/core/packages/gax/test/showcase-resumable-upload/client/src/v1beta1/resumable_upload_service_client.ts b/core/packages/gax/test/showcase-resumable-upload/client/src/v1beta1/resumable_upload_service_client.ts new file mode 100644 index 000000000000..857e947473ee --- /dev/null +++ b/core/packages/gax/test/showcase-resumable-upload/client/src/v1beta1/resumable_upload_service_client.ts @@ -0,0 +1,436 @@ +// 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 +// +// https://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. +// +// ** This file is automatically generated by gapic-generator-typescript. ** +// ** https://github.com/googleapis/gapic-generator-typescript ** +// ** All changes to this file may be overwritten. ** + +/* global window */ +import type * as gax from 'google-gax'; +import type { + Callback, + CallOptions, + Descriptors, + ClientOptions, +} from 'google-gax'; + +import * as protos from '../../protos/protos'; +import jsonProtos = require('../../protos/protos.json'); +import { loggingUtils as logging, decodeAnyProtosInArray } from 'google-gax'; + +/** + * Client JSON configuration object, loaded from + * `src/v1beta1/resumable_upload_service_client_config.json`. + * This file defines retry strategy and timeouts for all API methods in this library. + */ +import * as gapicConfig from './resumable_upload_service_client_config.json'; +const version = require('../../../package.json').version; + +/** + * A service showcasing universal resumable upload protocol support. + * @class + * @memberof v1beta1 + */ +export class ResumableUploadServiceClient { + private _terminated = false; + private _opts: ClientOptions; + private _providedCustomServicePath: boolean; + private _gaxModule: typeof gax | typeof gax.fallback; + private _gaxGrpc: gax.GrpcClient | gax.fallback.GrpcClient; + private _fallbackRest?: gax.GrpcClient | gax.fallback.GrpcClient; + private _protos: {}; + private _defaults: { [method: string]: gax.CallSettings }; + private _universeDomain: string; + private _servicePath: string; + private _log = logging.log('showcase'); + + auth: gax.GoogleAuth; + descriptors: Descriptors = { + page: {}, + stream: {}, + longrunning: {}, + batching: {}, + resumableUpload: {}, + }; + warn: (code: string, message: string, warnType?: string) => void; + innerApiCalls: { [name: string]: Function }; + resumableUploadServiceStub?: Promise<{ [name: string]: Function }>; + + /** + * Construct an instance of ResumableUploadServiceClient. + * + * @param {object} [options] - The configuration object. + * The options accepted by the constructor are described in detail + * in [this document](https://github.com/googleapis/gax-nodejs/blob/main/client-libraries.md#creating-the-client-instance). + * The common options are: + * @param {object} [options.credentials] - Credentials object. + * @param {string} [options.credentials.client_email] + * @param {string} [options.credentials.private_key] + * @param {string} [options.email] - Account email address. Required when + * using a .pem or .p12 keyFilename. + * @param {string} [options.keyFilename] - Full path to the a .json, .pem, or + * .p12 key downloaded from the Google Developers Console. If you provide + * a path to a JSON file, the projectId option below is not necessary. + * NOTE: .pem and .p12 require you to specify options.email as well. + * @param {number} [options.port] - The port on which to connect to + * the remote host. + * @param {string} [options.projectId] - The project ID from the Google + * Developer's Console, e.g. 'grape-spaceship-123'. We will also check + * the environment variable GCLOUD_PROJECT for your project ID. If your + * app is running in an environment which supports + * {@link https://cloud.google.com/docs/authentication/application-default-credentials Application Default Credentials}, + * your project ID will be detected automatically. + * @param {string} [options.apiEndpoint] - The domain name of the + * API remote host. + * @param {gax.ClientConfig} [options.clientConfig] - Client configuration override. + * Follows the structure of {@link gapicConfig}. + * @param {boolean} [options.fallback] - Use HTTP/1.1 REST mode. + * For more information, please check the + * {@link https://github.com/googleapis/gax-nodejs/blob/main/client-libraries.md#http11-rest-api-mode documentation}. + * @param {gax} [gaxInstance]: loaded instance of `google-gax`. Useful if you + * need to avoid loading the default gRPC version and want to use the fallback + * HTTP implementation. Load only fallback version and pass it to the constructor: + * ``` + * const gax = require('google-gax/build/src/fallback'); // avoids loading google-gax with gRPC + * const client = new ResumableUploadServiceClient({fallback: true}, gax); + * ``` + */ + constructor( + opts?: ClientOptions, + gaxInstance?: typeof gax | typeof gax.fallback, + ) { + // Ensure that options include all the required fields. + const staticMembers = this + .constructor as typeof ResumableUploadServiceClient; + if ( + opts?.universe_domain && + opts?.universeDomain && + opts?.universe_domain !== opts?.universeDomain + ) { + throw new Error( + 'Please set either universe_domain or universeDomain, but not both.', + ); + } + const universeDomainEnvVar = + typeof process === 'object' && typeof process.env === 'object' + ? process.env['GOOGLE_CLOUD_UNIVERSE_DOMAIN'] + : undefined; + this._universeDomain = + opts?.universeDomain ?? + opts?.universe_domain ?? + universeDomainEnvVar ?? + 'googleapis.com'; + this._servicePath = 'localhost'; + const servicePath = + opts?.servicePath || opts?.apiEndpoint || this._servicePath; + this._providedCustomServicePath = !!( + opts?.servicePath || opts?.apiEndpoint + ); + const port = opts?.port || staticMembers.port; + const clientConfig = opts?.clientConfig ?? {}; + const fallback = + opts?.fallback ?? + (typeof window !== 'undefined' && typeof window?.fetch === 'function'); + opts = Object.assign({ servicePath, port, clientConfig, fallback }, opts); + + // If scopes are unset in options and we're connecting to a non-default endpoint, set scopes just in case. + if (servicePath !== this._servicePath && !('scopes' in opts)) { + opts['scopes'] = staticMembers.scopes; + } + + // Load google-gax module synchronously if needed + if (!gaxInstance) { + gaxInstance = require('google-gax') as typeof gax; + } + + // Choose either gRPC or proto-over-HTTP implementation of google-gax. + this._gaxModule = opts.fallback ? gaxInstance.fallback : gaxInstance; + + // Create a `gaxGrpc` object, with any grpc-specific options sent to the client. + this._gaxGrpc = new this._gaxModule.GrpcClient(opts); + + // This service contains resumable upload methods, which are HTTPS-only. + // Make sure the REST transport is available even when the client was + // configured for gRPC. + this._fallbackRest = opts.fallback + ? this._gaxGrpc + : new gaxInstance.fallback.GrpcClient({ ...opts, fallback: true }); + + // Save options to use in initialize() method. + this._opts = opts; + + // Save the auth object to the client, for use by other methods. + this.auth = this._gaxGrpc.auth as gax.GoogleAuth; + + // Set useJWTAccessWithScope on the auth object. + this.auth.useJWTAccessWithScope = true; + + // Set defaultServicePath on the auth object. + this.auth.defaultServicePath = this._servicePath; + + // Set the default scopes in auth client if needed. + if (servicePath === this._servicePath) { + this.auth.defaultScopes = staticMembers.scopes; + } + + // Determine the client header string. + const clientHeader = [`gax/${this._gaxModule.version}`, `gapic/${version}`]; + if (typeof process === 'object' && 'versions' in process) { + clientHeader.push(`gl-node/${process.versions.node}`); + } else { + clientHeader.push(`gl-web/${this._gaxModule.version}`); + } + if (!opts.fallback) { + clientHeader.push(`grpc/${this._gaxGrpc.grpcVersion}`); + } else { + clientHeader.push(`rest/${this._gaxGrpc.grpcVersion}`); + } + if (opts.libName && opts.libVersion) { + clientHeader.push(`${opts.libName}/${opts.libVersion}`); + } + // Load the applicable protos. + this._protos = this._gaxGrpc.loadProtoJSON(jsonProtos); + + // Some methods on this API support resumable uploads; provide + // descriptors for these methods. + this.descriptors.resumableUpload = { + uploadMedia: new this._gaxModule.ResumableUploadDescriptor( + '/resumable/upload', + ), + }; + + // Put together the default options sent with requests. + this._defaults = this._gaxGrpc.constructSettings( + 'google.showcase.v1beta1.ResumableUploadService', + gapicConfig as gax.ClientConfig, + opts.clientConfig || {}, + { 'x-goog-api-client': clientHeader.join(' ') }, + ); + + // Set up a dictionary of "inner API calls"; the core implementation + // of calling the API is handled in `google-gax`, with this code + // merely providing the destination and request information. + this.innerApiCalls = {}; + + // Add a warn function to the client constructor so it can be easily tested. + this.warn = this._gaxModule.warn; + } + + /** + * Initialize the client. + * Performs asynchronous operations (such as authentication) and prepares the client. + * This function will be called automatically when any class method is called for the + * first time, but if you need to initialize it before calling an actual method, + * feel free to call initialize() directly. + * + * You can await on this method if you want to make sure the client is initialized. + * + * @returns {Promise} A promise that resolves to an authenticated service stub. + */ + initialize() { + // If the client stub promise is already initialized, return immediately. + if (this.resumableUploadServiceStub) { + return this.resumableUploadServiceStub; + } + + // Put together the "service stub" for + // google.showcase.v1beta1.ResumableUploadService. + this.resumableUploadServiceStub = this._gaxGrpc.createStub( + this._opts.fallback + ? (this._protos as protobuf.Root).lookupService( + 'google.showcase.v1beta1.ResumableUploadService', + ) + : // eslint-disable-next-line @typescript-eslint/no-explicit-any + (this._protos as any).google.showcase.v1beta1.ResumableUploadService, + this._opts, + this._providedCustomServicePath, + ) as Promise<{ [method: string]: Function }>; + + // Resumable upload methods do not use the gRPC/REST service stub; the + // The ResumableUploadSession performs its own HTTP requests. + this.innerApiCalls['uploadMedia'] = this._gaxModule.createApiCall( + this._gaxModule.resumableUploadStub, + this._defaults['uploadMedia'], + this.descriptors.resumableUpload!['uploadMedia'], + this._opts.fallback, + ); + + return this.resumableUploadServiceStub; + } + + /** + * The DNS address for this API service. + * @deprecated Use the apiEndpoint method of the client instance. + * @returns {string} The DNS address for this service. + */ + static get servicePath() { + if ( + typeof process === 'object' && + typeof process.emitWarning === 'function' + ) { + process.emitWarning( + 'Static servicePath is deprecated, please use the instance method instead.', + 'DeprecationWarning', + ); + } + return 'localhost'; + } + + /** + * The DNS address for this API service - same as servicePath. + * @deprecated Use the apiEndpoint method of the client instance. + * @returns {string} The DNS address for this service. + */ + static get apiEndpoint() { + if ( + typeof process === 'object' && + typeof process.emitWarning === 'function' + ) { + process.emitWarning( + 'Static apiEndpoint is deprecated, please use the instance method instead.', + 'DeprecationWarning', + ); + } + return 'localhost'; + } + + /** + * The DNS address for this API service. + * @returns {string} The DNS address for this service. + */ + get apiEndpoint() { + return this._servicePath; + } + + get universeDomain() { + return this._universeDomain; + } + + /** + * The port for this API service. + * @returns {number} The default port for this service. + */ + static get port() { + return 7469; + } + + /** + * The scopes needed to make gRPC calls for every method defined + * in this service. + * @returns {string[]} List of default scopes. + */ + static get scopes() { + return []; + } + + getProjectId(): Promise; + getProjectId(callback: Callback): void; + /** + * Return the project ID used by this class. + * @returns {Promise} A promise that resolves to string containing the project ID. + */ + getProjectId( + callback?: Callback, + ): Promise | void { + if (callback) { + this.auth.getProjectId(callback); + return; + } + return this.auth.getProjectId(); + } + + // ------------------- + // -- Service calls -- + // ------------------- + /** + * Creates a {@link gax.ResumableSource} backed by a local file. + * + * This delegates to google-gax so generated clients do not depend on the + * Node.js `fs` module at import time. The source can be passed as the + * `uploadSource` parameter of {@link gax.ResumableUploadSession#start}. + * + * @param {string} filePath - Path to the local file to upload. + * @returns {gax.ResumableSource} A seekable upload source for the file. + */ + getResumableSource(filePath: string): gax.ResumableSource { + const gaxModule = this._gaxModule as typeof gax; + return gaxModule.resumableSourceFromFile(filePath); + } + /** + * A method with media_upload annotation enabled. + * + * @param {object} [request] - The request object. + * @param {object} [options] - Optional parameters. The upload source, chunk + * size, progress callback, and resume URL are passed to + * {@link gax.ResumableUploadSession#start} instead. + * @returns {Promise} A resumable upload session. + * Call `.start(uploadParams)` with the source to upload, then await + * `.finished()` for the final RPC response. + */ + uploadMedia( + request?: protos.google.showcase.v1beta1.IUploadMediaRequest, + options?: CallOptions, + ): Promise { + request = request || {}; + options = options || {}; + if (!this._opts.fallback && this._opts.sslCreds) { + return Promise.reject( + new this._gaxModule.GoogleError( + 'Resumable upload methods require HTTP(S) authentication and ' + + 'cannot be used with gRPC channel credentials. Configure the ' + + 'client without `sslCreds`, or use `fallback: true`.', + ), + ); + } + this.initialize().catch((err) => { + throw err; + }); + this._log.info('uploadMedia request %j', request); + return ( + this.innerApiCalls['uploadMedia'](request, { + ...options, + resumableUpload: { + auth: this._fallbackRest!.auth as gax.GoogleAuth, + servicePath: this._opts.servicePath ?? this._servicePath, + servicePort: this._opts.port || 443, + protocol: this._opts.protocol || 'https', + rpc: this._gaxModule.protobuf.Root.fromJSON(jsonProtos).lookupService( + 'google.showcase.v1beta1.ResumableUploadService', + ).methods['UploadMedia'], + request, + uploadPrefix: '/resumable/upload', + numericEnums: this._opts.numericEnums, + minifyJson: this._opts.minifyJson, + }, + }) as Promise<[gax.ResumableUploadSession]> + ).then(([session]) => session); + } + + /** + * Terminate the gRPC channel and close the client. + * + * The client will no longer be usable and all future behavior is undefined. + * @returns {Promise} A promise that resolves when the client is closed. + */ + close(): Promise { + if (this.resumableUploadServiceStub && !this._terminated) { + return this.resumableUploadServiceStub.then((stub) => { + this._log.info('ending gRPC channel'); + this._terminated = true; + stub.close(); + }); + } + return Promise.resolve(); + } +} diff --git a/core/packages/gax/test/showcase-resumable-upload/client/src/v1beta1/resumable_upload_service_client_config.json b/core/packages/gax/test/showcase-resumable-upload/client/src/v1beta1/resumable_upload_service_client_config.json new file mode 100644 index 000000000000..ca6004c306a2 --- /dev/null +++ b/core/packages/gax/test/showcase-resumable-upload/client/src/v1beta1/resumable_upload_service_client_config.json @@ -0,0 +1,30 @@ +{ + "interfaces": { + "google.showcase.v1beta1.ResumableUploadService": { + "retry_codes": { + "non_idempotent": [], + "idempotent": [ + "DEADLINE_EXCEEDED", + "UNAVAILABLE" + ] + }, + "retry_params": { + "default": { + "initial_retry_delay_millis": 100, + "retry_delay_multiplier": 1.3, + "max_retry_delay_millis": 60000, + "initial_rpc_timeout_millis": 60000, + "rpc_timeout_multiplier": 1, + "max_rpc_timeout_millis": 60000, + "total_timeout_millis": 600000 + } + }, + "methods": { + "UploadMedia": { + "retry_codes_name": "non_idempotent", + "retry_params_name": "default" + } + } + } + } +} diff --git a/core/packages/gax/test/showcase-resumable-upload/client/src/v1beta1/resumable_upload_service_proto_list.json b/core/packages/gax/test/showcase-resumable-upload/client/src/v1beta1/resumable_upload_service_proto_list.json new file mode 100644 index 000000000000..1d19034e74ad --- /dev/null +++ b/core/packages/gax/test/showcase-resumable-upload/client/src/v1beta1/resumable_upload_service_proto_list.json @@ -0,0 +1,3 @@ +[ + "../../protos/google/showcase/v1beta1/resumable_upload.proto" +] diff --git a/core/packages/gax/test/showcase-resumable-upload/client/tsconfig.json b/core/packages/gax/test/showcase-resumable-upload/client/tsconfig.json new file mode 100644 index 000000000000..b88741ab40ce --- /dev/null +++ b/core/packages/gax/test/showcase-resumable-upload/client/tsconfig.json @@ -0,0 +1,18 @@ +{ + "extends": "../../../node_modules/gts/tsconfig-google.json", + "compilerOptions": { + "rootDir": ".", + "outDir": "build", + "resolveJsonModule": true, + "lib": [ + "es2023", + "dom" + ] + }, + "include": [ + "src/*.ts", + "src/**/*.ts", + "src/**/*.json", + "protos/protos.json" + ] +} diff --git a/core/packages/gax/test/showcase-resumable-upload/run.sh b/core/packages/gax/test/showcase-resumable-upload/run.sh new file mode 100755 index 000000000000..59123503f1c1 --- /dev/null +++ b/core/packages/gax/test/showcase-resumable-upload/run.sh @@ -0,0 +1,113 @@ +#!/usr/bin/env bash +# 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. + +# Runs sample.js against the gapic-showcase ResumableUploadService using the +# google-gax checkout in this monorepo. +# +# Env overrides: +# SHOWCASE_VERSION gapic-showcase release to download (default: 0.43.1) +# SHOWCASE_BIN path to an existing gapic-showcase binary +# SHOWCASE_PORT port for the showcase server (default: 7469) +# UPLOAD_FILE file to upload (default: a generated 512 KiB file) + +set -euo pipefail + +HARNESS_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +CLIENT_DIR="$HARNESS_DIR/client" +GAX_DIR="$(cd "$HARNESS_DIR/../.." && pwd)" +REPO_ROOT="$(cd "$GAX_DIR/../../.." && pwd)" +TOOLS_DIR="$REPO_ROOT/core/packages/tools" + +SHOWCASE_VERSION="${SHOWCASE_VERSION:-0.43.1}" +SHOWCASE_PORT="${SHOWCASE_PORT:-7469}" +SHOWCASE_BIN="${SHOWCASE_BIN:-}" + +# Make the local google-gax checkout importable by the sample and client. +mkdir -p "$HARNESS_DIR/node_modules" +ln -sfn "$GAX_DIR" "$HARNESS_DIR/node_modules/google-gax" + +echo "Compiling google-gax from $GAX_DIR" +(cd "$GAX_DIR" && npm run compile) + +if [[ ! -f "$TOOLS_DIR/build/src/compileProtos.js" ]]; then + echo "Compiling gapic-tools from $TOOLS_DIR" + (cd "$TOOLS_DIR" && npm run compile) +fi + +echo "Compiling the generated showcase client" +(cd "$CLIENT_DIR" && node "$TOOLS_DIR/build/src/compileProtos.js" src) +(cd "$CLIENT_DIR" && "$GAX_DIR/node_modules/.bin/tsc" -p .) +(cd "$CLIENT_DIR" && cp -R protos build/) + +DOWNLOAD_DIR="${TMPDIR:-/tmp}/gapic-showcase-$SHOWCASE_VERSION" +if [[ -z "$SHOWCASE_BIN" ]]; then + os="$(uname -s | tr '[:upper:]' '[:lower:]')" + case "$(uname -m)" in + x86_64|amd64) arch="amd64" ;; + arm64|aarch64) arch="arm64" ;; + *) echo "Unsupported architecture: $(uname -m)" >&2; exit 1 ;; + esac + + if [[ "$os" != "darwin" && "$os" != "linux" ]]; then + echo "Unsupported OS: $os (only darwin and linux are supported by this script)" >&2 + exit 1 + fi + + SHOWCASE_BIN="$DOWNLOAD_DIR/gapic-showcase" + if [[ ! -x "$SHOWCASE_BIN" ]]; then + mkdir -p "$DOWNLOAD_DIR" + tarball="$DOWNLOAD_DIR/gapic-showcase-$SHOWCASE_VERSION-$os-$arch.tar.gz" + echo "Downloading gapic-showcase $SHOWCASE_VERSION from GitHub releases" + curl -fsSL \ + "https://github.com/googleapis/gapic-showcase/releases/download/v$SHOWCASE_VERSION/gapic-showcase-$SHOWCASE_VERSION-$os-$arch.tar.gz" \ + -o "$tarball" + tar -xzf "$tarball" -C "$DOWNLOAD_DIR" + fi +fi + +LOG_FILE="$DOWNLOAD_DIR/showcase-$SHOWCASE_PORT.log" +echo "Starting gapic-showcase on port $SHOWCASE_PORT" +"$SHOWCASE_BIN" run --port ":$SHOWCASE_PORT" >"$LOG_FILE" 2>&1 & +SERVER_PID=$! +cleanup() { + kill "$SERVER_PID" 2>/dev/null || true + wait "$SERVER_PID" 2>/dev/null || true +} +trap cleanup EXIT + +for _ in $(seq 1 100); do + if curl -sS --max-time 1 "http://127.0.0.1:$SHOWCASE_PORT/" >/dev/null 2>&1; then + break + fi + sleep 0.1 +done +if ! kill -0 "$SERVER_PID" 2>/dev/null; then + echo "gapic-showcase failed to start; see $LOG_FILE" >&2 + exit 1 +fi + +PAYLOAD_FILE="${UPLOAD_FILE:-}" +PAYLOAD_CLEANUP="" +if [[ -z "$PAYLOAD_FILE" ]]; then + PAYLOAD_FILE="$(mktemp "${TMPDIR:-/tmp}/showcase-upload.XXXXXX")" + dd if=/dev/zero of="$PAYLOAD_FILE" bs=1024 count=512 2>/dev/null + PAYLOAD_CLEANUP="$PAYLOAD_FILE" +fi +if [[ -n "$PAYLOAD_CLEANUP" ]]; then + trap 'cleanup; rm -f "$PAYLOAD_CLEANUP"' EXIT +fi + +echo "Running sample.js" +(cd "$HARNESS_DIR" && SHOWCASE_PORT="$SHOWCASE_PORT" UPLOAD_FILE="$PAYLOAD_FILE" node sample.js) diff --git a/core/packages/gax/test/showcase-resumable-upload/sample.js b/core/packages/gax/test/showcase-resumable-upload/sample.js new file mode 100644 index 000000000000..c024c44d042b --- /dev/null +++ b/core/packages/gax/test/showcase-resumable-upload/sample.js @@ -0,0 +1,79 @@ +// 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. + +// Runs a resumable upload against the gapic-showcase ResumableUploadService +// using the generated client in ./client. + +'use strict'; + +const fs = require('fs'); +const path = require('path'); +const {GoogleAuth, googleAuthLibrary} = require('google-gax'); +const {ResumableUploadServiceClient} = require('./client'); + +async function main() { + const filePath = process.env.UPLOAD_FILE; + const port = Number(process.env.SHOWCASE_PORT || 7469); + if (!filePath) { + throw new Error('Set UPLOAD_FILE to the path of the file to upload.'); + } + if (!fs.existsSync(filePath)) { + throw new Error(`Upload file does not exist: ${filePath}`); + } + + const size = fs.statSync(filePath).size; + const client = new ResumableUploadServiceClient({ + servicePath: '127.0.0.1', + port, + protocol: 'http', + auth: new GoogleAuth({ + authClient: new googleAuthLibrary.PassThroughClient(), + }), + }); + + try { + console.log( + `Uploading ${filePath} (${size} bytes) through ${client.apiEndpoint}` + ); + + const session = await client.uploadMedia({ + name: path.basename(filePath), + }); + + await session.start({ + uploadSource: client.getResumableSource(filePath), + chunkSize: 512 * 1024, + onProgress: status => { + console.log(` ${status.bytesUploaded} / ${size} bytes committed`); + }, + }); + + console.log(`Upload session: ${session.uploadUrl}`); + const response = await session.finished(); + const uploadedSize = Number(response.size); + if (uploadedSize !== size) { + throw new Error( + `Uploaded size mismatch: expected ${size}, got ${response.size}` + ); + } + console.log(`Upload complete: ${JSON.stringify(response)}`); + } finally { + await client.close(); + } +} + +main().catch(err => { + console.error(err); + process.exitCode = 1; +}); From d24c3076b199bb9f542f90785c7e7923abd47d29 Mon Sep 17 00:00:00 2001 From: feywind <57276408+feywind@users.noreply.github.com> Date: Thu, 10 Sep 2026 18:39:20 -0400 Subject: [PATCH 3/3] test(gax): keep the showcase harness client out of presubmit lint The checked-in generated client is linted and type-checked as its own package, but CI installs the published google-gax (which does not have the resumable upload APIs yet) and never compiles protos/protos, so the client reported 12 type errors plus a promise/always-return error on every run. Treat it as a fixture instead: - move client/ to fixtures/, a path segment the monorepo linter ignores - rename its tsconfig.json to tsconfig.client.json, so the package detection walks up to google-gax's tsconfig and skips these files - update run.sh, sample.js, the client package.json and the harness README Verified by running the harness's own compile steps (compileProtos plus tsc -p tsconfig.client.json) against the local google-gax checkout. --- core/packages/gax/.gitignore | 2 +- .../packages/gax/test/showcase-resumable-upload/README.md | 8 ++++++-- .../{client => fixtures}/package.json | 2 +- .../protos/google/showcase/v1beta1/resumable_upload.proto | 0 .../{client => fixtures}/src/index.ts | 0 .../{client => fixtures}/src/v1beta1/index.ts | 0 .../src/v1beta1/resumable_upload_service_client.ts | 0 .../v1beta1/resumable_upload_service_client_config.json | 0 .../src/v1beta1/resumable_upload_service_proto_list.json | 0 .../tsconfig.json => fixtures/tsconfig.client.json} | 0 core/packages/gax/test/showcase-resumable-upload/run.sh | 4 ++-- .../packages/gax/test/showcase-resumable-upload/sample.js | 4 ++-- 12 files changed, 12 insertions(+), 8 deletions(-) rename core/packages/gax/test/showcase-resumable-upload/{client => fixtures}/package.json (88%) rename core/packages/gax/test/showcase-resumable-upload/{client => fixtures}/protos/google/showcase/v1beta1/resumable_upload.proto (100%) rename core/packages/gax/test/showcase-resumable-upload/{client => fixtures}/src/index.ts (100%) rename core/packages/gax/test/showcase-resumable-upload/{client => fixtures}/src/v1beta1/index.ts (100%) rename core/packages/gax/test/showcase-resumable-upload/{client => fixtures}/src/v1beta1/resumable_upload_service_client.ts (100%) rename core/packages/gax/test/showcase-resumable-upload/{client => fixtures}/src/v1beta1/resumable_upload_service_client_config.json (100%) rename core/packages/gax/test/showcase-resumable-upload/{client => fixtures}/src/v1beta1/resumable_upload_service_proto_list.json (100%) rename core/packages/gax/test/showcase-resumable-upload/{client/tsconfig.json => fixtures/tsconfig.client.json} (100%) diff --git a/core/packages/gax/.gitignore b/core/packages/gax/.gitignore index 4d28f76e1bdc..3f5b21bb889a 100644 --- a/core/packages/gax/.gitignore +++ b/core/packages/gax/.gitignore @@ -23,4 +23,4 @@ dist/ *.tgz **/*.tgz test/showcase-echo-client/protos/protos.* -test/showcase-resumable-upload/client/protos/protos.* +test/showcase-resumable-upload/fixtures/protos/protos.* diff --git a/core/packages/gax/test/showcase-resumable-upload/README.md b/core/packages/gax/test/showcase-resumable-upload/README.md index b2bcad05fb8a..bd1985901bbc 100644 --- a/core/packages/gax/test/showcase-resumable-upload/README.md +++ b/core/packages/gax/test/showcase-resumable-upload/README.md @@ -3,10 +3,14 @@ This directory contains a small end-to-end example of the resumable upload support added in the `scotty-1` work: -* `client/` — a generated `ResumableUploadServiceClient` for the real +* `fixtures/` — a generated `ResumableUploadServiceClient` for the real [gapic-showcase](https://github.com/googleapis/gapic-showcase) `ResumableUploadService`, produced by the generator in this repo with - `--resumable_upload_methods=ResumableUploadService.UploadMedia`. + `--resumable_upload_methods=ResumableUploadService.UploadMedia`. It is a + fixture rather than client source, so it lives under `fixtures/` and keeps + its `tsconfig.client.json`: the monorepo presubmit linter skips both, since + the generated client needs the resumable upload APIs from this checkout and + protos that only `run.sh` compiles. * `sample.js` — example code using `client.uploadMedia()`, `client.getResumableSource()`, and `session.start()`. * `run.sh` — downloads/starts a gapic-showcase server, builds the local diff --git a/core/packages/gax/test/showcase-resumable-upload/client/package.json b/core/packages/gax/test/showcase-resumable-upload/fixtures/package.json similarity index 88% rename from core/packages/gax/test/showcase-resumable-upload/client/package.json rename to core/packages/gax/test/showcase-resumable-upload/fixtures/package.json index 41484e192bf8..24c737aa18a7 100644 --- a/core/packages/gax/test/showcase-resumable-upload/client/package.json +++ b/core/packages/gax/test/showcase-resumable-upload/fixtures/package.json @@ -10,7 +10,7 @@ "build/protos" ], "scripts": { - "compile": "tsc -p . && cp -r protos build/ && minifyProtoJson", + "compile": "tsc -p tsconfig.client.json && cp -r protos build/ && minifyProtoJson", "compile-protos": "compileProtos src", "prepare": "npm run compile-protos && npm run compile" }, diff --git a/core/packages/gax/test/showcase-resumable-upload/client/protos/google/showcase/v1beta1/resumable_upload.proto b/core/packages/gax/test/showcase-resumable-upload/fixtures/protos/google/showcase/v1beta1/resumable_upload.proto similarity index 100% rename from core/packages/gax/test/showcase-resumable-upload/client/protos/google/showcase/v1beta1/resumable_upload.proto rename to core/packages/gax/test/showcase-resumable-upload/fixtures/protos/google/showcase/v1beta1/resumable_upload.proto diff --git a/core/packages/gax/test/showcase-resumable-upload/client/src/index.ts b/core/packages/gax/test/showcase-resumable-upload/fixtures/src/index.ts similarity index 100% rename from core/packages/gax/test/showcase-resumable-upload/client/src/index.ts rename to core/packages/gax/test/showcase-resumable-upload/fixtures/src/index.ts diff --git a/core/packages/gax/test/showcase-resumable-upload/client/src/v1beta1/index.ts b/core/packages/gax/test/showcase-resumable-upload/fixtures/src/v1beta1/index.ts similarity index 100% rename from core/packages/gax/test/showcase-resumable-upload/client/src/v1beta1/index.ts rename to core/packages/gax/test/showcase-resumable-upload/fixtures/src/v1beta1/index.ts diff --git a/core/packages/gax/test/showcase-resumable-upload/client/src/v1beta1/resumable_upload_service_client.ts b/core/packages/gax/test/showcase-resumable-upload/fixtures/src/v1beta1/resumable_upload_service_client.ts similarity index 100% rename from core/packages/gax/test/showcase-resumable-upload/client/src/v1beta1/resumable_upload_service_client.ts rename to core/packages/gax/test/showcase-resumable-upload/fixtures/src/v1beta1/resumable_upload_service_client.ts diff --git a/core/packages/gax/test/showcase-resumable-upload/client/src/v1beta1/resumable_upload_service_client_config.json b/core/packages/gax/test/showcase-resumable-upload/fixtures/src/v1beta1/resumable_upload_service_client_config.json similarity index 100% rename from core/packages/gax/test/showcase-resumable-upload/client/src/v1beta1/resumable_upload_service_client_config.json rename to core/packages/gax/test/showcase-resumable-upload/fixtures/src/v1beta1/resumable_upload_service_client_config.json diff --git a/core/packages/gax/test/showcase-resumable-upload/client/src/v1beta1/resumable_upload_service_proto_list.json b/core/packages/gax/test/showcase-resumable-upload/fixtures/src/v1beta1/resumable_upload_service_proto_list.json similarity index 100% rename from core/packages/gax/test/showcase-resumable-upload/client/src/v1beta1/resumable_upload_service_proto_list.json rename to core/packages/gax/test/showcase-resumable-upload/fixtures/src/v1beta1/resumable_upload_service_proto_list.json diff --git a/core/packages/gax/test/showcase-resumable-upload/client/tsconfig.json b/core/packages/gax/test/showcase-resumable-upload/fixtures/tsconfig.client.json similarity index 100% rename from core/packages/gax/test/showcase-resumable-upload/client/tsconfig.json rename to core/packages/gax/test/showcase-resumable-upload/fixtures/tsconfig.client.json diff --git a/core/packages/gax/test/showcase-resumable-upload/run.sh b/core/packages/gax/test/showcase-resumable-upload/run.sh index 59123503f1c1..8f68a0042798 100755 --- a/core/packages/gax/test/showcase-resumable-upload/run.sh +++ b/core/packages/gax/test/showcase-resumable-upload/run.sh @@ -25,7 +25,7 @@ set -euo pipefail HARNESS_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" -CLIENT_DIR="$HARNESS_DIR/client" +CLIENT_DIR="$HARNESS_DIR/fixtures" GAX_DIR="$(cd "$HARNESS_DIR/../.." && pwd)" REPO_ROOT="$(cd "$GAX_DIR/../../.." && pwd)" TOOLS_DIR="$REPO_ROOT/core/packages/tools" @@ -48,7 +48,7 @@ fi echo "Compiling the generated showcase client" (cd "$CLIENT_DIR" && node "$TOOLS_DIR/build/src/compileProtos.js" src) -(cd "$CLIENT_DIR" && "$GAX_DIR/node_modules/.bin/tsc" -p .) +(cd "$CLIENT_DIR" && "$GAX_DIR/node_modules/.bin/tsc" -p tsconfig.client.json) (cd "$CLIENT_DIR" && cp -R protos build/) DOWNLOAD_DIR="${TMPDIR:-/tmp}/gapic-showcase-$SHOWCASE_VERSION" diff --git a/core/packages/gax/test/showcase-resumable-upload/sample.js b/core/packages/gax/test/showcase-resumable-upload/sample.js index c024c44d042b..4fef6c48c809 100644 --- a/core/packages/gax/test/showcase-resumable-upload/sample.js +++ b/core/packages/gax/test/showcase-resumable-upload/sample.js @@ -13,14 +13,14 @@ // limitations under the License. // Runs a resumable upload against the gapic-showcase ResumableUploadService -// using the generated client in ./client. +// using the generated client in ./fixtures. 'use strict'; const fs = require('fs'); const path = require('path'); const {GoogleAuth, googleAuthLibrary} = require('google-gax'); -const {ResumableUploadServiceClient} = require('./client'); +const {ResumableUploadServiceClient} = require('./fixtures'); async function main() { const filePath = process.env.UPLOAD_FILE;