Resume on a full fleet fails with ResourceExhausted, not Aborted

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
@@ -389,12 +399,16 @@ func (w *ActorWorkflow) validateAssignedWorker(ctx context.Context, actorRef res
 	return worker, 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

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 {
@@ -2351,43 +2351,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)
+			}
+		})
 	}
 }
 

docs/metrics/registry/metrics.yaml

@@ -950,8 +950,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