From ea3c9cbd3694b498ec490e60e6d9e4b36e2c145d Mon Sep 17 00:00:00 2001 From: Du Bin Date: Sat, 15 Aug 2026 14:21:27 +0000 Subject: [PATCH] atenet: give resume callers independent park budgets Replace the fixed-budget singleflight with an explicit per-Actor flight whose execution deadline follows the latest caller while each request retains its own wait budget. Wake an accumulated long backoff when fresh demand arrives and document the resulting lifecycle.\n\nFixes #613 --- cmd/atenet/internal/router/cmd.go | 2 +- cmd/atenet/internal/router/ingress/resumer.go | 315 +++++++++++++----- .../internal/router/ingress/resumer_test.go | 117 ++++++- docs/request-parking.md | 47 +-- 4 files changed, 369 insertions(+), 112 deletions(-) diff --git a/cmd/atenet/internal/router/cmd.go b/cmd/atenet/internal/router/cmd.go index ebca821f55..64add6c26c 100644 --- a/cmd/atenet/internal/router/cmd.go +++ b/cmd/atenet/internal/router/cmd.go @@ -72,7 +72,7 @@ func NewRouterCmd() *cobra.Command { cmd.Flags().BoolVar(&cfg.Auth.AteapiUseTokenAuth, "ateapi-use-token-auth", false, "Authenticate to ateapi with the Bearer token from --ateapi-token-file instead of the client certificate from --ateapi-client-cert.") cmd.Flags().StringVar(&cfg.Auth.AteapiTokenFile, "ateapi-token-file", "", "Projected SA token file used as Bearer credential. Required with --ateapi-use-token-auth, ignored otherwise.") cmd.Flags().DurationVar(&cfg.RouteTimeout, "route-timeout", defaultRouteTimeout, "Envoy's end-to-end timeout on the workload route, bounding one request from the ingress listener to the actor's response. Raise it for actors whose turns legitimately run long — a harness relaying an LLM completion holds the request open for the whole generation. This does not cover the resume that may precede the request; see --parked-request-budget") - cmd.Flags().DurationVar(&cfg.ParkedRequest.Budget, "parked-request-budget", ingress.DefaultParkedRequestBudget, "Maximum time a resume flight keeps a request parked (held and retried) waiting for its actor to become routable; concurrent requests for the same actor share one flight and its budget") + cmd.Flags().DurationVar(&cfg.ParkedRequest.Budget, "parked-request-budget", ingress.DefaultParkedRequestBudget, "Maximum time each request is parked (held and retried) waiting for its actor to become routable; concurrent requests for the same actor share a control-plane retry loop without sharing wait budgets") cmd.Flags().IntVar(&cfg.ParkedRequest.Max, "parked-request-max", ingress.DefaultParkedRequestMax, "Maximum number of requests that may be parked simultaneously; excess requests are shed with 503. 0 disables parking (requests fail fast on worker-pool saturation)") cmd.Flags().DurationVar(&cfg.ParkedRequest.RetryInterval, "parked-request-retry-interval", ingress.DefaultParkedRequestRetryInterval, "Delay before a parked request's first resume retry") cmd.Flags().Float64Var(&cfg.ParkedRequest.RetryFactor, "parked-request-retry-factor", ingress.DefaultParkedRequestRetryFactor, "Multiplier applied to the retry delay after each attempt; must be >= 1") diff --git a/cmd/atenet/internal/router/ingress/resumer.go b/cmd/atenet/internal/router/ingress/resumer.go index ce4bd943ff..eda0998d57 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" @@ -44,9 +44,8 @@ const failFastResumeBudget = 15 * time.Second // the delay reaches Cap, which would end retries long before the parking budget // (a Cap of 2s stops the loop in ~7 steps regardless of the budget). A gentle // Factor keeps the gap small on its own — from 100ms at the default 1.1 the gap -// only grows to ~0.5s over a 5s budget — while Steps is set high so the budget -// context passed to ExponentialBackoffWithContext, not the step count, bounds -// the wait. +// only grows to ~0.5s over a 5s budget — while Steps is set high so flight and +// caller deadlines, not the step count, bound retries. func resumeBackoff(interval time.Duration, factor, jitter float64) wait.Backoff { return wait.Backoff{ Steps: math.MaxInt32, @@ -56,17 +55,16 @@ func resumeBackoff(interval time.Duration, factor, jitter float64) wait.Backoff } } -// budgetExhaustedError marks a resume that was still blocked on a retryable -// condition (e.g. "no free workers available") when the parking budget elapsed. -// It wraps the last retryable error, so the HTTP boundary still maps the -// underlying gRPC status faithfully (503 with the capacity message), while the -// parking metrics can report budget exhaustion as its own outcome. +// budgetExhaustedError marks a caller whose parking budget elapsed. It wraps +// the last retryable error when one is available, so the HTTP boundary can +// preserve that status (for example, a capacity 503). If the first RPC consumes +// the whole budget, it wraps context.DeadlineExceeded instead and maps to 504. type budgetExhaustedError struct{ lastErr error } func (e *budgetExhaustedError) Error() string { return e.lastErr.Error() } func (e *budgetExhaustedError) Unwrap() error { return e.lastErr } -// ResumeOutcome indicates the singleflight execution state of an actor resumption request. +// ResumeOutcome indicates the shared-flight execution state of an actor resumption request. type ResumeOutcome string const ( @@ -81,29 +79,138 @@ type resumeCallResult struct { // false if the actor was already running resumed bool // leaderID is the unique request ID (reqID) of the leader that initiated - // the singleflight execution. It helps disambiguates the leader caller + // the shared execution. It helps disambiguate the leader caller // (ResumeOutcomeTriggered) from joiner callers (ResumeOutcomeJoined). leaderID uint64 err error } +// resumeFlight owns one control-plane retry loop shared by requests for an +// Actor. Its deadline tracks the latest joining caller, while each caller has +// an independent timer and can stop waiting earlier. +type resumeFlight struct { + ctx context.Context + cancel context.CancelFunc + done chan struct{} + joined chan struct{} + + mu sync.Mutex + deadline time.Time + timer *time.Timer + expired bool + finished bool + lastRetryErr error + result *resumeCallResult +} + +func newResumeFlight(deadline time.Time) *resumeFlight { + ctx, cancel := context.WithCancel(context.Background()) + f := &resumeFlight{ + ctx: ctx, + cancel: cancel, + done: make(chan struct{}), + joined: make(chan struct{}, 1), + deadline: deadline, + } + f.timer = time.AfterFunc(time.Until(deadline), f.expire) + return f +} + +// join extends the shared loop through the caller's deadline and notifies the +// retry scheduler that fresh demand arrived. The notification is coalesced: +// ten simultaneous joiners need one backoff adjustment, not ten RPCs. +func (f *resumeFlight) join(deadline time.Time) bool { + f.mu.Lock() + if f.finished || f.expired { + f.mu.Unlock() + return false + } + if !time.Now().Before(f.deadline) { + f.expired = true + f.mu.Unlock() + f.cancel() + return false + } + if deadline.After(f.deadline) { + f.deadline = deadline + f.timer.Reset(time.Until(deadline)) + } + f.mu.Unlock() + + select { + case f.joined <- struct{}{}: + default: + } + return true +} + +func (f *resumeFlight) expire() { + f.mu.Lock() + if f.finished || f.expired { + f.mu.Unlock() + return + } + if remaining := time.Until(f.deadline); remaining > 0 { + // A previous timer callback can race with join extending the deadline. + // Re-check the guarded value instead of canceling the extended flight. + f.timer.Reset(remaining) + f.mu.Unlock() + return + } + f.expired = true + f.mu.Unlock() + f.cancel() +} + +func (f *resumeFlight) setLastRetryErr(err error) { + f.mu.Lock() + f.lastRetryErr = err + f.mu.Unlock() +} + +func (f *resumeFlight) budgetError() error { + f.mu.Lock() + defer f.mu.Unlock() + if f.lastRetryErr == nil { + // The caller's park budget still expired even though an in-flight first RPC + // occupied the whole window before producing a retryable error. + return &budgetExhaustedError{lastErr: context.DeadlineExceeded} + } + return &budgetExhaustedError{lastErr: f.lastRetryErr} +} + +// finish publishes the retry loop's one terminal result. Deadline expiry only +// cancels ctx; the loop must observe that cancellation and return before done +// is closed, after which joiners can safely read result. +func (f *resumeFlight) finish(result *resumeCallResult) { + f.mu.Lock() + f.finished = true + f.result = result + f.timer.Stop() + close(f.done) + f.mu.Unlock() + f.cancel() +} + // ActorResumer coordinates safe, deduplicated resumption of actors. type ActorResumer struct { apiClient ateapipb.ControlClient - flight singleflight.Group + + mu sync.Mutex + flights map[string]*resumeFlight // parkEnabled makes transient worker-pool saturation (FailedPrecondition) // retryable, so a request is parked and retried until budget rather than // failing immediately. parkEnabled bool - // budget bounds the total time a single resume operation retries before the - // underlying error is returned. + // budget is each caller's maximum parking time. The shared flight tracks the + // latest caller deadline, but callers stop waiting on their own timers. budget time.Duration - // backoff paces the retries within the budget. + // backoff paces the shared retry loop. 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 shared-flight outcome classification. nextID uint64 } @@ -112,8 +219,9 @@ type resumerOption func(*ActorResumer) // withParking configures parking behavior from cfg. When parking is enabled, // FailedPrecondition ("no free workers available") becomes retryable and the -// resume is retried, at cfg's retry cadence, for up to cfg's budget. When -// disabled, the resumer applies fail-fast-on-capacity behavior. +// shared resume is retried at cfg's cadence, while each caller waits for at +// most cfg's budget. When disabled, the resumer applies fail-fast-on-capacity +// behavior. func withParking(cfg ParkedRequestConfig) resumerOption { cfg = cfg.Normalized() return func(r *ActorResumer) { @@ -128,6 +236,7 @@ func withParking(cfg ParkedRequestConfig) resumerOption { func NewActorResumer(apiClient ateapipb.ControlClient, opts ...resumerOption) *ActorResumer { r := &ActorResumer{ apiClient: apiClient, + flights: make(map[string]*resumeFlight), budget: failFastResumeBudget, backoff: resumeBackoff(DefaultParkedRequestRetryInterval, DefaultParkedRequestRetryFactor, DefaultParkedRequestRetryJitter), @@ -138,6 +247,93 @@ func NewActorResumer(apiClient ateapipb.ControlClient, opts ...resumerOption) *A return r } +func (r *ActorResumer) joinOrStartFlight(actorRef resources.ActorRef, reqID uint64, deadline time.Time) *resumeFlight { + key := actorRef.String() + r.mu.Lock() + if flight := r.flights[key]; flight != nil && flight.join(deadline) { + r.mu.Unlock() + return flight + } + + flight := newResumeFlight(deadline) + r.flights[key] = flight + r.mu.Unlock() + + go func() { + flight.finish(r.runResumeFlight(flight, actorRef, reqID)) + + r.mu.Lock() + if r.flights[key] == flight { + delete(r.flights, key) + } + r.mu.Unlock() + }() + return flight +} + +func (r *ActorResumer) runResumeFlight(flight *resumeFlight, actorRef resources.ActorRef, leaderID uint64) *resumeCallResult { + backoff := r.backoff + +retry: + for { + resumeResp, err := r.apiClient.ResumeActor(flight.ctx, &ateapipb.ResumeActorRequest{ + Actor: actorRef.ToObjectRef(), + }) + if err == nil { + return &resumeCallResult{ + actor: resumeResp.GetActor(), + resumed: resumeResp.GetResumed(), + leaderID: leaderID, + } + } + + if flight.ctx.Err() != nil { + return &resumeCallResult{leaderID: leaderID, err: flight.budgetError()} + } + if !r.retryable(err) { + return &resumeCallResult{leaderID: leaderID, err: err} + } + flight.setLastRetryErr(err) + + delay := backoff.Step() + retryAt := time.Now().Add(delay) + timer := time.NewTimer(delay) + for { + select { + case <-flight.ctx.Done(): + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + return &resumeCallResult{leaderID: leaderID, err: flight.budgetError()} + case <-flight.joined: + // If accumulated backoff puts the next attempt more than one initial + // interval away, bring it forward and restart backoff growth. Leave an + // already-sooner attempt and its future growth unchanged. The channel's + // single slot coalesces simultaneous joins. + freshBackoff := r.backoff + freshDelay := freshBackoff.Step() + freshRetryAt := time.Now().Add(freshDelay) + if freshRetryAt.Before(retryAt) { + backoff = freshBackoff + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + timer.Reset(freshDelay) + retryAt = freshRetryAt + } + case <-timer.C: + continue retry + } + } + } +} + // retryable reports whether err warrants another resume attempt while the // request remains parked. A concurrent-resume conflict (Aborted) is always // retried. Transient pool saturation (FailedPrecondition, "no free workers @@ -167,80 +363,21 @@ func (r *ActorResumer) ResumeActor(ctx context.Context, actorRef resources.Actor defer span.End() 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() - - backoff := r.backoff - - var resumeResp *ateapipb.ResumeActorResponse - var lastRetryErr error - - err := wait.ExponentialBackoffWithContext(bgCtx, backoff, func(ctx context.Context) (bool, error) { - var err error - resumeResp, err = r.apiClient.ResumeActor(ctx, &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 - }) - - if err != nil { - // If the budget elapsed while we were still retrying a transient error, - // 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. - // - // Gate on bgCtx itself, not on errors.Is(err, context.DeadlineExceeded): - // when the deadline lands during an in-flight ResumeActor RPC, gRPC - // surfaces a *status* error with code DeadlineExceeded that does not - // match the context sentinel, which would misreport budget exhaustion - // as a 504. bgCtx is this loop's only deadline source, so checking it - // covers both landing spots (mid-RPC and between retries). - 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 - } - - return &resumeCallResult{ - actor: resumeResp.GetActor(), - resumed: resumeResp.GetResumed(), - leaderID: reqID, - }, nil - }) + callerDeadline := time.Now().Add(r.budget) + budgetTimer := time.NewTimer(time.Until(callerDeadline)) + defer budgetTimer.Stop() + flight := r.joinOrStartFlight(actorRef, reqID, callerDeadline) 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"). The detached flight keeps + // running so another caller can still share it. 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 <-budgetTimer.C: + return nil, ResumeOutcomeNone, flight.budgetError() + case <-flight.done: + callRes := flight.result // On error, return ResumeOutcomeNone ("none") so the failure is tagged // under the 'outcome' label rather than misreported as an activation. @@ -248,7 +385,7 @@ func (r *ActorResumer) ResumeActor(ctx context.Context, actorRef resources.Actor return nil, ResumeOutcomeNone, callRes.err } - // Disambiguate singleflight resume outcome: + // Disambiguate shared 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). diff --git a/cmd/atenet/internal/router/ingress/resumer_test.go b/cmd/atenet/internal/router/ingress/resumer_test.go index 225e592577..ebe49bbf23 100644 --- a/cmd/atenet/internal/router/ingress/resumer_test.go +++ b/cmd/atenet/internal/router/ingress/resumer_test.go @@ -18,6 +18,7 @@ import ( "context" "errors" "sync" + "sync/atomic" "testing" "testing/synctest" "time" @@ -154,7 +155,7 @@ func TestActorResumer_ResumeActor(t *testing.T) { } }) - t.Run("SingleflightDeduplication_Disambiguation", func(t *testing.T) { + t.Run("SharedFlightDeduplication_Disambiguation", func(t *testing.T) { var resumeCalled int var mu sync.Mutex @@ -219,7 +220,7 @@ func TestActorResumer_ResumeActor(t *testing.T) { mu.Lock() defer mu.Unlock() if resumeCalled != 1 { - t.Errorf("expected ResumeActor called exactly once by singleflight, got %d", resumeCalled) + t.Errorf("expected ResumeActor called exactly once by the shared flight, got %d", resumeCalled) } }) } @@ -436,11 +437,121 @@ func TestActorResumer_Parking(t *testing.T) { // contract from both sides: a caller that disconnects while parked gets // context.Canceled (classified as the `canceled` outcome) WITHOUT aborting the // shared in-flight resume, which keeps running and serves a later caller from -// the same single RPC. +// the same RPC. func TestActorResumer_CallerCancelDoesNotAbortFlight(t *testing.T) { synctest.Test(t, testCallerCancelDoesNotAbortFlight) } +type asyncResumeResult struct { + actor *ateapipb.Actor + outcome ResumeOutcome + err error +} + +func resumeAsync(r *ActorResumer, actorRef resources.ActorRef) <-chan asyncResumeResult { + result := make(chan asyncResumeResult, 1) + go func() { + actor, outcome, err := r.ResumeActor(context.Background(), actorRef) + result <- asyncResumeResult{actor: actor, outcome: outcome, err: err} + }() + return result +} + +func TestActorResumer_LateJoinerOutlivesLeaderOnSameFlight(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-late-shared"} + start := time.Now() + mock := &resumerMockClient{ + resumeFn: func(context.Context, *ateapipb.ResumeActorRequest, ...grpc.CallOption) (*ateapipb.ResumeActorResponse, error) { + if time.Since(start) >= 1700*time.Millisecond { + return &ateapipb.ResumeActorResponse{ + Actor: &ateapipb.Actor{Status: ateapipb.Actor_STATUS_RUNNING}, + Resumed: true, + }, nil + } + return nil, status.Error(codes.FailedPrecondition, "no free workers available") + }, + } + resumer := NewActorResumer(mock, withParking(ParkedRequestConfig{ + Max: 2, + Budget: time.Second, + RetryInterval: 100 * time.Millisecond, + RetryFactor: 1, + })) + + firstResult := resumeAsync(resumer, actorRef) + time.Sleep(750 * time.Millisecond) + secondResult := resumeAsync(resumer, actorRef) + + first := <-firstResult + var exhausted *budgetExhaustedError + if !errors.As(first.err, &exhausted) { + t.Fatalf("first caller error = %v, want budgetExhaustedError", first.err) + } + + second := <-secondResult + if second.err != nil { + t.Fatalf("late caller failed: %v", second.err) + } + if second.actor.GetStatus() != ateapipb.Actor_STATUS_RUNNING { + t.Errorf("late caller actor status = %v, want RUNNING", second.actor.GetStatus()) + } + if second.outcome != ResumeOutcomeJoined { + t.Errorf("late caller outcome = %q, want %q", second.outcome, ResumeOutcomeJoined) + } + + synctest.Wait() + resumer.mu.Lock() + defer resumer.mu.Unlock() + if _, exists := resumer.flights[actorRef.String()]; exists { + t.Fatal("successful flight remained registered") + } + }) +} + +func TestActorResumer_LateJoinerShortensAccumulatedBackoff(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-late-backoff"} + start := time.Now() + var calls atomic.Int32 + mock := &resumerMockClient{ + resumeFn: func(context.Context, *ateapipb.ResumeActorRequest, ...grpc.CallOption) (*ateapipb.ResumeActorResponse, error) { + calls.Add(1) + if time.Since(start) >= 800*time.Millisecond { + return &ateapipb.ResumeActorResponse{ + Actor: &ateapipb.Actor{Status: ateapipb.Actor_STATUS_RUNNING}, + Resumed: true, + }, nil + } + return nil, status.Error(codes.FailedPrecondition, "no free workers available") + }, + } + resumer := NewActorResumer(mock, withParking(ParkedRequestConfig{ + Max: 2, + Budget: time.Second, + RetryInterval: 100 * time.Millisecond, + RetryFactor: 100, + })) + + firstResult := resumeAsync(resumer, actorRef) + // The second attempt at 100ms leaves the shared loop in a 10s backoff. + // A caller arriving at 750ms brings the next attempt forward to 850ms. + time.Sleep(750 * time.Millisecond) + secondResult := resumeAsync(resumer, actorRef) + + first, second := <-firstResult, <-secondResult + if first.err != nil || second.err != nil { + t.Fatalf("resume results: first=%v, second=%v", first.err, second.err) + } + if first.outcome != ResumeOutcomeTriggered || second.outcome != ResumeOutcomeJoined { + t.Errorf("outcomes = (%q, %q), want (%q, %q)", first.outcome, second.outcome, ResumeOutcomeTriggered, ResumeOutcomeJoined) + } + if calls.Load() != 3 { + t.Errorf("ResumeActor calls = %d, want 2 before the join and 1 after it", calls.Load()) + } + }) +} + func testCallerCancelDoesNotAbortFlight(t *testing.T) { const ( testActorName = "actor-cancel" diff --git a/docs/request-parking.md b/docs/request-parking.md index 8378c7f80b..77d2608861 100644 --- a/docs/request-parking.md +++ b/docs/request-parking.md @@ -39,9 +39,10 @@ exponential backoff until either - the resume succeeds (the actor is `RUNNING` and has a worker IP) — the request is then routed normally; or -- the **park budget** (`--parked-request-budget`, default `5s`) elapses — the - underlying capacity error is returned, surfacing as `503 "actor - unavailable: no free workers available"`. +- the **park budget** (`--parked-request-budget`, default `5s`) elapses. If a + retryable error was observed, that underlying error is preserved (for + example, capacity remains a `503`); if the first RPC occupied the whole + budget, the request ends with `504`. To bound resource use and provide backpressure, the router admits requests to a **parking lot** of fixed capacity (`--parked-request-max`, default `1024`). Each @@ -62,18 +63,25 @@ 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 -control-plane RPC. - -**The park budget is per-flight, not per-request.** The budget clock starts -when a flight's first caller begins the resume; every later request for the -same actor joins that flight and shares its remaining budget and outcome. A -request that joins late may therefore see `budget_exhausted` after waiting far -less than a full budget itself — the accepted cost of collapsing a hot actor's -requests into one control-plane call. (`parking.wait.duration` records each -request's *own* parked time, so sub-budget `budget_exhausted` samples are -expected under sustained saturation.) +shared flight, so a hot actor consumes N parking slots without starting an +independent sequence of `ResumeActor` RPCs for every request. When an expired +flight is replaced, cancellation of its last RPC may briefly overlap the new +flight; ateapi's per-Actor lock makes this handoff safe and retryable. + +**The park budget is per-request.** Each caller receives the full configured +budget from the time it joins. A late caller extends the shared flight's +execution deadline, while every caller still stops waiting on its own timer. +The join also resets accumulated exponential growth and ensures the next retry +is no farther away than one initial retry interval; simultaneous joins coalesce +into one adjustment rather than issuing one RPC each. Thus de-duplication does +not make a newly arrived request inherit either an almost-expired wait budget +or a long backoff accumulated before it arrived. + +A shared flight can remain alive while requests for that Actor keep arriving: +each arrival moves its execution deadline to cover that caller. Once arrivals +stop, it expires no later than one configured budget after the last join. A +caller cancellation does not immediately abort the detached flight, so another +request arriving within that bounded window can still share its work. ### What is *not* parked @@ -111,15 +119,16 @@ 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-budget` | `5s` | Park budget for each request; requests for one actor share the control-plane retry loop, not the remaining wait budget. | | `--parked-request-max` | `1024` | Max concurrent parked/in-flight resume requests; excess 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. | | `--extproc-max-requests` | `0` (auto) | Envoy circuit-breaker `max_requests` for the ext_proc cluster. `0` derives twice `--parked-request-max` (min `1024`); explicit values must be `>= --parked-request-max` (enforced at startup). The excess is fast-path headroom (see Behavior). | -The retry backoff deliberately has no cap and no attempt limit: the budget alone -bounds the wait. +The retry backoff deliberately has no cap and no attempt limit. Each caller's +budget bounds its own wait; later arrivals may extend the lifetime of the shared +flight. ## Observability @@ -135,7 +144,7 @@ bounds the wait. | `outcome` | When it is set | | ------------------ | --------------------------------------------------------------------------- | | `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. | + | `budget_exhausted` | The caller's park budget elapsed. A saved retryable cause is preserved (for example, pool saturation remains a 503); if the first RPC consumed the whole budget, it maps to 504 instead. | | `canceled` | The client disconnected while parked (request context canceled). | | `timeout` | The request's own deadline expired while parked (distinct from the park budget). | | `error` | The resume failed with a non-retryable error (`NotFound`, `PermissionDenied`, ...). |