Release v0.26.0

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{},
 	}
 }
 

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)

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)
+}

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,
+			},
+		},
+	}
+}

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) {