feat(storage): expose highWaterMark option for File stream operations

handwritten/storage/src/file.ts

@@ -337,6 +337,11 @@ export interface CreateResumableUploadCallback {
 export interface CreateWriteStreamOptions extends CreateResumableUploadOptions {
   contentType?: string;
   gzip?: string | boolean;
+  /**
+   * Stream `highWaterMark` in bytes. Configures the maximum number of bytes
+   * buffered in the underlying writable stream before backpressure is applied.
+   */
+  highWaterMark?: number;
   resumable?: boolean;
   timeout?: number;
   validation?: string | boolean;
@@ -547,6 +552,11 @@ export interface CreateReadStreamOptions {
   start?: number;
   end?: number;
   decompress?: boolean;
+  /**
+   * Stream `highWaterMark` in bytes. Configures the maximum number of bytes
+   * buffered in the underlying readable stream.
+   */
+  highWaterMark?: number;
   [GCCL_GCS_CMD_KEY]?: string;
 }
 
@@ -1612,6 +1622,9 @@ class File extends ServiceObject<File, FileMetadata> {
    *     received data. By default this option is set to `true`.
    *     Applicable in cases where the data was uploaded with
    *     `gzip: true` option. See {@link File#createWriteStream}.
+   * @property {number} [highWaterMark] Stream `highWaterMark` in bytes.
+   *     Configures the maximum number of bytes buffered in the underlying
+   *     readable stream.
    */
   /**
    * Create a readable stream to read the contents of the remote file. It can be
@@ -1682,12 +1695,13 @@ class File extends ServiceObject<File, FileMetadata> {
    */
   createReadStream(options: CreateReadStreamOptions = {}): Readable {
     options = Object.assign({decompress: true}, options);
+    const {highWaterMark} = options;
     const rangeRequest =
       typeof options.start === 'number' || typeof options.end === 'number';
     const tailRequest = options.end! < 0;
 
     let validateStream: HashStreamValidator | undefined = undefined;
-    const throughStream = new PassThroughShim();
+    const throughStream = new PassThroughShim({highWaterMark});
 
     let crc32c = true;
     let md5 = false;
@@ -2063,6 +2077,9 @@ class File extends ServiceObject<File, FileMetadata> {
    *     If set to `auto`, the contentType is used to determine if the file
    * should be gzipped. This will set `options.metadata.contentEncoding` to
    * `gzip` if necessary.
+   * @property {number} [highWaterMark] Stream `highWaterMark` in bytes.
+   *     Configures the maximum number of bytes buffered in the underlying
+   *     writable stream before backpressure is applied.
    * @property {object} [metadata] See the examples below or
    *     {@link https://cloud.google.com/storage/docs/json_api/v1/objects/insert#request_properties_JSON| Objects: insert request body}
    *     for more details.
@@ -2218,6 +2235,7 @@ class File extends ServiceObject<File, FileMetadata> {
    * fs.createWriteStream({uri, resumeCRC32C});
    */
   createWriteStream(options: CreateWriteStreamOptions = {}): Writable {
+    const {highWaterMark} = options;
     options.metadata ??= {};
 
     if (options.contentType) {
@@ -2261,6 +2279,7 @@ class File extends ServiceObject<File, FileMetadata> {
 
     // A stream for consumer to write to
     const writeStream = new Writable({
+      highWaterMark,
       final(cb) {
         // Set the pipeline callback to this callback so the pipeline's results
         // can be populated to the consumer
@@ -2283,13 +2302,13 @@ class File extends ServiceObject<File, FileMetadata> {
       emitStream.destroy();
     });
 
-    const emitStream = new PassThroughShim();
+    const emitStream = new PassThroughShim({highWaterMark});
 
     // If `writeStream` is destroyed before the `writing` event, `emitStream` will not have any listeners. This prevents an unhandled error.
     const noop = () => {};
     emitStream.on('error', noop);
 
-    const fileWriteStream = duplexify();
+    const fileWriteStream = duplexify(undefined, undefined, {highWaterMark});
     let fileWriteStreamMetadataReceived = false;
 
     // Handing off emitted events to users
@@ -4873,7 +4892,9 @@ class File extends ServiceObject<File, FileMetadata> {
       options.preconditionOpts,
     );
 
-    const writeStream = new ProgressStream();
+    const writeStream = new ProgressStream({
+      highWaterMark: options.highWaterMark,
+    });
     writeStream.on('progress', evt => dup.emit('progress', evt));
     dup.setWritable(writeStream);
 

handwritten/storage/test/file.ts

@@ -17,6 +17,8 @@ import assert from 'assert';
 import {
   Bucket,
   CRC32C,
+  CreateReadStreamOptions,
+  CreateWriteStreamOptions,
   File,
   GaxiosError,
   GaxiosOptionsPrepared,
@@ -1179,6 +1181,12 @@ describe('File', () => {
         .resume();
     });
 
+    it('should pass highWaterMark to the returned stream', () => {
+      const options: CreateReadStreamOptions = {highWaterMark: 12345};
+      const stream = file.createReadStream(options);
+      assert.strictEqual(stream.readableHighWaterMark, 12345);
+    });
+
     describe('authenticating', () => {
       it('should create an authenticated request', () => {
         file.storageTransport.makeRequest = sandbox.stub().callsFake(opts => {
@@ -2259,6 +2267,48 @@ describe('File', () => {
       writable.write('data');
     });
 
+    it('should set highWaterMark on the returned stream', () => {
+      const options: CreateWriteStreamOptions = {highWaterMark: 12345};
+      const writable = file.createWriteStream(options);
+      assert.strictEqual(writable.writableHighWaterMark, 12345);
+    });
+
+    it('should propagate highWaterMark to resumable upload', done => {
+      const highWaterMark = 12345;
+      const options: CreateWriteStreamOptions = {
+        highWaterMark,
+        resumable: true,
+      };
+
+      file.startResumableUpload_ = sandbox.stub().callsFake((dup, opts) => {
+        assert.strictEqual(dup.writableHighWaterMark, highWaterMark);
+        assert.strictEqual(opts.highWaterMark, highWaterMark);
+        done();
+      });
+
+      const writable = file.createWriteStream(options);
+      assert.strictEqual(writable.writableHighWaterMark, highWaterMark);
+      writable.write('data');
+    });
+
+    it('should propagate highWaterMark to simple upload', done => {
+      const highWaterMark = 12345;
+      const options: CreateWriteStreamOptions = {
+        highWaterMark,
+        resumable: false,
+      };
+
+      file.startSimpleUpload_ = sandbox.stub().callsFake((dup, opts) => {
+        assert.strictEqual(dup.writableHighWaterMark, highWaterMark);
+        assert.strictEqual(opts.highWaterMark, highWaterMark);
+        done();
+      });
+
+      const writable = file.createWriteStream(options);
+      assert.strictEqual(writable.writableHighWaterMark, highWaterMark);
+      writable.write('data');
+    });
+
     it('should alias contentType to metadata object', done => {
       const contentType = 'text/html';
       const writable = file.createWriteStream({contentType});
@@ -5748,6 +5798,7 @@ describe('File', () => {
           uri: 'http://resumable-uri',
           userProject: 'user-project-id',
           chunkSize: 262144, // 256 KiB
+          highWaterMark: 12345,
         };
 
         const resumableUpload = {
@@ -5765,6 +5816,7 @@ describe('File', () => {
             assert.strictEqual(opts.uri, options.uri);
             assert.strictEqual(opts.userProject, options.userProject);
             assert.strictEqual(opts.chunkSize, options.chunkSize);
+            assert.strictEqual(opts.highWaterMark, options.highWaterMark);
 
             setImmediate(done);
             return new PassThrough();
@@ -5775,6 +5827,18 @@ describe('File', () => {
         file.startResumableUpload_(duplexify(), options);
       });
 
+      it('should pass highWaterMark to the upload stream', done => {
+        const dup = duplexify();
+        const highWaterMark = 12345;
+
+        dup.setWritable = (stream: Duplex) => {
+          assert.strictEqual(stream.writableHighWaterMark, highWaterMark);
+          done();
+        };
+
+        file.startResumableUpload_(dup, {highWaterMark});
+      });
+
       it('should emit the response', done => {
         const resp = {};
         const uploadStream = new PassThrough();
@@ -6035,6 +6099,34 @@ describe('File', () => {
       await file.startSimpleUpload_(duplexify(), options);
     });
 
+    it('should pass highWaterMark if set', async () => {
+      const options: CreateWriteStreamOptions = {
+        highWaterMark: 12345,
+      };
+      const dup = duplexify();
+      let setWritableStream: Duplex | undefined;
+      dup.setWritable = sandbox.stub().callsFake((stream: Duplex) => {
+        setWritableStream = stream;
+      });
+      file.storageTransport.makeRequest = sandbox
+        .stub()
+        .callsFake((options_: StorageRequestOptions) => {
+          const mediaPart = options_.multipart?.[1];
+          assert.strictEqual(mediaPart?.content, setWritableStream);
+          assert.strictEqual(
+            setWritableStream?.writableHighWaterMark,
+            options.highWaterMark,
+          );
+          assert.strictEqual(
+            setWritableStream?.readableHighWaterMark,
+            options.highWaterMark,
+          );
+        })
+        .resolves({});
+
+      await file.startSimpleUpload_(dup, options);
+    });
+
     describe('request', () => {
       describe('error', () => {
         const error = new Error('Error.');