From e456e964c2e4840459d6297f1a9fa3873d5ff408 Mon Sep 17 00:00:00 2001 From: Eric Curtin Date: Sun, 18 Oct 2026 22:02:44 +0100 Subject: [PATCH 1/2] ateapi: report a full fleet as ResourceExhausted when assignment retries run out The last attempt now decides: a full worker is no_capacity, a version conflict stays Aborted. Signed-off-by: Eric Curtin --- .../internal/controlapi/workflow_resume.go | 34 +++-- .../controlapi/workflow_resume_test.go | 120 +++++++++++++----- docs/metrics/registry/metrics.yaml | 6 +- 3 files changed, 116 insertions(+), 44 deletions(-) diff --git a/cmd/ateapi/internal/controlapi/workflow_resume.go b/cmd/ateapi/internal/controlapi/workflow_resume.go index 8f43df3dfe..80f2b93f32 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume.go @@ -256,6 +256,9 @@ func (w *ActorWorkflow) ensureWorkerAssigned(ctx context.Context, actorRef resou // ctxErr marks the loop ending between attempts, where no attempt ran and // so none recorded. var ctxErr error + // lostErr is the error of the last attempt that lost a race. It decides + // how the loop reports that the retries ran out. + var lostErr error // lastAttempt times only the last attempt that ran, as every other record // on the histogram does. The backoff can sleep once more after the last // attempt before it gives up, so the time since that attempt began is too @@ -275,7 +278,8 @@ func (w *ActorWorkflow) ensureWorkerAssigned(ctx context.Context, actorRef resou assignedActor, assignedWorker = attemptActor, attemptWorker return true, nil } - if errors.Is(attemptErr, store.ErrVersionConflict) || errors.Is(attemptErr, errWorkerFilledUp) { + if lostRace(attemptErr) { + lostErr = attemptErr if attemptActor != nil { actor = attemptActor // retry with the refreshed actor } @@ -300,12 +304,18 @@ func (w *ActorWorkflow) ensureWorkerAssigned(ctx context.Context, actorRef resou return assignedActor, assignedWorker, nil case ctxErr != nil: if attempted { - w.recordExhaustedAssignment(ctx, lastAttempt, actor, ctxErr) + w.recordExhaustedAssignment(ctx, lastAttempt, actor, ateattr.SchedulerOutcomeError, ctxErr) } return nil, nil, err + case exhausted && errors.Is(lostErr, errWorkerFilledUp): + // Every worker the scheduler offered was full: a lack of room, not a + // conflict, so the caller sees ResourceExhausted. + err = apierror.ResourceExhausted("%w", errNoWorkerRoom) + w.recordExhaustedAssignment(ctx, lastAttempt, actor, ateattr.SchedulerOutcomeNoCapacity, err) + return nil, nil, err case exhausted: err = apierror.Aborted("concurrent update conflict, please retry: %w", store.ErrVersionConflict) - w.recordExhaustedAssignment(ctx, lastAttempt, actor, err) + w.recordExhaustedAssignment(ctx, lastAttempt, actor, ateattr.SchedulerOutcomeError, err) return nil, nil, err default: // assignWorkerAttempt recorded this one itself. @@ -326,8 +336,8 @@ var assignmentBackoff = wait.Backoff{ // recordExhaustedAssignment writes the one record of an assignment loop that // ended without assigning. No worker was assigned, so no pool is named. -func (w *ActorWorkflow) recordExhaustedAssignment(ctx context.Context, elapsed time.Duration, actor *ateapi.Actor, err error) { - w.instruments.recordSchedulerAssignment(ctx, elapsed, ateattr.SchedulerOutcomeError, "", "", snapshotSandboxClass(actor), err) +func (w *ActorWorkflow) recordExhaustedAssignment(ctx context.Context, elapsed time.Duration, actor *ateapi.Actor, outcome string, err error) { + w.instruments.recordSchedulerAssignment(ctx, elapsed, outcome, "", "", snapshotSandboxClass(actor), err) } // validateAssignedWorker checks a RESUMING actor's persisted assignment @@ -444,12 +454,16 @@ func (w *ActorWorkflow) workerHoldingStaleClaim(ctx context.Context, actor *atea return nil, nil } -// schedulerRecordable excludes the two errors the assignment loop retries -// transparently. Counting a retried attempt would inflate the error rate and -// double-count the eventual success; ensureWorkerAssigned records a loop that -// never succeeds. +// lostRace reports the two errors the assignment loop retries transparently. +func lostRace(err error) bool { + return errors.Is(err, store.ErrVersionConflict) || errors.Is(err, errWorkerFilledUp) +} + +// schedulerRecordable excludes a retried attempt. Counting it would inflate +// the error rate and double-count the eventual success; ensureWorkerAssigned +// records a loop that never succeeds. func schedulerRecordable(err error) bool { - return !errors.Is(err, store.ErrVersionConflict) && !errors.Is(err, errWorkerFilledUp) + return !lostRace(err) } // assignWorkerAttempt makes one attempt at claiming a worker for the actor diff --git a/cmd/ateapi/internal/controlapi/workflow_resume_test.go b/cmd/ateapi/internal/controlapi/workflow_resume_test.go index eb2e4a8072..d3fa77eb42 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume_test.go @@ -2457,43 +2457,99 @@ func TestEnsureWorkerAssigned_ContentionRecordsOneObservation(t *testing.T) { // TestEnsureWorkerAssigned_ExhaustedBackoffRecordsOneError covers the other // end: every attempt loses the race, the retries run out, and the resume // fails. The histogram carries one observation for the resume, and not one -// for each attempt. error.type is the code RPCService.ResumeActor returns, so -// the scheduler histogram agrees with the router. The returned error carries -// that code too, so ate.actor.lifecycle.operation.duration, which classifies -// the same error, also records Aborted. The record times the last attempt and -// not the backoff, as every other record on the histogram does. +// for each attempt. The last attempt decides what the resume reports. A full +// worker is a lack of room: ResourceExhausted, and the no_capacity outcome with +// no error.type, so the scheduler histogram agrees with the router. A version +// conflict is Aborted, and error.type is that code. The returned error carries +// the same code, so ate.actor.lifecycle.operation.duration, which classifies +// it, agrees too. The record times the last attempt and not the backoff, as +// every other record on the histogram does. func TestEnsureWorkerAssigned_ExhaustedBackoffRecordsOneError(t *testing.T) { - sched := &filledUpScheduler{ - workers: []*ateapi.Worker{contendedWorker("full-pod")}, - full: map[string]bool{"full-pod": true}, + const pod = "full-pod" + tests := []struct { + name string + // goneUntil is the first Schedule call at which the worker is back in + // the store. Before it, the bind fails with a version conflict. A + // negative value keeps the worker out of the store. + goneUntil int + // goneFrom is the first Schedule call at which the worker has left the + // store again. Zero leaves it in. + goneFrom int + wantCode codes.Code + wantOutcome string + }{ + { + name: "full workers", + wantCode: codes.ResourceExhausted, + wantOutcome: ateattr.SchedulerOutcomeNoCapacity, + }, + { + name: "version conflicts", + goneUntil: -1, + wantCode: codes.Aborted, + wantOutcome: ateattr.SchedulerOutcomeError + "/" + codes.Aborted.String(), + }, + { + name: "conflict, then full workers", + goneUntil: 2, + wantCode: codes.ResourceExhausted, + wantOutcome: ateattr.SchedulerOutcomeNoCapacity, + }, + { + name: "full workers, then conflicts", + goneFrom: 3, + wantCode: codes.Aborted, + wantOutcome: ateattr.SchedulerOutcomeError + "/" + codes.Aborted.String(), + }, } - w, actor, reader := newContendedWorkflow(t, sched) + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + sched := &filledUpScheduler{ + workers: []*ateapi.Worker{contendedWorker(pod)}, + full: map[string]bool{pod: true}, + } + w, actor, reader := newContendedWorkflow(t, sched) + workers := w.store.(*contentionStore).workers + worker := sched.workers[0] + name := worker.GetMetadata().GetName() + if tt.goneUntil != 0 { + delete(workers, name) + } + sched.onSchedule = func(calls int) { + if tt.goneUntil > 0 && calls == tt.goneUntil { + workers[name] = worker + } + if tt.goneFrom > 0 && calls == tt.goneFrom { + delete(workers, name) + } + } - _, _, err := w.ensureWorkerAssigned(context.Background(), resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, &ateapi.ActorTemplate{}) - if err == nil { - t.Fatal("ensureWorkerAssigned succeeded, want a failure after the backoff ended") - } - if got := apierror.Code(err); got != codes.Aborted { - t.Errorf("apierror.Code(err) = %v, want %v (err: %v)", got, codes.Aborted, err) - } - if !errors.Is(err, store.ErrVersionConflict) { - t.Errorf("ensureWorkerAssigned error = %v, want one that wraps store.ErrVersionConflict for RPCService.ResumeActor", err) - } + _, _, err := w.ensureWorkerAssigned(context.Background(), resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, &ateapi.ActorTemplate{}) + if err == nil { + t.Fatal("ensureWorkerAssigned succeeded, want a failure after the backoff ended") + } + if got := apierror.Code(err); got != tt.wantCode { + t.Errorf("apierror.Code(err) = %v, want %v (err: %v)", got, tt.wantCode, err) + } + if tt.wantCode == codes.Aborted && !errors.Is(err, store.ErrVersionConflict) { + t.Errorf("ensureWorkerAssigned error = %v, want one that wraps store.ErrVersionConflict for RPCService.ResumeActor", err) + } - got, outcomes := assignmentObservations(t, reader) - if got != 1 { - t.Fatalf("observations on ate.scheduler.assignment.duration = %d %v, want 1 for the whole resume", got, outcomes) - } - want := ateattr.SchedulerOutcomeError + "/" + codes.Aborted.String() - if outcomes[0] != want { - t.Errorf("recorded outcome %q, want %q: error.type must be the code ResumeActor returns", outcomes[0], want) - } + got, outcomes := assignmentObservations(t, reader) + if got != 1 { + t.Fatalf("observations on ate.scheduler.assignment.duration = %d %v, want 1 for the whole resume", got, outcomes) + } + if outcomes[0] != tt.wantOutcome { + t.Errorf("recorded outcome %q, want %q", outcomes[0], tt.wantOutcome) + } - // One attempt against the stubs takes microseconds, so a record that - // includes the sleep after the last attempt goes above half of it. - _, sleep := replayBackoff(t, assignmentBackoff) - if recorded := assignmentSum(t, reader); recorded >= sleep.Seconds()/2 { - t.Errorf("recorded duration = %.3fs, want less than %.3fs: the record must time the last attempt, not the backoff", recorded, sleep.Seconds()/2) + // One attempt against the stubs takes microseconds, so a record that + // includes the sleep after the last attempt goes above half of it. + _, sleep := replayBackoff(t, assignmentBackoff) + if recorded := assignmentSum(t, reader); recorded >= sleep.Seconds()/2 { + t.Errorf("recorded duration = %.3fs, want less than %.3fs: the record must time the last attempt, not the backoff", recorded, sleep.Seconds()/2) + } + }) } } diff --git a/docs/metrics/registry/metrics.yaml b/docs/metrics/registry/metrics.yaml index 21faa5bf9f..b30d46a9df 100644 --- a/docs/metrics/registry/metrics.yaml +++ b/docs/metrics/registry/metrics.yaml @@ -937,8 +937,10 @@ groups: brief: The time of one attempt of the scheduler to assign a worker during a resume. note: > One resume gives one record. ateapi retries an attempt that loses a race - for a worker, and records only the last attempt. If the retries run out, - the outcome is error, and error.type is the code that the caller gets. + for a worker, and records only the last attempt. If the retries run out + on a full worker, the outcome is no_capacity. If they run out on a version + conflict, the outcome is error, and error.type is the code that the caller + gets. The buckets are smaller than the buckets of the lifecycle histogram. This step is a read of the cache and some writes to the store. It is not a restore of some seconds. But the buckets go to 5 s. Thus you can see a From 94475019b789208334c94dbdb7e029db4a182457 Mon Sep 17 00:00:00 2001 From: Eric Curtin Date: Sat, 31 Oct 2026 22:11:12 +0000 Subject: [PATCH 2/2] controlapi: contentionStore returns ErrWorkerNotFound for a missing worker BindActorToWorker reports a missing Actor as ErrNotFound now. Signed-off-by: Eric Curtin --- cmd/ateapi/internal/controlapi/workflow_resume_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cmd/ateapi/internal/controlapi/workflow_resume_test.go b/cmd/ateapi/internal/controlapi/workflow_resume_test.go index 2d448c39fd..db636ea5f7 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume_test.go @@ -2181,7 +2181,7 @@ type contentionStore struct { func (s *contentionStore) BindActorToWorker(_ context.Context, workerName string, _ *ateapi.ActorAssignment, admit func(*ateapi.Worker) error, _ resources.ActorRef, _ store.Precondition, mutate func(*ateapi.Actor) error) (*ateapi.Actor, error) { worker, ok := s.workers[workerName] if !ok { - return nil, store.ErrNotFound + return nil, store.ErrWorkerNotFound } if admit != nil { if err := admit(worker); err != nil {