Suspend and pause preserve concurrent actor crashes and report failure

cmd/ateapi/internal/controlapi/workflow_pause.go

@@ -229,8 +229,8 @@ func (w *ActorWorkflow) ensureAteletPaused(ctx context.Context, actorRef resourc
 // can never be resumed from local state, so it commits CRASHED instead and
 // fails the step with FailedPrecondition. It re-reads the actor first so an
 // out-of-band transition (e.g. the syncer crashing the actor after its worker
-// died) is not overwritten: with no assignment left there is nothing to
-// finalize.
+// died) is not overwritten: an actor that is no longer SUSPENDING is left as
+// it is and the step fails with FailedPrecondition.
 func (w *ActorWorkflow) ensurePausedFinalized(ctx context.Context, actorRef resources.ActorRef) (_ *ateapipb.Actor, err error) {
 	ctx, done := stepSpan(ctx, "FinalizePaused")
 	defer func() { err = done(err) }()
@@ -239,6 +239,9 @@ func (w *ActorWorkflow) ensurePausedFinalized(ctx context.Context, actorRef reso
 	if err != nil {
 		return nil, err
 	}
+	if state := latestActor.GetStatus().GetState(); state != ateapipb.ActorState_ACTOR_STATE_SUSPENDING {
+		return nil, apierror.FailedPrecondition("FinalizePaused prerequisite not met for Actor: %s (got: %v, want %s)", actorRef, state, ateapipb.ActorState_ACTOR_STATE_SUSPENDING)
+	}
 
 	// 1. Free the worker (if it hasn't been freed yet)
 	if assignment := latestActor.GetStatus().GetWorkerAssignment(); assignment != nil {

cmd/ateapi/internal/controlapi/workflow_pause_test.go

@@ -122,6 +122,7 @@ func TestPauseActor_Finalized_AlreadyCrashed(t *testing.T) {
 	updateDuringCheckpoint(t, atelet, st, actorRef, func(s *ateapipb.ActorStatus) {
 		s.State = ateapipb.ActorState_ACTOR_STATE_CRASHED
 		s.Crash = originalCrash
+		s.WorkerAssignment = nil
 	})
 
 	if _, err := w.PauseActor(ctx, actorRef); apierror.Code(err) != codes.FailedPrecondition {

cmd/ateapi/internal/controlapi/workflow_suspend.go

@@ -385,8 +385,9 @@ func (w *ActorWorkflow) ensureVolumesDetached(ctx context.Context, actor *ateapi
 // When the latest snapshot has no durable snapshot to record, it commits
 // CRASHED instead and fails the step.
 // It re-reads the actor first so an out-of-band transition (e.g. the syncer
-// crashing the actor after its worker died) is not overwritten: with no
-// assignment left there is nothing to finalize.
+// crashing the actor after its worker died) is not overwritten: an actor that
+// is no longer SUSPENDING is left as it is and the step fails with
+// FailedPrecondition.
 func (w *ActorWorkflow) ensureSuspendedFinalized(ctx context.Context, actorRef resources.ActorRef) (_ *ateapipb.Actor, err error) {
 	ctx, done := stepSpan(ctx, "FinalizeSuspended")
 	defer func() { err = done(err) }()
@@ -413,6 +414,9 @@ func (w *ActorWorkflow) ensureSuspendedFinalized(ctx context.Context, actorRef r
 	if err != nil {
 		return nil, err
 	}
+	if state := latestActor.GetStatus().GetState(); state != ateapipb.ActorState_ACTOR_STATE_SUSPENDING {
+		return nil, apierror.FailedPrecondition("FinalizeSuspended prerequisite not met for Actor: %s (got: %v, want %s)", actorRef, state, ateapipb.ActorState_ACTOR_STATE_SUSPENDING)
+	}
 	if snapshotAtLatestGeneration(latestActor.GetStatus()).GetDurableSnapshot() == nil {
 		return nil, crashActorAndFail(ctx, w.store, actorRef, ateattr.OperationSuspend, crashMessageDurableSnapshotMissing)
 	}

cmd/ateapi/internal/controlapi/workflow_suspend_test.go

@@ -1315,18 +1315,10 @@ func TestSuspendAndPause_CharacteristicMatrix(t *testing.T) {
 	}
 }
 
-// TestSuspendAndPause_FinalizeWithoutAssignment pins where the two finalize
-// steps differ when the DeleteWorker workflow crashes the actor, clearing its
-// worker assignment, between the atelet call and finalize. Both sides pin
-// bugs tracked in #2676: the suspend commits SUSPENDED over the crash, and
-// the pause keeps the crash but returns OK for an actor left CRASHED. The fix
-// changes the sync wantState to CRASHED and both codes to FailedPrecondition,
-// replacing the err != nil fatal below.
+// TestSuspendAndPause_FinalizeWithoutAssignment verifies that both workflows
+// preserve an out-of-band crash and return FailedPrecondition when the crash
+// clears the worker assignment between the atelet call and finalize's read.
 func TestSuspendAndPause_FinalizeWithoutAssignment(t *testing.T) {
-	wantState := map[string]ateapipb.ActorState{
-		"sync":  ateapipb.ActorState_ACTOR_STATE_SUSPENDED,
-		"async": ateapipb.ActorState_ACTOR_STATE_CRASHED,
-	}
 	for _, mode := range suspendModes {
 		t.Run(mode.name, func(t *testing.T) {
 			ctx := context.Background()
@@ -1343,21 +1335,29 @@ func TestSuspendAndPause_FinalizeWithoutAssignment(t *testing.T) {
 					WorkerAssignment: wireTestAssignment(),
 				},
 			})
+			// What a worker delete or the syncer commits: CRASHED with the
+			// crash recorded and the assignment cleared, landing after the
+			// atelet call and before finalize re-reads the actor.
+			originalCrash := newActorCrash(ateattr.OperationDelete, "original crash reason")
 			updateDuringCheckpoint(t, atelet, persistence, actorRef, func(s *ateapipb.ActorStatus) {
 				s.State = ateapipb.ActorState_ACTOR_STATE_CRASHED
+				s.Crash = originalCrash
 				s.WorkerAssignment = nil
 			})
 
 			returned, err := mode.run(w, ctx, actorRef)
-			// #2676: both modes should return FailedPrecondition here.
-			if err != nil {
-				t.Fatalf("%s: %v", mode.name, err)
+			if got := apierror.Code(err); got != codes.FailedPrecondition {
+				t.Fatalf("%s: code = %v (err %v), want FailedPrecondition", mode.name, got, err)
+			}
+			if returned != nil {
+				t.Errorf("%s: returned actor = %v, want nil", mode.name, returned)
 			}
-			if got := returned.GetStatus().GetState(); got != wantState[mode.name] {
-				t.Errorf("returned state = %v, want %v", got, wantState[mode.name])
+			stored := mustReadActor(t, ctx, persistence, actorRef)
+			if got := stored.GetStatus().GetState(); got != ateapipb.ActorState_ACTOR_STATE_CRASHED {
+				t.Errorf("%s: stored state = %v, want CRASHED", mode.name, got)
 			}
-			if got := mustReadActor(t, ctx, persistence, actorRef).GetStatus().GetState(); got != wantState[mode.name] {
-				t.Errorf("stored state = %v, want %v", got, wantState[mode.name])
+			if got, want := stored.GetStatus().GetCrash().GetMessage(), originalCrash.GetMessage(); got != want {
+				t.Errorf("%s: crash message = %q, want %q kept", mode.name, got, want)
 			}
 		})
 	}

internal/e2e/suites/demo/termination_test.go

@@ -22,6 +22,8 @@ import (
 	"github.com/agent-substrate/substrate/internal/e2e"
 	"github.com/agent-substrate/substrate/internal/resources"
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -142,7 +144,8 @@ func waitForWorkerRemoved(ctx context.Context, t *testing.T, clients *e2e.Client
 
 // TestGracefulWorkerTerminationSuspend exercises the case where a worker pod is
 // deleted (evicted), and while the container is in its SIGTERM shutdown phase,
-// we initiate a suspend. Suspend should succeed.
+// we initiate a suspend. Suspend either completes before worker deletion crashes
+// the actor, or reports that concurrent crash without overwriting it.
 func TestGracefulWorkerTerminationSuspend(t *testing.T) {
 	t.Parallel()
 	nsObj := e2e.CreateNamespace(t)
@@ -195,6 +198,10 @@ func TestGracefulWorkerTerminationSuspend(t *testing.T) {
 		t.Fatalf("running actor has no bound worker pod: ns=%q name=%q", podNS, podName)
 	}
 	t.Logf("Actor %q bound to worker pod %s/%s", actorID, podNS, podName)
+	previousSnapshotURI := durableSnapshotURI(running.GetStatus())
+	if previousSnapshotURI == "" {
+		t.Fatal("running actor has no completed durable snapshot")
+	}
 
 	// Evict the worker pod. The kubelet sends SIGTERM to worker, which propagates
 	// it into the sandbox; the container hangs for 30s.
@@ -207,19 +214,29 @@ func TestGracefulWorkerTerminationSuspend(t *testing.T) {
 
 	// Suspend the actor.
 	t.Logf("Suspending actor %q during termination", actorID)
-	if _, err := clients.SubstrateAPI.SuspendActor(ctx, &ateapipb.SuspendActorRequest{
+	_, suspendErr := clients.SubstrateAPI.SuspendActor(ctx, &ateapipb.SuspendActorRequest{
 		Actor: &ateapipb.ObjectRef{Atespace: demoAtespace, Name: actorID},
-	}); err != nil {
-		t.Fatalf("failed to suspend Actor: %v", err)
+	})
+	wantState := ateapipb.ActorState_ACTOR_STATE_SUSPENDED
+	switch status.Code(suspendErr) {
+	case codes.OK:
+	case codes.FailedPrecondition, codes.Aborted:
+		// Worker deletion can crash the actor before finalize reads it, or
+		// after that read but before its version-guarded update. Both must
+		// preserve the crash and report failure instead of returning success.
+		wantState = ateapipb.ActorState_ACTOR_STATE_CRASHED
+		t.Logf("Suspend reported concurrent worker deletion: %v", suspendErr)
+	default:
+		t.Fatalf("failed to suspend Actor: %v", suspendErr)
 	}
 
 	// The worker record must eventually be removed once the pod is gone.
 	if err := waitForWorkerRemoved(ctx, t, clients, podName, 60*time.Second); err != nil {
 		t.Fatalf("worker %s not removed after pod deletion: %v", podName, err)
 	}
 
-	// Verify the actor lands in ACTOR_STATE_SUSPENDED
-	waitForActorState(ctx, t, clients, actorID, ateapipb.ActorState_ACTOR_STATE_SUSPENDED)
+	// The RPC result and persisted state must agree after the worker is gone.
+	waitForActorState(ctx, t, clients, actorID, wantState)
 
 	// Verify the pod assignment was cleared.
 	actor, err := clients.SubstrateAPI.GetActor(ctx, &ateapipb.GetActorRequest{
@@ -231,4 +248,15 @@ func TestGracefulWorkerTerminationSuspend(t *testing.T) {
 	if pod := actor.GetStatus().GetWorkerAssignment().GetWorkerPod(); pod != "" {
 		t.Errorf("actor still bound to worker pod %q, expected empty", pod)
 	}
+	if got := durableSnapshotURI(actor.GetStatus()); got == "" {
+		t.Error("actor lost its completed durable snapshot")
+	} else if wantState == ateapipb.ActorState_ACTOR_STATE_CRASHED && got != previousSnapshotURI {
+		t.Errorf("crashed actor snapshot URI = %q, want previous snapshot %q", got, previousSnapshotURI)
+	}
+	if wantState == ateapipb.ActorState_ACTOR_STATE_CRASHED && actor.GetStatus().GetCrash().GetMessage() == "" {
+		t.Error("crashed actor lost its crash reason")
+	}
+	if wantState == ateapipb.ActorState_ACTOR_STATE_SUSPENDED && actor.GetStatus().GetCrash() != nil {
+		t.Errorf("successful suspend retained a crash record: %v", actor.GetStatus().GetCrash())
+	}
 }