From ce08ffe0784bb0b29d5577d79fc0fe6a83b9d63e Mon Sep 17 00:00:00 2001 From: cloudonly Date: Thu, 15 Oct 2026 16:48:07 +0800 Subject: [PATCH 1/2] Finalize leaves an actor that is no longer SUSPENDING alone ensureSuspendedFinalized and ensurePausedFinalized re-read the actor so that an out-of-band transition is not overwritten, but only the version precondition guarded that, and it covers a crash landing after the read, not before it. A worker delete or the syncer crashes an actor without taking the actor lease, and clears its worker assignment doing so. When that landed between the atelet call and finalize's read, suspend committed SUSPENDED over the crash and left the stale crash on the record, and pause found no assignment to free and reported success for a crashed actor. Both steps now require SUSPENDING and otherwise return FailedPrecondition without touching the record, as the mark steps do, so each workflow ends CRASHED and the caller is told. Cover it through the full workflows: flip the test that pinned the old outcome, and give the already-crashed pause test the shape a real crash leaves (assignment cleared) instead of one that only reached FailedPrecondition through the node-unknown branch. --- .../internal/controlapi/workflow_pause.go | 7 ++-- .../controlapi/workflow_pause_test.go | 1 + .../internal/controlapi/workflow_suspend.go | 8 +++-- .../controlapi/workflow_suspend_test.go | 36 +++++++++---------- 4 files changed, 30 insertions(+), 22 deletions(-) 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) } }) } From 61ba1786c8adadd27e3875751a16ddd876891d6e Mon Sep 17 00:00:00 2001 From: cloudonly Date: Fri, 16 Oct 2026 06:33:14 +0800 Subject: [PATCH 2/2] e2e: preserve crash outcomes during worker termination Worker deletion can crash the actor before suspend finalization. Check that the RPC result matches the persisted state and that a reported crash retains its completed snapshot and reason. --- internal/e2e/suites/demo/termination_test.go | 40 +++++++++++++++++--- 1 file changed, 34 insertions(+), 6 deletions(-) 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()) + } }