diff --git a/agent/internal/container/runtime.go b/agent/internal/container/runtime.go index 8cb55e6e..a0993d9a 100644 --- a/agent/internal/container/runtime.go +++ b/agent/internal/container/runtime.go @@ -71,8 +71,6 @@ func Deploy(config *DeployConfig) (*DeployResult, error) { image := config.Image - exec.Command("podman", "rm", "-f", config.Name).Run() - logFunc("stdout", fmt.Sprintf("Pulling image: %s", image)) pullCmd := exec.Command("podman", "pull", "--tls-verify=false", image) @@ -92,6 +90,7 @@ func Deploy(config *DeployConfig) (*DeployResult, error) { } args := buildPodmanRunArgs(config, image) + exec.Command("podman", "rm", "-f", config.Name).Run() logFunc("stdout", fmt.Sprintf("Starting container: %s", config.Name)) diff --git a/web/actions/projects.ts b/web/actions/projects.ts index a3b9ed25..56c886a9 100644 --- a/web/actions/projects.ts +++ b/web/actions/projects.ts @@ -1161,9 +1161,6 @@ export async function updateServiceConfig( if (config.ports) { if (config.ports.remove && config.ports.remove.length > 0) { for (const portId of config.ports.remove) { - await db - .delete(deploymentPorts) - .where(eq(deploymentPorts.servicePortId, portId)); await db.delete(servicePorts).where(eq(servicePorts.id, portId)); } } diff --git a/web/app/api/projects/[id]/services/route.ts b/web/app/api/projects/[id]/services/route.ts index 7f69ca9d..35628784 100644 --- a/web/app/api/projects/[id]/services/route.ts +++ b/web/app/api/projects/[id]/services/route.ts @@ -1,6 +1,6 @@ export const dynamic = "force-dynamic"; -import { and, desc, eq, isNull } from "drizzle-orm"; +import { and, desc, eq, inArray, isNull } from "drizzle-orm"; import { headers } from "next/headers"; import { db } from "@/db"; import { @@ -12,12 +12,14 @@ import { servers, servicePorts, serviceReplicas, + serviceRevisions, services, serviceVolumes, volumeBackups, } from "@/db/schema"; import { auth } from "@/lib/auth"; import { getTimestamp } from "@/lib/date"; +import { revisionSpecToDeployedConfig } from "@/lib/service-config"; export async function GET( request: Request, @@ -114,6 +116,40 @@ export async function GET( : Promise.resolve(null), ]); + const activeDeployment = serviceDeployments.find( + (deployment) => + deployment.trafficState === "active" && + deployment.runtimeDesiredState !== "removed", + ); + const activeRevision = activeDeployment + ? await db + .select({ specification: serviceRevisions.specification }) + .from(serviceRevisions) + .where(eq(serviceRevisions.id, activeDeployment.serviceRevisionId)) + .then((rows) => rows[0]) + : null; + const revisionServers = activeRevision + ? await db + .select({ id: servers.id, name: servers.name }) + .from(servers) + .where( + inArray( + servers.id, + activeRevision.specification.placements.map( + (placement) => placement.serverId, + ), + ), + ) + : []; + const activeConfig = activeRevision + ? revisionSpecToDeployedConfig( + activeRevision.specification, + Object.fromEntries( + revisionServers.map((server) => [server.id, server.name]), + ), + ) + : null; + const deploymentsWithDetails = await Promise.all( serviceDeployments.map(async (deployment) => { const [depPorts, server] = await Promise.all([ @@ -121,13 +157,9 @@ export async function GET( .select({ id: deploymentPorts.id, hostPort: deploymentPorts.hostPort, - containerPort: servicePorts.port, + containerPort: deploymentPorts.containerPort, }) .from(deploymentPorts) - .innerJoin( - servicePorts, - eq(deploymentPorts.servicePortId, servicePorts.id), - ) .where(eq(deploymentPorts.deploymentId, deployment.id)), db .select({ name: servers.name, wireguardIp: servers.wireguardIp }) @@ -205,6 +237,7 @@ export async function GET( volumes, lockedServer, latestBuild, + activeConfig, deletionBackupFallback, }; }), diff --git a/web/components/service/service-layout-client.tsx b/web/components/service/service-layout-client.tsx index 7d4d0fe4..15db8498 100644 --- a/web/components/service/service-layout-client.tsx +++ b/web/components/service/service-layout-client.tsx @@ -7,11 +7,7 @@ import useSWR from "swr"; import type { ServiceWithDetails as Service } from "@/db/types"; import { fetcher } from "@/lib/fetcher"; import type { ConfigChange } from "@/lib/service-config"; -import { - buildCurrentConfig, - diffConfigs, - parseDeployedConfig, -} from "@/lib/service-config"; +import { buildCurrentConfig, diffConfigs } from "@/lib/service-config"; import { cn } from "@/lib/utils"; const ACTIVE_BUILD_STATUSES = [ @@ -77,7 +73,7 @@ export function ServiceLayoutClient({ const pendingChanges = useMemo(() => { if (!service) return []; - const deployed = parseDeployedConfig(service.deployedConfig); + const deployed = service.activeConfig ?? null; const replicas = (service.configuredReplicas || []).map((r) => ({ serverId: r.serverId, serverName: r.serverName, diff --git a/web/db/schema.ts b/web/db/schema.ts index a3bc22ca..5727e283 100644 --- a/web/db/schema.ts +++ b/web/db/schema.ts @@ -2,6 +2,7 @@ import { relations, sql } from "drizzle-orm"; import { bigint, boolean, + foreignKey, index, integer, jsonb, @@ -9,8 +10,10 @@ import { real, text, timestamp, + unique, uniqueIndex, } from "drizzle-orm/pg-core"; +import type { ServiceRevisionSpec } from "@/lib/service-revision-spec"; export const user = pgTable("user", { id: text("id").primaryKey(), @@ -417,7 +420,6 @@ export const services = pgTable("services", { serverlessWakeTimeoutSeconds: integer("serverless_wake_timeout_seconds") .notNull() .default(300), - deployedConfig: text("deployed_config"), deploymentSchedule: text("deployment_schedule"), lastScheduledDeploymentRunAt: timestamp("last_scheduled_deployment_run_at", { withTimezone: true, @@ -547,6 +549,29 @@ export const secrets = pgTable("secrets", { .notNull(), }); +export const serviceRevisions = pgTable( + "service_revisions", + { + id: text("id").primaryKey(), + serviceId: text("service_id") + .notNull() + .references(() => services.id, { onDelete: "cascade" }), + specification: jsonb("specification") + .$type() + .notNull(), + createdAt: timestamp("created_at", { withTimezone: true }) + .defaultNow() + .notNull(), + }, + (table) => [ + unique("service_revisions_id_service_id_unique").on( + table.id, + table.serviceId, + ), + index("service_revisions_service_id_idx").on(table.serviceId), + ], +); + export const deployments = pgTable( "deployments", { @@ -554,6 +579,7 @@ export const deployments = pgTable( serviceId: text("service_id") .notNull() .references(() => services.id, { onDelete: "cascade" }), + serviceRevisionId: text("service_revision_id").notNull(), serverId: text("server_id") .notNull() .references(() => servers.id, { onDelete: "cascade" }), @@ -611,12 +637,18 @@ export const deployments = pgTable( index("deployments_container_id_idx").on(table.containerId), index("deployments_rollout_id_idx").on(table.rolloutId), index("deployments_service_id_idx").on(table.serviceId), + index("deployments_service_revision_id_idx").on(table.serviceRevisionId), index("deployments_server_id_idx").on(table.serverId), index("deployments_runtime_desired_state_idx").on( table.runtimeDesiredState, ), index("deployments_traffic_state_idx").on(table.trafficState), index("deployments_observed_phase_idx").on(table.observedPhase), + foreignKey({ + name: "deployments_service_revision_service_fk", + columns: [table.serviceRevisionId, table.serviceId], + foreignColumns: [serviceRevisions.id, serviceRevisions.serviceId], + }).onDelete("no action"), ], ); @@ -627,6 +659,7 @@ export const rollouts = pgTable( serviceId: text("service_id") .notNull() .references(() => services.id, { onDelete: "cascade" }), + serviceRevisionId: text("service_revision_id"), status: text("status", { enum: ["queued", "in_progress", "completed", "failed", "rolled_back"], }) @@ -638,7 +671,15 @@ export const rollouts = pgTable( .notNull(), completedAt: timestamp("completed_at", { withTimezone: true }), }, - (table) => [index("rollouts_service_id_idx").on(table.serviceId)], + (table) => [ + index("rollouts_service_id_idx").on(table.serviceId), + index("rollouts_service_revision_id_idx").on(table.serviceRevisionId), + foreignKey({ + name: "rollouts_service_revision_service_fk", + columns: [table.serviceRevisionId, table.serviceId], + foreignColumns: [serviceRevisions.id, serviceRevisions.serviceId], + }).onDelete("no action"), + ], ); export const deploymentPorts = pgTable("deployment_ports", { @@ -646,9 +687,7 @@ export const deploymentPorts = pgTable("deployment_ports", { deploymentId: text("deployment_id") .notNull() .references(() => deployments.id, { onDelete: "cascade" }), - servicePortId: text("service_port_id") - .notNull() - .references(() => servicePorts.id, { onDelete: "cascade" }), + containerPort: integer("container_port").notNull(), hostPort: integer("host_port").notNull(), createdAt: timestamp("created_at", { withTimezone: true }) .defaultNow() diff --git a/web/db/types.ts b/web/db/types.ts index 32c4d446..6a8d6f44 100644 --- a/web/db/types.ts +++ b/web/db/types.ts @@ -1,3 +1,4 @@ +import type { DeployedConfig } from "@/lib/service-config"; import type { builds, deploymentPorts, @@ -54,6 +55,7 @@ export type HealthStats = { }; export type ServiceWithDetails = Service & { + activeConfig?: DeployedConfig | null; ports: ServicePort[]; configuredReplicas: Array< ServiceReplica & { serverName: string; serverIsProxy: boolean } diff --git a/web/lib/agent-status.ts b/web/lib/agent-status.ts index dc4a362b..51243b29 100644 --- a/web/lib/agent-status.ts +++ b/web/lib/agent-status.ts @@ -3,11 +3,11 @@ import { db } from "@/db"; import { type AgentHealth, type ContainerHealth, - deploymentPorts, deployments, type NetworkHealth, rollouts, servers, + serviceRevisions, services, workQueue, } from "@/db/schema"; @@ -29,7 +29,7 @@ import { import { inngest } from "@/lib/inngest/client"; import { inngestEvents } from "@/lib/inngest/events"; import { getServerlessWakeFailureUpdate } from "@/lib/serverless-wake-failures"; -import { getDeployedServerlessConfig } from "@/lib/service-config"; +import type { ServiceRevisionSpec } from "@/lib/service-revision-spec"; import { ingestRolloutLog } from "@/lib/victoria-logs"; import { enqueueWork } from "@/lib/work-queue"; @@ -153,17 +153,16 @@ async function applyDeploymentErrors( rolloutId: deployments.rolloutId, observedPhase: deployments.observedPhase, serverlessWakeFailureCount: deployments.serverlessWakeFailureCount, - serverlessEnabled: services.serverlessEnabled, - serverlessSleepAfterSeconds: services.serverlessSleepAfterSeconds, - serverlessWakeTimeoutSeconds: services.serverlessWakeTimeoutSeconds, - stateful: services.stateful, - deployedConfig: services.deployedConfig, + revisionSpec: serviceRevisions.specification, rolloutStatus: rollouts.status, serverName: servers.name, }) .from(deployments) .innerJoin(servers, eq(deployments.serverId, servers.id)) - .innerJoin(services, eq(deployments.serviceId, services.id)) + .innerJoin( + serviceRevisions, + eq(deployments.serviceRevisionId, serviceRevisions.id), + ) .leftJoin(rollouts, eq(deployments.rolloutId, rollouts.id)) .where( and( @@ -192,8 +191,7 @@ async function applyDeploymentErrors( .set( isServerlessWakeDeployment ? getServerlessWakeFailureUpdate({ - serverlessEnabled: - getDeployedServerlessConfig(deployment).enabled, + serverlessEnabled: deployment.revisionSpec.serverless.enabled, currentFailureCount: deployment.serverlessWakeFailureCount, failedStage: "serverless_wake", }) @@ -284,16 +282,15 @@ async function applyServerlessTransitions( trafficState: deployments.trafficState, observedPhase: deployments.observedPhase, serverlessWakeFailureCount: deployments.serverlessWakeFailureCount, - serverlessEnabled: services.serverlessEnabled, - serverlessSleepAfterSeconds: services.serverlessSleepAfterSeconds, - serverlessWakeTimeoutSeconds: services.serverlessWakeTimeoutSeconds, - stateful: services.stateful, - deployedConfig: services.deployedConfig, + revisionSpec: serviceRevisions.specification, serverIsProxy: servers.isProxy, serverName: servers.name, }) .from(deployments) - .innerJoin(services, eq(deployments.serviceId, services.id)) + .innerJoin( + serviceRevisions, + eq(deployments.serviceRevisionId, serviceRevisions.id), + ) .innerJoin(servers, eq(deployments.serverId, servers.id)) .where(eq(deployments.id, transition.deploymentId)) .then((rows) => rows[0]); @@ -418,7 +415,7 @@ async function applyServerlessTransitions( .update(deployments) .set( getServerlessWakeFailureUpdate({ - serverlessEnabled: getDeployedServerlessConfig(deployment).enabled, + serverlessEnabled: deployment.revisionSpec.serverless.enabled, currentFailureCount: deployment.serverlessWakeFailureCount, failedStage: "serverless_wake", }), @@ -565,10 +562,7 @@ function getInvalidServerlessTransitionReason({ runtimeDesiredState: string; trafficState: string; observedPhase: string; - serverlessEnabled: boolean; - serverlessSleepAfterSeconds: number; - serverlessWakeTimeoutSeconds: number; - deployedConfig: string | null; + revisionSpec: ServiceRevisionSpec; serverIsProxy: boolean; } | undefined; @@ -577,7 +571,7 @@ function getInvalidServerlessTransitionReason({ if (deployment.serverId !== serverId) return "deployment belongs to another server"; if (!deployment.serverIsProxy) return "server is not a proxy"; - if (!getDeployedServerlessConfig(deployment).enabled) { + if (!deployment.revisionSpec.serverless.enabled) { return "service is not serverless"; } if (deployment.runtimeDesiredState === "removed") { @@ -764,9 +758,6 @@ export async function applyStatusReport( console.log( `[status:${serverId.slice(0, 8)}] deployment ${dep.id.slice(0, 8)} was removed and container gone, deleting`, ); - await db - .delete(deploymentPorts) - .where(eq(deploymentPorts.deploymentId, dep.id)); await db.delete(deployments).where(eq(deployments.id, dep.id)); } else if ( isObservedActiveContainer(dep.observedPhase) && @@ -817,13 +808,20 @@ export async function applyStatusReport( `[health:recover] found stuck deployment ${stuckDeployment.id}, attaching container ${container.containerId}`, ); - const service = await db - .select() - .from(services) - .where(eq(services.id, stuckDeployment.serviceId)) - .then((r) => r[0]); - - const hasHealthCheck = service?.healthCheckCmd != null; + const [service, revision] = await Promise.all([ + db + .select() + .from(services) + .where(eq(services.id, stuckDeployment.serviceId)) + .then((r) => r[0]), + db + .select({ specification: serviceRevisions.specification }) + .from(serviceRevisions) + .where(eq(serviceRevisions.id, stuckDeployment.serviceRevisionId)) + .then((r) => r[0]), + ]); + + const hasHealthCheck = revision?.specification.healthCheck != null; const newStatus = hasHealthCheck ? "starting" : "healthy"; await db @@ -913,17 +911,17 @@ export async function applyStatusReport( deployment.runtimeDesiredState === "running" && ["sleeping", "stopped"].includes(deployment.observedPhase) ) { - const service = await db - .select() - .from(services) - .where(eq(services.id, deployment.serviceId)) + const revision = await db + .select({ specification: serviceRevisions.specification }) + .from(serviceRevisions) + .where(eq(serviceRevisions.id, deployment.serviceRevisionId)) .then((r) => r[0]); - if (service && getDeployedServerlessConfig(service).enabled) { + if (revision?.specification.serverless.enabled) { Object.assign( updateFields, getStaleStoppedServerlessReportUpdate({ - hasHealthCheck: service.healthCheckCmd != null, + hasHealthCheck: revision.specification.healthCheck != null, healthStatus, }), ); @@ -938,13 +936,20 @@ export async function applyStatusReport( continue; } - const service = await db - .select() - .from(services) - .where(eq(services.id, deployment.serviceId)) - .then((r) => r[0]); + const [revision, service] = await Promise.all([ + db + .select({ specification: serviceRevisions.specification }) + .from(serviceRevisions) + .where(eq(serviceRevisions.id, deployment.serviceRevisionId)) + .then((r) => r[0]), + db + .select({ migrationStatus: services.migrationStatus }) + .from(services) + .where(eq(services.id, deployment.serviceId)) + .then((r) => r[0]), + ]); - const hasHealthCheck = service?.healthCheckCmd != null; + const hasHealthCheck = revision?.specification.healthCheck != null; const newStatus = hasHealthCheck ? "starting" : "healthy"; updateFields.observedPhase = newStatus; if (deployment.observedPhase === "waking") { diff --git a/web/lib/agent/expected-state.ts b/web/lib/agent/expected-state.ts index cb13d987..db5c4b36 100644 --- a/web/lib/agent/expected-state.ts +++ b/web/lib/agent/expected-state.ts @@ -3,11 +3,9 @@ import { db } from "@/db"; import { deploymentPorts, deployments, - secrets, servers, - servicePorts, + serviceRevisions, services, - serviceVolumes, } from "@/db/schema"; import { getAllCertificatesForDomains } from "@/lib/acme-manager"; import { @@ -17,30 +15,35 @@ import { runtimeExpectedStates, } from "@/lib/deployment-status"; import { - getDeployedServerlessConfig, - getDeployedServicePorts, - isDeployedServerlessService, -} from "@/lib/service-config"; -import { slugify } from "@/lib/utils"; + SERVICE_REVISION_SCHEMA_VERSION, + type ServiceRevisionSecret, + type ServiceRevisionSpec, +} from "@/lib/service-revision-spec"; import { getWireGuardPeers } from "@/lib/wireguard"; type Server = typeof servers.$inferSelect; type Service = typeof services.$inferSelect; type Deployment = typeof deployments.$inferSelect; -type ServicePort = typeof servicePorts.$inferSelect; +type ServiceRevision = typeof serviceRevisions.$inferSelect; const SERVERLESS_GATEWAY_PORT = 18080; -type RouteServicePort = Pick< - ServicePort, - | "id" - | "serviceId" - | "port" - | "isPublic" - | "domain" - | "protocol" - | "externalPort" - | "tlsPassthrough" ->; +type RouteServicePort = { + id: string; + serviceId: string; + port: number; + isPublic: boolean; + domain: string | null; + protocol: "http" | "tcp" | "udp"; + externalPort: number | null; + tlsPassthrough: boolean; +}; + +export type RuntimeServiceRevision = { + id: string; + name: string; + revisionId: string; + specification: ServiceRevisionSpec; +}; export type ExpectedContainer = { deploymentId: string; @@ -127,18 +130,6 @@ type DeploymentPortRow = { containerPort: number; }; -type SecretRow = { - serviceId: string; - key: string; - encryptedValue: string; -}; - -type VolumeRow = { - serviceId: string; - name: string; - containerPath: string; -}; - type RoutableDeploymentRow = { serviceId: string; ipAddress: string | null; @@ -172,16 +163,19 @@ export async function getServer(serverId: string) { export async function buildAgentExpectedState( server: Server, ): Promise { - const allServices = await getActiveServices(); - const containers = await buildExpectedContainers(server.id); - const dnsRecords = await buildDnsRecords(allServices); - const traefikConfig = await buildTraefikConfig(server, allServices); + const runtimeServices = await getRuntimeServiceRevisions(); + const [containers, dnsRecords, traefikConfig, wireguardPeers] = + await Promise.all([ + buildExpectedContainers(server.id), + buildDnsRecords(runtimeServices), + buildTraefikConfig(server, runtimeServices), + getWireGuardPeers(server.id, server.privateIp), + ]); const serverless = await buildServerlessExpectedState( server, - allServices, + runtimeServices, containers, ); - const wireguardPeers = await getWireGuardPeers(server.id, server.privateIp); return { serverName: server.name, @@ -193,8 +187,45 @@ export async function buildAgentExpectedState( }; } -async function getActiveServices() { - return db.select().from(services).where(isNull(services.deletedAt)); +async function getRuntimeServiceRevisions(): Promise { + const rows = await db + .select({ + serviceId: services.id, + serviceName: services.name, + revisionId: serviceRevisions.id, + specification: serviceRevisions.specification, + }) + .from(deployments) + .innerJoin(services, eq(deployments.serviceId, services.id)) + .innerJoin( + serviceRevisions, + eq(deployments.serviceRevisionId, serviceRevisions.id), + ) + .where( + and( + isNull(services.deletedAt), + inArray(deployments.runtimeDesiredState, runtimeExpectedStates), + inArray(deployments.trafficState, activeTrafficStates), + ), + ); + + const runtimeServices = new Map(); + for (const row of rows) { + if (row.specification.schemaVersion !== SERVICE_REVISION_SCHEMA_VERSION) { + throw new Error(`Service ${row.serviceId} uses an unsupported revision`); + } + const existing = runtimeServices.get(row.serviceId); + if (existing && existing.revisionId !== row.revisionId) { + throw new Error(`Service ${row.serviceId} has multiple active revisions`); + } + runtimeServices.set(row.serviceId, { + id: row.serviceId, + name: row.serviceName, + revisionId: row.revisionId, + specification: row.specification, + }); + } + return [...runtimeServices.values()].sort((a, b) => a.id.localeCompare(b.id)); } async function buildExpectedContainers( @@ -211,42 +242,28 @@ async function buildExpectedContainers( ); const serviceIds = unique(serverDeployments.map((dep) => dep.serviceId)); + const revisionIds = unique( + serverDeployments.map((dep) => dep.serviceRevisionId), + ); if (serviceIds.length === 0) return []; - const [activeServices, depPorts, serviceSecrets, volumes] = await Promise.all( - [ - db - .select() - .from(services) - .where( - and(inArray(services.id, serviceIds), isNull(services.deletedAt)), - ), - fetchDeploymentPorts(serverDeployments.map((dep) => dep.id)), - db - .select({ - serviceId: secrets.serviceId, - key: secrets.key, - encryptedValue: secrets.encryptedValue, - }) - .from(secrets) - .where(inArray(secrets.serviceId, serviceIds)), - db - .select({ - serviceId: serviceVolumes.serviceId, - name: serviceVolumes.name, - containerPath: serviceVolumes.containerPath, - }) - .from(serviceVolumes) - .where(inArray(serviceVolumes.serviceId, serviceIds)), - ], - ); + const [activeServices, revisions, depPorts] = await Promise.all([ + db + .select() + .from(services) + .where(and(inArray(services.id, serviceIds), isNull(services.deletedAt))), + db + .select() + .from(serviceRevisions) + .where(inArray(serviceRevisions.id, revisionIds)), + fetchDeploymentPorts(serverDeployments.map((dep) => dep.id)), + ]); return buildExpectedContainersFromRows({ deployments: serverDeployments, services: activeServices, + revisions, deploymentPorts: depPorts, - secrets: serviceSecrets, - volumes, }); } @@ -257,25 +274,27 @@ async function fetchDeploymentPorts(deploymentIds: string[]) { .select({ deploymentId: deploymentPorts.deploymentId, hostPort: deploymentPorts.hostPort, - containerPort: servicePorts.port, + containerPort: deploymentPorts.containerPort, }) .from(deploymentPorts) - .innerJoin(servicePorts, eq(deploymentPorts.servicePortId, servicePorts.id)) - .where(inArray(deploymentPorts.deploymentId, deploymentIds)); + .where(inArray(deploymentPorts.deploymentId, deploymentIds)) + .orderBy( + deploymentPorts.deploymentId, + deploymentPorts.containerPort, + deploymentPorts.hostPort, + ); } export function buildExpectedContainersFromRows({ deployments: deploymentRows, services: serviceRows, + revisions: revisionRows, deploymentPorts: deploymentPortRows, - secrets: secretRows, - volumes: volumeRows, }: { deployments: Deployment[]; services: Service[]; + revisions: ServiceRevision[]; deploymentPorts: DeploymentPortRow[]; - secrets: SecretRow[]; - volumes: VolumeRow[]; }): ExpectedContainer[] { const servicesById = new Map( serviceRows.map((service) => [service.id, service]), @@ -284,16 +303,61 @@ export function buildExpectedContainersFromRows({ deploymentPortRows, (port) => port.deploymentId, ); - const secretsByServiceId = groupBy(secretRows, (secret) => secret.serviceId); - const volumesByServiceId = groupBy(volumeRows, (volume) => volume.serviceId); + const revisionsById = new Map( + revisionRows.map((revision) => [revision.id, revision]), + ); return deploymentRows .slice() .sort((a, b) => a.id.localeCompare(b.id)) .flatMap((dep) => { const service = servicesById.get(dep.serviceId); - if (!service) return []; - + const revision = revisionsById.get(dep.serviceRevisionId); + if (!service) { + console.error( + `[expected-state] deployment ${dep.id} omitted because its service is deleted`, + ); + return []; + } + if (!revision) { + throw new Error(`Deployment ${dep.id} has no service revision`); + } + if ( + revision.specification.schemaVersion !== SERVICE_REVISION_SCHEMA_VERSION + ) { + throw new Error( + `Deployment ${dep.id} uses an unsupported service revision`, + ); + } + const specification = revision.specification; + const ports = (portsByDeploymentId.get(dep.id) ?? []) + .slice() + .sort( + (a, b) => + a.containerPort - b.containerPort || a.hostPort - b.hostPort, + ) + .map((port) => ({ + containerPort: port.containerPort, + hostPort: port.hostPort, + })); + const expectedContainerPorts = specification.ports + .map((port) => port.containerPort) + .sort((a, b) => a - b); + const allocatedContainerPorts = ports.map((port) => port.containerPort); + if ( + JSON.stringify(expectedContainerPorts) !== + JSON.stringify(allocatedContainerPorts) + ) { + throw new Error(`Deployment ${dep.id} has incomplete port allocation`); + } + const env = buildEnv(specification.secrets); + const volumes = specification.volumes + .slice() + .sort((a, b) => a.containerPath.localeCompare(b.containerPath)) + .map((volume) => ({ + name: volume.name, + containerPath: volume.containerPath, + })); return [ { deploymentId: dep.id, @@ -302,28 +366,16 @@ export function buildExpectedContainersFromRows({ name: `${dep.serviceId}-${dep.id.slice(0, 8)}`, desiredState: dep.runtimeDesiredState === "stopped" ? "stopped" : "running", - image: normalizeImage(service.image), + image: normalizeImage(specification.image), ipAddress: dep.ipAddress, - ports: (portsByDeploymentId.get(dep.id) ?? []) - .slice() - .sort((a, b) => a.containerPort - b.containerPort) - .map((p) => ({ - containerPort: p.containerPort, - hostPort: p.hostPort, - })), - publishLocalPorts: isDeployedServerlessService(service), - env: buildEnv(secretsByServiceId.get(dep.serviceId) ?? []), - startCommand: service.startCommand || null, - healthCheck: buildHealthCheck(service), - volumes: (volumesByServiceId.get(dep.serviceId) ?? []) - .slice() - .sort((a, b) => a.containerPath.localeCompare(b.containerPath)) - .map((v) => ({ - name: v.name, - containerPath: v.containerPath, - })), - resourceCpuLimit: service.resourceCpuLimit, - resourceMemoryLimitMb: service.resourceMemoryLimitMb, + ports, + publishLocalPorts: specification.serverless.enabled, + env, + startCommand: specification.startCommand, + healthCheck: specification.healthCheck, + volumes, + resourceCpuLimit: specification.resourceLimits.cpuCores, + resourceMemoryLimitMb: specification.resourceLimits.memoryMb, }, ]; }); @@ -331,42 +383,39 @@ export function buildExpectedContainersFromRows({ async function buildServerlessExpectedState( server: Server, - allServices: Service[], + allServices: RuntimeServiceRevision[], containers: ExpectedContainer[], ): Promise<{ routes: ServerlessRoute[] }> { if (!server.isProxy) { return { routes: [] }; } - const serverlessServices = allServices.filter(isDeployedServerlessService); + const serverlessServices = allServices.filter( + (service) => service.specification.serverless.enabled, + ); if (serverlessServices.length === 0) return { routes: [] }; const serviceIds = serverlessServices.map((service) => service.id); - const [ports, deploymentRows] = await Promise.all([ - db - .select() - .from(servicePorts) - .where(inArray(servicePorts.serviceId, serviceIds)), - db - .select({ - id: deployments.id, - serviceId: deployments.serviceId, - serverId: deployments.serverId, - ipAddress: deployments.ipAddress, - runtimeDesiredState: deployments.runtimeDesiredState, - trafficState: deployments.trafficState, - observedPhase: deployments.observedPhase, - serverIsProxy: servers.isProxy, - }) - .from(deployments) - .innerJoin(servers, eq(deployments.serverId, servers.id)) - .where( - and( - inArray(deployments.serviceId, serviceIds), - inArray(deployments.runtimeDesiredState, runtimeExpectedStates), - ), + const deploymentRows = await db + .select({ + id: deployments.id, + serviceId: deployments.serviceId, + serverId: deployments.serverId, + ipAddress: deployments.ipAddress, + runtimeDesiredState: deployments.runtimeDesiredState, + trafficState: deployments.trafficState, + observedPhase: deployments.observedPhase, + serverIsProxy: servers.isProxy, + }) + .from(deployments) + .innerJoin(servers, eq(deployments.serverId, servers.id)) + .where( + and( + inArray(deployments.serviceId, serviceIds), + inArray(deployments.runtimeDesiredState, runtimeExpectedStates), ), - ]); + ); + const ports = buildRuntimeRoutePorts(serverlessServices); return { routes: buildServerlessRoutesFromRows({ @@ -387,7 +436,7 @@ export function buildServerlessRoutesFromRows({ containers, }: { serverId: string; - services: Service[]; + services: RuntimeServiceRevision[]; ports: RouteServicePort[]; deployments: ServerlessDeploymentRow[]; containers: ExpectedContainer[]; @@ -403,10 +452,10 @@ export function buildServerlessRoutesFromRows({ return serviceRows .flatMap((service) => - getRuntimeServicePortsForRoutes( + (portsByServiceId.get(service.id) ?? []).map((port) => ({ service, - portsByServiceId.get(service.id) ?? [], - ).map((port) => ({ service, port })), + port, + })), ) .sort((a, b) => compareServicePorts(a.port, b.port)) .flatMap(({ service, port }) => { @@ -451,10 +500,9 @@ export function buildServerlessRoutesFromRows({ serviceId: service.id, domain: port.domain, port: port.port, - sleepAfterSeconds: - getDeployedServerlessConfig(service).sleepAfterSeconds, + sleepAfterSeconds: service.specification.serverless.sleepAfterSeconds, wakeTimeoutSeconds: - getDeployedServerlessConfig(service).wakeTimeoutSeconds, + service.specification.serverless.wakeTimeoutSeconds, localDeploymentIds, upstreams, }, @@ -462,7 +510,7 @@ export function buildServerlessRoutesFromRows({ }); } -async function buildDnsRecords(allServices: Service[]) { +async function buildDnsRecords(allServices: RuntimeServiceRevision[]) { const serviceIds = allServices.map((service) => service.id); if (serviceIds.length === 0) return []; @@ -495,25 +543,21 @@ async function buildDnsRecords(allServices: Service[]) { if (ips.length === 0) return []; - const hostname = service.hostname || slugify(service.name); - return [{ name: `${hostname}.internal`, ips }]; + return [{ name: `${service.specification.hostname}.internal`, ips }]; }) .sort((a, b) => a.name.localeCompare(b.name)); } -async function buildTraefikConfig(server: Server, allServices: Service[]) { +async function buildTraefikConfig( + server: Server, + allServices: RuntimeServiceRevision[], +) { const emptyConfig = { httpRoutes: [], tcpRoutes: [], udpRoutes: [] }; if (!server.isProxy) return emptyConfig; const serviceIds = allServices.map((service) => service.id); - const [ports, routableDeployments, proxyHostedServerlessDeployments] = + const [routableDeployments, proxyHostedServerlessDeployments] = await Promise.all([ - serviceIds.length > 0 - ? db - .select() - .from(servicePorts) - .where(inArray(servicePorts.serviceId, serviceIds)) - : Promise.resolve([]), serviceIds.length > 0 ? db .select({ @@ -556,7 +600,7 @@ async function buildTraefikConfig(server: Server, allServices: Service[]) { proxyHostedServerlessDeployments, }); - const routePorts = buildRuntimeRoutePorts(allServices, ports); + const routePorts = buildRuntimeRoutePorts(allServices); const routes = buildTraefikRoutes({ serverId: server.id, ports: routePorts, @@ -687,7 +731,7 @@ export function buildServerlessTraefikRouteSets({ proxyHostedServerlessDeployments, }: { serverId: string; - services: Service[]; + services: RuntimeServiceRevision[]; proxyHostedServerlessDeployments: ProxyHostedServerlessDeploymentRow[]; }) { const proxyHostedServerlessServiceIds = new Set( @@ -702,7 +746,7 @@ export function buildServerlessTraefikRouteSets({ serviceRows .filter( (service) => - isDeployedServerlessService(service) && + service.specification.serverless.enabled && proxyHostedServerlessServiceIds.has(service.id), ) .map((service) => service.id), @@ -732,19 +776,7 @@ export function buildTraefikCertificateDomains(ports: RouteServicePort[]) { ).sort(); } -function buildHealthCheck(service: Service): ExpectedContainer["healthCheck"] { - if (!service.healthCheckCmd) return null; - - return { - cmd: service.healthCheckCmd, - interval: service.healthCheckInterval ?? 10, - timeout: service.healthCheckTimeout ?? 5, - retries: service.healthCheckRetries ?? 3, - startPeriod: service.healthCheckStartPeriod ?? 30, - }; -} - -function buildEnv(secretRows: SecretRow[]) { +function buildEnv(secretRows: ServiceRevisionSecret[]) { const env: Record = {}; for (const secret of secretRows .slice() @@ -796,44 +828,20 @@ function groupBy(items: T[], keyFn: (item: T) => K) { } export function buildRuntimeRoutePorts( - serviceRows: Service[], - portRows: ServicePort[], + serviceRows: RuntimeServiceRevision[], ): RouteServicePort[] { - const portsByServiceId = groupBy(portRows, (port) => port.serviceId); return serviceRows.flatMap((service) => - getRuntimeServicePortsForRoutes( - service, - portsByServiceId.get(service.id) ?? [], - ), - ); -} - -function getRuntimeServicePortsForRoutes( - service: Service, - livePorts: RouteServicePort[], -): RouteServicePort[] { - const ports = isDeployedServerlessService(service) - ? getDeployedServicePorts(service, livePorts) - : livePorts; - return ports.map((port, index) => { - const routePort = port as Partial; - return { - id: - typeof routePort.id === "string" - ? routePort.id - : `${service.id}:deployed:${index}`, + service.specification.ports.map((port, index) => ({ + id: `${service.revisionId}:${index}`, serviceId: service.id, - port: port.port, + port: port.containerPort, isPublic: port.isPublic, domain: port.domain, - protocol: port.protocol ?? "http", - externalPort: - typeof routePort.externalPort === "number" - ? routePort.externalPort - : null, - tlsPassthrough: routePort.tlsPassthrough ?? false, - }; - }); + protocol: port.protocol, + externalPort: port.externalPort, + tlsPassthrough: port.tlsPassthrough, + })), + ); } function compareServicePorts(a: RouteServicePort, b: RouteServicePort) { diff --git a/web/lib/deploy-service.ts b/web/lib/deploy-service.ts index 4f78e9d2..aa420c20 100644 --- a/web/lib/deploy-service.ts +++ b/web/lib/deploy-service.ts @@ -1,4 +1,3 @@ -import { randomUUID } from "node:crypto"; import { eq } from "drizzle-orm"; import { revalidatePath } from "next/cache"; import { db } from "@/db"; @@ -7,6 +6,7 @@ import { rollouts, serviceReplicas } from "@/db/schema"; import { inngest } from "@/lib/inngest/client"; import { inngestEvents } from "@/lib/inngest/events"; import { startMigrationInternal } from "@/lib/migrations"; +import { createRolloutWithServiceRevision } from "@/lib/service-revisions"; export async function deployServiceInternal(serviceId: string) { const service = await getService(serviceId); @@ -48,14 +48,7 @@ export async function deployServiceInternal(serviceId: string) { } } - const rolloutId = randomUUID(); - - await db.insert(rollouts).values({ - id: rolloutId, - serviceId, - status: "queued", - currentStage: "queued", - }); + const { rolloutId } = await createRolloutWithServiceRevision(serviceId); try { await inngest.send( diff --git a/web/lib/inngest/functions/rollout-helpers.ts b/web/lib/inngest/functions/rollout-helpers.ts index 20273c2d..3943b9f6 100644 --- a/web/lib/inngest/functions/rollout-helpers.ts +++ b/web/lib/inngest/functions/rollout-helpers.ts @@ -1,19 +1,15 @@ import { randomUUID } from "node:crypto"; import { and, eq, inArray } from "drizzle-orm"; import { db } from "@/db"; -import { getService } from "@/db/queries"; import { deploymentPorts, deployments, - secrets, + rollouts, servers, - servicePorts, - serviceReplicas, services, - serviceVolumes, } from "@/db/schema"; import { getCertificate, issueCertificate } from "@/lib/acme-manager"; -import { buildCurrentConfig, getDeployedStateful } from "@/lib/service-config"; +import type { ServiceRevisionSpec } from "@/lib/service-revision-spec"; import { assignContainerIp } from "@/lib/wireguard"; import { enqueueWork } from "@/lib/work-queue"; @@ -23,7 +19,8 @@ const PORT_RANGE_END = 32767; export type Placement = { serverId: string; replicas: number }; export type DeploymentContext = { - service: NonNullable>>; + revisionId: string; + specification: ServiceRevisionSpec; placements: Placement[]; serverMap: Map< string, @@ -33,20 +30,6 @@ export type DeploymentContext = { isRollingUpdate: boolean; }; -export function isActiveDeploymentForRollout( - deployment: { trafficState: string }, - service: { - deployedConfig?: string | null; - stateful?: boolean | null; - serverlessEnabled?: boolean | null; - serverlessSleepAfterSeconds?: number | null; - serverlessWakeTimeoutSeconds?: number | null; - }, -) { - void service; - return deployment.trafficState === "active"; -} - export function normalizeImage(image: string): string { if (!image.includes("/")) { return `docker.io/library/${image}`; @@ -64,14 +47,14 @@ async function getUsedPorts(serverId: string): Promise> { .innerJoin(deployments, eq(deploymentPorts.deploymentId, deployments.id)) .where(eq(deployments.serverId, serverId)); - return new Set(existingPorts.map((p) => p.hostPort)); + return new Set(existingPorts.map((port) => port.hostPort)); } export async function allocateHostPorts( serverId: string, count: number, ): Promise { - const usedPorts = await getUsedPorts(serverId); + const unavailablePorts = await getUsedPorts(serverId); const allocated: number[] = []; for ( @@ -79,7 +62,7 @@ export async function allocateHostPorts( port <= PORT_RANGE_END && allocated.length < count; port++ ) { - if (!usedPorts.has(port)) { + if (!unavailablePorts.has(port)) { allocated.push(port); } } @@ -91,22 +74,16 @@ export async function allocateHostPorts( return allocated; } -export async function calculateServicePlacements( - service: NonNullable>>, -): Promise<{ +export function calculateRevisionPlacements( + specification: ServiceRevisionSpec, +): { placements: Placement[]; totalReplicas: number; - migrationNeeded?: { targetServerId: string }; -}> { - const configuredReplicas = await db - .select({ - serverId: serviceReplicas.serverId, - replicas: serviceReplicas.count, - }) - .from(serviceReplicas) - .where(eq(serviceReplicas.serviceId, service.id)); - - const placements = configuredReplicas.filter((p) => p.replicas > 0); +} { + const placements = specification.placements.map((placement) => ({ + serverId: placement.serverId, + replicas: placement.count, + })); const totalReplicas = placements.reduce((sum, p) => sum + p.replicas, 0); if (totalReplicas < 1) { @@ -116,7 +93,7 @@ export async function calculateServicePlacements( throw new Error("Maximum 10 replicas allowed"); } - if (service.stateful) { + if (specification.stateful) { if (totalReplicas !== 1) { throw new Error("Stateful services can only have exactly 1 replica"); } @@ -127,14 +104,6 @@ export async function calculateServicePlacements( "Stateful services must be deployed to exactly one server", ); } - - const targetServerId = serverIds[0]; - if (service.lockedServerId && service.lockedServerId !== targetServerId) { - if (service.migrationStatus) { - throw new Error("Migration already in progress"); - } - return { placements, totalReplicas, migrationNeeded: { targetServerId } }; - } } return { placements, totalReplicas }; @@ -178,63 +147,28 @@ export async function validateServers( return serverMap; } -export async function prepareRollingUpdate( - serviceId: string, -): Promise<{ deploymentIds: string[] }> { - const service = await getService(serviceId); - if (!service) { - return { deploymentIds: [] }; - } - - const existingDeployments = await db - .select() - .from(deployments) - .where(eq(deployments.serviceId, serviceId)); - - const runningDeployments = existingDeployments.filter((d) => - isActiveDeploymentForRollout(d, service), - ); - - return { deploymentIds: runningDeployments.map((d) => d.id) }; -} - export async function cleanupTerminalDeployments( serviceId: string, ): Promise { - const terminalDeployments = await db - .select() - .from(deployments) + await db + .delete(deployments) .where( and( eq(deployments.serviceId, serviceId), eq(deployments.runtimeDesiredState, "removed"), ), ); - - for (const dep of terminalDeployments) { - await db - .delete(deploymentPorts) - .where(eq(deploymentPorts.deploymentId, dep.id)); - await db.delete(deployments).where(eq(deployments.id, dep.id)); - } } export async function cleanupExistingDeployments( serviceId: string, ): Promise<{ deletedCount: number }> { - const existingDeployments = await db - .select() - .from(deployments) - .where(eq(deployments.serviceId, serviceId)); + const deletedDeployments = await db + .delete(deployments) + .where(eq(deployments.serviceId, serviceId)) + .returning({ id: deployments.id }); - for (const dep of existingDeployments) { - await db - .delete(deploymentPorts) - .where(eq(deploymentPorts.deploymentId, dep.id)); - await db.delete(deployments).where(eq(deployments.id, dep.id)); - } - - return { deletedCount: existingDeployments.length }; + return { deletedCount: deletedDeployments.length }; } export type CertificateProvisioningResult = { @@ -244,17 +178,12 @@ export type CertificateProvisioningResult = { failedDomains: string[]; }; -export async function issueCertificatesForService( - serviceId: string, +export async function issueCertificatesForRevision( + specification: ServiceRevisionSpec, ): Promise { - const servicePortsList = await db - .select() - .from(servicePorts) - .where(eq(servicePorts.serviceId, serviceId)); - const domainsNeedingCerts = Array.from( new Set( - servicePortsList + specification.ports .filter((p) => p.isPublic && p.domain) .map((p) => (p.domain as string).trim()) .filter(Boolean), @@ -304,40 +233,9 @@ export async function createDeploymentRecords( serviceId: string, context: DeploymentContext, ): Promise<{ deploymentIds: string[] }> { - const { service, placements, serverMap, totalReplicas } = context; - - const servicePortsList = await db - .select() - .from(servicePorts) - .where(eq(servicePorts.serviceId, serviceId)); - - const serviceSecrets = await db - .select() - .from(secrets) - .where(eq(secrets.serviceId, serviceId)); - - const volumes = await db - .select() - .from(serviceVolumes) - .where(eq(serviceVolumes.serviceId, serviceId)); - - const env: Record = {}; - for (const secret of serviceSecrets) { - env[secret.key] = secret.encryptedValue; - } - - const updateData: { replicas: number; lockedServerId?: string } = { - replicas: totalReplicas, - }; - - if (service.stateful && !service.lockedServerId) { - updateData.lockedServerId = placements.map((p) => p.serverId)[0]; - } - - await db.update(services).set(updateData).where(eq(services.id, serviceId)); + const { revisionId, specification, placements, serverMap } = context; const deploymentIds: string[] = []; - let replicaIndex = 0; for (const placement of placements) { if (placement.replicas <= 0) continue; @@ -348,70 +246,43 @@ export async function createDeploymentRecords( } for (let i = 0; i < placement.replicas; i++) { - replicaIndex++; + const deploymentId = randomUUID(); const hostPorts = await allocateHostPorts( server.id, - servicePortsList.length, + specification.ports.length, ); const ipAddress = await assignContainerIp(server.id); - const deploymentId = randomUUID(); - deploymentIds.push(deploymentId); - - await db.insert(deployments).values({ - id: deploymentId, - serviceId, - serverId: server.id, - ipAddress, - runtimeDesiredState: "running", - trafficState: "candidate", - observedPhase: "pending", - rolloutId, - }); - - const portMappings: { containerPort: number; hostPort: number }[] = []; - for (let j = 0; j < servicePortsList.length; j++) { - const sp = servicePortsList[j]; - const hostPort = hostPorts[j]; - - await db.insert(deploymentPorts).values({ - id: randomUUID(), - deploymentId, - servicePortId: sp.id, - hostPort, + await db.transaction(async (tx) => { + await tx.insert(deployments).values({ + id: deploymentId, + serviceId, + serviceRevisionId: revisionId, + serverId: server.id, + ipAddress, + runtimeDesiredState: "running", + trafficState: "candidate", + observedPhase: "pending", + rolloutId, }); - portMappings.push({ containerPort: sp.port, hostPort }); - } + if (specification.ports.length > 0) { + await tx.insert(deploymentPorts).values( + specification.ports.map((port, index) => ({ + id: randomUUID(), + deploymentId, + containerPort: port.containerPort, + hostPort: hostPorts[index], + })), + ); + } + }); - const healthCheck = service.healthCheckCmd - ? { - cmd: service.healthCheckCmd, - interval: service.healthCheckInterval ?? 10, - timeout: service.healthCheckTimeout ?? 5, - retries: service.healthCheckRetries ?? 3, - startPeriod: service.healthCheckStartPeriod ?? 30, - } - : null; - - const volumeMounts = volumes.map((v) => ({ - name: v.name, - containerPath: v.containerPath, - })); + deploymentIds.push(deploymentId); await enqueueWork(server.id, "reconcile", { reason: "rollout_deployment_created", deploymentId, - serviceId, - serviceName: service.name, - image: normalizeImage(service.image), - portMappings, - wireguardIp: server.wireguardIp, - ipAddress, - name: `${serviceId}-${replicaIndex}`, - healthCheck, - env, - volumeMounts, }); } } @@ -419,76 +290,82 @@ export async function createDeploymentRecords( return { deploymentIds }; } -export async function saveDeployedConfig( +export async function completeRollout( + rolloutId: string, serviceId: string, - context: DeploymentContext, -): Promise { - const { placements, serverMap } = context; - - const servicePortsList = await db - .select() - .from(servicePorts) - .where(eq(servicePorts.serviceId, serviceId)); - - const serviceSecrets = await db - .select() - .from(secrets) - .where(eq(secrets.serviceId, serviceId)); - - const volumes = await db - .select() - .from(serviceVolumes) - .where(eq(serviceVolumes.serviceId, serviceId)); - - const replicaConfigs = placements - .filter((p) => p.replicas > 0) - .map((p) => ({ - serverId: p.serverId, - serverName: serverMap.get(p.serverId)?.name ?? "Unknown", - count: p.replicas, - })); - - const portConfigs = servicePortsList.map((p) => ({ - port: p.port, - isPublic: p.isPublic, - domain: p.domain, - protocol: p.protocol, - tlsPassthrough: p.tlsPassthrough, - })); - - const updatedService = await getService(serviceId); - if (updatedService) { - const deployedConfig = buildCurrentConfig( - updatedService, - replicaConfigs, - portConfigs, - serviceSecrets, - volumes, - ); + context: Omit, +): Promise<{ completed: boolean; stoppedCount: number }> { + const { placements, specification, totalReplicas, isRollingUpdate } = context; + const lockedServerId = specification.stateful + ? placements[0]?.serverId + : undefined; + + return db.transaction(async (tx) => { + const rollout = await tx + .select({ status: rollouts.status }) + .from(rollouts) + .where(eq(rollouts.id, rolloutId)) + .for("update") + .then((rows) => rows[0]); + if (rollout?.status !== "in_progress") { + return { completed: false, stoppedCount: 0 }; + } - await db + const stoppedDeployments = isRollingUpdate + ? await tx + .update(deployments) + .set({ + runtimeDesiredState: "removed", + trafficState: "inactive", + }) + .where( + and( + eq(deployments.serviceId, serviceId), + eq(deployments.trafficState, "draining"), + ), + ) + .returning({ id: deployments.id }) + : []; + + await tx .update(services) - .set({ deployedConfig: JSON.stringify(deployedConfig) }) + .set({ + replicas: totalReplicas, + ...(lockedServerId ? { lockedServerId } : {}), + }) .where(eq(services.id, serviceId)); - } + + await tx + .update(rollouts) + .set({ + status: "completed", + currentStage: "completed", + completedAt: new Date(), + }) + .where(eq(rollouts.id, rolloutId)); + return { completed: true, stoppedCount: stoppedDeployments.length }; + }); } export async function checkForRollingUpdate( serviceId: string, + specification: ServiceRevisionSpec, ): Promise { - const service = await getService(serviceId); - if (!service || getDeployedStateful(service)) { + if (specification.stateful) { return false; } - const existingDeployments = await db - .select() + const existingDeployment = await db + .select({ id: deployments.id }) .from(deployments) - .where(eq(deployments.serviceId, serviceId)); - - const runningDeployments = existingDeployments.filter((d) => - isActiveDeploymentForRollout(d, service), - ); + .where( + and( + eq(deployments.serviceId, serviceId), + eq(deployments.trafficState, "active"), + ), + ) + .limit(1) + .then((rows) => rows[0]); - return runningDeployments.length > 0; + return existingDeployment != null; } diff --git a/web/lib/inngest/functions/rollout-workflow.ts b/web/lib/inngest/functions/rollout-workflow.ts index d2f6d5ee..ccafa198 100644 --- a/web/lib/inngest/functions/rollout-workflow.ts +++ b/web/lib/inngest/functions/rollout-workflow.ts @@ -2,18 +2,19 @@ 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"; +import type { ServiceRevisionSpec } from "@/lib/service-revision-spec"; +import { getRolloutServiceRevision } from "@/lib/service-revisions"; import { ingestRolloutLog } from "@/lib/victoria-logs"; import { inngest } from "../client"; import { inngestEvents } from "../events"; import { - calculateServicePlacements, + calculateRevisionPlacements, checkForRollingUpdate, cleanupExistingDeployments, cleanupTerminalDeployments, + completeRollout, createDeploymentRecords, - issueCertificatesForService, - prepareRollingUpdate, - saveDeployedConfig, + issueCertificatesForRevision, validateServers, } from "./rollout-helpers"; import { handleRolloutFailure } from "./rollout-utils"; @@ -193,6 +194,19 @@ export const rolloutWorkflow = inngest.createFunction( return { status: "failed", rolloutId, reason: "queue_timeout" }; } + const revision = await step.run("load-service-revision", async () => { + const rolloutRevision = await getRolloutServiceRevision(rolloutId); + const executionSpecification: ServiceRevisionSpec = { + ...rolloutRevision.specification, + secrets: [], + }; + return { + id: rolloutRevision.id, + specification: executionSpecification, + }; + }); + const specification = revision.specification; + await step.run("log-rollout-started", async () => { await ingestRolloutLog( rolloutId, @@ -203,12 +217,8 @@ export const rolloutWorkflow = inngest.createFunction( }); const placementResult = await step.run("load-placements", async () => { - const service = await getService(serviceId); - if (!service) { - throw new Error("Service not found"); - } try { - const result = await calculateServicePlacements(service); + const result = calculateRevisionPlacements(specification); await ingestRolloutLog( rolloutId, serviceId, @@ -286,20 +296,10 @@ export const rolloutWorkflow = inngest.createFunction( }); const isRollingUpdate = await step.run("check-rolling-update", async () => { - return checkForRollingUpdate(serviceId); + return checkForRollingUpdate(serviceId, specification); }); - if (isRollingUpdate) { - await step.run("prepare-rolling-update", async () => { - await prepareRollingUpdate(serviceId); - await ingestRolloutLog( - rolloutId, - serviceId, - "preparing", - "Prepared rolling update", - ); - }); - } else { + if (!isRollingUpdate) { await step.run("cleanup-existing", async () => { const { deletedCount } = await cleanupExistingDeployments(serviceId); if (deletedCount > 0) { @@ -319,7 +319,7 @@ export const rolloutWorkflow = inngest.createFunction( .set({ currentStage: "certificates" }) .where(eq(rollouts.id, rolloutId)); try { - const result = await issueCertificatesForService(serviceId); + const result = await issueCertificatesForRevision(specification); if (result.issuedDomains.length > 0) { await ingestRolloutLog( rolloutId, @@ -360,15 +360,11 @@ export const rolloutWorkflow = inngest.createFunction( .set({ currentStage: "deploying" }) .where(eq(rollouts.id, rolloutId)); - const service = await getService(serviceId); - if (!service) { - throw new Error("Service not found"); - } - const serverMap = await validateServers(placements); const result = await createDeploymentRecords(rolloutId, serviceId, { - service, + revisionId: revision.id, + specification, placements, serverMap, totalReplicas, @@ -386,8 +382,7 @@ export const rolloutWorkflow = inngest.createFunction( }); await step.run("start-health-check", async () => { - const service = await getService(serviceId); - const hasHealthCheck = service?.healthCheckCmd != null; + const hasHealthCheck = specification.healthCheck != null; await db .update(rollouts) @@ -583,84 +578,33 @@ export const rolloutWorkflow = inngest.createFunction( return { status: "rolled_back", rolloutId, reason: "dns_sync_timeout" }; } - if (isRollingUpdate) { - await step.run("stop-old-deployments", async () => { - const stoppedDeploymentsWithoutContainers = await db - .update(deployments) - .set({ - runtimeDesiredState: "removed", - trafficState: "inactive", - }) - .where( - and( - eq(deployments.serviceId, serviceId), - eq(deployments.trafficState, "draining"), - isNull(deployments.containerId), - ), - ) - .returning({ id: deployments.id }); - - const stoppingDeployments = await db - .update(deployments) - .set({ - runtimeDesiredState: "removed", - trafficState: "inactive", - }) - .where( - and( - eq(deployments.serviceId, serviceId), - eq(deployments.trafficState, "draining"), - ), - ) - .returning({ id: deployments.id }); - - const stoppedCount = - stoppedDeploymentsWithoutContainers.length + - stoppingDeployments.length; - if (stoppedCount > 0) { - await ingestRolloutLog( - rolloutId, - serviceId, - "dns_sync", - `Stopping ${stoppedCount} old deployment(s) after DNS sync`, - ); - } - }); - } - - await step.run("save-deployed-config", async () => { - const service = await getService(serviceId); - if (!service) { - throw new Error("Service not found"); - } - - const serverMap = await validateServers(placements); - - await saveDeployedConfig(serviceId, { - service, + const rolloutCompleted = await step.run("complete-rollout", async () => { + const result = await completeRollout(rolloutId, serviceId, { + specification, placements, - serverMap, totalReplicas, isRollingUpdate, }); - }); - - await step.run("complete-rollout", async () => { - await db - .update(rollouts) - .set({ - status: "completed", - currentStage: "completed", - completedAt: new Date(), - }) - .where(eq(rollouts.id, rolloutId)); + if (!result.completed) return false; + if (result.stoppedCount > 0) { + await ingestRolloutLog( + rolloutId, + serviceId, + "dns_sync", + `Stopping ${result.stoppedCount} old deployment(s) after DNS sync`, + ); + } await ingestRolloutLog( rolloutId, serviceId, "completed", "Rollout completed successfully", ); + return true; }); + if (!rolloutCompleted) { + return { status: "cancelled", rolloutId }; + } return { status: "completed", rolloutId }; }, diff --git a/web/lib/service-config.ts b/web/lib/service-config.ts index 742aa2aa..ece73e89 100644 --- a/web/lib/service-config.ts +++ b/web/lib/service-config.ts @@ -1,3 +1,5 @@ +import type { ServiceRevisionSpec } from "@/lib/service-revision-spec"; + export type ReplicaConfig = { serverId: string; serverName: string; @@ -537,17 +539,6 @@ export function diffConfigs( return changes; } -export function parseDeployedConfig( - json: string | null, -): DeployedConfig | null { - if (!json) return null; - try { - return JSON.parse(json) as DeployedConfig; - } catch { - return null; - } -} - export function normalizeServerlessConfig( config: ServerlessConfig | undefined, ): ServerlessConfig { @@ -580,49 +571,43 @@ export function getCurrentServerlessConfig(service: { }; } -export function getDeployedServerlessConfig(service: { - deployedConfig?: string | null; - serverlessEnabled?: boolean | null; - serverlessSleepAfterSeconds?: number | null; - serverlessWakeTimeoutSeconds?: number | null; -}): ServerlessConfig { - const deployed = parseDeployedConfig(service.deployedConfig ?? null); - return deployed?.serverless - ? normalizeServerlessConfig(deployed.serverless) - : getCurrentServerlessConfig(service); -} - -export function getDeployedStateful(service: { - deployedConfig?: string | null; - stateful?: boolean | null; -}) { - const deployed = parseDeployedConfig(service.deployedConfig ?? null); - return deployed?.stateful ?? service.stateful ?? false; -} - -export function isDeployedServerlessService(service: { - deployedConfig?: string | null; - stateful?: boolean | null; - serverlessEnabled?: boolean | null; - serverlessSleepAfterSeconds?: number | null; - serverlessWakeTimeoutSeconds?: number | null; -}) { - return getDeployedServerlessConfig(service).enabled; -} - -export function getDeployedServicePorts( - service: { id: string; deployedConfig?: string | null }, - livePorts: PortConfig[], -): PortConfig[] { - const deployed = parseDeployedConfig(service.deployedConfig ?? null); - if (!deployed?.ports) { - return livePorts; - } - return deployed.ports.map((port) => ({ - port: port.port, - isPublic: port.isPublic, - domain: port.domain, - protocol: port.protocol ?? "http", - tlsPassthrough: port.tlsPassthrough, - })); +export function revisionSpecToDeployedConfig( + specification: ServiceRevisionSpec, + serverNames: Record, +): DeployedConfig { + return { + source: { type: "image", image: specification.image }, + hostname: specification.hostname, + stateful: specification.stateful, + placement: { + replicas: specification.placements.reduce( + (sum, placement) => sum + placement.count, + 0, + ), + }, + replicas: specification.placements.map((placement) => ({ + serverId: placement.serverId, + serverName: serverNames[placement.serverId] ?? "Unknown", + count: placement.count, + })), + healthCheck: specification.healthCheck, + startCommand: specification.startCommand, + resourceLimits: { + cpuCores: specification.resourceLimits.cpuCores, + memoryMb: specification.resourceLimits.memoryMb, + }, + ports: specification.ports.map((port) => ({ + port: port.containerPort, + isPublic: port.isPublic, + domain: port.domain, + protocol: port.protocol, + tlsPassthrough: port.tlsPassthrough, + })), + serverless: specification.serverless, + secrets: specification.secrets.map((secret) => ({ + key: secret.key, + updatedAt: secret.updatedAt, + })), + volumes: specification.volumes, + }; } diff --git a/web/lib/service-revision-spec.ts b/web/lib/service-revision-spec.ts new file mode 100644 index 00000000..a8f7aeed --- /dev/null +++ b/web/lib/service-revision-spec.ts @@ -0,0 +1,203 @@ +export const SERVICE_REVISION_SCHEMA_VERSION = 1 as const; + +export function getDefaultServiceHostname(name: string): string { + return name + .toLowerCase() + .replace(/[^a-z0-9]+/g, "-") + .replace(/^-|-$/g, ""); +} + +export type ServiceRevisionHealthCheck = { + cmd: string; + interval: number; + timeout: number; + retries: number; + startPeriod: number; +}; + +export type ServiceRevisionPlacement = { + serverId: string; + count: number; +}; + +export type ServiceRevisionPort = { + containerPort: number; + isPublic: boolean; + domain: string | null; + protocol: "http" | "tcp" | "udp"; + externalPort: number | null; + tlsPassthrough: boolean; +}; + +export type ServiceRevisionSecret = { + key: string; + encryptedValue: string; + updatedAt: string; +}; + +export type ServiceRevisionVolume = { + name: string; + containerPath: string; +}; + +export type ServiceRevisionSpec = { + schemaVersion: typeof SERVICE_REVISION_SCHEMA_VERSION; + image: string; + hostname: string; + stateful: boolean; + serverless: { + enabled: boolean; + sleepAfterSeconds: number; + wakeTimeoutSeconds: number; + }; + healthCheck: ServiceRevisionHealthCheck | null; + startCommand: string | null; + resourceLimits: { + cpuCores: number | null; + memoryMb: number | null; + }; + placements: ServiceRevisionPlacement[]; + ports: ServiceRevisionPort[]; + secrets: ServiceRevisionSecret[]; + volumes: ServiceRevisionVolume[]; +}; + +export type ServiceRevisionDraft = { + service: { + name: string; + image: string; + hostname: string | null; + stateful: boolean | null; + serverlessEnabled: boolean | null; + serverlessSleepAfterSeconds: number | null; + serverlessWakeTimeoutSeconds: number | null; + healthCheckCmd: string | null; + healthCheckInterval: number | null; + healthCheckTimeout: number | null; + healthCheckRetries: number | null; + healthCheckStartPeriod: number | null; + startCommand: string | null; + resourceCpuLimit: number | null; + resourceMemoryLimitMb: number | null; + }; + placements: Array<{ serverId: string; count: number }>; + ports: Array<{ + port: number; + isPublic: boolean; + domain: string | null; + protocol: "http" | "tcp" | "udp" | null; + externalPort: number | null; + tlsPassthrough: boolean | null; + }>; + secrets: Array<{ + key: string; + encryptedValue: string; + updatedAt: Date | string; + }>; + volumes: Array<{ name: string; containerPath: string }>; +}; + +function validateServiceRevisionSpec(specification: ServiceRevisionSpec) { + const totalReplicas = specification.placements.reduce( + (sum, placement) => sum + placement.count, + 0, + ); + + if (totalReplicas < 1) { + throw new Error("At least one replica is required"); + } + if (totalReplicas > 10) { + throw new Error("Maximum 10 replicas allowed"); + } + if (specification.stateful && totalReplicas !== 1) { + throw new Error("Stateful services can only have exactly 1 replica"); + } + if (specification.stateful && specification.placements.length !== 1) { + throw new Error("Stateful services must be deployed to exactly one server"); + } +} + +function compareStrings(a: string, b: string) { + return a.localeCompare(b, "en"); +} + +export function buildServiceRevisionSpec( + draft: ServiceRevisionDraft, +): ServiceRevisionSpec { + const { service } = draft; + + const specification: ServiceRevisionSpec = { + schemaVersion: SERVICE_REVISION_SCHEMA_VERSION, + image: service.image.trim(), + hostname: + service.hostname?.trim() || getDefaultServiceHostname(service.name), + stateful: service.stateful ?? false, + serverless: { + enabled: service.serverlessEnabled ?? false, + sleepAfterSeconds: Math.max( + service.serverlessSleepAfterSeconds ?? 300, + 120, + ), + wakeTimeoutSeconds: service.serverlessWakeTimeoutSeconds ?? 300, + }, + healthCheck: service.healthCheckCmd + ? { + cmd: service.healthCheckCmd, + interval: service.healthCheckInterval ?? 10, + timeout: service.healthCheckTimeout ?? 5, + retries: service.healthCheckRetries ?? 3, + startPeriod: service.healthCheckStartPeriod ?? 30, + } + : null, + startCommand: service.startCommand?.trim() || null, + resourceLimits: { + cpuCores: service.resourceCpuLimit, + memoryMb: service.resourceMemoryLimitMb, + }, + placements: draft.placements + .filter((placement) => placement.count > 0) + .map((placement) => ({ + serverId: placement.serverId, + count: placement.count, + })) + .sort((a, b) => compareStrings(a.serverId, b.serverId)), + ports: draft.ports + .map((port) => ({ + containerPort: port.port, + isPublic: port.isPublic, + domain: port.domain?.trim() || null, + protocol: port.protocol ?? "http", + externalPort: port.externalPort, + tlsPassthrough: port.tlsPassthrough ?? false, + })) + .sort( + (a, b) => + a.containerPort - b.containerPort || + compareStrings(a.protocol, b.protocol) || + compareStrings(a.domain ?? "", b.domain ?? "") || + (a.externalPort ?? 0) - (b.externalPort ?? 0), + ), + secrets: draft.secrets + .map((secret) => ({ + key: secret.key, + encryptedValue: secret.encryptedValue, + updatedAt: + secret.updatedAt instanceof Date + ? secret.updatedAt.toISOString() + : secret.updatedAt, + })) + .sort((a, b) => compareStrings(a.key, b.key)), + volumes: draft.volumes + .map((volume) => ({ + name: volume.name, + containerPath: volume.containerPath, + })) + .sort( + (a, b) => + compareStrings(a.name, b.name) || + compareStrings(a.containerPath, b.containerPath), + ), + }; + validateServiceRevisionSpec(specification); + return specification; +} diff --git a/web/lib/service-revisions.ts b/web/lib/service-revisions.ts new file mode 100644 index 00000000..eb43fafa --- /dev/null +++ b/web/lib/service-revisions.ts @@ -0,0 +1,113 @@ +import { randomUUID } from "node:crypto"; +import { and, eq, isNull } from "drizzle-orm"; +import { db } from "@/db"; +import { + rollouts, + secrets, + servicePorts, + serviceReplicas, + serviceRevisions, + services, + serviceVolumes, +} from "@/db/schema"; +import { + buildServiceRevisionSpec, + SERVICE_REVISION_SCHEMA_VERSION, +} from "@/lib/service-revision-spec"; + +export async function createRolloutWithServiceRevision(serviceId: string) { + return db.transaction( + async (tx) => { + const service = await tx + .select() + .from(services) + .where(and(eq(services.id, serviceId), isNull(services.deletedAt))) + .then((rows) => rows[0]); + + if (!service) { + throw new Error("Service not found"); + } + + const placements = await tx + .select({ + serverId: serviceReplicas.serverId, + count: serviceReplicas.count, + }) + .from(serviceReplicas) + .where(eq(serviceReplicas.serviceId, serviceId)); + const ports = await tx + .select() + .from(servicePorts) + .where(eq(servicePorts.serviceId, serviceId)); + const revisionSecrets = await tx + .select({ + key: secrets.key, + encryptedValue: secrets.encryptedValue, + updatedAt: secrets.updatedAt, + }) + .from(secrets) + .where(eq(secrets.serviceId, serviceId)); + const volumes = await tx + .select({ + name: serviceVolumes.name, + containerPath: serviceVolumes.containerPath, + }) + .from(serviceVolumes) + .where(eq(serviceVolumes.serviceId, serviceId)); + + const specification = buildServiceRevisionSpec({ + service, + placements, + ports, + secrets: revisionSecrets, + volumes, + }); + const revision = await tx + .insert(serviceRevisions) + .values({ id: randomUUID(), serviceId, specification }) + .returning() + .then((rows) => rows[0]); + if (!revision) { + throw new Error("Failed to create service revision"); + } + + const rolloutId = randomUUID(); + await tx.insert(rollouts).values({ + id: rolloutId, + serviceId, + serviceRevisionId: revision.id, + status: "queued", + currentStage: "queued", + }); + + return { rolloutId, revision }; + }, + { isolationLevel: "repeatable read" }, + ); +} + +export async function getRolloutServiceRevision(rolloutId: string) { + const result = await db + .select({ + revision: serviceRevisions, + }) + .from(rollouts) + .innerJoin( + serviceRevisions, + eq(rollouts.serviceRevisionId, serviceRevisions.id), + ) + .where(eq(rollouts.id, rolloutId)) + .then((rows) => rows[0]); + + if (!result) { + throw new Error("Rollout revision not found"); + } + if ( + result.revision.specification.schemaVersion !== + SERVICE_REVISION_SCHEMA_VERSION + ) { + throw new Error("Unsupported service revision schema version"); + } + + return result.revision; +} diff --git a/web/tests/expected-state.test.ts b/web/tests/expected-state.test.ts index 4848957d..fcc15421 100644 --- a/web/tests/expected-state.test.ts +++ b/web/tests/expected-state.test.ts @@ -14,27 +14,52 @@ import { buildTraefikCertificateDomains, buildTraefikRoutes, } from "@/lib/agent/expected-state"; +import type { ServiceRevisionSpec } from "@/lib/service-revision-spec"; describe("expected-state pure builders", () => { - const deployedServerlessConfig = JSON.stringify({ - source: { type: "image", image: "nginx" }, - stateful: false, - replicas: [], - healthCheck: null, - ports: [ - { - port: 3000, - isPublic: true, - domain: "sleepy.example.com", - protocol: "http", + function revision( + serviceId: string, + overrides: Partial = {}, + ) { + return { + id: `rev_${serviceId}`, + name: serviceId, + serviceId, + revisionId: `rev_${serviceId}`, + specification: { + schemaVersion: 1, + image: "nginx", + hostname: serviceId, + stateful: false, + serverless: { + enabled: false, + sleepAfterSeconds: 300, + wakeTimeoutSeconds: 300, + }, + healthCheck: null, + startCommand: null, + resourceLimits: { cpuCores: null, memoryMb: null }, + placements: [], + ports: [], + secrets: [], + volumes: [], + ...overrides, }, - ], - serverless: { - enabled: true, - sleepAfterSeconds: 300, - wakeTimeoutSeconds: 120, - }, - }); + } as any; + } + + function runtimeRevision( + serviceId: string, + overrides: Partial = {}, + ) { + const revisionRow = revision(serviceId, overrides); + return { + id: serviceId, + name: serviceId, + revisionId: revisionRow.id, + specification: revisionRow.specification, + }; + } it("groups container inputs by deployment and service deterministically", () => { const containers = buildExpectedContainersFromRows({ @@ -42,6 +67,7 @@ describe("expected-state pure builders", () => { { id: "dep_bbbbbbbb", serviceId: "svc_1", + serviceRevisionId: "rev_svc_1", ipAddress: "10.0.0.2", runtimeDesiredState: "stopped", trafficState: "active", @@ -50,12 +76,65 @@ describe("expected-state pure builders", () => { { id: "dep_aaaaaaaa", serviceId: "svc_1", + serviceRevisionId: "rev_svc_1", ipAddress: "10.0.0.1", runtimeDesiredState: "running", trafficState: "active", observedPhase: "running", }, ] as any, + revisions: [ + revision("svc_1", { + startCommand: "npm start", + healthCheck: { + cmd: "curl -f /health", + interval: 10, + timeout: 5, + retries: 3, + startPeriod: 30, + }, + resourceLimits: { cpuCores: 1, memoryMb: 512 }, + serverless: { + enabled: true, + sleepAfterSeconds: 300, + wakeTimeoutSeconds: 300, + }, + ports: [ + { + containerPort: 80, + isPublic: true, + domain: "api.example.com", + protocol: "http", + externalPort: null, + tlsPassthrough: false, + }, + { + containerPort: 3000, + isPublic: false, + domain: null, + protocol: "http", + externalPort: null, + tlsPassthrough: false, + }, + ], + secrets: [ + { + key: "ZED", + encryptedValue: "last", + updatedAt: "2026-07-01T00:00:00.000Z", + }, + { + key: "ALPHA", + encryptedValue: "first", + updatedAt: "2026-07-01T00:00:00.000Z", + }, + ], + volumes: [ + { name: "cache", containerPath: "/var/cache" }, + { name: "data", containerPath: "/data" }, + ], + }), + ], services: [ { id: "svc_1", @@ -75,14 +154,8 @@ describe("expected-state pure builders", () => { deploymentPorts: [ { deploymentId: "dep_aaaaaaaa", hostPort: 30001, containerPort: 3000 }, { deploymentId: "dep_aaaaaaaa", hostPort: 80, containerPort: 80 }, - ] as any, - secrets: [ - { serviceId: "svc_1", key: "ZED", encryptedValue: "last" }, - { serviceId: "svc_1", key: "ALPHA", encryptedValue: "first" }, - ] as any, - volumes: [ - { serviceId: "svc_1", name: "cache", containerPath: "/var/cache" }, - { serviceId: "svc_1", name: "data", containerPath: "/data" }, + { deploymentId: "dep_bbbbbbbb", hostPort: 30002, containerPort: 3000 }, + { deploymentId: "dep_bbbbbbbb", hostPort: 81, containerPort: 80 }, ] as any, }); @@ -118,6 +191,73 @@ describe("expected-state pure builders", () => { }); }); + it("rejects partial expected state when a deployment revision is missing", () => { + expect(() => + buildExpectedContainersFromRows({ + deployments: [ + { + id: "dep_missing_revision", + serviceId: "svc_1", + serviceRevisionId: "rev_missing", + runtimeDesiredState: "running", + }, + ] as any, + services: [{ id: "svc_1", name: "api" }] as any, + revisions: [], + deploymentPorts: [], + }), + ).toThrow("Deployment dep_missing_revision has no service revision"); + }); + + it("omits deployments whose service was soft-deleted", () => { + expect( + buildExpectedContainersFromRows({ + deployments: [ + { + id: "dep_deleted_service", + serviceId: "svc_1", + serviceRevisionId: "rev_svc_1", + runtimeDesiredState: "running", + }, + ] as any, + services: [], + revisions: [revision("svc_1")], + deploymentPorts: [], + }), + ).toEqual([]); + }); + + it("rejects partial expected state when deployment ports are incomplete", () => { + expect(() => + buildExpectedContainersFromRows({ + deployments: [ + { + id: "dep_incomplete_ports", + serviceId: "svc_1", + serviceRevisionId: "rev_svc_1", + runtimeDesiredState: "running", + }, + ] as any, + services: [{ id: "svc_1", name: "api" }] as any, + revisions: [ + revision("svc_1", { + ports: [ + { + containerPort: 3000, + isPublic: false, + domain: null, + protocol: "http", + externalPort: null, + tlsPassthrough: false, + }, + ], + }), + ], + deploymentPorts: [], + }), + ).toThrow("Deployment dep_incomplete_ports has incomplete port allocation"); + }); + it("keeps HTTP local upstreams before remote upstreams", () => { const routes = buildTraefikRoutes({ serverId: "server_local", @@ -162,12 +302,22 @@ describe("expected-state pure builders", () => { { id: "dep_sleeping", serviceId: "svc_private", + serviceRevisionId: "rev_svc_private", ipAddress: "10.0.0.10", runtimeDesiredState: "running", trafficState: "active", observedPhase: "running", }, ] as any, + revisions: [ + revision("svc_private", { + serverless: { + enabled: true, + sleepAfterSeconds: 300, + wakeTimeoutSeconds: 300, + }, + }), + ], services: [ { id: "svc_private", @@ -177,8 +327,6 @@ describe("expected-state pure builders", () => { }, ] as any, deploymentPorts: [], - secrets: [], - volumes: [], }); expect(containers[0]).toMatchObject({ @@ -193,12 +341,14 @@ describe("expected-state pure builders", () => { { id: "dep_stopped", serviceId: "svc_private", + serviceRevisionId: "rev_svc_private", ipAddress: "10.0.0.10", runtimeDesiredState: "stopped", trafficState: "active", observedPhase: "sleeping", }, ] as any, + revisions: [revision("svc_private")], services: [ { id: "svc_private", @@ -208,8 +358,6 @@ describe("expected-state pure builders", () => { }, ] as any, deploymentPorts: [], - secrets: [], - volumes: [], }); expect(containers[0]).toMatchObject({ @@ -225,12 +373,22 @@ describe("expected-state pure builders", () => { { id: "dep_sleeping", serviceId: "svc_public", + serviceRevisionId: "rev_svc_public", ipAddress: "10.0.0.10", runtimeDesiredState: "stopped", trafficState: "active", observedPhase: "sleeping", }, ] as any, + revisions: [ + revision("svc_public", { + serverless: { + enabled: true, + sleepAfterSeconds: 300, + wakeTimeoutSeconds: 300, + }, + }), + ], services: [ { id: "svc_public", @@ -240,8 +398,6 @@ describe("expected-state pure builders", () => { }, ] as any, deploymentPorts: [], - secrets: [], - volumes: [], }); expect(containers[0]).toMatchObject({ @@ -251,30 +407,37 @@ describe("expected-state pure builders", () => { }); }); - it("keeps deployed serverless sleeping while live serverless settings are disabled", () => { + it("keeps revision serverless behavior while draft settings are disabled", () => { const containers = buildExpectedContainersFromRows({ deployments: [ { id: "dep_sleeping", serviceId: "svc_public", + serviceRevisionId: "rev_svc_public", ipAddress: "10.0.0.10", runtimeDesiredState: "stopped", trafficState: "active", observedPhase: "sleeping", }, ] as any, + revisions: [ + revision("svc_public", { + serverless: { + enabled: true, + sleepAfterSeconds: 300, + wakeTimeoutSeconds: 120, + }, + }), + ], services: [ { id: "svc_public", name: "public-api", image: "nginx", serverlessEnabled: false, - deployedConfig: deployedServerlessConfig, }, ] as any, deploymentPorts: [], - secrets: [], - volumes: [], }); expect(containers[0]).toMatchObject({ @@ -290,6 +453,7 @@ describe("expected-state pure builders", () => { { id: "dep_draining", serviceId: "svc_public", + serviceRevisionId: "rev_svc_public", ipAddress: "10.0.0.10", runtimeDesiredState: "stopped", trafficState: "draining", @@ -297,6 +461,15 @@ describe("expected-state pure builders", () => { containerId: null, }, ] as any, + revisions: [ + revision("svc_public", { + serverless: { + enabled: true, + sleepAfterSeconds: 300, + wakeTimeoutSeconds: 300, + }, + }), + ], services: [ { id: "svc_public", @@ -306,8 +479,6 @@ describe("expected-state pure builders", () => { }, ] as any, deploymentPorts: [], - secrets: [], - volumes: [], }); expect(containers[0]).toMatchObject({ @@ -347,12 +518,14 @@ describe("expected-state pure builders", () => { const services: Parameters< typeof buildServerlessTraefikRouteSets >[0]["services"] = [ - { - id: "svc_serverless", - serverlessEnabled: true, - stateful: false, - }, - ] as Parameters[0]["services"]; + runtimeRevision("svc_serverless", { + serverless: { + enabled: true, + sleepAfterSeconds: 300, + wakeTimeoutSeconds: 300, + }, + }), + ]; const ports: Parameters[0]["ports"] = [ { id: "port_1", @@ -536,14 +709,14 @@ describe("expected-state pure builders", () => { const routes = buildServerlessRoutesFromRows({ serverId: "proxy_1", services: [ - { - id: "svc_1", - serverlessEnabled: true, - stateful: false, - serverlessSleepAfterSeconds: 300, - serverlessWakeTimeoutSeconds: 120, - }, - ] as any, + runtimeRevision("svc_1", { + serverless: { + enabled: true, + sleepAfterSeconds: 300, + wakeTimeoutSeconds: 120, + }, + }), + ], ports: [ { id: "port_1", @@ -619,14 +792,15 @@ describe("expected-state pure builders", () => { const routes = buildServerlessRoutesFromRows({ serverId: "proxy_1", services: [ - { - id: "svc_stateful", - serverlessEnabled: true, + runtimeRevision("svc_stateful", { stateful: true, - serverlessSleepAfterSeconds: 300, - serverlessWakeTimeoutSeconds: 120, - }, - ] as any, + serverless: { + enabled: true, + sleepAfterSeconds: 300, + wakeTimeoutSeconds: 120, + }, + }), + ], ports: [ { id: "port_1", @@ -674,14 +848,14 @@ describe("expected-state pure builders", () => { const routes = buildServerlessRoutesFromRows({ serverId: "proxy_1", services: [ - { - id: "svc_1", - serverlessEnabled: true, - stateful: false, - serverlessSleepAfterSeconds: 300, - serverlessWakeTimeoutSeconds: 120, - }, - ] as any, + runtimeRevision("svc_1", { + serverless: { + enabled: true, + sleepAfterSeconds: 300, + wakeTimeoutSeconds: 120, + }, + }), + ], ports: [ { id: "port_1", @@ -726,20 +900,28 @@ describe("expected-state pure builders", () => { ).toEqual(["dep_new"]); }); - it("keeps wake metadata from deployed serverless config while live settings are disabled", () => { - const routes = buildServerlessRoutesFromRows({ - serverId: "proxy_1", - services: [ + it("keeps wake metadata and ports pinned to the active revision", () => { + const service = runtimeRevision("svc_1", { + serverless: { + enabled: true, + sleepAfterSeconds: 300, + wakeTimeoutSeconds: 120, + }, + ports: [ { - id: "svc_1", - serverlessEnabled: false, - stateful: false, - serverlessSleepAfterSeconds: 60, - serverlessWakeTimeoutSeconds: 60, - deployedConfig: deployedServerlessConfig, + containerPort: 3000, + isPublic: true, + domain: "sleepy.example.com", + protocol: "http", + externalPort: null, + tlsPassthrough: false, }, - ] as any, - ports: [], + ], + }); + const routes = buildServerlessRoutesFromRows({ + serverId: "proxy_1", + services: [service], + ports: buildRuntimeRoutePorts([service]), deployments: [ { id: "dep_sleeping", @@ -770,18 +952,26 @@ describe("expected-state pure builders", () => { ]); }); - it("keeps Traefik gateway route from deployed serverless ports while live ports are removed", () => { - const ports = buildRuntimeRoutePorts( - [ - { - id: "svc_public", - serverlessEnabled: false, - stateful: false, - deployedConfig: deployedServerlessConfig, + it("keeps Traefik gateway routes pinned to revision ports", () => { + const ports = buildRuntimeRoutePorts([ + runtimeRevision("svc_public", { + serverless: { + enabled: true, + sleepAfterSeconds: 300, + wakeTimeoutSeconds: 120, }, - ] as any, - [], - ); + ports: [ + { + containerPort: 3000, + isPublic: true, + domain: "sleepy.example.com", + protocol: "http", + externalPort: null, + tlsPassthrough: false, + }, + ], + }), + ]); const routes = buildTraefikRoutes({ serverId: "proxy_1", diff --git a/web/tests/rollout-helpers.test.ts b/web/tests/rollout-helpers.test.ts deleted file mode 100644 index 65f96954..00000000 --- a/web/tests/rollout-helpers.test.ts +++ /dev/null @@ -1,39 +0,0 @@ -import { describe, expect, it, vi } from "vitest"; - -vi.mock("@/db", () => ({ db: {} })); -vi.mock("@/db/queries", () => ({ getService: vi.fn() })); -vi.mock("@/lib/acme-manager", () => ({ - getCertificate: vi.fn(), - issueCertificate: vi.fn(), -})); -vi.mock("@/lib/wireguard", () => ({ - assignContainerIp: vi.fn(), -})); -vi.mock("@/lib/work-queue", () => ({ - enqueueWork: vi.fn(), -})); - -import { isActiveDeploymentForRollout } from "@/lib/inngest/functions/rollout-helpers"; - -describe("rollout helpers", () => { - it("treats active traffic deployments as the live rollout version", () => { - expect( - isActiveDeploymentForRollout( - { trafficState: "active" }, - { serverlessEnabled: true }, - ), - ).toBe(true); - expect( - isActiveDeploymentForRollout( - { trafficState: "candidate" }, - { serverlessEnabled: true }, - ), - ).toBe(false); - expect( - isActiveDeploymentForRollout( - { trafficState: "draining" }, - { serverlessEnabled: false }, - ), - ).toBe(false); - }); -}); diff --git a/web/tests/service-config.test.ts b/web/tests/service-config.test.ts index 95444145..8609fed8 100644 --- a/web/tests/service-config.test.ts +++ b/web/tests/service-config.test.ts @@ -2,10 +2,9 @@ import { describe, expect, it } from "vitest"; import { type DeployedConfig, diffConfigs, - getDeployedServerlessConfig, getCurrentServerlessConfig, - isDeployedServerlessService, MIN_SERVERLESS_SLEEP_AFTER_SECONDS, + revisionSpecToDeployedConfig, } from "@/lib/service-config"; function deployedConfig( @@ -27,24 +26,7 @@ function deployedConfig( } describe("service config", () => { - it("uses deployed serverless settings as the runtime mode", () => { - const service = { - serverlessEnabled: false, - serverlessSleepAfterSeconds: 60, - serverlessWakeTimeoutSeconds: 60, - stateful: true, - deployedConfig: JSON.stringify(deployedConfig()), - }; - - expect(getDeployedServerlessConfig(service)).toMatchObject({ - enabled: true, - sleepAfterSeconds: 300, - wakeTimeoutSeconds: 120, - }); - expect(isDeployedServerlessService(service)).toBe(true); - }); - - it("enforces the minimum serverless sleep timeout for legacy config", () => { + it("enforces the minimum serverless sleep timeout", () => { expect( getCurrentServerlessConfig({ serverlessEnabled: true, @@ -54,33 +36,34 @@ describe("service config", () => { ).toMatchObject({ sleepAfterSeconds: MIN_SERVERLESS_SLEEP_AFTER_SECONDS, }); - expect( - getDeployedServerlessConfig({ - deployedConfig: JSON.stringify( - deployedConfig({ - serverless: { - enabled: true, - sleepAfterSeconds: 60, - wakeTimeoutSeconds: 120, - }, - }), - ), - }), - ).toMatchObject({ - sleepAfterSeconds: MIN_SERVERLESS_SLEEP_AFTER_SECONDS, - }); }); - it("allows deployed stateful services to be serverless", () => { - const service = { - serverlessEnabled: true, - stateful: true, - serverlessSleepAfterSeconds: 300, - serverlessWakeTimeoutSeconds: 120, - deployedConfig: JSON.stringify(deployedConfig({ stateful: true })), - }; + it("converts an immutable revision for pending-change comparisons", () => { + const config = revisionSpecToDeployedConfig( + { + schemaVersion: 1, + image: "nginx", + hostname: "api", + stateful: false, + serverless: { + enabled: false, + sleepAfterSeconds: 300, + wakeTimeoutSeconds: 300, + }, + healthCheck: null, + startCommand: null, + resourceLimits: { cpuCores: null, memoryMb: null }, + placements: [{ serverId: "server-1", count: 1 }], + ports: [], + secrets: [], + volumes: [], + }, + { "server-1": "Sydney" }, + ); - expect(isDeployedServerlessService(service)).toBe(true); + expect(config.replicas).toEqual([ + { serverId: "server-1", serverName: "Sydney", count: 1 }, + ]); }); it("reports serverless changes as pending config", () => { diff --git a/web/tests/service-revision-spec.test.ts b/web/tests/service-revision-spec.test.ts new file mode 100644 index 00000000..24bc3624 --- /dev/null +++ b/web/tests/service-revision-spec.test.ts @@ -0,0 +1,107 @@ +import { describe, expect, it } from "vitest"; +import { + buildServiceRevisionSpec, + type ServiceRevisionDraft, +} from "@/lib/service-revision-spec"; + +function draft( + overrides: Partial = {}, +): ServiceRevisionDraft { + return { + service: { + name: "API Service", + image: "nginx:latest", + hostname: "api.internal", + stateful: false, + serverlessEnabled: false, + serverlessSleepAfterSeconds: 300, + serverlessWakeTimeoutSeconds: 300, + healthCheckCmd: "curl -f http://localhost/health", + healthCheckInterval: 10, + healthCheckTimeout: 5, + healthCheckRetries: 3, + healthCheckStartPeriod: 30, + startCommand: null, + resourceCpuLimit: null, + resourceMemoryLimitMb: null, + }, + placements: [ + { serverId: "server-b", count: 1 }, + { serverId: "server-a", count: 2 }, + ], + ports: [ + { + port: 443, + isPublic: true, + domain: "api.example.com", + protocol: "tcp", + externalPort: 443, + tlsPassthrough: true, + }, + { + port: 80, + isPublic: false, + domain: null, + protocol: "http", + externalPort: null, + tlsPassthrough: false, + }, + ], + secrets: [ + { + key: "TOKEN", + encryptedValue: "ciphertext-2", + updatedAt: "2026-07-01T00:00:00.000Z", + }, + { + key: "API_KEY", + encryptedValue: "ciphertext-1", + updatedAt: "2026-07-01T00:00:00.000Z", + }, + ], + volumes: [ + { name: "logs", containerPath: "/logs" }, + { name: "data", containerPath: "/data" }, + ], + ...overrides, + }; +} + +describe("service revision specification", () => { + it("normalizes draft row ordering", () => { + const first = buildServiceRevisionSpec(draft()); + const reorderedDraft = draft(); + reorderedDraft.placements.reverse(); + reorderedDraft.ports.reverse(); + reorderedDraft.secrets.reverse(); + reorderedDraft.volumes.reverse(); + const second = buildServiceRevisionSpec(reorderedDraft); + + expect(second).toEqual(first); + }); + + it("normalizes defaults once", () => { + const input = draft(); + input.service.serverlessSleepAfterSeconds = 30; + input.service.healthCheckInterval = null; + input.ports[0].protocol = null; + input.ports[0].tlsPassthrough = null; + + const spec = buildServiceRevisionSpec(input); + + expect(spec.serverless.sleepAfterSeconds).toBe(120); + expect(spec.healthCheck?.interval).toBe(10); + expect(spec.ports[1]).toMatchObject({ + protocol: "http", + tlsPassthrough: false, + }); + }); + + it("rejects an invalid replica layout before it can be persisted", () => { + const input = draft({ placements: [] }); + + expect(() => buildServiceRevisionSpec(input)).toThrow( + "At least one replica is required", + ); + }); +});