diff --git a/cmd/ateapi/internal/controlapi/workflow_resume.go b/cmd/ateapi/internal/controlapi/workflow_resume.go index 3720cd59b..d9b9a03bc 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume.go @@ -244,7 +244,22 @@ func (s *AssignWorkerStep) Execute(ctx context.Context, input *ResumeInput, stat updatedActor, err := s.store.UpdateActor(ctx, state.Actor, state.Actor.GetMetadata().GetVersion()) if err != nil { - return err + if !errors.Is(err, store.ErrPersistenceRetry) { + return err + } + // refresh the version of actor to avoid always failure in rest retries. + fresh, gerr := s.store.GetActor(ctx, input.Atespace, input.ActorName) + if gerr != nil { + slog.WarnContext(ctx, "Failed to refresh actor after assignment conflict", slog.Any("err", gerr)) + return err + } + switch fresh.GetStatus() { + case ateapipb.Actor_STATUS_SUSPENDED, ateapipb.Actor_STATUS_PAUSED: + state.Actor = fresh + return err + default: + return status.Errorf(codes.Aborted, "actor %s is %s and can no longer be resumed", input.ActorName, fresh.GetStatus()) + } } state.Actor = updatedActor state.Worker = assignedWorker diff --git a/cmd/ateapi/internal/controlapi/workflow_resume_test.go b/cmd/ateapi/internal/controlapi/workflow_resume_test.go index 2593a4e20..d65a78aec 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume_test.go @@ -16,9 +16,12 @@ package controlapi import ( "context" + "errors" + "sync" "testing" "time" + "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" "github.com/agent-substrate/substrate/cmd/ateapi/internal/store/storetest" "github.com/agent-substrate/substrate/cmd/ateapi/internal/workercache" atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" @@ -404,6 +407,117 @@ func TestAssignWorkerStep_RetryAfterConflictPicksFreshWorker(t *testing.T) { } } +// conflictInjectingStore wraps a store and runs inject exactly once, +// immediately before the first UpdateActor, simulating a concurrent writer +// racing the step's read-modify-write window. +type conflictInjectingStore struct { + store.Interface + once sync.Once + inject func() +} + +func (c *conflictInjectingStore) UpdateActor(ctx context.Context, actor *ateapipb.Actor, expectedVersion int64) (*ateapipb.Actor, error) { + c.once.Do(c.inject) + return c.Interface.UpdateActor(ctx, actor, expectedVersion) +} + +// seedAssignFixture stores one free gvisor worker and a SUSPENDED actor and +// returns the actor plus a started worker cache. +func seedAssignFixture(t *testing.T, ctx context.Context, persistence store.Interface) (*ateapipb.Actor, *workercache.Cache) { + t.Helper() + if err := persistence.CreateWorker(ctx, &ateapipb.Worker{ + WorkerNamespace: "worker-ns", + WorkerPool: "pool", + WorkerPod: "pod-1", + SandboxClass: "gvisor", + }); err != nil { + t.Fatalf("CreateWorker: %v", err) + } + actor, err := persistence.CreateActor(ctx, &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "id1"}, + Status: ateapipb.Actor_STATUS_SUSPENDED, + }) + if err != nil { + t.Fatalf("CreateActor: %v", err) + } + cacheCtx, cancel := context.WithCancel(ctx) + t.Cleanup(cancel) + wc := workercache.New(persistence, time.Minute) + if err := wc.Start(cacheCtx); err != nil { + t.Fatalf("workercache.Start: %v", err) + } + return actor, wc +} + +// runAssignConflict seeds the fixture, applies mutate as a concurrent write +// racing the step's first UpdateActor, executes the step once, and returns +// the stored actor, the state, the store, and the Execute error. +func runAssignConflict(t *testing.T, mutate func(*ateapipb.Actor)) (*ateapipb.Actor, *ResumeState, store.Interface, error) { + t.Helper() + ctx := context.Background() + persistence := newTestPersistence(t) + actor, wc := seedAssignFixture(t, ctx, persistence) + + st := &conflictInjectingStore{Interface: persistence, inject: func() { + fresh, err := persistence.GetActor(ctx, "team-a", "id1") + if err != nil { + t.Errorf("inject GetActor: %v", err) + return + } + mutate(fresh) + if _, err := persistence.UpdateActor(ctx, fresh, fresh.GetMetadata().GetVersion()); err != nil { + t.Errorf("inject UpdateActor: %v", err) + } + }} + + step := &AssignWorkerStep{store: st, workerCache: wc} + state := &ResumeState{ + Actor: actor, + ActorTemplate: &atev1alpha1.ActorTemplate{ + Spec: atev1alpha1.ActorTemplateSpec{SandboxClass: atev1alpha1.SandboxClassGvisor}, + }, + } + execErr := step.Execute(ctx, &ResumeInput{ActorName: "id1", Atespace: "team-a"}, state) + + stored, err := persistence.GetActor(ctx, "team-a", "id1") + if err != nil { + t.Fatalf("GetActor: %v", err) + } + return stored, state, persistence, execErr +} + +// TestAssignWorkerStep_ConflictRefreshesActor verifies the actor write's +// conflict handling within a single Execute: a concurrent spec write leaves +// ErrPersistenceRetry with state.Actor refreshed, so the engine's backoff +// retry applies on top of the concurrent write instead of replaying the same +// stale version until the backoff is exhausted; an actor that left the +// resumable states aborts the retry instead of being resurrected to RESUMING. +func TestAssignWorkerStep_ConflictRefreshesActor(t *testing.T) { + t.Run("spec write refreshes state.Actor", func(t *testing.T) { + _, state, _, err := runAssignConflict(t, func(a *ateapipb.Actor) { + a.WorkerSelector = &ateapipb.Selector{MatchLabels: map[string]string{"team": "blue"}} + }) + if !errors.Is(err, store.ErrPersistenceRetry) { + t.Fatalf("Execute: %v, want ErrPersistenceRetry", err) + } + if got := state.Actor.GetWorkerSelector().GetMatchLabels()["team"]; got != "blue" { + t.Errorf("state.Actor selector team = %q, want blue (refreshed with the concurrent write)", got) + } + }) + + t.Run("crash aborts instead of resurrecting", func(t *testing.T) { + stored, _, _, err := runAssignConflict(t, func(a *ateapipb.Actor) { + a.Status = ateapipb.Actor_STATUS_CRASHED + }) + if got := status.Code(err); got != codes.Aborted { + t.Fatalf("status.Code(err) = %v, want %v (err: %v)", got, codes.Aborted, err) + } + if stored.GetStatus() != ateapipb.Actor_STATUS_CRASHED { + t.Errorf("status = %v, want CRASHED (retry must not resurrect the actor)", stored.GetStatus()) + } + }) +} + // TestResumeActorWorkflow_RejectedAndIdempotentPaths covers the two // short-circuit paths of the resume workflow: rejection by AssignWorkerStep's // CheckPrerequisite and the IsComplete idempotent fast-forward.