From a9daec7e663b8437b757511f7e86c61beda7895b Mon Sep 17 00:00:00 2001 From: Techulus Agent <291950465+techulus-agent@users.noreply.github.com> Date: Tue, 7 Jul 2026 23:50:57 +1000 Subject: [PATCH 01/11] Improve serverless wake readiness logs --- agent/internal/agent/run.go | 5 +- agent/internal/serverless/gateway.go | 309 ++++++++++++++++++++-- agent/internal/serverless/gateway_test.go | 62 ++++- 3 files changed, 347 insertions(+), 29 deletions(-) diff --git a/agent/internal/agent/run.go b/agent/internal/agent/run.go index ac10ee3c..220a2982 100644 --- a/agent/internal/agent/run.go +++ b/agent/internal/agent/run.go @@ -117,13 +117,14 @@ func (a *Agent) RequestStatusReport(reason string) { } func (a *Agent) reportStatus(reason string) { + startedAt := time.Now() report := a.BuildStatusReport(true) reportedDeploymentErrorCount := len(report.DeploymentErrors) completed, active := a.SnapshotWorkStatus() serverlessTransitions := a.SnapshotServerlessTransitions() response, err := a.Client.ReportStatus(report, completed, active, serverlessTransitions) if err != nil { - log.Printf("[status] failed to report (%s): %v", reason, err) + log.Printf("[status] failed to report (%s) latency=%s: %v", reason, time.Since(startedAt).Round(time.Millisecond), err) return } a.ClearReportedDeploymentErrors(reportedDeploymentErrorCount) @@ -131,7 +132,7 @@ func (a *Agent) reportStatus(reason string) { a.AcknowledgeWorkResults(response.AcceptedWorkItemResults, response.RejectedWorkItemResults) a.LogRejectedActiveWorkItems(response.RejectedActiveWorkItems) a.AcceptLeasedWorkItems(response.WorkItems) - log.Printf("[status] reported (%s)", reason) + log.Printf("[status] reported (%s) latency=%s", reason, time.Since(startedAt).Round(time.Millisecond)) } func nextStatusReportDelay() time.Duration { diff --git a/agent/internal/serverless/gateway.go b/agent/internal/serverless/gateway.go index c7002585..5c449520 100644 --- a/agent/internal/serverless/gateway.go +++ b/agent/internal/serverless/gateway.go @@ -28,6 +28,7 @@ var ( wakePollInterval = 500 * time.Millisecond idleTimerSeedInterval = 15 * time.Second upstreamDialTimeout = 250 * time.Millisecond + wakeWaitLogInterval = 2 * time.Second checkUpstreamReady = tcpUpstreamReady ) @@ -70,6 +71,30 @@ type activityState struct { sleepTimer *time.Timer } +type upstreamReadiness struct { + ready bool + latency time.Duration + err error +} + +type upstreamResolution struct { + ready []agenthttp.ServerlessUpstream + sleepingLocalIDs []string + waiting []upstreamWaitReason +} + +type upstreamWaitReason struct { + deploymentID string + serviceID string + containerID string + upstreamURL string + reason string + state string + health string + dialLatency time.Duration + err error +} + func NewGateway(runtime Runtime) *Gateway { return &Gateway{ runtime: runtime, @@ -267,9 +292,17 @@ func (g *Gateway) resolveUpstreams(host string) ([]agenthttp.ServerlessUpstream, } func (g *Gateway) readyUpstreams(route *agenthttp.ServerlessRoute, state *agenthttp.ExpectedState) ([]agenthttp.ServerlessUpstream, []string, error) { + resolution, err := g.inspectUpstreams(route, state) + if err != nil { + return nil, nil, err + } + return resolution.ready, resolution.sleepingLocalIDs, nil +} + +func (g *Gateway) inspectUpstreams(route *agenthttp.ServerlessRoute, state *agenthttp.ExpectedState) (upstreamResolution, error) { actualContainers, err := g.runtime.ListServerlessContainers() if err != nil { - return nil, nil, fmt.Errorf("failed to list local containers: %w", err) + return upstreamResolution{}, fmt.Errorf("failed to list local containers: %w", err) } expectedByDeploymentID := expectedContainersByDeploymentID(state) @@ -287,24 +320,67 @@ func (g *Gateway) readyUpstreams(route *agenthttp.ServerlessRoute, state *agenth upstreamsByURL[upstream.Url] = upstream } - sleepingLocalIDs := []string{} + resolution := upstreamResolution{} for _, deploymentID := range route.LocalDeploymentIDs { expected, ok := expectedByDeploymentID[deploymentID] if !ok { + resolution.waiting = append(resolution.waiting, upstreamWaitReason{ + deploymentID: deploymentID, + serviceID: route.ServiceID, + reason: "missing_expected_state", + }) continue } - actual, isRunning := actualByDeploymentID[deploymentID] - if isRunning && g.isContainerReady(actual, expected) { + actual, exists := actualByDeploymentID[deploymentID] + if exists && actual.State == "running" { + ready, waitReason := g.containerReady(actual, expected) + if !ready { + resolution.waiting = append(resolution.waiting, waitReason) + continue + } if upstream, ok := localUpstream(route, expected); ok { - if checkUpstreamReady(upstream.Url) { + readiness := checkUpstreamReady(upstream.Url) + if readiness.ready { upstreamsByURL[upstream.Url] = upstream + } else { + resolution.waiting = append(resolution.waiting, upstreamWaitReason{ + deploymentID: expected.DeploymentID, + serviceID: expected.ServiceID, + containerID: actual.ID, + upstreamURL: upstream.Url, + reason: "upstream_unreachable", + state: actual.State, + dialLatency: readiness.latency, + err: readiness.err, + }) } + } else { + resolution.waiting = append(resolution.waiting, upstreamWaitReason{ + deploymentID: expected.DeploymentID, + serviceID: expected.ServiceID, + containerID: actual.ID, + reason: "missing_local_upstream", + state: actual.State, + }) } continue } if expected.DesiredState == "stopped" || g.runtime.HasPendingServerlessSleep(deploymentID) { - sleepingLocalIDs = append(sleepingLocalIDs, deploymentID) + resolution.sleepingLocalIDs = append(resolution.sleepingLocalIDs, deploymentID) + resolution.waiting = append(resolution.waiting, upstreamWaitReason{ + deploymentID: expected.DeploymentID, + serviceID: expected.ServiceID, + reason: "sleeping", + state: containerState(actual, exists), + }) + continue } + resolution.waiting = append(resolution.waiting, upstreamWaitReason{ + deploymentID: expected.DeploymentID, + serviceID: expected.ServiceID, + reason: "container_not_running", + state: containerState(actual, exists), + }) } upstreams := make([]agenthttp.ServerlessUpstream, 0, len(upstreamsByURL)) @@ -317,7 +393,8 @@ func (g *Gateway) readyUpstreams(route *agenthttp.ServerlessRoute, state *agenth upstreams = append(upstreams, upstream) } sortUpstreams(upstreams) - return upstreams, sleepingLocalIDs, nil + resolution.ready = upstreams + return resolution, nil } func (g *Gateway) wakeLocalDeployments(route *agenthttp.ServerlessRoute, state *agenthttp.ExpectedState, deploymentIDs []string) []string { @@ -377,45 +454,61 @@ func (g *Gateway) wakeLocalDeployments(route *agenthttp.ServerlessRoute, state * func (g *Gateway) waitForReadyUpstreams(route *agenthttp.ServerlessRoute, wakeTimeoutSeconds int, startedAt time.Time, wokenDeploymentIDs []string) ([]agenthttp.ServerlessUpstream, error) { timeout := wakeTimeout(wakeTimeoutSeconds) deadline := startedAt.Add(timeout) + attempt := 0 + nextWaitLogAt := time.Time{} + lastWaitKey := "" for { + attempt++ state := g.runtime.ExpectedState() - ready, _, err := g.readyUpstreams(route, state) + resolution, err := g.inspectUpstreams(route, state) if err != nil { return nil, err } + ready := resolution.ready if len(ready) > 0 { pendingIDs := pendingWakeDeploymentIDs(wokenDeploymentIDs, ready) if len(pendingIDs) > 0 { go g.waitForWokenDeployments(route, route.WakeTimeoutSeconds, startedAt, pendingIDs) } log.Printf( - "[serverless-gateway] wake ready host=%s upstreams=%d latency=%s", + "[serverless-gateway] wake ready host=%s upstreams=%d attempts=%d latency=%s", route.Domain, len(ready), + attempt, roundDuration(time.Since(startedAt)), ) return ready, nil } + waitSummary := summarizeWaitReasons(resolution.waiting) + waitKey := summarizeWaitReasonKeys(resolution.waiting) if time.Now().After(deadline) { pendingIDs := pendingWakeDeploymentIDs(wokenDeploymentIDs, ready) - g.queueWakeTimeouts(route, pendingIDs, startedAt) + g.queueWakeTimeouts(route, pendingIDs, startedAt, waitSummary) if len(ready) > 0 { log.Printf( - "[serverless-gateway] wake partially ready host=%s upstreams=%d latency=%s", + "[serverless-gateway] wake partially ready host=%s upstreams=%d attempts=%d latency=%s", route.Domain, len(ready), + attempt, roundDuration(time.Since(startedAt)), ) return ready, nil } log.Printf( - "[serverless-gateway] wake timed out host=%s latency=%s", + "[serverless-gateway] wake timed out host=%s attempts=%d latency=%s last_wait=%q", route.Domain, + attempt, roundDuration(time.Since(startedAt)), + waitSummary, ) return nil, fmt.Errorf("timed out waiting for local serverless wake") } + if shouldLogWakeWait(nextWaitLogAt, lastWaitKey, waitKey) { + logWakeWait(route, resolution.waiting, attempt, time.Since(startedAt)) + nextWaitLogAt = time.Now().Add(wakeWaitLogInterval) + lastWaitKey = waitKey + } time.Sleep(wakePollInterval) } } @@ -423,28 +516,41 @@ func (g *Gateway) waitForReadyUpstreams(route *agenthttp.ServerlessRoute, wakeTi func (g *Gateway) waitForWokenDeployments(route *agenthttp.ServerlessRoute, wakeTimeoutSeconds int, startedAt time.Time, wokenDeploymentIDs []string) { timeout := wakeTimeout(wakeTimeoutSeconds) deadline := startedAt.Add(timeout) + attempt := 0 + nextWaitLogAt := time.Time{} + lastWaitKey := "" for { + attempt++ state := g.runtime.ExpectedState() - ready, _, err := g.readyUpstreams(route, state) + resolution, err := g.inspectUpstreams(route, state) if err != nil { log.Printf("[serverless-gateway] wake monitor failed host=%s error=%v", route.Domain, err) return } + ready := resolution.ready pendingIDs := pendingWakeDeploymentIDs(wokenDeploymentIDs, ready) if len(pendingIDs) == 0 { log.Printf( - "[serverless-gateway] wake ready host=%s deployments=%d latency=%s", + "[serverless-gateway] wake ready host=%s deployments=%d attempts=%d latency=%s", route.Domain, len(wokenDeploymentIDs), + attempt, roundDuration(time.Since(startedAt)), ) return } + waitSummary := summarizeWaitReasons(resolution.waiting) + waitKey := summarizeWaitReasonKeys(resolution.waiting) if time.Now().After(deadline) { - g.queueWakeTimeouts(route, pendingIDs, startedAt) + g.queueWakeTimeouts(route, pendingIDs, startedAt, waitSummary) return } + if shouldLogWakeWait(nextWaitLogAt, lastWaitKey, waitKey) { + logWakeWait(route, filterWaitReasons(resolution.waiting, pendingIDs), attempt, time.Since(startedAt)) + nextWaitLogAt = time.Now().Add(wakeWaitLogInterval) + lastWaitKey = waitKey + } time.Sleep(wakePollInterval) } } @@ -466,15 +572,134 @@ func pendingWakeDeploymentIDs(wokenDeploymentIDs []string, ready []agenthttp.Ser return pendingIDs } -func (g *Gateway) queueWakeTimeouts(route *agenthttp.ServerlessRoute, deploymentIDs []string, startedAt time.Time) { +func shouldLogWakeWait(nextLogAt time.Time, lastSummary string, currentSummary string) bool { + if currentSummary == "" { + return false + } + if currentSummary != lastSummary { + return true + } + return nextLogAt.IsZero() || time.Now().After(nextLogAt) +} + +func logWakeWait(route *agenthttp.ServerlessRoute, reasons []upstreamWaitReason, attempt int, elapsed time.Duration) { + if len(reasons) == 0 { + log.Printf( + "[serverless-gateway] wake waiting host=%s attempt=%d elapsed=%s reason=no_ready_upstreams", + route.Domain, + attempt, + roundDuration(elapsed), + ) + return + } + for _, reason := range reasons { + fields := []string{ + fmt.Sprintf("host=%s", route.Domain), + fmt.Sprintf("deployment=%s", reason.deploymentID), + fmt.Sprintf("service=%s", reason.serviceID), + fmt.Sprintf("attempt=%d", attempt), + fmt.Sprintf("elapsed=%s", roundDuration(elapsed)), + fmt.Sprintf("reason=%s", reason.reason), + } + if reason.containerID != "" { + fields = append(fields, fmt.Sprintf("container=%s", reason.containerID)) + } + if reason.upstreamURL != "" { + fields = append(fields, fmt.Sprintf("upstream=%s", reason.upstreamURL)) + } + if reason.state != "" { + fields = append(fields, fmt.Sprintf("state=%s", reason.state)) + } + if reason.health != "" { + fields = append(fields, fmt.Sprintf("health=%s", reason.health)) + } + if reason.dialLatency > 0 { + fields = append(fields, fmt.Sprintf("dial_latency=%s", roundDuration(reason.dialLatency))) + } + if reason.err != nil { + fields = append(fields, fmt.Sprintf("error=%q", compactError(reason.err))) + } + log.Printf("[serverless-gateway] wake waiting %s", strings.Join(fields, " ")) + } +} + +func filterWaitReasons(reasons []upstreamWaitReason, deploymentIDs []string) []upstreamWaitReason { + if len(reasons) == 0 || len(deploymentIDs) == 0 { + return nil + } + pending := map[string]struct{}{} + for _, deploymentID := range deploymentIDs { + pending[deploymentID] = struct{}{} + } + filtered := []upstreamWaitReason{} + for _, reason := range reasons { + if _, ok := pending[reason.deploymentID]; ok { + filtered = append(filtered, reason) + } + } + return filtered +} + +func summarizeWaitReasons(reasons []upstreamWaitReason) string { + return formatWaitReasons(reasons, true) +} + +func summarizeWaitReasonKeys(reasons []upstreamWaitReason) string { + return formatWaitReasons(reasons, false) +} + +func formatWaitReasons(reasons []upstreamWaitReason, includeDialLatency bool) string { + if len(reasons) == 0 { + return "" + } + parts := make([]string, 0, len(reasons)) + for _, reason := range reasons { + part := fmt.Sprintf("deployment=%s reason=%s", reason.deploymentID, reason.reason) + if reason.upstreamURL != "" { + part += fmt.Sprintf(" upstream=%s", reason.upstreamURL) + } + if reason.state != "" { + part += fmt.Sprintf(" state=%s", reason.state) + } + if reason.health != "" { + part += fmt.Sprintf(" health=%s", reason.health) + } + if includeDialLatency && reason.dialLatency > 0 { + part += fmt.Sprintf(" dial_latency=%s", roundDuration(reason.dialLatency)) + } + if reason.err != nil { + part += fmt.Sprintf(" error=%q", compactError(reason.err)) + } + parts = append(parts, part) + } + sort.Strings(parts) + return strings.Join(parts, "; ") +} + +func compactError(err error) string { + if err == nil { + return "" + } + message := strings.TrimSpace(err.Error()) + if len(message) <= 240 { + return message + } + return message[:240] + "..." +} + +func (g *Gateway) queueWakeTimeouts(route *agenthttp.ServerlessRoute, deploymentIDs []string, startedAt time.Time, waitSummary string) { for _, deploymentID := range deploymentIDs { errMessage := fmt.Sprintf("timed out waiting %s for local serverless wake", roundDuration(time.Since(startedAt))) + if waitSummary != "" { + errMessage = fmt.Sprintf("%s; last wait: %s", errMessage, waitSummary) + } log.Printf( - "[serverless-gateway] wake timed out host=%s deployment=%s service=%s latency=%s", + "[serverless-gateway] wake timed out host=%s deployment=%s service=%s latency=%s last_wait=%q", route.Domain, deploymentID, route.ServiceID, roundDuration(time.Since(startedAt)), + waitSummary, ) g.runtime.QueueServerlessTransition(agenthttp.ServerlessTransition{ Type: "wake_failed", @@ -484,24 +709,47 @@ func (g *Gateway) queueWakeTimeouts(route *agenthttp.ServerlessRoute, deployment } } -func (g *Gateway) isContainerReady(actual container.Container, expected agenthttp.ExpectedContainer) bool { +func (g *Gateway) containerReady(actual container.Container, expected agenthttp.ExpectedContainer) (bool, upstreamWaitReason) { if actual.State != "running" { - return false + return false, upstreamWaitReason{ + deploymentID: expected.DeploymentID, + serviceID: expected.ServiceID, + containerID: actual.ID, + reason: "container_not_running", + state: actual.State, + } } if expected.HealthCheck == nil || expected.HealthCheck.Cmd == "" { - return true + return true, upstreamWaitReason{} } health := g.runtime.GetServerlessContainerHealth(actual.ID) - return health == "healthy" || health == "none" + if health == "healthy" || health == "none" { + return true, upstreamWaitReason{} + } + return false, upstreamWaitReason{ + deploymentID: expected.DeploymentID, + serviceID: expected.ServiceID, + containerID: actual.ID, + reason: "health_not_ready", + state: actual.State, + health: health, + } } -func tcpUpstreamReady(address string) bool { +func tcpUpstreamReady(address string) upstreamReadiness { + startedAt := time.Now() conn, err := net.DialTimeout("tcp", address, upstreamDialTimeout) if err != nil { - return false + return upstreamReadiness{ + latency: time.Since(startedAt), + err: err, + } } conn.Close() - return true + return upstreamReadiness{ + ready: true, + latency: time.Since(startedAt), + } } func (g *Gateway) cachedUpstreams(host string) ([]agenthttp.ServerlessUpstream, bool) { @@ -725,7 +973,8 @@ func (g *Gateway) sleepService(serviceID string) { ) continue } - if expected.DesiredState != "stopped" && !g.isContainerReady(actual, expected) { + containerReady, _ := g.containerReady(actual, expected) + if expected.DesiredState != "stopped" && !containerReady { log.Printf( "[serverless-gateway] sleep skipped service=%s deployment=%s reason=container_not_ready", serviceID, @@ -888,6 +1137,16 @@ func actualContainersByDeploymentID(containers []container.Container) map[string return containersByDeploymentID } +func containerState(actual container.Container, exists bool) string { + if !exists { + return "missing" + } + if actual.State == "" { + return "unknown" + } + return actual.State +} + func localUpstream(route *agenthttp.ServerlessRoute, expected agenthttp.ExpectedContainer) (agenthttp.ServerlessUpstream, bool) { if expected.IPAddress != "" { return agenthttp.ServerlessUpstream{ diff --git a/agent/internal/serverless/gateway_test.go b/agent/internal/serverless/gateway_test.go index ad1e3dec..79c4678c 100644 --- a/agent/internal/serverless/gateway_test.go +++ b/agent/internal/serverless/gateway_test.go @@ -1,8 +1,11 @@ package serverless import ( + "bytes" "context" "errors" + "log" + "strings" "sync" "testing" "time" @@ -12,7 +15,7 @@ import ( ) func init() { - checkUpstreamReady = func(string) bool { return true } + checkUpstreamReady = func(string) upstreamReadiness { return upstreamReadiness{ready: true} } } type fakeRuntime struct { @@ -431,7 +434,9 @@ func TestWakeTimeoutWhenLocalPortNeverOpens(t *testing.T) { previousPoll := wakePollInterval previousReady := checkUpstreamReady wakePollInterval = time.Millisecond - checkUpstreamReady = func(string) bool { return false } + checkUpstreamReady = func(string) upstreamReadiness { + return upstreamReadiness{err: errors.New("connection refused")} + } t.Cleanup(func() { wakePollInterval = previousPoll checkUpstreamReady = previousReady @@ -460,6 +465,59 @@ func TestWakeTimeoutWhenLocalPortNeverOpens(t *testing.T) { } } +func TestWakeWaitingLogsReadinessReason(t *testing.T) { + previousPoll := wakePollInterval + previousLogInterval := wakeWaitLogInterval + previousReady := checkUpstreamReady + wakePollInterval = time.Millisecond + wakeWaitLogInterval = time.Hour + checkUpstreamReady = func(string) upstreamReadiness { + return upstreamReadiness{ + latency: time.Millisecond, + err: errors.New("connection refused"), + } + } + t.Cleanup(func() { + wakePollInterval = previousPoll + wakeWaitLogInterval = previousLogInterval + checkUpstreamReady = previousReady + }) + + var logs bytes.Buffer + previousOutput := log.Writer() + previousFlags := log.Flags() + log.SetOutput(&logs) + log.SetFlags(0) + t.Cleanup(func() { + log.SetOutput(previousOutput) + log.SetFlags(previousFlags) + }) + + state := testExpectedState("stopped") + state.Serverless.Routes[0].Upstreams = nil + state.Serverless.Routes[0].WakeTimeoutSeconds = 1 + runtime := &fakeRuntime{state: state} + gateway := NewGateway(runtime) + + _, err := gateway.getUpstreams(context.Background(), "app.example.com") + if err == nil { + t.Fatal("getUpstreams succeeded, want timeout error") + } + + output := logs.String() + for _, want := range []string{ + "wake waiting", + "host=app.example.com", + "deployment=dep_local", + "reason=upstream_unreachable", + "error=\"connection refused\"", + } { + if !strings.Contains(output, want) { + t.Fatalf("log output missing %q:\n%s", want, output) + } + } +} + func testExpectedState(localDesiredState string) *agenthttp.ExpectedState { state := &agenthttp.ExpectedState{ Containers: []agenthttp.ExpectedContainer{ From 85a77a465ea67f5d2c1773e9f803cefc396a33d8 Mon Sep 17 00:00:00 2001 From: Techulus Agent <291950465+techulus-agent@users.noreply.github.com> Date: Wed, 8 Jul 2026 09:23:51 +1000 Subject: [PATCH 02/11] Use loopback ports for serverless wake routing --- agent/internal/container/runtime.go | 85 +++++++++++++---------- agent/internal/container/runtime_test.go | 45 ++++++++++++ agent/internal/container/types.go | 31 +++++---- agent/internal/http/client.go | 1 + agent/internal/reconcile/reconcile.go | 29 ++++---- agent/internal/serverless/gateway.go | 63 +++++++++++------ agent/internal/serverless/gateway_test.go | 79 +++++++++++++++++++++ web/lib/agent/expected-state.ts | 2 + web/tests/expected-state.test.ts | 4 ++ 9 files changed, 252 insertions(+), 87 deletions(-) create mode 100644 agent/internal/container/runtime_test.go diff --git a/agent/internal/container/runtime.go b/agent/internal/container/runtime.go index 912e694f..660b66b6 100644 --- a/agent/internal/container/runtime.go +++ b/agent/internal/container/runtime.go @@ -91,6 +91,47 @@ func Deploy(config *DeployConfig) (*DeployResult, error) { logFunc("stdout", fmt.Sprintf("Created volume directory: %s", vm.HostPath)) } + args := buildPodmanRunArgs(config, image) + + logFunc("stdout", fmt.Sprintf("Starting container: %s", config.Name)) + + runCmd := exec.Command("podman", args...) + output, err := runCmd.CombinedOutput() + if err != nil { + logFunc("stderr", fmt.Sprintf("Start failed: %s", string(output))) + return nil, fmt.Errorf("failed to run container: %s: %w", string(output), err) + } + + containerID := strings.TrimSpace(string(output)) + logFunc("stdout", fmt.Sprintf("Container started: %s", containerID)) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + logFunc("stdout", "Verifying container is running...") + err = retry.WithBackoff(ctx, retry.DeployBackoff, func() (bool, error) { + running, err := IsContainerRunning(containerID) + if err != nil { + return false, err + } + return running, nil + }) + + if err != nil { + logsCmd := exec.Command("podman", "logs", "--tail", "50", containerID) + logsOutput, _ := logsCmd.CombinedOutput() + logFunc("stderr", fmt.Sprintf("Container failed to stay running. Logs:\n%s", string(logsOutput))) + return nil, fmt.Errorf("container failed to stay running after start: %w", err) + } + + logFunc("stdout", "Container verified running") + + return &DeployResult{ + ContainerID: containerID, + }, nil +} + +func buildPodmanRunArgs(config *DeployConfig, image string) []string { args := []string{ "run", "-d", "--name", config.Name, @@ -117,6 +158,12 @@ func Deploy(config *DeployConfig) (*DeployResult, error) { if config.IPAddress != "" { args = append(args, "--network", NetworkName, "--ip", config.IPAddress) + if config.PublishLocalPorts { + for _, pm := range config.PortMappings { + portMapping := fmt.Sprintf("127.0.0.1:%d:%d", pm.HostPort, pm.ContainerPort) + args = append(args, "-p", portMapping) + } + } } else { for _, pm := range config.PortMappings { portMapping := fmt.Sprintf("%s:%d:%d", config.WireGuardIP, pm.HostPort, pm.ContainerPort) @@ -154,43 +201,7 @@ func Deploy(config *DeployConfig) (*DeployResult, error) { } else { args = append(args, image) } - - logFunc("stdout", fmt.Sprintf("Starting container: %s", config.Name)) - - runCmd := exec.Command("podman", args...) - output, err := runCmd.CombinedOutput() - if err != nil { - logFunc("stderr", fmt.Sprintf("Start failed: %s", string(output))) - return nil, fmt.Errorf("failed to run container: %s: %w", string(output), err) - } - - containerID := strings.TrimSpace(string(output)) - logFunc("stdout", fmt.Sprintf("Container started: %s", containerID)) - - ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) - defer cancel() - - logFunc("stdout", "Verifying container is running...") - err = retry.WithBackoff(ctx, retry.DeployBackoff, func() (bool, error) { - running, err := IsContainerRunning(containerID) - if err != nil { - return false, err - } - return running, nil - }) - - if err != nil { - logsCmd := exec.Command("podman", "logs", "--tail", "50", containerID) - logsOutput, _ := logsCmd.CombinedOutput() - logFunc("stderr", fmt.Sprintf("Container failed to stay running. Logs:\n%s", string(logsOutput))) - return nil, fmt.Errorf("container failed to stay running after start: %w", err) - } - - logFunc("stdout", "Container verified running") - - return &DeployResult{ - ContainerID: containerID, - }, nil + return args } func Stop(containerID string) error { diff --git a/agent/internal/container/runtime_test.go b/agent/internal/container/runtime_test.go new file mode 100644 index 00000000..790d75fd --- /dev/null +++ b/agent/internal/container/runtime_test.go @@ -0,0 +1,45 @@ +package container + +import ( + "slices" + "testing" +) + +func TestBuildPodmanRunArgsPublishesLoopbackPortsWithStaticIP(t *testing.T) { + args := buildPodmanRunArgs(&DeployConfig{ + Name: "svc-dep", + Image: "docker.io/library/nginx:latest", + ServiceID: "svc", + ServiceName: "api", + DeploymentID: "dep", + IPAddress: "10.200.1.2", + PublishLocalPorts: true, + PortMappings: []PortMapping{ + {ContainerPort: 80, HostPort: 30080}, + }, + }, "docker.io/library/nginx:latest") + + for _, want := range []string{"--network", NetworkName, "--ip", "10.200.1.2", "-p", "127.0.0.1:30080:80"} { + if !slices.Contains(args, want) { + t.Fatalf("args missing %q: %+v", want, args) + } + } +} + +func TestBuildPodmanRunArgsDoesNotPublishStaticIPPortsByDefault(t *testing.T) { + args := buildPodmanRunArgs(&DeployConfig{ + Name: "svc-dep", + Image: "docker.io/library/nginx:latest", + ServiceID: "svc", + ServiceName: "api", + DeploymentID: "dep", + IPAddress: "10.200.1.2", + PortMappings: []PortMapping{ + {ContainerPort: 80, HostPort: 30080}, + }, + }, "docker.io/library/nginx:latest") + + if slices.Contains(args, "-p") { + t.Fatalf("args unexpectedly publish ports: %+v", args) + } +} diff --git a/agent/internal/container/types.go b/agent/internal/container/types.go index 53f1a03c..f0571589 100644 --- a/agent/internal/container/types.go +++ b/agent/internal/container/types.go @@ -26,21 +26,22 @@ type VolumeMount struct { type BuildLogFunc func(stream string, message string) type DeployConfig struct { - Name string - Image string - ServiceID string - ServiceName string - DeploymentID string - WireGuardIP string - IPAddress string - PortMappings []PortMapping - HealthCheck *HealthCheck - Env map[string]string - VolumeMounts []VolumeMount - StartCommand string - CPULimit *float64 - MemoryLimitMb *int - LogFunc BuildLogFunc + Name string + Image string + ServiceID string + ServiceName string + DeploymentID string + WireGuardIP string + IPAddress string + PortMappings []PortMapping + PublishLocalPorts bool + HealthCheck *HealthCheck + Env map[string]string + VolumeMounts []VolumeMount + StartCommand string + CPULimit *float64 + MemoryLimitMb *int + LogFunc BuildLogFunc } type DeployResult struct { diff --git a/agent/internal/http/client.go b/agent/internal/http/client.go index 8ec46747..22ee9dd4 100644 --- a/agent/internal/http/client.go +++ b/agent/internal/http/client.go @@ -77,6 +77,7 @@ type ExpectedContainer struct { Image string `json:"image"` IPAddress string `json:"ipAddress"` Ports []PortMapping `json:"ports"` + PublishLocalPorts bool `json:"publishLocalPorts"` Env map[string]string `json:"env"` StartCommand string `json:"startCommand"` HealthCheck *HealthCheck `json:"healthCheck"` diff --git a/agent/internal/reconcile/reconcile.go b/agent/internal/reconcile/reconcile.go index db9e159e..65635afc 100644 --- a/agent/internal/reconcile/reconcile.go +++ b/agent/internal/reconcile/reconcile.go @@ -64,20 +64,21 @@ func (r *Reconciler) Deploy(exp agenthttp.ExpectedContainer) error { } _, err := container.Deploy(&container.DeployConfig{ - Name: exp.Name, - Image: exp.Image, - ServiceID: exp.ServiceID, - ServiceName: exp.ServiceName, - DeploymentID: exp.DeploymentID, - IPAddress: exp.IPAddress, - PortMappings: portMappings, - HealthCheck: healthCheck, - Env: decryptedEnv, - VolumeMounts: volumeMounts, - StartCommand: exp.StartCommand, - CPULimit: exp.ResourceCPULimit, - MemoryLimitMb: exp.ResourceMemoryLimitMb, - LogFunc: func(stream, message string) { log.Printf("[deploy:%s] %s", stream, message) }, + Name: exp.Name, + Image: exp.Image, + ServiceID: exp.ServiceID, + ServiceName: exp.ServiceName, + DeploymentID: exp.DeploymentID, + IPAddress: exp.IPAddress, + PortMappings: portMappings, + PublishLocalPorts: exp.PublishLocalPorts, + HealthCheck: healthCheck, + Env: decryptedEnv, + VolumeMounts: volumeMounts, + StartCommand: exp.StartCommand, + CPULimit: exp.ResourceCPULimit, + MemoryLimitMb: exp.ResourceMemoryLimitMb, + LogFunc: func(stream, message string) { log.Printf("[deploy:%s] %s", stream, message) }, }) return err diff --git a/agent/internal/serverless/gateway.go b/agent/internal/serverless/gateway.go index 5c449520..6c6d447b 100644 --- a/agent/internal/serverless/gateway.go +++ b/agent/internal/serverless/gateway.go @@ -338,21 +338,31 @@ func (g *Gateway) inspectUpstreams(route *agenthttp.ServerlessRoute, state *agen resolution.waiting = append(resolution.waiting, waitReason) continue } - if upstream, ok := localUpstream(route, expected); ok { - readiness := checkUpstreamReady(upstream.Url) - if readiness.ready { - upstreamsByURL[upstream.Url] = upstream - } else { - resolution.waiting = append(resolution.waiting, upstreamWaitReason{ - deploymentID: expected.DeploymentID, - serviceID: expected.ServiceID, - containerID: actual.ID, - upstreamURL: upstream.Url, - reason: "upstream_unreachable", - state: actual.State, - dialLatency: readiness.latency, - err: readiness.err, - }) + upstreams := localUpstreamCandidates(route, expected) + if len(upstreams) > 0 { + upstreamReady := false + waitReasons := make([]upstreamWaitReason, 0, len(upstreams)) + for _, upstream := range upstreams { + readiness := checkUpstreamReady(upstream.Url) + if readiness.ready { + upstreamsByURL[upstream.Url] = upstream + upstreamReady = true + break + } else { + waitReasons = append(waitReasons, upstreamWaitReason{ + deploymentID: expected.DeploymentID, + serviceID: expected.ServiceID, + containerID: actual.ID, + upstreamURL: upstream.Url, + reason: "upstream_unreachable", + state: actual.State, + dialLatency: readiness.latency, + err: readiness.err, + }) + } + } + if !upstreamReady { + resolution.waiting = append(resolution.waiting, waitReasons...) } } else { resolution.waiting = append(resolution.waiting, upstreamWaitReason{ @@ -1147,25 +1157,36 @@ func containerState(actual container.Container, exists bool) string { return actual.State } -func localUpstream(route *agenthttp.ServerlessRoute, expected agenthttp.ExpectedContainer) (agenthttp.ServerlessUpstream, bool) { +func localUpstreamCandidates(route *agenthttp.ServerlessRoute, expected agenthttp.ExpectedContainer) []agenthttp.ServerlessUpstream { + upstreams := []agenthttp.ServerlessUpstream{} + if expected.PublishLocalPorts { + upstreams = append(upstreams, localLoopbackUpstreams(route, expected)...) + } if expected.IPAddress != "" { - return agenthttp.ServerlessUpstream{ + upstreams = append(upstreams, agenthttp.ServerlessUpstream{ DeploymentID: expected.DeploymentID, Url: fmt.Sprintf("%s:%d", expected.IPAddress, route.Port), Local: true, - }, true + }) } + if expected.IPAddress == "" && !expected.PublishLocalPorts { + upstreams = append(upstreams, localLoopbackUpstreams(route, expected)...) + } + return upstreams +} +func localLoopbackUpstreams(route *agenthttp.ServerlessRoute, expected agenthttp.ExpectedContainer) []agenthttp.ServerlessUpstream { + upstreams := []agenthttp.ServerlessUpstream{} for _, port := range expected.Ports { if port.ContainerPort == route.Port && port.HostPort > 0 { - return agenthttp.ServerlessUpstream{ + upstreams = append(upstreams, agenthttp.ServerlessUpstream{ DeploymentID: expected.DeploymentID, Url: fmt.Sprintf("127.0.0.1:%d", port.HostPort), Local: true, - }, true + }) } } - return agenthttp.ServerlessUpstream{}, false + return upstreams } func hasRunningLocalDeployment(deploymentIDs []string, actualByDeploymentID map[string]container.Container) bool { diff --git a/agent/internal/serverless/gateway_test.go b/agent/internal/serverless/gateway_test.go index 79c4678c..84853dfe 100644 --- a/agent/internal/serverless/gateway_test.go +++ b/agent/internal/serverless/gateway_test.go @@ -5,6 +5,7 @@ import ( "context" "errors" "log" + "slices" "strings" "sync" "testing" @@ -244,6 +245,84 @@ func TestSleepingLocalDeploymentWakesAndReturnsLocalUpstream(t *testing.T) { } } +func TestSleepingLocalDeploymentPrefersPublishedLoopbackUpstream(t *testing.T) { + previousPoll := wakePollInterval + previousReady := checkUpstreamReady + wakePollInterval = time.Millisecond + var checked []string + checkUpstreamReady = func(address string) upstreamReadiness { + checked = append(checked, address) + if address == "127.0.0.1:31000" { + return upstreamReadiness{ready: true} + } + t.Fatalf("unexpected readiness check for %s", address) + return upstreamReadiness{} + } + t.Cleanup(func() { + wakePollInterval = previousPoll + checkUpstreamReady = previousReady + }) + + state := testExpectedState("stopped") + state.Containers[0].PublishLocalPorts = true + state.Containers[0].Ports = []agenthttp.PortMapping{ + {ContainerPort: 3000, HostPort: 31000}, + } + state.Serverless.Routes[0].Upstreams = nil + runtime := &fakeRuntime{state: state} + gateway := NewGateway(runtime) + + upstreams, err := gateway.getUpstreams(context.Background(), "app.example.com") + if err != nil { + t.Fatalf("getUpstreams failed: %v", err) + } + if len(upstreams) != 1 || upstreams[0].Url != "127.0.0.1:31000" { + t.Fatalf("upstreams = %+v, want loopback upstream", upstreams) + } + if len(checked) != 1 || checked[0] != "127.0.0.1:31000" { + t.Fatalf("checked = %+v, want only loopback check", checked) + } +} + +func TestSleepingLocalDeploymentFallsBackToStaticIPWhenLoopbackUnreachable(t *testing.T) { + previousPoll := wakePollInterval + previousReady := checkUpstreamReady + wakePollInterval = time.Millisecond + var checked []string + checkUpstreamReady = func(address string) upstreamReadiness { + checked = append(checked, address) + if address == "10.0.0.10:3000" { + return upstreamReadiness{ready: true} + } + return upstreamReadiness{err: errors.New("connection refused")} + } + t.Cleanup(func() { + wakePollInterval = previousPoll + checkUpstreamReady = previousReady + }) + + state := testExpectedState("stopped") + state.Containers[0].PublishLocalPorts = true + state.Containers[0].Ports = []agenthttp.PortMapping{ + {ContainerPort: 3000, HostPort: 31000}, + } + state.Serverless.Routes[0].Upstreams = nil + runtime := &fakeRuntime{state: state} + gateway := NewGateway(runtime) + + upstreams, err := gateway.getUpstreams(context.Background(), "app.example.com") + if err != nil { + t.Fatalf("getUpstreams failed: %v", err) + } + if len(upstreams) != 1 || upstreams[0].Url != "10.0.0.10:3000" { + t.Fatalf("upstreams = %+v, want static IP fallback", upstreams) + } + wantChecked := []string{"127.0.0.1:31000", "10.0.0.10:3000"} + if !slices.Equal(checked, wantChecked) { + t.Fatalf("checked = %+v, want %+v", checked, wantChecked) + } +} + func TestSleepingLocalDeploymentStartsExistingStoppedContainer(t *testing.T) { previousPoll := wakePollInterval wakePollInterval = time.Millisecond diff --git a/web/lib/agent/expected-state.ts b/web/lib/agent/expected-state.ts index 7f68931d..74c6f4a5 100644 --- a/web/lib/agent/expected-state.ts +++ b/web/lib/agent/expected-state.ts @@ -51,6 +51,7 @@ export type ExpectedContainer = { image: string; ipAddress: string | null; ports: Array<{ containerPort: number; hostPort: number }>; + publishLocalPorts: boolean; env: Record; startCommand: string | null; healthCheck: { @@ -319,6 +320,7 @@ export function buildExpectedContainersFromRows({ containerPort: p.containerPort, hostPort: p.hostPort, })), + publishLocalPorts: isDeployedServerlessService(service), env: buildEnv(secretsByServiceId.get(dep.serviceId) ?? []), startCommand: service.startCommand || null, healthCheck: buildHealthCheck(service), diff --git a/web/tests/expected-state.test.ts b/web/tests/expected-state.test.ts index 29c90c87..4848957d 100644 --- a/web/tests/expected-state.test.ts +++ b/web/tests/expected-state.test.ts @@ -98,6 +98,7 @@ describe("expected-state pure builders", () => { { containerPort: 80, hostPort: 80 }, { containerPort: 3000, hostPort: 30001 }, ], + publishLocalPorts: true, env: { ALPHA: "first", ZED: "last" }, volumes: [ { name: "data", containerPath: "/data" }, @@ -214,6 +215,7 @@ describe("expected-state pure builders", () => { expect(containers[0]).toMatchObject({ deploymentId: "dep_stopped", desiredState: "stopped", + publishLocalPorts: false, }); }); @@ -245,6 +247,7 @@ describe("expected-state pure builders", () => { expect(containers[0]).toMatchObject({ deploymentId: "dep_sleeping", desiredState: "stopped", + publishLocalPorts: true, }); }); @@ -277,6 +280,7 @@ describe("expected-state pure builders", () => { expect(containers[0]).toMatchObject({ deploymentId: "dep_sleeping", desiredState: "stopped", + publishLocalPorts: true, }); }); From 339afa73d63cd041ae81a3fe048c3fe14e82a3c8 Mon Sep 17 00:00:00 2001 From: Techulus Agent <291950465+techulus-agent@users.noreply.github.com> Date: Wed, 8 Jul 2026 10:13:06 +1000 Subject: [PATCH 03/11] Fix serverless wake readiness regression --- agent/internal/serverless/gateway.go | 292 ++++++++++-------- agent/internal/serverless/gateway_test.go | 341 ++++++++++++++++++---- 2 files changed, 456 insertions(+), 177 deletions(-) diff --git a/agent/internal/serverless/gateway.go b/agent/internal/serverless/gateway.go index 6c6d447b..9bc4a7b3 100644 --- a/agent/internal/serverless/gateway.go +++ b/agent/internal/serverless/gateway.go @@ -28,7 +28,6 @@ var ( wakePollInterval = 500 * time.Millisecond idleTimerSeedInterval = 15 * time.Second upstreamDialTimeout = 250 * time.Millisecond - wakeWaitLogInterval = 2 * time.Second checkUpstreamReady = tcpUpstreamReady ) @@ -340,28 +339,10 @@ func (g *Gateway) inspectUpstreams(route *agenthttp.ServerlessRoute, state *agen } upstreams := localUpstreamCandidates(route, expected) if len(upstreams) > 0 { - upstreamReady := false - waitReasons := make([]upstreamWaitReason, 0, len(upstreams)) - for _, upstream := range upstreams { - readiness := checkUpstreamReady(upstream.Url) - if readiness.ready { - upstreamsByURL[upstream.Url] = upstream - upstreamReady = true - break - } else { - waitReasons = append(waitReasons, upstreamWaitReason{ - deploymentID: expected.DeploymentID, - serviceID: expected.ServiceID, - containerID: actual.ID, - upstreamURL: upstream.Url, - reason: "upstream_unreachable", - state: actual.State, - dialLatency: readiness.latency, - err: readiness.err, - }) - } - } - if !upstreamReady { + upstream, waitReasons, upstreamReady := probeLocalUpstreams(upstreams, expected, actual) + if upstreamReady { + upstreamsByURL[upstream.Url] = upstream + } else { resolution.waiting = append(resolution.waiting, waitReasons...) } } else { @@ -462,49 +443,52 @@ func (g *Gateway) wakeLocalDeployments(route *agenthttp.ServerlessRoute, state * } func (g *Gateway) waitForReadyUpstreams(route *agenthttp.ServerlessRoute, wakeTimeoutSeconds int, startedAt time.Time, wokenDeploymentIDs []string) ([]agenthttp.ServerlessUpstream, error) { + domain := route.Domain timeout := wakeTimeout(wakeTimeoutSeconds) deadline := startedAt.Add(timeout) attempt := 0 - nextWaitLogAt := time.Time{} - lastWaitKey := "" for { attempt++ state := g.runtime.ExpectedState() + currentRoute := findRoute(state, domain) + if currentRoute == nil { + log.Printf( + "[serverless-gateway] wake stale host=%s reason=route_missing attempts=%d latency=%s", + domain, + attempt, + roundDuration(time.Since(startedAt)), + ) + return nil, fmt.Errorf("serverless route changed during wake") + } + route = currentRoute + currentWokenIDs := currentWakeDeploymentIDs(route, state, wokenDeploymentIDs) resolution, err := g.inspectUpstreams(route, state) if err != nil { return nil, err } ready := resolution.ready if len(ready) > 0 { - pendingIDs := pendingWakeDeploymentIDs(wokenDeploymentIDs, ready) + pendingIDs := pendingWakeDeploymentIDs(currentWokenIDs, ready) if len(pendingIDs) > 0 { go g.waitForWokenDeployments(route, route.WakeTimeoutSeconds, startedAt, pendingIDs) } + logWakeReady(route, ready, attempt, startedAt) + return ready, nil + } + if len(currentWokenIDs) == 0 { log.Printf( - "[serverless-gateway] wake ready host=%s upstreams=%d attempts=%d latency=%s", - route.Domain, - len(ready), + "[serverless-gateway] wake stale host=%s reason=deployments_replaced attempts=%d latency=%s", + domain, attempt, roundDuration(time.Since(startedAt)), ) - return ready, nil + return nil, fmt.Errorf("serverless route changed during wake") } - waitSummary := summarizeWaitReasons(resolution.waiting) - waitKey := summarizeWaitReasonKeys(resolution.waiting) + waitSummary := summarizeWaitReasons(filterWaitReasons(resolution.waiting, currentWokenIDs)) if time.Now().After(deadline) { - pendingIDs := pendingWakeDeploymentIDs(wokenDeploymentIDs, ready) + pendingIDs := pendingWakeDeploymentIDs(currentWokenIDs, ready) g.queueWakeTimeouts(route, pendingIDs, startedAt, waitSummary) - if len(ready) > 0 { - log.Printf( - "[serverless-gateway] wake partially ready host=%s upstreams=%d attempts=%d latency=%s", - route.Domain, - len(ready), - attempt, - roundDuration(time.Since(startedAt)), - ) - return ready, nil - } log.Printf( "[serverless-gateway] wake timed out host=%s attempts=%d latency=%s last_wait=%q", route.Domain, @@ -514,53 +498,56 @@ func (g *Gateway) waitForReadyUpstreams(route *agenthttp.ServerlessRoute, wakeTi ) return nil, fmt.Errorf("timed out waiting for local serverless wake") } - if shouldLogWakeWait(nextWaitLogAt, lastWaitKey, waitKey) { - logWakeWait(route, resolution.waiting, attempt, time.Since(startedAt)) - nextWaitLogAt = time.Now().Add(wakeWaitLogInterval) - lastWaitKey = waitKey - } time.Sleep(wakePollInterval) } } func (g *Gateway) waitForWokenDeployments(route *agenthttp.ServerlessRoute, wakeTimeoutSeconds int, startedAt time.Time, wokenDeploymentIDs []string) { + domain := route.Domain timeout := wakeTimeout(wakeTimeoutSeconds) deadline := startedAt.Add(timeout) attempt := 0 - nextWaitLogAt := time.Time{} - lastWaitKey := "" for { attempt++ state := g.runtime.ExpectedState() + currentRoute := findRoute(state, domain) + if currentRoute == nil { + log.Printf( + "[serverless-gateway] wake monitor stale host=%s reason=route_missing attempts=%d latency=%s", + domain, + attempt, + roundDuration(time.Since(startedAt)), + ) + return + } + route = currentRoute + currentWokenIDs := currentWakeDeploymentIDs(route, state, wokenDeploymentIDs) + if len(currentWokenIDs) == 0 { + log.Printf( + "[serverless-gateway] wake monitor stale host=%s reason=deployments_replaced attempts=%d latency=%s", + domain, + attempt, + roundDuration(time.Since(startedAt)), + ) + return + } resolution, err := g.inspectUpstreams(route, state) if err != nil { log.Printf("[serverless-gateway] wake monitor failed host=%s error=%v", route.Domain, err) return } ready := resolution.ready - pendingIDs := pendingWakeDeploymentIDs(wokenDeploymentIDs, ready) + pendingIDs := pendingWakeDeploymentIDs(currentWokenIDs, ready) if len(pendingIDs) == 0 { - log.Printf( - "[serverless-gateway] wake ready host=%s deployments=%d attempts=%d latency=%s", - route.Domain, - len(wokenDeploymentIDs), - attempt, - roundDuration(time.Since(startedAt)), - ) + logWakeReadyDeployments(route, ready, len(currentWokenIDs), attempt, startedAt) return } - waitSummary := summarizeWaitReasons(resolution.waiting) - waitKey := summarizeWaitReasonKeys(resolution.waiting) + waitSummary := summarizeWaitReasons(filterWaitReasons(resolution.waiting, pendingIDs)) if time.Now().After(deadline) { g.queueWakeTimeouts(route, pendingIDs, startedAt, waitSummary) return } - if shouldLogWakeWait(nextWaitLogAt, lastWaitKey, waitKey) { - logWakeWait(route, filterWaitReasons(resolution.waiting, pendingIDs), attempt, time.Since(startedAt)) - nextWaitLogAt = time.Now().Add(wakeWaitLogInterval) - lastWaitKey = waitKey - } time.Sleep(wakePollInterval) } } @@ -582,57 +569,6 @@ func pendingWakeDeploymentIDs(wokenDeploymentIDs []string, ready []agenthttp.Ser return pendingIDs } -func shouldLogWakeWait(nextLogAt time.Time, lastSummary string, currentSummary string) bool { - if currentSummary == "" { - return false - } - if currentSummary != lastSummary { - return true - } - return nextLogAt.IsZero() || time.Now().After(nextLogAt) -} - -func logWakeWait(route *agenthttp.ServerlessRoute, reasons []upstreamWaitReason, attempt int, elapsed time.Duration) { - if len(reasons) == 0 { - log.Printf( - "[serverless-gateway] wake waiting host=%s attempt=%d elapsed=%s reason=no_ready_upstreams", - route.Domain, - attempt, - roundDuration(elapsed), - ) - return - } - for _, reason := range reasons { - fields := []string{ - fmt.Sprintf("host=%s", route.Domain), - fmt.Sprintf("deployment=%s", reason.deploymentID), - fmt.Sprintf("service=%s", reason.serviceID), - fmt.Sprintf("attempt=%d", attempt), - fmt.Sprintf("elapsed=%s", roundDuration(elapsed)), - fmt.Sprintf("reason=%s", reason.reason), - } - if reason.containerID != "" { - fields = append(fields, fmt.Sprintf("container=%s", reason.containerID)) - } - if reason.upstreamURL != "" { - fields = append(fields, fmt.Sprintf("upstream=%s", reason.upstreamURL)) - } - if reason.state != "" { - fields = append(fields, fmt.Sprintf("state=%s", reason.state)) - } - if reason.health != "" { - fields = append(fields, fmt.Sprintf("health=%s", reason.health)) - } - if reason.dialLatency > 0 { - fields = append(fields, fmt.Sprintf("dial_latency=%s", roundDuration(reason.dialLatency))) - } - if reason.err != nil { - fields = append(fields, fmt.Sprintf("error=%q", compactError(reason.err))) - } - log.Printf("[serverless-gateway] wake waiting %s", strings.Join(fields, " ")) - } -} - func filterWaitReasons(reasons []upstreamWaitReason, deploymentIDs []string) []upstreamWaitReason { if len(reasons) == 0 || len(deploymentIDs) == 0 { return nil @@ -650,12 +586,73 @@ func filterWaitReasons(reasons []upstreamWaitReason, deploymentIDs []string) []u return filtered } -func summarizeWaitReasons(reasons []upstreamWaitReason) string { - return formatWaitReasons(reasons, true) +func currentWakeDeploymentIDs(route *agenthttp.ServerlessRoute, state *agenthttp.ExpectedState, deploymentIDs []string) []string { + if route == nil || len(deploymentIDs) == 0 { + return nil + } + + expectedByDeploymentID := expectedContainersByDeploymentID(state) + routeIDs := map[string]struct{}{} + for _, deploymentID := range route.LocalDeploymentIDs { + routeIDs[deploymentID] = struct{}{} + } + + currentIDs := []string{} + for _, deploymentID := range deploymentIDs { + if _, ok := routeIDs[deploymentID]; !ok { + continue + } + if _, ok := expectedByDeploymentID[deploymentID]; !ok { + continue + } + currentIDs = append(currentIDs, deploymentID) + } + return currentIDs } -func summarizeWaitReasonKeys(reasons []upstreamWaitReason) string { - return formatWaitReasons(reasons, false) +func logWakeReady(route *agenthttp.ServerlessRoute, ready []agenthttp.ServerlessUpstream, attempt int, startedAt time.Time) { + fields := []string{ + fmt.Sprintf("host=%s", route.Domain), + fmt.Sprintf("upstreams=%d", len(ready)), + } + if localUpstreams := formatLocalUpstreamURLs(ready); localUpstreams != "" { + fields = append(fields, fmt.Sprintf("local_upstreams=%s", localUpstreams)) + } + fields = append(fields, + fmt.Sprintf("attempts=%d", attempt), + fmt.Sprintf("latency=%s", roundDuration(time.Since(startedAt))), + ) + log.Printf("[serverless-gateway] wake ready %s", strings.Join(fields, " ")) +} + +func logWakeReadyDeployments(route *agenthttp.ServerlessRoute, ready []agenthttp.ServerlessUpstream, deployments int, attempt int, startedAt time.Time) { + fields := []string{ + fmt.Sprintf("host=%s", route.Domain), + fmt.Sprintf("deployments=%d", deployments), + } + if localUpstreams := formatLocalUpstreamURLs(ready); localUpstreams != "" { + fields = append(fields, fmt.Sprintf("local_upstreams=%s", localUpstreams)) + } + fields = append(fields, + fmt.Sprintf("attempts=%d", attempt), + fmt.Sprintf("latency=%s", roundDuration(time.Since(startedAt))), + ) + log.Printf("[serverless-gateway] wake ready %s", strings.Join(fields, " ")) +} + +func formatLocalUpstreamURLs(upstreams []agenthttp.ServerlessUpstream) string { + urls := []string{} + for _, upstream := range upstreams { + if upstream.Local { + urls = append(urls, upstream.Url) + } + } + sort.Strings(urls) + return strings.Join(urls, ",") +} + +func summarizeWaitReasons(reasons []upstreamWaitReason) string { + return formatWaitReasons(reasons, true) } func formatWaitReasons(reasons []upstreamWaitReason, includeDialLatency bool) string { @@ -1189,6 +1186,61 @@ func localLoopbackUpstreams(route *agenthttp.ServerlessRoute, expected agenthttp return upstreams } +type upstreamProbeResult struct { + upstream agenthttp.ServerlessUpstream + readiness upstreamReadiness +} + +func probeLocalUpstreams(upstreams []agenthttp.ServerlessUpstream, expected agenthttp.ExpectedContainer, actual container.Container) (agenthttp.ServerlessUpstream, []upstreamWaitReason, bool) { + if len(upstreams) == 0 { + return agenthttp.ServerlessUpstream{}, nil, false + } + + check := checkUpstreamReady + results := make(chan upstreamProbeResult, len(upstreams)) + for _, upstream := range upstreams { + upstream := upstream + go func() { + results <- upstreamProbeResult{ + upstream: upstream, + readiness: check(upstream.Url), + } + }() + } + + waitReasons := make([]upstreamWaitReason, 0, len(upstreams)) + for range upstreams { + result := <-results + if result.readiness.ready { + return result.upstream, nil, true + } + waitReasons = append(waitReasons, upstreamWaitReason{ + deploymentID: expected.DeploymentID, + serviceID: expected.ServiceID, + containerID: actual.ID, + upstreamURL: result.upstream.Url, + reason: "upstream_unreachable", + state: actual.State, + dialLatency: result.readiness.latency, + err: result.readiness.err, + }) + } + sortWaitReasons(waitReasons) + return agenthttp.ServerlessUpstream{}, waitReasons, false +} + +func sortWaitReasons(reasons []upstreamWaitReason) { + sort.Slice(reasons, func(i, j int) bool { + if reasons[i].deploymentID != reasons[j].deploymentID { + return reasons[i].deploymentID < reasons[j].deploymentID + } + if reasons[i].reason != reasons[j].reason { + return reasons[i].reason < reasons[j].reason + } + return reasons[i].upstreamURL < reasons[j].upstreamURL + }) +} + func hasRunningLocalDeployment(deploymentIDs []string, actualByDeploymentID map[string]container.Container) bool { for _, deploymentID := range deploymentIDs { if actual, ok := actualByDeploymentID[deploymentID]; ok && actual.State == "running" { diff --git a/agent/internal/serverless/gateway_test.go b/agent/internal/serverless/gateway_test.go index 84853dfe..45dc0e12 100644 --- a/agent/internal/serverless/gateway_test.go +++ b/agent/internal/serverless/gateway_test.go @@ -134,6 +134,36 @@ func (f *fakeRuntime) snapshotContainers() []container.Container { return append([]container.Container(nil), f.containers...) } +func (f *fakeRuntime) setState(state *agenthttp.ExpectedState, containers []container.Container) { + f.mu.Lock() + defer f.mu.Unlock() + f.state = state + f.containers = append([]container.Container(nil), containers...) +} + +func receiveProbe(t *testing.T, probes <-chan string) string { + t.Helper() + select { + case probe := <-probes: + return probe + case <-time.After(100 * time.Millisecond): + t.Fatal("timed out waiting for readiness probe") + return "" + } +} + +func assertProbeSet(t *testing.T, got []string, want ...string) { + t.Helper() + if len(got) != len(want) { + t.Fatalf("probes = %+v, want %+v", got, want) + } + for _, probe := range want { + if !slices.Contains(got, probe) { + t.Fatalf("probes = %+v, want %+v", got, want) + } + } +} + func TestGetUpstreamsReturnsWhenFollowerContextIsCancelled(t *testing.T) { state := testExpectedState("stopped") g := NewGateway(&fakeRuntime{state: state}) @@ -249,16 +279,16 @@ func TestSleepingLocalDeploymentPrefersPublishedLoopbackUpstream(t *testing.T) { previousPoll := wakePollInterval previousReady := checkUpstreamReady wakePollInterval = time.Millisecond - var checked []string + releaseStatic := make(chan struct{}) checkUpstreamReady = func(address string) upstreamReadiness { - checked = append(checked, address) if address == "127.0.0.1:31000" { return upstreamReadiness{ready: true} } - t.Fatalf("unexpected readiness check for %s", address) - return upstreamReadiness{} + <-releaseStatic + return upstreamReadiness{err: errors.New("connection refused")} } t.Cleanup(func() { + close(releaseStatic) wakePollInterval = previousPoll checkUpstreamReady = previousReady }) @@ -279,18 +309,13 @@ func TestSleepingLocalDeploymentPrefersPublishedLoopbackUpstream(t *testing.T) { if len(upstreams) != 1 || upstreams[0].Url != "127.0.0.1:31000" { t.Fatalf("upstreams = %+v, want loopback upstream", upstreams) } - if len(checked) != 1 || checked[0] != "127.0.0.1:31000" { - t.Fatalf("checked = %+v, want only loopback check", checked) - } } func TestSleepingLocalDeploymentFallsBackToStaticIPWhenLoopbackUnreachable(t *testing.T) { previousPoll := wakePollInterval previousReady := checkUpstreamReady wakePollInterval = time.Millisecond - var checked []string checkUpstreamReady = func(address string) upstreamReadiness { - checked = append(checked, address) if address == "10.0.0.10:3000" { return upstreamReadiness{ready: true} } @@ -317,9 +342,94 @@ func TestSleepingLocalDeploymentFallsBackToStaticIPWhenLoopbackUnreachable(t *te if len(upstreams) != 1 || upstreams[0].Url != "10.0.0.10:3000" { t.Fatalf("upstreams = %+v, want static IP fallback", upstreams) } - wantChecked := []string{"127.0.0.1:31000", "10.0.0.10:3000"} - if !slices.Equal(checked, wantChecked) { - t.Fatalf("checked = %+v, want %+v", checked, wantChecked) +} + +func TestPublishedLoopbackAndStaticIPAreProbedConcurrently(t *testing.T) { + previousPoll := wakePollInterval + previousReady := checkUpstreamReady + wakePollInterval = time.Millisecond + probes := make(chan string, 2) + release := make(chan struct{}) + var releaseOnce sync.Once + checkUpstreamReady = func(address string) upstreamReadiness { + probes <- address + <-release + return upstreamReadiness{err: errors.New("connection refused")} + } + t.Cleanup(func() { + releaseOnce.Do(func() { close(release) }) + wakePollInterval = previousPoll + checkUpstreamReady = previousReady + }) + + state := testExpectedState("running") + state.Containers[0].PublishLocalPorts = true + state.Containers[0].Ports = []agenthttp.PortMapping{ + {ContainerPort: 3000, HostPort: 31000}, + } + state.Serverless.Routes[0].Upstreams = nil + runtime := &fakeRuntime{ + state: state, + containers: []container.Container{ + {ID: "ctr-local", State: "running", DeploymentID: "dep_local", ServiceID: "svc_1"}, + }, + } + gateway := NewGateway(runtime) + + done := make(chan struct{}) + go func() { + _, _ = gateway.inspectUpstreams(&state.Serverless.Routes[0], state) + close(done) + }() + + got := []string{receiveProbe(t, probes)} + select { + case probe := <-probes: + got = append(got, probe) + case <-time.After(100 * time.Millisecond): + t.Fatalf("only saw probes %+v; want loopback and static IP to start concurrently", got) + } + assertProbeSet(t, got, "127.0.0.1:31000", "10.0.0.10:3000") + + releaseOnce.Do(func() { close(release) }) + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("timed out waiting for concurrent probes to finish") + } +} + +func TestPublishedLocalPortsWithoutMatchingHostPortUsesStaticIP(t *testing.T) { + previousPoll := wakePollInterval + previousReady := checkUpstreamReady + wakePollInterval = time.Millisecond + checkUpstreamReady = func(address string) upstreamReadiness { + if address != "10.0.0.10:3000" { + t.Errorf("checked %s, want only static IP", address) + return upstreamReadiness{err: errors.New("unexpected upstream")} + } + return upstreamReadiness{ready: true} + } + t.Cleanup(func() { + wakePollInterval = previousPoll + checkUpstreamReady = previousReady + }) + + state := testExpectedState("stopped") + state.Containers[0].PublishLocalPorts = true + state.Containers[0].Ports = []agenthttp.PortMapping{ + {ContainerPort: 8080, HostPort: 31000}, + } + state.Serverless.Routes[0].Upstreams = nil + runtime := &fakeRuntime{state: state} + gateway := NewGateway(runtime) + + upstreams, err := gateway.getUpstreams(context.Background(), "app.example.com") + if err != nil { + t.Fatalf("getUpstreams failed: %v", err) + } + if len(upstreams) != 1 || upstreams[0].Url != "10.0.0.10:3000" { + t.Fatalf("upstreams = %+v, want static IP upstream", upstreams) } } @@ -467,49 +577,106 @@ func TestPendingSleepCanWakeBeforeExpectedStateSettles(t *testing.T) { } } -func TestWakeTimeoutQueuesWakeFailedTransition(t *testing.T) { +func TestBlockingWakeRefreshesRouteAfterRedeploy(t *testing.T) { previousPoll := wakePollInterval + previousReady := checkUpstreamReady wakePollInterval = time.Millisecond + checkUpstreamReady = func(address string) upstreamReadiness { + if address == "10.0.0.11:3000" { + return upstreamReadiness{ready: true} + } + return upstreamReadiness{err: errors.New("connection refused")} + } t.Cleanup(func() { wakePollInterval = previousPoll + checkUpstreamReady = previousReady }) - state := testExpectedState("stopped") - state.Containers[0].HealthCheck = &agenthttp.HealthCheck{ - Cmd: "curl http://localhost:3000/health", - Interval: 1, - Timeout: 1, - Retries: 1, - } - state.Serverless.Routes[0].Upstreams = nil - state.Serverless.Routes[0].WakeTimeoutSeconds = 1 + oldState := testExpectedStateWithLocalDeployment("running", "dep_local", "10.0.0.10") + oldState.Serverless.Routes[0].Upstreams = nil + newState := testExpectedStateWithLocalDeployment("running", "dep_new", "10.0.0.11") + newState.Serverless.Routes[0].Upstreams = nil + runtime := &fakeRuntime{ - state: state, - healthStatus: "unhealthy", + state: oldState, + containers: []container.Container{ + {ID: "ctr-old", State: "running", DeploymentID: "dep_local", ServiceID: "svc_1"}, + }, + } + var swapped sync.Once + runtime.afterList = func() { + swapped.Do(func() { + runtime.setState(newState, []container.Container{ + {ID: "ctr-new", State: "running", DeploymentID: "dep_new", ServiceID: "svc_1"}, + }) + }) } gateway := NewGateway(runtime) - _, err := gateway.getUpstreams(context.Background(), "app.example.com") - if err == nil { - t.Fatal("getUpstreams succeeded, want timeout error") + upstreams, err := gateway.waitForReadyUpstreams(&oldState.Serverless.Routes[0], 5, time.Now(), []string{"dep_local"}) + if err != nil { + t.Fatalf("waitForReadyUpstreams failed: %v", err) + } + if len(upstreams) != 1 || upstreams[0].Url != "10.0.0.11:3000" || upstreams[0].DeploymentID != "dep_new" { + t.Fatalf("upstreams = %+v, want redeployed local upstream", upstreams) } + transitions, _, _ := runtime.snapshot() + if len(transitions) != 0 { + t.Fatalf("transitions = %+v, want no stale wake_failed transition", transitions) + } +} - transitions, _, deployCalls := runtime.snapshot() - if deployCalls != 1 { - t.Fatalf("deployCalls = %d, want 1", deployCalls) +func TestBlockingWakeReturnsWhenRouteDisappearsDuringRedeploy(t *testing.T) { + previousPoll := wakePollInterval + previousReady := checkUpstreamReady + wakePollInterval = time.Millisecond + checkUpstreamReady = func(string) upstreamReadiness { + return upstreamReadiness{err: errors.New("connection refused")} } - if len(transitions) != 2 { - t.Fatalf("transitions = %+v, want wake_started and wake_failed", transitions) + t.Cleanup(func() { + wakePollInterval = previousPoll + checkUpstreamReady = previousReady + }) + + oldState := testExpectedState("running") + oldState.Serverless.Routes[0].Upstreams = nil + newState := testExpectedState("running") + newState.Serverless.Routes = nil + runtime := &fakeRuntime{ + state: oldState, + containers: []container.Container{ + {ID: "ctr-old", State: "running", DeploymentID: "dep_local", ServiceID: "svc_1"}, + }, } - if transitions[0].Type != "wake_started" || transitions[1].Type != "wake_failed" { - t.Fatalf("transitions = %+v, want wake_started then wake_failed", transitions) + var swapped sync.Once + runtime.afterList = func() { + swapped.Do(func() { + runtime.setState(newState, nil) + }) } - if transitions[1].DeploymentID != "dep_local" || transitions[1].Error == "" { - t.Fatalf("wake_failed transition = %+v, want dep_local with error", transitions[1]) + gateway := NewGateway(runtime) + + done := make(chan error, 1) + go func() { + _, err := gateway.waitForReadyUpstreams(&oldState.Serverless.Routes[0], 5, time.Now(), []string{"dep_local"}) + done <- err + }() + + select { + case err := <-done: + if err == nil { + t.Fatal("waitForReadyUpstreams succeeded, want stale route error") + } + case <-time.After(200 * time.Millisecond): + t.Fatal("timed out waiting for stale route to return") + } + transitions, _, _ := runtime.snapshot() + if len(transitions) != 0 { + t.Fatalf("transitions = %+v, want no stale wake_failed transition", transitions) } } -func TestWakeTimeoutWhenLocalPortNeverOpens(t *testing.T) { +func TestWakeMonitorDropsStaleDeploymentAfterRedeploy(t *testing.T) { previousPoll := wakePollInterval previousReady := checkUpstreamReady wakePollInterval = time.Millisecond @@ -521,10 +688,63 @@ func TestWakeTimeoutWhenLocalPortNeverOpens(t *testing.T) { checkUpstreamReady = previousReady }) + oldState := testExpectedStateWithLocalDeployment("running", "dep_local", "10.0.0.10") + oldState.Serverless.Routes[0].Upstreams = nil + newState := testExpectedStateWithLocalDeployment("running", "dep_new", "10.0.0.11") + newState.Serverless.Routes[0].Upstreams = nil + runtime := &fakeRuntime{ + state: oldState, + containers: []container.Container{ + {ID: "ctr-old", State: "running", DeploymentID: "dep_local", ServiceID: "svc_1"}, + }, + } + var swapped sync.Once + runtime.afterList = func() { + swapped.Do(func() { + runtime.setState(newState, []container.Container{ + {ID: "ctr-new", State: "running", DeploymentID: "dep_new", ServiceID: "svc_1"}, + }) + }) + } + gateway := NewGateway(runtime) + + done := make(chan struct{}) + go func() { + gateway.waitForWokenDeployments(&oldState.Serverless.Routes[0], 5, time.Now(), []string{"dep_local"}) + close(done) + }() + + select { + case <-done: + case <-time.After(200 * time.Millisecond): + t.Fatal("timed out waiting for stale wake monitor to return") + } + transitions, _, _ := runtime.snapshot() + if len(transitions) != 0 { + t.Fatalf("transitions = %+v, want no stale wake_failed transition", transitions) + } +} + +func TestWakeTimeoutQueuesWakeFailedTransition(t *testing.T) { + previousPoll := wakePollInterval + wakePollInterval = time.Millisecond + t.Cleanup(func() { + wakePollInterval = previousPoll + }) + state := testExpectedState("stopped") + state.Containers[0].HealthCheck = &agenthttp.HealthCheck{ + Cmd: "curl http://localhost:3000/health", + Interval: 1, + Timeout: 1, + Retries: 1, + } state.Serverless.Routes[0].Upstreams = nil state.Serverless.Routes[0].WakeTimeoutSeconds = 1 - runtime := &fakeRuntime{state: state} + runtime := &fakeRuntime{ + state: state, + healthStatus: "unhealthy", + } gateway := NewGateway(runtime) _, err := gateway.getUpstreams(context.Background(), "app.example.com") @@ -542,23 +762,20 @@ func TestWakeTimeoutWhenLocalPortNeverOpens(t *testing.T) { if transitions[0].Type != "wake_started" || transitions[1].Type != "wake_failed" { t.Fatalf("transitions = %+v, want wake_started then wake_failed", transitions) } + if transitions[1].DeploymentID != "dep_local" || transitions[1].Error == "" { + t.Fatalf("wake_failed transition = %+v, want dep_local with error", transitions[1]) + } } -func TestWakeWaitingLogsReadinessReason(t *testing.T) { +func TestWakeTimeoutWhenLocalPortNeverOpens(t *testing.T) { previousPoll := wakePollInterval - previousLogInterval := wakeWaitLogInterval previousReady := checkUpstreamReady wakePollInterval = time.Millisecond - wakeWaitLogInterval = time.Hour checkUpstreamReady = func(string) upstreamReadiness { - return upstreamReadiness{ - latency: time.Millisecond, - err: errors.New("connection refused"), - } + return upstreamReadiness{err: errors.New("connection refused")} } t.Cleanup(func() { wakePollInterval = previousPoll - wakeWaitLogInterval = previousLogInterval checkUpstreamReady = previousReady }) @@ -583,17 +800,18 @@ func TestWakeWaitingLogsReadinessReason(t *testing.T) { t.Fatal("getUpstreams succeeded, want timeout error") } - output := logs.String() - for _, want := range []string{ - "wake waiting", - "host=app.example.com", - "deployment=dep_local", - "reason=upstream_unreachable", - "error=\"connection refused\"", - } { - if !strings.Contains(output, want) { - t.Fatalf("log output missing %q:\n%s", want, output) - } + transitions, _, deployCalls := runtime.snapshot() + if deployCalls != 1 { + t.Fatalf("deployCalls = %d, want 1", deployCalls) + } + if len(transitions) != 2 { + t.Fatalf("transitions = %+v, want wake_started and wake_failed", transitions) + } + if transitions[0].Type != "wake_started" || transitions[1].Type != "wake_failed" { + t.Fatalf("transitions = %+v, want wake_started then wake_failed", transitions) + } + if strings.Contains(logs.String(), "wake waiting") { + t.Fatalf("log output contains verbose wake waiting logs:\n%s", logs.String()) } } @@ -631,3 +849,12 @@ func testExpectedState(localDesiredState string) *agenthttp.ExpectedState { } return state } + +func testExpectedStateWithLocalDeployment(localDesiredState string, deploymentID string, ipAddress string) *agenthttp.ExpectedState { + state := testExpectedState(localDesiredState) + state.Containers[0].DeploymentID = deploymentID + state.Containers[0].Name = "svc_1-" + deploymentID + state.Containers[0].IPAddress = ipAddress + state.Serverless.Routes[0].LocalDeploymentIDs = []string{deploymentID} + return state +} From 8e648b659a35bccc75038b59fbb9d753b4ecbc92 Mon Sep 17 00:00:00 2001 From: Techulus Agent <291950465+techulus-agent@users.noreply.github.com> Date: Wed, 8 Jul 2026 12:52:26 +1000 Subject: [PATCH 04/11] Show sleeping serverless deployments in UI --- .../details/service-details-overview.tsx | 8 ++++- web/components/ui/canvas-wrapper.tsx | 9 +++++- web/tests/canvas-status-color.test.ts | 31 +++++++++++++++++++ 3 files changed, 46 insertions(+), 2 deletions(-) create mode 100644 web/tests/canvas-status-color.test.ts diff --git a/web/components/service/details/service-details-overview.tsx b/web/components/service/details/service-details-overview.tsx index a3330a2b..505a9a14 100644 --- a/web/components/service/details/service-details-overview.tsx +++ b/web/components/service/details/service-details-overview.tsx @@ -78,7 +78,7 @@ type SourceInfo = { type ServiceStatus = { label: string; - tone: "live" | "progress" | "warning" | "muted"; + tone: "live" | "progress" | "warning" | "sleeping" | "muted"; }; type ChartRow = { @@ -119,6 +119,7 @@ const STATUS_TONE_CLASSES: Record< live: { dot: "bg-teal-500", text: "text-teal-700 dark:text-teal-400" }, progress: { dot: "bg-blue-500", text: "text-blue-700 dark:text-blue-400" }, warning: { dot: "bg-red-500", text: "text-red-700 dark:text-red-400" }, + sleeping: { dot: "bg-cyan-500", text: "text-cyan-700 dark:text-cyan-400" }, muted: { dot: "bg-muted-foreground", text: "text-muted-foreground" }, }; @@ -779,6 +780,11 @@ function getServiceStatus( return { label: "Needs attention", tone: "warning" }; } } + for (const deployment of service.deployments || []) { + if (deployment.observedPhase === "sleeping") { + return { label: "Sleeping", tone: "sleeping" }; + } + } return { label: service.deployments.length > 0 ? "Stopped" : "Not deployed", diff --git a/web/components/ui/canvas-wrapper.tsx b/web/components/ui/canvas-wrapper.tsx index e863f06f..439e7556 100644 --- a/web/components/ui/canvas-wrapper.tsx +++ b/web/components/ui/canvas-wrapper.tsx @@ -40,6 +40,12 @@ const statusColorMap: Record = { dot: "bg-slate-400", text: "text-slate-500", }, + sleeping: { + bg: "bg-cyan-500/5", + border: "border-cyan-500/30", + dot: "bg-cyan-500", + text: "text-cyan-700 dark:text-cyan-400", + }, failed: { bg: "bg-rose-500/5", border: "border-rose-500/30", @@ -79,10 +85,10 @@ export function getStatusColorFromDeployments( d.observedPhase === "waking", ); const hasFailed = deployments.some((d) => d.observedPhase === "failed"); + const hasSleeping = deployments.some((d) => d.observedPhase === "sleeping"); const hasStopped = deployments.some( (d) => d.observedPhase === "stopped" || - d.observedPhase === "sleeping" || d.runtimeDesiredState === "removed", ); const hasUnknown = deployments.some((d) => d.observedPhase === "unknown"); @@ -91,6 +97,7 @@ export function getStatusColorFromDeployments( if (hasPending) return statusColorMap.pending; if (hasFailed) return statusColorMap.failed; if (hasUnknown) return statusColorMap.unknown; + if (hasSleeping) return statusColorMap.sleeping; if (hasStopped) return statusColorMap.stopped; return defaultColors; } diff --git a/web/tests/canvas-status-color.test.ts b/web/tests/canvas-status-color.test.ts new file mode 100644 index 00000000..6622a105 --- /dev/null +++ b/web/tests/canvas-status-color.test.ts @@ -0,0 +1,31 @@ +import { describe, expect, it } from "vitest"; +import { + getStatusColor, + getStatusColorFromDeployments, +} from "@/components/ui/canvas-wrapper"; + +describe("canvas status colors", () => { + it("uses a distinct color for sleeping deployments", () => { + expect(getStatusColorFromDeployments([{ observedPhase: "sleeping" }])).toEqual( + getStatusColor("sleeping"), + ); + expect(getStatusColorFromDeployments([{ observedPhase: "sleeping" }])).not.toEqual( + getStatusColor("stopped"), + ); + }); + + it("keeps active and failed states higher priority than sleeping", () => { + expect( + getStatusColorFromDeployments([ + { observedPhase: "sleeping" }, + { observedPhase: "healthy" }, + ]), + ).toEqual(getStatusColor("running")); + expect( + getStatusColorFromDeployments([ + { observedPhase: "sleeping" }, + { observedPhase: "failed" }, + ]), + ).toEqual(getStatusColor("failed")); + }); +}); From 651d9c4be8760fdb4ad89de08b586c562db2c7bc Mon Sep 17 00:00:00 2001 From: Techulus Agent <291950465+techulus-agent@users.noreply.github.com> Date: Wed, 8 Jul 2026 13:05:49 +1000 Subject: [PATCH 05/11] Show waking serverless deployments in UI --- .../details/service-details-overview.tsx | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) diff --git a/web/components/service/details/service-details-overview.tsx b/web/components/service/details/service-details-overview.tsx index 505a9a14..be26b73d 100644 --- a/web/components/service/details/service-details-overview.tsx +++ b/web/components/service/details/service-details-overview.tsx @@ -24,6 +24,7 @@ import { Badge } from "@/components/ui/badge"; import { Card } from "@/components/ui/card"; import { Skeleton } from "@/components/ui/skeleton"; import type { ServiceWithDetails as Service } from "@/db/types"; +import { isObservedStarting } from "@/lib/deployment-status"; import { fetcher } from "@/lib/fetcher"; import { cn } from "@/lib/utils"; @@ -759,6 +760,7 @@ function getServiceStatus( runningDeployments: number, ): ServiceStatus { const latestRollout = service.rollouts?.[0]; + const deployments = service.deployments || []; if (service.migrationStatus) return { label: "Migrating", tone: "progress" }; if ( @@ -775,19 +777,29 @@ function getServiceStatus( } if (runningDeployments > 0) return { label: "Live", tone: "live" }; - for (const deployment of service.deployments || []) { + for (const deployment of deployments) { if (deployment.observedPhase === "failed") { return { label: "Needs attention", tone: "warning" }; } } - for (const deployment of service.deployments || []) { + for (const deployment of deployments) { + if (deployment.observedPhase === "waking") { + return { label: "Waking", tone: "progress" }; + } + } + for (const deployment of deployments) { + if (isObservedStarting(deployment.observedPhase)) { + return { label: "Starting", tone: "progress" }; + } + } + for (const deployment of deployments) { if (deployment.observedPhase === "sleeping") { return { label: "Sleeping", tone: "sleeping" }; } } return { - label: service.deployments.length > 0 ? "Stopped" : "Not deployed", + label: deployments.length > 0 ? "Stopped" : "Not deployed", tone: "muted", }; } From 2d4120c5dc4ab3e0859c632d0b34eeb2e851fc45 Mon Sep 17 00:00:00 2001 From: Techulus Agent <291950465+techulus-agent@users.noreply.github.com> Date: Wed, 8 Jul 2026 16:40:32 +1000 Subject: [PATCH 06/11] Preserve waking state during serverless wake --- web/lib/agent-status.ts | 7 +++++++ web/tests/agent-status.test.ts | 10 ++++++++++ 2 files changed, 17 insertions(+) diff --git a/web/lib/agent-status.ts b/web/lib/agent-status.ts index dde83bc6..dc4a362b 100644 --- a/web/lib/agent-status.ts +++ b/web/lib/agent-status.ts @@ -66,6 +66,7 @@ export function shouldAttachReportedContainer(observedPhase: ObservedPhase) { export function getStoppedContainerReportUpdate(deployment: { runtimeDesiredState: string; + observedPhase?: string; }) { if (deployment.runtimeDesiredState === "stopped") { return { @@ -74,6 +75,12 @@ export function getStoppedContainerReportUpdate(deployment: { healthStatus: null, }; } + if (deployment.observedPhase === "waking") { + return { + observedPhase: "waking" as const, + healthStatus: null, + }; + } return { observedPhase: "stopped" as const, diff --git a/web/tests/agent-status.test.ts b/web/tests/agent-status.test.ts index ea56421c..f2c976e1 100644 --- a/web/tests/agent-status.test.ts +++ b/web/tests/agent-status.test.ts @@ -48,6 +48,16 @@ describe("agent status serverless attachment", () => { observedPhase: "stopped", healthStatus: "none", }); + + expect( + getStoppedContainerReportUpdate({ + runtimeDesiredState: "running", + observedPhase: "waking", + }), + ).toEqual({ + observedPhase: "waking", + healthStatus: null, + }); }); it("restores stale stopped serverless observations from live running reports", () => { From fd1afd1351eede89ff287c0927353ecf6dbf5f60 Mon Sep 17 00:00:00 2001 From: Techulus Agent <291950465+techulus-agent@users.noreply.github.com> Date: Wed, 8 Jul 2026 16:48:17 +1000 Subject: [PATCH 07/11] Raise serverless sleep minimum to 120 seconds --- web/actions/projects.ts | 7 ++++- .../service/details/serverless-section.tsx | 7 +++-- web/lib/service-config.ts | 28 ++++++++++++------ web/tests/service-config.test.ts | 29 +++++++++++++++++++ 4 files changed, 58 insertions(+), 13 deletions(-) diff --git a/web/actions/projects.ts b/web/actions/projects.ts index 1b6e9c82..828a2575 100644 --- a/web/actions/projects.ts +++ b/web/actions/projects.ts @@ -47,6 +47,7 @@ import { nameSchema, volumeNameSchema, } from "@/lib/schemas"; +import { MIN_SERVERLESS_SLEEP_AFTER_SECONDS } from "@/lib/service-config"; import type { PortConfig, HealthCheckConfig as ServiceHealthCheckConfig, @@ -1002,7 +1003,11 @@ export async function updateServiceSchedule( const serverlessSettingsSchema = z.object({ enabled: z.boolean(), - sleepAfterSeconds: z.number().int().min(60).max(86_400), + sleepAfterSeconds: z + .number() + .int() + .min(MIN_SERVERLESS_SLEEP_AFTER_SECONDS) + .max(86_400), wakeTimeoutSeconds: z.number().int().min(10).max(900), }); diff --git a/web/components/service/details/serverless-section.tsx b/web/components/service/details/serverless-section.tsx index 17664b35..94814241 100644 --- a/web/components/service/details/serverless-section.tsx +++ b/web/components/service/details/serverless-section.tsx @@ -9,6 +9,7 @@ import { Input } from "@/components/ui/input"; import { Item, ItemContent, ItemMedia, ItemTitle } from "@/components/ui/item"; import { Switch } from "@/components/ui/switch"; import type { ServiceWithDetails as Service } from "@/db/types"; +import { MIN_SERVERLESS_SLEEP_AFTER_SECONDS } from "@/lib/service-config"; export const ServerlessSection = memo(function ServerlessSection({ service, @@ -51,10 +52,10 @@ export const ServerlessSection = memo(function ServerlessSection({ } if ( !Number.isInteger(parsed.sleepAfterSeconds) || - parsed.sleepAfterSeconds < 60 || + parsed.sleepAfterSeconds < MIN_SERVERLESS_SLEEP_AFTER_SECONDS || parsed.sleepAfterSeconds > 86_400 ) { - return "Sleep timeout must be between 60 seconds and 24 hours"; + return `Sleep timeout must be between ${MIN_SERVERLESS_SLEEP_AFTER_SECONDS} seconds and 24 hours`; } if ( !Number.isInteger(parsed.wakeTimeoutSeconds) || @@ -121,7 +122,7 @@ export const ServerlessSection = memo(function ServerlessSection({ ({ key: s.key, updatedAt: @@ -553,8 +553,12 @@ export function normalizeServerlessConfig( ): ServerlessConfig { return { enabled: config?.enabled ?? false, - sleepAfterSeconds: config?.sleepAfterSeconds ?? 300, - wakeTimeoutSeconds: config?.wakeTimeoutSeconds ?? 300, + sleepAfterSeconds: Math.max( + config?.sleepAfterSeconds ?? DEFAULT_SERVERLESS_SLEEP_AFTER_SECONDS, + MIN_SERVERLESS_SLEEP_AFTER_SECONDS, + ), + wakeTimeoutSeconds: + config?.wakeTimeoutSeconds ?? DEFAULT_SERVERLESS_WAKE_TIMEOUT_SECONDS, }; } @@ -565,8 +569,14 @@ export function getCurrentServerlessConfig(service: { }): ServerlessConfig { return { enabled: service.serverlessEnabled ?? false, - sleepAfterSeconds: service.serverlessSleepAfterSeconds ?? 300, - wakeTimeoutSeconds: service.serverlessWakeTimeoutSeconds ?? 300, + sleepAfterSeconds: Math.max( + service.serverlessSleepAfterSeconds ?? + DEFAULT_SERVERLESS_SLEEP_AFTER_SECONDS, + MIN_SERVERLESS_SLEEP_AFTER_SECONDS, + ), + wakeTimeoutSeconds: + service.serverlessWakeTimeoutSeconds ?? + DEFAULT_SERVERLESS_WAKE_TIMEOUT_SECONDS, }; } diff --git a/web/tests/service-config.test.ts b/web/tests/service-config.test.ts index 98a1c55b..95444145 100644 --- a/web/tests/service-config.test.ts +++ b/web/tests/service-config.test.ts @@ -3,7 +3,9 @@ import { type DeployedConfig, diffConfigs, getDeployedServerlessConfig, + getCurrentServerlessConfig, isDeployedServerlessService, + MIN_SERVERLESS_SLEEP_AFTER_SECONDS, } from "@/lib/service-config"; function deployedConfig( @@ -42,6 +44,33 @@ describe("service config", () => { expect(isDeployedServerlessService(service)).toBe(true); }); + it("enforces the minimum serverless sleep timeout for legacy config", () => { + expect( + getCurrentServerlessConfig({ + serverlessEnabled: true, + serverlessSleepAfterSeconds: 60, + serverlessWakeTimeoutSeconds: 120, + }), + ).toMatchObject({ + sleepAfterSeconds: MIN_SERVERLESS_SLEEP_AFTER_SECONDS, + }); + expect( + getDeployedServerlessConfig({ + deployedConfig: JSON.stringify( + deployedConfig({ + serverless: { + enabled: true, + sleepAfterSeconds: 60, + wakeTimeoutSeconds: 120, + }, + }), + ), + }), + ).toMatchObject({ + sleepAfterSeconds: MIN_SERVERLESS_SLEEP_AFTER_SECONDS, + }); + }); + it("allows deployed stateful services to be serverless", () => { const service = { serverlessEnabled: true, From 51fa1c3a73004bfe6815520f66a3e9e8d3d65e9e Mon Sep 17 00:00:00 2001 From: Techulus Agent <291950465+techulus-agent@users.noreply.github.com> Date: Wed, 8 Jul 2026 16:49:56 +1000 Subject: [PATCH 08/11] Tighten service request metric typography --- .../service/details/service-details-overview.tsx | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/web/components/service/details/service-details-overview.tsx b/web/components/service/details/service-details-overview.tsx index be26b73d..e385f341 100644 --- a/web/components/service/details/service-details-overview.tsx +++ b/web/components/service/details/service-details-overview.tsx @@ -198,13 +198,13 @@ function RequestStatsPanel({
{isLoading ? (
- +
) : hasPublicHttp ? (
-

+

{hasMetricData ? formatCompactNumber(stats.totalRequests) : "-"} @@ -214,7 +214,7 @@ function RequestStatsPanel({

-

+

{hasMetricData ? formatRate(getAverageRequestsPerSecond(stats)) : "-"} @@ -226,7 +226,9 @@ function RequestStatsPanel({

) : ( <> -

-

+

+ - +

no public HTTP ingress

From 5c8386ebe991974f01b329fb3eb3fe3b62cbe091 Mon Sep 17 00:00:00 2001 From: Techulus Agent <291950465+techulus-agent@users.noreply.github.com> Date: Wed, 8 Jul 2026 17:03:45 +1000 Subject: [PATCH 09/11] Show server role in placement UI --- .../service/details/replicas-section.tsx | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/web/components/service/details/replicas-section.tsx b/web/components/service/details/replicas-section.tsx index acb024cf..d11523ff 100644 --- a/web/components/service/details/replicas-section.tsx +++ b/web/components/service/details/replicas-section.tsx @@ -19,16 +19,16 @@ import type { ServiceWithDetails as Service, } from "@/db/types"; -type ServerInfo = Pick; +type ServerInfo = Pick; type ServerWithStatus = ServerInfo & { status: string }; const fetcher = async (url: string): Promise => { const res = await fetch(url); const servers: ServerWithStatus[] = await res.json(); - return servers.map(({ id, name, wireguardIp }) => ({ + return servers.map(({ id, name, isProxy }) => ({ id, name, - wireguardIp, + isProxy, })); }; @@ -143,6 +143,8 @@ export const ReplicasSection = memo(function ReplicasSection({ ? 1 : 0 : Object.values(localReplicas).reduce((sum, n) => sum + n, 0); + const formatServerRole = (server: ServerInfo) => + server.isProxy ? "Proxy + compute node" : "Compute-only node"; if (service.stateful) { return ( @@ -216,13 +218,13 @@ export const ReplicasSection = memo(function ReplicasSection({

{server.name}

- {server.wireguardIp} + {formatServerRole(server)}

@@ -282,8 +284,8 @@ export const ReplicasSection = memo(function ReplicasSection({ >

{server.name}

-

- {server.wireguardIp} +

+ {formatServerRole(server)}

From dd3c405ecc63ac9c09b14738356bd51203890934 Mon Sep 17 00:00:00 2001 From: Techulus Agent <291950465+techulus-agent@users.noreply.github.com> Date: Wed, 8 Jul 2026 17:06:23 +1000 Subject: [PATCH 10/11] Refine placement node role copy --- web/components/service/details/replicas-section.tsx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/web/components/service/details/replicas-section.tsx b/web/components/service/details/replicas-section.tsx index d11523ff..82ea6a21 100644 --- a/web/components/service/details/replicas-section.tsx +++ b/web/components/service/details/replicas-section.tsx @@ -144,7 +144,7 @@ export const ReplicasSection = memo(function ReplicasSection({ : 0 : Object.values(localReplicas).reduce((sum, n) => sum + n, 0); const formatServerRole = (server: ServerInfo) => - server.isProxy ? "Proxy + compute node" : "Compute-only node"; + server.isProxy ? "Proxy node" : "Worker node"; if (service.stateful) { return ( From 8bef3d6a14ef016d2bd927370c51d14e1c10ade2 Mon Sep 17 00:00:00 2001 From: Arjun Komath Date: Wed, 8 Jul 2026 18:18:25 +1000 Subject: [PATCH 11/11] Fix manifest completion race in build workflow --- compose.dev.yml | 5 +- web/.env.example | 3 + .../api/v1/agent/builds/[id]/status/route.ts | 1 + web/lib/inngest/functions/build-workflow.ts | 80 ++++++++++++++++--- web/next.config.ts | 5 ++ 5 files changed, 80 insertions(+), 14 deletions(-) diff --git a/compose.dev.yml b/compose.dev.yml index 4419b83a..070d6302 100644 --- a/compose.dev.yml +++ b/compose.dev.yml @@ -19,8 +19,11 @@ services: build: context: ./registry dockerfile: Dockerfile + environment: + REGISTRY_USERNAME: docker + REGISTRY_PASSWORD: docker ports: - - "5000:5000" + - "5002:5000" volumes: - registry-data:/var/lib/registry healthcheck: diff --git a/web/.env.example b/web/.env.example index 212cc136..24da6e3e 100644 --- a/web/.env.example +++ b/web/.env.example @@ -11,6 +11,9 @@ ENCRYPTION_KEY=your-64-character-hex-string # Public URL APP_URL=http://localhost:3000 +# Local dev origins allowed by Next.js, comma-separated (optional) +NEXT_ALLOWED_DEV_ORIGINS=192.168.1.10,dev.local + # Runtime release tag used for control-plane update checks (tip/dev disables prompts) TECHULUS_CLOUD_VERSION=dev diff --git a/web/app/api/v1/agent/builds/[id]/status/route.ts b/web/app/api/v1/agent/builds/[id]/status/route.ts index da67cf2f..0e53d83e 100644 --- a/web/app/api/v1/agent/builds/[id]/status/route.ts +++ b/web/app/api/v1/agent/builds/[id]/status/route.ts @@ -241,6 +241,7 @@ export async function POST( images: [archImageUri], finalImageUri: baseImageUri, serviceId: shouldDeploy ? build.serviceId : undefined, + buildId, }); await db diff --git a/web/lib/inngest/functions/build-workflow.ts b/web/lib/inngest/functions/build-workflow.ts index d9c1ec43..b66ebb03 100644 --- a/web/lib/inngest/functions/build-workflow.ts +++ b/web/lib/inngest/functions/build-workflow.ts @@ -1,10 +1,46 @@ import { and, eq, isNull } from "drizzle-orm"; import { db } from "@/db"; -import { builds, serviceReplicas, services } from "@/db/schema"; +import { builds, serviceReplicas, services, workQueue } from "@/db/schema"; import { deployServiceInternal } from "@/lib/deploy-service"; import { inngest } from "../client"; import { inngestEvents } from "../events"; +async function hasCompletedManifestWorkItem({ + serviceId, + buildId, + buildGroupId, +}: { + serviceId: string; + buildId?: string; + buildGroupId?: string | null; +}) { + const completedManifestItems = await db + .select({ payload: workQueue.payload }) + .from(workQueue) + .where( + and( + eq(workQueue.type, "create_manifest"), + eq(workQueue.status, "completed"), + ), + ); + + return completedManifestItems.some((item) => { + try { + const payload = JSON.parse(item.payload) as { + serviceId?: string; + buildId?: string; + buildGroupId?: string; + }; + + if (payload.serviceId !== serviceId) return false; + if (buildGroupId) return payload.buildGroupId === buildGroupId; + return !payload.buildGroupId && payload.buildId === buildId; + } catch { + return false; + } + }); +} + export const buildWorkflow = inngest.createFunction( { id: "build-workflow", @@ -42,14 +78,23 @@ export const buildWorkflow = inngest.createFunction( return { status: "failed", reason: result.data.error, buildId }; } - const manifestResult = await step.waitForEvent("wait-manifest", { - event: inngestEvents.manifestCompleted, - timeout: "10m", - if: `async.data.serviceId == "${serviceId}"`, + const manifestAlreadyCompleted = await step.run("check-existing-manifest", async () => { + return hasCompletedManifestWorkItem({ + serviceId, + buildId, + }); }); - if (!manifestResult) { - return { status: "completed_no_manifest", buildId }; + if (!manifestAlreadyCompleted) { + const manifestResult = await step.waitForEvent("wait-manifest", { + event: inngestEvents.manifestCompleted, + timeout: "10m", + if: `async.data.serviceId == "${serviceId}"`, + }); + + if (!manifestResult) { + return { status: "completed_no_manifest", buildId }; + } } const shouldDeploy = await step.run("check-auto-deploy", async () => { @@ -118,14 +163,23 @@ export const buildWorkflow = inngest.createFunction( return { status: "failed", reason: "build_failed", buildGroupId }; } - const manifestResult = await step.waitForEvent("wait-group-manifest", { - event: inngestEvents.manifestCompleted, - timeout: "10m", - if: `async.data.buildGroupId == "${buildGroupId}"`, + const manifestAlreadyCompleted = await step.run("check-existing-group-manifest", async () => { + return hasCompletedManifestWorkItem({ + serviceId, + buildGroupId, + }); }); - if (!manifestResult) { - return { status: "completed_no_manifest", buildGroupId }; + if (!manifestAlreadyCompleted) { + const manifestResult = await step.waitForEvent("wait-group-manifest", { + event: inngestEvents.manifestCompleted, + timeout: "10m", + if: `async.data.buildGroupId == "${buildGroupId}"`, + }); + + if (!manifestResult) { + return { status: "completed_no_manifest", buildGroupId }; + } } const shouldDeploy = await step.run("check-auto-deploy-group", async () => { diff --git a/web/next.config.ts b/web/next.config.ts index 398b0d5c..4629536c 100644 --- a/web/next.config.ts +++ b/web/next.config.ts @@ -1,7 +1,12 @@ import type { NextConfig } from "next"; +const allowedDevOrigins = process.env.NEXT_ALLOWED_DEV_ORIGINS?.split(",") + .map((origin) => origin.trim()) + .filter(Boolean); + const nextConfig: NextConfig = { output: "standalone", + ...(allowedDevOrigins?.length ? { allowedDevOrigins } : {}), }; export default nextConfig;