Retry and rate-pace TestOps uploads

packages/cli/src/commands/open.ts

@@ -10,6 +10,7 @@ import { Command, Option } from "clipanion";
 import { red } from "yoctocolors";
 
 import { findFilesByGlobs } from "./../utils/fileSystem.js";
+import { notifySignals, waitForAbort } from "./../utils/signals.js";
 import { generate } from "./commons/generate.js";
 
 export class OpenCommand extends Command {
@@ -83,12 +84,14 @@ export class OpenCommand extends Command {
       });
 
       // clean up temp report directory on ctrl-c
-      process.on("SIGINT", async () => {
+      const notifier = notifySignals(["SIGINT", "SIGTERM"]);
+
+      void waitForAbort(notifier.signal).then(async () => {
         try {
           await rm(config.output, { recursive: true });
         } catch {}
 
-        process.exit(0);
+        process.exit(notifier.info()?.code ?? 0);
       });
 
       await serve({

packages/cli/src/utils/index.ts

@@ -3,3 +3,4 @@ export * from "./terminal.js";
 export * from "./logs.js";
 export * from "./execution-context.js";
 export * from "./fileSystem.js";
+export * from "./signals.js";

packages/cli/src/utils/signals.ts

@@ -0,0 +1,92 @@
+import process from "node:process";
+
+export const SIGNAL_EXIT_CODES: Partial<Record<NodeJS.Signals, number>> = {
+  SIGINT: 130,
+  SIGTERM: 143,
+};
+
+export const GRACEFUL_SHUTDOWN_TIMEOUT_MS = 60_000;
+export const COMMAND_TERMINATION_GRACE_MS = 5_000;
+
+export type SignalInfo = {
+  signal: NodeJS.Signals;
+  code: number;
+  receivedAt: number;
+  deadline: number;
+};
+
+const exitCodeForSignal = (signal: NodeJS.Signals): number => SIGNAL_EXIT_CODES[signal] ?? 1;
+
+export type SignalNotifier = {
+  /** aborts the moment the first SIGINT/SIGTERM is received */
+  signal: AbortSignal;
+  /** the signal that triggered the abort, once received */
+  info: () => SignalInfo | undefined;
+  /** stop listening for signals */
+  dispose: () => void;
+};
+
+export const notifySignals = (
+  signals: NodeJS.Signals[] = ["SIGINT", "SIGTERM"],
+  onRepeat?: (signal: NodeJS.Signals) => void,
+): SignalNotifier => {
+  const controller = new AbortController();
+  let info: SignalInfo | undefined;
+
+  const handler = (signal: NodeJS.Signals) => {
+    if (info) {
+      onRepeat?.(signal);
+      return;
+    }
+
+    const receivedAt = Date.now();
+
+    info = {
+      signal,
+      code: exitCodeForSignal(signal),
+      receivedAt,
+      deadline: receivedAt + GRACEFUL_SHUTDOWN_TIMEOUT_MS,
+    };
+
+    controller.abort();
+  };
+
+  for (const signal of signals) {
+    process.on(signal, handler);
+  }
+
+  return {
+    signal: controller.signal,
+    info: () => info,
+    dispose: () => {
+      for (const signal of signals) {
+        process.off(signal, handler);
+      }
+    },
+  };
+};
+
+/** time left until the graceful-shutdown deadline, floored at 0; undefined if no signal received yet */
+export const gracefulShutdownRemaining = (info: SignalInfo | undefined): number | undefined => {
+  if (!info) {
+    return undefined;
+  }
+
+  return Math.max(0, info.deadline - Date.now());
+};
+
+export const boundedTerminationSignal = (
+  info: SignalInfo | undefined,
+  graceMs: number = COMMAND_TERMINATION_GRACE_MS,
+): AbortSignal => {
+  const remaining = gracefulShutdownRemaining(info);
+  const timeoutMs = remaining === undefined ? graceMs : Math.min(remaining, graceMs);
+
+  return AbortSignal.timeout(timeoutMs);
+};
+
+/** resolves once the given signal aborts */
+export const waitForAbort = (signal: AbortSignal): Promise<void> =>
+  signal.aborted
+    ? Promise.resolve()
+    : new Promise((resolve) => signal.addEventListener("abort", () => resolve(), { once: true }));

packages/plugin-testops/README.md

@@ -80,6 +80,8 @@ The plugin accepts the following options:
 | `autocloseLaunch`  | When `true` (default), the launch is closed automatically when the plugin finishes; set to `false` to keep the launch open | `boolean` | `true`          |
 | `gitFlow`          | When `true`, collect Git metadata for TestOps Git Flow on CI uploads (opt-in)                                 | `boolean` | `false`         |
 | `ancestorLimit`    | How many ancestor commits to attach to the launch for history linking in TestOps | `number`  | `100`           |
+| `uploadRateLimit`  | Caps how fast uploads are sent to TestOps within a rolling time window (requests/files/bytes per window); pass `false` to disable pacing entirely | `{ windowMs: number; maxRequestsPerWindow?: number; maxFilesPerWindow?: number; maxBytesPerWindow?: number } \| false` | `20` req/s, `1000` files/s, `1 GiB`/s |
+| `reopenClosedLaunch` | When `true`, a launch that TestOps reports as closed is reopened automatically instead of failing the upload | `boolean` | `false`         |
 
 ### Using options from environment variables
 

packages/plugin-testops/src/client.ts

@@ -34,6 +34,7 @@ import type {
   UploadResultsResponseDto,
 } from "./model.js";
 import type { TestOpsFixtureResult } from "./model.js";
+import { UploadPacer } from "./uploadPacer.js";
 import { toUploadFixturesResultsDto } from "./utils/fixtures.js";
 import { testStatusToLaunchStatus } from "./utils/launches.js";
 import { normalizeTestStepsResults, toUploadResultsDto } from "./utils/testResults.js";
@@ -58,6 +59,20 @@ class TestOpsClientError extends AxiosError<{
 const CHUNK_SIZE = 100;
 const BULK_UPLOAD_CHUNK_SIZE = 1000;
 
+// best-effort: a Buffer/Blob's size is known upfront, a stream's isn't without consuming it,
+// so streamed attachments just don't contribute to the byte budget
+const attachmentByteLength = (content: AttachmentForUpload["content"]): number => {
+  if (Buffer.isBuffer(content)) {
+    return content.length;
+  }
+
+  if (typeof Blob !== "undefined" && content instanceof Blob) {
+    return content.size;
+  }
+
+  return 0;
+};
+
 export class TestOpsClient {
   #baseUrl: string;
   #logger = new Logger("TestOpsClient");
@@ -68,6 +83,7 @@ export class TestOpsClient {
   #session?: TestOpsSession;
   #uploadInProgress: boolean = false;
   #uploadLimit: number = 1;
+  #uploadPacer: UploadPacer;
   #namedEnvsIdsByEnv: Map<string, TestOpsNamedEnv> = new Map();
 
   constructor(params: TestOpsClientParams) {
@@ -98,6 +114,8 @@ export class TestOpsClient {
     if (params.limit) {
       this.#uploadLimit = params.limit;
     }
+
+    this.#uploadPacer = new UploadPacer(params.uploadRateLimit);
   }
 
   isTestOpsClientError(error: unknown): error is TestOpsClientError {
@@ -126,6 +144,12 @@ export class TestOpsClient {
     this.#logger.verbose("Launch closed");
   }
 
+  async reopenLaunch(launchId: number): Promise<void> {
+    this.#logger.verbose("Reopening closed launch…");
+    await this.#client.post(`/api/launch/${launchId}/reopen`);
+    this.#logger.verbose("Launch reopened");
+  }
+
   async createLaunchCategoriesBulk(
     launchId: number,
     items: LaunchCategoryBulkItem[],
@@ -339,6 +363,7 @@ export class TestOpsClient {
     }
 
     const formData = new FormData();
+    let totalBytes = 0;
 
     for (const attachmentLink of attachments) {
       const attachment = await attachmentsResolver(attachmentLink);
@@ -347,12 +372,16 @@ export class TestOpsClient {
         continue;
       }
 
+      totalBytes += attachmentByteLength(attachment.content);
+
       formData.append("file", attachment.content, {
         filename: attachment.originalFileName,
         contentType: attachment.contentType,
       });
     }
 
+    await this.#uploadPacer.wait({ requests: 1, files: attachments.length, bytes: totalBytes });
+
     await this.#client.post("/api/launch/attachment", {
       body: formData,
       onUploadProgress(progressEvent) {
@@ -374,6 +403,8 @@ export class TestOpsClient {
       throw new Error("Launch isn't created! Call createLaunch first");
     }
 
+    await this.#uploadPacer.wait({ requests: 1 });
+
     await this.#client.post("/api/launch/error/bulk", {
       body: {
         launchId: this.#launch.id,
@@ -408,51 +439,42 @@ export class TestOpsClient {
     const uploadedTrs: TestResult[] = [];
     const envNamesById = new Map(environments.map(({ id, name }) => [id, name]));
 
-    try {
-      for (const trsChunk of trsChunks) {
-        const chunkEnvs = new Map<string, EnvironmentIdentity>();
+    for (const trsChunk of trsChunks) {
+      const chunkEnvs = new Map<string, EnvironmentIdentity>();
 
-        for (const tr of trsChunk) {
-          const environmentId = tr.environment;
+      for (const tr of trsChunk) {
+        const environmentId = tr.environment;
 
-          if (environmentId && !this.#namedEnvsIdsByEnv.has(environmentId)) {
-            chunkEnvs.set(environmentId, {
-              id: environmentId,
-              name: envNamesById.get(environmentId) ?? environmentId,
-            });
-          }
+        if (environmentId && !this.#namedEnvsIdsByEnv.has(environmentId)) {
+          chunkEnvs.set(environmentId, {
+            id: environmentId,
+            name: envNamesById.get(environmentId) ?? environmentId,
+          });
         }
+      }
 
-        if (chunkEnvs.size > 0) {
-          await this.createNamedEnvs(Array.from(chunkEnvs.values()));
-        }
+      if (chunkEnvs.size > 0) {
+        await this.createNamedEnvs(Array.from(chunkEnvs.values()));
+      }
 
-        const reportIdsToTestOpsIds = await this.#postTestResultsChunk(trsChunk);
+      await this.#uploadPacer.wait({ requests: 1, files: trsChunk.length });
 
-        uploadedTrs.push(...trsChunk.filter((tr) => typeof reportIdsToTestOpsIds[tr.id] === "number"));
+      const reportIdsToTestOpsIds = await this.#postTestResultsChunk(trsChunk);
 
-        await this.#uploadChunkAttachmentsAndFixtures(
-          trsChunk,
-          reportIdsToTestOpsIds,
-          attachmentsResolver,
-          fixturesResolver,
-          uploadLimitFn,
-          onProgress,
-        );
-      }
+      uploadedTrs.push(...trsChunk.filter((tr) => typeof reportIdsToTestOpsIds[tr.id] === "number"));
 
-      this.#logger.verbose("Test results upload completed");
-    } catch (error) {
-      if (this.isTestOpsClientError(error)) {
-        this.#logger.error(`Failed to upload test results: ${error.response?.data.message}`);
-        this.#logger.debug(error.response.data);
-      } else if (error instanceof Error) {
-        this.#logger.error(`Failed to upload test results: ${error.message}`);
-      } else {
-        this.#logger.error("Failed to upload test results");
-      }
+      await this.#uploadChunkAttachmentsAndFixtures(
+        trsChunk,
+        reportIdsToTestOpsIds,
+        attachmentsResolver,
+        fixturesResolver,
+        uploadLimitFn,
+        onProgress,
+      );
     }
 
+    this.#logger.verbose("Test results upload completed");
+
     return uploadedTrs;
   }
 
@@ -519,17 +541,30 @@ export class TestOpsClient {
             return;
           }
 
-          const attachments = await attachmentsResolver(tr);
-          const fixtures = (await fixturesResolver(tr))
-            .filter((fixture) => validateExecutableName(fixture.name))
-            .map((fixture) => ({
-              ...fixture,
-              ...(fixture.steps ? { steps: normalizeTestStepsResults(fixture.steps) } : {}),
-            }));
-
-          await this.#uploadAttachmentsForResult(testOpsId, attachments as AttachmentForUpload[]);
-          await this.#uploadFixturesForResult(testOpsId, fixtures);
-          onProgress?.();
+          try {
+            const attachments = await attachmentsResolver(tr);
+            const fixtures = (await fixturesResolver(tr))
+              .filter((fixture) => validateExecutableName(fixture.name))
+              .map((fixture) => ({
+                ...fixture,
+                ...(fixture.steps ? { steps: normalizeTestStepsResults(fixture.steps) } : {}),
+              }));
+
+            await this.#uploadAttachmentsForResult(testOpsId, attachments as AttachmentForUpload[]);
+            await this.#uploadFixturesForResult(testOpsId, fixtures);
+          } catch (error) {
+            // a subordinate failure (resolver or fixture upload) shouldn't invalidate the test
+            // result itself, which TestOps has already acknowledged by this point
+            if (this.isTestOpsClientError(error)) {
+              this.#logger.error(`Failed to upload fixtures for result ${testOpsId}: ${error.response?.data.message}`);
+            } else if (error instanceof Error) {
+              this.#logger.error(`Failed to upload fixtures for result ${testOpsId}: ${error.message}`);
+            } else {
+              this.#logger.error(`Failed to upload fixtures for result ${testOpsId}`);
+            }
+          } finally {
+            onProgress?.();
+          }
         }),
       ),
     );
@@ -545,14 +580,20 @@ export class TestOpsClient {
     for (const attachmentsChunk of attachmentsChunks) {
       const formData = new FormData();
 
+      let chunkBytes = 0;
+
       for (const att of attachmentsChunk) {
+        chunkBytes += attachmentByteLength(att.content);
+
         formData.append("file", att.content, {
           filename: att.originalFileName,
           contentType: att.contentType,
         });
       }
 
       try {
+        await this.#uploadPacer.wait({ requests: 1, files: attachmentsChunk.length, bytes: chunkBytes });
+
         await this.#client.post(`/api/upload/test-result/${testOpsResultId}/attachment`, {
           body: formData,
           headers: formData.getHeaders(),
@@ -578,6 +619,8 @@ export class TestOpsClient {
 
     const body = toUploadFixturesResultsDto(fixtures);
 
+    await this.#uploadPacer.wait({ requests: 1 });
+
     await this.#client.post(`/api/upload/test-result/${testOpsResultId}/test-fixture-result`, {
       body,
     });
@@ -610,6 +653,8 @@ export class TestOpsClient {
       return item;
     });
 
+    await this.#uploadPacer.wait({ requests: 1 });
+
     await this.#client.post("/api/launch/quality-gate/bulk", {
       body: {
         launchId: this.#launch.id,