Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions agentplane/runner_gateway/environments/lifecycle/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,9 @@ Resume tokens bind the Sandbox namespace, name, UID and observed wake generation
preconditions, not capabilities. Writes use resource-version optimistic locking and recheck policy,
readiness and live Identity incarnations after conflicts. A stale generation returns state changed.
An error after a patch may have reached the API server is reported as an unknown outcome and is never
retried by the service. A shared `WakeBudget` reserves a bounded attempt keyed by effective caller
identity, Sandbox UID and next generation.
retried by the service. Automatic `RequestWake` reserves a bounded attempt from the shared
`WakeBudget`, keyed by effective caller identity, Sandbox UID and next generation. Explicit
`ResumeEnvironment` does not reserve wake budget.

`REQUESTED` reports that desired state was accepted. The Sandbox Controller owns stopping, startup,
readiness and execution-incarnation transitions; lifecycle acceptance does not imply readiness.
Expand Down
9 changes: 5 additions & 4 deletions agentplane/runner_gateway/environments/lifecycle/SPEC.md
Original file line number Diff line number Diff line change
Expand Up @@ -53,10 +53,11 @@ The manager rereads and reauthorizes on every resource-version conflict, includi
the live caller Identity and the target's live bound Identity when a grant selector uses it. A stale
token returns `STATE_CHANGED` with current eligibility and generation rather than retrying against a
later suspension. It bounds conflict retries. A patch error may have uncertain effect, so the manager
returns a typed `OUTCOME_UNKNOWN` status and does not replay the patch. The shared wake budget is
reserved before patching and is keyed by effective caller namespace/name/UID (an Identity by the
manager's namespace, its name and UID, or a service by exact SPIFFE ID), target Sandbox UID and next
generation.
returns a typed `OUTCOME_UNKNOWN` status and does not replay the patch. Only automatic `RequestWake`
reserves the shared wake budget before patching, keyed by effective caller namespace/name/UID
(an Identity by the manager's namespace, its name and UID, or a service by exact SPIFFE ID),
target Sandbox UID and next generation. Explicit `ResumeEnvironment`, for both operator callers
and Identity grants, does not reserve budget and is unaffected by budget exhaustion or unavailability.

`REQUESTED` means the desired running state and next wake generation were accepted. It does not mean
the environment is ready. `ALREADY_RUNNING` means no wake mutation was needed. A current-generation
Expand Down
16 changes: 8 additions & 8 deletions agentplane/runner_gateway/environments/lifecycle/manager.go
Original file line number Diff line number Diff line change
Expand Up @@ -189,8 +189,6 @@ func (s *Manager) ResumeEnvironment(
return resumeResponse(target, assessment.GetTargetGeneration(), environmentpb.ResumeOutcome_RESUME_OUTCOME_REQUESTED, eligibility, assessment), nil
case environmentlifecyclepb.WakeState_WAKE_STATE_NOT_NEEDED:
return resumeResponse(target, assessment.GetTargetGeneration(), environmentpb.ResumeOutcome_RESUME_OUTCOME_ALREADY_RUNNING, eligibility, assessment), nil
case environmentlifecyclepb.WakeState_WAKE_STATE_RATE_LIMITED:
return nil, rpcWithAssessment(codes.ResourceExhausted, "resume wake budget exhausted", assessment)
case environmentlifecyclepb.WakeState_WAKE_STATE_DENIED:
current, err := s.readTarget(ctx, target)
if err != nil {
Expand Down Expand Up @@ -281,12 +279,14 @@ func (s *Manager) applyResume(
}

nextGeneration := view.Spec.WakeGeneration + 1
delay, err := s.wakeBudget.ReserveWakeAttempt(ctx, s.budgetKey(caller), target.Uid, nextGeneration)
if err != nil {
return failureAssessment(codes.Unavailable, "shared wake budget is unavailable", environmentlifecyclepb.WakeReason_WAKE_REASON_BACKEND_STATE_UNAVAILABLE, view.Spec.WakeGeneration), nil
}
if delay > 0 {
return rateLimited(delay, view.Spec.WakeGeneration), nil
if automatic {
delay, err := s.wakeBudget.ReserveWakeAttempt(ctx, s.budgetKey(caller), target.Uid, nextGeneration)
if err != nil {
return failureAssessment(codes.Unavailable, "shared wake budget is unavailable", environmentlifecyclepb.WakeReason_WAKE_REASON_BACKEND_STATE_UNAVAILABLE, view.Spec.WakeGeneration), nil
}
if delay > 0 {
return rateLimited(delay, view.Spec.WakeGeneration), nil
}
}
updated := view.DeepCopy()
updated.Spec.Running = true
Expand Down
86 changes: 56 additions & 30 deletions agentplane/runner_gateway/environments/lifecycle/service_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -86,13 +86,10 @@ func (r testReviews) Create(_ context.Context, request *authenticationv1.TokenRe
}

type testWakeBudget struct {
mu sync.Mutex
delay time.Duration
err error
calls []budgetCall
barrier int
arrived int
release chan struct{}
mu sync.Mutex
delay time.Duration
err error
calls []budgetCall
}

type budgetCall struct {
Expand All @@ -103,20 +100,9 @@ type budgetCall struct {

func (b *testWakeBudget) ReserveWakeAttempt(_ context.Context, caller lifecycle.WakeBudgetKey, target string, generation int64) (time.Duration, error) {
b.mu.Lock()
defer b.mu.Unlock()
b.calls = append(b.calls, budgetCall{caller: caller, target: target, gen: generation})
delay, err := b.delay, b.err
if b.barrier == 0 {
b.mu.Unlock()
return delay, err
}
b.arrived++
if b.arrived == b.barrier {
close(b.release)
}
release := b.release
b.mu.Unlock()
<-release
return delay, err
return b.delay, b.err
}

type testPatchClient struct {
Expand Down Expand Up @@ -340,7 +326,7 @@ func TestOwnExecutionAccountCannotResumeIdleSuspendedSandbox(t *testing.T) {
}
}

func TestIdentityGrantCanResumeForeignSandboxAndUsesIdentityBudget(t *testing.T) {
func TestIdentityGrantCanResumeForeignSandboxWithoutWakeBudget(t *testing.T) {
sandbox := suspendedSandbox(&sandboxapi.ResumePolicy{}, true, sandboxapi.SuspendedByIdle, 4)
sandbox.Spec.Identity = &sandboxapi.IdentityRef{Name: "target-agent"}
objects := testObjects(t, sandbox, executionUID)
Expand All @@ -360,8 +346,8 @@ func TestIdentityGrantCanResumeForeignSandboxAndUsesIdentityBudget(t *testing.T)
response.GetEligibility().GetReason() != environmentlifecyclepb.ResumeEligibilityReason_RESUME_ELIGIBILITY_REASON_ALLOWED_IDENTITY_GRANT {
t.Fatalf("granted foreign resume = %+v, %v", response, err)
}
if len(h.budget.calls) != 1 || h.budget.calls[0].caller != (lifecycle.WakeBudgetKey{Namespace: testNamespace, Name: executionIdentityName, UID: executionIdentityUID}) {
t.Fatalf("resume budget caller = %+v, want effective Identity", h.budget.calls)
if len(h.budget.calls) != 0 {
t.Fatalf("explicit Identity resume reserved wake budget: %+v", h.budget.calls)
}
}

Expand Down Expand Up @@ -686,6 +672,35 @@ func TestDirectPodBoundOperatorCanResumeRequestedStopWithoutToken(t *testing.T)
}
}

func TestExplicitResumeDoesNotReserveWakeBudget(t *testing.T) {
for _, operator := range []bool{true, false} {
for _, unavailable := range []bool{false, true} {
t.Run(fmt.Sprintf("operator=%t/budgetUnavailable=%t", operator, unavailable), func(t *testing.T) {
objects := testIdentityObjects(t, suspendedSandbox(&sandboxapi.ResumePolicy{}, true, sandboxapi.SuspendedByIdle, 4), executionUID)
identity := &policyapi.Identity{ObjectMeta: metav1.ObjectMeta{Namespace: testNamespace, Name: executionIdentityName, UID: types.UID(executionIdentityUID)}}
rules := identityResumeRules(identity, []policyapi.EnvironmentSelector{{BoundIdentity: &policyapi.BoundIdentitySelector{Self: true}}}, nil)
h := newIdentityHarness(t, objects, []auth.Account{operatorAccount}, rules, func() error { return nil })
if unavailable {
h.budget.err = errors.New("budget storage unavailable")
} else {
h.budget.delay = 30 * time.Second
}
ctx := delegatedContext(t)
if operator {
ctx = directContext(t, "operator")
}
response, err := environmentpb.NewEnvironmentServiceClient(h.conn).ResumeEnvironment(ctx, resumeRequest(testSandboxUID, 4))
if err != nil || response.GetOutcome() != environmentpb.ResumeOutcome_RESUME_OUTCOME_REQUESTED || response.GetWakeGeneration() != 5 {
t.Fatalf("explicit resume with exhausted/unavailable budget = %+v, %v", response, err)
}
if len(h.budget.calls) != 0 {
t.Fatalf("explicit resume reserved wake budget: %+v", h.budget.calls)
}
})
}
}
}

func TestIncompleteCurrentGenerationIdleSuspensionIsStillStopping(t *testing.T) {
t.Run("current generation is still stopping", func(t *testing.T) {
sandbox := suspendedSandbox(&sandboxapi.ResumePolicy{}, true, sandboxapi.SuspendedByIdle, 4)
Expand Down Expand Up @@ -746,17 +761,28 @@ func TestOldGenerationDoesNotAdvanceASecondWakeAndConcurrentConflictRereads(t *t
if err != nil || second.GetOutcome() != environmentpb.ResumeOutcome_RESUME_OUTCOME_STATE_CHANGED || second.GetWakeGeneration() != 4 {
t.Fatalf("stale resume = %+v, %v", second, err)
}
if len(h.budget.calls) != 1 {
t.Fatalf("old token consumed %d budget reservations, want 1", len(h.budget.calls))
if len(h.budget.calls) != 0 {
t.Fatalf("explicit resume reserved wake budget: %+v", h.budget.calls)
}
})
t.Run("two simultaneous requests converge on one generation", func(t *testing.T) {
sandbox := suspendedSandbox(&sandboxapi.ResumePolicy{AllowOwnServiceAccount: true}, true, sandboxapi.SuspendedByIdle, 8)
sandbox.ResourceVersion = "1"
objects := testObjects(t, sandbox, executionUID)
base := testObjects(t, sandbox, executionUID)
var mu sync.Mutex
arrived := 0
release := make(chan struct{})
objects := testPatchClient{WithWatch: base, patch: func(ctx context.Context, object client.Object, patch client.Patch, options ...client.PatchOption) error {
mu.Lock()
arrived++
if arrived == 2 {
close(release)
}
mu.Unlock()
<-release
return base.Patch(ctx, object, patch, options...)
}}
h := newHarness(t, objects, []auth.Account{operatorAccount})
h.budget.barrier = 2
h.budget.release = make(chan struct{})
client := environmentpb.NewEnvironmentServiceClient(h.conn)
type result struct {
response *environmentpb.ResumeEnvironmentResponse
Expand Down Expand Up @@ -807,8 +833,8 @@ func TestOldGenerationDoesNotAdvanceASecondWakeAndConcurrentConflictRereads(t *t
if err != nil || response.GetOutcome() != environmentpb.ResumeOutcome_RESUME_OUTCOME_REQUESTED || response.GetWakeGeneration() != 6 {
t.Fatalf("resume after conflict = %+v, %v", response, err)
}
if conflicts != 1 || len(h.budget.calls) != 2 || h.budget.calls[0].gen != 6 || h.budget.calls[1].gen != 6 {
t.Fatalf("conflict retries=%d, budget calls=%+v; want two reservations for generation 6", conflicts, h.budget.calls)
if conflicts != 1 || len(h.budget.calls) != 0 {
t.Fatalf("conflict retries=%d, budget calls=%+v; want one conflict and no reservations", conflicts, h.budget.calls)
}
})
}
Expand Down
4 changes: 2 additions & 2 deletions agentplane/runner_gateway/environments/lifecycle/types.go
Original file line number Diff line number Diff line change
Expand Up @@ -45,8 +45,8 @@ type WakeBudgetKey struct {
ServiceSPIFFEID string
}

// WakeBudget reserves a bounded wake attempt without depending on Dispatcher storage.
// Implementations share reservations across replicas and explicit and automatic wake paths.
// WakeBudget reserves a bounded automatic wake attempt without depending on Dispatcher storage.
// Implementations share reservations across replicas. Explicit resume does not reserve budget.
type WakeBudget interface {
ReserveWakeAttempt(context.Context, WakeBudgetKey, string, int64) (time.Duration, error)
}
Expand Down
10 changes: 5 additions & 5 deletions agentplane/runner_gateway/environments/service_svid_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -40,11 +40,11 @@ func (c serviceCreateWithUIDClient) Create(ctx context.Context, object client.Ob
}

type serviceRoleWakeBudget struct {
caller lifecycle.WakeBudgetKey
calls int
}

func (b *serviceRoleWakeBudget) ReserveWakeAttempt(_ context.Context, caller lifecycle.WakeBudgetKey, _ string, _ int64) (time.Duration, error) {
b.caller = caller
func (b *serviceRoleWakeBudget) ReserveWakeAttempt(_ context.Context, _ lifecycle.WakeBudgetKey, _ string, _ int64) (time.Duration, error) {
b.calls++
return 0, nil
}

Expand Down Expand Up @@ -166,8 +166,8 @@ func TestEnvironmentServiceServiceRolesStayMethodScoped(t *testing.T) {
if err != nil || response.GetOutcome() != environmentpb.ResumeOutcome_RESUME_OUTCOME_REQUESTED {
t.Fatalf("service operator resume = %v, %v", response, err)
}
if budget.caller != (lifecycle.WakeBudgetKey{ServiceSPIFFEID: operatorID}) {
t.Fatalf("operator wake budget caller = %+v, want SPIFFE identity only", budget.caller)
if budget.calls != 0 {
t.Fatalf("explicit service operator resume reserved wake budget %d times", budget.calls)
}
var updated sandboxapi.Sandbox
if err := objects.Get(t.Context(), types.NamespacedName{Namespace: discoveryNamespace, Name: sandboxName}, &updated); err != nil {
Expand Down
Loading