diff --git a/cmd/atenet/internal/router/dataplane.go b/cmd/atenet/internal/router/dataplane.go index 3ce608ce32..d72a153945 100644 --- a/cmd/atenet/internal/router/dataplane.go +++ b/cmd/atenet/internal/router/dataplane.go @@ -20,7 +20,6 @@ import ( "log/slog" "net" "strconv" - "time" "golang.org/x/sync/errgroup" @@ -98,7 +97,7 @@ func (s *RouterServer) newEnvoyXdsServer(ctx context.Context, parkCfg ingress.Pa if parkCfg.Enabled() { // Envoy must keep a parked request open at least as long as the router // will hold it; add a margin so the router surfaces its own 503 first. - xdsSrv.SetExtProcMessageTimeout(parkCfg.Budget + 5*time.Second) + xdsSrv.SetExtProcMessageTimeout(parkCfg.Budget + ingress.ExtProcTimeoutMargin) } xdsSrv.SetTlsConfig(s.cfg.HttpsPort, s.cfg.EnvoyCertPath) diff --git a/cmd/atenet/internal/router/ingress/errors.go b/cmd/atenet/internal/router/ingress/errors.go index 6218d44dfa..949ee0897b 100644 --- a/cmd/atenet/internal/router/ingress/errors.go +++ b/cmd/atenet/internal/router/ingress/errors.go @@ -73,6 +73,13 @@ func parkingFullErr(actorID string, retryErr error) error { "actor %q unavailable: router at capacity", actorID) } +// stillResumingErr tells the client the actor is still resuming and to retry. +// It wraps context.DeadlineExceeded so metrics count it as a timeout. +func stillResumingErr(actorRef resources.ActorRef) error { + return extproc.WrapReqError(envoy_type.StatusCode_ServiceUnavailable, context.DeadlineExceeded, + "actor %s is still resuming, retry", actorRef) +} + // mapResumeError translates an ActorResumer error into a client-facing // denial. It maps gRPC status codes to appropriate HTTP status codes and // short, human-readable bodies. The original error is preserved via Unwrap diff --git a/cmd/atenet/internal/router/ingress/resumer.go b/cmd/atenet/internal/router/ingress/resumer.go index 0ccbff2cea..7657a20bf2 100644 --- a/cmd/atenet/internal/router/ingress/resumer.go +++ b/cmd/atenet/internal/router/ingress/resumer.go @@ -37,6 +37,15 @@ import ( // conflicts are retried; capacity errors fail immediately. const failFastResumeBudget = 15 * time.Second +// ExtProcTimeoutMargin is how much longer than the park budget Envoy waits for +// the router's reply. +const ExtProcTimeoutMargin = 5 * time.Second + +// routerExtProcTimeout is how much longer than the park budget the router +// waits before answering. It must be less than ExtProcTimeoutMargin: if +// Envoy's timeout fires first, Envoy sends the client an empty 504. +const routerExtProcTimeout = ExtProcTimeoutMargin - time.Second + // resumeBackoff builds the backoff between resume attempts while a request is // parked, from the configured retry parameters. // @@ -272,12 +281,19 @@ func flightResult(bgCtx context.Context, resumeResp *ateapi.ResumeActorResponse, // flight is retrying (or already was), the caller must hold a parking-lot slot // to keep waiting and is shed with a 503 when the lot is full. func (r *ActorResumer) awaitFlight(ctx context.Context, f *resumeActorFlight, actorRef resources.ActorRef, reqID uint64) (*ateapi.Actor, ResumeOutcome, error) { + // If the flight is still running just before Envoy's ext_proc timeout, the + // router ends the wait itself and sends the client a 503. Without this, + // Envoy would end the request with a 504 and no useful error message. The flight + // keeps going, and later requests join it. + maxWait := r.budget + routerExtProcTimeout + waitCtx, cancel := context.WithTimeout(ctx, maxWait) + defer cancel() + select { - case <-ctx.Done(): - // The caller's request context was canceled before the shared resume - // completed. The flight continues and may still activate the actor, so - // the outcome is "unknown", not "none". - return nil, ResumeOutcomeUnknown, ctx.Err() + case <-waitCtx.Done(): + // The caller stopped waiting, but the flight may still resume the + // actor, so the outcome is "unknown". + return nil, ResumeOutcomeUnknown, waitErr(ctx, actorRef) case <-f.done: // Fast path: the flight finished without ever retrying, or before this // caller saw retrying. The lot is never touched. @@ -304,8 +320,8 @@ func (r *ActorResumer) awaitFlight(ctx context.Context, f *resumeActorFlight, ac defer func() { release(parkOutcomeFor(finalErr)) }() select { - case <-ctx.Done(): - finalErr = ctx.Err() + case <-waitCtx.Done(): + finalErr = waitErr(ctx, actorRef) return nil, ResumeOutcomeUnknown, finalErr case <-f.done: actor, outcome, err := f.callerResult(reqID) @@ -314,6 +330,15 @@ func (r *ActorResumer) awaitFlight(ctx context.Context, f *resumeActorFlight, ac } } +// waitErr returns ctx's error if the request itself ended, or stillResumingErr +// if the router stopped waiting first. +func waitErr(ctx context.Context, actorRef resources.ActorRef) error { + if err := ctx.Err(); err != nil { + return err + } + return stillResumingErr(actorRef) +} + // enterLot admits the caller to the parking lot, treating a nil lot as // unbounded (no admission control). shedOutcome labels the rejection if the // lot is full. diff --git a/cmd/atenet/internal/router/ingress/resumer_test.go b/cmd/atenet/internal/router/ingress/resumer_test.go index c37a0f9254..48f59d0d74 100644 --- a/cmd/atenet/internal/router/ingress/resumer_test.go +++ b/cmd/atenet/internal/router/ingress/resumer_test.go @@ -505,6 +505,73 @@ func TestActorResumer_Parking(t *testing.T) { }) }) + t.Run("CallerStopsWaitingBeforeEnvoyTimeout", func(t *testing.T) { + for _, tc := range []struct { + name string + // firstErr, if set, fails the first attempt so the caller parks + // before the slow one. + firstErr error + wantCalls int32 + }{ + {"Resolving", nil, 1}, + {"Parked", status.Error(codes.ResourceExhausted, "no worker has room for the actor"), 2}, + } { + t.Run(tc.name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + const budget = 5 * time.Second + var calls atomic.Int32 + proceed := make(chan struct{}) + mock := &resumerMockClient{ + resumeFn: func(context.Context, *ateapi.ResumeActorRequest, ...grpc.CallOption) (*ateapi.ResumeActorResponse, error) { + if calls.Add(1) == 1 && tc.firstErr != nil { + return nil, tc.firstErr + } + // A restore slower than Envoy's timeout. + <-proceed + return &ateapi.ResumeActorResponse{ + Actor: &ateapi.Actor{Metadata: &ateapi.ResourceMetadata{Name: testActorName}, Status: &ateapi.ActorStatus{State: ateapi.ActorState_ACTOR_STATE_RUNNING, WorkerAssignment: &ateapi.WorkerAssignment{WorkerPodIps: []string{expectedIP}}}}, + Resumed: true, + }, nil + }, + } + resumer := NewActorResumer(mock, withParking(ParkedRequestConfig{Max: 2, Budget: budget})) + + // Envoy closes the stream at its timeout. + envoyTimeout := budget + ExtProcTimeoutMargin + ctx, cancel := context.WithTimeout(t.Context(), envoyTimeout) + defer cancel() + start := time.Now() + _, _, err := resumer.ResumeActor(ctx, testActorRef) + if waited := time.Since(start); waited >= envoyTimeout { + t.Errorf("caller waited %v, want less than Envoy's %v timeout", waited, envoyTimeout) + } + var reqErr *extproc.ReqError + if !errors.As(err, &reqErr) || reqErr.StatusCode != int(envoy_type.StatusCode_ServiceUnavailable) { + t.Fatalf("expected the router's 503, got %v", err) + } + if got := parkOutcomeFor(err); got != parkOutcomeTimeout { + t.Errorf("outcome = %q, want %q", got, parkOutcomeTimeout) + } + + // A later caller joins the running attempt and is served. + errCh := make(chan error, 1) + go func() { + _, _, err := resumer.ResumeActor(t.Context(), testActorRef) + errCh <- err + }() + synctest.Wait() + close(proceed) + if err := <-errCh; err != nil { + t.Fatalf("later caller: unexpected error: %v", err) + } + if got := calls.Load(); got != tc.wantCalls { + t.Errorf("expected %d resume attempts, got %d", tc.wantCalls, got) + } + }) + }) + } + }) + t.Run("LateNonRetryableErrorIsPreserved", func(t *testing.T) { for _, tc := range []struct { name string diff --git a/docs/concepts/request-parking.md b/docs/concepts/request-parking.md index 5ff2528a94..1cb95b1866 100644 --- a/docs/concepts/request-parking.md +++ b/docs/concepts/request-parking.md @@ -54,23 +54,22 @@ keeps retrying with exponential backoff until either **The budget bounds retries, not a committed resume.** When the budget elapses the router stops starting new resume attempts, but an attempt already in flight is **never canceled**: by then the control plane has committed work to -it, and canceling would discard an in-progress restore. Instead the router -waits for that attempt's real result: a restore that overshoots the budget -(routine under node contention) is **served late** rather than failed, and a -late retryable error still surfaces as the capacity `503`. The attempt is -bounded by the control plane's own server-side RPC deadline, and Envoy's -ext_proc message timeout (`budget + 5s`) remains the ceiling on how long a -client is held either way. +it, and canceling would discard an in-progress restore. Instead a request keeps +waiting for that attempt until 1s before Envoy's ext_proc message timeout +(`budget + 5s`). A restore that overshoots the budget (routine under node +contention) is **served late** if it finishes by then, and a late retryable +error still surfaces as the capacity `503`. If the attempt is still running, +the router ends the wait and answers `503 "actor is still resuming, +retry"`. **Worst-case occupancy.** A parked request's lot slot is held until its caller -stops waiting; with the Envoy dataplane that is bounded by the ext_proc -message timeout (`budget + 5s`). The resume attempt itself can outlive every -caller: it carries no client-side deadline and runs until ateapi's -server-side maximum RPC deadline, holding that actor's flight entry. +stops waiting, which is at most `budget + 4s`. The resume attempt itself can +outlive every caller: it carries no client-side deadline and runs until +ateapi's server-side maximum RPC deadline, holding that actor's flight entry. New requests for the same actor during that window do not start another -control-plane call — they join the in-flight attempt, and if it has not -resolved by their own stream deadline they are ended by the dataplane's -timeout rather than a router verdict. +control-plane call. They join the in-flight attempt, wait up to the same +`budget + 4s` from their own arrival, and get the same "still resuming" `503` +if it has not finished. To bound resource use and provide backpressure, parked requests are admitted to a **parking lot** of fixed capacity (`--parked-request-max`, default `1024`). A @@ -140,8 +139,9 @@ Envoy sidecar) alive until in-flight streams finish, and the ext_proc drain deadline (`--drain-timeout`) defaults to a value derived from `--parked-request-budget` and is validated at startup to be `>=` the budget — so a parked request always gets its full budget and a normal verdict (routed -`200` or capacity `503`) even mid-termination. See the graceful-shutdown knobs -(`--drain-delay`, `--drain-timeout`) in `manifests/ate-install/atenet-router.yaml`. +`200` or the router's `503`) even mid-termination. See the graceful-shutdown +knobs (`--drain-delay`, `--drain-timeout`) in +`manifests/ate-install/atenet-router.yaml`. ## Configuration @@ -174,7 +174,7 @@ bounds the wait. | `served` | The resume succeeded and the request was routed to its worker. | | `budget_exhausted` | The park budget elapsed while the resume was still blocked on a retryable condition (pool saturated, a concurrent operation holding the actor, or the control plane unavailable) — the signal that capacity, not a fault, is the bottleneck. | | `canceled` | The client disconnected while parked (request context canceled). | - | `timeout` | The request's own deadline expired while parked (distinct from the park budget). | + | `timeout` | The request's own deadline expired while parked, or the router stopped waiting because the resume was still running (the "still resuming" `503`). | | `error` | The resume failed with a non-retryable error (`NotFound`, `PermissionDenied`, ...). | - `atenet.router.parking.rejected` — counter: requests shed because the lot was