rename ErrPersistenceRetry to ErrVersionConflict to reflect the actual behavior

cmd/ateapi/internal/controlapi/delete_actor.go

@@ -61,7 +61,7 @@ func (s *Service) DeleteActor(ctx context.Context, req *ateapipb.DeleteActorRequ
 			}
 			return nil, status.Errorf(codes.FailedPrecondition, "Actor %s is not suspended", req.GetActor().GetName())
 		}
-		if errors.Is(err, store.ErrPersistenceRetry) {
+		if errors.Is(err, store.ErrVersionConflict) {
 			return nil, status.Error(codes.Aborted, "concurrent update conflict, please retry")
 		}
 		return nil, fmt.Errorf("while deleting actor from DB: %w", err)

cmd/ateapi/internal/controlapi/pause_actor.go

@@ -34,7 +34,7 @@ func (s *Service) PauseActor(ctx context.Context, req *ateapipb.PauseActorReques
 
 	actor, err := s.actorWorkflow.PauseActor(ctx, req.GetActor().GetAtespace(), req.GetActor().GetName())
 	if err != nil {
-		if errors.Is(err, store.ErrPersistenceRetry) {
+		if errors.Is(err, store.ErrVersionConflict) {
 			return nil, status.Error(codes.Aborted, "concurrent update conflict, please retry")
 		}
 		if errors.Is(err, store.ErrNotFound) {

cmd/ateapi/internal/controlapi/resume_actor.go

@@ -34,7 +34,7 @@ func (s *Service) ResumeActor(ctx context.Context, req *ateapipb.ResumeActorRequ
 
 	actor, err := s.actorWorkflow.ResumeActor(ctx, req.GetActor().GetAtespace(), req.GetActor().GetName(), req.GetBoot())
 	if err != nil {
-		if errors.Is(err, store.ErrPersistenceRetry) {
+		if errors.Is(err, store.ErrVersionConflict) {
 			return nil, status.Error(codes.Aborted, "concurrent update conflict, please retry")
 		}
 		if errors.Is(err, store.ErrNotFound) {

cmd/ateapi/internal/controlapi/suspend_actor.go

@@ -34,7 +34,7 @@ func (s *Service) SuspendActor(ctx context.Context, req *ateapipb.SuspendActorRe
 
 	actor, err := s.actorWorkflow.SuspendActor(ctx, req.GetActor().GetAtespace(), req.GetActor().GetName())
 	if err != nil {
-		if errors.Is(err, store.ErrPersistenceRetry) {
+		if errors.Is(err, store.ErrVersionConflict) {
 			return nil, status.Error(codes.Aborted, "concurrent update conflict, please retry")
 		}
 		if errors.Is(err, store.ErrNotFound) {

cmd/ateapi/internal/controlapi/syncer.go

@@ -229,7 +229,7 @@ func (s *WorkerPoolSyncer) releaseActorOnDeadWorker(ctx context.Context, namespa
 	actor.AteomPodIp = ""
 	actor.InProgressSnapshot = ""
 	actor.WorkerPoolName = ""
-	if _, err := s.persistence.UpdateActor(ctx, actor, actor.GetMetadata().GetVersion()); err != nil && !errors.Is(err, store.ErrPersistenceRetry) {
+	if _, err := s.persistence.UpdateActor(ctx, actor, actor.GetMetadata().GetVersion()); err != nil && !errors.Is(err, store.ErrVersionConflict) {
 		return err
 	}
 	return nil

cmd/ateapi/internal/controlapi/update_actor.go

@@ -43,7 +43,7 @@ func (s *Service) UpdateActor(ctx context.Context, req *ateapipb.UpdateActorRequ
 
 	updated, err := s.persistence.UpdateActor(ctx, actor, actor.GetMetadata().GetVersion())
 	if err != nil {
-		if errors.Is(err, store.ErrPersistenceRetry) {
+		if errors.Is(err, store.ErrVersionConflict) {
 			return nil, status.Error(codes.Aborted, "concurrent update conflict, please retry")
 		}
 		return nil, fmt.Errorf("while updating actor: %w", err)

cmd/ateapi/internal/controlapi/workflow.go

@@ -56,7 +56,7 @@ type WorkflowStep[Params any, Context any] interface {
 	Execute(ctx context.Context, params Params, wCtx Context) error
 
 	// RetryBackoff returns an optional backoff configuration for this step.
-	// If non-nil, the workflow orchestrator automatically retries Execute() on persistence conflicts.
+	// If non-nil, the workflow orchestrator automatically retries Execute() on version conflicts.
 	RetryBackoff() *wait.Backoff
 }
 
@@ -120,7 +120,7 @@ func runStep[Params any, Context any](ctx context.Context, params Params, wCtx C
 		if execErr == nil {
 			return true, nil
 		}
-		if errors.Is(execErr, store.ErrPersistenceRetry) {
+		if errors.Is(execErr, store.ErrVersionConflict) {
 			return false, nil // retryable
 		}
 		return false, execErr // fatal

cmd/ateapi/internal/controlapi/workflow_resume.go

@@ -222,7 +222,7 @@ 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 {
-		if !errors.Is(err, store.ErrPersistenceRetry) {
+		if !errors.Is(err, store.ErrVersionConflict) {
 			return err
 		}
 		// refresh the version of actor to avoid always failure in rest retries.

cmd/ateapi/internal/controlapi/workflow_resume_test.go

@@ -310,14 +310,14 @@ func seedAssignFixture(t *testing.T, ctx context.Context, persistence store.Inte
 
 // TestAssignWorkerStep_ConflictRefreshesActor verifies the actor write's
 // conflict handling within a single Execute: a concurrent spec write leaves
-// ErrPersistenceRetry with state.Actor refreshed.
+// ErrVersionConflict with state.Actor refreshed.
 func TestAssignWorkerStep_ConflictRefreshesActor(t *testing.T) {
 	t.Parallel()
 	tests := []struct {
 		name string
 		// mutate is the racing concurrent write applied to the fresh actor.
 		mutate func(fresh *ateapipb.Actor)
-		// wantRetry means Execute surfaces ErrPersistenceRetry with
+		// wantRetry means Execute surfaces ErrVersionConflict with
 		// state.Actor refreshed to the injected write; otherwise Aborted.
 		wantRetry bool
 		// wantStoredStatus is the persisted status after Execute.
@@ -371,8 +371,8 @@ func TestAssignWorkerStep_ConflictRefreshesActor(t *testing.T) {
 			err := step.Execute(ctx, &ResumeInput{ActorName: "id1", Atespace: "team-a"}, state)
 
 			if tc.wantRetry {
-				if !errors.Is(err, store.ErrPersistenceRetry) {
-					t.Fatalf("Execute: %v, want ErrPersistenceRetry", err)
+				if !errors.Is(err, store.ErrVersionConflict) {
+					t.Fatalf("Execute: %v, want ErrVersionConflict", err)
 				}
 				if got := state.Actor.GetMetadata().GetVersion(); got != injected.GetMetadata().GetVersion() {
 					t.Errorf("state.Actor version = %d, want %d (refreshed for the retry)", got, injected.GetMetadata().GetVersion())

cmd/ateapi/internal/controlapi/workflow_test.go

@@ -128,7 +128,7 @@ func TestRunWorkflow_CheckPrerequisiteOrdering(t *testing.T) {
 }
 
 // TestRunWorkflow_RetryOnPersistenceConflict verifies runStep's retry loop:
-// with a RetryBackoff, Execute is retried on store.ErrPersistenceRetry without
+// with a RetryBackoff, Execute is retried on store.ErrVersionConflict without
 // re-running IsComplete or CheckPrerequisite; any other error — or a nil
 // backoff — fails the step on the first attempt.
 func TestRunWorkflow_RetryOnPersistenceConflict(t *testing.T) {
@@ -142,7 +142,7 @@ func TestRunWorkflow_RetryOnPersistenceConflict(t *testing.T) {
 		steps := []WorkflowStep[struct{}, struct{}]{
 			&stubStep{
 				name:        "s1",
-				executeErrs: []error{store.ErrPersistenceRetry, store.ErrPersistenceRetry},
+				executeErrs: []error{store.ErrVersionConflict, store.ErrVersionConflict},
 				backoff:     backoff,
 				calls:       &calls,
 			},
@@ -181,7 +181,7 @@ func TestRunWorkflow_RetryOnPersistenceConflict(t *testing.T) {
 	t.Run("persistent conflict exhausts backoff", func(t *testing.T) {
 		var calls []string
 		steps := []WorkflowStep[struct{}, struct{}]{
-			&stubStep{name: "s1", executeErr: store.ErrPersistenceRetry, backoff: backoff, calls: &calls},
+			&stubStep{name: "s1", executeErr: store.ErrVersionConflict, backoff: backoff, calls: &calls},
 		}
 		if err := RunWorkflow(ctx, struct{}{}, struct{}{}, steps); err == nil {
 			t.Fatal("RunWorkflow: expected error after exhausting retries, got nil")
@@ -194,11 +194,11 @@ func TestRunWorkflow_RetryOnPersistenceConflict(t *testing.T) {
 	t.Run("nil backoff means no retry", func(t *testing.T) {
 		var calls []string
 		steps := []WorkflowStep[struct{}, struct{}]{
-			&stubStep{name: "s1", executeErrs: []error{store.ErrPersistenceRetry}, calls: &calls},
+			&stubStep{name: "s1", executeErrs: []error{store.ErrVersionConflict}, calls: &calls},
 		}
 		err := RunWorkflow(ctx, struct{}{}, struct{}{}, steps)
-		if !errors.Is(err, store.ErrPersistenceRetry) {
-			t.Fatalf("RunWorkflow error = %v, want wrapping store.ErrPersistenceRetry", err)
+		if !errors.Is(err, store.ErrVersionConflict) {
+			t.Fatalf("RunWorkflow error = %v, want wrapping store.ErrVersionConflict", err)
 		}
 		if got := countCalls(calls, "s1.Execute"); got != 1 {
 			t.Errorf("Execute called %d times, want 1; calls = %v", got, calls)

cmd/ateapi/internal/store/ateredis/ateredis.go

@@ -429,7 +429,7 @@ func (s *Persistence) UpdateWorker(ctx context.Context, worker *ateapipb.Worker,
 		}
 
 		if currentWorker.GetVersion() != expectedVersion {
-			return store.ErrPersistenceRetry
+			return store.ErrVersionConflict
 		}
 		dbWorker.Version = currentWorker.GetVersion() + 1
 		if currentWorker.GetWorkerNamespace() != dbWorker.GetWorkerNamespace() {
@@ -460,8 +460,8 @@ func (s *Persistence) UpdateWorker(ctx context.Context, worker *ateapipb.Worker,
 		if errors.Is(err, store.ErrNotFound) {
 			return store.ErrNotFound
 		}
-		if errors.Is(err, store.ErrPersistenceRetry) || errors.Is(err, redis.TxFailedErr) {
-			return store.ErrPersistenceRetry
+		if errors.Is(err, store.ErrVersionConflict) || errors.Is(err, redis.TxFailedErr) {
+			return store.ErrVersionConflict
 		}
 		return fmt.Errorf("while executing update worker transaction: %w", err)
 	}
@@ -517,7 +517,7 @@ func (s *Persistence) DeleteActor(ctx context.Context, atespace, name string) (*
 
 	if err != nil {
 		if errors.Is(err, redis.TxFailedErr) {
-			return nil, store.ErrPersistenceRetry
+			return nil, store.ErrVersionConflict
 		}
 		return nil, err
 	}
@@ -547,7 +547,7 @@ func (s *Persistence) UpdateActor(ctx context.Context, actor *ateapipb.Actor, ex
 		}
 
 		if currentActor.GetMetadata().GetVersion() != expectedVersion {
-			return store.ErrPersistenceRetry
+			return store.ErrVersionConflict
 		}
 		if currentActor.GetMetadata().GetName() != dbActor.GetMetadata().GetName() {
 			return fmt.Errorf("name is immutable")
@@ -580,8 +580,8 @@ func (s *Persistence) UpdateActor(ctx context.Context, actor *ateapipb.Actor, ex
 		if errors.Is(err, store.ErrNotFound) {
 			return nil, store.ErrNotFound
 		}
-		if errors.Is(err, store.ErrPersistenceRetry) || errors.Is(err, redis.TxFailedErr) {
-			return nil, store.ErrPersistenceRetry
+		if errors.Is(err, store.ErrVersionConflict) || errors.Is(err, redis.TxFailedErr) {
+			return nil, store.ErrVersionConflict
 		}
 		return nil, fmt.Errorf("while executing update actor transaction: %w", err)
 	}

cmd/ateapi/internal/store/ateredis/ateredis_test.go

@@ -229,8 +229,8 @@ func TestUpdateActor_Conflict(t *testing.T) {
 	// Try to update instance 2 (which has stale version)
 	actor2.Status = ateapipb.Actor_STATUS_SUSPENDED
 	_, err = s.UpdateActor(ctx, actor2, actor2.GetMetadata().GetVersion())
-	if !errors.Is(err, store.ErrPersistenceRetry) {
-		t.Errorf("expected ErrPersistenceRetry, got %v", err)
+	if !errors.Is(err, store.ErrVersionConflict) {
+		t.Errorf("expected ErrVersionConflict, got %v", err)
 	}
 }
 
@@ -607,8 +607,8 @@ func TestUpdateWorker_Conflict(t *testing.T) {
 	// Try to update instance 2
 	worker2.Assignment = &ateapipb.Assignment{Actor: &ateapipb.ObjectRef{Name: "session-2"}}
 	err = s.UpdateWorker(ctx, worker2, worker2.Version)
-	if !errors.Is(err, store.ErrPersistenceRetry) {
-		t.Errorf("expected ErrPersistenceRetry, got %v", err)
+	if !errors.Is(err, store.ErrVersionConflict) {
+		t.Errorf("expected ErrVersionConflict, got %v", err)
 	}
 }
 

cmd/ateapi/internal/store/store.go

@@ -30,8 +30,8 @@ var (
 	// ErrAlreadyExists indicates that the object already exists in the DB.
 	ErrAlreadyExists = errors.New("persistence: already exists")
 
-	// ErrPersistenceRetry is the error returned when the persistence layer needs to retry.
-	ErrPersistenceRetry = errors.New("persistence: retry status")
+	// ErrVersionConflict indicates an update's expected version did not match the stored one.
+	ErrVersionConflict = errors.New("persistence: version conflict")
 
 	// ErrFailedPrecondition indicates the object is not in the required state for the operation.
 	ErrFailedPrecondition = errors.New("persistence: failed precondition")
@@ -52,7 +52,7 @@ type Interface interface {
 
 	// Updates actor state with optimistic concurrency check and returns the stored
 	// resource with advanced metadata (version, update_time). The input is not
-	// mutated. Returns ErrNotFound if missing, or ErrPersistenceRetry on version mismatch.
+	// mutated. Returns ErrNotFound if missing, or ErrVersionConflict on version mismatch.
 	UpdateActor(ctx context.Context, actor *ateapipb.Actor, expectedVersion int64) (*ateapipb.Actor, error)
 
 	// Removes an actor and returns the deleted resource. Returns ErrNotFound if
@@ -88,7 +88,7 @@ type Interface interface {
 	// Registers a new idle worker. Returns ErrAlreadyExists if already registered.
 	CreateWorker(ctx context.Context, worker *ateapipb.Worker) error
 
-	// Updates worker state with optimistic concurrency check. Returns ErrNotFound if missing, or ErrPersistenceRetry on version mismatch.
+	// Updates worker state with optimistic concurrency check. Returns ErrNotFound if missing, or ErrVersionConflict on version mismatch.
 	UpdateWorker(ctx context.Context, worker *ateapipb.Worker, expectedVersion int64) error
 
 	// Removes a worker. Idempotent: does nothing if worker is not found.