diff --git a/.agents/setup b/.agents/setup index 81a7777d..59ad2500 100755 --- a/.agents/setup +++ b/.agents/setup @@ -3,6 +3,7 @@ set -euo pipefail repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" mise_bin="$HOME/.local/bin/mise" +export PATH="$HOME/.local/bin:$PATH" echo "Installing mise..." if [[ ! -x "$mise_bin" ]]; then diff --git a/AGENT.md b/AGENT.md index 001a0b94..e70f8f06 100644 --- a/AGENT.md +++ b/AGENT.md @@ -37,6 +37,11 @@ An open container deployment platform. See README.md for architecture. high-value critical behavior, serious regression risk, or contracts that would be costly to break. Keep tests focused; avoid low-signal harnesses. +## Communication + +- Keep responses concise and to the point. Avoid verbose responses unless + explicitly asked. + ## ⚠️ Critical restrictions - **NEVER run the Node application** (`next dev`, `next start`, `pnpm dev`), Go Agent or Go CLI diff --git a/agent/cmd/agent/main.go b/agent/cmd/agent/main.go index 51e58a6f..4605ad7c 100644 --- a/agent/cmd/agent/main.go +++ b/agent/cmd/agent/main.go @@ -282,7 +282,7 @@ func main() { } } - reconciler := reconcile.NewReconciler(config.EncryptionKey, dataDir) + reconciler := reconcile.NewReconciler(config.EncryptionKey, dataDir, config.RegistryInsecure) client := agenthttp.NewClient(controlPlaneURL, config.ServerID, signingKeyPair, dataDir) var logCollector *logs.Collector diff --git a/agent/internal/container/runtime.go b/agent/internal/container/runtime.go index f5d320f9..412fd36d 100644 --- a/agent/internal/container/runtime.go +++ b/agent/internal/container/runtime.go @@ -75,7 +75,7 @@ func Deploy(config *DeployConfig) (*DeployResult, error) { logFunc("stdout", fmt.Sprintf("Pulling image: %s", image)) - pullCmd := exec.Command("podman", "pull", "--tls-verify=false", image) + pullCmd := exec.Command("podman", buildPodmanPullArgs(config)...) pullOutput, err := pullCmd.CombinedOutput() if err != nil { logFunc("stderr", fmt.Sprintf("Pull failed: %s", string(pullOutput))) @@ -132,6 +132,14 @@ func Deploy(config *DeployConfig) (*DeployResult, error) { }, nil } +func buildPodmanPullArgs(config *DeployConfig) []string { + args := []string{"pull"} + if config.RegistryInsecure { + args = append(args, "--tls-verify=false") + } + return append(args, config.Image) +} + func buildPodmanRunArgs(config *DeployConfig, image string) []string { networkMAC := StableMACAddress(config.IPAddress) diff --git a/agent/internal/container/runtime_test.go b/agent/internal/container/runtime_test.go index 1fcf4b18..629febf4 100644 --- a/agent/internal/container/runtime_test.go +++ b/agent/internal/container/runtime_test.go @@ -5,6 +5,36 @@ import ( "testing" ) +func TestBuildPodmanPullArgs(t *testing.T) { + tests := []struct { + name string + insecure bool + want []string + }{ + { + name: "does not disable TLS by default", + want: []string{"pull", "registry.example.com/app:latest"}, + }, + { + name: "disables TLS verification when configured", + insecure: true, + want: []string{"pull", "--tls-verify=false", "registry.example.com/app:latest"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := buildPodmanPullArgs(&DeployConfig{ + Image: "registry.example.com/app:latest", + RegistryInsecure: tt.insecure, + }) + if !slices.Equal(got, tt.want) { + t.Fatalf("buildPodmanPullArgs() = %q, want %q", got, tt.want) + } + }) + } +} + func TestBuildPodmanRunArgsPublishesLoopbackPortsWithStaticIP(t *testing.T) { args := buildPodmanRunArgs(&DeployConfig{ Name: "svc-dep", diff --git a/agent/internal/container/types.go b/agent/internal/container/types.go index f0571589..ea1730f7 100644 --- a/agent/internal/container/types.go +++ b/agent/internal/container/types.go @@ -28,6 +28,7 @@ type BuildLogFunc func(stream string, message string) type DeployConfig struct { Name string Image string + RegistryInsecure bool ServiceID string ServiceName string DeploymentID string diff --git a/agent/internal/reconcile/reconcile.go b/agent/internal/reconcile/reconcile.go index 65635afc..1817ff5f 100644 --- a/agent/internal/reconcile/reconcile.go +++ b/agent/internal/reconcile/reconcile.go @@ -11,14 +11,16 @@ import ( ) type Reconciler struct { - encryptionKey string - dataDir string + encryptionKey string + dataDir string + registryInsecure bool } -func NewReconciler(encryptionKey, dataDir string) *Reconciler { +func NewReconciler(encryptionKey, dataDir string, registryInsecure bool) *Reconciler { return &Reconciler{ - encryptionKey: encryptionKey, - dataDir: dataDir, + encryptionKey: encryptionKey, + dataDir: dataDir, + registryInsecure: registryInsecure, } } @@ -66,6 +68,7 @@ func (r *Reconciler) Deploy(exp agenthttp.ExpectedContainer) error { _, err := container.Deploy(&container.DeployConfig{ Name: exp.Name, Image: exp.Image, + RegistryInsecure: r.registryInsecure, ServiceID: exp.ServiceID, ServiceName: exp.ServiceName, DeploymentID: exp.DeploymentID, diff --git a/cli/internal/manifest/manifest.go b/cli/internal/manifest/manifest.go index eb680d2a..876a6238 100644 --- a/cli/internal/manifest/manifest.go +++ b/cli/internal/manifest/manifest.go @@ -220,8 +220,8 @@ func Validate(m Manifest) error { if m.Service.StartCommand != nil && *m.Service.StartCommand == "" { return errors.New("service.startCommand cannot be blank") } - if m.Service.Replicas < 1 || m.Service.Replicas > 10 { - return errors.New("service.replicas must be between 1 and 10") + if m.Service.Replicas < 1 || m.Service.Replicas > 32 { + return errors.New("service.replicas must be between 1 and 32") } if m.Service.Placement == nil { return errors.New("service.placement is required") @@ -249,8 +249,8 @@ func Validate(m Manifest) error { } total += server.Count } - if total < 1 || total > 10 { - return errors.New("service.placement manual total must be between 1 and 10") + if total < 1 || total > 32 { + return errors.New("service.placement manual total must be between 1 and 32") } if total != m.Service.Replicas { return errors.New("service.placement manual total must equal service.replicas") diff --git a/cli/internal/manifest/manifest_test.go b/cli/internal/manifest/manifest_test.go index 57cdcbb7..b7c5103a 100644 --- a/cli/internal/manifest/manifest_test.go +++ b/cli/internal/manifest/manifest_test.go @@ -66,7 +66,7 @@ func TestPlacementRoundTripAndValidation(t *testing.T) { {"blank server", &Placement{Mode: "manual", Servers: []PlacementServer{{ServerID: " ", Count: 1}}}, 1}, {"duplicate server", &Placement{Mode: "manual", Servers: []PlacementServer{{ServerID: "a", Count: 1}, {ServerID: "a", Count: 1}}}, 2}, {"nonpositive count", &Placement{Mode: "manual", Servers: []PlacementServer{{ServerID: "a", Count: 0}}}, 1}, - {"total exceeds limit", &Placement{Mode: "manual", Servers: []PlacementServer{{ServerID: "a", Count: 11}}}, 10}, + {"total exceeds limit", &Placement{Mode: "manual", Servers: []PlacementServer{{ServerID: "a", Count: 33}}}, 32}, {"total differs from replicas", &Placement{Mode: "manual", Servers: []PlacementServer{{ServerID: "a", Count: 1}}}, 2}, } for _, tc := range tests { @@ -82,6 +82,26 @@ func TestPlacementRoundTripAndValidation(t *testing.T) { } } +func TestReplicaLimit(t *testing.T) { + for _, placement := range []*Placement{ + {Mode: "automatic"}, + {Mode: "manual", Servers: []PlacementServer{{ServerID: "a", Count: 32}}}, + } { + m := base() + m.Service.Replicas = 32 + m.Service.Placement = placement + if err := Validate(m); err != nil { + t.Fatalf("32 replicas rejected: %v", err) + } + } + + m := base() + m.Service.Replicas = 33 + if err := Validate(m); err == nil { + t.Fatal("33 replicas accepted") + } +} + func TestPlacementIsRequired(t *testing.T) { _, err := Parse([]byte(`apiVersion: v1 service: diff --git a/docs/services/scaling.mdx b/docs/services/scaling.mdx index 13ab4163..1ed8c62b 100644 --- a/docs/services/scaling.mdx +++ b/docs/services/scaling.mdx @@ -7,7 +7,7 @@ description: "Replicas, placement, and server pinning." Each service can run multiple replicas across your cluster. Configure how many replicas run on each server from the service settings. -Replica count ranges from 1 to 10 per service. +Replica count ranges from 1 to 32 per service. ## Serverless scaling @@ -59,7 +59,7 @@ You can also manually lock any service to a specific server by setting the locke - Stateful services are limited to 1 replica. - Stateful services are always pinned to their locked server. - Stateful services do not automatically fail over to another server. -- Maximum 10 replicas per service. +- Maximum 32 replicas per service. - Serverless scaling requires a public HTTP service domain. - Sleep and wake are proxy-local; serverless replicas must be placed on proxy nodes. - Serverless traffic must be routed only to proxy nodes that own a local proxy replica for that service. diff --git a/web/actions/projects.ts b/web/actions/projects.ts index 27d3f370..204f73df 100644 --- a/web/actions/projects.ts +++ b/web/actions/projects.ts @@ -1067,7 +1067,7 @@ export type ServiceConfigUpdate = { const placementInputSchema = z.discriminatedUnion("mode", [ z.strictObject({ mode: z.literal("automatic"), - replicas: z.number().int().min(1).max(10), + replicas: z.number().int().min(1).max(32), }), z .strictObject({ @@ -1076,7 +1076,7 @@ const placementInputSchema = z.discriminatedUnion("mode", [ .array( z.strictObject({ serverId: z.string().min(1), - count: z.number().int().min(1).max(10), + count: z.number().int().min(1).max(32), }), ) .min(1), @@ -1092,10 +1092,10 @@ const placementInputSchema = z.discriminatedUnion("mode", [ path: ["placements"], }); const total = value.placements.reduce((sum, item) => sum + item.count, 0); - if (total > 10) + if (total > 32) context.addIssue({ code: "custom", - message: "Total replicas must be between 1 and 10", + message: "Total replicas must be between 1 and 32", path: ["placements"], }); }), diff --git a/web/app/api/projects/[id]/services/route.ts b/web/app/api/projects/[id]/services/route.ts index 1137888d..e695a620 100644 --- a/web/app/api/projects/[id]/services/route.ts +++ b/web/app/api/projects/[id]/services/route.ts @@ -63,9 +63,11 @@ export async function GET( replicas, serviceSecrets, serviceRollouts, + activeRollout, volumes, lockedServer, latestBuild, + activeBuild, githubRepo, ] = await Promise.all([ db @@ -99,6 +101,18 @@ export async function GET( .where(eq(rollouts.serviceId, service.id)) .orderBy(desc(rollouts.createdAt)) .limit(1), + db + .select() + .from(rollouts) + .where( + and( + eq(rollouts.serviceId, service.id), + inArray(rollouts.status, ["queued", "in_progress"]), + ), + ) + .orderBy(desc(rollouts.createdAt)) + .limit(1) + .then((r) => r[0] || null), db .select() .from(serviceVolumes) @@ -119,6 +133,24 @@ export async function GET( .limit(1) .then((r) => r[0] || null) : Promise.resolve(null), + db + .select({ id: builds.id, status: builds.status }) + .from(builds) + .where( + and( + eq(builds.serviceId, service.id), + inArray(builds.status, [ + "pending", + "claimed", + "cloning", + "building", + "pushing", + ]), + ), + ) + .orderBy(desc(builds.createdAt)) + .limit(1) + .then((r) => r[0] || null), service.sourceType === "github" ? db .select() @@ -262,9 +294,11 @@ export async function GET( deployments: deploymentsWithDetails, secrets: serviceSecrets, rollouts: serviceRollouts, + activeRollout, volumes, lockedServer, latestBuild, + activeBuild, hasGithubAppRepo: githubRepo !== null, activeConfig, currentSource, diff --git a/web/app/api/v1/agent/register/route.ts b/web/app/api/v1/agent/register/route.ts index 7801be88..a06bc793 100644 --- a/web/app/api/v1/agent/register/route.ts +++ b/web/app/api/v1/agent/register/route.ts @@ -6,7 +6,11 @@ import { HOUR_IN_MILLISECONDS, subtractMilliseconds } from "@/lib/date"; import { EncryptionKeyUnavailableError, resolveEncryptionKey } from "@/lib/kms"; import { agentRegisterSchema } from "@/lib/schemas"; import { formatZodErrors } from "@/lib/utils"; -import { assignSubnet } from "@/lib/wireguard"; +import { + assignSubnet, + SUBNET_ALLOCATION_CONSTRAINTS, + withAllocationRetry, +} from "@/lib/wireguard"; const TOKEN_EXPIRY_HOURS = 24; @@ -60,38 +64,46 @@ export async function POST(request: NextRequest) { const encryptionKeyBuffer = await resolveEncryptionKey(); const encryptionKey = encryptionKeyBuffer.toString("hex"); - const { subnetId, wireguardIp } = await assignSubnet(); + const allocation = await withAllocationRetry( + () => + db.transaction(async (tx) => { + const { subnetId, wireguardIp } = await assignSubnet(tx); + const claimedServers = await tx + .update(servers) + .set({ + wireguardPublicKey, + signingPublicKey, + subnetId, + wireguardIp, + publicIp: publicIp || null, + privateIp: privateIp || null, + isProxy: isProxy === true, + tokenUsedAt: now, + status: "online", + lastHeartbeat: now, + }) + .where( + and( + eq(servers.id, server.id), + eq(servers.agentToken, token), + isNull(servers.tokenUsedAt), + gt(servers.tokenCreatedAt, expiryThreshold), + ), + ) + .returning({ id: servers.id }); - const claimedServers = await db - .update(servers) - .set({ - wireguardPublicKey, - signingPublicKey, - subnetId, - wireguardIp, - publicIp: publicIp || null, - privateIp: privateIp || null, - isProxy: isProxy === true, - tokenUsedAt: now, - status: "online", - lastHeartbeat: now, - }) - .where( - and( - eq(servers.id, server.id), - eq(servers.agentToken, token), - isNull(servers.tokenUsedAt), - gt(servers.tokenCreatedAt, expiryThreshold), - ), - ) - .returning({ id: servers.id }); + return claimedServers.length > 0 ? { subnetId, wireguardIp } : null; + }), + SUBNET_ALLOCATION_CONSTRAINTS, + ); - if (claimedServers.length === 0) { + if (!allocation) { return NextResponse.json( { error: "Invalid, expired, or already used token" }, { status: 401 }, ); } + const { subnetId, wireguardIp } = allocation; return NextResponse.json({ serverId: server.id, @@ -103,7 +115,7 @@ export async function POST(request: NextRequest) { registryUrl: process.env.REGISTRY_URL ?? null, registryUsername: process.env.REGISTRY_USERNAME ?? null, registryPassword: process.env.REGISTRY_PASSWORD ?? null, - registryInsecure: process.env.REGISTRY_INSECURE !== "false", + registryInsecure: process.env.REGISTRY_INSECURE === "true", }); } catch (error) { console.error("Agent registration error:", error); diff --git a/web/components/core/summary-card.tsx b/web/components/core/summary-card.tsx index 6d0990b1..104063cd 100644 --- a/web/components/core/summary-card.tsx +++ b/web/components/core/summary-card.tsx @@ -55,7 +55,7 @@ export function SummaryCardStat({ {label} - + {children} ); diff --git a/web/components/service/details/deployment-progress.tsx b/web/components/service/details/deployment-progress.tsx index 239e5787..faebfb71 100644 --- a/web/components/service/details/deployment-progress.tsx +++ b/web/components/service/details/deployment-progress.tsx @@ -94,23 +94,28 @@ export function getBarState( }; } - if ( + const activeBuild = + service.activeBuild === undefined && service.latestBuild && ACTIVE_BUILD_STATUSES.includes(service.latestBuild.status) - ) { + ? service.latestBuild + : service.activeBuild; + + if (activeBuild) { return { mode: "building", - buildId: service.latestBuild.id, - buildStatus: service.latestBuild.status, + buildId: activeBuild.id, + buildStatus: activeBuild.status, }; } const latestRollout = service.rollouts?.[0]; const activeRollout = - latestRollout?.status === "queued" || - latestRollout?.status === "in_progress" + service.activeRollout === undefined && + (latestRollout?.status === "queued" || + latestRollout?.status === "in_progress") ? latestRollout - : undefined; + : service.activeRollout; if (activeRollout) { const currentStage = diff --git a/web/components/service/details/replicas-section.tsx b/web/components/service/details/replicas-section.tsx index 80cec112..be1632c8 100644 --- a/web/components/service/details/replicas-section.tsx +++ b/web/components/service/details/replicas-section.tsx @@ -13,6 +13,7 @@ import { EmptyTitle, } from "@/components/ui/empty"; import { Input } from "@/components/ui/input"; +import { Slider } from "@/components/ui/slider"; import { Spinner } from "@/components/ui/spinner"; import { Tabs, TabsList, TabsTrigger } from "@/components/ui/tabs"; import type { @@ -198,7 +199,7 @@ export const ReplicasSection = memo(function ReplicasSection({ setIsEditing(true); setLocalReplicas((prev) => ({ ...prev, - [serverId]: Math.max(0, Math.min(10, Math.floor(value))), + [serverId]: Math.max(0, Math.min(32, Math.floor(value))), })); }, []); @@ -211,7 +212,7 @@ export const ReplicasSection = memo(function ReplicasSection({ (sum, count) => sum + count, 0, ); - setDesiredReplicas(Math.max(1, Math.min(10, manualTotal || 1))); + setDesiredReplicas(Math.max(1, Math.min(32, manualTotal || 1))); } setPlacementMode(nextMode); }; @@ -268,7 +269,7 @@ export const ReplicasSection = memo(function ReplicasSection({ : null; }; - const manualTotalIsValid = totalReplicas >= 1 && totalReplicas <= 10; + const manualTotalIsValid = totalReplicas >= 1 && totalReplicas <= 32; if (service.stateful) { return ( @@ -401,41 +402,36 @@ export const ReplicasSection = memo(function ReplicasSection({ {placementMode === "automatic" ? (
-
- -

- The control plane distributes replicas evenly across healthy - {service.serverlessEnabled ? " proxy nodes" : " nodes"} and - moves them after failures. -

+
+
+

Desired replicas

+

+ The control plane distributes replicas evenly across healthy + {service.serverlessEnabled ? " proxy nodes" : " nodes"} and + moves them after failures. +

+
+ + {desiredReplicas} + +
+
+ { + setIsEditing(true); + setDesiredReplicas(value); + }} + /> +
+ 1 + 32 +
- { - setIsEditing(true); - setDesiredReplicas( - Math.max( - 1, - Math.min(10, Math.floor(event.target.valueAsNumber || 1)), - ), - ); - }} - className="w-24" - aria-describedby="automatic-replica-range" - /> -

- Choose between 1 and 10 replicas. -

{hasChanges ? (
{!manualTotalIsValid && (

- Manual placement requires 1 to 10 replicas in total. + Manual placement requires 1 to 32 replicas in total.

)} {hasChanges && ( diff --git a/web/components/service/details/service-details-overview.tsx b/web/components/service/details/service-details-overview.tsx index 8d01f750..06ef8bc1 100644 --- a/web/components/service/details/service-details-overview.tsx +++ b/web/components/service/details/service-details-overview.tsx @@ -752,18 +752,24 @@ function getServiceStatus( ): ServiceStatus { const latestRollout = service.rollouts?.[0]; const deployments = service.deployments || []; - - if (service.migrationStatus) return { label: "Migrating", tone: "progress" }; - if ( + const activeBuild = + service.activeBuild === undefined && service.latestBuild && ACTIVE_BUILD_STATUSES.has(service.latestBuild.status) - ) { + ? service.latestBuild + : service.activeBuild; + const activeRollout = + service.activeRollout === undefined && + (latestRollout?.status === "queued" || + latestRollout?.status === "in_progress") + ? latestRollout + : service.activeRollout; + + if (service.migrationStatus) return { label: "Migrating", tone: "progress" }; + if (activeBuild) { return { label: "Building", tone: "progress" }; } - if ( - latestRollout?.status === "queued" || - latestRollout?.status === "in_progress" - ) { + if (activeRollout) { return { label: "Deploying", tone: "progress" }; } if (runningDeployments > 0) return { label: "Live", tone: "live" }; diff --git a/web/components/service/service-layout-client.tsx b/web/components/service/service-layout-client.tsx index 07cecb9d..c0237fd4 100644 --- a/web/components/service/service-layout-client.tsx +++ b/web/components/service/service-layout-client.tsx @@ -57,11 +57,20 @@ export function ServiceLayoutClient({ onSuccess: (data) => { const svc = data?.find((s) => s.id === serviceId); if (!svc) return; + const hasActiveBuild = + svc.activeBuild === undefined + ? svc.latestBuild != null && + ACTIVE_BUILD_STATUSES.includes(svc.latestBuild.status) + : svc.activeBuild !== null; + const latestRollout = svc.rollouts?.[0]; + const hasActiveRollout = + svc.activeRollout === undefined + ? latestRollout?.status === "queued" || + latestRollout?.status === "in_progress" + : svc.activeRollout !== null; const isActive = - (svc.latestBuild != null && - ACTIVE_BUILD_STATUSES.includes(svc.latestBuild.status)) || - svc.rollouts?.[0]?.status === "queued" || - svc.rollouts?.[0]?.status === "in_progress" || + hasActiveBuild || + hasActiveRollout || !!svc.migrationStatus || svc.deployments.some((d) => IN_PROGRESS_DEPLOY_STATUSES.includes(d.observedPhase), diff --git a/web/components/ui/slider.tsx b/web/components/ui/slider.tsx new file mode 100644 index 00000000..ead1fcec --- /dev/null +++ b/web/components/ui/slider.tsx @@ -0,0 +1,60 @@ +import { Slider as SliderPrimitive } from "@base-ui/react/slider"; + +import { cn } from "@/lib/utils"; + +type SliderProps = + SliderPrimitive.Root.Props & { + getThumbAriaLabel?: (index: number) => string; + }; + +function Slider({ + className, + defaultValue, + value, + min = 0, + max = 100, + "aria-label": ariaLabel, + getThumbAriaLabel, + ...props +}: SliderProps) { + const effectiveValue = value ?? defaultValue; + const thumbCount = Array.isArray(effectiveValue) ? effectiveValue.length : 1; + + return ( + + + + + + {Array.from({ length: thumbCount }, (_, index) => ( + + ))} + + + ); +} + +export { Slider }; diff --git a/web/db/schema.ts b/web/db/schema.ts index f7e4ab86..ca7122ea 100644 --- a/web/db/schema.ts +++ b/web/db/schema.ts @@ -321,8 +321,8 @@ export const servers = pgTable("servers", { name: text("name").notNull(), publicIp: text("public_ip"), privateIp: text("private_ip"), - subnetId: integer("subnet_id"), - wireguardIp: text("wireguard_ip"), + subnetId: integer("subnet_id").unique("servers_subnet_id_unique"), + wireguardIp: text("wireguard_ip").unique("servers_wireguard_ip_unique"), wireguardPublicKey: text("wireguard_public_key"), signingPublicKey: text("signing_public_key"), isProxy: boolean("is_proxy").default(false).notNull(), @@ -692,6 +692,9 @@ export const deployments = pgTable( 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), + uniqueIndex("deployments_server_id_ip_address_unique_idx") + .on(table.serverId, table.ipAddress) + .where(sql`${table.ipAddress} is not null`), foreignKey({ name: "deployments_service_revision_service_fk", columns: [table.serviceRevisionId, table.serviceId], diff --git a/web/db/types.ts b/web/db/types.ts index 02bb6ca3..4860a6f8 100644 --- a/web/db/types.ts +++ b/web/db/types.ts @@ -65,8 +65,10 @@ export type ServiceWithDetails = Service & { volumes?: ServiceVolume[]; secrets?: Array & { updatedAt: Date | string }>; rollouts?: Rollout[]; + activeRollout?: Rollout | null; lockedServer?: Pick | null; latestBuild?: Pick | null; + activeBuild?: Pick | null; hasGithubAppRepo?: boolean; deletionBackupFallback?: { volumeCount: number; diff --git a/web/lib/compose-parser.ts b/web/lib/compose-parser.ts index 5adc7345..2662baa2 100644 --- a/web/lib/compose-parser.ts +++ b/web/lib/compose-parser.ts @@ -557,7 +557,7 @@ export function parseComposeYaml(yamlContent: string): ComposeParseResult { let replicas = serviceConfig.deploy?.replicas ?? 1; if (replicas < 1) replicas = 1; - if (replicas > 10) replicas = 10; + if (replicas > 32) replicas = 32; if (stateful && replicas > 1) { warnings.push({ service: serviceName, diff --git a/web/lib/inngest/functions/rollout-helpers.ts b/web/lib/inngest/functions/rollout-helpers.ts index 79d30979..60322d35 100644 --- a/web/lib/inngest/functions/rollout-helpers.ts +++ b/web/lib/inngest/functions/rollout-helpers.ts @@ -13,12 +13,18 @@ import { getPublishedContainerPorts, type ServiceRevisionSpec, } from "@/lib/service-revision-spec"; -import { assignContainerIp } from "@/lib/wireguard"; +import { + assignContainerIp, + CONTAINER_IP_ALLOCATION_CONSTRAINTS, + withAllocationRetry, +} from "@/lib/wireguard"; import { enqueueWork } from "@/lib/work-queue"; const PORT_RANGE_START = 30000; const PORT_RANGE_END = 32767; +type RolloutTransaction = Parameters[0]>[0]; + export type Placement = { serverId: string; replicas: number }; export function automaticPlacementIneligibilityReason( @@ -41,8 +47,8 @@ export function distributeReplicas( ): Placement[] { const ids = [...new Set(serverIds)].sort((a, b) => a.localeCompare(b)); if (ids.length === 0) throw new Error("No eligible servers for deployment"); - if (!Number.isInteger(replicas) || replicas < 1 || replicas > 10) - throw new Error("Replica count must be between 1 and 10"); + if (!Number.isInteger(replicas) || replicas < 1 || replicas > 32) + throw new Error("Replica count must be between 1 and 32"); const counts = new Map(ids.map((id) => [id, 0])); for (let index = 0; index < replicas; index++) { const id = ids[index % ids.length]; @@ -65,8 +71,11 @@ export type DeploymentContext = { isRollingUpdate: boolean; }; -async function getUsedPorts(serverId: string): Promise> { - const existingPorts = await db +async function getUsedPorts( + tx: RolloutTransaction, + serverId: string, +): Promise> { + const existingPorts = await tx .select({ hostPort: deploymentPorts.hostPort }) .from(deploymentPorts) .innerJoin(deployments, eq(deploymentPorts.deploymentId, deployments.id)) @@ -76,10 +85,11 @@ async function getUsedPorts(serverId: string): Promise> { } export async function allocateHostPorts( + tx: RolloutTransaction, serverId: string, count: number, ): Promise { - const unavailablePorts = await getUsedPorts(serverId); + const unavailablePorts = await getUsedPorts(tx, serverId); const allocated: number[] = []; for ( @@ -114,8 +124,8 @@ export function calculateRevisionPlacements( if (totalReplicas < 1) { throw new Error("At least one replica is required"); } - if (totalReplicas > 10) { - throw new Error("Maximum 10 replicas allowed"); + if (totalReplicas > 32) { + throw new Error("Maximum 32 replicas allowed"); } if (specification.stateful) { @@ -356,38 +366,10 @@ export async function createDeploymentRecords( ): Promise<{ deploymentIds: string[] }> { const { revisionId, specification, placements, serverMap } = context; - const existingDeployments = await db - .select({ - id: deployments.id, - serviceId: deployments.serviceId, - serviceRevisionId: deployments.serviceRevisionId, - serverId: deployments.serverId, - }) - .from(deployments) - .where(eq(deployments.rolloutId, rolloutId)); const requestedReplicasByServer = new Map( placements.map((placement) => [placement.serverId, placement.replicas]), ); - const existingDeploymentsByServer = Map.groupBy( - existingDeployments, - (deployment) => deployment.serverId, - ); - for (const deployment of existingDeployments) { - if ( - deployment.serviceId !== serviceId || - deployment.serviceRevisionId !== revisionId || - !requestedReplicasByServer.has(deployment.serverId) - ) { - throw new Error("Rollout deployment idempotency conflict"); - } - } - for (const [serverId, existing] of existingDeploymentsByServer) { - if (existing.length > (requestedReplicasByServer.get(serverId) ?? 0)) { - throw new Error("Rollout deployment idempotency conflict"); - } - } - - const deploymentIds = existingDeployments.map((deployment) => deployment.id); + const deploymentIds = new Set(); const publishedContainerPorts = getPublishedContainerPorts( specification.ports, ); @@ -400,68 +382,112 @@ export async function createDeploymentRecords( throw new Error(`Server ${placement.serverId} not found`); } - const existingReplicaCount = - existingDeploymentsByServer.get(placement.serverId)?.length ?? 0; - for (let i = existingReplicaCount; i < placement.replicas; i++) { + for (let i = 0; i < placement.replicas; i++) { const deploymentId = randomUUID(); - const hostPorts = await allocateHostPorts( - server.id, - publishedContainerPorts.length, - ); - const ipAddress = await assignContainerIp(server.id); - await db.transaction(async (tx) => { - await tx.execute( - sql`SELECT pg_advisory_xact_lock(hashtext(${serviceId}))`, - ); - const [rollout] = await tx - .select({ status: rollouts.status }) - .from(rollouts) - .where(eq(rollouts.id, rolloutId)) - .for("update"); - if (rollout?.status !== "in_progress") { - throw new Error("Rollout is no longer in progress"); - } - - await tx.insert(deployments).values({ - id: deploymentId, - serviceId, - serviceRevisionId: revisionId, - serverId: server.id, - ipAddress, - runtimeDesiredState: "running", - trafficState: "candidate", - observedPhase: "pending", - rolloutId, - }); - - if (publishedContainerPorts.length > 0) { - await tx.insert(deploymentPorts).values( - publishedContainerPorts.map((containerPort, index) => ({ - id: randomUUID(), + const currentDeploymentIds = await withAllocationRetry( + () => + db.transaction(async (tx) => { + await tx.execute( + sql`SELECT pg_advisory_xact_lock(hashtext(${serviceId}))`, + ); + const [rollout] = await tx + .select({ status: rollouts.status }) + .from(rollouts) + .where(eq(rollouts.id, rolloutId)) + .for("update"); + if (rollout?.status !== "in_progress") { + throw new Error("Rollout is no longer in progress"); + } + + const existingDeployments = await tx + .select({ + id: deployments.id, + serviceId: deployments.serviceId, + serviceRevisionId: deployments.serviceRevisionId, + serverId: deployments.serverId, + }) + .from(deployments) + .where(eq(deployments.rolloutId, rolloutId)); + const existingDeploymentsByServer = Map.groupBy( + existingDeployments, + (deployment) => deployment.serverId, + ); + for (const deployment of existingDeployments) { + if ( + deployment.serviceId !== serviceId || + deployment.serviceRevisionId !== revisionId || + !requestedReplicasByServer.has(deployment.serverId) + ) { + throw new Error("Rollout deployment idempotency conflict"); + } + } + for (const [serverId, existing] of existingDeploymentsByServer) { + if ( + existing.length > (requestedReplicasByServer.get(serverId) ?? 0) + ) { + throw new Error("Rollout deployment idempotency conflict"); + } + } + if ( + (existingDeploymentsByServer.get(server.id)?.length ?? 0) >= + placement.replicas + ) { + return existingDeployments.map((deployment) => deployment.id); + } + + const ipAddress = await assignContainerIp(tx, server.id); + const hostPorts = await allocateHostPorts( + tx, + server.id, + publishedContainerPorts.length, + ); + await tx.insert(deployments).values({ + id: deploymentId, + serviceId, + serviceRevisionId: revisionId, + serverId: server.id, + ipAddress, + runtimeDesiredState: "running", + trafficState: "candidate", + observedPhase: "pending", + rolloutId, + }); + + if (publishedContainerPorts.length > 0) { + await tx.insert(deploymentPorts).values( + publishedContainerPorts.map((containerPort, index) => ({ + id: randomUUID(), + deploymentId, + containerPort, + hostPort: hostPorts[index], + })), + ); + } + + await enqueueWork( + server.id, + "reconcile", + { + reason: "rollout_deployment_created", + deploymentId, + }, + { tx }, + ); + + return [ + ...existingDeployments.map((deployment) => deployment.id), deploymentId, - containerPort, - hostPort: hostPorts[index], - })), - ); - } - - await enqueueWork( - server.id, - "reconcile", - { - reason: "rollout_deployment_created", - deploymentId, - }, - { tx }, - ); - }); + ]; + }), + CONTAINER_IP_ALLOCATION_CONSTRAINTS, + ); - deploymentIds.push(deploymentId); + for (const id of currentDeploymentIds) deploymentIds.add(id); } } - return { deploymentIds }; + return { deploymentIds: [...deploymentIds] }; } export async function completeRollout( diff --git a/web/lib/inngest/functions/rollout-workflow.ts b/web/lib/inngest/functions/rollout-workflow.ts index 7316bed3..3b723990 100644 --- a/web/lib/inngest/functions/rollout-workflow.ts +++ b/web/lib/inngest/functions/rollout-workflow.ts @@ -24,7 +24,7 @@ import { handleRolloutFailure } from "./rollout-utils"; const PREFLIGHT_FAILURE_MESSAGES = [ "At least one replica is required", - "Maximum 10 replicas allowed", + "Maximum 32 replicas allowed", "No servers selected for deployment", "Stateful services can only have exactly 1 replica", "Stateful services must be deployed to exactly one server", diff --git a/web/lib/public-api.ts b/web/lib/public-api.ts index 6bc13d87..75fbfae6 100644 --- a/web/lib/public-api.ts +++ b/web/lib/public-api.ts @@ -554,7 +554,7 @@ const hostnameSchema = z export const placementSchema = z.discriminatedUnion("mode", [ z.strictObject({ mode: z.literal("automatic"), - replicas: z.number().int().min(1).max(10), + replicas: z.number().int().min(1).max(32), }), z .strictObject({ @@ -563,7 +563,7 @@ export const placementSchema = z.discriminatedUnion("mode", [ .array( z.strictObject({ serverId: z.string().min(1), - count: z.number().int().min(1).max(10), + count: z.number().int().min(1).max(32), }), ) .min(1), @@ -578,10 +578,10 @@ export const placementSchema = z.discriminatedUnion("mode", [ message: "Server IDs must be unique", path: ["placements"], }); - if (value.placements.reduce((sum, item) => sum + item.count, 0) > 10) + if (value.placements.reduce((sum, item) => sum + item.count, 0) > 32) context.addIssue({ code: "custom", - message: "Total replicas must be between 1 and 10", + message: "Total replicas must be between 1 and 32", path: ["placements"], }); }), diff --git a/web/lib/service-revision-changes.ts b/web/lib/service-revision-changes.ts index cdb0448c..cdfa0b31 100644 --- a/web/lib/service-revision-changes.ts +++ b/web/lib/service-revision-changes.ts @@ -85,7 +85,7 @@ const serviceRevisionSpecSchema = z z.strictObject({ mode: z.literal("manual") }), z.strictObject({ mode: z.literal("automatic"), - replicas: z.number().int().min(1).max(10), + replicas: z.number().int().min(1).max(32), }), ]), ...serviceRevisionSpecFields, diff --git a/web/lib/service-revision-spec.ts b/web/lib/service-revision-spec.ts index 0bad7661..315784b1 100644 --- a/web/lib/service-revision-spec.ts +++ b/web/lib/service-revision-spec.ts @@ -238,8 +238,8 @@ function validateServiceRevisionSpec( if (totalReplicas < 1 && !allowNoPlacements) { throw new Error("At least one replica is required"); } - if (totalReplicas > 10) { - throw new Error("Maximum 10 replicas allowed"); + if (totalReplicas > 32) { + throw new Error("Maximum 32 replicas allowed"); } if ( specification.placement.mode === "automatic" && diff --git a/web/lib/wireguard.ts b/web/lib/wireguard.ts index cb564a71..458a0c04 100644 --- a/web/lib/wireguard.ts +++ b/web/lib/wireguard.ts @@ -1,8 +1,22 @@ -import { db } from "@/db"; -import { servers, deployments } from "@/db/schema"; -import { eq, isNotNull, and, ne } from "drizzle-orm"; -import { WIREGUARD_SUBNET_PREFIX, CONTAINER_SUBNET_PREFIX } from "./constants"; +import { and, eq, isNotNull, ne, sql } from "drizzle-orm"; +import { DrizzleQueryError } from "drizzle-orm/errors"; import { Address4 } from "ip-address"; +import { db } from "@/db"; +import { deployments, servers } from "@/db/schema"; +import { CONTAINER_SUBNET_PREFIX, WIREGUARD_SUBNET_PREFIX } from "./constants"; + +type AllocationTransaction = Parameters< + Parameters[0] +>[0]; + +export const SUBNET_ALLOCATION_CONSTRAINTS = new Set([ + "servers_subnet_id_unique", + "servers_wireguard_ip_unique", +]); +export const CONTAINER_IP_ALLOCATION_CONSTRAINTS = new Set([ + "deployments_server_id_ip_address_unique_idx", +]); +const ALLOCATION_RETRY_LIMIT = 3; function sameSubnet(ip1: string, ip2: string, prefix: number = 16): boolean { if (!ip1 || !ip2) return false; @@ -15,11 +29,38 @@ function sameSubnet(ip1: string, ip2: string, prefix: number = 16): boolean { } } -export async function assignSubnet(): Promise<{ +export async function withAllocationRetry( + operation: () => Promise, + constraints: ReadonlySet, +): Promise { + for (let attempt = 1; ; attempt++) { + try { + return await operation(); + } catch (error) { + const databaseError = ( + error instanceof DrizzleQueryError ? error.cause : error + ) as { code?: string; constraint?: string }; + if ( + databaseError.code !== "23505" || + !databaseError.constraint || + !constraints.has(databaseError.constraint) || + attempt >= ALLOCATION_RETRY_LIMIT + ) { + throw error; + } + } + } +} + +export async function assignSubnet(tx: AllocationTransaction): Promise<{ subnetId: number; wireguardIp: string; }> { - const existingServers = await db + await tx.execute( + sql`SELECT pg_advisory_xact_lock(hashtext('subnet_allocation'))`, + ); + + const existingServers = await tx .select({ subnetId: servers.subnetId }) .from(servers) .where(isNotNull(servers.subnetId)); @@ -36,8 +77,15 @@ export async function assignSubnet(): Promise<{ throw new Error("No available subnets"); } -export async function assignContainerIp(serverId: string): Promise { - const server = await db +export async function assignContainerIp( + tx: AllocationTransaction, + serverId: string, +): Promise { + await tx.execute( + sql`SELECT pg_advisory_xact_lock(hashtext('container_ip_allocation'), hashtext(${serverId}))`, + ); + + const server = await tx .select({ subnetId: servers.subnetId }) .from(servers) .where(eq(servers.id, serverId)) @@ -47,7 +95,7 @@ export async function assignContainerIp(serverId: string): Promise { throw new Error("Server does not have a subnet assigned"); } - const existingDeployments = await db + const existingDeployments = await tx .select({ ipAddress: deployments.ipAddress }) .from(deployments) .where( diff --git a/web/tests/autoplacement.test.ts b/web/tests/autoplacement.test.ts index eae08e90..8c46c676 100644 --- a/web/tests/autoplacement.test.ts +++ b/web/tests/autoplacement.test.ts @@ -41,12 +41,18 @@ describe("automatic placement distribution", () => { ).toEqual(counts); }); - it("stacks ten replicas deterministically on two servers", () => { - expect(distributeReplicas(["b", "a"], 10)).toEqual([ - { serverId: "a", replicas: 5 }, - { serverId: "b", replicas: 5 }, + it("stacks 32 replicas deterministically on two servers", () => { + expect(distributeReplicas(["b", "a"], 32)).toEqual([ + { serverId: "a", replicas: 16 }, + { serverId: "b", replicas: 16 }, ]); }); + + it("rejects more than 32 replicas", () => { + expect(() => distributeReplicas(["a"], 33)).toThrow( + "Replica count must be between 1 and 32", + ); + }); }); describe("automatic placement eligibility diagnostics", () => { diff --git a/web/tests/deployment-progress.test.ts b/web/tests/deployment-progress.test.ts new file mode 100644 index 00000000..d5a60c58 --- /dev/null +++ b/web/tests/deployment-progress.test.ts @@ -0,0 +1,125 @@ +import { describe, expect, it } from "vitest"; +import { getBarState } from "@/components/service/details/deployment-progress"; +import type { ServiceWithDetails } from "@/db/types"; +import type { ConfigChange } from "@/lib/service-config"; + +const pendingChanges: ConfigChange[] = [ + { field: "Image", from: "old", to: "new" }, +]; + +function service( + overrides: Partial = {}, +): ServiceWithDetails { + return { + migrationStatus: null, + latestBuild: null, + activeBuild: null, + rollouts: [], + activeRollout: null, + deployments: [], + configuredReplicas: [{ count: 1 }], + replicas: 1, + placementMode: "manual", + ...overrides, + } as ServiceWithDetails; +} + +const completedRollout = { + id: "rollout-completed", + status: "completed" as const, +}; + +const activeRollout = { + id: "rollout-active", + status: "in_progress" as const, + currentStage: "health_check", +}; + +describe("deployment progress state", () => { + it("uses an active sibling build when the latest platform build completed", () => { + expect( + getBarState( + service({ + latestBuild: { id: "build-completed", status: "completed" }, + activeBuild: { id: "build-active", status: "building" }, + }), + pendingChanges, + ), + ).toEqual({ + mode: "building", + buildId: "build-active", + buildStatus: "building", + }); + }); + + it("uses an active rollout when the latest rollout is terminal", () => { + expect( + getBarState( + service({ + rollouts: [completedRollout] as ServiceWithDetails["rollouts"], + activeRollout: activeRollout as ServiceWithDetails["activeRollout"], + }), + pendingChanges, + ), + ).toMatchObject({ + mode: "deploying", + rolloutId: "rollout-active", + stage: "health_check", + }); + }); + + it("shows pending changes only when active build and rollout are absent", () => { + expect(getBarState(service(), pendingChanges)).toEqual({ + mode: "ready", + hasChanges: true, + changesCount: 1, + }); + }); + + it("preserves active latest-record fallbacks for legacy payloads", () => { + expect( + getBarState( + service({ + activeBuild: undefined, + latestBuild: { id: "legacy-build", status: "pushing" }, + }), + pendingChanges, + ), + ).toMatchObject({ mode: "building", buildId: "legacy-build" }); + + expect( + getBarState( + service({ + activeRollout: undefined, + rollouts: [activeRollout] as ServiceWithDetails["rollouts"], + }), + pendingChanges, + ), + ).toMatchObject({ mode: "deploying", rolloutId: "rollout-active" }); + }); + + it("keeps migration ahead of build and rollout activity", () => { + expect( + getBarState( + service({ + migrationStatus: "stopping", + activeBuild: { id: "build-active", status: "building" }, + activeRollout: activeRollout as ServiceWithDetails["activeRollout"], + }), + pendingChanges, + ), + ).toMatchObject({ mode: "deploying", stage: "migrating" }); + }); + + it("keeps build activity ahead of rollout activity", () => { + expect( + getBarState( + service({ + activeBuild: { id: "build-active", status: "building" }, + activeRollout: activeRollout as ServiceWithDetails["activeRollout"], + }), + pendingChanges, + ), + ).toMatchObject({ mode: "building", buildId: "build-active" }); + }); +}); diff --git a/web/tests/public-api-source.test.ts b/web/tests/public-api-source.test.ts index 7cfcb4c7..22f97aea 100644 --- a/web/tests/public-api-source.test.ts +++ b/web/tests/public-api-source.test.ts @@ -154,7 +154,9 @@ describe("public API placement schema", () => { }); it.each([ + { mode: "automatic", replicas: 32 }, { mode: "automatic", replicas: 3 }, + { mode: "manual", placements: [{ serverId: "server-1", count: 32 }] }, { mode: "manual", placements: [{ serverId: "server-1", count: 2 }] }, ])("accepts valid placement intent", (placement) => { expect( @@ -165,12 +167,13 @@ describe("public API placement schema", () => { it.each([ { mode: "automatic", replicas: 0 }, + { mode: "automatic", replicas: 33 }, { mode: "manual", placements: [] }, { mode: "manual", placements: [ - { serverId: "a", count: 6 }, - { serverId: "b", count: 5 }, + { serverId: "a", count: 17 }, + { serverId: "b", count: 16 }, ], }, { diff --git a/web/tests/service-revision-spec.test.ts b/web/tests/service-revision-spec.test.ts index 6bb02475..b8e1755c 100644 --- a/web/tests/service-revision-spec.test.ts +++ b/web/tests/service-revision-spec.test.ts @@ -250,14 +250,24 @@ describe("service revision specification", () => { it("snapshots automatic placement intent without resolved placements", () => { const input = draft({ volumes: [] }); input.service.placementMode = "automatic"; - input.service.replicas = 4; + input.service.replicas = 32; expect(buildServiceRevisionSpec(input)).toMatchObject({ - placement: { mode: "automatic", replicas: 4 }, + placement: { mode: "automatic", replicas: 32 }, placements: [], }); }); + it("rejects more than 32 automatic replicas", () => { + const input = draft({ volumes: [] }); + input.service.placementMode = "automatic"; + input.service.replicas = 33; + + expect(() => buildServiceRevisionSpec(input)).toThrow( + "Maximum 32 replicas allowed", + ); + }); + it("rejects automatic placement for stateful and volume-backed services", () => { const stateful = draft({ volumes: [] }); stateful.service.stateful = true;