feat: prepare 2.6.1-rc5 shared databases and build SSE

server/app.ts

@@ -1,9 +1,10 @@
 import type { KubernetesObject } from "@kubernetes/client-node";
 import { randomUUID } from "node:crypto";
 import type { ComposeSpecification } from "../schema/docker.d";
-import { DatabaseReconciliationError } from "../lib/database";
+import { DATABASE_RECONCILE_PHASES, DatabaseReconciliationError } from "../lib/database";
 import type { Operation as PublicOperation } from "../shared/api";
 import type { BuildRequest, Sha256Digest } from "../shared/build-protocol";
+import { KUBER_VERSION, KUBER_VERSION_HEADER } from "../shared/version";
 import {
   createToken,
   hashToken,
@@ -28,6 +29,7 @@ import {
   BuildValidationError,
   type BuildController,
 } from "./build-controller";
+import { BuildEventStreamHub } from "./build-event-stream";
 import { KubernetesLogError, type LogService } from "./log-service";
 import {
   ExecService,
@@ -142,6 +144,11 @@ export type UnknownFailureLog = {
 export type RequestErrorLog = Omit<UnknownFailureLog, "event"> & {
   event: string;
   operationId?: string;
+  phase?: string;
+  errorClass?: string;
+  providerStatus?: number;
+  providerCode?: string;
+  topFrame?: string;
 };
 
 export interface AppLogger {
@@ -200,6 +207,62 @@ class HttpError extends Error {
   }
 }
 
+const DATABASE_PHASES = new Set<string>(DATABASE_RECONCILE_PHASES);
+const DATABASE_ERROR_CLASSES = new Set([
+  "Error", "TypeError", "RangeError", "SyntaxError", "AbortError",
+  "ApiException", "ResponseError", "FetchError", "TimeoutError",
+]);
+const DATABASE_ERROR_CODES = new Set([
+  "ECONNREFUSED", "ECONNRESET", "ETIMEDOUT", "EHOSTUNREACH",
+  "ENOTFOUND", "EAI_AGAIN", "ABORT_ERR", "UND_ERR_CONNECT_TIMEOUT",
+]);
+
+/** Never serialize an unknown error: its message, stack and provider fields may contain secrets. */
+function databaseFailureDiagnostics(error: unknown): Record<string, unknown> {
+  let current = error;
+  let phase = "operation execution";
+  const seen = new Set<unknown>();
+  for (
+    let depth = 0;
+    depth < 5 && current && typeof current === "object" && !seen.has(current);
+    depth++
+  ) {
+    seen.add(current);
+    if (
+      current instanceof DatabaseReconciliationError &&
+      DATABASE_PHASES.has(current.phase)
+    ) {
+      phase = current.phase;
+    }
+    const cause = (current as { cause?: unknown }).cause;
+    if (!cause || typeof cause !== "object" || seen.has(cause)) break;
+    current = cause;
+  }
+  const source = current && typeof current === "object"
+    ? current as Record<string, unknown> : {};
+  const errorClass = current instanceof Error && DATABASE_ERROR_CLASSES.has(current.name)
+    ? current.name
+    : "UnknownError";
+  const status = [source.statusCode, source.status, source.code].find(
+    (value) => typeof value === "number" && Number.isInteger(value) &&
+      value >= 400 && value < 600,
+  );
+  const errorCode = typeof source.code === "string" && DATABASE_ERROR_CODES.has(source.code)
+    ? source.code
+    : undefined;
+  // Only the first frame and a fixed set of our source files; never log raw stacks or paths.
+  const firstFrame = current instanceof Error ? current.stack?.split("\n")[1] : undefined;
+  const frame = firstFrame &&
+    /(?:^|\/)\b((?:lib\/database|server\/management|server\/app)\.ts):(\d+):(\d+)\b/.exec(firstFrame);
+  return {
+    phase,
+    errorClass,
+    ...(status !== undefined && { providerStatus: status }),
+    ...(errorCode && { providerCode: errorCode }),
+    ...(frame && { topFrame: `${frame[1]}:${frame[2]}:${frame[3]}` }),
+  };
+}
+
 function isRecord(value: unknown): value is Record<string, unknown> {
   return typeof value === "object" && value !== null && !Array.isArray(value);
 }
@@ -282,6 +345,9 @@ export function createApp(
   const now = options.now ?? Date.now;
   const makeRequestId = options.requestId ?? randomUUID;
   const logger = options.logger ?? defaultAppLogger;
+  const buildEventStreams = options.builds
+    ? new BuildEventStreamHub(options.builds)
+    : undefined;
   const bodyLimit = options.jsonBodyLimit ?? DEFAULT_JSON_LIMIT;
   const allowedOrigins = new Set(options.allowedOrigins ?? []);
   const hasOriginConfiguration = options.allowedOrigins !== undefined;
@@ -1150,25 +1216,43 @@ export function createApp(
         });
       } catch (error) {
         if (!operationStarted) throw error;
-        const message = error instanceof Error ? error.message : String(error);
         const leaseLost =
           leaseOwnershipLost ||
           (error instanceof HttpError && error.code === "WORKSPACE_LEASE_LOST");
+        const databaseFailure =
+          action === "databases.reconcile" && !leaseLost
+            ? error instanceof DatabaseReconciliationError
+              ? error
+              : new DatabaseReconciliationError("operation execution", error)
+            : undefined;
+        if (databaseFailure) {
+          logger.error({
+            event: "operation.database_reconcile.failed",
+            requestId: requestIds.get(request) ?? makeRequestId(),
+            method: "POST",
+            pathname: "/api/v2/workspaces/:workspaceId/databases",
+            operationId: operation.metadata.name,
+            status: 500,
+            code: "DATABASE_RECONCILE_FAILED",
+            errorName: "DatabaseReconciliationError",
+            message: "Database reconciliation failed",
+            ...databaseFailureDiagnostics(databaseFailure),
+          });
+        }
+        const message =
+          databaseFailure?.message ??
+          (error instanceof Error ? error.message : String(error));
         const failed = await transitionOperationToFailure(
           operation.metadata.name,
           {
             code: leaseLost
               ? "WORKSPACE_LEASE_LOST"
-              : action === "databases.reconcile" &&
-                  error instanceof DatabaseReconciliationError
+              : databaseFailure
                 ? "DATABASE_RECONCILE_FAILED"
                 : "OPERATION_FAILED",
             message: leaseLost
               ? "Workspace operation lease ownership was lost"
-              : action === "databases.reconcile" &&
-                  !(error instanceof DatabaseReconciliationError)
-                ? "Database reconciliation failed"
-                : message,
+              : message,
           },
         );
         const failure = failed.status.error;
@@ -1470,13 +1554,31 @@ export function createApp(
       );
       if (!action && request.method === "GET")
         return response(await builds.getBuildStatus(id));
-      if (action === "events" && request.method === "GET")
+      if (action === "events" && request.method === "GET") {
+        if (request.headers.get("accept")?.split(",").some((value) =>
+          value.trim().split(";")[0]?.trim().toLowerCase() === "text/event-stream",
+        )) {
+          const cursor = request.headers.get("last-event-id") ?? url.searchParams.get("after");
+          if (
+            cursor !== null &&
+            (!/^(0|[1-9]\d*)$/.test(cursor) ||
+              !Number.isSafeInteger(Number(cursor)))
+          )
+            throw new HttpError(
+              400,
+              "Invalid cursor",
+              "INVALID_QUERY",
+              "Event cursor must be a non-negative safe integer",
+            );
+          return buildEventStreams!.open(id, request);
+        }
         return response(
           await builds.getBuildEvents(
             id,
             nonNegativeInteger(url.searchParams.get("after"), "after") ?? 0,
           ),
         );
+      }
       if (action === "reconcile" && request.method === "POST")
         return response(await builds.reconcileBuild(id));
       if (action === "cancel" && request.method === "POST")
@@ -2553,6 +2655,7 @@ export function createApp(
           result = problem(normalized, requestId);
         }
         result.headers.set("x-request-id", requestId);
+        result.headers.set(KUBER_VERSION_HEADER, KUBER_VERSION);
         if (origin && originAllowed) {
           result.headers.set("access-control-allow-origin", origin);
           result.headers.set("vary", "Origin");
@@ -2689,6 +2792,7 @@ export function execProblem(error: unknown, requestId = ""): Response {
     },
   );
   result.headers.set("content-type", "application/problem+json");
+  result.headers.set(KUBER_VERSION_HEADER, KUBER_VERSION);
   return result;
 }
 

server/build-event-stream.ts

@@ -0,0 +1,255 @@
+import type { BuildController } from "./build-controller";
+import type { BuildStatus } from "../shared/build-protocol";
+
+const terminal = (status: BuildStatus) =>
+  status.state === "succeeded" || status.state === "failed";
+
+function cursor(value: string | null): number {
+  if (value === null) return 0;
+  if (!/^(0|[1-9]\d*)$/.test(value) || !Number.isSafeInteger(Number(value)))
+    throw new RangeError("Event cursor must be a non-negative safe integer");
+  return Number(value);
+}
+
+type Subscriber = {
+  sequence: number;
+  status?: string;
+  lastActivity: number;
+  controller: ReadableStreamDefaultController<Uint8Array>;
+  stop: () => void;
+};
+
+type Group = {
+  members: Set<Subscriber>;
+  abort: AbortController;
+  nextReconcileAt: number;
+  reconciling: boolean;
+};
+
+/** One hub per app/controller instance; call only after the route's capability and workspace checks.
+ * The authenticated GET /api/v2/builds/:id/events route should branch on
+ * Accept: text/event-stream, then return hub.open(id, request); retain JSON otherwise.
+ * A hub coalesces reconciliation and persisted event reads per physical build on this
+ * replica. BuildController's lease coordinates reconciliation across replicas; its
+ * persisted event store supplies logs even when a different replica owns the job.
+ */
+export class BuildEventStreamHub {
+  private readonly builds = new Map<string, Group>();
+
+  constructor(
+    private readonly controller: Pick<
+      BuildController,
+      "getBuildStatus" | "getBuildEvents" | "reconcileBuild"
+    >,
+    private readonly options: {
+      pollMs?: number;
+      heartbeatMs?: number;
+      reconcileMs?: number;
+    } = {},
+  ) {}
+
+  async open(id: string, request: Request): Promise<Response> {
+    const header = request.headers.get("last-event-id");
+    const query = new URL(request.url).searchParams.get("after");
+    const after = cursor(header ?? query);
+    // Validate existence before committing response headers.
+    await this.controller.getBuildStatus(id);
+    let subscriber: Subscriber;
+    const stream = new ReadableStream<Uint8Array>(
+      {
+        start: (controller) => {
+          let group: Group;
+          const stop = () => {
+            request.signal.removeEventListener("abort", stop);
+            group.members.delete(subscriber);
+            if (group.members.size === 0 && this.builds.get(id) === group) {
+              this.builds.delete(id);
+              group.abort.abort();
+            }
+            try {
+              controller.close();
+            } catch {
+              /* Already closed or cancelled. */
+            }
+          };
+          subscriber = {
+            sequence: after,
+            lastActivity: Date.now(),
+            controller,
+            stop,
+          };
+          const existing = this.builds.get(id);
+          if (existing) {
+            group = existing;
+            group.members.add(subscriber);
+          } else {
+            group = {
+              members: new Set([subscriber]),
+              abort: new AbortController(),
+              nextReconcileAt: 0,
+              reconciling: false,
+            };
+            this.builds.set(id, group);
+            void this.run(id, group);
+          }
+          request.signal.addEventListener("abort", stop, { once: true });
+          if (request.signal.aborted) stop();
+        },
+        cancel: () => subscriber.stop(),
+      },
+      { highWaterMark: 16 },
+    );
+    return new Response(stream, {
+      headers: {
+        "content-type": "text/event-stream; charset=utf-8",
+        "cache-control": "no-cache, no-transform",
+        connection: "keep-alive",
+        "x-accel-buffering": "no",
+      },
+    });
+  }
+
+  private async run(id: string, group: Group): Promise<void> {
+    const encoder = new TextEncoder();
+    const pollMs = this.options.pollMs ?? 1_000;
+    const reconcileMs = this.options.reconcileMs ?? 7_000;
+    const { members } = group;
+    // Bun's default HTTP idle timeout is 10 seconds; keep bytes flowing well
+    // inside it even while Kubernetes reconciliation makes no progress.
+    const heartbeatMs = this.options.heartbeatMs ?? 5_000;
+    const heartbeat = setInterval(() => {
+      if (group.abort.signal.aborted) return;
+      for (const member of members) {
+        if (Date.now() - member.lastActivity < heartbeatMs) continue;
+        if ((member.controller.desiredSize ?? 0) <= 0) member.stop();
+        else {
+          member.controller.enqueue(encoder.encode(": heartbeat\n\n"));
+          member.lastActivity = Date.now();
+        }
+      }
+    }, heartbeatMs);
+    heartbeat.unref?.();
+    try {
+      while (!group.abort.signal.aborted) {
+        try {
+          const initial = await this.controller.getBuildStatus(id);
+          if (group.abort.signal.aborted) break;
+          // Reconciliation is shared per build and never blocks persisted progress.
+          // The controller's lease still coordinates attempts across replicas.
+          if (
+            !terminal(initial) &&
+            !group.reconciling &&
+            Date.now() >= group.nextReconcileAt
+          ) {
+            group.reconciling = true;
+            group.nextReconcileAt = Date.now() + reconcileMs;
+            const signal = group.abort.signal;
+            let onAbort = () => {};
+            const aborted = new Promise<void>((resolve) => {
+              onAbort = resolve;
+              signal.addEventListener("abort", onAbort, { once: true });
+            });
+            const reconciliation = this.controller
+              .reconcileBuild(id, { signal })
+              .catch(() => {
+                // Another replica may hold the lease; keep observing its progress.
+              });
+            // Drop this group's references even if an external API ignores abort
+            // and leaves the reconciliation promise unresolved indefinitely.
+            void Promise.race([reconciliation, aborted]).finally(() => {
+              signal.removeEventListener("abort", onAbort);
+              group.reconciling = false;
+            });
+          }
+          const status = await this.controller.getBuildStatus(id);
+          if (group.abort.signal.aborted) break;
+          const oldest = Math.min(
+            ...[...members].map((member) => member.sequence),
+          );
+          const events = await this.controller.getBuildEvents(id, oldest);
+          if (group.abort.signal.aborted) break;
+          const firstLogSequence = events.find(
+            (event) => event.type === "log",
+          )?.sequence;
+          for (const member of members) {
+            const send = (frame: string) => {
+              // A slow consumer reconnects from its last delivered sequence instead
+              // of retaining an unbounded in-memory backlog on this replica.
+              if ((member.controller.desiredSize ?? 0) <= 0) {
+                member.stop();
+                return false;
+              }
+              member.controller.enqueue(encoder.encode(frame));
+              member.lastActivity = Date.now();
+              return true;
+            };
+            if (
+              firstLogSequence !== undefined &&
+              firstLogSequence > member.sequence + 1
+            ) {
+              const gap = {
+                type: "gap",
+                after: member.sequence,
+                before: firstLogSequence,
+                missing: firstLogSequence - member.sequence - 1,
+                message: `Build log history was trimmed; ${firstLogSequence - member.sequence - 1} log event(s) before sequence ${firstLogSequence} are unavailable.`,
+              };
+              if (
+                !send(
+                  `id: ${firstLogSequence - 1}\nevent: gap\ndata: ${JSON.stringify(gap)}\n\n`,
+                )
+              )
+                continue;
+              // Advance over the unavailable range so reconnects do not repeat
+              // the marker. The cursor now refers to the last unavailable
+              // sequence; delivered log records retain their normal IDs.
+              member.sequence = firstLogSequence - 1;
+            }
+            for (const event of events) {
+              if (event.type !== "log" || event.sequence <= member.sequence)
+                continue;
+              if (
+                !send(
+                  `id: ${event.sequence}\nevent: log\ndata: ${JSON.stringify(event)}\n\n`,
+                )
+              )
+                break;
+              member.sequence = event.sequence;
+            }
+            if (!members.has(member)) continue;
+            const fingerprint = JSON.stringify(status);
+            if (member.status !== fingerprint) {
+              if (
+                !send(
+                  `event: status\ndata: ${JSON.stringify({ type: "status", status })}\n\n`,
+                )
+              )
+                continue;
+              member.status = fingerprint;
+            }
+            if (terminal(status)) member.stop();
+          }
+        } catch {
+          // A transient API/store failure closes streams so clients reconnect with
+          // their last sequence; no silent terminal state is fabricated.
+          for (const member of members) member.stop();
+        }
+        if (group.abort.signal.aborted) break;
+        await new Promise<void>((resolve) => {
+          const stop = () => {
+            clearTimeout(timer);
+            resolve();
+          };
+          const timer = setTimeout(() => {
+            group.abort.signal.removeEventListener("abort", stop);
+            resolve();
+          }, pollMs);
+          group.abort.signal.addEventListener("abort", stop, { once: true });
+        });
+      }
+    } finally {
+      clearInterval(heartbeat);
+      group.abort.abort();
+    }
+  }
+}

server/management.ts

@@ -8,7 +8,9 @@ import {
   sortResources,
 } from "../lib/apply";
 import {
+  DATABASE_CLUSTER,
   DATABASE_NAMESPACE,
+  DatabaseReconciliationError,
   getComposePostgresClaims,
   getRoleCredentials,
   listManagedDatabaseResources,
@@ -494,8 +496,15 @@ export function createManagementService(dependencies: ManagementDependencies) {
         );
 
     if (full) {
+      // Database CRs may have consumers in other projects. Labels are not a
+      // reference count, including when this workspace created the CR.
+      const databases = await dependencies.listDatabaseResources(workspace.project);
+      retainedResources.push(...databases.filter(
+        (resource) => resource.kind === "Database" &&
+          resource.metadata?.labels?.[WORKSPACE_UID_LABEL] === workspace.uid,
+      ));
       deleteResources.push(
-        ...(await dependencies.listDatabaseResources(workspace.project)),
+        ...databases.filter((resource) => resource.kind !== "Database"),
         ...(await dependencies.listStorageResources(workspace.project)),
       );
       if (!safety.namespaceUid) {
@@ -640,19 +649,56 @@ export function createManagementService(dependencies: ManagementDependencies) {
       compose: ComposeSpecification,
       execution?: OperationExecution,
     ) {
-      throwIfExecutionAborted(execution);
-      await assertSafe(workspace, true);
-      const environment = await dependencies.reconcileDatabases(
-        workspace.project,
-        compose,
-        execution,
-      );
-      await ownExternalResources(
-        workspace,
-        DATABASE_NAMESPACE,
-        await dependencies.listDatabaseResources(workspace.project),
-        execution,
-      );
+      try {
+        throwIfExecutionAborted(execution);
+        await assertSafe(workspace, true);
+      } catch (error) {
+        throw new DatabaseReconciliationError("namespace precheck", error);
+      }
+      let environment: Record<string, Record<string, string>>;
+      try {
+        environment = await dependencies.reconcileDatabases(
+          workspace.project,
+          compose,
+          execution,
+        );
+      } catch (error) {
+        if (error instanceof DatabaseReconciliationError) throw error;
+        throw new DatabaseReconciliationError("database dependency", error);
+      }
+      let resources: KubernetesObject[];
+      try {
+        resources = await dependencies.listDatabaseResources(workspace.project);
+      } catch (error) {
+        throw new DatabaseReconciliationError("database resource listing", error);
+      }
+      try {
+        const claims = getComposePostgresClaims(compose);
+        const claimed = new Set(claims.map((claim) => claim.database));
+        for (const resource of resources) {
+          // Only a declared database may be shared. Never adopt its foreign
+          // workspace UID or apply its metadata through this field manager.
+          const existingOwner = resource.metadata?.labels?.[WORKSPACE_UID_LABEL];
+          if (resource.kind === "Database" && existingOwner &&
+            existingOwner !== workspace.uid &&
+            claimed.has(resource.metadata?.name ?? "")) {
+            if (resource.metadata?.namespace !== DATABASE_NAMESPACE) {
+              throw new Error("Database is outside expected namespace");
+            }
+            const claim = claims.find((item) => item.database === resource.metadata?.name);
+            const spec = (resource as KubernetesObject & {
+              spec?: { owner?: string; cluster?: { name?: string } };
+            }).spec;
+            if (spec?.owner !== claim?.username || spec?.cluster?.name !== DATABASE_CLUSTER) {
+              throw new Error("Database does not match the declared claim");
+            }
+            continue;
+          }
+          await ownExternalResources(workspace, DATABASE_NAMESPACE, [resource], execution);
+        }
+      } catch (error) {
+        throw new DatabaseReconciliationError("database resource ownership", error);
+      }
       return environment;
     },