fix: stabilize live deployment progress

command/up.ts

@@ -73,6 +73,7 @@ export type OperationResumeOptions = {
   now?: () => number;
   sleep?: (milliseconds: number) => Promise<void>;
   onEvent?: (event: ResourceOperationEvent) => void;
+  signal?: AbortSignal;
 };
 
 type OperationEventsResponse = {
@@ -335,6 +336,41 @@ function operationResumeDeadline(
   return now + Math.max(rolloutTimeoutMs, 0) + OPERATION_RESUME_GRACE_MS;
 }
 
+function abortReason(signal: AbortSignal): unknown {
+  return (
+    signal.reason ?? new DOMException("The operation was aborted", "AbortError")
+  );
+}
+
+function throwIfAborted(signal: AbortSignal | undefined): void {
+  if (signal?.aborted) throw abortReason(signal);
+}
+
+function abortableDelay(
+  milliseconds: number,
+  signal: AbortSignal | undefined,
+): Promise<void> {
+  throwIfAborted(signal);
+  return new Promise<void>((resolve, reject) => {
+    let timeout: ReturnType<typeof setTimeout> | undefined;
+    const cleanup = () => {
+      if (timeout !== undefined) clearTimeout(timeout);
+      signal?.removeEventListener("abort", abort);
+    };
+    const complete = () => {
+      cleanup();
+      resolve();
+    };
+    const abort = () => {
+      cleanup();
+      reject(abortReason(signal!));
+    };
+    timeout = setTimeout(complete, milliseconds);
+    signal?.addEventListener("abort", abort, { once: true });
+    if (signal?.aborted) abort();
+  });
+}
+
 async function resumeManagedOperation(
   project: string,
   request: ApiRequester,
@@ -344,18 +380,22 @@ async function resumeManagedOperation(
   options: OperationResumeOptions,
 ): Promise<void> {
   const now = options.now ?? Date.now;
-  const sleep = options.sleep ?? ((milliseconds) => Bun.sleep(milliseconds));
   const deadline = operationResumeDeadline(rolloutTimeoutMs, now());
   const headers = new Headers(init.headers);
   headers.set("idempotency-key", randomUUID());
   headers.set("prefer", "respond-async");
-  let operationInit = { ...init, headers };
+  let operationInit = {
+    ...init,
+    headers,
+    signal: options.signal ?? init.signal,
+  };
   let operationId: string | undefined;
   let backoffMs = OPERATION_RESUME_INITIAL_BACKOFF_MS;
   let restartRetryPending = false;
   let eventCursor = 0;
 
   for (;;) {
+    throwIfAborted(options.signal);
     if (restartRetryPending && now() >= deadline)
       throw new Error("Timed out while reconnecting to resume the operation");
     try {
@@ -365,7 +405,7 @@ async function resumeManagedOperation(
           project,
           request,
           `/operations/${encodeURIComponent(operationId)}`,
-          {},
+          { signal: options.signal },
         );
         status = operation.status;
       } else {
@@ -386,7 +426,7 @@ async function resumeManagedOperation(
             project,
             request,
             `/operations/${encodeURIComponent(operationId)}/events?after=${eventCursor}`,
-            {},
+            { signal: options.signal },
           );
           if (hasOperationEventCursorGap(events, eventCursor))
             throw new OperationProgressCursorGapError();
@@ -402,6 +442,7 @@ async function resumeManagedOperation(
           }
         } catch (error) {
           if (error instanceof OperationProgressCursorGapError) throw error;
+          throwIfAborted(options.signal);
           // Event polling is additive; never delay resumable operation status polling.
         }
       }
@@ -413,10 +454,15 @@ async function resumeManagedOperation(
         eventCursor = 0;
         restartRetryPending = true;
         headers.set("idempotency-key", randomUUID());
-        operationInit = { ...init, headers };
+        operationInit = {
+          ...init,
+          headers,
+          signal: options.signal ?? init.signal,
+        };
       } else if (status.state === "failed" || status.state === "cancelled")
         throw operationFailure(operationId, status);
     } catch (error) {
+      throwIfAborted(options.signal);
       if (
         error instanceof KuberApiError &&
         error.code === "OPERATION_INTERRUPTED"
@@ -426,7 +472,11 @@ async function resumeManagedOperation(
         eventCursor = 0;
         restartRetryPending = true;
         headers.set("idempotency-key", randomUUID());
-        operationInit = { ...init, headers };
+        operationInit = {
+          ...init,
+          headers,
+          signal: options.signal ?? init.signal,
+        };
       } else {
         if (!isRecoverableConnectionInterruption(error))
           throw adoptionHint(project, error);
@@ -437,7 +487,12 @@ async function resumeManagedOperation(
     const remainingMs = deadline - now();
     if (remainingMs <= 0)
       throw new Error("Timed out while reconnecting to resume the operation");
-    await sleep(Math.min(backoffMs, remainingMs));
+    const delayMs = Math.min(backoffMs, remainingMs);
+    if (options.sleep) {
+      throwIfAborted(options.signal);
+      await options.sleep(delayMs);
+      throwIfAborted(options.signal);
+    } else await abortableDelay(delayMs, options.signal);
     backoffMs = Math.min(backoffMs * 2, OPERATION_RESUME_MAX_BACKOFF_MS);
   }
 }
@@ -561,52 +616,59 @@ function resourceOperationTaskTitle(
   return `${action} ${resource.kind}/${resource.name}`;
 }
 
-export async function runLiveResourceOperation(
+export function runLiveResourceOperation(
   task: ListrTaskWrapper<any, any, any>,
   phase: ResourceOperationEvent["data"]["phase"],
   targets: ResourceOperationTarget[],
   operation: (
     onEvent: (event: ResourceOperationEvent) => void,
+    signal: AbortSignal,
   ) => Promise<void>,
   options: LiveResourceOperationOptions = {},
-): Promise<void> {
-  if (targets.length === 0) return operation(() => {});
+): Listr<ResourceOperationTarget, any, any> | Promise<void> {
+  if (targets.length === 0) return operation(() => {}, task.signal);
   const active = new Map<
     string,
     ListrTaskWrapper<ResourceOperationTarget, any, any>
   >();
-  const complete: Array<() => void> = [];
+  const complete = new Set<() => void>();
   let started = 0;
-  let markStarted!: () => void;
-  const ready = new Promise<void>((resolve) => {
-    markStarted = resolve;
-  });
-  const children = task.newListr<ResourceOperationTarget>(
+  return task.newListr<ResourceOperationTarget>(
     targets.map((target) => ({
       title: resourceOperationTaskTitle(phase, target),
-      task: (_target, child) => {
+      task: async (_target, child) => {
         active.set(resourceOperationKey(target), child);
         options.onTaskStarted?.(target, child);
         started += 1;
-        if (started === targets.length) markStarted();
-        return new Promise<void>((resolve) => complete.push(resolve));
+        if (started !== targets.length)
+          return new Promise<void>((resolve) => {
+            const completeTask = () => {
+              task.signal.removeEventListener("abort", completeTask);
+              complete.delete(completeTask);
+              resolve();
+            };
+            complete.add(completeTask);
+            if (task.signal.aborted) completeTask();
+            else
+              task.signal.addEventListener("abort", completeTask, {
+                once: true,
+              });
+          });
+        try {
+          await operation((event) => {
+            const resource = event.data.resource;
+            const activeTask = active.get(resourceOperationKey(resource));
+            if (!activeTask) return;
+            activeTask.output = event.data.state;
+            task.output = resourceOperationEventTitle(event);
+          }, task.signal);
+        } finally {
+          for (const resolve of [...complete]) resolve();
+        }
       },
     })),
+    { concurrent: true },
   );
-  const childRun = children.run();
-  await ready;
-  try {
-    await operation((event) => {
-      const resource = event.data.resource;
-      const child = active.get(resourceOperationKey(resource));
-      if (!child) return;
-      child.output = event.data.state;
-      task.output = resourceOperationEventTitle(event);
-    });
-  } finally {
-    for (const resolve of complete) resolve();
-    await childRun;
-  }
 }
 
 export async function runUp(
@@ -801,14 +863,14 @@ export async function runUp(
                   phaseTask,
                   "apply",
                   desired,
-                  (onEvent) =>
+                  (onEvent, signal) =>
                     resumeManagedOperation(
                       project,
                       request,
                       `${resourcePath}/apply`,
                       { method: "POST", json: { resources: plan.desired } },
                       config.rolloutTimeoutMs,
-                      { onEvent },
+                      { onEvent, signal },
                     ),
                 ),
             },
@@ -829,7 +891,7 @@ export async function runUp(
                   phaseTask,
                   "wait",
                   deployments,
-                  (onEvent) =>
+                  (onEvent, signal) =>
                     resumeManagedOperation(
                       project,
                       request,
@@ -844,7 +906,7 @@ export async function runUp(
                         },
                       },
                       config.rolloutTimeoutMs,
-                      { onEvent },
+                      { onEvent, signal },
                     ),
                 ),
             },
@@ -856,14 +918,14 @@ export async function runUp(
                   phaseTask,
                   "delete",
                   stale,
-                  (onEvent) =>
+                  (onEvent, signal) =>
                     resumeManagedOperation(
                       project,
                       request,
                       `${resourcePath}/delete`,
                       { method: "POST", json: { resources: plan.stale } },
                       config.rolloutTimeoutMs,
-                      { onEvent },
+                      { onEvent, signal },
                     ),
                 ),
             },