[build-tools][worker] Collect device run session activity events

packages/build-tools/src/gcs/LoggerStream.ts

@@ -29,6 +29,8 @@ class GCSLoggerStream extends Writable {
   private buffer: string[] = [];
   private writePromise?: Promise<any>;
   private cleanUpCalled: boolean = false;
+  private lastWriteError?: Error;
+  private lastUploadError?: Error;
 
   constructor({ logger, uploadMethod, options }: GCSLoggerStream.Config) {
     super();
@@ -62,7 +64,7 @@ class GCSLoggerStream extends Writable {
     return (await this.flush(true)) as string;
   }
 
-  public async cleanUp(): Promise<void> {
+  public async cleanUp({ failOnError = false }: { failOnError?: boolean } = {}): Promise<void> {
     if (this.cleanUpCalled) {
       this.logger.info('Cleanup already called');
       return;
@@ -77,14 +79,20 @@ class GCSLoggerStream extends Writable {
     }
 
     await this.safeWriteToFile(true);
+    const finalWriteError = this.lastWriteError;
     await fs.close(this.fileHandle);
     this.fileHandle = undefined;
 
     await this.flush(true);
+    const finalUploadError = this.lastUploadError;
 
     await fs.remove(this.temporaryLogsPath);
     await fs.remove(this.temporaryCompressedLogsPath);
     this.logger.info('Cleaning up GCS log stream');
+
+    if (failOnError && (finalWriteError || finalUploadError)) {
+      throw finalWriteError ?? finalUploadError;
+    }
   }
 
   public write(rec: any): boolean {
@@ -114,7 +122,9 @@ class GCSLoggerStream extends Writable {
       await this.writePromise;
       this.buffer = this.buffer.slice(buffer.length);
       this.hasChangesToUpload = true;
+      this.lastWriteError = undefined;
     } catch (err) {
+      this.lastWriteError = toError(err);
       this.logger.error({ err, origin: 'gcs-logger' }, 'Failed to write logs to file');
     } finally {
       this.writePromise = undefined;
@@ -141,9 +151,11 @@ class GCSLoggerStream extends Writable {
     });
     return await this.upload()
       .then(result => {
+        this.lastUploadError = undefined;
         return result;
       })
       .catch(err => {
+        this.lastUploadError = toError(err);
         this.logger.error({ err }, 'Failed to upload logs file to GCS');
       })
       .then(result => {
@@ -169,6 +181,7 @@ class GCSLoggerStream extends Writable {
       return await GCS.uploadWithSignedUrl({
         signedUrl: this.uploadMethod.signedUrl,
         srcGeneratorAsync,
+        requestTimeoutMs: this.options.uploadTimeoutMs,
       });
     }
 
@@ -221,6 +234,7 @@ namespace GCSLoggerStream {
   export interface Options {
     uploadIntervalMs: number;
     compress?: CompressionMethod;
+    uploadTimeoutMs?: number;
   }
 
   export type UploadMethod =
@@ -234,4 +248,8 @@ namespace GCSLoggerStream {
   }
 }
 
+function toError(value: unknown): Error {
+  return value instanceof Error ? value : new Error(String(value));
+}
+
 export default GCSLoggerStream;

packages/build-tools/src/gcs/__tests__/LoggerStream.test.ts

@@ -0,0 +1,96 @@
+jest.unmock('fs');
+jest.unmock('node:fs');
+jest.unmock('fs/promises');
+jest.unmock('node:fs/promises');
+
+import { bunyan } from '@expo/logger';
+import fetch from 'node-fetch';
+import { Readable } from 'node:stream';
+
+import GCS from '../client';
+import GCSLoggerStream from '../LoggerStream';
+
+const logger = {
+  info: jest.fn(),
+  error: jest.fn(),
+} as unknown as bunyan;
+
+function createLoggerStream({ uploadFile }: { uploadFile: jest.Mock }): GCSLoggerStream {
+  return new GCSLoggerStream({
+    uploadMethod: {
+      client: { uploadFile } as unknown as GCS,
+      key: 'worker.ndjson',
+      customTime: null,
+    },
+    options: {
+      uploadIntervalMs: 60_000,
+    },
+    logger,
+  });
+}
+
+describe(GCSLoggerStream, () => {
+  beforeEach(() => {
+    jest.clearAllMocks();
+  });
+
+  it('fails required cleanup when the final GCS upload fails', async () => {
+    const uploadError = new Error('GCS unavailable');
+    const stream = createLoggerStream({
+      uploadFile: jest.fn().mockRejectedValue(uploadError),
+    });
+    await stream.init();
+    stream.write({ msg: 'structured activity' });
+
+    await expect(stream.cleanUp({ failOnError: true })).rejects.toBe(uploadError);
+  });
+
+  it('keeps final upload best-effort for existing jobs', async () => {
+    const stream = createLoggerStream({
+      uploadFile: jest.fn().mockRejectedValue(new Error('GCS unavailable')),
+    });
+    await stream.init();
+    stream.write({ msg: 'ordinary build log' });
+
+    await expect(stream.cleanUp()).resolves.toBeUndefined();
+  });
+
+  it('succeeds when a prior upload failed but the final upload persisted', async () => {
+    const uploadFile = jest
+      .fn()
+      .mockRejectedValueOnce(new Error('transient GCS failure'))
+      .mockResolvedValue({ Location: 'https://logs.example.test/worker.ndjson' });
+    const stream = createLoggerStream({ uploadFile });
+    await stream.init();
+    stream.write({ msg: 'structured activity' });
+
+    await expect(stream.cleanUp({ failOnError: true })).resolves.toBeUndefined();
+    expect(uploadFile).toHaveBeenCalledTimes(2);
+  });
+});
+
+describe(GCS.uploadWithSignedUrl, () => {
+  it('aborts a log upload that exceeds its request deadline', async () => {
+    jest.mocked(fetch).mockImplementation(
+      async (_url, init) =>
+        await new Promise((_resolve, reject) => {
+          init?.signal?.addEventListener(
+            'abort',
+            () => {
+              reject(new Error('upload aborted'));
+            },
+            { once: true }
+          );
+        })
+    );
+
+    await expect(
+      GCS.uploadWithSignedUrl({
+        signedUrl: { url: 'https://gcs.example.test/logs', headers: {} },
+        srcGeneratorAsync: async () => Readable.from(['log']),
+        retries: 0,
+        requestTimeoutMs: 10,
+      })
+    ).rejects.toThrow('upload aborted');
+  });
+});

packages/build-tools/src/gcs/client.ts

@@ -22,6 +22,7 @@ interface UploadWithSignedUrlParams {
   srcGeneratorAsync: () => Promise<Readable>;
   retryIntervalMs?: RetryOptions['retryIntervalMs'];
   retries?: RetryOptions['retries'];
+  requestTimeoutMs?: number;
 }
 
 class GCS {
@@ -36,6 +37,7 @@ class GCS {
     srcGeneratorAsync,
     retries = 2,
     retryIntervalMs = 30_000,
+    requestTimeoutMs,
   }: UploadWithSignedUrlParams): Promise<string> {
     let resp;
     try {
@@ -46,6 +48,7 @@ class GCS {
             method: 'PUT',
             headers: signedUrl.headers,
             body: src,
+            ...(requestTimeoutMs ? { signal: AbortSignal.timeout(requestTimeoutMs) } : {}),
           });
         },
         {