From ee8eabe74631f1b13f8eaaf5e2d2b406449f07e0 Mon Sep 17 00:00:00 2001 From: Vicente Ferrara Date: Tue, 13 Oct 2026 22:34:09 +0000 Subject: [PATCH 1/3] ate-env-api: probe the guest and retry on Unavailable until a suspended environment wakes, bounded by --wake-budget --- cmd/ate-env-api/main.go | 2 + internal/apiservice/server.go | 22 ++++- internal/apiservice/wakeretry.go | 96 +++++++++++++++++++ internal/apiservice/wakeretry_test.go | 89 +++++++++++++++++ .../internaltest/fakecontrol/fakecontrol.go | 16 ++++ 5 files changed, 220 insertions(+), 5 deletions(-) create mode 100644 internal/apiservice/wakeretry.go create mode 100644 internal/apiservice/wakeretry_test.go diff --git a/cmd/ate-env-api/main.go b/cmd/ate-env-api/main.go index ef429421..d6e71006 100644 --- a/cmd/ate-env-api/main.go +++ b/cmd/ate-env-api/main.go @@ -41,6 +41,7 @@ func main() { hostSuffix = flag.String("host-suffix", ate.DefaultHostSuffix, "atenet router host suffix for actor routing") skipVerify = flag.Bool("skip-verify", true, "skip TLS certificate verification on the control plane connection") tokenFile = flag.String("ateapi-token-file", "", "file with a bearer token for the control plane, e.g. a projected ServiceAccount token with the ateapi audience") + wakeBudget = flag.Duration("wake-budget", apiservice.DefaultWakeBudget, "how long a guest call waits for a suspended environment to wake before failing; cold image pulls and cross-region snapshot reads exceed the router's own parking budget") ) flag.Parse() @@ -68,6 +69,7 @@ func main() { log.Printf("default atespace %q ready", ate.DefaultAtespace) apisvc := apiservice.New(client, *atenet, *hostSuffix) + apisvc.WakeBudget = *wakeBudget defer apisvc.Close() grpcServer := grpc.NewServer() diff --git a/internal/apiservice/server.go b/internal/apiservice/server.go index aad189ef..a818401f 100644 --- a/internal/apiservice/server.go +++ b/internal/apiservice/server.go @@ -22,6 +22,7 @@ import ( "errors" "io" "strings" + "time" "github.com/agent-substrate/env/internal/ate" ateenvv1alpha "github.com/agent-substrate/env/proto/ateenv/v1alpha" @@ -55,6 +56,10 @@ type Server struct { client *ate.Client routerAddr string hostSuffix string + + // WakeBudget bounds how long a guest call waits for a suspended + // environment to wake; see ensureAwake. Zero means DefaultWakeBudget. + WakeBudget time.Duration } // New creates a new Server. @@ -68,6 +73,7 @@ func New(client *ate.Client, routerAddr, hostSuffix string) *Server { client: client, routerAddr: routerAddr, hostSuffix: hostSuffix, + WakeBudget: DefaultWakeBudget, } } @@ -261,7 +267,7 @@ func (s *Server) processClient(ctx context.Context) (ateenvv1alpha.ProcessServic if err != nil { return nil, nil, nil, err } - conn, err := s.guestConn(atespace, envID) + conn, err := s.guestConn(ctx, atespace, envID) if err != nil { return nil, nil, nil, err } @@ -371,7 +377,7 @@ func (s *Server) ReadFile(req *ateenvv1alpha.ReadFileRequest, stream grpc.Server if err != nil { return err } - conn, err := s.guestConn(atespace, envID) + conn, err := s.guestConn(ctx, atespace, envID) if err != nil { return err } @@ -403,7 +409,7 @@ func (s *Server) WriteFile(stream grpc.ClientStreamingServer[ateenvv1alpha.Write if err != nil { return err } - conn, err := s.guestConn(atespace, envID) + conn, err := s.guestConn(ctx, atespace, envID) if err != nil { return err } @@ -455,7 +461,9 @@ func envFromContext(ctx context.Context) (string, string, error) { return ids[0], atespace, nil } -// guestConn dials the environment's guest through the router. Calls on the +// guestConn dials the environment's guest through the router and waits, +// within WakeBudget, for the guest to answer, so a suspended environment is +// awake before the caller's request goes through (see ensureAwake). Calls on the // returned connection take the caller's context as is, for its deadline and // cancellation only: the interceptors below are the only writer of // ate-target-actor, and no caller metadata is forwarded. Forwarding it once @@ -463,7 +471,7 @@ func envFromContext(ctx context.Context) (string, string, error) { // (agent-substrate/substrate#2308 answers duplicate ate-target-actor headers // with "invalid actor reference"), and it would let a client pass routing // headers through the proxy. Nothing in the guest reads caller metadata. -func (s *Server) guestConn(atespace, id string) (*grpc.ClientConn, error) { +func (s *Server) guestConn(ctx context.Context, atespace, id string) (*grpc.ClientConn, error) { if s.routerAddr == "" { return nil, status.Error(codes.FailedPrecondition, "router address not configured for guest operations") } @@ -486,6 +494,10 @@ func (s *Server) guestConn(atespace, id string) (*grpc.ClientConn, error) { if err != nil { return nil, status.Errorf(codes.Internal, "dialing guest router for %s/%s: %v", atespace, id, err) } + if err := s.ensureAwake(ctx, conn, atespace, id); err != nil { + _ = conn.Close() + return nil, err + } return conn, nil } diff --git a/internal/apiservice/wakeretry.go b/internal/apiservice/wakeretry.go new file mode 100644 index 00000000..8ba1d782 --- /dev/null +++ b/internal/apiservice/wakeretry.go @@ -0,0 +1,96 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package apiservice + +import ( + "context" + "log" + "time" + + ateenvv1alpha "github.com/agent-substrate/env/proto/ateenv/v1alpha" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// DefaultWakeBudget bounds how long a guest call waits for its environment +// to wake before the error reaches the caller. The first request that +// reaches the router is what resumes a suspended actor, and the router holds +// that request only for its own parking budget, a few seconds. A wake that +// has to pull the environment's image onto the node or read its snapshot +// across a region takes longer than that: 18 s for a cold image pull and 7 s +// for a 1 GiB snapshot read across regions, measured on GKE. +const DefaultWakeBudget = 2 * time.Minute + +const ( + wakeRetryInitialBackoff = 200 * time.Millisecond + wakeRetryMaxBackoff = 2 * time.Second +) + +// ensureAwake sends a trivial request to the environment's guest over conn +// and repeats it while the router answers Unavailable, which is what an +// expired parking budget and a guest that is not yet listening both look +// like. It returns when the guest answers anything else, when the actor has +// crashed, when ctx ends, or when the wake budget is spent. Because the probe +// is the first request through the router, it is also what wakes the actor; +// the calls that follow on conn find it up. +func (s *Server) ensureAwake(ctx context.Context, conn *grpc.ClientConn, atespace, id string) error { + budget := s.WakeBudget + if budget <= 0 { + budget = DefaultWakeBudget + } + deadline := time.Now().Add(budget) + probe := ateenvv1alpha.NewProcessServiceClient(conn) + backoff := wakeRetryInitialBackoff + for attempt := 1; ; attempt++ { + _, err := probe.GetProcess(ctx, &ateenvv1alpha.GetProcessRequest{ProcessId: "ate-env-wake-probe"}) + if status.Code(err) != codes.Unavailable { + // Anything else, NotFound for the probe id included, came from + // the guest itself. + return nil + } + crashed, cerr := s.actorCrashed(ctx, atespace, id) + if cerr != nil { + return cerr + } + if crashed { + return status.Errorf(codes.FailedPrecondition, "environment %s/%s crashed while waking: %v", atespace, id, err) + } + remaining := time.Until(deadline) + if remaining <= 0 { + return status.Errorf(codes.Unavailable, "environment %s/%s did not wake within %s (%d attempts): %v", atespace, id, budget, attempt, err) + } + if attempt == 1 { + log.Printf("waiting for environment %s/%s to wake (budget %s)", atespace, id, budget) + } + select { + case <-ctx.Done(): + return status.FromContextError(ctx.Err()).Err() + case <-time.After(min(backoff, remaining)): + } + backoff = min(backoff*2, wakeRetryMaxBackoff) + } +} + +// actorCrashed reports whether the control plane has the actor in a crashed +// state, in which case waiting for a wake is pointless. +func (s *Server) actorCrashed(ctx context.Context, atespace, id string) (bool, error) { + actor, err := s.client.Get(ctx, atespace, id) + if err != nil { + return false, toGRPCError(err) + } + return actor.GetStatus().GetState() == ateapipb.ActorState_ACTOR_STATE_CRASHED, nil +} diff --git a/internal/apiservice/wakeretry_test.go b/internal/apiservice/wakeretry_test.go new file mode 100644 index 00000000..ebc72ed9 --- /dev/null +++ b/internal/apiservice/wakeretry_test.go @@ -0,0 +1,89 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package apiservice_test + +import ( + "context" + "testing" + "time" + + "github.com/agent-substrate/env/guest" + "github.com/agent-substrate/env/internal/apiservice" + ateenvv1alpha "github.com/agent-substrate/env/proto/ateenv/v1alpha" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" +) + +// registerGuest creates an environment and serves a real guest for it on the +// fake router, returning the metadata context that addresses it. +func registerGuest(t *testing.T, te *testEnv, id string) context.Context { + t.Helper() + ctx := context.Background() + if _, err := te.envClient.CreateEnvironment(ctx, &ateenvv1alpha.CreateEnvironmentRequest{Id: id}); err != nil { + t.Fatalf("CreateEnvironment: %v", err) + } + srv, cleanup, err := guest.NewServer(guest.Config{Workspace: t.TempDir(), EnableProcess: true, EnableFileSystem: true}) + if err != nil { + t.Fatalf("guest.NewServer: %v", err) + } + t.Cleanup(cleanup) + te.router.Register(id, srv) + return metadata.AppendToOutgoingContext(ctx, "x-env-id", id, "x-env-atespace", apiservice.DefaultAtespace) +} + +// A guest call made while the environment is still suspended is held until +// the environment is running, instead of failing with the router's 503. +func TestGuestCallWaitsForWake(t *testing.T) { + te := newFullTestEnv(t) + envCtx := registerGuest(t, te, "wake-test") + te.control.SetState("wake-test", ateapipb.ActorState_ACTOR_STATE_SUSPENDED) + + // The fake router has no resume path of its own; stand in for it. + go func() { + time.Sleep(600 * time.Millisecond) + te.control.SetState("wake-test", ateapipb.ActorState_ACTOR_STATE_RUNNING) + }() + + start := time.Now() + proc, err := te.procClient.StartProcess(envCtx, &ateenvv1alpha.StartProcessRequest{Command: []string{"sh", "-c", "echo awake"}}) + if err != nil { + t.Fatalf("StartProcess while suspended: %v", err) + } + if proc.GetProcessId() == "" { + t.Fatal("StartProcess returned no process id") + } + if elapsed := time.Since(start); elapsed < 500*time.Millisecond { + t.Errorf("call returned after %s, before the environment was running", elapsed) + } +} + +// A crashed environment fails the call at once rather than consuming the +// wake budget. +func TestGuestCallFailsFastOnCrash(t *testing.T) { + te := newFullTestEnv(t) + envCtx := registerGuest(t, te, "crash-test") + te.control.SetState("crash-test", ateapipb.ActorState_ACTOR_STATE_CRASHED) + + start := time.Now() + _, err := te.procClient.StartProcess(envCtx, &ateenvv1alpha.StartProcessRequest{Command: []string{"sh", "-c", "true"}}) + if status.Code(err) != codes.FailedPrecondition { + t.Fatalf("StartProcess on a crashed environment: code %v (%v), want FailedPrecondition", status.Code(err), err) + } + if elapsed := time.Since(start); elapsed > 5*time.Second { + t.Errorf("crash took %s to report; the wake budget must not apply", elapsed) + } +} diff --git a/internal/internaltest/fakecontrol/fakecontrol.go b/internal/internaltest/fakecontrol/fakecontrol.go index ff5300fc..efda16ec 100644 --- a/internal/internaltest/fakecontrol/fakecontrol.go +++ b/internal/internaltest/fakecontrol/fakecontrol.go @@ -179,6 +179,22 @@ func (s *Server) Status(name string) ateapipb.ActorState { return s.find(name).GetStatus().GetState() } +// SetState overrides an actor's state, for tests that need a state the fake's +// own transitions never produce, such as CRASHED, or need to hold an actor +// SUSPENDED after creation. +func (s *Server) SetState(name string, state ateapipb.ActorState) { + s.mu.Lock() + defer s.mu.Unlock() + a := s.find(name) + if a == nil { + return + } + if a.Status == nil { + a.Status = &ateapipb.ActorStatus{} + } + a.Status.State = state +} + // find returns the actor with the given name, whichever atespace holds it, or // nil when no actor has that name. The helpers above address actors by name // alone so tests do not have to know which atespace a client placed them in. From 3df2928beb0761be696a449ec360cb677b0273dc Mon Sep 17 00:00:00 2001 From: Vicente Ferrara Date: Wed, 14 Oct 2026 18:00:41 +0000 Subject: [PATCH 2/3] ate-env-api: cap the wake wait at --wake-running-budget once the actor is RUNNING, name the state in the error, test both budgets --- README.md | 5 ++ cmd/ate-env-api/main.go | 16 +++--- internal/apiservice/server.go | 12 +++-- internal/apiservice/server_test.go | 2 + internal/apiservice/wakeretry.go | 74 ++++++++++++++++++--------- internal/apiservice/wakeretry_test.go | 46 +++++++++++++++++ 6 files changed, 121 insertions(+), 34 deletions(-) diff --git a/README.md b/README.md index b0badfab..6956c329 100644 --- a/README.md +++ b/README.md @@ -88,6 +88,11 @@ ate-env manifest \ kubectl get pods -n ate-env ``` +`ate-env-api` waits for a suspended environment to wake before forwarding a guest call, up to +`--wake-budget` (default 2m; cold image pulls and cross-region snapshot reads exceed the router's own +parking budget) and, once the control plane reports the environment RUNNING, up to +`--wake-running-budget` (default 20s). Both are flags on the API deployment. + ### 2. Register the ActorTemplate in Substrate Substrate manages ActorTemplates directly in its control plane rather than Kubernetes CRDs. Use `ate-env manifest template` to generate the Substrate ActorTemplate manifest: diff --git a/cmd/ate-env-api/main.go b/cmd/ate-env-api/main.go index d6e71006..ad1a5f0b 100644 --- a/cmd/ate-env-api/main.go +++ b/cmd/ate-env-api/main.go @@ -35,13 +35,14 @@ import ( func main() { var ( - listen = flag.String("listen", "0.0.0.0:7777", "address to serve the API on") - ateapi = flag.String("ateapi", ate.DefaultControlAddr, "address of the ateapi gRPC control plane") - atenet = flag.String("atenet", ate.DefaultRouterAddr, "address of the atenet HTTP router") - hostSuffix = flag.String("host-suffix", ate.DefaultHostSuffix, "atenet router host suffix for actor routing") - skipVerify = flag.Bool("skip-verify", true, "skip TLS certificate verification on the control plane connection") - tokenFile = flag.String("ateapi-token-file", "", "file with a bearer token for the control plane, e.g. a projected ServiceAccount token with the ateapi audience") - wakeBudget = flag.Duration("wake-budget", apiservice.DefaultWakeBudget, "how long a guest call waits for a suspended environment to wake before failing; cold image pulls and cross-region snapshot reads exceed the router's own parking budget") + listen = flag.String("listen", "0.0.0.0:7777", "address to serve the API on") + ateapi = flag.String("ateapi", ate.DefaultControlAddr, "address of the ateapi gRPC control plane") + atenet = flag.String("atenet", ate.DefaultRouterAddr, "address of the atenet HTTP router") + hostSuffix = flag.String("host-suffix", ate.DefaultHostSuffix, "atenet router host suffix for actor routing") + skipVerify = flag.Bool("skip-verify", true, "skip TLS certificate verification on the control plane connection") + tokenFile = flag.String("ateapi-token-file", "", "file with a bearer token for the control plane, e.g. a projected ServiceAccount token with the ateapi audience") + wakeBudget = flag.Duration("wake-budget", apiservice.DefaultWakeBudget, "how long a guest call waits for a suspended environment to wake before failing; cold image pulls and cross-region snapshot reads exceed the router's own parking budget") + wakeRunningBudget = flag.Duration("wake-running-budget", apiservice.DefaultWakeRunningBudget, "how long a guest call waits when the environment is already RUNNING but the router still answers Unavailable (guest not yet listening vs. path down)") ) flag.Parse() @@ -70,6 +71,7 @@ func main() { apisvc := apiservice.New(client, *atenet, *hostSuffix) apisvc.WakeBudget = *wakeBudget + apisvc.WakeRunningBudget = *wakeRunningBudget defer apisvc.Close() grpcServer := grpc.NewServer() diff --git a/internal/apiservice/server.go b/internal/apiservice/server.go index a818401f..a8791870 100644 --- a/internal/apiservice/server.go +++ b/internal/apiservice/server.go @@ -60,6 +60,9 @@ type Server struct { // WakeBudget bounds how long a guest call waits for a suspended // environment to wake; see ensureAwake. Zero means DefaultWakeBudget. WakeBudget time.Duration + // WakeRunningBudget bounds the wait once the control plane reports the + // environment RUNNING; see ensureAwake. Zero means DefaultWakeRunningBudget. + WakeRunningBudget time.Duration } // New creates a new Server. @@ -70,10 +73,11 @@ func New(client *ate.Client, routerAddr, hostSuffix string) *Server { routerAddr = strings.TrimPrefix(routerAddr, "http://") routerAddr = strings.TrimPrefix(routerAddr, "https://") return &Server{ - client: client, - routerAddr: routerAddr, - hostSuffix: hostSuffix, - WakeBudget: DefaultWakeBudget, + client: client, + routerAddr: routerAddr, + hostSuffix: hostSuffix, + WakeBudget: DefaultWakeBudget, + WakeRunningBudget: DefaultWakeRunningBudget, } } diff --git a/internal/apiservice/server_test.go b/internal/apiservice/server_test.go index 5ee1d5fd..d25bdb3c 100644 --- a/internal/apiservice/server_test.go +++ b/internal/apiservice/server_test.go @@ -42,6 +42,7 @@ type testEnv struct { conn *grpc.ClientConn router *fakerouter.Router control *fakecontrol.Server + srv *apiservice.Server } func newTestEnv(t *testing.T) (ateenvv1alpha.EnvironmentServiceClient, *fakecontrol.Server) { @@ -105,6 +106,7 @@ func newFullTestEnv(t *testing.T) *testEnv { conn: conn, router: router, control: control, + srv: srv, } } diff --git a/internal/apiservice/wakeretry.go b/internal/apiservice/wakeretry.go index 8ba1d782..e9d08d2b 100644 --- a/internal/apiservice/wakeretry.go +++ b/internal/apiservice/wakeretry.go @@ -17,6 +17,7 @@ package apiservice import ( "context" "log" + "strings" "time" ateenvv1alpha "github.com/agent-substrate/env/proto/ateenv/v1alpha" @@ -26,15 +27,24 @@ import ( "google.golang.org/grpc/status" ) -// DefaultWakeBudget bounds how long a guest call waits for its environment -// to wake before the error reaches the caller. The first request that -// reaches the router is what resumes a suspended actor, and the router holds -// that request only for its own parking budget, a few seconds. A wake that -// has to pull the environment's image onto the node or read its snapshot -// across a region takes longer than that: 18 s for a cold image pull and 7 s -// for a 1 GiB snapshot read across regions, measured on GKE. +// DefaultWakeBudget bounds how long a guest call waits for an environment +// that the control plane reports as suspended or in transition (resuming, +// pausing, suspending) to wake before the error reaches the caller. The first +// request that reaches the router is what resumes a suspended actor, and the +// router holds that request only for its own parking budget, a few seconds. A +// wake that has to pull the environment's image onto the node or read its +// snapshot across a region takes longer than that: 18 s for a cold image pull +// and 7 s for a 1 GiB snapshot read across regions, measured on GKE. const DefaultWakeBudget = 2 * time.Minute +// DefaultWakeRunningBudget bounds the wait when the control plane already +// reports the environment RUNNING but the router still answers Unavailable. +// That has two causes: the guest is not yet listening after a restore, which +// resolves in seconds, or the path to the worker is down, which does not. +// A short budget covers the first without turning the second into a +// two-minute hang per call. +const DefaultWakeRunningBudget = 20 * time.Second + const ( wakeRetryInitialBackoff = 200 * time.Millisecond wakeRetryMaxBackoff = 2 * time.Second @@ -44,15 +54,23 @@ const ( // and repeats it while the router answers Unavailable, which is what an // expired parking budget and a guest that is not yet listening both look // like. It returns when the guest answers anything else, when the actor has -// crashed, when ctx ends, or when the wake budget is spent. Because the probe -// is the first request through the router, it is also what wakes the actor; -// the calls that follow on conn find it up. +// crashed, when ctx ends, or when the budget for the actor's current state +// is spent: WakeBudget while the actor is suspended or in transition, +// WakeRunningBudget once the control plane reports it RUNNING. Because the +// probe is the first request through the router, it is also what wakes the +// actor; the calls that follow on conn find it up. The state is read once per +// attempt, and the backoff caps attempts at roughly one per two seconds, so a +// full two-minute wait costs about sixty control-plane reads. func (s *Server) ensureAwake(ctx context.Context, conn *grpc.ClientConn, atespace, id string) error { budget := s.WakeBudget if budget <= 0 { budget = DefaultWakeBudget } - deadline := time.Now().Add(budget) + runningBudget := s.WakeRunningBudget + if runningBudget <= 0 { + runningBudget = DefaultWakeRunningBudget + } + start := time.Now() probe := ateenvv1alpha.NewProcessServiceClient(conn) backoff := wakeRetryInitialBackoff for attempt := 1; ; attempt++ { @@ -62,19 +80,24 @@ func (s *Server) ensureAwake(ctx context.Context, conn *grpc.ClientConn, atespac // the guest itself. return nil } - crashed, cerr := s.actorCrashed(ctx, atespace, id) - if cerr != nil { - return cerr + state, serr := s.actorState(ctx, atespace, id) + if serr != nil { + return serr } - if crashed { + if state == ateapipb.ActorState_ACTOR_STATE_CRASHED { return status.Errorf(codes.FailedPrecondition, "environment %s/%s crashed while waking: %v", atespace, id, err) } - remaining := time.Until(deadline) + allowed := budget + if state == ateapipb.ActorState_ACTOR_STATE_RUNNING { + allowed = runningBudget + } + remaining := allowed - time.Since(start) if remaining <= 0 { - return status.Errorf(codes.Unavailable, "environment %s/%s did not wake within %s (%d attempts): %v", atespace, id, budget, attempt, err) + return status.Errorf(codes.Unavailable, "environment %s/%s did not answer within %s while %s (%d attempts): %v", + atespace, id, allowed, shortState(state), attempt, err) } if attempt == 1 { - log.Printf("waiting for environment %s/%s to wake (budget %s)", atespace, id, budget) + log.Printf("waiting for environment %s/%s to wake (state %s, budget %s)", atespace, id, shortState(state), allowed) } select { case <-ctx.Done(): @@ -85,12 +108,17 @@ func (s *Server) ensureAwake(ctx context.Context, conn *grpc.ClientConn, atespac } } -// actorCrashed reports whether the control plane has the actor in a crashed -// state, in which case waiting for a wake is pointless. -func (s *Server) actorCrashed(ctx context.Context, atespace, id string) (bool, error) { +// actorState reads the actor's state from the control plane. A deleted +// environment surfaces here as NotFound (through toGRPCError), which is the +// right answer to a caller still holding its handle. +func (s *Server) actorState(ctx context.Context, atespace, id string) (ateapipb.ActorState, error) { actor, err := s.client.Get(ctx, atespace, id) if err != nil { - return false, toGRPCError(err) + return ateapipb.ActorState_ACTOR_STATE_UNSPECIFIED, toGRPCError(err) } - return actor.GetStatus().GetState() == ateapipb.ActorState_ACTOR_STATE_CRASHED, nil + return actor.GetStatus().GetState(), nil +} + +func shortState(st ateapipb.ActorState) string { + return strings.TrimPrefix(st.String(), "ACTOR_STATE_") } diff --git a/internal/apiservice/wakeretry_test.go b/internal/apiservice/wakeretry_test.go index ebc72ed9..28805bb2 100644 --- a/internal/apiservice/wakeretry_test.go +++ b/internal/apiservice/wakeretry_test.go @@ -16,6 +16,7 @@ package apiservice_test import ( "context" + "strings" "testing" "time" @@ -87,3 +88,48 @@ func TestGuestCallFailsFastOnCrash(t *testing.T) { t.Errorf("crash took %s to report; the wake budget must not apply", elapsed) } } + +// An environment that stays suspended past the wake budget fails with +// Unavailable, naming the state and the attempt count. +func TestGuestCallFailsWhenWakeBudgetExpires(t *testing.T) { + te := newFullTestEnv(t) + envCtx := registerGuest(t, te, "budget-test") + te.control.SetState("budget-test", ateapipb.ActorState_ACTOR_STATE_SUSPENDED) + te.srv.WakeBudget = 1500 * time.Millisecond + + start := time.Now() + _, err := te.procClient.StartProcess(envCtx, &ateenvv1alpha.StartProcessRequest{Command: []string{"sh", "-c", "true"}}) + if status.Code(err) != codes.Unavailable { + t.Fatalf("code %v (%v), want Unavailable", status.Code(err), err) + } + if msg := status.Convert(err).Message(); !strings.Contains(msg, "did not answer within") || !strings.Contains(msg, "SUSPENDED") { + t.Errorf("message %q should name the budget and the state", msg) + } + if elapsed := time.Since(start); elapsed < 1400*time.Millisecond || elapsed > 6*time.Second { + t.Errorf("gave up after %s, want about the 1.5 s budget", elapsed) + } +} + +// An environment the control plane reports RUNNING, whose guest the router +// still cannot reach, fails after the shorter running budget rather than +// holding the call for the full wake budget. +func TestGuestCallCapsWaitWhenRunning(t *testing.T) { + te := newFullTestEnv(t) + envCtx := registerGuest(t, te, "running-test") + te.control.SetState("running-test", ateapipb.ActorState_ACTOR_STATE_RUNNING) + te.router.Running = func(string) bool { return false } // path down: 503 regardless of state + te.srv.WakeBudget = time.Minute + te.srv.WakeRunningBudget = time.Second + + start := time.Now() + _, err := te.procClient.StartProcess(envCtx, &ateenvv1alpha.StartProcessRequest{Command: []string{"sh", "-c", "true"}}) + if status.Code(err) != codes.Unavailable { + t.Fatalf("code %v (%v), want Unavailable", status.Code(err), err) + } + if msg := status.Convert(err).Message(); !strings.Contains(msg, "RUNNING") { + t.Errorf("message %q should name the RUNNING state", msg) + } + if elapsed := time.Since(start); elapsed > 5*time.Second { + t.Errorf("waited %s; the running budget (1 s) must apply, not the wake budget (1 min)", elapsed) + } +} From 4685b2bf1dd0bd598d7e92365d96d701a99a2962 Mon Sep 17 00:00:00 2001 From: Vicente Ferrara Date: Thu, 15 Oct 2026 21:52:35 +0000 Subject: [PATCH 3/3] ate-env-api: start the running budget at the first attempt that sees the actor RUNNING, and name the budget that tripped --- internal/apiservice/wakeretry.go | 37 +++++++++++++++++---------- internal/apiservice/wakeretry_test.go | 28 ++++++++++++++++++++ 2 files changed, 52 insertions(+), 13 deletions(-) diff --git a/internal/apiservice/wakeretry.go b/internal/apiservice/wakeretry.go index e9d08d2b..fda16dc9 100644 --- a/internal/apiservice/wakeretry.go +++ b/internal/apiservice/wakeretry.go @@ -54,13 +54,15 @@ const ( // and repeats it while the router answers Unavailable, which is what an // expired parking budget and a guest that is not yet listening both look // like. It returns when the guest answers anything else, when the actor has -// crashed, when ctx ends, or when the budget for the actor's current state -// is spent: WakeBudget while the actor is suspended or in transition, -// WakeRunningBudget once the control plane reports it RUNNING. Because the -// probe is the first request through the router, it is also what wakes the -// actor; the calls that follow on conn find it up. The state is read once per -// attempt, and the backoff caps attempts at roughly one per two seconds, so a -// full two-minute wait costs about sixty control-plane reads. +// crashed, when ctx ends, or when a budget is spent: WakeBudget bounds the +// whole wait, and WakeRunningBudget bounds the part of it that starts with +// the first attempt that sees the actor RUNNING, so a long suspended phase +// does not eat the time the guest needs to start listening after the state +// flips. Because the probe is the first request through the router, it is +// also what wakes the actor; the calls that follow on conn find it up. The +// state is read once per attempt, and the backoff caps attempts at roughly +// one per two seconds, so a full two-minute wait costs about sixty +// control-plane reads. func (s *Server) ensureAwake(ctx context.Context, conn *grpc.ClientConn, atespace, id string) error { budget := s.WakeBudget if budget <= 0 { @@ -71,6 +73,7 @@ func (s *Server) ensureAwake(ctx context.Context, conn *grpc.ClientConn, atespac runningBudget = DefaultWakeRunningBudget } start := time.Now() + var runningSince time.Time // first attempt that saw the actor RUNNING probe := ateenvv1alpha.NewProcessServiceClient(conn) backoff := wakeRetryInitialBackoff for attempt := 1; ; attempt++ { @@ -87,17 +90,25 @@ func (s *Server) ensureAwake(ctx context.Context, conn *grpc.ClientConn, atespac if state == ateapipb.ActorState_ACTOR_STATE_CRASHED { return status.Errorf(codes.FailedPrecondition, "environment %s/%s crashed while waking: %v", atespace, id, err) } - allowed := budget + allowed, since, which := budget, start, "wake budget" if state == ateapipb.ActorState_ACTOR_STATE_RUNNING { - allowed = runningBudget + if runningSince.IsZero() { + runningSince = time.Now() + } + allowed, since, which = runningBudget, runningSince, "running budget" + } else { + runningSince = time.Time{} + } + remaining := allowed - time.Since(since) + if overall := budget - time.Since(start); overall < remaining { + remaining, allowed, which = overall, budget, "wake budget" } - remaining := allowed - time.Since(start) if remaining <= 0 { - return status.Errorf(codes.Unavailable, "environment %s/%s did not answer within %s while %s (%d attempts): %v", - atespace, id, allowed, shortState(state), attempt, err) + return status.Errorf(codes.Unavailable, "environment %s/%s did not answer within the %s of %s while %s (%d attempts): %v", + atespace, id, which, allowed, shortState(state), attempt, err) } if attempt == 1 { - log.Printf("waiting for environment %s/%s to wake (state %s, budget %s)", atespace, id, shortState(state), allowed) + log.Printf("waiting for environment %s/%s to wake (state %s, %s %s)", atespace, id, shortState(state), which, allowed) } select { case <-ctx.Done(): diff --git a/internal/apiservice/wakeretry_test.go b/internal/apiservice/wakeretry_test.go index 28805bb2..78ac34b4 100644 --- a/internal/apiservice/wakeretry_test.go +++ b/internal/apiservice/wakeretry_test.go @@ -133,3 +133,31 @@ func TestGuestCallCapsWaitWhenRunning(t *testing.T) { t.Errorf("waited %s; the running budget (1 s) must apply, not the wake budget (1 min)", elapsed) } } + +// A wake whose suspended phase outlasts the running budget must still get +// the running budget once the actor is RUNNING: the guest listens a moment +// after the state flips. +func TestGuestCallRunningBudgetStartsAtRunning(t *testing.T) { + te := newFullTestEnv(t) + envCtx := registerGuest(t, te, "late-running-test") + te.control.SetState("late-running-test", ateapipb.ActorState_ACTOR_STATE_SUSPENDED) + te.srv.WakeBudget = time.Minute + te.srv.WakeRunningBudget = 3 * time.Second + + start := time.Now() + flip := start.Add(2500 * time.Millisecond) // the control plane reports RUNNING; the next attempt is at ~3.0 s + listen := start.Add(4500 * time.Millisecond) // the guest answers 2 s later, inside the 3 s running budget + te.router.Running = func(string) bool { return time.Now().After(listen) } + go func() { + time.Sleep(time.Until(flip)) + te.control.SetState("late-running-test", ateapipb.ActorState_ACTOR_STATE_RUNNING) + }() + + _, err := te.procClient.StartProcess(envCtx, &ateenvv1alpha.StartProcessRequest{Command: []string{"sh", "-c", "true"}}) + if err != nil { + t.Fatalf("StartProcess after a 2.5 s suspended phase and a 2 s listen lag: %v", err) + } + if elapsed := time.Since(start); elapsed < 4400*time.Millisecond { + t.Errorf("call returned after %s, before the guest was listening", elapsed) + } +}