diff --git a/cmd/ateapi/internal/controlapi/workflow_pause.go b/cmd/ateapi/internal/controlapi/workflow_pause.go index 072b721afb..d3bb75f61c 100644 --- a/cmd/ateapi/internal/controlapi/workflow_pause.go +++ b/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 { diff --git a/cmd/ateapi/internal/controlapi/workflow_pause_test.go b/cmd/ateapi/internal/controlapi/workflow_pause_test.go index 76be4a5088..2fb91e13d7 100644 --- a/cmd/ateapi/internal/controlapi/workflow_pause_test.go +++ b/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 { diff --git a/cmd/ateapi/internal/controlapi/workflow_suspend.go b/cmd/ateapi/internal/controlapi/workflow_suspend.go index b2655daa16..f8ae72fa36 100644 --- a/cmd/ateapi/internal/controlapi/workflow_suspend.go +++ b/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) } diff --git a/cmd/ateapi/internal/controlapi/workflow_suspend_test.go b/cmd/ateapi/internal/controlapi/workflow_suspend_test.go index 8bed842237..4f629f4874 100644 --- a/cmd/ateapi/internal/controlapi/workflow_suspend_test.go +++ b/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) } }) } diff --git a/internal/e2e/suites/demo/termination_test.go b/internal/e2e/suites/demo/termination_test.go index 763179cc96..369c258eb0 100644 --- a/internal/e2e/suites/demo/termination_test.go +++ b/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,10 +214,20 @@ 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. @@ -218,8 +235,8 @@ func TestGracefulWorkerTerminationSuspend(t *testing.T) { 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()) + } }