feat: release 2.5.1

server/kubernetes-state.ts

@@ -15,7 +15,8 @@ import {
 import { createHash } from "node:crypto";
 import { createKubernetesHttpLibrary } from "../lib/k8s-http";
 import { sanitizeKubernetesObject } from "../lib/kubernetes-object";
-import { type AuditEvent, type AuditPersistence } from "./audit-store";
+import { waitForRollout } from "../lib/rollout";
+import { MAX_AUDIT_EVENTS, type AuditEvent, type AuditPersistence } from "./audit-store";
 import {
   managementDependencies,
   type ManagementDependencies,
@@ -942,28 +943,75 @@ export class KubernetesOperationPersistence implements OperationPersistence {
 export class KubernetesAuditPersistence implements AuditPersistence {
   constructor(private readonly objects = createKubernetesClients().objects) {}
 
-  async append(event: AuditEvent): Promise<void> {
-    await createObject(
-      this.objects,
-      stateObject(
-        "ConfigMap",
-        event.metadata.name,
-        "audit",
-        event,
-        event.spec.workspaceId,
-      ),
+  private readonly name = "audit-log";
+
+  private async readRing(): Promise<DataObject | undefined> {
+    try {
+      return (await this.objects.read({
+        apiVersion: "v1", kind: "ConfigMap",
+        metadata: { name: this.name, namespace: KUBER_STATE_NAMESPACE },
+      })) as DataObject;
+    } catch (error) {
+      if (isNotFound(error)) return;
+      throw error;
+    }
+  }
+
+  private async legacy(): Promise<DataObject[]> {
+    const result = await this.objects.list(
+      "v1", "ConfigMap", KUBER_STATE_NAMESPACE, undefined, undefined,
+      undefined, undefined, `${TYPE_LABEL}=audit`,
     );
+    return result.items.map((item) => ({
+      apiVersion: "v1", kind: "ConfigMap", ...item,
+    })) as DataObject[];
+  }
+
+  private async reconcile(add?: AuditEvent): Promise<AuditEvent[]> {
+    for (let attempt = 0; attempt < 128; attempt++) {
+      const [ring, legacy] = await Promise.all([this.readRing(), this.legacy()]);
+      const current = ring && parsePayload<{ kind: string; items: AuditEvent[] }>(ring);
+      const events = new Map<string, AuditEvent>();
+      for (const event of [...(current?.items ?? []), ...legacy.map(parsePayload<AuditEvent>).filter((event): event is AuditEvent => event?.kind === "AuditEvent"), ...(add ? [add] : [])]) {
+        if (event?.kind === "AuditEvent") events.set(event.metadata.uid || event.metadata.name, event);
+      }
+      const ordered = [...events.values()].sort((a, b) =>
+        a.metadata.creationTimestamp.localeCompare(b.metadata.creationTimestamp) ||
+        a.metadata.name.localeCompare(b.metadata.name),
+      );
+      let retained = ordered.slice(-MAX_AUDIT_EVENTS);
+      while (retained.length && Buffer.byteLength(JSON.stringify({ kind: "AuditLog", items: retained })) > 768 * 1024) retained.shift();
+      const value = stateObject("ConfigMap", this.name, "audit-log", { kind: "AuditLog", items: retained });
+      try {
+        if (ring) {
+          value.metadata!.resourceVersion = ring.metadata?.resourceVersion;
+          await this.objects.replace(value);
+        } else {
+          await this.objects.create(value);
+        }
+      } catch (error) {
+        if (isConflict(error)) continue;
+        throw error;
+      }
+      // Never remove legacy records until the merged ring has been committed.
+      for (const item of legacy.slice(0, 100)) {
+        const name = item.metadata?.name;
+        if (name) await deleteObject(this.objects, "ConfigMap", name);
+      }
+      return retained;
+    }
+    throw new Error("Audit log was concurrently modified; retry limit reached");
+  }
+
+  async append(event: AuditEvent): Promise<void> {
+    await this.reconcile(event);
   }
 
   async list(workspaceId?: string): Promise<AuditEvent[]> {
-    return (await list(this.objects, "ConfigMap", "audit", workspaceId))
-      .map((item) => parsePayload<AuditEvent>(item))
-      .filter((item): item is AuditEvent => item?.kind === "AuditEvent")
-      .sort((a, b) =>
-        a.metadata.creationTimestamp.localeCompare(
-          b.metadata.creationTimestamp,
-        ),
-      );
+    const events = await this.reconcile();
+    return events.filter((event) =>
+      workspaceId ? event.spec.workspaceId === workspaceId : true,
+    );
   }
 }
 
@@ -1032,34 +1080,10 @@ function resource(identity: ResourceIdentity): KubernetesObject {
   };
 }
 
-async function sleepUntilExecutionCancelled(
-  delayMs: number,
-  execution?: OperationExecution,
-): Promise<void> {
-  const signal = execution?.signal;
-  if (!signal) {
-    await Bun.sleep(delayMs);
-    return;
-  }
-  if (signal.aborted)
-    throw new Error("Workspace operation execution was cancelled");
-  await new Promise<void>((resolve, reject) => {
-    const timer = setTimeout(() => {
-      signal.removeEventListener("abort", cancelSleep);
-      resolve();
-    }, delayMs);
-    const cancelSleep = () => {
-      clearTimeout(timer);
-      reject(new Error("Workspace operation execution was cancelled"));
-    };
-    signal.addEventListener("abort", cancelSleep, { once: true });
-  });
-}
-
 export function createKubernetesManagementDependencies(
   clients = createKubernetesClients(),
 ): ManagementDependencies {
-  const { objects, apps } = clients;
+  const { objects, apps, core } = clients;
   const revision = (replicaSet: V1ReplicaSet): number | undefined => {
     const value = Number(
       replicaSet.metadata?.annotations?.["deployment.kubernetes.io/revision"],
@@ -1171,30 +1195,17 @@ export function createKubernetesManagementDependencies(
       name,
       timeoutMs = 300_000,
       execution?: OperationExecution,
-    ) => {
-      const started = Date.now();
-      while (Date.now() - started < timeoutMs) {
-        if (execution?.signal?.aborted)
-          throw new Error("Workspace operation execution was cancelled");
-        const deployment = await apps.readNamespacedDeployment({
-          namespace: project,
-          name,
-        });
-        const desired = deployment.spec?.replicas ?? 1;
-        if (
-          (deployment.status?.observedGeneration ?? 0) >=
-            (deployment.metadata?.generation ?? 0) &&
-          (deployment.status?.updatedReplicas ?? 0) === desired &&
-          (deployment.status?.availableReplicas ?? 0) === desired &&
-          (deployment.status?.unavailableReplicas ?? 0) === 0
-        )
-          return;
-        if (execution?.signal?.aborted)
-          throw new Error("Workspace operation execution was cancelled");
-        await sleepUntilExecutionCancelled(2_000, execution);
-      }
-      throw new Error(`Timed out waiting for deployment ${name} rollout`);
-    },
+    ) =>
+      waitForRollout(
+        name,
+        Date.now() + timeoutMs,
+        () => apps.readNamespacedDeployment({ namespace: project, name }),
+        async () =>
+          (await apps.listNamespacedReplicaSet({ namespace: project })).items,
+        async () =>
+          (await core.listNamespacedPod({ namespace: project })).items,
+        execution?.signal,
+      ),
     planRollback: async (project, names) => {
       const deployments = await overrides.listDeployments!(project);
       const selected = names