From 8fc32393f47ee0e10143de2c364e9bd5c1d8691a Mon Sep 17 00:00:00 2001 From: Amp Date: Tue, 28 Jul 2026 07:58:54 +0000 Subject: [PATCH 01/11] Support multiple domains per HTTP port Amp-Thread-ID: https://ampcode.com/threads/T-019fa6e6-5aa9-7137-a803-db807ff3b7aa Co-authored-by: Arjun Komath --- agent/internal/logs/traefik_collector.go | 7 +- agent/internal/logs/traefik_collector_test.go | 11 ++ agent/internal/traefik/l4.go | 20 ++- agent/internal/traefik/routes.go | 26 ++- agent/internal/traefik/routes_test.go | 56 +++++++ cli/internal/manifest/manifest.go | 25 ++- cli/internal/manifest/manifest_test.go | 27 ++- web/actions/projects.ts | 157 +++++++++++------- .../service/details/networking-section.tsx | 2 +- web/lib/agent/expected-state.ts | 26 ++- web/lib/inngest/functions/rollout-helpers.ts | 16 +- web/lib/port-allocation.ts | 9 +- web/lib/public-api.ts | 47 ++++-- web/lib/service-revision-changes.ts | 15 +- web/lib/service-revision-spec.ts | 78 +++++++++ web/lib/victoria-metrics.ts | 2 +- web/tests/expected-state.test.ts | 118 +++++++++++++ web/tests/service-revision-spec.test.ts | 80 ++++++++- .../victoria-metrics-service-metrics.test.ts | 12 +- 19 files changed, 616 insertions(+), 118 deletions(-) create mode 100644 agent/internal/traefik/routes_test.go diff --git a/agent/internal/logs/traefik_collector.go b/agent/internal/logs/traefik_collector.go index 182f4622..5d85f279 100644 --- a/agent/internal/logs/traefik_collector.go +++ b/agent/internal/logs/traefik_collector.go @@ -210,11 +210,8 @@ func (c *TraefikCollector) processLine(line []byte) { } func extractServiceId(routerName string) string { - parts := strings.Split(routerName, "@") - if len(parts) > 0 && parts[0] != "" { - return parts[0] - } - return "" + name := strings.SplitN(routerName, "@", 2)[0] + return strings.SplitN(name, "--", 2)[0] } func (c *TraefikCollector) enqueue(entry HTTPLogEntry) { diff --git a/agent/internal/logs/traefik_collector_test.go b/agent/internal/logs/traefik_collector_test.go index 45f7ee1a..7bd0fd2b 100644 --- a/agent/internal/logs/traefik_collector_test.go +++ b/agent/internal/logs/traefik_collector_test.go @@ -39,3 +39,14 @@ func TestTraefikCollectorProcessLineQueuesRetainedFields(t *testing.T) { t.Fatalf("queued entries = %#v, want %#v", collector.queue, want) } } + +func TestExtractServiceIdFromRouteSpecificRouterName(t *testing.T) { + for input, want := range map[string]string{ + "service-42--app.example.com@file": "service-42", + "service-42@docker": "service-42", + } { + if got := extractServiceId(input); got != want { + t.Fatalf("extractServiceId(%q) = %q, want %q", input, got, want) + } + } +} diff --git a/agent/internal/traefik/l4.go b/agent/internal/traefik/l4.go index 443f38ce..f0f2c125 100644 --- a/agent/internal/traefik/l4.go +++ b/agent/internal/traefik/l4.go @@ -78,11 +78,15 @@ func UpdateHttpRoutesWithL4(httpRoutes []TraefikRoute, tcpRoutes []TraefikTCPRou if len(route.Upstreams) == 0 { continue } + routeName := httpRouteName(route) + if _, exists := config.HTTP.Routers[routeName]; exists { + return fmt.Errorf("duplicate HTTP route %s", routeName) + } - config.HTTP.Routers[route.ServiceId] = routerWithMiddleware{ + config.HTTP.Routers[routeName] = routerWithMiddleware{ Rule: fmt.Sprintf("Host(`%s`)", route.Domain), EntryPoints: []string{"websecure"}, - Service: route.ServiceId, + Service: routeName, TLS: &tlsConfig{}, Middlewares: middlewareNames, } @@ -96,7 +100,7 @@ func UpdateHttpRoutesWithL4(httpRoutes []TraefikRoute, tcpRoutes []TraefikTCPRou servers[i] = srv } - config.HTTP.Services[route.ServiceId] = service{ + config.HTTP.Services[routeName] = service{ LoadBalancer: loadBalancer{ Servers: servers, }, @@ -167,11 +171,11 @@ func UpdateHttpRoutesWithL4(httpRoutes []TraefikRoute, tcpRoutes []TraefikTCPRou return fmt.Errorf("failed to marshal traefik config: %w", err) } - if err := os.MkdirAll(traefikDynamicDir, 0755); err != nil { + if err := os.MkdirAll(dynamicConfigDir, 0755); err != nil { return fmt.Errorf("failed to create dynamic config dir: %w", err) } - routesPath := filepath.Join(traefikDynamicDir, routesFileName) + routesPath := filepath.Join(dynamicConfigDir, routesFileName) tmpPath := routesPath + ".tmp" if err := os.WriteFile(tmpPath, data, 0644); err != nil { @@ -187,6 +191,10 @@ func UpdateHttpRoutesWithL4(httpRoutes []TraefikRoute, tcpRoutes []TraefikTCPRou return nil } +func httpRouteName(route TraefikRoute) string { + return route.ServiceId + "--" + route.ID +} + func HashTCPRoutes(routes []TraefikTCPRoute) string { sortedRoutes := make([]TraefikTCPRoute, len(routes)) copy(sortedRoutes, routes) @@ -318,7 +326,7 @@ func GetCurrentL4ConfigHash() string { } func readCurrentFullConfig() (*traefikFullConfigWithMiddlewares, error) { - routesPath := filepath.Join(traefikDynamicDir, routesFileName) + routesPath := filepath.Join(dynamicConfigDir, routesFileName) data, err := os.ReadFile(routesPath) if err != nil { if os.IsNotExist(err) { diff --git a/agent/internal/traefik/routes.go b/agent/internal/traefik/routes.go index acb8c88b..ab8b26d6 100644 --- a/agent/internal/traefik/routes.go +++ b/agent/internal/traefik/routes.go @@ -13,13 +13,21 @@ func HashRoutes(routes []TraefikRoute) string { sortedRoutes := make([]TraefikRoute, len(routes)) copy(sortedRoutes, routes) sort.Slice(sortedRoutes, func(i, j int) bool { - return sortedRoutes[i].ServiceId < sortedRoutes[j].ServiceId + if sortedRoutes[i].ServiceId != sortedRoutes[j].ServiceId { + return sortedRoutes[i].ServiceId < sortedRoutes[j].ServiceId + } + if sortedRoutes[i].ID != sortedRoutes[j].ID { + return sortedRoutes[i].ID < sortedRoutes[j].ID + } + return sortedRoutes[i].Domain < sortedRoutes[j].Domain }) var sb strings.Builder for _, r := range sortedRoutes { sb.WriteString(r.ServiceId) sb.WriteString(":") + sb.WriteString(r.ID) + sb.WriteString(":") sb.WriteString(r.Domain) sb.WriteString(":") sortedUpstreams := make([]Upstream, len(r.Upstreams)) @@ -53,11 +61,12 @@ func GetCurrentConfigHash() string { } var routes []TraefikRoute - for serviceId, router := range config.HTTP.Routers { + for routerName, router := range config.HTTP.Routers { domain := extractDomainFromRule(router.Rule) + serviceId, routeID := parseHTTPRouteName(routerName) var upstreams []Upstream - if svc, exists := config.HTTP.Services[serviceId]; exists { + if svc, exists := config.HTTP.Services[router.Service]; exists { for _, server := range svc.LoadBalancer.Servers { url := strings.TrimPrefix(server.URL, "http://") weight := 1 @@ -72,7 +81,7 @@ func GetCurrentConfigHash() string { } routes = append(routes, TraefikRoute{ - ID: serviceId, + ID: routeID, Domain: domain, Upstreams: upstreams, ServiceId: serviceId, @@ -84,6 +93,15 @@ func GetCurrentConfigHash() string { return HashRoutesWithServerName(routes, serverName) } +func parseHTTPRouteName(name string) (string, string) { + name = strings.SplitN(name, "@", 2)[0] + parts := strings.SplitN(name, "--", 2) + if len(parts) == 2 { + return parts[0], parts[1] + } + return name, name +} + func extractForwardedServerName(middlewares map[string]middleware) string { if mw, exists := middlewares["forwarded_server"]; exists && mw.Headers != nil { if value, ok := mw.Headers.CustomRequestHeaders["X-Forwarded-Server"]; ok { diff --git a/agent/internal/traefik/routes_test.go b/agent/internal/traefik/routes_test.go new file mode 100644 index 00000000..a4c5b001 --- /dev/null +++ b/agent/internal/traefik/routes_test.go @@ -0,0 +1,56 @@ +package traefik + +import "testing" + +func TestHTTPAliasesGenerateDistinctRoutesAndConvergentHash(t *testing.T) { + originalDir := dynamicConfigDir + t.Cleanup(func() { dynamicConfigDir = originalDir }) + dynamicConfigDir = t.TempDir() + + routes := []TraefikRoute{ + { + ID: "app.example.com", + Domain: "app.example.com", + ServiceId: "service-42", + Upstreams: []Upstream{{URL: "10.0.0.1:3000", Weight: 5}}, + }, + { + ID: "www.example.com", + Domain: "www.example.com", + ServiceId: "service-42", + Upstreams: []Upstream{{URL: "10.0.0.1:3000", Weight: 5}}, + }, + } + + if err := UpdateHttpRoutesWithL4(routes, nil, nil, ""); err != nil { + t.Fatal(err) + } + config, err := readCurrentFullConfig() + if err != nil { + t.Fatal(err) + } + if got := len(config.HTTP.Routers); got != 2 { + t.Fatalf("generated %d HTTP routers, want 2", got) + } + if got := len(config.HTTP.Services); got != 2 { + t.Fatalf("generated %d HTTP services, want 2", got) + } + for _, route := range routes { + name := httpRouteName(route) + router, exists := config.HTTP.Routers[name] + if !exists { + t.Fatalf("router %q was not generated", name) + } + if router.Service != name { + t.Fatalf("router %q targets %q, want %q", name, router.Service, name) + } + } + + if got, want := GetCurrentConfigHash(), HashRoutesWithServerName(routes, ""); got != want { + t.Fatalf("current config hash %q, want %q", got, want) + } + reversed := []TraefikRoute{routes[1], routes[0]} + if HashRoutes(reversed) != HashRoutes(routes) { + t.Fatal("HTTP route hash depends on input order") + } +} diff --git a/cli/internal/manifest/manifest.go b/cli/internal/manifest/manifest.go index 76136a77..d929bb54 100644 --- a/cli/internal/manifest/manifest.go +++ b/cli/internal/manifest/manifest.go @@ -264,15 +264,12 @@ func Validate(m Manifest) error { return errors.New("service.placement.mode must be automatic or manual") } } - seenPorts := make(map[int]struct{}, len(m.Service.Ports)) + portsByNumber := make(map[int][]Port, len(m.Service.Ports)) + seenDomains := make(map[string]struct{}, len(m.Service.Ports)) for i, p := range m.Service.Ports { if p.ContainerPort < 1 || p.ContainerPort > 65535 { return fmt.Errorf("service.ports[%d].containerPort must be between 1 and 65535", i) } - if _, exists := seenPorts[p.ContainerPort]; exists { - return fmt.Errorf("service.ports[%d].containerPort must be unique", i) - } - seenPorts[p.ContainerPort] = struct{}{} if p.Domain != nil && strings.TrimSpace(*p.Domain) == "" { return fmt.Errorf("service.ports[%d].domain cannot be blank", i) } @@ -282,6 +279,24 @@ func Validate(m Manifest) error { if !p.Public && p.Domain != nil { return fmt.Errorf("service.ports[%d].domain cannot be set for internal ports", i) } + if p.Domain != nil { + domain := strings.ToLower(strings.TrimSpace(*p.Domain)) + if _, exists := seenDomains[domain]; exists { + return fmt.Errorf("service.ports[%d].domain must be unique", i) + } + seenDomains[domain] = struct{}{} + } + portsByNumber[p.ContainerPort] = append(portsByNumber[p.ContainerPort], p) + } + for containerPort, ports := range portsByNumber { + if len(ports) < 2 { + continue + } + for _, port := range ports { + if !port.Public || port.Domain == nil { + return fmt.Errorf("service port %d can only be repeated for public domains", containerPort) + } + } } if h := m.Service.HealthCheck; h != nil && (h.Cmd == "" || h.Interval < 1 || h.Timeout < 1 || h.Retries < 1 || h.StartPeriod < 0) { return errors.New("service.healthCheck contains invalid values") diff --git a/cli/internal/manifest/manifest_test.go b/cli/internal/manifest/manifest_test.go index 83e3a025..b18b5709 100644 --- a/cli/internal/manifest/manifest_test.go +++ b/cli/internal/manifest/manifest_test.go @@ -125,7 +125,32 @@ func TestRejectWindowsAbsoluteRootDir(t *testing.T) { func TestRejectDuplicatePorts(t *testing.T) { m := base() m.Service.Ports = []Port{{ContainerPort: 8080}, {ContainerPort: 8080}} - if err := Validate(m); err == nil || !strings.Contains(err.Error(), "must be unique") { + if err := Validate(m); err == nil || !strings.Contains(err.Error(), "can only be repeated") { + t.Fatalf("error = %v", err) + } +} + +func TestAllowsMultipleDomainsOnOnePort(t *testing.T) { + m := base() + first := "app.example.com" + second := "www.example.com" + m.Service.Ports = []Port{ + {ContainerPort: 8080, Public: true, Domain: &first}, + {ContainerPort: 8080, Public: true, Domain: &second}, + } + if err := Validate(m); err != nil { + t.Fatalf("valid port aliases rejected: %v", err) + } +} + +func TestRejectsDuplicatePortDomains(t *testing.T) { + m := base() + domain := "app.example.com" + m.Service.Ports = []Port{ + {ContainerPort: 8080, Public: true, Domain: &domain}, + {ContainerPort: 8080, Public: true, Domain: &domain}, + } + if err := Validate(m); err == nil || !strings.Contains(err.Error(), "domain must be unique") { t.Fatalf("error = %v", err) } } diff --git a/web/actions/projects.ts b/web/actions/projects.ts index 3f678905..d47a5b39 100644 --- a/web/actions/projects.ts +++ b/web/actions/projects.ts @@ -51,6 +51,7 @@ import type { HealthCheckConfig as ServiceHealthCheckConfig, } from "@/lib/service-config"; import { MIN_SERVERLESS_SLEEP_AFTER_SECONDS } from "@/lib/service-config"; +import { findServicePortValidationIssue } from "@/lib/service-revision-spec"; import type { DeleteConfirmation } from "@/lib/two-factor"; import { getZodErrorMessage, slugify } from "@/lib/utils"; import { @@ -1083,78 +1084,114 @@ 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(servicePorts).where(eq(servicePorts.id, portId)); - } - } - - if (config.ports.add && config.ports.add.length > 0) { - const existing = await db - .select() - .from(servicePorts) - .where(eq(servicePorts.serviceId, serviceId)); - - for (const port of config.ports.add) { - const protocol = port.protocol || "http"; - + try { + await db.transaction(async (tx) => { + await tx.execute( + sql`SELECT pg_advisory_xact_lock(hashtext(${serviceId}))`, + ); + const existing = await tx + .select() + .from(servicePorts) + .where(eq(servicePorts.serviceId, serviceId)); + const removedIds = new Set(config.ports?.remove ?? []); if ( - existing.some((p) => p.port === port.port && p.protocol === protocol) + [...removedIds].some((id) => !existing.some((port) => port.id === id)) ) { - throw new Error(`Port ${port.port} (${protocol}) already exists`); + throw new Error("Port not found"); } - if (port.isPublic) { - if (protocol === "http") { - if (!port.domain) { - throw new Error("Domain is required for public HTTP ports"); - } - - const domain = port.domain.trim().toLowerCase(); - if (!domain) { - throw new Error("Invalid domain"); - } - - const existingDomain = await db - .select() - .from(servicePorts) - .where(eq(servicePorts.domain, domain)); - - if (existingDomain.length > 0) { - throw new Error("Domain already in use"); - } - - await db.insert(servicePorts).values({ - id: randomUUID(), - serviceId, - port: port.port, - isPublic: true, - domain, - protocol: "http", - }); - } else if (protocol === "tcp" || protocol === "udp") { - const externalPort = await allocatePort(protocol); - - await db.insert(servicePorts).values({ - id: randomUUID(), - serviceId, - port: port.port, - isPublic: true, - protocol, - externalPort, - tlsPassthrough: port.tlsPassthrough ?? false, - }); + const additions: (typeof servicePorts.$inferInsert)[] = []; + const reservedExternalPorts = { + tcp: new Set(), + udp: new Set(), + }; + if ( + (config.ports?.add ?? []).some( + (port) => + port.isPublic && + (port.protocol === "tcp" || port.protocol === "udp"), + ) + ) { + await tx.execute( + sql`SELECT pg_advisory_xact_lock(hashtext('service_port_external_allocation'))`, + ); + } + for (const port of config.ports?.add ?? []) { + const protocol = port.protocol ?? "http"; + const domain = port.domain?.trim().toLowerCase() || null; + const externalPort = + port.isPublic && (protocol === "tcp" || protocol === "udp") + ? await allocatePort(protocol, reservedExternalPorts[protocol]) + : null; + if ( + externalPort !== null && + (protocol === "tcp" || protocol === "udp") + ) { + reservedExternalPorts[protocol].add(externalPort); } - } else { - await db.insert(servicePorts).values({ + additions.push({ id: randomUUID(), serviceId, port: port.port, - isPublic: false, + isPublic: port.isPublic, + domain: protocol === "http" && port.isPublic ? domain : null, protocol, + externalPort, + tlsPassthrough: + protocol === "tcp" ? (port.tlsPassthrough ?? false) : false, }); } + + const finalPorts = [ + ...existing.filter((port) => !removedIds.has(port.id)), + ...additions, + ]; + const issue = findServicePortValidationIssue( + finalPorts.map((port) => ({ + containerPort: port.port, + isPublic: port.isPublic ?? false, + domain: port.domain ?? null, + protocol: port.protocol ?? "http", + })), + ); + if (issue) throw new Error(issue.message); + + for (const port of additions) { + if (!port.domain) continue; + const conflict = await tx + .select({ serviceId: servicePorts.serviceId }) + .from(servicePorts) + .where(eq(servicePorts.domain, port.domain)) + .limit(1) + .then((rows) => rows[0]); + if (conflict && conflict.serviceId !== serviceId) { + throw new Error("Domain already in use"); + } + } + + if (removedIds.size > 0) { + await tx + .delete(servicePorts) + .where( + and( + eq(servicePorts.serviceId, serviceId), + inArray(servicePorts.id, [...removedIds]), + ), + ); + } + if (additions.length > 0) { + await tx.insert(servicePorts).values(additions); + } + }); + } catch (error) { + if ( + (error as { code?: string; constraint?: string }).code === "23505" && + (error as { constraint?: string }).constraint === + "service_ports_domain_unique" + ) { + throw new Error("Domain already in use"); } + throw error; } } diff --git a/web/components/service/details/networking-section.tsx b/web/components/service/details/networking-section.tsx index 7d202c87..69e72424 100644 --- a/web/components/service/details/networking-section.tsx +++ b/web/components/service/details/networking-section.tsx @@ -124,7 +124,7 @@ export const NetworkingSection = memo(function NetworkingSection({ port <= 65535; const canAdd = isValidPort && - !httpPorts.some((p) => p.port === port) && + !httpPorts.some((p) => p.port === port && !p.isPublic) && isValidDomain && !httpPorts.some((p) => p.domain === normalizedDomain) && !isSaving; diff --git a/web/lib/agent/expected-state.ts b/web/lib/agent/expected-state.ts index db0ada54..0ac7bfc7 100644 --- a/web/lib/agent/expected-state.ts +++ b/web/lib/agent/expected-state.ts @@ -17,9 +17,10 @@ import { } from "@/lib/deployment-status"; import { selectRoutingSyncRolloutIds } from "@/lib/routing-sync"; import { parseServiceRevisionSpec } from "@/lib/service-revision-changes"; -import type { - ServiceRevisionSecret, - ServiceRevisionSpec, +import { + getPublishedContainerPorts, + type ServiceRevisionSecret, + type ServiceRevisionSpec, } from "@/lib/service-revision-spec"; import { getWireGuardPeers } from "@/lib/wireguard"; @@ -269,7 +270,9 @@ async function buildExpectedContainers( ), ); - const serviceIds = [...new Set(serverDeployments.map((dep) => dep.serviceId))]; + const serviceIds = [ + ...new Set(serverDeployments.map((dep) => dep.serviceId)), + ]; const revisionIds = [ ...new Set(serverDeployments.map((dep) => dep.serviceRevisionId)), ]; @@ -357,13 +360,18 @@ export function buildExpectedContainersFromRows({ (a, b) => a.containerPort - b.containerPort || a.hostPort - b.hostPort, ) + .filter( + (port, index, sortedPorts) => + index === 0 || + sortedPorts[index - 1].containerPort !== port.containerPort, + ) .map((port) => ({ containerPort: port.containerPort, hostPort: port.hostPort, })); - const expectedContainerPorts = specification.ports - .map((port) => port.containerPort) - .sort((a, b) => a - b); + const expectedContainerPorts = getPublishedContainerPorts( + specification.ports, + ); const allocatedContainerPorts = ports.map((port) => port.containerPort); if ( JSON.stringify(expectedContainerPorts) !== @@ -855,6 +863,8 @@ function compareServicePorts(a: RouteServicePort, b: RouteServicePort) { return ( a.serviceId.localeCompare(b.serviceId) || a.protocol.localeCompare(b.protocol) || - a.port - b.port + a.port - b.port || + (a.domain ?? "").localeCompare(b.domain ?? "") || + a.id.localeCompare(b.id) ); } diff --git a/web/lib/inngest/functions/rollout-helpers.ts b/web/lib/inngest/functions/rollout-helpers.ts index 0b7e153a..4b4c8cc0 100644 --- a/web/lib/inngest/functions/rollout-helpers.ts +++ b/web/lib/inngest/functions/rollout-helpers.ts @@ -9,7 +9,10 @@ import { services, } from "@/db/schema"; import { getCertificate, issueCertificate } from "@/lib/acme-manager"; -import type { ServiceRevisionSpec } from "@/lib/service-revision-spec"; +import { + getPublishedContainerPorts, + type ServiceRevisionSpec, +} from "@/lib/service-revision-spec"; import { assignContainerIp } from "@/lib/wireguard"; import { enqueueWork } from "@/lib/work-queue"; @@ -381,6 +384,9 @@ export async function createDeploymentRecords( } const deploymentIds = existingDeployments.map((deployment) => deployment.id); + const publishedContainerPorts = getPublishedContainerPorts( + specification.ports, + ); for (const placement of placements) { if (placement.replicas <= 0) continue; @@ -396,7 +402,7 @@ export async function createDeploymentRecords( const deploymentId = randomUUID(); const hostPorts = await allocateHostPorts( server.id, - specification.ports.length, + publishedContainerPorts.length, ); const ipAddress = await assignContainerIp(server.id); @@ -425,12 +431,12 @@ export async function createDeploymentRecords( rolloutId, }); - if (specification.ports.length > 0) { + if (publishedContainerPorts.length > 0) { await tx.insert(deploymentPorts).values( - specification.ports.map((port, index) => ({ + publishedContainerPorts.map((containerPort, index) => ({ id: randomUUID(), deploymentId, - containerPort: port.containerPort, + containerPort, hostPort: hostPorts[index], })), ); diff --git a/web/lib/port-allocation.ts b/web/lib/port-allocation.ts index 19535930..b18492d5 100644 --- a/web/lib/port-allocation.ts +++ b/web/lib/port-allocation.ts @@ -1,13 +1,16 @@ +import { asc, eq } from "drizzle-orm"; import { db } from "@/db"; import { servicePorts } from "@/db/schema"; -import { asc, eq } from "drizzle-orm"; const TCP_PORT_START = 10000; const TCP_PORT_END = 10999; const UDP_PORT_START = 11000; const UDP_PORT_END = 11999; -export async function allocatePort(protocol: "tcp" | "udp"): Promise { +export async function allocatePort( + protocol: "tcp" | "udp", + reserved: ReadonlySet = new Set(), +): Promise { const portStart = protocol === "tcp" ? TCP_PORT_START : UDP_PORT_START; const portEnd = protocol === "tcp" ? TCP_PORT_END : UDP_PORT_END; @@ -22,7 +25,7 @@ export async function allocatePort(protocol: "tcp" | "udp"): Promise { ); for (let port = portStart; port <= portEnd; port++) { - if (!usedSet.has(port)) { + if (!usedSet.has(port) && !reserved.has(port)) { return port; } } diff --git a/web/lib/public-api.ts b/web/lib/public-api.ts index 5f80af4f..beb5011a 100644 --- a/web/lib/public-api.ts +++ b/web/lib/public-api.ts @@ -18,6 +18,7 @@ import { validateDockerImageInternal } from "@/lib/docker-image"; import { getServiceTotalReplicas } from "@/lib/service-config"; import { parseServiceRevisionSpec } from "@/lib/service-revision-changes"; import { + findServicePortValidationIssue, getDefaultServiceHostname, getServiceRevisionTotalReplicas, } from "@/lib/service-revision-spec"; @@ -654,7 +655,7 @@ export async function patchConfiguration( } } - return db.transaction(async (tx) => { + const update = db.transaction(async (tx) => { await tx.execute( sql`SELECT pg_advisory_xact_lock(hashtext(${service.id}))`, ); @@ -795,18 +796,20 @@ export async function patchConfiguration( 400, ); } - if ( - new Set(input.ports.map((port) => port.containerPort)).size !== - input.ports.length - ) { - domainError("Port numbers must be unique", "DUPLICATE_PORT", 400); + const portIssue = findServicePortValidationIssue( + input.ports.map((port) => ({ + containerPort: port.containerPort, + isPublic: port.public, + domain: port.domain ?? null, + protocol: "http" as const, + })), + ); + if (portIssue) { + domainError(portIssue.message, portIssue.code, 400); } const domains = input.ports.flatMap((port) => port.public && port.domain ? [port.domain] : [], ); - if (new Set(domains).size !== domains.length) { - domainError("Port domains must be unique", "DUPLICATE_DOMAIN", 400); - } for (const domain of domains) { const duplicate = await tx .select({ id: servicePorts.id }) @@ -965,13 +968,23 @@ export async function patchConfiguration( if (input.ports) { const currentPorts = ports .map((port) => [port.port, port.isPublic, port.domain] as const) - .toSorted((a, b) => a[0] - b[0]); + .toSorted( + (a, b) => + a[0] - b[0] || + Number(a[1]) - Number(b[1]) || + (a[2] ?? "").localeCompare(b[2] ?? ""), + ); const desiredPorts = input.ports .map( (port) => [port.containerPort, port.public, port.domain ?? null] as const, ) - .toSorted((a, b) => a[0] - b[0]); + .toSorted( + (a, b) => + a[0] - b[0] || + Number(a[1]) - Number(b[1]) || + (a[2] ?? "").localeCompare(b[2] ?? ""), + ); if (changed("ports", currentPorts, desiredPorts)) { await tx .delete(servicePorts) @@ -996,4 +1009,16 @@ export async function patchConfiguration( changes, }; }); + try { + return await update; + } catch (error) { + if ( + (error as { code?: string; constraint?: string }).code === "23505" && + (error as { constraint?: string }).constraint === + "service_ports_domain_unique" + ) { + domainError("Port domain is already in use", "DOMAIN_CONFLICT"); + } + throw error; + } } diff --git a/web/lib/service-revision-changes.ts b/web/lib/service-revision-changes.ts index 21cd9000..127f2a82 100644 --- a/web/lib/service-revision-changes.ts +++ b/web/lib/service-revision-changes.ts @@ -6,6 +6,7 @@ import type { ServiceRevisionPort, ServiceRevisionSpec, } from "@/lib/service-revision-spec"; +import { validateServiceRevisionPorts } from "@/lib/service-revision-spec"; const serviceRevisionSpecFields = { image: z.string(), @@ -143,9 +144,19 @@ export function parseServiceRevisionSpec(value: unknown): ServiceRevisionSpec { const version = (value as { schemaVersion?: unknown } | null)?.schemaVersion; if (version === 2) { const legacy = serviceRevisionSpecV2Schema.parse(value); - return { ...legacy, schemaVersion: 3, placement: { mode: "manual" } }; + const specification = { + ...legacy, + schemaVersion: 3 as const, + placement: { mode: "manual" as const }, + }; + validateServiceRevisionPorts(specification.ports); + return specification; } - return serviceRevisionSpecSchema.parse(value) as ServiceRevisionSpec; + const specification = serviceRevisionSpecSchema.parse( + value, + ) as ServiceRevisionSpec; + validateServiceRevisionPorts(specification.ports); + return specification; } function compareStrings(a: string, b: string): number { diff --git a/web/lib/service-revision-spec.ts b/web/lib/service-revision-spec.ts index 5b289cb4..89467da5 100644 --- a/web/lib/service-revision-spec.ts +++ b/web/lib/service-revision-spec.ts @@ -33,6 +33,83 @@ export type ServiceRevisionPort = { tlsPassthrough: boolean; }; +export type ServicePortValidationIssue = { + code: "DUPLICATE_DOMAIN" | "DUPLICATE_PORT" | "INVALID_DOMAIN"; + message: string; +}; + +export function findServicePortValidationIssue( + ports: Pick< + ServiceRevisionPort, + "containerPort" | "isPublic" | "domain" | "protocol" + >[], +): ServicePortValidationIssue | null { + const domains = new Set(); + const portsByNumberAndProtocol = Map.groupBy( + ports, + (port) => `${port.containerPort}/${port.protocol}`, + ); + + for (const port of ports) { + if (port.protocol === "http" && port.isPublic && !port.domain) { + return { + code: "INVALID_DOMAIN", + message: "Public HTTP ports require a domain", + }; + } + if ((!port.isPublic || port.protocol !== "http") && port.domain) { + return { + code: "INVALID_DOMAIN", + message: "Only public HTTP ports can define a domain", + }; + } + if (port.domain) { + const domain = port.domain.toLowerCase(); + if (domains.has(domain)) { + return { + code: "DUPLICATE_DOMAIN", + message: "Port domains must be unique", + }; + } + domains.add(domain); + } + } + + for (const group of portsByNumberAndProtocol.values()) { + if ( + group.length > 1 && + !group.every( + (port) => port.protocol === "http" && port.isPublic && port.domain, + ) + ) { + return { + code: "DUPLICATE_PORT", + message: `Port ${group[0].containerPort} (${group[0].protocol}) can only be repeated for public HTTP domains`, + }; + } + } + + return null; +} + +export function validateServiceRevisionPorts( + ports: Pick< + ServiceRevisionPort, + "containerPort" | "isPublic" | "domain" | "protocol" + >[], +) { + const issue = findServicePortValidationIssue(ports); + if (issue) throw new Error(issue.message); +} + +export function getPublishedContainerPorts( + ports: Pick[], +): number[] { + return [...new Set(ports.map((port) => port.containerPort))].sort( + (a, b) => a - b, + ); +} + export type ServiceRevisionSecret = { key: string; encryptedValue: string; @@ -143,6 +220,7 @@ function validateServiceRevisionSpec( specification: ServiceRevisionSpec, allowNoPlacements: boolean, ) { + validateServiceRevisionPorts(specification.ports); const totalReplicas = getServiceRevisionTotalReplicas(specification); if (totalReplicas < 1 && !allowNoPlacements) { diff --git a/web/lib/victoria-metrics.ts b/web/lib/victoria-metrics.ts index 301b1773..ad7a4ee9 100644 --- a/web/lib/victoria-metrics.ts +++ b/web/lib/victoria-metrics.ts @@ -643,7 +643,7 @@ export function getMetricWindow( } export function buildTraefikServiceMatcher(serviceId: string): string { - return `^${escapePromRegex(serviceId)}(@file)?$`; + return `^${escapePromRegex(serviceId)}(?:--[^@]+)?(@file)?$`; } export function formatPromDuration(seconds: number): string { diff --git a/web/tests/expected-state.test.ts b/web/tests/expected-state.test.ts index 138355a0..6cf80a5b 100644 --- a/web/tests/expected-state.test.ts +++ b/web/tests/expected-state.test.ts @@ -259,6 +259,107 @@ describe("expected-state pure builders", () => { ).toThrow("Deployment dep_incomplete_ports has incomplete port allocation"); }); + it("publishes one physical port for multiple HTTP domain aliases", () => { + const ports = ["app.example.com", "www.example.com"].map((domain) => ({ + containerPort: 3000, + isPublic: true, + domain, + protocol: "http" as const, + externalPort: null, + tlsPassthrough: false, + })); + const containers = buildExpectedContainersFromRows({ + deployments: [ + { + id: "dep_aliases", + serviceId: "svc_aliases", + serviceRevisionId: "rev_svc_aliases", + runtimeDesiredState: "running", + }, + ] as any, + services: [{ id: "svc_aliases", name: "aliases" }] as any, + revisions: [revision("svc_aliases", { ports })], + deploymentPorts: [ + { deploymentId: "dep_aliases", containerPort: 3000, hostPort: 31000 }, + ] as any, + }); + + expect(containers[0]?.ports).toEqual([ + { containerPort: 3000, hostPort: 31000 }, + ]); + }); + + it("deduplicates historical physical mappings for cross-protocol ports", () => { + const containers = buildExpectedContainersFromRows({ + deployments: [ + { + id: "dep_protocols", + serviceId: "svc_protocols", + serviceRevisionId: "rev_svc_protocols", + runtimeDesiredState: "running", + }, + ] as any, + services: [{ id: "svc_protocols", name: "protocols" }] as any, + revisions: [ + revision("svc_protocols", { + ports: [ + { + containerPort: 3000, + isPublic: true, + domain: "app.example.com", + protocol: "http", + externalPort: null, + tlsPassthrough: false, + }, + { + containerPort: 3000, + isPublic: true, + domain: null, + protocol: "tcp", + externalPort: 10000, + tlsPassthrough: false, + }, + ], + }), + ], + deploymentPorts: [ + { deploymentId: "dep_protocols", containerPort: 3000, hostPort: 31000 }, + { deploymentId: "dep_protocols", containerPort: 3000, hostPort: 31001 }, + ] as any, + }); + + expect(containers[0]?.ports).toEqual([ + { containerPort: 3000, hostPort: 31000 }, + ]); + }); + + it("builds one HTTP route per domain alias", () => { + const routes = buildTraefikRoutes({ + serverId: "server_1", + ports: ["app.example.com", "www.example.com"].map((domain, index) => ({ + id: `port_${index}`, + serviceId: "svc_1", + port: 3000, + isPublic: true, + protocol: "http", + domain, + })) as any, + routableDeployments: [ + { serviceId: "svc_1", serverId: "server_1", ipAddress: "10.0.0.1" }, + ] as any, + }); + + expect(routes.httpRoutes.map((route) => route.domain)).toEqual([ + "app.example.com", + "www.example.com", + ]); + expect( + routes.httpRoutes.every( + (route) => route.upstreams[0]?.url === "10.0.0.1:3000", + ), + ).toBe(true); + }); + it("keeps HTTP local upstreams before remote upstreams", () => { const routes = buildTraefikRoutes({ serverId: "server_local", @@ -870,6 +971,14 @@ describe("expected-state pure builders", () => { externalPort: null, tlsPassthrough: false, }, + { + containerPort: 3000, + isPublic: true, + domain: "www.sleepy.example.com", + protocol: "http", + externalPort: null, + tlsPassthrough: false, + }, ], }); const routes = buildServerlessRoutesFromRows({ @@ -903,6 +1012,15 @@ describe("expected-state pure builders", () => { localDeploymentIds: ["dep_sleeping"], upstreams: [], }, + { + serviceId: "svc_1", + domain: "www.sleepy.example.com", + port: 3000, + sleepAfterSeconds: 300, + wakeTimeoutSeconds: 120, + localDeploymentIds: ["dep_sleeping"], + upstreams: [], + }, ]); }); diff --git a/web/tests/service-revision-spec.test.ts b/web/tests/service-revision-spec.test.ts index 1bf4831b..fb888ce1 100644 --- a/web/tests/service-revision-spec.test.ts +++ b/web/tests/service-revision-spec.test.ts @@ -33,7 +33,7 @@ function draft( { port: 443, isPublic: true, - domain: "api.example.com", + domain: null, protocol: "tcp", externalPort: 443, tlsPassthrough: true, @@ -85,6 +85,7 @@ describe("service revision specification", () => { input.service.serverlessSleepAfterSeconds = 30; input.service.healthCheckInterval = null; input.ports[0].protocol = null; + input.ports[0].domain = "api.example.com"; input.ports[0].tlsPassthrough = null; const spec = buildServiceRevisionSpec(input); @@ -105,6 +106,83 @@ describe("service revision specification", () => { ); }); + it("allows distinct public HTTP domains to share a container port", () => { + const input = draft({ + ports: [ + { + port: 3000, + isPublic: true, + domain: "app.example.com", + protocol: "http", + externalPort: null, + tlsPassthrough: false, + }, + { + port: 3000, + isPublic: true, + domain: "www.example.com", + protocol: "http", + externalPort: null, + tlsPassthrough: false, + }, + ], + }); + + expect(buildServiceRevisionSpec(input).ports).toHaveLength(2); + }); + + it("allows different protocols to share a numeric container port", () => { + const input = draft({ + ports: [ + { + port: 3000, + isPublic: true, + domain: "app.example.com", + protocol: "http", + externalPort: null, + tlsPassthrough: false, + }, + { + port: 3000, + isPublic: true, + domain: null, + protocol: "tcp", + externalPort: 10000, + tlsPassthrough: false, + }, + ], + }); + + expect(buildServiceRevisionSpec(input).ports).toHaveLength(2); + }); + + it("rejects repeated container ports that are not HTTP aliases", () => { + const input = draft({ + ports: [ + { + port: 3000, + isPublic: true, + domain: "app.example.com", + protocol: "http", + externalPort: null, + tlsPassthrough: false, + }, + { + port: 3000, + isPublic: false, + domain: null, + protocol: "http", + externalPort: null, + tlsPassthrough: false, + }, + ], + }); + + expect(() => buildServiceRevisionSpec(input)).toThrow( + "can only be repeated for public HTTP domains", + ); + }); + it("snapshots GitHub source provenance and the reserved runtime image", () => { const spec = buildServiceRevisionSpec(draft(), { image: "registry.test/project/service:revision-1", diff --git a/web/tests/victoria-metrics-service-metrics.test.ts b/web/tests/victoria-metrics-service-metrics.test.ts index cf44d606..9cec578d 100644 --- a/web/tests/victoria-metrics-service-metrics.test.ts +++ b/web/tests/victoria-metrics-service-metrics.test.ts @@ -25,9 +25,11 @@ describe("VictoriaMetrics service metrics", () => { it("matches Traefik service labels with optional provider suffix", () => { expect(buildTraefikServiceMatcher(SERVICE_ID)).toBe( - `^${SERVICE_ID}(@file)?$`, + `^${SERVICE_ID}(?:--[^@]+)?(@file)?$`, + ); + expect(buildTraefikServiceMatcher("svc.1")).toBe( + "^svc\\.1(?:--[^@]+)?(@file)?$", ); - expect(buildTraefikServiceMatcher("svc.1")).toBe("^svc\\.1(@file)?$"); }); it("creates an empty metrics payload", () => { @@ -129,7 +131,7 @@ describe("VictoriaMetrics service metrics", () => { expect(instantTimes).toEqual([String(END_TS), String(END_TS)]); expect(queries.some((query) => query.includes("LogSQL"))).toBe(false); expect(queries).toContain( - `sum by (code) (increase(traefik_service_requests_total{service=~"^${SERVICE_ID}(@file)?$"}[5m]))`, + `sum by (code) (increase(traefik_service_requests_total{service=~"^${SERVICE_ID}(?:--[^@]+)?(@file)?$"}[5m]))`, ); expect(queries).toContain( `sum(avg_over_time(techulus_service_cpu_usage_percent{service_id="${SERVICE_ID}"}[5m]))`, @@ -138,10 +140,10 @@ describe("VictoriaMetrics service metrics", () => { `sum(avg_over_time(techulus_service_memory_usage_percent{service_id="${SERVICE_ID}"}[5m]))`, ); expect(queries).toContain( - `sum(increase(traefik_service_requests_bytes_total{service=~"^${SERVICE_ID}(@file)?$"}[1d]))`, + `sum(increase(traefik_service_requests_bytes_total{service=~"^${SERVICE_ID}(?:--[^@]+)?(@file)?$"}[1d]))`, ); expect(queries).toContain( - `sum(increase(traefik_service_responses_bytes_total{service=~"^${SERVICE_ID}(@file)?$"}[1d]))`, + `sum(increase(traefik_service_responses_bytes_total{service=~"^${SERVICE_ID}(?:--[^@]+)?(@file)?$"}[1d]))`, ); expect( starts.every((start) => start === String(END_TS - 24 * 60 * 60 + 5 * 60)), From 265cd794068110b893b3e8bb1c9b5bd24e057429 Mon Sep 17 00:00:00 2001 From: Amp Date: Tue, 28 Jul 2026 07:58:56 +0000 Subject: [PATCH 02/11] Support automatic serverless placement Amp-Thread-ID: https://ampcode.com/threads/T-019fa5f7-6f82-77d6-a291-8ee1896c3a4c Co-authored-by: Arjun Komath --- docs/api/public-api.mdx | 2 +- docs/services/scaling.mdx | 7 ++- docs/services/volumes.mdx | 4 +- web/actions/projects.ts | 51 +++++++++------ .../service/details/replicas-section.tsx | 19 +++--- .../service/details/serverless-section.tsx | 4 +- .../service/details/volumes-section.tsx | 17 ++++- web/lib/inngest/functions/rollout-helpers.ts | 4 ++ web/lib/inngest/functions/rollout-workflow.ts | 10 ++- web/lib/public-api.ts | 4 +- web/lib/scheduler.ts | 5 +- web/lib/service-revision-changes.ts | 15 ++++- web/lib/service-revision-spec.ts | 10 +-- web/tests/expected-state.test.ts | 62 +------------------ web/tests/service-revision-changes.test.ts | 34 +++++++++- web/tests/service-revision-spec.test.ts | 48 ++++++++++++-- 16 files changed, 175 insertions(+), 121 deletions(-) diff --git a/docs/api/public-api.mdx b/docs/api/public-api.mdx index 5fb9c8ae..2f2dd5cc 100644 --- a/docs/api/public-api.mdx +++ b/docs/api/public-api.mdx @@ -209,7 +209,7 @@ Use manual placement to choose exact servers: } ``` -Manual placement requires online servers with WireGuard configured. Serverless services require proxy servers. Automatic placement is not available for stateful, serverless, or volume-backed services. Submit replica changes through `placement`; the API rejects a top-level `replicas` field. +Manual placement requires online servers with WireGuard configured. Serverless services require proxy servers, including when automatic placement is used. Automatic placement is not available for stateful or volume-backed services. Submit replica changes through `placement`; the API rejects a top-level `replicas` field. The API only manages stateless services with HTTP ports. Existing volumes, stateful mode, TCP or UDP ports, TLS passthrough, or invalid resource limits return a conflict with an actionable code. diff --git a/docs/services/scaling.mdx b/docs/services/scaling.mdx index cb1e3cb3..61966591 100644 --- a/docs/services/scaling.mdx +++ b/docs/services/scaling.mdx @@ -42,9 +42,9 @@ stores the desired replica count and distributes replicas across online, configured nodes during rollout. Manual placement selects exact target servers and replica counts. -Serverless services currently require manual placement on proxy nodes. Automatic -serverless placement will remain unavailable until every healthy proxy can route -requests through the owning proxy's wake gateway. +Serverless services support automatic placement across online, configured proxy +nodes. Public ingress must use health checks to avoid proxy nodes that do not +currently own a replica for the service. ## Server Pinning @@ -61,6 +61,7 @@ You can also manually lock any service to a specific server by setting the locke - Stateful services do not automatically fail over to another server. - Maximum 10 replicas per service. - Serverless scaling requires a public HTTP service domain. +- Serverless services must be stateless and cannot use volumes. - 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. - Proxy agents report sleep and wake transitions through normal status reports. diff --git a/docs/services/volumes.mdx b/docs/services/volumes.mdx index d516e02e..ad9bd6ee 100644 --- a/docs/services/volumes.mdx +++ b/docs/services/volumes.mdx @@ -18,7 +18,8 @@ Each volume has a name and a container path: When you add a volume, the service automatically becomes **stateful**. Stateful services are locked to a single server and limited to 1 replica so the container always mounts the same local data path. When the last volume is removed, the service reverts to stateless. -Stateful services can use serverless scaling when they have a public HTTP domain. Because volumes are local to one server, only a proxy-hosted stateful replica can sleep and wake on request. A stateful replica placed on a worker node stays always on. +Serverless scaling is not available for services with volumes. Disable serverless +before adding a volume. ## Volume Backups @@ -47,6 +48,7 @@ You can restore a volume from any completed backup. The restore process download ## Limitations - Services with volumes are locked to a single server. +- Services with volumes cannot use serverless scaling. - Replica count is fixed at 1 for stateful services. - Volume data lives on the host filesystem and is not replicated to other servers. - If the server is lost, data is only recoverable from completed backups. diff --git a/web/actions/projects.ts b/web/actions/projects.ts index 3f678905..13db96c4 100644 --- a/web/actions/projects.ts +++ b/web/actions/projects.ts @@ -925,10 +925,8 @@ export async function updateServiceServerlessSettings( } if (validated.enabled) { - if (service.placementMode === "automatic") { - throw new Error( - "Switch to manual placement before enabling serverless", - ); + if (service.stateful) { + throw new Error("Serverless services must be stateless"); } const publicHttpPorts = await tx .select({ id: servicePorts.id }) @@ -949,24 +947,38 @@ export async function updateServiceServerlessSettings( ); } - const configuredReplicas = await tx - .select({ - count: serviceReplicas.count, - serverIsProxy: servers.isProxy, - }) - .from(serviceReplicas) - .innerJoin(servers, eq(serviceReplicas.serverId, servers.id)) - .where(eq(serviceReplicas.serviceId, serviceId)); - const totalConfiguredReplicas = configuredReplicas.reduce( - (total, replica) => total + replica.count, - 0, - ); + const [configuredReplicas, volume] = await Promise.all([ + tx + .select({ + count: serviceReplicas.count, + serverIsProxy: servers.isProxy, + }) + .from(serviceReplicas) + .innerJoin(servers, eq(serviceReplicas.serverId, servers.id)) + .where(eq(serviceReplicas.serviceId, serviceId)), + tx + .select({ id: serviceVolumes.id }) + .from(serviceVolumes) + .where(eq(serviceVolumes.serviceId, serviceId)) + .limit(1), + ]); + if (volume.length > 0) { + throw new Error("Serverless services cannot use volumes"); + } + const totalConfiguredReplicas = + service.placementMode === "automatic" + ? service.replicas + : configuredReplicas.reduce( + (total, replica) => total + replica.count, + 0, + ); if (totalConfiguredReplicas < 1) { throw new Error("Serverless services require at least one replica"); } if ( + service.placementMode === "manual" && configuredReplicas.some( (replica) => replica.count > 0 && !replica.serverIsProxy, ) @@ -1177,10 +1189,6 @@ export async function updateServiceConfig( if (!currentService) throw new Error("Service not found"); if (placement.mode === "automatic") { - if (currentService.serverlessEnabled) - throw new Error( - "Automatic placement is not supported for serverless services", - ); const volume = await tx .select({ id: serviceVolumes.id }) .from(serviceVolumes) @@ -1419,6 +1427,9 @@ export async function addServiceVolume( .where(and(eq(services.id, serviceId), isNull(services.deletedAt))) .then((rows) => rows[0]); if (!service) throw new Error("Service not found"); + if (service.serverlessEnabled) { + throw new Error("Disable serverless before adding a volume"); + } if (service.placementMode === "automatic") { throw new Error("Switch to manual placement before adding a volume"); } diff --git a/web/components/service/details/replicas-section.tsx b/web/components/service/details/replicas-section.tsx index 280ddee3..ce54091d 100644 --- a/web/components/service/details/replicas-section.tsx +++ b/web/components/service/details/replicas-section.tsx @@ -53,7 +53,7 @@ export const ReplicasSection = memo(function ReplicasSection({ ); const [selectedServerId, setSelectedServerId] = useState(null); const [placementMode, setPlacementMode] = useState( - service.serverlessEnabled ? "manual" : service.placementMode, + service.placementMode, ); const [desiredReplicas, setDesiredReplicas] = useState(service.replicas); const [isEditing, setIsEditing] = useState(false); @@ -86,9 +86,7 @@ export const ReplicasSection = memo(function ReplicasSection({ } } setLocalReplicas(replicaMap); - setPlacementMode( - service.serverlessEnabled ? "manual" : service.placementMode, - ); + setPlacementMode(service.placementMode); setDesiredReplicas(service.replicas); } }, [ @@ -97,7 +95,6 @@ export const ReplicasSection = memo(function ReplicasSection({ service.stateful, service.lockedServerId, service.placementMode, - service.serverlessEnabled, service.replicas, isEditing, ]); @@ -205,7 +202,6 @@ export const ReplicasSection = memo(function ReplicasSection({ const handleModeChange = (mode: string) => { const nextMode = mode as PlacementMode; - if (nextMode === "automatic" && service.serverlessEnabled) return; if (nextMode === placementMode) return; setIsEditing(true); if (nextMode === "automatic") { @@ -384,11 +380,9 @@ export const ReplicasSection = memo(function ReplicasSection({
- {!service.serverlessEnabled && ( - - Automatic - - )} + + Automatic + Manual @@ -403,7 +397,8 @@ export const ReplicasSection = memo(function ReplicasSection({

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

replica.count > 0 && !replica.serverIsProxy, ); const unavailableReason = - service.placementMode === "automatic" - ? "Switch to manual placement before enabling serverless" + service.stateful || (service.volumes?.length ?? 0) > 0 + ? "Serverless services must be stateless and cannot use volumes" : !hasPublicHttpEndpoint ? "Add a public HTTP port with a domain to enable serverless" : hasWorkerReplica diff --git a/web/components/service/details/volumes-section.tsx b/web/components/service/details/volumes-section.tsx index 53a334b8..c9a6c2c8 100644 --- a/web/components/service/details/volumes-section.tsx +++ b/web/components/service/details/volumes-section.tsx @@ -22,6 +22,9 @@ export const VolumesSection = memo(function VolumesSection({ const [error, setError] = useState(null); const volumes = service.volumes || []; + const volumeAddUnavailableReason = service.serverlessEnabled + ? "Disable serverless before adding a volume" + : null; const handleAdd = async () => { if (!name || !containerPath) return; @@ -107,23 +110,35 @@ export const VolumesSection = memo(function VolumesSection({ {error && (

{error}

)} + {volumeAddUnavailableReason && ( +

+ {volumeAddUnavailableReason}. +

+ )}
setName(e.target.value)} + disabled={!!volumeAddUnavailableReason} className="flex-1" /> setContainerPath(e.target.value)} + disabled={!!volumeAddUnavailableReason} className="flex-1" />
- - + + {displayedEndpoints.map((endpoint, index) => ( + + {index > 0 ? ", " : null} + + + ))} {formatPortSummary(service.ports || [])} - {primaryEndpoint.kind !== "private" ? ( + {displayedEndpoints[0]?.kind !== "private" ? ( {`${service.hostname || service.name}.internal`} @@ -896,18 +903,28 @@ function formatInstanceSummary(overview: OverviewData): string { return `${overview.runningDeployments}/${configured} running`; } -function getPrimaryEndpoint(endpoints: EndpointItem[]): EndpointItem { - return ( - endpoints.find((endpoint) => endpoint.kind === "public") ?? - endpoints.find((endpoint) => endpoint.kind === "tcp") ?? +function getDisplayedEndpoints(endpoints: EndpointItem[]): EndpointItem[] { + const publicEndpoints = endpoints.filter( + (endpoint) => endpoint.kind === "public", + ); + if (publicEndpoints.length > 0) { + return publicEndpoints.sort((a, b) => a.label.localeCompare(b.label)); + } + + const tcpEndpoints = endpoints.filter((endpoint) => endpoint.kind === "tcp"); + if (tcpEndpoints.length > 0) { + return tcpEndpoints.sort((a, b) => a.label.localeCompare(b.label)); + } + + return [ endpoints[0] ?? { key: "none", kind: "private", typeLabel: "Private", label: "No endpoint", target: "Internal DNS", - } - ); + }, + ]; } function formatPortSummary(ports: Service["ports"]): string { diff --git a/web/lib/service-config.ts b/web/lib/service-config.ts index 51394867..f7f30dd8 100644 --- a/web/lib/service-config.ts +++ b/web/lib/service-config.ts @@ -538,49 +538,20 @@ export function diffConfigs( }); } - const deployedPortsMap = new Map( - (deployed.ports || []).map((p) => [p.port, p]), - ); - const currentPortsMap = new Map( - (current.ports || []).map((p) => [p.port, p]), - ); - - for (const [port, currentPort] of currentPortsMap) { - const deployedPort = deployedPortsMap.get(port); - const portType = currentPort.isPublic ? "public" : "internal"; - const portDesc = currentPort.domain - ? `${portType}, ${currentPort.domain}` - : portType; - - if (!deployedPort) { + const deployedPortsByNumber = groupPortsByNumber(deployed.ports || []); + const currentPortsByNumber = groupPortsByNumber(current.ports || []); + const portNumbers = Array.from( + new Set([...deployedPortsByNumber.keys(), ...currentPortsByNumber.keys()]), + ).sort((a, b) => a - b); + + for (const port of portNumbers) { + const deployedDesc = describePorts(deployedPortsByNumber.get(port) || []); + const currentDesc = describePorts(currentPortsByNumber.get(port) || []); + if (deployedDesc !== currentDesc) { changes.push({ field: `Port ${port}`, - from: "(none)", - to: portDesc, - }); - } else { - const deployedType = deployedPort.isPublic ? "public" : "internal"; - const deployedDesc = deployedPort.domain - ? `${deployedType}, ${deployedPort.domain}` - : deployedType; - - if (deployedDesc !== portDesc) { - changes.push({ - field: `Port ${port}`, - from: deployedDesc, - to: portDesc, - }); - } - } - } - - for (const [port, deployedPort] of deployedPortsMap) { - if (!currentPortsMap.has(port)) { - const deployedType = deployedPort.isPublic ? "public" : "internal"; - changes.push({ - field: `Port ${port}`, - from: deployedType, - to: "(removed)", + from: deployedDesc || "(none)", + to: currentDesc || "(removed)", }); } } @@ -659,6 +630,30 @@ export function diffConfigs( return changes; } +function groupPortsByNumber(ports: PortConfig[]): Map { + const grouped = new Map(); + for (const port of ports) { + const entries = grouped.get(port.port) || []; + entries.push(port); + grouped.set(port.port, entries); + } + return grouped; +} + +function describePorts(ports: PortConfig[]): string { + return ports + .map((port) => { + const details = [port.isPublic ? "public" : "internal"]; + const protocol = port.protocol || "http"; + if (protocol !== "http") details.push(protocol.toUpperCase()); + if (port.domain) details.push(port.domain); + if (port.tlsPassthrough) details.push("TLS passthrough"); + return details.join(", "); + }) + .sort((a, b) => a.localeCompare(b)) + .join("; "); +} + export function normalizeServerlessConfig( config: ServerlessConfig | undefined, ): ServerlessConfig { diff --git a/web/tests/service-config.test.ts b/web/tests/service-config.test.ts index 95933302..db76a47b 100644 --- a/web/tests/service-config.test.ts +++ b/web/tests/service-config.test.ts @@ -234,6 +234,52 @@ describe("service config", () => { expect(diffConfigs(deployed, current)).toEqual([]); }); + it("compares multiple domains on one port without depending on order", () => { + const first = { + port: 8080, + isPublic: true, + domain: "app.example.com", + protocol: "http" as const, + }; + const second = { + port: 8080, + isPublic: true, + domain: "www.example.com", + protocol: "http" as const, + }; + + expect( + diffConfigs( + deployedConfig({ ports: [first, second] }), + deployedConfig({ ports: [second, first] }), + ), + ).toEqual([]); + }); + + it("reports changes to one of multiple domains on a shared port", () => { + const unchanged = { + port: 8080, + isPublic: true, + domain: "app.example.com", + protocol: "http" as const, + }; + + expect( + diffConfigs( + deployedConfig({ + ports: [unchanged, { ...unchanged, domain: "old.example.com" }], + }), + deployedConfig({ + ports: [{ ...unchanged, domain: "new.example.com" }, unchanged], + }), + ), + ).toContainEqual({ + field: "Port 8080", + from: "public, app.example.com; public, old.example.com", + to: "public, app.example.com; public, new.example.com", + }); + }); + it("reports serverless changes as pending config", () => { const changes = diffConfigs(deployedConfig(), { source: { type: "image", image: "nginx" }, From a9021d23979af1dd1292987b153c8e3d762db37d Mon Sep 17 00:00:00 2001 From: Amp Date: Tue, 28 Jul 2026 12:23:46 +0000 Subject: [PATCH 10/11] Simplify multiple-domain implementation Amp-Thread-ID: https://ampcode.com/threads/T-019fa6e6-5aa9-7137-a803-db807ff3b7aa Co-authored-by: Arjun Komath --- agent/internal/metrics/traefik.go | 18 ++------ agent/internal/routeowners/registry.go | 10 +---- agent/internal/routeowners/registry_test.go | 1 - agent/internal/traefik/l4.go | 21 +--------- agent/internal/traefik/routes.go | 6 +-- web/lib/public-api.ts | 46 ++++++++++----------- web/lib/service-config.ts | 20 ++++----- 7 files changed, 39 insertions(+), 83 deletions(-) diff --git a/agent/internal/metrics/traefik.go b/agent/internal/metrics/traefik.go index cb4382d4..c00385b3 100644 --- a/agent/internal/metrics/traefik.go +++ b/agent/internal/metrics/traefik.go @@ -3,8 +3,6 @@ package metrics import ( "bytes" "fmt" - "sort" - "strings" dto "github.com/prometheus/client_model/go" "github.com/prometheus/common/expfmt" @@ -21,6 +19,8 @@ func EnrichTraefik(data []byte, owners *routeowners.Registry) ([]byte, error) { if err != nil { return nil, fmt.Errorf("parse Prometheus metrics: %w", err) } + var output bytes.Buffer + encoder := expfmt.NewEncoder(&output, expfmt.NewFormat(expfmt.TypeTextPlain)) for _, family := range families { for _, metric := range family.Metric { labels := metric.Label[:0] @@ -40,20 +40,10 @@ func EnrichTraefik(data []byte, owners *routeowners.Registry) ([]byte, error) { name, value := "service_id", serviceID metric.Label = append(metric.Label, &dto.LabelPair{Name: &name, Value: &value}) } - sort.Slice(metric.Label, func(i, j int) bool { return metric.Label[i].GetName() < metric.Label[j].GetName() }) } - } - var output bytes.Buffer - encoder := expfmt.NewEncoder(&output, expfmt.NewFormat(expfmt.TypeTextPlain)) - names := make([]string, 0, len(families)) - for name := range families { - names = append(names, name) - } - sort.Strings(names) - for _, name := range names { - if err := encoder.Encode(families[name]); err != nil { + if err := encoder.Encode(family); err != nil { return nil, err } } - return []byte(strings.TrimSpace(output.String()) + "\n"), nil + return output.Bytes(), nil } diff --git a/agent/internal/routeowners/registry.go b/agent/internal/routeowners/registry.go index 7bbe80b2..9f04f9a7 100644 --- a/agent/internal/routeowners/registry.go +++ b/agent/internal/routeowners/registry.go @@ -16,24 +16,16 @@ func NewRegistry() *Registry { } func (r *Registry) Merge(owners map[string]string) { - if r == nil { - return - } r.mu.Lock() defer r.mu.Unlock() for resource, serviceID := range owners { if resource != "" && serviceID != "" { - if _, exists := r.owners[resource]; !exists { - r.owners[resource] = serviceID - } + r.owners[resource] = serviceID } } } func (r *Registry) Lookup(resource string) (string, bool) { - if r == nil { - return "", false - } resource = strings.TrimSuffix(resource, "@file") r.mu.RLock() defer r.mu.RUnlock() diff --git a/agent/internal/routeowners/registry_test.go b/agent/internal/routeowners/registry_test.go index 5082f282..13dcff4b 100644 --- a/agent/internal/routeowners/registry_test.go +++ b/agent/internal/routeowners/registry_test.go @@ -6,7 +6,6 @@ func TestRegistryRetainsOwnersAndHandlesFileSuffix(t *testing.T) { registry := NewRegistry() registry.Merge(map[string]string{"http-old": "service-old"}) registry.Merge(map[string]string{ - "http-old": "service-other", "http-new": "service-new", }) diff --git a/agent/internal/traefik/l4.go b/agent/internal/traefik/l4.go index 0a00dc32..5e94d5c8 100644 --- a/agent/internal/traefik/l4.go +++ b/agent/internal/traefik/l4.go @@ -120,25 +120,8 @@ func WriteRoutesConfig(compiled *RoutesConfig) error { return fmt.Errorf("failed to create dynamic config dir: %w", err) } routesPath := filepath.Join(dynamicConfigDir, routesFileName) - tmp, err := os.CreateTemp(dynamicConfigDir, routesFileName+".tmp-") - if err != nil { - return fmt.Errorf("failed to create temp config: %w", err) - } - tmpPath := tmp.Name() - defer os.Remove(tmpPath) - if err := tmp.Chmod(0644); err != nil { - tmp.Close() - return err - } - if _, err := tmp.Write(data); err != nil { - tmp.Close() - return fmt.Errorf("failed to write temp config: %w", err) - } - if err := tmp.Close(); err != nil { - return err - } - if err := os.Rename(tmpPath, routesPath); err != nil { - return fmt.Errorf("failed to rename config file: %w", err) + if err := atomicWrite(routesPath, data, 0644); err != nil { + return fmt.Errorf("failed to write routes config: %w", err) } log.Printf("[traefik] routes updated successfully") return nil diff --git a/agent/internal/traefik/routes.go b/agent/internal/traefik/routes.go index def9c94c..dd87006d 100644 --- a/agent/internal/traefik/routes.go +++ b/agent/internal/traefik/routes.go @@ -14,11 +14,7 @@ type RoutesConfig struct { } func resourceName(kind, serviceID, routeID string) string { - canonical, _ := json.Marshal(struct { - ServiceID string `json:"serviceId"` - RouteID string `json:"routeId"` - }{serviceID, routeID}) - hash := sha256.Sum256(canonical) + hash := sha256.Sum256([]byte(serviceID + "\x00" + routeID)) return kind + "-" + hex.EncodeToString(hash[:]) } diff --git a/web/lib/public-api.ts b/web/lib/public-api.ts index beb5011a..b193eac7 100644 --- a/web/lib/public-api.ts +++ b/web/lib/public-api.ts @@ -655,7 +655,7 @@ export async function patchConfiguration( } } - const update = db.transaction(async (tx) => { + return db.transaction(async (tx) => { await tx.execute( sql`SELECT pg_advisory_xact_lock(hashtext(${service.id}))`, ); @@ -990,16 +990,28 @@ export async function patchConfiguration( .delete(servicePorts) .where(eq(servicePorts.serviceId, service.id)); if (input.ports.length > 0) { - await tx.insert(servicePorts).values( - input.ports.map((port) => ({ - id: randomUUID(), - serviceId: service.id, - port: port.containerPort, - isPublic: port.public, - domain: port.public ? (port.domain ?? null) : null, - protocol: "http" as const, - })), - ); + try { + await tx.insert(servicePorts).values( + input.ports.map((port) => ({ + id: randomUUID(), + serviceId: service.id, + port: port.containerPort, + isPublic: port.public, + domain: port.public ? (port.domain ?? null) : null, + protocol: "http" as const, + })), + ); + } catch (error) { + if ( + (error as { code?: string; constraint?: string }).code === + "23505" && + (error as { constraint?: string }).constraint === + "service_ports_domain_unique" + ) { + domainError("Port domain is already in use", "DOMAIN_CONFLICT"); + } + throw error; + } } } } @@ -1009,16 +1021,4 @@ export async function patchConfiguration( changes, }; }); - try { - return await update; - } catch (error) { - if ( - (error as { code?: string; constraint?: string }).code === "23505" && - (error as { constraint?: string }).constraint === - "service_ports_domain_unique" - ) { - domainError("Port domain is already in use", "DOMAIN_CONFLICT"); - } - throw error; - } } diff --git a/web/lib/service-config.ts b/web/lib/service-config.ts index f7f30dd8..5692fd72 100644 --- a/web/lib/service-config.ts +++ b/web/lib/service-config.ts @@ -538,8 +538,14 @@ export function diffConfigs( }); } - const deployedPortsByNumber = groupPortsByNumber(deployed.ports || []); - const currentPortsByNumber = groupPortsByNumber(current.ports || []); + const deployedPortsByNumber = Map.groupBy( + deployed.ports || [], + (port) => port.port, + ); + const currentPortsByNumber = Map.groupBy( + current.ports || [], + (port) => port.port, + ); const portNumbers = Array.from( new Set([...deployedPortsByNumber.keys(), ...currentPortsByNumber.keys()]), ).sort((a, b) => a - b); @@ -630,16 +636,6 @@ export function diffConfigs( return changes; } -function groupPortsByNumber(ports: PortConfig[]): Map { - const grouped = new Map(); - for (const port of ports) { - const entries = grouped.get(port.port) || []; - entries.push(port); - grouped.set(port.port, entries); - } - return grouped; -} - function describePorts(ports: PortConfig[]): string { return ports .map((port) => { From 8ca9bbf94561d9f9284f3bd3b162efa59e405171 Mon Sep 17 00:00:00 2001 From: Amp Date: Tue, 28 Jul 2026 12:57:26 +0000 Subject: [PATCH 11/11] Preserve client browser compatibility Amp-Thread-ID: https://ampcode.com/threads/T-019fa6e6-5aa9-7137-a803-db807ff3b7aa Co-authored-by: Arjun Komath --- web/lib/service-config.ts | 20 ++++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/web/lib/service-config.ts b/web/lib/service-config.ts index 5692fd72..f7f30dd8 100644 --- a/web/lib/service-config.ts +++ b/web/lib/service-config.ts @@ -538,14 +538,8 @@ export function diffConfigs( }); } - const deployedPortsByNumber = Map.groupBy( - deployed.ports || [], - (port) => port.port, - ); - const currentPortsByNumber = Map.groupBy( - current.ports || [], - (port) => port.port, - ); + const deployedPortsByNumber = groupPortsByNumber(deployed.ports || []); + const currentPortsByNumber = groupPortsByNumber(current.ports || []); const portNumbers = Array.from( new Set([...deployedPortsByNumber.keys(), ...currentPortsByNumber.keys()]), ).sort((a, b) => a - b); @@ -636,6 +630,16 @@ export function diffConfigs( return changes; } +function groupPortsByNumber(ports: PortConfig[]): Map { + const grouped = new Map(); + for (const port of ports) { + const entries = grouped.get(port.port) || []; + entries.push(port); + grouped.set(port.port, entries); + } + return grouped; +} + function describePorts(ports: PortConfig[]): string { return ports .map((port) => {