diff --git a/cmd/atenet/internal/router/ingress/errors.go b/cmd/atenet/internal/router/ingress/errors.go index 1c3bc8956..7c1e18201 100644 --- a/cmd/atenet/internal/router/ingress/errors.go +++ b/cmd/atenet/internal/router/ingress/errors.go @@ -51,10 +51,16 @@ func statusDescription(err error) string { return status.Convert(err).Message() } +// errParkingLotFull is returned by ActorResumer.ResumeActor when a request +// reached its park transition but the lot had no free slot. mapResumeError +// turns it into the client-facing 503 "router at capacity" denial. +var errParkingLotFull = errors.New("parking lot full") + // parkingFullErr returns a 503 denial signaling that the router's parking lot -// is at capacity, so the request was shed without waiting. Clients should retry. -func parkingFullErr(actorID string) error { - return extproc.NewReqError(envoy_type.StatusCode_ServiceUnavailable, +// is at capacity, so the request was shed rather than parked. Clients should +// retry. The cause is preserved for log inspection via Unwrap. +func parkingFullErr(actorID string, cause error) error { + return extproc.WrapReqError(envoy_type.StatusCode_ServiceUnavailable, cause, "actor %q unavailable: router at capacity", actorID) } @@ -70,6 +76,12 @@ func mapResumeError(actorRef resources.ActorRef, err error) error { return nil } + // A caller shed at its park transition because the lot was full: the + // resume itself did not fail, the router declined to hold the request. + if errors.Is(err, errParkingLotFull) { + return parkingFullErr(actorRef.String(), err) + } + re := &extproc.ReqError{Cause: err} // Bare context sentinels reach here when the request's own context ends diff --git a/cmd/atenet/internal/router/ingress/flight.go b/cmd/atenet/internal/router/ingress/flight.go new file mode 100644 index 000000000..d7fa68f14 --- /dev/null +++ b/cmd/atenet/internal/router/ingress/flight.go @@ -0,0 +1,103 @@ +// 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 ingress + +import ( + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// resumeFlight is one in-flight, per-actor resume that concurrent requests +// share. It replaces a singleflight.Group entry so callers can observe the +// flight mid-run: parked exposes the park transition, which singleflight's +// result-only channel cannot. Its lifecycle is park (at most once, flight +// goroutine only) followed by ActorResumer.publish (exactly once). +type resumeFlight struct { + // parked is closed at most once, by park, at the flight's first retryable + // error — the moment the flight stops resolving and starts waiting. A + // caller woken by it (or attaching after it) must hold a parking-lot slot + // to keep waiting. Never closed on the fast path. + parked chan struct{} + // parkedSignaled guards the parked close. Only the flight goroutine calls + // park, so a plain bool suffices. + parkedSignaled bool + // done is closed exactly once, by publish, after result is written and the + // flight is deleted from the registry. The write-result-then-close order + // is what lets every caller read result without further synchronization; + // the delete-then-close order is what keeps a completed flight unjoinable + // (the next request for the actor starts a fresh flight). + done chan struct{} + // result is the shared outcome, written exactly once before done closes. + result *resumeCallResult +} + +func newResumeFlight() *resumeFlight { + return &resumeFlight{parked: make(chan struct{}), done: make(chan struct{})} +} + +// park signals the flight's park transition: it stopped resolving and started +// waiting, so from here on callers must hold parking-lot slots to keep +// waiting. Idempotent; only the flight goroutine may call it. +func (f *resumeFlight) park() { + if f.parkedSignaled { + return + } + f.parkedSignaled = true + close(f.parked) +} + +// callerResult classifies f's completed outcome for one caller. It must only +// be called after f.done is closed. +func (f *resumeFlight) callerResult(reqID uint64) (*ateapipb.Actor, ResumeOutcome, error) { + res := f.result + if res == nil { + return nil, ResumeOutcomeNone, status.Error(codes.Internal, "resume call returned nil result") + } + + // On error, return ResumeOutcomeNone ("none") so the failure is tagged + // under the 'outcome' label rather than misreported as an activation. + if res.err != nil { + return nil, ResumeOutcomeNone, res.err + } + + // Disambiguate the shared-flight resume outcome: + // - ResumeOutcomeNone ("none"): resumed == false, actor was already active/running. + // - ResumeOutcomeTriggered ("triggered"): Cold activation leader (resumed == true, caller's reqID == leaderID). + // - ResumeOutcomeJoined ("joined"): Cold activation joiner (resumed == true, caller's reqID != leaderID). + outcome := ResumeOutcomeNone + if res.resumed { + if res.leaderID == reqID { + outcome = ResumeOutcomeTriggered + } else { + outcome = ResumeOutcomeJoined + } + } + + return res.actor, outcome, nil +} + +// publish completes f with result and makes it unjoinable. The order is +// load-bearing: result before done (the channel close is what makes the write +// visible to callers), and registry delete before done (so no caller can +// attach to a completed flight — the next request for the actor starts a +// fresh one, preserving forget-on-completion semantics). +func (r *ActorResumer) publish(f *resumeFlight, key string, result *resumeCallResult) { + f.result = result + r.mu.Lock() + delete(r.flights, key) + r.mu.Unlock() + close(f.done) +} diff --git a/cmd/atenet/internal/router/ingress/ingress.go b/cmd/atenet/internal/router/ingress/ingress.go index 09a7ebb95..76dc88351 100644 --- a/cmd/atenet/internal/router/ingress/ingress.go +++ b/cmd/atenet/internal/router/ingress/ingress.go @@ -72,9 +72,12 @@ type Handler struct { } func New(apiClient ateapipb.ControlClient, parkCfg ParkedRequestConfig, parkMetrics *ParkingMetrics) *Handler { + // The lot is shared: the resumer charges it at each caller's park + // transition; the handler keeps a reference only for the status page. + lot := newParkingLot(parkCfg, parkMetrics) return &Handler{ - resumer: NewActorResumer(apiClient, withParking(parkCfg)), - parking: newParkingLot(parkCfg, parkMetrics), + resumer: NewActorResumer(apiClient, withParking(parkCfg), withParkingLot(lot)), + parking: lot, } } @@ -116,19 +119,13 @@ func (h *Handler) HandleRequestHeaders(ctx context.Context, md *extproc.RequestM } } - // Admit the request to the parking lot before resuming. While resume is - // in-flight the request occupies a slot; if the actor's worker pool is - // momentarily saturated the resumer parks (retries) here rather than failing - // fast. A full lot sheds the request immediately so the router applies - // backpressure instead of queueing without bound. - release, ok := h.parking.enter(ctx) - if !ok { - return extproc.Result{}, parkingFullErr(actorRef.String()) - } - + // The resumer parks the request if the actor's worker pool is momentarily + // saturated, retrying rather than failing fast. Parking-lot admission + // happens inside, at the park transition: a request resolved on the first + // attempt never occupies a slot, so a full lot sheds only requests that + // would actually wait — never traffic to already-running actors. slog.InfoContext(ctx, "ResumeActor", slog.Any("actor", actorRef)) actor, resumeOutcome, err := h.resumer.ResumeActor(ctx, actorRef) - release(parkOutcomeFor(err)) if err != nil { return extproc.Result{Resume: string(resumeOutcome)}, mapResumeError(actorRef, err) } diff --git a/cmd/atenet/internal/router/ingress/ingress_test.go b/cmd/atenet/internal/router/ingress/ingress_test.go index 6b00a2548..47f152b48 100644 --- a/cmd/atenet/internal/router/ingress/ingress_test.go +++ b/cmd/atenet/internal/router/ingress/ingress_test.go @@ -21,7 +21,9 @@ import ( "errors" "log/slog" "strings" + "sync/atomic" "testing" + "testing/synctest" "time" corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" @@ -338,17 +340,31 @@ func TestHandleRequestHeadersHandlesConnectMethod(t *testing.T) { } } -// TestHandleRequestHeaders_ParkingLotFull verifies that when the parking lot is at capacity -// the request is shed with a 503 before any resume is attempted. -func TestHandleRequestHeaders_ParkingLotFull(t *testing.T) { +// TestHandleRequestHeaders_FullLotServesRunningActor pins the design guarantee +// from docs/request-parking.md: a saturated parking lot cannot starve requests +// to already-running actors, at any lot size. A request whose resume resolves +// on the first attempt never occupies a slot, so it is served even with the +// lot at capacity (issue #1081). +func TestHandleRequestHeaders_FullLotServesRunningActor(t *testing.T) { const testUUID = "123e4567-e89b-12d3-a456-426614174000" authority := testUUID + ".team-a.actors.resources.substrate.ate.dev" var resumeCalled bool clientMock := &mockClient{ - resumeFn: func(ctx context.Context, in *ateapipb.ResumeActorRequest, opts ...grpc.CallOption) (*ateapipb.ResumeActorResponse, error) { + resumeFn: func( + ctx context.Context, + in *ateapipb.ResumeActorRequest, + opts ...grpc.CallOption, + ) (*ateapipb.ResumeActorResponse, error) { resumeCalled = true - return &ateapipb.ResumeActorResponse{Actor: &ateapipb.Actor{Status: &ateapipb.ActorStatus{WorkerAssignment: &ateapipb.WorkerAssignment{WorkerPodIp: "10.0.0.1"}}}}, nil + return &ateapipb.ResumeActorResponse{ + Actor: &ateapipb.Actor{ + Status: &ateapipb.ActorStatus{ + State: ateapipb.ActorState_ACTOR_STATE_RUNNING, + WorkerAssignment: &ateapipb.WorkerAssignment{WorkerPodIp: "10.0.0.1"}, + }, + }, + }, nil }, } @@ -365,21 +381,77 @@ func TestHandleRequestHeaders_ParkingLotFull(t *testing.T) { &corev3.HeaderValue{Key: ":authority", Value: authority}, ) - _, err := h.HandleRequestHeaders(context.Background(), md) - if err == nil { - t.Fatal("expected error when parking lot is full") - } - var reqErr *extproc.ReqError - if !errors.As(err, &reqErr) { - t.Fatalf("expected *extproc.ReqError, got %T (%v)", err, err) + res, err := h.HandleRequestHeaders(context.Background(), md) + if err != nil { + t.Fatalf("a running actor must be served despite a full lot, got: %v", err) } - if reqErr.StatusCode != int(envoy_type.StatusCode_ServiceUnavailable) { - t.Errorf("status code = %d, want %d (503)", reqErr.StatusCode, envoy_type.StatusCode_ServiceUnavailable) + if !resumeCalled { + t.Error("the resume lookup must still run when the lot is full") } - if !strings.Contains(reqErr.Error(), "router at capacity") { - t.Errorf("error body = %q, want it to mention capacity", reqErr.Error()) + const wantTarget = "10.0.0.1:443" + if res.Target != wantTarget { + t.Errorf("target = %q, want %q", res.Target, wantTarget) } - if resumeCalled { - t.Error("resume must not be attempted for a shed request") + if got := h.parking.activeCount(); got != 1 { + t.Errorf("a first-attempt resolution must not occupy a slot; active = %d, want 1 (the priming entry)", got) } } + +// TestHandleRequestHeaders_FullLotShedsParkedRequest verifies the other half +// of lot admission: when the lot is full, a request whose resume actually +// parks (first retryable failure) is shed with 503 "router at capacity" — at +// the park transition, after its single initial attempt, not before any. +func TestHandleRequestHeaders_FullLotShedsParkedRequest(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + const testUUID = "123e4567-e89b-12d3-a456-426614174000" + authority := testUUID + ".team-a.actors.resources.substrate.ate.dev" + + var resumeCalls atomic.Int32 + clientMock := &mockClient{ + resumeFn: func( + ctx context.Context, + in *ateapipb.ResumeActorRequest, + opts ...grpc.CallOption, + ) (*ateapipb.ResumeActorResponse, error) { + resumeCalls.Add(1) + return nil, status.Error(codes.ResourceExhausted, "no free workers available") + }, + } + + h := New(clientMock, ParkedRequestConfig{Budget: 500 * time.Millisecond, Max: 1}, nil) + release, ok := h.parking.enter(context.Background()) + if !ok { + t.Fatal("priming enter should be admitted") + } + defer release(parkOutcomeServed) + + md := requestMetadata(t, authority, + &corev3.HeaderValue{Key: ":authority", Value: authority}, + ) + + _, err := h.HandleRequestHeaders(context.Background(), md) + if err == nil { + t.Fatal("expected error when a parked request finds the lot full") + } + var reqErr *extproc.ReqError + if !errors.As(err, &reqErr) { + t.Fatalf("expected *extproc.ReqError, got %T (%v)", err, err) + } + if reqErr.StatusCode != int(envoy_type.StatusCode_ServiceUnavailable) { + t.Errorf("status code = %d, want %d (503)", reqErr.StatusCode, envoy_type.StatusCode_ServiceUnavailable) + } + if !strings.Contains(reqErr.Error(), "router at capacity") { + t.Errorf("error body = %q, want it to mention capacity", reqErr.Error()) + } + // Shedding happens at the park transition: exactly one attempt has run + // when the caller is turned away. + if got := resumeCalls.Load(); got != 1 { + t.Errorf("expected the caller shed after exactly 1 attempt, got %d", got) + } + + // The abandoned flight retries on until its budget, like any flight + // whose callers left; sleep (fake time) past the budget so it exits + // before the bubble does. + time.Sleep(600 * time.Millisecond) + }) +} diff --git a/cmd/atenet/internal/router/ingress/parking.go b/cmd/atenet/internal/router/ingress/parking.go index 85c394b7b..1e0da0b35 100644 --- a/cmd/atenet/internal/router/ingress/parking.go +++ b/cmd/atenet/internal/router/ingress/parking.go @@ -135,9 +135,12 @@ func DefaultParkedRequestConfig() ParkedRequestConfig { } } -// parkingLot is a bounded, non-blocking admission gate for resume-gated -// requests. Each admitted request holds a slot for the duration of its resume -// attempt; when the lot is full further requests are shed immediately so the +// parkingLot is a bounded, non-blocking admission gate for parked requests. +// A caller enters at its resume flight's park transition — the first +// retryable failure — and holds the slot for the rest of its wait; requests +// resolved on the flight's first attempt never enter, so a saturated lot +// cannot starve traffic to already-running actors (issue #1081). When the lot +// is full, callers reaching their park transition are shed immediately so the // router applies backpressure instead of accumulating waiters without bound. // // With parking disabled (Max <= 0) enter always admits and performs no @@ -154,12 +157,12 @@ func newParkingLot(cfg ParkedRequestConfig, m *ParkingMetrics) *parkingLot { return &parkingLot{cfg: cfg, metrics: m} } -// enter attempts to reserve a parking slot. On success it returns a release -// func and ok=true; the caller MUST invoke release exactly once (passing the -// request outcome, e.g. parkOutcomeServed) when the resume attempt completes. -// ok=false means the lot is full and the request should be shed without -// waiting. When parking is disabled every request is admitted and no slot -// accounting or metrics are recorded. +// enter reserves a parking slot for a caller whose resume flight just parked. +// On success it returns a release func and ok=true; the caller MUST invoke +// release exactly once (passing the request outcome, e.g. parkOutcomeServed) +// when its wait ends. ok=false means the lot is full and the request should +// be shed without waiting further. When parking is disabled every request is +// admitted and no slot accounting or metrics are recorded. func (l *parkingLot) enter(ctx context.Context) (release func(outcome parkOutcome), ok bool) { if !l.cfg.Enabled() { return func(parkOutcome) {}, true diff --git a/cmd/atenet/internal/router/ingress/resumer.go b/cmd/atenet/internal/router/ingress/resumer.go index b515e19e5..d62dd25ae 100644 --- a/cmd/atenet/internal/router/ingress/resumer.go +++ b/cmd/atenet/internal/router/ingress/resumer.go @@ -17,6 +17,7 @@ package ingress import ( "context" "math" + "sync" "sync/atomic" "time" @@ -26,7 +27,6 @@ import ( "github.com/agent-substrate/substrate/pkg/proto/ateapipb" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/trace" - "golang.org/x/sync/singleflight" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "k8s.io/apimachinery/pkg/util/wait" @@ -90,7 +90,19 @@ type resumeCallResult struct { // ActorResumer coordinates safe, deduplicated resumption of actors. type ActorResumer struct { apiClient ateapipb.ControlClient - flight singleflight.Group + + // mu guards flights, the per-actor registry of in-flight resumes. + mu sync.Mutex + // flights deduplicates concurrent resumes per actor, singleflight-style: + // the first caller creates the flight and later callers attach to it. An + // entry is removed the moment its flight completes. + flights map[string]*resumeFlight + + // lot bounds how many callers may wait on parked flights at once. A slot + // is taken only at a flight's park transition — never for the fast path — + // so a saturated lot cannot starve requests to already-running actors + // (issue #1081). A nil lot admits everyone. + lot *parkingLot // parkEnabled makes transient worker-pool saturation (FailedPrecondition) // retryable, so a request is parked and retried until budget rather than @@ -103,7 +115,7 @@ type ActorResumer struct { backoff wait.Backoff // nextID is a counter assigned to each incoming ResumeActor call. // Used as a unique ID to identify requests (reqID) and disambiguate the - // leader vs joiners for singleflight outcome classification. + // leader vs joiners for flight outcome classification. nextID uint64 } @@ -125,9 +137,17 @@ func withParking(cfg ParkedRequestConfig) resumerOption { } } +// withParkingLot bounds concurrent parked callers with lot. A caller acquires +// a slot only at its flight's park transition; when the lot is full at that +// moment the caller is shed with errParkingLotFull instead of waiting. +func withParkingLot(lot *parkingLot) resumerOption { + return func(r *ActorResumer) { r.lot = lot } +} + func NewActorResumer(apiClient ateapipb.ControlClient, opts ...resumerOption) *ActorResumer { r := &ActorResumer{ apiClient: apiClient, + flights: make(map[string]*resumeFlight), budget: failFastResumeBudget, backoff: resumeBackoff(DefaultParkedRequestRetryInterval, DefaultParkedRequestRetryFactor, DefaultParkedRequestRetryJitter), @@ -162,7 +182,9 @@ func (r *ActorResumer) retryable(err error) bool { // ResumeActor ensures the requested actor is running. It deduplicates concurrent // requests within the process and, when parking is enabled, holds the request -// while retrying transient failures until the budget elapses. +// while retrying transient failures until the budget elapses. A caller occupies +// a parking-lot slot only while its flight is actually parked; a resume that +// resolves on the first attempt never touches the lot. func (r *ActorResumer) ResumeActor(ctx context.Context, actorRef resources.ActorRef) (*ateapipb.Actor, ResumeOutcome, error) { ctx, span := otel.Tracer(extproc.ServiceName).Start(ctx, "ResumeActor", trace.WithAttributes(ateattr.ActorRefAttributes(actorRef)...)) @@ -170,110 +192,148 @@ func (r *ActorResumer) ResumeActor(ctx context.Context, actorRef resources.Actor reqID := atomic.AddUint64(&r.nextID, 1) - ch := r.flight.DoChan(actorRef.String(), func() (interface{}, error) { - // We detach the context from the first caller using a fixed background budget. - // This guarantees that if Caller 1 disconnects or times out, the underlying - // resume operation continues running for Caller 2 and Caller 3 without failing. - // - // The budget is therefore per-FLIGHT, not per-caller: its clock starts with - // the first caller, and later callers de-duplicated onto this flight share - // its remaining budget and outcome. A late joiner can see budget_exhausted - // after waiting far less than a full budget itself — the accepted cost of - // one control-plane RPC per hot actor (see docs/request-parking.md). - bgCtx, bgCancel := context.WithTimeout(context.Background(), r.budget) - defer bgCancel() - // The budget bounds the RETRY LOOP only — it never cancels an - // in-flight ResumeActor. ateapi durably claims the worker and marks - // the actor RESUMING before the expensive snapshot restore begins, - // rolls back neither on cancellation, and nothing reclaims a RESUMING - // actor whose worker pod is alive — a budget cancel therefore throws - // the restore away and strands the worker (#675). An attempt still - // running when the budget elapses is waited for and its real result - // classified below; ateapi's own server-side RPC deadline bounds it. - attemptCtx := context.WithoutCancel(bgCtx) - - backoff := r.backoff - - var resumeResp *ateapipb.ResumeActorResponse - var lastRetryErr error - - err := wait.ExponentialBackoffWithContext(bgCtx, backoff, func(context.Context) (bool, error) { - var err error - resumeResp, err = r.apiClient.ResumeActor(attemptCtx, &ateapipb.ResumeActorRequest{ - Actor: actorRef.ToObjectRef(), - }) - if err == nil { - return true, nil - } - - if r.retryable(err) { - lastRetryErr = err // remember it in case the budget elapses - return false, nil // park: retry until the budget elapses - } - return false, err - }) + key := actorRef.String() + r.mu.Lock() + f, ok := r.flights[key] + if !ok { + f = newResumeFlight() + r.flights[key] = f + go r.runFlight(f, key, actorRef, reqID) + } + r.mu.Unlock() + + return r.awaitFlight(ctx, f, reqID) +} - if err != nil { - // If the budget elapsed while we were still blocked on a retryable - // condition, surface that underlying error rather than the generic - // wait/deadline error so the HTTP boundary maps it faithfully - // (e.g. 503 "no free workers available") instead of a misleading - // timeout. The wrapper marks the exhaustion explicitly for the - // parking wait-duration metric. - // - // wait.Interrupted covers the budget landing between retries; the - // bgCtx check covers an attempt that came back with a retryable - // error only after the budget had already elapsed (the loop then - // exits with the context error). The RPC itself is never canceled, - // so the loop cannot end before its first attempt has completed — - // lastRetryErr is always the attempt's real answer here, and a - // definitive error (NotFound, ...) still passes through untouched. - if lastRetryErr != nil && (bgCtx.Err() != nil || wait.Interrupted(err)) { - return &resumeCallResult{leaderID: reqID, err: &budgetExhaustedError{lastErr: lastRetryErr}}, nil - } - return &resumeCallResult{leaderID: reqID, err: err}, nil +// runFlight executes one shared resume for actorRef and publishes the outcome +// to every caller attached to f. reqID identifies the caller that created the +// flight (the cold-activation leader). +func (r *ActorResumer) runFlight(f *resumeFlight, key string, actorRef resources.ActorRef, reqID uint64) { + // We detach the context from the first caller using a fixed background budget. + // This guarantees that if Caller 1 disconnects or times out, the underlying + // resume operation continues running for Caller 2 and Caller 3 without failing. + // + // The budget is therefore per-FLIGHT, not per-caller: its clock starts with + // the first caller, and later callers de-duplicated onto this flight share + // its remaining budget and outcome. A late joiner can see budget_exhausted + // after waiting far less than a full budget itself — the accepted cost of + // one control-plane RPC per hot actor (see docs/request-parking.md). + bgCtx, bgCancel := context.WithTimeout(context.Background(), r.budget) + defer bgCancel() + // The budget bounds the RETRY LOOP only — it never cancels an + // in-flight ResumeActor. ateapi durably claims the worker and marks + // the actor RESUMING before the expensive snapshot restore begins, + // rolls back neither on cancellation, and nothing reclaims a RESUMING + // actor whose worker pod is alive — a budget cancel therefore throws + // the restore away and strands the worker (#675). An attempt still + // running when the budget elapses is waited for and its real result + // classified below; ateapi's own server-side RPC deadline bounds it. + attemptCtx := context.WithoutCancel(bgCtx) + + backoff := r.backoff + + var resumeResp *ateapipb.ResumeActorResponse + var lastRetryErr error + + err := wait.ExponentialBackoffWithContext(bgCtx, backoff, func(context.Context) (bool, error) { + var err error + resumeResp, err = r.apiClient.ResumeActor(attemptCtx, &ateapipb.ResumeActorRequest{ + Actor: actorRef.ToObjectRef(), + }) + if err == nil { + return true, nil } - return &resumeCallResult{ - actor: resumeResp.GetActor(), - resumed: resumeResp.GetResumed(), - leaderID: reqID, - }, nil + if r.retryable(err) { + f.park() + lastRetryErr = err // remember it in case the budget elapses + return false, nil // park: retry until the budget elapses + } + return false, err }) + r.publish(f, key, flightResult(bgCtx, resumeResp, err, lastRetryErr, reqID)) +} + +// flightResult classifies the retry loop's terminal state into the shared +// result every caller attached to the flight receives. bgCtx is the flight's +// budget context, consulted only for whether the budget expired. +func flightResult(bgCtx context.Context, resumeResp *ateapipb.ResumeActorResponse, err, lastRetryErr error, reqID uint64) *resumeCallResult { + result := &resumeCallResult{leaderID: reqID} + switch { + case err == nil: + result.actor = resumeResp.GetActor() + result.resumed = resumeResp.GetResumed() + // If the budget elapsed while we were still blocked on a retryable + // condition, surface that underlying error rather than the generic + // wait/deadline error so the HTTP boundary maps it faithfully + // (e.g. 503 "no free workers available") instead of a misleading + // timeout. The wrapper marks the exhaustion explicitly for the + // parking wait-duration metric. + // + // wait.Interrupted covers the budget landing between retries; the + // bgCtx check covers an attempt that came back with a retryable + // error only after the budget had already elapsed (the loop then + // exits with the context error). The RPC itself is never canceled, + // so the loop cannot end before its first attempt has completed — + // lastRetryErr is always the attempt's real answer here, and a + // definitive error (NotFound, ...) still passes through untouched. + case lastRetryErr != nil && (bgCtx.Err() != nil || wait.Interrupted(err)): + result.err = &budgetExhaustedError{lastErr: lastRetryErr} + default: + result.err = err + } + return result +} + +// awaitFlight waits for f's outcome on behalf of one caller. The wait is +// two-phase: while the flight is resolving the caller holds nothing; once the +// flight parks (or if it already has), the caller must hold a parking-lot slot +// to keep waiting and is shed with errParkingLotFull when the lot is full. +func (r *ActorResumer) awaitFlight(ctx context.Context, f *resumeFlight, reqID uint64) (*ateapipb.Actor, ResumeOutcome, error) { select { case <-ctx.Done(): - // The caller's request context was canceled before the singleflight resume completed. - // Return early with ResumeOutcomeNone ("none") + // The caller's request context was canceled before the shared resume + // completed. Return early with ResumeOutcomeNone ("none"). return nil, ResumeOutcomeNone, ctx.Err() - case res := <-ch: - callRes, _ := res.Val.(*resumeCallResult) - if callRes == nil { - if res.Err != nil { - return nil, ResumeOutcomeNone, res.Err - } - return nil, ResumeOutcomeNone, status.Error(codes.Internal, "resume call returned nil result") - } + case <-f.done: + // Fast path: the flight resolved without ever parking (or finished + // before this caller reacted to parking) — the lot is never touched. + return f.callerResult(reqID) + case <-f.parked: + } - // On error, return ResumeOutcomeNone ("none") so the failure is tagged - // under the 'outcome' label rather than misreported as an activation. - if callRes.err != nil { - return nil, ResumeOutcomeNone, callRes.err - } + // The flight parked. If its result raced in anyway, serve it without + // charging the lot. + select { + case <-f.done: + return f.callerResult(reqID) + default: + } - // Disambiguate singleflight resume outcome: - // - ResumeOutcomeNone ("none"): resumed == false, actor was already active/running. - // - ResumeOutcomeTriggered ("triggered"): Cold activation leader (resumed == true, caller's reqID == leaderID). - // - ResumeOutcomeJoined ("joined"): Cold activation joiner (resumed == true, caller's reqID != leaderID). - outcome := ResumeOutcomeNone - if callRes.resumed { - if callRes.leaderID == reqID { - outcome = ResumeOutcomeTriggered - } else { - outcome = ResumeOutcomeJoined - } - } + release, ok := r.enterLot(ctx) + if !ok { + return nil, ResumeOutcomeNone, errParkingLotFull + } + var finalErr error + defer func() { release(parkOutcomeFor(finalErr)) }() + + select { + case <-ctx.Done(): + finalErr = ctx.Err() + return nil, ResumeOutcomeNone, finalErr + case <-f.done: + actor, outcome, err := f.callerResult(reqID) + finalErr = err + return actor, outcome, err + } +} - return callRes.actor, outcome, nil +// enterLot admits the caller to the parking lot, treating a nil lot as +// unbounded (no admission control). +func (r *ActorResumer) enterLot(ctx context.Context) (func(parkOutcome), bool) { + if r.lot == nil { + return func(parkOutcome) {}, true } + return r.lot.enter(ctx) } diff --git a/cmd/atenet/internal/router/ingress/resumer_test.go b/cmd/atenet/internal/router/ingress/resumer_test.go index 648328046..7a2a9a50c 100644 --- a/cmd/atenet/internal/router/ingress/resumer_test.go +++ b/cmd/atenet/internal/router/ingress/resumer_test.go @@ -572,6 +572,343 @@ func testCallerCancelDoesNotAbortFlight(t *testing.T) { } } +// TestActorResumer_LotAdmission pins WHEN a caller occupies a parking-lot +// slot: never while its flight is resolving, always while it is parked, and +// shed at the park transition when the lot is full (issue #1081). Timed cases +// run inside synctest bubbles so the parked retry loop's waits are fake time. +func TestActorResumer_LotAdmission(t *testing.T) { + const ( + testActorName = "actor-lot" + testAtespace = "team-a" + expectedIP = "10.0.0.99" + ) + testActorRef := resources.ActorRef{Atespace: testAtespace, Name: testActorName} + runningResp := func() *ateapipb.ResumeActorResponse { + return &ateapipb.ResumeActorResponse{ + Actor: &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Name: testActorName}, + Status: &ateapipb.ActorStatus{ + State: ateapipb.ActorState_ACTOR_STATE_RUNNING, + WorkerAssignment: &ateapipb.WorkerAssignment{WorkerPodIp: expectedIP}, + }, + }, + Resumed: true, + } + } + + t.Run("FastFlightNeverEntersLot", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + mock := &resumerMockClient{ + resumeFn: func( + ctx context.Context, + in *ateapipb.ResumeActorRequest, + opts ...grpc.CallOption, + ) (*ateapipb.ResumeActorResponse, error) { + return runningResp(), nil + }, + } + cfg := ParkedRequestConfig{Max: 1, Budget: 5 * time.Second} + lot := newParkingLot(cfg, nil) + // Fill the only slot: any lot entry would shed, so success proves + // the fast path never asked. + release, ok := lot.enter(context.Background()) + if !ok { + t.Fatal("priming enter should be admitted") + } + defer release(parkOutcomeServed) + + resumer := NewActorResumer(mock, withParking(cfg), withParkingLot(lot)) + actor, _, err := resumer.ResumeActor(context.Background(), testActorRef) + if err != nil { + t.Fatalf("a first-attempt resolution must be served despite a full lot: %v", err) + } + if actor.GetStatus().GetWorkerAssignment().GetWorkerPodIp() != expectedIP { + t.Errorf("expected IP %q, got %q", expectedIP, actor.GetStatus().GetWorkerAssignment().GetWorkerPodIp()) + } + if got := lot.activeCount(); got != 1 { + t.Errorf("fast path must not take a slot; active = %d, want 1 (the priming entry)", got) + } + }) + }) + + t.Run("ParkTransitionAcquiresSlot", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + cfg := ParkedRequestConfig{Max: 2, Budget: 5 * time.Second} + lot := newParkingLot(cfg, nil) + var mu sync.Mutex + var calls, activeDuringRetry int + mock := &resumerMockClient{ + resumeFn: func( + ctx context.Context, + in *ateapipb.ResumeActorRequest, + opts ...grpc.CallOption, + ) (*ateapipb.ResumeActorResponse, error) { + mu.Lock() + calls++ + n := calls + mu.Unlock() + if n == 1 { + return nil, status.Error(codes.ResourceExhausted, "no free workers available") + } + // By the retry, the parked caller must already hold its + // slot: the bubble advances past the backoff sleep only + // once every goroutine — the caller included — is blocked. + mu.Lock() + activeDuringRetry = lot.activeCount() + mu.Unlock() + return runningResp(), nil + }, + } + + resumer := NewActorResumer(mock, withParking(cfg), withParkingLot(lot)) + actor, _, err := resumer.ResumeActor(context.Background(), testActorRef) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if actor.GetStatus().GetWorkerAssignment().GetWorkerPodIp() != expectedIP { + t.Errorf("expected IP %q, got %q", expectedIP, actor.GetStatus().GetWorkerAssignment().GetWorkerPodIp()) + } + mu.Lock() + defer mu.Unlock() + if activeDuringRetry != 1 { + t.Errorf("caller must hold a slot while its flight is parked; active during retry = %d, want 1", activeDuringRetry) + } + if got := lot.activeCount(); got != 0 { + t.Errorf("slot must be released when the wait ends; active = %d, want 0", got) + } + }) + }) + + t.Run("ShedWhenLotFullAtParkTransition", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + cfg := ParkedRequestConfig{Max: 1, Budget: 500 * time.Millisecond} + lot := newParkingLot(cfg, nil) + release, ok := lot.enter(context.Background()) + if !ok { + t.Fatal("priming enter should be admitted") + } + defer release(parkOutcomeServed) + + var mu sync.Mutex + var calls int + mock := &resumerMockClient{ + resumeFn: func( + ctx context.Context, + in *ateapipb.ResumeActorRequest, + opts ...grpc.CallOption, + ) (*ateapipb.ResumeActorResponse, error) { + mu.Lock() + calls++ + mu.Unlock() + return nil, status.Error(codes.ResourceExhausted, "no free workers available") + }, + } + + resumer := NewActorResumer(mock, withParking(cfg), withParkingLot(lot)) + _, outcome, err := resumer.ResumeActor(context.Background(), testActorRef) + if !errors.Is(err, errParkingLotFull) { + t.Fatalf("expected errParkingLotFull, got %v", err) + } + if outcome != ResumeOutcomeNone { + t.Errorf("shed caller outcome = %q, want %q", outcome, ResumeOutcomeNone) + } + // The caller was turned away at the transition: exactly one attempt + // had run. + mu.Lock() + if calls != 1 { + t.Errorf("expected the caller shed after exactly 1 attempt, got %d", calls) + } + mu.Unlock() + + // Its abandoned flight retries on until the budget; sleep (fake + // time) past it so the flight exits before the bubble does. + time.Sleep(600 * time.Millisecond) + }) + }) + + t.Run("JoinerToParkedFlightNeedsSlot", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + cfg := ParkedRequestConfig{Max: 1, Budget: 5 * time.Second} + lot := newParkingLot(cfg, nil) + var mu sync.Mutex + var calls int + proceed := make(chan struct{}) + mock := &resumerMockClient{ + resumeFn: func( + ctx context.Context, + in *ateapipb.ResumeActorRequest, + opts ...grpc.CallOption, + ) (*ateapipb.ResumeActorResponse, error) { + mu.Lock() + calls++ + n := calls + mu.Unlock() + if n == 1 { + return nil, status.Error(codes.ResourceExhausted, "no free workers available") + } + <-proceed + return runningResp(), nil + }, + } + resumer := NewActorResumer(mock, withParking(cfg), withParkingLot(lot)) + + // The leader parks and takes the lot's only slot. + type result struct { + actor *ateapipb.Actor + outcome ResumeOutcome + err error + } + leaderCh := make(chan result, 1) + go func() { + a, o, err := resumer.ResumeActor(context.Background(), testActorRef) + leaderCh <- result{a, o, err} + }() + // Wait blocks until the leader is durably parked again — past the + // non-blocking lot entry, i.e. holding the slot. + synctest.Wait() + + // A joiner attaching to the already-parked flight must take its own + // slot; the lot is full, so it is shed while the leader keeps waiting. + _, outcome, err := resumer.ResumeActor(context.Background(), testActorRef) + if !errors.Is(err, errParkingLotFull) { + t.Fatalf("joiner: expected errParkingLotFull, got %v", err) + } + if outcome != ResumeOutcomeNone { + t.Errorf("joiner outcome = %q, want %q", outcome, ResumeOutcomeNone) + } + + close(proceed) + res := <-leaderCh + if res.err != nil { + t.Fatalf("leader: unexpected error: %v", res.err) + } + if res.outcome != ResumeOutcomeTriggered { + t.Errorf("leader outcome = %q, want %q", res.outcome, ResumeOutcomeTriggered) + } + if res.actor.GetStatus().GetWorkerAssignment().GetWorkerPodIp() != expectedIP { + t.Errorf("leader IP = %q, want %q", res.actor.GetStatus().GetWorkerAssignment().GetWorkerPodIp(), expectedIP) + } + if got := lot.activeCount(); got != 0 { + t.Errorf("all slots must be released; active = %d, want 0", got) + } + }) + }) + + t.Run("BudgetExhaustionReleasesSlot", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + cfg := ParkedRequestConfig{Max: 1, Budget: 1 * time.Second} + lot := newParkingLot(cfg, nil) + mock := &resumerMockClient{ + resumeFn: func( + ctx context.Context, + in *ateapipb.ResumeActorRequest, + opts ...grpc.CallOption, + ) (*ateapipb.ResumeActorResponse, error) { + return nil, status.Error(codes.ResourceExhausted, "no free workers available") + }, + } + + resumer := NewActorResumer(mock, withParking(cfg), withParkingLot(lot)) + _, _, err := resumer.ResumeActor(context.Background(), testActorRef) + var budget *budgetExhaustedError + if !errors.As(err, &budget) { + t.Fatalf("expected budget exhaustion, got %T (%v)", err, err) + } + if got := status.Code(err); got != codes.ResourceExhausted { + t.Errorf("expected the underlying capacity code to surface, got %v", got) + } + if got := lot.activeCount(); got != 0 { + t.Errorf("slot must be released on budget exhaustion; active = %d, want 0", got) + } + }) + }) + + t.Run("CancelWhileParkedReleasesSlot", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + cfg := ParkedRequestConfig{Max: 1, Budget: 5 * time.Second} + lot := newParkingLot(cfg, nil) + proceed := make(chan struct{}) + var mu sync.Mutex + var calls int + mock := &resumerMockClient{ + resumeFn: func( + ctx context.Context, + in *ateapipb.ResumeActorRequest, + opts ...grpc.CallOption, + ) (*ateapipb.ResumeActorResponse, error) { + mu.Lock() + calls++ + n := calls + mu.Unlock() + if n == 1 { + return nil, status.Error(codes.ResourceExhausted, "no free workers available") + } + <-proceed + return runningResp(), nil + }, + } + resumer := NewActorResumer(mock, withParking(cfg), withParkingLot(lot)) + + ctx, cancel := context.WithCancel(context.Background()) + errCh := make(chan error, 1) + go func() { + _, _, err := resumer.ResumeActor(ctx, testActorRef) + errCh <- err + }() + synctest.Wait() + if got := lot.activeCount(); got != 1 { + t.Fatalf("parked caller must hold a slot; active = %d, want 1", got) + } + + cancel() + if err := <-errCh; !errors.Is(err, context.Canceled) { + t.Fatalf("expected context.Canceled, got %v", err) + } + if got := lot.activeCount(); got != 0 { + t.Errorf("slot must be released when the caller disconnects; active = %d, want 0", got) + } + + // Let the abandoned flight run out: it wakes from its backoff + // (fake time), finds proceed closed, completes, and exits before + // the bubble does. + close(proceed) + time.Sleep(200 * time.Millisecond) + }) + }) + + t.Run("CompletedFlightIsForgotten", func(t *testing.T) { + var mu sync.Mutex + var calls int + mock := &resumerMockClient{ + resumeFn: func( + ctx context.Context, + in *ateapipb.ResumeActorRequest, + opts ...grpc.CallOption, + ) (*ateapipb.ResumeActorResponse, error) { + mu.Lock() + calls++ + mu.Unlock() + return runningResp(), nil + }, + } + resumer := NewActorResumer(mock, withParking(ParkedRequestConfig{Max: 1, Budget: time.Second})) + for i := 0; i < 2; i++ { + _, outcome, err := resumer.ResumeActor(context.Background(), testActorRef) + if err != nil { + t.Fatalf("call %d: unexpected error: %v", i, err) + } + if outcome != ResumeOutcomeTriggered { + t.Errorf("call %d: outcome = %q, want %q (each sequential call starts a fresh flight)", i, outcome, ResumeOutcomeTriggered) + } + } + mu.Lock() + defer mu.Unlock() + if calls != 2 { + t.Errorf("sequential calls must not share a completed flight; RPCs = %d, want 2", calls) + } + }) +} + func TestResumeBackoffHasNoCap(t *testing.T) { // Regression: the resume backoff must NOT set wait.Backoff.Cap. delay() zeroes // Steps the moment the delay reaches Cap, which would end parking retries far diff --git a/docs/request-parking.md b/docs/request-parking.md index 4a90c74ba..2747167ac 100644 --- a/docs/request-parking.md +++ b/docs/request-parking.md @@ -58,17 +58,20 @@ client is held either way. 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 singleflight entry. +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. -To bound resource use and provide backpressure, the router admits requests to a -**parking lot** of fixed capacity (`--parked-request-max`, default `1024`). Each -in-flight resume occupies one slot. When the lot is full, further requests are -shed immediately with `503 "actor unavailable: router at capacity"` rather -than queueing without bound. +To bound resource use and provide backpressure, parked requests are admitted to +a **parking lot** of fixed capacity (`--parked-request-max`, default `1024`). A +request occupies a slot only from the moment it actually parks — its resume +flight's first retryable failure. A request resolved on the flight's first +attempt (the actor was already running) never occupies a slot. When the lot is +full, a request reaching its park transition is shed with `503 "actor +unavailable: router at capacity"` rather than queueing without bound — at the +cost of exactly the one resume attempt that revealed it would have to wait. Every parked request holds one ext_proc stream — one active request against Envoy's ext_proc cluster — for its entire wait, while ordinary requests hold @@ -77,14 +80,16 @@ is therefore the hard ceiling on concurrent parked requests. By default the router **derives** it as twice `--parked-request-max` (minimum `1024`), so the lot always fits and an equal share of **fast-path headroom** remains — a saturated lot cannot starve requests to already-running actors, at any lot -size. `--extproc-max-requests` overrides the derivation; explicit values are +size. The lot upholds the same guarantee on its side: admission happens at the +park transition, so fast-path requests never compete for slots (#1081). +`--extproc-max-requests` overrides the derivation; explicit values are validated `>= --parked-request-max` at startup, because a breaker below the lot would silently truncate it — Envoy would reject the overflow itself, with 503s that never reach the lot and never count in `parking.rejected`. Concurrent requests for the *same* actor are de-duplicated by the resumer's -`singleflight` group: they share a single in-flight `ResumeActor` call and all -park on its result, so a hot actor consumes N parking slots but only one +per-actor flight registry: they share a single in-flight `ResumeActor` call and +all park on its result, so a hot actor consumes N parking slots but only one control-plane RPC. **The park budget is per-flight, not per-request.** The budget clock starts @@ -98,15 +103,17 @@ expected under sustained saturation.) ### What is *not* parked -Only transient conditions — capacity (`FailedPrecondition`), concurrency -(`Aborted`), and control-plane unavailability (`Unavailable`) — are parked. -Errors that will not resolve by waiting are returned immediately (fail fast): +Only transient conditions — capacity (`ResourceExhausted`), transient actor +state (`FailedPrecondition`), concurrency (`Aborted`), and control-plane +unavailability (`Unavailable`) — are parked. Errors that will not resolve by +waiting are returned immediately (fail fast): | Resume result | Behavior | | -------------------------------------- | --------------------------------- | | `OK` | Route to worker | | `Aborted` (concurrent resume) | Retry (always) | -| `FailedPrecondition` (no free worker) | **Park & retry** (when enabled) | +| `ResourceExhausted` (no free worker) | **Park & retry** (when enabled) | +| `FailedPrecondition` (transient state) | **Park & retry** (when enabled) | | `Unavailable` (control-plane blip) | **Park & retry** (when enabled) | | `NotFound` | Fail fast → `404` | | `DeadlineExceeded` | Fail fast → `504` | @@ -133,7 +140,7 @@ so a parked request always gets its full budget and a normal verdict (routed | Flag | Default | Meaning | | -------------------------------- | ------- | ------------------------------------------------------------------ | | `--parked-request-budget` | `5s` | Park budget per resume *flight*; requests de-duplicated onto an in-flight resume share its remaining budget (see Behavior). | -| `--parked-request-max` | `1024` | Max concurrent parked/in-flight resume requests; excess shed (503). `0` disables parking. | +| `--parked-request-max` | `1024` | Max concurrent **parked** requests (a slot is taken at the park transition, never for a first-attempt lookup); requests parking beyond it are shed (503). `0` disables parking. | | `--parked-request-retry-interval` | `100ms` | Delay before a parked request's first resume retry. | | `--parked-request-retry-factor` | `1.1` | Multiplier applied to the retry delay after each attempt (>= 1). | | `--parked-request-retry-jitter` | `0.1` | Random fraction in `[0, 1)` added per retry to de-synchronize parked requests. | @@ -148,10 +155,11 @@ bounds the wait. - `atenet.router.parking.active` — up/down counter: requests currently parked. - `atenet.router.parking.wait.duration` — histogram (seconds) of time spent - parked. Recorded **exactly once per admitted request**, at the moment its - resume attempt completes; never recorded for shed requests (those only - increment `parking.rejected`) nor when parking is disabled. The `outcome` - label says how the park ended: + parked. Recorded **exactly once per parked request**, at the moment its wait + ends; never recorded for requests served on their flight's first attempt + (those never park), for shed requests (those only increment + `parking.rejected`), nor when parking is disabled. The `outcome` label says + how the park ended: | `outcome` | When it is set | | ------------------ | --------------------------------------------------------------------------- |