diff --git a/web/actions/projects.ts b/web/actions/projects.ts index 057e8788..00cdfd7f 100644 --- a/web/actions/projects.ts +++ b/web/actions/projects.ts @@ -1194,32 +1194,38 @@ export async function restartService(serviceId: string) { export async function abortRollout(serviceId: string) { await requireAuth(); - const updatedRollouts = await db - .update(rollouts) - .set({ - status: "failed", - currentStage: "aborted", - completedAt: new Date(), - }) + const activeRollouts = await db + .select({ id: rollouts.id, status: rollouts.status }) + .from(rollouts) .where( and( eq(rollouts.serviceId, serviceId), - eq(rollouts.status, "in_progress"), + inArray(rollouts.status, ["queued", "in_progress"]), ), - ) - .returning(); - - const inProgressRollout = updatedRollouts[0]; + ); - if (!inProgressRollout) { + if (activeRollouts.length === 0) { return { success: false, error: "No in-progress rollout found" }; } - await inngest.send( - inngestEvents.rolloutCancelled.create({ - rolloutId: inProgressRollout.id, - }), - ); + const activeRolloutIds = activeRollouts.map((rollout) => rollout.id); + + await db + .update(rollouts) + .set({ + status: "failed", + currentStage: "aborted", + completedAt: new Date(), + }) + .where(inArray(rollouts.id, activeRolloutIds)); + + for (const rolloutId of activeRolloutIds) { + await inngest.send( + inngestEvents.rolloutCancelled.create({ + rolloutId, + }), + ); + } await db .update(deployments) @@ -1231,10 +1237,13 @@ export async function abortRollout(serviceId: string) { ), ); - const rolloutDeployments = await db - .select() - .from(deployments) - .where(eq(deployments.rolloutId, inProgressRollout.id)); + const rolloutDeployments = + activeRolloutIds.length > 0 + ? await db + .select() + .from(deployments) + .where(inArray(deployments.rolloutId, activeRolloutIds)) + : []; const serverContainers = new Map(); @@ -1259,20 +1268,25 @@ export async function abortRollout(serviceId: string) { .where(eq(deploymentPorts.deploymentId, dep.id)); } - await db - .delete(deployments) - .where(eq(deployments.rolloutId, inProgressRollout.id)); - - const pendingWork = await db - .select({ id: workQueue.id, payload: workQueue.payload }) - .from(workQueue) - .where( - and( - eq(workQueue.status, "pending"), - inArray(workQueue.type, ["deploy", "reconcile"]), - inArray(workQueue.serverId, [...serverContainers.keys()]), - ), - ); + if (activeRolloutIds.length > 0) { + await db + .delete(deployments) + .where(inArray(deployments.rolloutId, activeRolloutIds)); + } + + const pendingWork = + serverContainers.size > 0 + ? await db + .select({ id: workQueue.id, payload: workQueue.payload }) + .from(workQueue) + .where( + and( + eq(workQueue.status, "pending"), + inArray(workQueue.type, ["deploy", "reconcile"]), + inArray(workQueue.serverId, [...serverContainers.keys()]), + ), + ) + : []; const rolloutDeploymentIds = new Set(rolloutDeployments.map((d) => d.id)); const workToDelete = pendingWork.filter((w) => { diff --git a/web/components/service/details/deployment-progress.tsx b/web/components/service/details/deployment-progress.tsx index 3fdbe4b0..48bbb925 100644 --- a/web/components/service/details/deployment-progress.tsx +++ b/web/components/service/details/deployment-progress.tsx @@ -21,6 +21,7 @@ type StageInfo = { const STAGES: StageInfo[] = [ { id: "migrating", label: "Migrating" }, + { id: "queued", label: "Queued" }, { id: "deploying", label: "Starting" }, { id: "health_check", label: "Checking Health" }, { id: "dns_sync", label: "Routing traffic" }, @@ -103,10 +104,16 @@ export function getBarState( const latestRollout = service.rollouts?.[0]; const activeRollout = - latestRollout?.status === "in_progress" ? latestRollout : undefined; + latestRollout?.status === "queued" || + latestRollout?.status === "in_progress" + ? latestRollout + : undefined; if (activeRollout) { - const currentStage = activeRollout.currentStage || "deploying"; + const currentStage = + activeRollout.status === "queued" + ? "queued" + : activeRollout.currentStage || "deploying"; return { mode: "deploying", stage: currentStage, diff --git a/web/components/service/details/rollout-details.tsx b/web/components/service/details/rollout-details.tsx index ef7eb90d..0d597ef7 100644 --- a/web/components/service/details/rollout-details.tsx +++ b/web/components/service/details/rollout-details.tsx @@ -3,21 +3,16 @@ import { ArrowLeft, CheckCircle2, + Clock, Loader2, RotateCcw, XCircle, } from "lucide-react"; import { useRouter } from "next/navigation"; -import { - Item, - ItemContent, - ItemDescription, - ItemTitle, -} from "@/components/ui/item"; +import { LogViewer } from "@/components/logs/log-viewer"; import { Button } from "@/components/ui/button"; import type { Rollout, RolloutStatus, Service } from "@/db/types"; import { formatRelativeTime } from "@/lib/date"; -import { LogViewer } from "@/components/logs/log-viewer"; type RolloutWithDates = Omit & { createdAt: string | Date; @@ -33,6 +28,12 @@ const STATUS_CONFIG: Record< label: string; } > = { + queued: { + icon: Clock, + color: "text-slate-500", + bgColor: "bg-slate-500/10", + label: "Queued", + }, in_progress: { icon: Loader2, color: "text-blue-500", @@ -60,6 +61,7 @@ const STATUS_CONFIG: Record< }; const STAGE_LABELS: Record = { + queued: "Queued", preparing: "Preparing", certificates: "Issuing Certificates", deploying: "Deploying", @@ -102,7 +104,9 @@ export function RolloutDetails({ const router = useRouter(); const config = STATUS_CONFIG[rollout.status as RolloutStatus]; const Icon = config.icon; - const isLive = rollout.status === "in_progress"; + const isLive = + rollout.status === "queued" || rollout.status === "in_progress"; + const isAnimated = rollout.status === "in_progress"; return (
@@ -123,7 +127,7 @@ export function RolloutDetails({ - + {config.label} {rollout.currentStage && isLive && ( diff --git a/web/components/service/details/rollout-history.tsx b/web/components/service/details/rollout-history.tsx index eafaeaad..67c305a9 100644 --- a/web/components/service/details/rollout-history.tsx +++ b/web/components/service/details/rollout-history.tsx @@ -1,8 +1,8 @@ "use client"; -import { useState } from "react"; import { CheckCircle2, Clock, Loader2, RotateCcw, XCircle } from "lucide-react"; import Link from "next/link"; +import { useState } from "react"; import useSWR from "swr"; import { Empty, @@ -39,6 +39,11 @@ const STATUS_CONFIG: Record< label: string; } > = { + queued: { + icon: Clock, + color: "text-slate-500", + label: "Queued", + }, in_progress: { icon: Loader2, color: "text-blue-500", @@ -77,6 +82,7 @@ function StatusBadge({ status }: { status: RolloutStatus }) { } const STAGE_LABELS: Record = { + queued: "Queued", preparing: "Preparing", certificates: "Issuing Certificates", deploying: "Deploying", @@ -123,7 +129,9 @@ export function RolloutHistory({ revalidateOnFocus: true, onSuccess: (data) => { setHasInProgress( - data?.rollouts?.some((r) => r.status === "in_progress") ?? false, + data?.rollouts?.some( + (r) => r.status === "queued" || r.status === "in_progress", + ) ?? false, ); }, }, @@ -168,9 +176,11 @@ export function RolloutHistory({ - {rollout.status === "in_progress" - ? `Deploying — ${formatStage(rollout.currentStage)}` - : STATUS_CONFIG[rollout.status].label} + {rollout.status === "queued" + ? "Queued" + : rollout.status === "in_progress" + ? `Deploying — ${formatStage(rollout.currentStage)}` + : STATUS_CONFIG[rollout.status].label} diff --git a/web/components/service/service-canvas.tsx b/web/components/service/service-canvas.tsx index bb8fffdf..5d1803a4 100644 --- a/web/components/service/service-canvas.tsx +++ b/web/components/service/service-canvas.tsx @@ -64,6 +64,8 @@ const CANVAS_WIDTH = 1320; const CANVAS_HEIGHT = 900; const MIN_CANVAS_SCALE = 0.5; const SNAP_GRID_SIZE = 24; +const CANVAS_DOT_PATTERN = + "radial-gradient(circle, color-mix(in oklab, var(--muted-foreground) 36%, transparent) 1px, transparent 1px)"; function getCanvasScale() { if (typeof window === "undefined") { @@ -706,7 +708,7 @@ export function ServiceCanvas({ " style={{ height: "calc(100vh - 3.5rem)", - backgroundImage: `radial-gradient(circle, rgb(161 161 170 / 0.2) 1px, transparent 1px)`, + backgroundImage: CANVAS_DOT_PATTERN, backgroundSize: "24px 24px", }} > @@ -764,7 +766,7 @@ export function ServiceCanvas({ " style={{ height: "calc(100vh - 5rem)", - backgroundImage: `radial-gradient(circle, rgb(161 161 170 / 0.3) 1px, transparent 1px)`, + backgroundImage: CANVAS_DOT_PATTERN, backgroundSize: "20px 20px", }} > @@ -842,7 +844,7 @@ export function ServiceCanvas({ " style={{ height: "calc(100vh - 3.5rem)", - backgroundImage: `radial-gradient(circle, rgb(161 161 170 / 0.2) 1px, transparent 1px)`, + backgroundImage: CANVAS_DOT_PATTERN, backgroundSize: "24px 24px", }} > @@ -855,34 +857,36 @@ export function ServiceCanvas({
-
+
- {services.map((service, index) => ( - - ))} +
+ {services.map((service, index) => ( + + ))} +
diff --git a/web/components/service/service-layout-client.tsx b/web/components/service/service-layout-client.tsx index 90e608cb..2e38b894 100644 --- a/web/components/service/service-layout-client.tsx +++ b/web/components/service/service-layout-client.tsx @@ -11,8 +11,8 @@ import { } from "react"; import useSWR from "swr"; import type { ServiceWithDetails as Service } from "@/db/types"; -import type { ConfigChange } from "@/lib/service-config"; import { fetcher } from "@/lib/fetcher"; +import type { ConfigChange } from "@/lib/service-config"; import { buildCurrentConfig, diffConfigs, @@ -68,6 +68,7 @@ export function ServiceLayoutClient({ const isActive = (svc.latestBuild != null && ACTIVE_BUILD_STATUSES.includes(svc.latestBuild.status)) || + svc.rollouts?.[0]?.status === "queued" || svc.rollouts?.[0]?.status === "in_progress" || !!svc.migrationStatus || svc.deployments.some((d) => diff --git a/web/db/schema.ts b/web/db/schema.ts index bd4126ee..da3751cd 100644 --- a/web/db/schema.ts +++ b/web/db/schema.ts @@ -492,10 +492,10 @@ export const rollouts = pgTable( .notNull() .references(() => services.id, { onDelete: "cascade" }), status: text("status", { - enum: ["in_progress", "completed", "failed", "rolled_back"], + enum: ["queued", "in_progress", "completed", "failed", "rolled_back"], }) .notNull() - .default("in_progress"), + .default("queued"), currentStage: text("current_stage"), createdAt: timestamp("created_at", { withTimezone: true }) .defaultNow() diff --git a/web/lib/deploy-service.ts b/web/lib/deploy-service.ts index 4e846dd9..4f78e9d2 100644 --- a/web/lib/deploy-service.ts +++ b/web/lib/deploy-service.ts @@ -53,16 +53,28 @@ export async function deployServiceInternal(serviceId: string) { await db.insert(rollouts).values({ id: rolloutId, serviceId, - status: "in_progress", + status: "queued", currentStage: "queued", }); - await inngest.send( - inngestEvents.rolloutCreated.create({ - rolloutId, - serviceId, - }), - ); + try { + await inngest.send( + inngestEvents.rolloutCreated.create({ + rolloutId, + serviceId, + }), + ); + } catch (error) { + await db + .update(rollouts) + .set({ + status: "failed", + currentStage: "enqueue_failed", + completedAt: new Date(), + }) + .where(eq(rollouts.id, rolloutId)); + throw error; + } return { rolloutId }; } diff --git a/web/lib/inngest/functions/rollout-workflow.ts b/web/lib/inngest/functions/rollout-workflow.ts index 39c24e82..b57cc455 100644 --- a/web/lib/inngest/functions/rollout-workflow.ts +++ b/web/lib/inngest/functions/rollout-workflow.ts @@ -1,4 +1,4 @@ -import { and, eq, inArray, isNull, ne, or } from "drizzle-orm"; +import { and, eq, inArray, isNull, lt, ne, or, sql } from "drizzle-orm"; import { db } from "@/db"; import { getService } from "@/db/queries"; import { deployments, rollouts, servers } from "@/db/schema"; @@ -29,6 +29,11 @@ const PREFLIGHT_FAILURE_MESSAGES = [ const PREFLIGHT_FAILURE_PREFIXES = ["Server "]; +const ROLLOUT_TURN_WAIT_ATTEMPTS = 360; +const ROLLOUT_TURN_WAIT_INTERVAL = "10s"; + +type RolloutTurnState = "acquired" | "waiting" | "terminal"; + function getPreflightFailureReason(error: unknown) { if (!(error instanceof Error)) return null; @@ -47,6 +52,63 @@ function getPreflightFailureReason(error: unknown) { return null; } +async function acquireRolloutTurn( + rolloutId: string, + serviceId: string, +): Promise { + return db.transaction(async (tx) => { + await tx.execute(sql`SELECT pg_advisory_xact_lock(hashtext(${serviceId}))`); + + const rollout = await tx + .select({ status: rollouts.status, createdAt: rollouts.createdAt }) + .from(rollouts) + .where(eq(rollouts.id, rolloutId)) + .then((rows) => rows[0]); + + if (!rollout) { + throw new Error("Rollout not found"); + } + + if (rollout.status !== "queued") { + return rollout.status === "in_progress" ? "acquired" : "terminal"; + } + + const blockingRollout = await tx + .select({ id: rollouts.id }) + .from(rollouts) + .where( + or( + and( + eq(rollouts.serviceId, serviceId), + eq(rollouts.status, "in_progress"), + ne(rollouts.id, rolloutId), + ), + and( + eq(rollouts.serviceId, serviceId), + eq(rollouts.status, "queued"), + lt(rollouts.createdAt, rollout.createdAt), + ), + ), + ) + .limit(1) + .then((rows) => rows[0]); + + if (blockingRollout) { + return "waiting"; + } + + await tx + .update(rollouts) + .set({ + status: "in_progress", + currentStage: "preparing", + }) + .where(eq(rollouts.id, rolloutId)); + + return "acquired"; + }); +} + export const rolloutWorkflow = inngest.createFunction( { id: "rollout-workflow", @@ -55,6 +117,27 @@ export const rolloutWorkflow = inngest.createFunction( cancelOn: [ { event: inngestEvents.rolloutCancelled, match: "data.rolloutId" }, ], + onFailure: async ({ event }) => { + const { rolloutId } = event.data.event.data as { + rolloutId?: string; + }; + + if (!rolloutId) return; + + await db + .update(rollouts) + .set({ + status: "failed", + currentStage: "workflow_failed", + completedAt: new Date(), + }) + .where( + and( + eq(rollouts.id, rolloutId), + inArray(rollouts.status, ["queued", "in_progress"]), + ), + ); + }, }, async ({ event, step }) => { const { rolloutId, serviceId } = event.data; @@ -66,11 +149,51 @@ export const rolloutWorkflow = inngest.createFunction( } }); - await step.run("mark-rollout-in-progress", async () => { - await db - .update(rollouts) - .set({ status: "in_progress", currentStage: "preparing" }) - .where(eq(rollouts.id, rolloutId)); + let acquiredTurn = false; + for (let attempt = 0; attempt < ROLLOUT_TURN_WAIT_ATTEMPTS; attempt++) { + const turnState = await step.run( + `acquire-rollout-turn-${attempt}`, + async () => { + return acquireRolloutTurn(rolloutId, serviceId); + }, + ); + + if (turnState === "terminal") { + return { status: "cancelled", rolloutId }; + } + + if (turnState === "acquired") { + acquiredTurn = true; + break; + } + + await step.sleep( + `wait-for-active-rollout-${attempt}`, + ROLLOUT_TURN_WAIT_INTERVAL, + ); + } + + if (!acquiredTurn) { + await step.run("mark-rollout-queue-timeout", async () => { + await db + .update(rollouts) + .set({ + status: "failed", + currentStage: "queue_timeout", + completedAt: new Date(), + }) + .where(eq(rollouts.id, rolloutId)); + await ingestRolloutLog( + rolloutId, + serviceId, + "queue_timeout", + "Timed out waiting for previous rollout to finish", + ); + }); + return { status: "failed", rolloutId, reason: "queue_timeout" }; + } + + await step.run("log-rollout-started", async () => { await ingestRolloutLog( rolloutId, serviceId, diff --git a/web/lib/scheduler.ts b/web/lib/scheduler.ts index d157c19e..e2edab16 100644 --- a/web/lib/scheduler.ts +++ b/web/lib/scheduler.ts @@ -192,7 +192,7 @@ export async function checkAndRunScheduledDeployments(): Promise { .where( and( eq(rollouts.serviceId, service.id), - eq(rollouts.status, "in_progress"), + inArray(rollouts.status, ["queued", "in_progress"]), ), ) .limit(1); @@ -233,37 +233,14 @@ export async function checkAndRunScheduledDeployments(): Promise { console.log("[scheduler] finished checking scheduled deployments"); } -const STALE_ITEM_THRESHOLD_MS = 15 * 60 * 1000; const OLD_ITEM_THRESHOLD_MS = 90 * 24 * 60 * 60 * 1000; export async function cleanupStaleItems(): Promise { - const staleThreshold = new Date(Date.now() - STALE_ITEM_THRESHOLD_MS); const workItemLeaseThreshold = new Date( Date.now() - WORK_QUEUE_LEASE_DURATION_MS, ); const oldThreshold = new Date(Date.now() - OLD_ITEM_THRESHOLD_MS); - const staleRollouts = await db - .update(rollouts) - .set({ - status: "failed", - currentStage: "timeout", - completedAt: new Date(), - }) - .where( - and( - eq(rollouts.status, "in_progress"), - lt(rollouts.createdAt, staleThreshold), - ), - ) - .returning({ id: rollouts.id }); - - if (staleRollouts.length > 0) { - console.log( - `[scheduler] cleaned up ${staleRollouts.length} stale rollouts`, - ); - } - // Pending work is intentionally retained so commands can run when an agent // reconnects. Only exhausted processing attempts are failed here. const staleWorkItems = await db diff --git a/web/lib/work-queue.ts b/web/lib/work-queue.ts index c6bb674a..90f2cd45 100644 --- a/web/lib/work-queue.ts +++ b/web/lib/work-queue.ts @@ -1,10 +1,11 @@ import { randomUUID } from "node:crypto"; import { and, eq, sql } from "drizzle-orm"; import { db } from "@/db"; -import { deployments, workQueue } from "@/db/schema"; +import { deployments, rollouts, servers, workQueue } from "@/db/schema"; import type { WorkQueue } from "@/db/types"; import { inngest } from "@/lib/inngest/client"; import { inngestEvents } from "@/lib/inngest/events"; +import { ingestRolloutLog } from "@/lib/victoria-logs"; export const WORK_QUEUE_MAX_ATTEMPTS = 3; export const WORK_QUEUE_LEASE_DURATION_MS = 2 * 60 * 1000; @@ -200,6 +201,11 @@ async function runWorkItemCompletionSideEffects( item: WorkQueue, result: WorkItemResult, ): Promise { + if ((item.type === "deploy" || item.type === "reconcile") && item.payload) { + await runDeploymentWorkCompletionSideEffects(item, result); + return; + } + if (item.type === "force_cleanup" && item.payload) { await runForceCleanupCompletionSideEffects(item, result); return; @@ -240,6 +246,79 @@ async function runWorkItemCompletionSideEffects( } } +async function runDeploymentWorkCompletionSideEffects( + item: WorkQueue, + result: WorkItemResult, +): Promise { + if (result.status !== "failed") return; + + try { + const payload = JSON.parse(item.payload) as { + deploymentId?: string; + }; + + if (!payload.deploymentId) return; + + const deployment = await db + .select({ + id: deployments.id, + serviceId: deployments.serviceId, + rolloutId: deployments.rolloutId, + rolloutStatus: rollouts.status, + serverName: servers.name, + }) + .from(deployments) + .innerJoin(servers, eq(deployments.serverId, servers.id)) + .innerJoin(rollouts, eq(deployments.rolloutId, rollouts.id)) + .where(eq(deployments.id, payload.deploymentId)) + .then((rows) => rows[0]); + + if ( + !deployment || + !deployment.rolloutId || + deployment.rolloutStatus !== "in_progress" + ) { + return; + } + + await db + .update(deployments) + .set({ + status: "failed", + failedStage: "deploying", + }) + .where(eq(deployments.id, deployment.id)); + + await ingestRolloutLog( + deployment.rolloutId, + deployment.serviceId, + "deploying", + `Deployment failed on server ${deployment.serverName}: ${formatWorkItemError(result.error)}`, + ); + + await inngest.send( + inngestEvents.resourceStatusChanged.create({ + type: "deployment", + id: deployment.id, + parentType: "rollout", + parentId: deployment.rolloutId, + }), + ); + } catch (error) { + console.error( + "[work-queue] failed to run deployment completion side effects:", + error, + ); + } +} + +function formatWorkItemError(error: string | undefined): string { + const fallback = "Agent failed to apply the deployment"; + if (!error?.trim()) return fallback; + + return error.trim().replace(/\s+/g, " ").slice(0, 1000); +} + async function runForceCleanupCompletionSideEffects( item: WorkQueue, result: WorkItemResult,