Fix unresolved streaming cache review comments

Author: Copilot

libraries/rush-lib/src/logic/buildCache/ICloudBuildCacheProvider.ts

@@ -18,8 +18,9 @@ export interface ICloudBuildCacheProvider {
    * cache entry into memory, if possible. The implementation should download the cache entry and write it
    * to the specified local file path.
    *
-   * @returns `true` if the cache entry was found and written to the file, `false` if it was
-   * not found. Throws on errors.
+   * @returns `true` if the cache entry was found and written to the file; otherwise `false`.
+   * Implementations typically log transfer failures and return `false`, but may still throw for
+   * unexpected errors.
    */
   tryDownloadCacheEntryToFileAsync?(
     terminal: ITerminal,

libraries/rush-lib/src/utilities/WebClient.ts

@@ -170,6 +170,10 @@ function _makeRawRequestAsync<TResponse>(
 
         handleResponse(response, redirected, resolve, reject);
       }).on('error', (error: Error) => {
+        if (body && !Buffer.isBuffer(body)) {
+          body.destroy(error);
+        }
+
         reject(error);
       });
 

libraries/rush-lib/src/utilities/test/WebClient.test.ts

@@ -1,6 +1,9 @@
 // Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license.
 // See LICENSE in the project root for license information.
 
+import { createServer, type Server } from 'node:http';
+import { Readable } from 'node:stream';
+
 import { WebClient } from '../WebClient';
 
 describe(WebClient.name, () => {
@@ -52,4 +55,40 @@ describe(WebClient.name, () => {
       expect(target).toMatchSnapshot();
     });
   });
+
+  describe(WebClient.prototype.fetchAsync.name, () => {
+    it('destroys a streamed request body if the request errors', async () => {
+      const server: Server = createServer((request) => {
+        request.socket.destroy();
+      });
+      await new Promise<void>((resolve, reject) => {
+        server.once('error', reject);
+        server.listen(0, '127.0.0.1', () => resolve());
+      });
+
+      const address = server.address();
+      if (!address || typeof address === 'string') {
+        throw new Error('Expected a TCP server address');
+      }
+
+      const webClient: WebClient = new WebClient();
+      const body: Readable = new Readable({
+        read() {
+          this.push(Buffer.alloc(64 * 1024));
+        }
+      });
+
+      await expect(
+        webClient.fetchAsync(`http://127.0.0.1:${address.port}`, {
+          verb: 'PUT',
+          body
+        })
+      ).rejects.toThrow();
+      expect(body.destroyed).toBe(true);
+
+      await new Promise<void>((resolve, reject) => {
+        server.close((error) => (error ? reject(error) : resolve()));
+      });
+    });
+  });
 });

rush-plugins/rush-amazon-s3-build-cache-plugin/src/AmazonS3Client.ts

@@ -70,6 +70,15 @@ const storageRetryOptions: IStorageRetryOptions = {
   retryPolicyType: StorageRetryPolicyType.EXPONENTIAL
 };
 
+function _isWebClientResponse(response: IWebClientResponseBase): response is IWebClientResponse {
+  const candidate: Partial<IWebClientResponse> = response;
+  return (
+    typeof candidate.getTextAsync === 'function' &&
+    typeof candidate.getJsonAsync === 'function' &&
+    typeof candidate.getBufferAsync === 'function'
+  );
+}
+
 /**
  * Computes the SHA-256 hash of a file on disk using streaming reads.
  */
@@ -216,6 +225,7 @@ export class AmazonS3Client {
 
     // Compute SHA-256 hash of the file before uploading so we can sign the payload
     const contentHash: string = await _hashFileAsync(localFilePath);
+    const { size } = await FileSystem.getStatisticsAsync(localFilePath);
     const entryStream: FileSystemReadStream = FileSystem.createReadStream(localFilePath);
 
     // Streaming uploads cannot be retried because the stream is consumed after the first attempt.
@@ -224,7 +234,8 @@ export class AmazonS3Client {
       objectName,
       entryStream as Readable,
       true,
-      contentHash
+      contentHash,
+      size
     );
     response.stream.resume();
     if (!response.ok) {
@@ -260,7 +271,7 @@ export class AmazonS3Client {
     getSuccessResult: () => T | Promise<T>,
     cleanup?: () => void
   ): Promise<RetryableRequestResponse<T | undefined>> {
-    const { ok, status, statusText } = response;
+    const { ok, status } = response;
     if (ok) {
       return {
         hasNetworkError: false,
@@ -288,16 +299,25 @@ export class AmazonS3Client {
       };
     } else if (status === 400 || status === 401 || status === 403) {
       cleanup?.();
-      throw new Error(`Amazon S3 responded with status code ${status} (${statusText})`);
+      throw await this._getGetResponseErrorAsync(response);
     } else {
       cleanup?.();
       return {
         hasNetworkError: true,
-        error: new Error(`Amazon S3 responded with status code ${status} (${statusText})`)
+        error: await this._getGetResponseErrorAsync(response)
       };
     }
   }
 
+  private async _getGetResponseErrorAsync(response: IWebClientResponseBase): Promise<Error> {
+    if (_isWebClientResponse(response)) {
+      return await this._getS3ErrorAsync(response);
+    }
+
+    const { status, statusText } = response;
+    return new Error(`Amazon S3 responded with status code ${status} (${statusText})`);
+  }
+
   private async _makeSignedRequestAsync(
     verb: 'GET' | 'PUT',
     objectName: string,
@@ -308,27 +328,33 @@ export class AmazonS3Client {
     objectName: string,
     body: Readable | undefined,
     stream: true,
-    contentHash?: string
+    contentHash?: string,
+    contentLength?: number
   ): Promise<IWebClientStreamResponse>;
   private async _makeSignedRequestAsync(
     verb: 'GET' | 'PUT',
     objectName: string,
     body?: Buffer | Readable,
     stream?: boolean,
-    contentHash?: string
+    contentHash?: string,
+    contentLength?: number
   ): Promise<IWebClientResponse | IWebClientStreamResponse> {
     // Use the provided content hash if available (e.g. pre-computed from a file on disk),
     // otherwise compute from the buffer body, or use the empty hash for GET requests.
     const bodyHash: string = contentHash ?? this._getBufferSha256(Buffer.isBuffer(body) ? body : undefined);
-    const { url, headers } = this._buildSignedRequest(verb, objectName, bodyHash);
+    const { url, headers } = this._buildSignedRequest(verb, objectName, bodyHash, contentLength);
 
-    const webFetchOptions: IGetFetchOptions | IFetchOptionsWithBody = {
-      verb,
-      headers
-    };
-    if (verb === 'PUT' && body) {
-      (webFetchOptions as IFetchOptionsWithBody).body = body;
-    }
+    const webFetchOptions: IGetFetchOptions | IFetchOptionsWithBody =
+      verb === 'GET'
+        ? {
+            verb: 'GET',
+            headers
+          }
+        : {
+            verb: 'PUT',
+            headers,
+            body
+          };
 
     if (stream) {
       return await this._webClient.fetchStreamAsync(url, webFetchOptions);
@@ -343,12 +369,16 @@ export class AmazonS3Client {
   private _buildSignedRequest(
     verb: 'GET' | 'PUT',
     objectName: string,
-    bodyHash: string
+    bodyHash: string,
+    contentLength?: number
   ): { url: string; headers: Record<string, string> } {
     const isoDateString: IIsoDateString = this._getIsoDateString();
     const headers: Record<string, string> = {};
     headers[DATE_HEADER_NAME] = isoDateString.dateTime;
     headers[CONTENT_HASH_HEADER_NAME] = bodyHash;
+    if (verb === 'PUT' && contentLength !== undefined) {
+      headers['Content-Length'] = `${contentLength}`;
+    }
 
     // the host can be e.g. https://s3.aws.com or http://localhost:9000
     const host: string = this._s3Endpoint.replace(protocolRegex, '');

rush-plugins/rush-amazon-s3-build-cache-plugin/src/test/AmazonS3Client.test.ts

@@ -661,6 +661,9 @@ describe(AmazonS3Client.name, () => {
       jest
         .spyOn(FileSystem, 'createWriteStreamAsync')
         .mockResolvedValue({} as unknown as Awaited<ReturnType<typeof FileSystem.createWriteStreamAsync>>);
+      jest
+        .spyOn(FileSystem, 'getStatisticsAsync')
+        .mockResolvedValue({ size: 123 } as Awaited<ReturnType<typeof FileSystem.getStatisticsAsync>>);
       // Return a Readable that immediately ends, so _hashFileAsync completes with the null hash
       jest.spyOn(FileSystem, 'createReadStream').mockReturnValue(
         new Readable({
@@ -836,6 +839,7 @@ describe(AmazonS3Client.name, () => {
         const [url, options] = spy.mock.calls[0];
         expect(url).toBe('http://localhost:9000/abc123');
         expect(options.verb).toBe('PUT');
+        expect(options.headers['Content-Length']).toBe('123');
         // Verify the content hash is a real SHA-256 hex string, NOT UNSIGNED-PAYLOAD
         expect(options.headers['x-amz-content-sha256']).toMatch(/^[0-9a-f]{64}$/);
         expect(options.headers['x-amz-date']).toBe('20200418T123242Z');
@@ -875,6 +879,34 @@ describe(AmazonS3Client.name, () => {
         expect(spy).toHaveBeenCalledTimes(1);
         spy.mockRestore();
       });
+
+      it('Preserves S3 error details for buffered GET failures', async () => {
+        jest.spyOn(WebClient.prototype, 'fetchAsync').mockReturnValue(
+          Promise.resolve({
+            getBufferAsync: () => Promise.resolve(Buffer.from('AccessDenied: missing permission')),
+            getTextAsync: () => Promise.resolve('AccessDenied: missing permission'),
+            getJsonAsync: () => Promise.reject(new Error('Not JSON')),
+            headers: {},
+            status: 403,
+            statusText: 'Forbidden',
+            ok: false,
+            redirected: false
+          })
+        );
+
+        const s3Client: AmazonS3Client = new AmazonS3Client(
+          {
+            accessKeyId: 'accessKeyId',
+            secretAccessKey: 'secretAccessKey',
+            sessionToken: undefined
+          },
+          DUMMY_OPTIONS,
+          webClient,
+          terminal
+        );
+
+        await expect(s3Client.getObjectAsync('abc123')).rejects.toThrow('AccessDenied: missing permission');
+      });
     });
   });
 });