From dcbe067a6f2f3ecb32cc16dcc702c3e452c59381 Mon Sep 17 00:00:00 2001 From: Haven Xia Date: Mon, 27 Jul 2026 18:55:58 -0700 Subject: [PATCH] ateapi: rename ErrPersistenceRetry to ErrVersionConflict The error is only returned on an optimistic-concurrency version mismatch. Rename it for that behavior so other retryable errors don't get folded into it (but should introduce their own error name). --- cmd/ateapi/internal/controlapi/delete_actor.go | 2 +- cmd/ateapi/internal/controlapi/pause_actor.go | 2 +- cmd/ateapi/internal/controlapi/resume_actor.go | 2 +- cmd/ateapi/internal/controlapi/suspend_actor.go | 2 +- cmd/ateapi/internal/controlapi/syncer.go | 2 +- cmd/ateapi/internal/controlapi/update_actor.go | 2 +- cmd/ateapi/internal/controlapi/workflow.go | 4 ++-- cmd/ateapi/internal/controlapi/workflow_resume.go | 2 +- .../internal/controlapi/workflow_resume_test.go | 8 ++++---- cmd/ateapi/internal/controlapi/workflow_test.go | 12 ++++++------ cmd/ateapi/internal/store/ateredis/ateredis.go | 14 +++++++------- .../internal/store/ateredis/ateredis_test.go | 8 ++++---- cmd/ateapi/internal/store/store.go | 8 ++++---- 13 files changed, 34 insertions(+), 34 deletions(-) diff --git a/cmd/ateapi/internal/controlapi/delete_actor.go b/cmd/ateapi/internal/controlapi/delete_actor.go index df8fac6e3..f0ea3a3c2 100644 --- a/cmd/ateapi/internal/controlapi/delete_actor.go +++ b/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) diff --git a/cmd/ateapi/internal/controlapi/pause_actor.go b/cmd/ateapi/internal/controlapi/pause_actor.go index 61418fbb2..1a5f3cffa 100644 --- a/cmd/ateapi/internal/controlapi/pause_actor.go +++ b/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) { diff --git a/cmd/ateapi/internal/controlapi/resume_actor.go b/cmd/ateapi/internal/controlapi/resume_actor.go index c8487c1b2..3bbf7a0d4 100644 --- a/cmd/ateapi/internal/controlapi/resume_actor.go +++ b/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) { diff --git a/cmd/ateapi/internal/controlapi/suspend_actor.go b/cmd/ateapi/internal/controlapi/suspend_actor.go index 60b761fba..ce66177ab 100644 --- a/cmd/ateapi/internal/controlapi/suspend_actor.go +++ b/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) { diff --git a/cmd/ateapi/internal/controlapi/syncer.go b/cmd/ateapi/internal/controlapi/syncer.go index 82854f826..6320523cd 100644 --- a/cmd/ateapi/internal/controlapi/syncer.go +++ b/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 diff --git a/cmd/ateapi/internal/controlapi/update_actor.go b/cmd/ateapi/internal/controlapi/update_actor.go index db3384121..aba4a8fe3 100644 --- a/cmd/ateapi/internal/controlapi/update_actor.go +++ b/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) diff --git a/cmd/ateapi/internal/controlapi/workflow.go b/cmd/ateapi/internal/controlapi/workflow.go index 7d7e189b8..855ec2aa4 100644 --- a/cmd/ateapi/internal/controlapi/workflow.go +++ b/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 diff --git a/cmd/ateapi/internal/controlapi/workflow_resume.go b/cmd/ateapi/internal/controlapi/workflow_resume.go index b1814fb2e..3025ae915 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume.go +++ b/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. diff --git a/cmd/ateapi/internal/controlapi/workflow_resume_test.go b/cmd/ateapi/internal/controlapi/workflow_resume_test.go index 94c12c3ba..d69c5067d 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume_test.go +++ b/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()) diff --git a/cmd/ateapi/internal/controlapi/workflow_test.go b/cmd/ateapi/internal/controlapi/workflow_test.go index afb5db8e2..58056b56f 100644 --- a/cmd/ateapi/internal/controlapi/workflow_test.go +++ b/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) diff --git a/cmd/ateapi/internal/store/ateredis/ateredis.go b/cmd/ateapi/internal/store/ateredis/ateredis.go index 161195ca8..467707e06 100644 --- a/cmd/ateapi/internal/store/ateredis/ateredis.go +++ b/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) } diff --git a/cmd/ateapi/internal/store/ateredis/ateredis_test.go b/cmd/ateapi/internal/store/ateredis/ateredis_test.go index a5a4e7927..fcdbf5834 100644 --- a/cmd/ateapi/internal/store/ateredis/ateredis_test.go +++ b/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) } } diff --git a/cmd/ateapi/internal/store/store.go b/cmd/ateapi/internal/store/store.go index ad7a1513c..3b6e8426c 100644 --- a/cmd/ateapi/internal/store/store.go +++ b/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.