From 775a5dfcae478147767285ef8ee8307da9b8a7d0 Mon Sep 17 00:00:00 2001 From: Anish Ramasekar Date: Mon, 17 Aug 2026 09:31:21 -0700 Subject: [PATCH] fix(ateapi): keep worker reclaimable when crashActor's release fails (#771) Fixes #608 `crashActor` collected the `releaseWorker` error but still marked the actor `CRASHED` and cleared its `WorkerAssignment`, stranding the still-assigned worker with no actor left to reclaim it. On a release failure, return instead so the actor stays intact and the workflow's retry reclaims the worker (`releaseWorker` is idempotent). - [x] Tests pass - [ ] Appropriate changes to documentation are included in the PR --- cmd/ateapi/internal/controlapi/crash.go | 16 ++++-- cmd/ateapi/internal/controlapi/crash_test.go | 59 ++++++++++++++++++++ 2 files changed, 70 insertions(+), 5 deletions(-) diff --git a/cmd/ateapi/internal/controlapi/crash.go b/cmd/ateapi/internal/controlapi/crash.go index 350f69762..4ec2ec36d 100644 --- a/cmd/ateapi/internal/controlapi/crash.go +++ b/cmd/ateapi/internal/controlapi/crash.go @@ -65,10 +65,17 @@ func crashActor(ctx context.Context, st crashActorStore, actorRef resources.Acto reason = ateattr.ReasonUnknown } - var errCollected []error + // Release the worker before moving the actor to the terminal CRASHED state. + // If the release fails we must not clear the actor's worker assignment or + // mark it CRASHED: doing so would strand the still-assigned worker with no + // actor referencing it, so nothing would ever retry the release and the + // worker slot would be consumed until its pod dies. Returning the error + // instead leaves the actor (and its assignment) intact so the caller retries + // crashActor, which re-attempts the release. releaseWorker is idempotent, so + // a retry after a release that already succeeded is a no-op. sandboxClass, err := releaseWorker(ctx, st, actor) if err != nil { - errCollected = append(errCollected, err) + return fmt.Errorf("while releasing worker to crash actor: %w", err) } // Snapshot crash attributes before pod and pool pointers are cleared below; @@ -85,8 +92,7 @@ func crashActor(ctx context.Context, st crashActorStore, actorRef resources.Acto return nil })) if err != nil { - errCollected = append(errCollected, fmt.Errorf("while marking actor crashed: %w", err)) - return errors.Join(errCollected...) + return fmt.Errorf("while marking actor crashed: %w", err) } // Increment metric only after a successful UpdateActor, and only if the actor was not already crashed. @@ -94,7 +100,7 @@ func crashActor(ctx context.Context, st crashActorStore, actorRef resources.Acto recordActorCrash(ctx, crashAttrs) } - return errors.Join(errCollected...) + return nil } // crashActorStore encapsulates the subset of store operations needed to crash diff --git a/cmd/ateapi/internal/controlapi/crash_test.go b/cmd/ateapi/internal/controlapi/crash_test.go index 620981e87..4b763b770 100644 --- a/cmd/ateapi/internal/controlapi/crash_test.go +++ b/cmd/ateapi/internal/controlapi/crash_test.go @@ -491,3 +491,62 @@ func assertCrashMetricDatapoint(t *testing.T, reader *sdkmetric.ManualReader, wa t.Errorf("did not find ate.actor.crashes metric with attrs: opName=%q, reason=%q, tmplNS=%q, tmplName=%q, workerPool=%q, sandboxClass=%q", wantOpName, wantReason, wantTmplNS, wantTmplName, wantWorkerPool, wantSandboxClass) } + +// failingUpdateWorkerStore wraps a store and fails every UpdateWorker call, +// simulating a transient state-store error while releasing a worker. +type failingUpdateWorkerStore struct { + store.Interface + err error +} + +func (f failingUpdateWorkerStore) UpdateWorker(context.Context, *ateapipb.Worker, int64) error { + return f.err +} + +// A transient failure releasing the worker must not move the actor to the +// terminal CRASHED state: doing so would strand the still-assigned worker with +// no actor left to drive a retry, permanently consuming the worker slot. +// crashActor must return the error with the actor and worker left intact so the +// caller retries and the worker is reclaimed. +func TestCrashActorReleaseFailureLeavesWorkerReclaimable(t *testing.T) { + ctx := context.Background() + actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"} + + st, cleanup := storetest.SetupTestStore(t) + defer cleanup() + seedActor(t, ctx, st, actorRef) + seedWorker(t, ctx, st, actorRef) + + releaseErr := errors.New("state store unavailable") + err := crashActor(ctx, failingUpdateWorkerStore{Interface: st, err: releaseErr}, actorRef, ateattr.OperationUnknown, ateattr.ReasonUnknown) + + if err == nil { + t.Fatal("crashActor() = nil, want error") + } + if !errors.Is(err, releaseErr) { + t.Errorf("crashActor() error = %v, want it to wrap %v", err, releaseErr) + } + + // The actor must stay RUNNING with its worker assignment intact, so a retry + // can re-release the worker. + got, gerr := st.GetActor(ctx, actorRef) + if gerr != nil { + t.Fatalf("GetActor() = %v, want nil", gerr) + } + if got.GetStatus() != ateapipb.Actor_STATUS_RUNNING { + t.Errorf("status = %v, want %v (actor must not be crashed when the release fails)", got.GetStatus(), ateapipb.Actor_STATUS_RUNNING) + } + if got.GetWorkerAssignment() == nil { + t.Error("WorkerAssignment cleared, want preserved so the release can be retried") + } + + // The worker must still be assigned to the actor (the failed release did not + // persist): it is not leaked, and a retry will reclaim it. + worker, werr := st.GetWorker(ctx, "ns", "pool", "pod") + if werr != nil { + t.Fatalf("GetWorker() = %v, want nil", werr) + } + if worker.GetAssignment() == nil { + t.Error("worker assignment = nil, want still assigned (release failed, must remain retriable)") + } +}