From d2e8bec95ebcfcc783e17ae968892f08444667a0 Mon Sep 17 00:00:00 2001 From: Arjun Komath Date: Tue, 7 Jul 2026 00:05:03 +1000 Subject: [PATCH] Stabilize serverless sleep wake transitions --- agent/internal/agent/agent.go | 8 +- agent/internal/agent/run.go | 2 +- agent/internal/agent/serverless.go | 115 ++++++++++- agent/internal/agent/serverless_test.go | 121 +++++++++++- agent/internal/http/client.go | 20 +- agent/internal/serverless/gateway.go | 18 +- agent/internal/serverless/gateway_test.go | 85 +++++++++ web/app/api/v1/agent/status/route.ts | 10 +- web/lib/agent-status.ts | 221 +++++++++++++++++++++- web/tests/agent-status.test.ts | 52 +++++ 10 files changed, 611 insertions(+), 41 deletions(-) diff --git a/agent/internal/agent/agent.go b/agent/internal/agent/agent.go index c28e88fa..6a3b4f0b 100644 --- a/agent/internal/agent/agent.go +++ b/agent/internal/agent/agent.go @@ -66,8 +66,8 @@ type Agent struct { deploymentDeployLocks map[string]*sync.Mutex serverlessMutex sync.Mutex pendingServerlessTransitions []agenthttp.ServerlessTransition - pendingServerlessSleep map[string]struct{} - pendingServerlessWake map[string]struct{} + pendingServerlessSleep map[string]serverlessTransitionGuard + pendingServerlessWake map[string]serverlessTransitionGuard expectedStateMutex sync.RWMutex latestExpectedState *agenthttp.ExpectedState Client *agenthttp.Client @@ -120,8 +120,8 @@ func NewAgent( IsProxy: isProxy, DisableDNS: disableDNS, deploymentDeployLocks: map[string]*sync.Mutex{}, - pendingServerlessSleep: map[string]struct{}{}, - pendingServerlessWake: map[string]struct{}{}, + pendingServerlessSleep: map[string]serverlessTransitionGuard{}, + pendingServerlessWake: map[string]serverlessTransitionGuard{}, } } diff --git a/agent/internal/agent/run.go b/agent/internal/agent/run.go index 58ead8ae..9b39c419 100644 --- a/agent/internal/agent/run.go +++ b/agent/internal/agent/run.go @@ -126,7 +126,7 @@ func (a *Agent) reportStatus(reason string) { return } a.ClearReportedDeploymentErrors(reportedDeploymentErrorCount) - a.ClearReportedServerlessTransitions(len(serverlessTransitions)) + a.AcknowledgeServerlessTransitions(response.ServerlessTransitionResults, len(serverlessTransitions)) a.AcknowledgeWorkResults(response.AcceptedWorkItemResults, response.RejectedWorkItemResults) a.LogRejectedActiveWorkItems(response.RejectedActiveWorkItems) a.AcceptLeasedWorkItems(response.WorkItems) diff --git a/agent/internal/agent/serverless.go b/agent/internal/agent/serverless.go index 9b033cc5..7bbbc702 100644 --- a/agent/internal/agent/serverless.go +++ b/agent/internal/agent/serverless.go @@ -1,13 +1,26 @@ package agent import ( + "crypto/rand" + "encoding/hex" + "fmt" "log" "sync" + "sync/atomic" + "time" "techulus/cloud-agent/internal/container" agenthttp "techulus/cloud-agent/internal/http" ) +var serverlessTransitionCounter atomic.Uint64 + +const serverlessTransitionGuardTTL = 2 * time.Minute + +type serverlessTransitionGuard struct { + createdAt time.Time +} + func (a *Agent) SetLatestExpectedState(state *agenthttp.ExpectedState) { a.expectedStateMutex.Lock() defer a.expectedStateMutex.Unlock() @@ -110,21 +123,25 @@ func (a *Agent) QueueServerlessTransition(transition agenthttp.ServerlessTransit if transition.Type == "" || transition.DeploymentID == "" { return } + if transition.ID == "" { + transition.ID = newServerlessTransitionID() + } a.serverlessMutex.Lock() if a.pendingServerlessSleep == nil { - a.pendingServerlessSleep = map[string]struct{}{} + a.pendingServerlessSleep = map[string]serverlessTransitionGuard{} } if a.pendingServerlessWake == nil { - a.pendingServerlessWake = map[string]struct{}{} + a.pendingServerlessWake = map[string]serverlessTransitionGuard{} } + guard := serverlessTransitionGuard{createdAt: time.Now()} switch transition.Type { case "sleep": - a.pendingServerlessSleep[transition.DeploymentID] = struct{}{} + a.pendingServerlessSleep[transition.DeploymentID] = guard delete(a.pendingServerlessWake, transition.DeploymentID) case "wake_started": delete(a.pendingServerlessSleep, transition.DeploymentID) - a.pendingServerlessWake[transition.DeploymentID] = struct{}{} + a.pendingServerlessWake[transition.DeploymentID] = guard case "wake_failed": delete(a.pendingServerlessWake, transition.DeploymentID) } @@ -156,6 +173,51 @@ func (a *Agent) ClearReportedServerlessTransitions(count int) { a.pendingServerlessTransitions = a.pendingServerlessTransitions[count:] } +func (a *Agent) AcknowledgeServerlessTransitions(results []agenthttp.ServerlessTransitionResult, reportedCount int) { + if len(results) == 0 { + a.ClearReportedServerlessTransitions(reportedCount) + return + } + + acknowledged := map[string]agenthttp.ServerlessTransitionResult{} + for _, result := range results { + if result.ID == "" { + continue + } + acknowledged[result.ID] = result + if result.Outcome == "rejected" { + log.Printf( + "[serverless] transition rejected type=%s deployment=%s reason=%s", + result.Type, + Truncate(result.DeploymentID, 8), + result.Reason, + ) + } + } + + a.serverlessMutex.Lock() + defer a.serverlessMutex.Unlock() + + pending := a.pendingServerlessTransitions[:0] + for _, transition := range a.pendingServerlessTransitions { + result, ok := acknowledged[transition.ID] + if !ok { + pending = append(pending, transition) + continue + } + + if result.Outcome == "rejected" { + switch transition.Type { + case "sleep": + delete(a.pendingServerlessSleep, transition.DeploymentID) + case "wake_started", "wake_failed": + delete(a.pendingServerlessWake, transition.DeploymentID) + } + } + } + a.pendingServerlessTransitions = pending +} + func (a *Agent) HasPendingServerlessSleep(deploymentID string) bool { a.serverlessMutex.Lock() defer a.serverlessMutex.Unlock() @@ -219,25 +281,58 @@ func (a *Agent) ReconcilePendingServerlessTransitionsWithExpected(state *agentht } } - for deploymentID := range a.pendingServerlessSleep { + now := time.Now() + for deploymentID, guard := range a.pendingServerlessSleep { if _, stillReporting := pendingSleepTransitions[deploymentID]; stillReporting { continue } - delete(a.pendingServerlessSleep, deploymentID) desiredState, ok := desiredByDeploymentID[deploymentID] + if ok && desiredState == "stopped" { + delete(a.pendingServerlessSleep, deploymentID) + continue + } + if serverlessGuardExpired(guard, now) { + log.Printf("[serverless] sleep transition guard for deployment %s expired after %s; allowing reconcile", Truncate(deploymentID, 8), roundServerlessGuardAge(now.Sub(guard.createdAt))) + delete(a.pendingServerlessSleep, deploymentID) + continue + } if ok && desiredState == "running" { - log.Printf("[serverless] sleep transition for deployment %s is not reflected in expected state; allowing reconcile", Truncate(deploymentID, 8)) + log.Printf("[serverless] sleep transition for deployment %s is not reflected in expected state; keeping reconcile suppressed", Truncate(deploymentID, 8)) } } - for deploymentID := range a.pendingServerlessWake { + for deploymentID, guard := range a.pendingServerlessWake { if _, stillReporting := pendingWakeTransitions[deploymentID]; stillReporting { continue } - delete(a.pendingServerlessWake, deploymentID) desiredState, ok := desiredByDeploymentID[deploymentID] + if ok && desiredState == "running" { + delete(a.pendingServerlessWake, deploymentID) + continue + } + if serverlessGuardExpired(guard, now) { + log.Printf("[serverless] wake transition guard for deployment %s expired after %s; allowing reconcile", Truncate(deploymentID, 8), roundServerlessGuardAge(now.Sub(guard.createdAt))) + delete(a.pendingServerlessWake, deploymentID) + continue + } if !ok || desiredState != "running" { - log.Printf("[serverless] wake transition for deployment %s is not reflected in expected state; allowing reconcile", Truncate(deploymentID, 8)) + log.Printf("[serverless] wake transition for deployment %s is not reflected in expected state; keeping reconcile suppressed", Truncate(deploymentID, 8)) } } } + +func newServerlessTransitionID() string { + var bytes [16]byte + if _, err := rand.Read(bytes[:]); err == nil { + return hex.EncodeToString(bytes[:]) + } + return fmt.Sprintf("fallback-%d", serverlessTransitionCounter.Add(1)) +} + +func serverlessGuardExpired(guard serverlessTransitionGuard, now time.Time) bool { + return !guard.createdAt.IsZero() && now.Sub(guard.createdAt) >= serverlessTransitionGuardTTL +} + +func roundServerlessGuardAge(age time.Duration) time.Duration { + return age.Round(time.Second) +} diff --git a/agent/internal/agent/serverless_test.go b/agent/internal/agent/serverless_test.go index 683570ec..c8ccd874 100644 --- a/agent/internal/agent/serverless_test.go +++ b/agent/internal/agent/serverless_test.go @@ -3,6 +3,7 @@ package agent import ( "slices" "testing" + "time" "techulus/cloud-agent/internal/container" agenthttp "techulus/cloud-agent/internal/http" @@ -11,9 +12,9 @@ import ( func TestPendingServerlessWakeDoesNotStopStaleStoppedExpectedContainer(t *testing.T) { agent := &Agent{ DisableDNS: true, - pendingServerlessSleep: map[string]struct{}{}, - pendingServerlessWake: map[string]struct{}{ - "dep_serverless": {}, + pendingServerlessSleep: map[string]serverlessTransitionGuard{}, + pendingServerlessWake: map[string]serverlessTransitionGuard{ + "dep_serverless": {createdAt: time.Now()}, }, } expected := &agenthttp.ExpectedState{ @@ -66,9 +67,9 @@ func TestServerlessGatewayCapabilityRequiresStartedGateway(t *testing.T) { func TestPendingServerlessWakeDoesNotSuppressContainerReport(t *testing.T) { agent := &Agent{ - pendingServerlessSleep: map[string]struct{}{}, - pendingServerlessWake: map[string]struct{}{ - "dep_serverless": {}, + pendingServerlessSleep: map[string]serverlessTransitionGuard{}, + pendingServerlessWake: map[string]serverlessTransitionGuard{ + "dep_serverless": {createdAt: time.Now()}, }, latestExpectedState: &agenthttp.ExpectedState{ Containers: []agenthttp.ExpectedContainer{ @@ -84,3 +85,111 @@ func TestPendingServerlessWakeDoesNotSuppressContainerReport(t *testing.T) { t.Fatal("container report was suppressed while wake transition is pending") } } + +func TestAcceptedSleepKeepsGuardUntilExpectedStateStops(t *testing.T) { + agent := &Agent{ + pendingServerlessSleep: map[string]serverlessTransitionGuard{}, + pendingServerlessWake: map[string]serverlessTransitionGuard{}, + } + agent.QueueServerlessTransition(agenthttp.ServerlessTransition{ + Type: "sleep", + DeploymentID: "dep_serverless", + ContainerID: "ctr_serverless", + }) + transitions := agent.SnapshotServerlessTransitions() + if len(transitions) != 1 || transitions[0].ID == "" { + t.Fatalf("transitions = %+v, want one transition with id", transitions) + } + + agent.AcknowledgeServerlessTransitions([]agenthttp.ServerlessTransitionResult{ + { + ID: transitions[0].ID, + Type: "sleep", + DeploymentID: "dep_serverless", + Outcome: "applied", + }, + }, len(transitions)) + + agent.ReconcilePendingServerlessTransitionsWithExpected(expectedServerlessState("running"), false) + if !agent.HasPendingServerlessSleep("dep_serverless") { + t.Fatal("sleep guard was cleared before expected state stopped") + } + + agent.ReconcilePendingServerlessTransitionsWithExpected(expectedServerlessState("stopped"), false) + if agent.HasPendingServerlessSleep("dep_serverless") { + t.Fatal("sleep guard was not cleared after expected state stopped") + } +} + +func TestRejectedSleepClearsGuard(t *testing.T) { + agent := &Agent{ + pendingServerlessSleep: map[string]serverlessTransitionGuard{}, + pendingServerlessWake: map[string]serverlessTransitionGuard{}, + } + agent.QueueServerlessTransition(agenthttp.ServerlessTransition{ + Type: "sleep", + DeploymentID: "dep_serverless", + ContainerID: "ctr_serverless", + }) + transitions := agent.SnapshotServerlessTransitions() + + agent.AcknowledgeServerlessTransitions([]agenthttp.ServerlessTransitionResult{ + { + ID: transitions[0].ID, + Type: "sleep", + DeploymentID: "dep_serverless", + Outcome: "rejected", + Reason: "deployment is not sleepable from starting", + }, + }, len(transitions)) + + if agent.HasPendingServerlessSleep("dep_serverless") { + t.Fatal("sleep guard was not cleared after explicit rejection") + } + if remaining := agent.SnapshotServerlessTransitions(); len(remaining) != 0 { + t.Fatalf("remaining transitions = %+v, want none", remaining) + } +} + +func TestSleepGuardExpiresWhenExpectedStateStaysRunning(t *testing.T) { + agent := &Agent{ + pendingServerlessSleep: map[string]serverlessTransitionGuard{ + "dep_serverless": { + createdAt: time.Now().Add(-serverlessTransitionGuardTTL - time.Second), + }, + }, + pendingServerlessWake: map[string]serverlessTransitionGuard{}, + } + + agent.ReconcilePendingServerlessTransitionsWithExpected(expectedServerlessState("running"), false) + if agent.HasPendingServerlessSleep("dep_serverless") { + t.Fatal("expired sleep guard was not cleared") + } +} + +func TestWakeGuardExpiresWhenExpectedStateStaysStopped(t *testing.T) { + agent := &Agent{ + pendingServerlessSleep: map[string]serverlessTransitionGuard{}, + pendingServerlessWake: map[string]serverlessTransitionGuard{ + "dep_serverless": { + createdAt: time.Now().Add(-serverlessTransitionGuardTTL - time.Second), + }, + }, + } + + agent.ReconcilePendingServerlessTransitionsWithExpected(expectedServerlessState("stopped"), false) + if agent.HasPendingServerlessWake("dep_serverless") { + t.Fatal("expired wake guard was not cleared") + } +} + +func expectedServerlessState(desiredState string) *agenthttp.ExpectedState { + return &agenthttp.ExpectedState{ + Containers: []agenthttp.ExpectedContainer{ + { + DeploymentID: "dep_serverless", + DesiredState: desiredState, + }, + }, + } +} diff --git a/agent/internal/http/client.go b/agent/internal/http/client.go index ec529e9a..9d513fbc 100644 --- a/agent/internal/http/client.go +++ b/agent/internal/http/client.go @@ -299,6 +299,7 @@ type ActiveWorkItem struct { } type ServerlessTransition struct { + ID string `json:"id,omitempty"` Type string `json:"type"` DeploymentID string `json:"deploymentId"` ContainerID string `json:"containerId,omitempty"` @@ -401,11 +402,20 @@ type RejectedWorkItemResult struct { } type StatusResponse struct { - OK bool `json:"ok"` - AcceptedWorkItemResults []string `json:"acceptedWorkItemResults"` - RejectedWorkItemResults []RejectedWorkItemResult `json:"rejectedWorkItemResults"` - RejectedActiveWorkItems []RejectedWorkItemResult `json:"rejectedActiveWorkItems"` - WorkItems []WorkQueueItem `json:"workItems"` + OK bool `json:"ok"` + AcceptedWorkItemResults []string `json:"acceptedWorkItemResults"` + RejectedWorkItemResults []RejectedWorkItemResult `json:"rejectedWorkItemResults"` + RejectedActiveWorkItems []RejectedWorkItemResult `json:"rejectedActiveWorkItems"` + ServerlessTransitionResults []ServerlessTransitionResult `json:"serverlessTransitionResults"` + WorkItems []WorkQueueItem `json:"workItems"` +} + +type ServerlessTransitionResult struct { + ID string `json:"id,omitempty"` + Type string `json:"type,omitempty"` + DeploymentID string `json:"deploymentId,omitempty"` + Outcome string `json:"outcome"` + Reason string `json:"reason,omitempty"` } func (c *Client) ReportStatus(report *StatusReport, completed []CompletedWorkItem, active []ActiveWorkItem, serverlessTransitions []ServerlessTransition) (*StatusResponse, error) { diff --git a/agent/internal/serverless/gateway.go b/agent/internal/serverless/gateway.go index 163ddbcc..4d8f8c18 100644 --- a/agent/internal/serverless/gateway.go +++ b/agent/internal/serverless/gateway.go @@ -27,6 +27,8 @@ const ( var ( wakePollInterval = 500 * time.Millisecond idleTimerSeedInterval = 15 * time.Second + upstreamDialTimeout = 250 * time.Millisecond + checkUpstreamReady = tcpUpstreamReady ) type Gateway struct { @@ -48,6 +50,7 @@ type Runtime interface { ListServerlessContainers() ([]container.Container, error) GetServerlessContainerHealth(containerID string) string QueueServerlessTransition(agenthttp.ServerlessTransition) + HasPendingServerlessSleep(deploymentID string) bool } type cachedUpstreams struct { @@ -293,11 +296,13 @@ func (g *Gateway) readyUpstreams(route *agenthttp.ServerlessRoute, state *agenth actual, isRunning := actualByDeploymentID[deploymentID] if isRunning && g.isContainerReady(actual, expected) { if upstream, ok := localUpstream(route, expected); ok { - upstreamsByURL[upstream.Url] = upstream + if checkUpstreamReady(upstream.Url) { + upstreamsByURL[upstream.Url] = upstream + } } continue } - if expected.DesiredState == "stopped" { + if expected.DesiredState == "stopped" || g.runtime.HasPendingServerlessSleep(deploymentID) { sleepingLocalIDs = append(sleepingLocalIDs, deploymentID) } } @@ -490,6 +495,15 @@ func (g *Gateway) isContainerReady(actual container.Container, expected agenthtt return health == "healthy" || health == "none" } +func tcpUpstreamReady(address string) bool { + conn, err := net.DialTimeout("tcp", address, upstreamDialTimeout) + if err != nil { + return false + } + conn.Close() + return true +} + func (g *Gateway) cachedUpstreams(host string) ([]agenthttp.ServerlessUpstream, bool) { g.mu.Lock() defer g.mu.Unlock() diff --git a/agent/internal/serverless/gateway_test.go b/agent/internal/serverless/gateway_test.go index 5bf165b2..c39123bb 100644 --- a/agent/internal/serverless/gateway_test.go +++ b/agent/internal/serverless/gateway_test.go @@ -11,6 +11,10 @@ import ( agenthttp "techulus/cloud-agent/internal/http" ) +func init() { + checkUpstreamReady = func(string) bool { return true } +} + type fakeRuntime struct { mu sync.Mutex state *agenthttp.ExpectedState @@ -24,6 +28,7 @@ type fakeRuntime struct { deployStarted chan struct{} deployStartedOnce sync.Once allowDeploy chan struct{} + pendingSleeps map[string]bool } func (f *fakeRuntime) ExpectedState() *agenthttp.ExpectedState { @@ -96,6 +101,21 @@ func (f *fakeRuntime) QueueServerlessTransition(transition agenthttp.ServerlessT f.mu.Lock() defer f.mu.Unlock() f.transitions = append(f.transitions, transition) + if transition.Type == "sleep" { + if f.pendingSleeps == nil { + f.pendingSleeps = map[string]bool{} + } + f.pendingSleeps[transition.DeploymentID] = true + } + if transition.Type == "wake_started" { + delete(f.pendingSleeps, transition.DeploymentID) + } +} + +func (f *fakeRuntime) HasPendingServerlessSleep(deploymentID string) bool { + f.mu.Lock() + defer f.mu.Unlock() + return f.pendingSleeps[deploymentID] } func (f *fakeRuntime) snapshot() ([]agenthttp.ServerlessTransition, []string, int) { @@ -333,6 +353,38 @@ func TestSleepHostUsesServiceActivityAcrossDomains(t *testing.T) { } } +func TestPendingSleepCanWakeBeforeExpectedStateSettles(t *testing.T) { + state := testExpectedState("running") + state.Serverless.Routes[0].Upstreams = nil + runtime := &fakeRuntime{ + state: state, + containers: []container.Container{ + {ID: "ctr-local", State: "exited", DeploymentID: "dep_local", ServiceID: "svc_1"}, + }, + pendingSleeps: map[string]bool{"dep_local": true}, + } + gateway := NewGateway(runtime) + + upstreams, err := gateway.getUpstreams(context.Background(), "app.example.com") + if err != nil { + t.Fatalf("getUpstreams returned error: %v", err) + } + if len(upstreams) != 1 || upstreams[0].DeploymentID != "dep_local" { + t.Fatalf("upstreams = %+v, want local dep_local", upstreams) + } + + transitions, _, deployCalls := runtime.snapshot() + if deployCalls != 1 { + t.Fatalf("deployCalls = %d, want 1", deployCalls) + } + if len(transitions) != 1 || transitions[0].Type != "wake_started" { + t.Fatalf("transitions = %+v, want one wake_started transition", transitions) + } + if runtime.HasPendingServerlessSleep("dep_local") { + t.Fatal("pending sleep was not cleared by wake_started") + } +} + func TestWakeTimeoutQueuesWakeFailedTransition(t *testing.T) { previousPoll := wakePollInterval wakePollInterval = time.Millisecond @@ -375,6 +427,39 @@ func TestWakeTimeoutQueuesWakeFailedTransition(t *testing.T) { } } +func TestWakeTimeoutWhenLocalPortNeverOpens(t *testing.T) { + previousPoll := wakePollInterval + previousReady := checkUpstreamReady + wakePollInterval = time.Millisecond + checkUpstreamReady = func(string) bool { return false } + t.Cleanup(func() { + wakePollInterval = previousPoll + checkUpstreamReady = previousReady + }) + + 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") + } + + 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) + } +} + func testExpectedState(localDesiredState string) *agenthttp.ExpectedState { state := &agenthttp.ExpectedState{ Containers: []agenthttp.ExpectedContainer{ diff --git a/web/app/api/v1/agent/status/route.ts b/web/app/api/v1/agent/status/route.ts index cf1ce4b6..a338f4ab 100644 --- a/web/app/api/v1/agent/status/route.ts +++ b/web/app/api/v1/agent/status/route.ts @@ -2,7 +2,6 @@ import { type NextRequest, NextResponse } from "next/server"; import { verifyAgentRequest } from "@/lib/agent-auth"; import { applyStatusReport, - type ServerlessTransition, type StatusReport, } from "@/lib/agent-status"; import { @@ -17,7 +16,7 @@ type StatusRequestBody = { statusReport?: StatusReport; completedWorkItems?: WorkItemResult[]; activeWorkItems?: ActiveWorkItem[]; - serverlessTransitions?: ServerlessTransition[]; + serverlessTransitions?: unknown[]; }; export async function POST(request: NextRequest) { @@ -47,7 +46,11 @@ export async function POST(request: NextRequest) { ? data.serverlessTransitions : []; - await applyStatusReport(serverId, data.statusReport, serverlessTransitions); + const { serverlessTransitionResults } = await applyStatusReport( + serverId, + data.statusReport, + serverlessTransitions, + ); const completedWorkItems = Array.isArray(data.completedWorkItems) ? data.completedWorkItems.filter(isValidWorkItemResult) @@ -71,6 +74,7 @@ export async function POST(request: NextRequest) { acceptedWorkItemResults: accepted, rejectedWorkItemResults: rejected, rejectedActiveWorkItems: rejectedActive, + serverlessTransitionResults, workItems: nextWorkItem ? [nextWorkItem] : [], }); } diff --git a/web/lib/agent-status.ts b/web/lib/agent-status.ts index d2a4a969..78bd7d6e 100644 --- a/web/lib/agent-status.ts +++ b/web/lib/agent-status.ts @@ -46,9 +46,17 @@ type DeploymentError = { }; export type ServerlessTransition = - | { type: "sleep"; deploymentId: string; containerId: string } - | { type: "wake_started"; deploymentId: string } - | { type: "wake_failed"; deploymentId: string; error: string }; + | { id?: string; type: "sleep"; deploymentId: string; containerId: string } + | { id?: string; type: "wake_started"; deploymentId: string } + | { id?: string; type: "wake_failed"; deploymentId: string; error: string }; + +export type ServerlessTransitionResult = { + id?: string; + type?: ServerlessTransition["type"]; + deploymentId?: string; + outcome: "applied" | "already_applied" | "rejected"; + reason?: string; +}; export function shouldAttachReportedContainer(observedPhase: ObservedPhase) { return (observedStartingPhases as readonly ObservedPhase[]).includes( @@ -73,6 +81,26 @@ export function getStoppedContainerReportUpdate(deployment: { }; } +export function getStaleStoppedServerlessReportUpdate({ + hasHealthCheck, + healthStatus, +}: { + hasHealthCheck: boolean; + healthStatus: ContainerStatus["healthStatus"]; +}) { + const recovered = healthStatus === "healthy" || healthStatus === "none"; + + return { + observedPhase: recovered ? ("healthy" as const) : ("starting" as const), + healthStatus: hasHealthCheck + ? recovered + ? healthStatus + : ("starting" as const) + : ("none" as const), + serverlessWakeFailureCount: 0, + }; +} + function isMigrationTargetStarting(status: string | null | undefined) { return status === "deploying_target" || status === "starting"; } @@ -217,13 +245,28 @@ async function applyDeploymentErrors( async function applyServerlessTransitions( serverId: string, - transitions: ServerlessTransition[], -) { - for (const transition of transitions) { - if (!isValidServerlessTransition(transition)) { + transitions: unknown[], +): Promise { + const results: ServerlessTransitionResult[] = []; + + for (const transitionValue of transitions) { + const resultBase = getServerlessTransitionResultBase(transitionValue); + if (!isValidServerlessTransition(transitionValue)) { console.log(`[serverless:status] rejected malformed transition`); + results.push({ + ...resultBase, + outcome: "rejected", + reason: "malformed transition", + }); continue; } + const transition = transitionValue; + + const validResultBase = { + id: transition.id, + type: transition.type, + deploymentId: transition.deploymentId, + }; const deployment = await db .select({ @@ -259,6 +302,13 @@ async function applyServerlessTransitions( console.log( `[serverless:status] rejected ${transition.type} for deployment ${transition.deploymentId}: ${invalidReason}`, ); + results.push({ + ...validResultBase, + outcome: isAlreadyAppliedServerlessTransition(transition, deployment) + ? "already_applied" + : "rejected", + reason: invalidReason ?? "deployment not found", + }); continue; } @@ -291,6 +341,22 @@ async function applyServerlessTransitions( transition.deploymentId, deployment.serviceId, ); + results.push({ ...validResultBase, outcome: "applied" }); + } else { + const outcome = isAlreadyAppliedServerlessTransition( + transition, + deployment, + ) + ? "already_applied" + : "rejected"; + const reason = + outcome === "already_applied" + ? "sleep already applied" + : "sleep update matched zero rows"; + console.log( + `[serverless:status] ${outcome} ${transition.type} for deployment ${transition.deploymentId}: ${reason}`, + ); + results.push({ ...validResultBase, outcome, reason }); } continue; } @@ -323,6 +389,22 @@ async function applyServerlessTransitions( transition.deploymentId, deployment.serviceId, ); + results.push({ ...validResultBase, outcome: "applied" }); + } else { + const outcome = isAlreadyAppliedServerlessTransition( + transition, + deployment, + ) + ? "already_applied" + : "rejected"; + const reason = + outcome === "already_applied" + ? "wake already applied" + : "wake update matched zero rows"; + console.log( + `[serverless:status] ${outcome} ${transition.type} for deployment ${transition.deploymentId}: ${reason}`, + ); + results.push({ ...validResultBase, outcome, reason }); } continue; } @@ -354,8 +436,92 @@ async function applyServerlessTransitions( transition.deploymentId, deployment.serviceId, ); + results.push({ ...validResultBase, outcome: "applied" }); + } else { + const outcome = isAlreadyAppliedServerlessTransition( + transition, + deployment, + ) + ? "already_applied" + : "rejected"; + const reason = + outcome === "already_applied" + ? "wake failure already applied" + : "wake_failed update matched zero rows"; + console.log( + `[serverless:status] ${outcome} ${transition.type} for deployment ${transition.deploymentId}: ${reason}`, + ); + results.push({ ...validResultBase, outcome, reason }); } } + + return results; +} + +function getServerlessTransitionResultBase( + transition: unknown, +): Omit { + if (!transition || typeof transition !== "object") { + return {}; + } + const candidate = transition as Record; + return { + id: typeof candidate.id === "string" ? candidate.id : undefined, + type: isServerlessTransitionType(candidate.type) + ? candidate.type + : undefined, + deploymentId: + typeof candidate.deploymentId === "string" + ? candidate.deploymentId + : undefined, + }; +} + +function isServerlessTransitionType( + value: unknown, +): value is ServerlessTransition["type"] { + return value === "sleep" || value === "wake_started" || value === "wake_failed"; +} + +export function getSleepTransitionDeploymentIds( + transitions: unknown[], +): Set { + return new Set( + transitions + .filter(isValidServerlessTransition) + .filter((transition) => transition.type === "sleep") + .map((transition) => transition.deploymentId), + ); +} + +function isAlreadyAppliedServerlessTransition( + transition: ServerlessTransition, + deployment: + | { + runtimeDesiredState: string; + observedPhase: string; + } + | undefined, +) { + if (!deployment) return false; + if (transition.type === "sleep") { + return ( + deployment.runtimeDesiredState === "stopped" && + deployment.observedPhase === "sleeping" + ); + } + if (transition.type === "wake_started") { + return ( + deployment.runtimeDesiredState === "running" && + ["waking", "starting", "healthy", "running"].includes( + deployment.observedPhase, + ) + ); + } + if (transition.type === "wake_failed") { + return ["sleeping", "failed"].includes(deployment.observedPhase); + } + return false; } function isValidServerlessTransition( @@ -479,7 +645,7 @@ export type StatusReport = { export async function applyStatusReport( serverId: string, report: StatusReport, - serverlessTransitions: ServerlessTransition[] = [], + serverlessTransitions: unknown[] = [], ) { const updateData: Record = { lastHeartbeat: new Date(), @@ -559,7 +725,12 @@ export async function applyStatusReport( }; await applyDeploymentErrors(serverId, report.deploymentErrors || []); - await applyServerlessTransitions(serverId, serverlessTransitions); + const serverlessTransitionResults = await applyServerlessTransitions( + serverId, + serverlessTransitions, + ); + const sleepTransitionDeploymentIds = + getSleepTransitionDeploymentIds(serverlessTransitions); const reportedDeploymentIds = report.containers .map((c) => c.deploymentId) @@ -590,7 +761,10 @@ export async function applyStatusReport( .delete(deploymentPorts) .where(eq(deploymentPorts.deploymentId, dep.id)); await db.delete(deployments).where(eq(deployments.id, dep.id)); - } else if (isObservedActiveContainer(dep.observedPhase)) { + } else if ( + isObservedActiveContainer(dep.observedPhase) && + !sleepTransitionDeploymentIds.has(dep.id) + ) { console.log( `[status:${serverId.slice(0, 8)}] deployment ${dep.id.slice(0, 8)} NOT reported, marking UNKNOWN`, ); @@ -727,6 +901,31 @@ export async function applyStatusReport( continue; } + if ( + container.status === "running" && + deployment.runtimeDesiredState === "running" && + ["sleeping", "stopped"].includes(deployment.observedPhase) + ) { + const service = await db + .select() + .from(services) + .where(eq(services.id, deployment.serviceId)) + .then((r) => r[0]); + + if (service && getDeployedServerlessConfig(service).enabled) { + Object.assign( + updateFields, + getStaleStoppedServerlessReportUpdate({ + hasHealthCheck: service.healthCheckCmd != null, + healthStatus, + }), + ); + console.log( + `[health:restore] serverless deployment ${deployment.id} restored from ${deployment.observedPhase} to ${updateFields.observedPhase}`, + ); + } + } + if (shouldAttachReportedContainer(deployment.observedPhase)) { if (container.status !== "running") { continue; @@ -1025,6 +1224,8 @@ export async function applyStatusReport( ); } } + + return { serverlessTransitionResults }; } function prepareAutohealRecreatePayload({ diff --git a/web/tests/agent-status.test.ts b/web/tests/agent-status.test.ts index b3df6253..ea56421c 100644 --- a/web/tests/agent-status.test.ts +++ b/web/tests/agent-status.test.ts @@ -18,6 +18,8 @@ vi.mock("@/lib/work-queue", () => ({ })); import { + getSleepTransitionDeploymentIds, + getStaleStoppedServerlessReportUpdate, getStoppedContainerReportUpdate, shouldAttachReportedContainer, } from "@/lib/agent-status"; @@ -47,4 +49,54 @@ describe("agent status serverless attachment", () => { healthStatus: "none", }); }); + + it("restores stale stopped serverless observations from live running reports", () => { + expect( + getStaleStoppedServerlessReportUpdate({ + hasHealthCheck: false, + healthStatus: "none", + }), + ).toEqual({ + observedPhase: "healthy", + healthStatus: "none", + serverlessWakeFailureCount: 0, + }); + + expect( + getStaleStoppedServerlessReportUpdate({ + hasHealthCheck: true, + healthStatus: "starting", + }), + ).toEqual({ + observedPhase: "starting", + healthStatus: "starting", + serverlessWakeFailureCount: 0, + }); + + expect( + getStaleStoppedServerlessReportUpdate({ + hasHealthCheck: true, + healthStatus: "healthy", + }), + ).toEqual({ + observedPhase: "healthy", + healthStatus: "healthy", + serverlessWakeFailureCount: 0, + }); + }); + + it("extracts sleep transition ids without trusting raw payload shape", () => { + expect( + Array.from( + getSleepTransitionDeploymentIds([ + null, + 42, + { type: "sleep", deploymentId: "", containerId: "ctr_empty" }, + { type: "sleep", deploymentId: "dep_sleep", containerId: "ctr_sleep" }, + { type: "wake_started", deploymentId: "dep_wake" }, + { type: "sleep", deploymentId: "dep_missing_container" }, + ]), + ), + ).toEqual(["dep_sleep"]); + }); });