ateapi,atenet: fail fast when resuming a CRASHED or DELETING actor

cmd/ateapi/internal/controlapi/workflow_resume.go

@@ -231,6 +231,25 @@ func (w *ActorWorkflow) ensureVolumesCreated(ctx context.Context, actorRef resou
 	return storedActor, nil
 }
 
+// assignWorkerPrerequisiteNotMetError is the FailedPrecondition for an actor
+// in a state ensureWorkerAssigned cannot start from. SUSPENDING and REVERTING
+// carry an ErrorInfo reason, since a concurrent operation moves the actor out
+// of them and a caller that fails fast on FailedPrecondition may retry.
+func assignWorkerPrerequisiteNotMetError(actorRef resources.ActorRef, actor *ateapi.Actor) error {
+	const format = "AssignWorker prerequisite not met for Actor: %s (got: %v, want %s)"
+	state := actor.GetStatus().GetState()
+	var reason apierror.Reason
+	switch state {
+	case ateapi.ActorState_ACTOR_STATE_SUSPENDING:
+		reason = apierror.ReasonActorSuspending
+	case ateapi.ActorState_ACTOR_STATE_REVERTING:
+		reason = apierror.ReasonActorReverting
+	default:
+		return apierror.FailedPrecondition(format, actorRef, state, ateapi.ActorState_ACTOR_STATE_SUSPENDED)
+	}
+	return apierror.WithReason(apierror.FailedPrecondition(format, actorRef, state, ateapi.ActorState_ACTOR_STATE_SUSPENDED), reason)
+}
+
 // ensureWorkerAssigned assigns a worker and leaves the actor RESUMING.
 // Re-entry validates the recorded assignment before reuse.
 // Version conflicts re-read the actor before retrying.
@@ -248,7 +267,7 @@ func (w *ActorWorkflow) ensureWorkerAssigned(ctx context.Context, actorRef resou
 		return actor, worker, nil
 	case ateapi.ActorState_ACTOR_STATE_SUSPENDED:
 	default:
-		return nil, nil, apierror.FailedPrecondition("AssignWorker prerequisite not met for Actor: %s (got: %v, want %s)", actorRef, actor.GetStatus().GetState(), ateapi.ActorState_ACTOR_STATE_SUSPENDED)
+		return nil, nil, assignWorkerPrerequisiteNotMetError(actorRef, actor)
 	}
 
 	var assignedActor *ateapi.Actor

cmd/ateapi/internal/controlapi/workflow_resume_test.go

@@ -41,6 +41,7 @@ import (
 	"go.ate.dev/substrate/internal/volume"
 	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	"google.golang.org/genproto/googleapis/rpc/errdetails"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials/insecure"
@@ -2233,6 +2234,53 @@ func TestEnsureVolumesCreated(t *testing.T) {
 	}
 }
 
+// ensureWorkerAssigned refuses every state but SUSPENDED and RESUMING with
+// FailedPrecondition, and attaches an ErrorInfo reason only for the states a
+// concurrent operation moves the actor out of, so a caller that fails fast
+// on FailedPrecondition may retry those.
+func TestEnsureWorkerAssigned(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name       string
+		state      ateapi.ActorState
+		wantReason apierror.Reason
+	}{
+		{"suspending actor carries ACTOR_SUSPENDING", ateapi.ActorState_ACTOR_STATE_SUSPENDING, apierror.ReasonActorSuspending},
+		{"reverting actor carries ACTOR_REVERTING", ateapi.ActorState_ACTOR_STATE_REVERTING, apierror.ReasonActorReverting},
+		{"crashed actor carries no reason", ateapi.ActorState_ACTOR_STATE_CRASHED, ""},
+		{"deleting actor carries no reason", ateapi.ActorState_ACTOR_STATE_DELETING, ""},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			actorRef := resources.ActorRef{Atespace: "team-a", Name: "id1"}
+			actor := &ateapi.Actor{
+				Metadata: &ateapi.ResourceMetadata{Atespace: "team-a", Name: "id1"},
+				Status:   &ateapi.ActorStatus{State: tt.state},
+			}
+			// The state switch runs before any store access.
+			w := &ActorWorkflow{}
+			_, _, err := w.ensureWorkerAssigned(context.Background(), actorRef, actor, nil)
+			if code := apierror.Code(err); code != codes.FailedPrecondition {
+				t.Fatalf("ensureWorkerAssigned code = %v, want FailedPrecondition; error = %v", code, err)
+			}
+			if !strings.Contains(err.Error(), "AssignWorker prerequisite not met") {
+				t.Errorf("ensureWorkerAssigned error = %v, want the AssignWorker prerequisite message", err)
+			}
+			st, _ := apierror.FromError(err)
+			var gotReason apierror.Reason
+			for _, d := range st.Details() {
+				if info, ok := d.(*errdetails.ErrorInfo); ok && info.GetDomain() == apierror.ErrorInfoDomain {
+					gotReason = apierror.Reason(info.GetReason())
+				}
+			}
+			if gotReason != tt.wantReason {
+				t.Errorf("ErrorInfo reason = %q, want %q", gotReason, tt.wantReason)
+			}
+		})
+	}
+}
+
 // filledUpScheduler hands out workers in order and refuses room on the ones
 // named in full. It stands in for a stale worker cache: Schedule offers a
 // worker that looked free, and the admit callback that BindActorToWorker runs

cmd/atenet/internal/router/ingress/errors.go

@@ -20,10 +20,12 @@ import (
 	"fmt"
 
 	envoy_type "github.com/envoyproxy/go-control-plane/envoy/type/v3"
+	"google.golang.org/genproto/googleapis/rpc/errdetails"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
 	"go.ate.dev/substrate/cmd/atenet/internal/router/extproc"
+	"go.ate.dev/substrate/internal/apierror"
 	"go.ate.dev/substrate/internal/ateattr"
 	"go.ate.dev/substrate/internal/resources"
 )
@@ -46,6 +48,28 @@ func statusDescription(err error) string {
 	return status.Convert(err).Message()
 }
 
+// isTransientResumeErr returns true if the error names an actor state a
+// concurrent operation moves the actor out of (SUSPENDING, REVERTING), so the
+// request may be parked and retried.
+func isTransientResumeErr(err error) bool {
+	type grpcStatus interface{ GRPCStatus() *status.Status }
+	var gs grpcStatus
+	if !errors.As(err, &gs) {
+		return false
+	}
+	for _, d := range gs.GRPCStatus().Details() {
+		info, ok := d.(*errdetails.ErrorInfo)
+		if !ok || info.GetDomain() != apierror.ErrorInfoDomain {
+			continue
+		}
+		switch apierror.Reason(info.GetReason()) {
+		case apierror.ReasonActorSuspending, apierror.ReasonActorReverting:
+			return true
+		}
+	}
+	return false
+}
+
 // shedOutcome returns the route outcome of a request that is shed while its
 // flight retries on retryErr: no_capacity for a ResourceExhausted retryErr,
 // unavailable otherwise. parkingFullErr and the parking.rejected label both

cmd/atenet/internal/router/ingress/errors_test.go

@@ -17,16 +17,68 @@ package ingress
 import (
 	"context"
 	"errors"
+	"fmt"
 	"testing"
 
 	envoy_type "github.com/envoyproxy/go-control-plane/envoy/type/v3"
+	"google.golang.org/genproto/googleapis/rpc/errdetails"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
 	"go.ate.dev/substrate/cmd/atenet/internal/router/extproc"
+	"go.ate.dev/substrate/internal/apierror"
 	"go.ate.dev/substrate/internal/resources"
 )
 
+// transientPrecondition builds the FailedPrecondition ateapi returns for an
+// actor in a state a concurrent operation moves it out of: desc with an
+// ErrorInfo detail carrying reason in apierror's domain.
+func transientPrecondition(t *testing.T, reason apierror.Reason, desc string) error {
+	t.Helper()
+	st, err := status.New(codes.FailedPrecondition, desc).WithDetails(&errdetails.ErrorInfo{Domain: apierror.ErrorInfoDomain, Reason: string(reason)})
+	if err != nil {
+		t.Fatalf("WithDetails: %v", err)
+	}
+	return st.Err()
+}
+
+// isTransientResumeErr is true only for apierror's two transient reasons in
+// its own domain; any other detail, a status with none, or a non-status error
+// fails fast.
+func TestIsTransientResumeErr(t *testing.T) {
+	t.Parallel()
+
+	withDetail := func(domain, reason string) error {
+		t.Helper()
+		st, err := status.New(codes.FailedPrecondition, "actor is "+reason).WithDetails(&errdetails.ErrorInfo{Domain: domain, Reason: reason})
+		if err != nil {
+			t.Fatalf("WithDetails: %v", err)
+		}
+		return st.Err()
+	}
+	tests := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{"plain FailedPrecondition", status.Error(codes.FailedPrecondition, "actor is ACTOR_STATE_CRASHED"), false},
+		{"ACTOR_SUSPENDING in the ate.dev domain", transientPrecondition(t, apierror.ReasonActorSuspending, "actor is ACTOR_STATE_SUSPENDING"), true},
+		{"ACTOR_REVERTING in the ate.dev domain", transientPrecondition(t, apierror.ReasonActorReverting, "actor is ACTOR_STATE_REVERTING"), true},
+		{"matching reason in another domain", withDetail("example.com", string(apierror.ReasonActorSuspending)), false},
+		{"unknown reason in the ate.dev domain", withDetail(apierror.ErrorInfoDomain, "ACTOR_CRASHED"), false},
+		{"wrapped transient status", fmt.Errorf("while resuming: %w", transientPrecondition(t, apierror.ReasonActorSuspending, "actor is ACTOR_STATE_SUSPENDING")), true},
+		{"non-status error", errors.New("resume failed"), false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			if got := isTransientResumeErr(tt.err); got != tt.want {
+				t.Errorf("isTransientResumeErr(%v) = %t, want %t", tt.err, got, tt.want)
+			}
+		})
+	}
+}
+
 func TestActorNotFoundErr(t *testing.T) {
 	t.Parallel()
 
@@ -84,6 +136,18 @@ func TestMapResumeError(t *testing.T) {
 
 	actorRef := resources.ActorRef{Atespace: "team-a", Name: "ctr6"}
 
+	// otherDomainPrecondition carries an ErrorInfo whose reason matches but
+	// whose domain is not apierror's; the router maps it like any other
+	// FailedPrecondition.
+	otherDomainPrecondition := func(t *testing.T) error {
+		t.Helper()
+		st, err := status.New(codes.FailedPrecondition, "actor is ACTOR_STATE_SUSPENDING").WithDetails(&errdetails.ErrorInfo{Domain: "example.com", Reason: string(apierror.ReasonActorSuspending)})
+		if err != nil {
+			t.Fatalf("WithDetails: %v", err)
+		}
+		return st.Err()
+	}
+
 	tests := []struct {
 		name     string
 		err      error
@@ -102,6 +166,24 @@ func TestMapResumeError(t *testing.T) {
 			wantCode: envoy_type.StatusCode_ServiceUnavailable,
 			wantBody: `actor team-a/ctr6 unavailable: no worker has room for the actor`,
 		},
+		{
+			name:     "FailedPrecondition with the ACTOR_SUSPENDING reason after the budget maps to 503 and preserves desc",
+			err:      &budgetExhaustedError{lastErr: transientPrecondition(t, apierror.ReasonActorSuspending, "actor team-a/ctr6 is ACTOR_STATE_SUSPENDING")},
+			wantCode: envoy_type.StatusCode_ServiceUnavailable,
+			wantBody: `actor team-a/ctr6 unavailable: actor team-a/ctr6 is ACTOR_STATE_SUSPENDING`,
+		},
+		{
+			name:     "FailedPrecondition with the ACTOR_REVERTING reason maps to 503",
+			err:      transientPrecondition(t, apierror.ReasonActorReverting, "actor team-a/ctr6 is ACTOR_STATE_REVERTING"),
+			wantCode: envoy_type.StatusCode_ServiceUnavailable,
+			wantBody: `actor team-a/ctr6 unavailable: actor team-a/ctr6 is ACTOR_STATE_REVERTING`,
+		},
+		{
+			name:     "FailedPrecondition with a reason from another domain maps to 503",
+			err:      otherDomainPrecondition(t),
+			wantCode: envoy_type.StatusCode_ServiceUnavailable,
+			wantBody: `actor team-a/ctr6 unavailable: actor is ACTOR_STATE_SUSPENDING`,
+		},
 		{
 			name:     "Unavailable maps to 503",
 			err:      status.Error(codes.Unavailable, "control-plane down"),

cmd/atenet/internal/router/ingress/ingress_test.go

@@ -35,6 +35,7 @@ import (
 
 	"go.ate.dev/substrate/api/ateapi/v1"
 	"go.ate.dev/substrate/cmd/atenet/internal/router/extproc"
+	"go.ate.dev/substrate/internal/apierror"
 	"go.ate.dev/substrate/internal/ateattr"
 	"go.ate.dev/substrate/internal/atenet"
 	atunnelingress "go.ate.dev/substrate/internal/atunnel/ingress"
@@ -305,6 +306,14 @@ func TestHandleRequestHeaders(t *testing.T) {
 			expectedErrStr: `actor team-a/123e4567-e89b-12d3-a456-426614174000 unavailable: no worker has room for the actor`,
 			expectedStatus: envoy_type.StatusCode_ServiceUnavailable,
 		},
+		{
+			name:           "FailedPrecondition with a transient reason and parking disabled maps to 503 at once",
+			authority:      testUUID + ".team-a.actors.resources.substrate.ate.dev",
+			resumeErr:      transientPrecondition(t, apierror.ReasonActorSuspending, "actor is ACTOR_STATE_SUSPENDING"),
+			expectErr:      true,
+			expectedErrStr: `actor team-a/123e4567-e89b-12d3-a456-426614174000 unavailable: actor is ACTOR_STATE_SUSPENDING`,
+			expectedStatus: envoy_type.StatusCode_ServiceUnavailable,
+		},
 		{
 			name:           "NotFound maps to 404",
 			authority:      testUUID + ".team-a.actors.resources.substrate.ate.dev",
@@ -809,7 +818,7 @@ func TestHandleRequestHeaders_FullLotShedsParkedRequest(t *testing.T) {
 		},
 		{
 			name:        "FailedPrecondition is dropped",
-			resumeErr:   status.Error(codes.FailedPrecondition, "actor is suspending"),
+			resumeErr:   transientPrecondition(t, apierror.ReasonActorSuspending, "actor is suspending"),
 			wantCode:    codes.Unknown,
 			wantOutcome: ateattr.RouterOutcomeUnavailable,
 		},

cmd/atenet/internal/router/ingress/resumer.go

@@ -103,9 +103,8 @@ type ActorResumer struct {
 	// never on the fast path (#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
-	// failing immediately.
+	// parkEnabled lets retryable park a request on a transient condition and
+	// retry it until budget rather than failing immediately.
 	parkEnabled bool
 	// budget bounds the total time a single resume operation retries before the
 	// underlying error is returned.
@@ -164,16 +163,20 @@ func NewActorResumer(apiClient ateapi.ControlClient, opts ...resumerOption) *Act
 // ateapi rolling restart) are retried only when parking is enabled, turning a
 // momentary condition into a bounded wait instead of an immediate failure — a
 // parked request should ride out a blip, not fail on it with budget remaining.
-// FailedPrecondition stays retryable too: it no longer carries saturation, but
-// it does cover states a concurrent operation can move the actor out of.
+// FailedPrecondition is retried only when parking is enabled and the actor is
+// in a state a concurrent operation moves it out of (SUSPENDING, REVERTING);
+// see [isTransientResumeErr]. Every other FailedPrecondition, such as a
+// CRASHED actor, fails fast, as does a transient one while parking is disabled.
 // All other codes (NotFound, DeadlineExceeded, PermissionDenied, ...) are
 // returned to the caller so the HTTP boundary can map them with full fidelity.
 func (r *ActorResumer) retryable(err error) bool {
 	switch status.Code(err) {
 	case codes.Aborted:
 		return true
-	case codes.ResourceExhausted, codes.FailedPrecondition, codes.Unavailable:
+	case codes.ResourceExhausted, codes.Unavailable:
 		return r.parkEnabled
+	case codes.FailedPrecondition:
+		return r.parkEnabled && isTransientResumeErr(err)
 	default:
 		return false
 	}