From 025063eb8307fa2ded18ea144e0934adde311957 Mon Sep 17 00:00:00 2001 From: Tim Hockin Date: Sat, 22 Aug 2026 17:55:45 -0700 Subject: [PATCH] Compartmentalize AtespaceExists This needs to be a property of the storage layer. It is in PG but not in redis. This commit moves it out of the store interface and down into redis. It's still not CORRECT but it constrains it. This requires a lot of tests to create the atespace before creating other resources. These tests are wrong in the face of a correct storage layer, anyway. --- .../actoridentity/actoridentity_test.go | 8 ++ cmd/ateapi/internal/controlapi/actor.go | 9 -- .../internal/controlapi/actor_template.go | 8 -- cmd/ateapi/internal/controlapi/actor_test.go | 17 ++++ cmd/ateapi/internal/controlapi/crash_test.go | 12 +++ cmd/ateapi/internal/controlapi/service.go | 1 - cmd/ateapi/internal/controlapi/syncer_test.go | 23 +++++ .../controlapi/workflow_pause_test.go | 16 ++++ .../controlapi/workflow_resume_test.go | 9 ++ .../controlapi/workflow_suspend_test.go | 36 +++++++ .../controlapi/workflow_testutil_test.go | 8 ++ cmd/ateapi/internal/store/atepg/atepg.go | 8 -- .../internal/store/ateredis/ateredis.go | 22 ++++- .../internal/store/ateredis/ateredis_test.go | 96 +++++++++++++++++-- cmd/ateapi/internal/store/store.go | 3 - .../internal/store/storecontract/contract.go | 15 --- 16 files changed, 237 insertions(+), 54 deletions(-) diff --git a/cmd/ateapi/internal/actoridentity/actoridentity_test.go b/cmd/ateapi/internal/actoridentity/actoridentity_test.go index ef973d7e46..171f969da7 100644 --- a/cmd/ateapi/internal/actoridentity/actoridentity_test.go +++ b/cmd/ateapi/internal/actoridentity/actoridentity_test.go @@ -247,6 +247,14 @@ func seedActor(t *testing.T, ctx context.Context, st store.Interface, f actorFix t.Helper() actorRef := resources.ActorRef{Atespace: testAtespace, Name: testActorName} + + atespace := &ateapipb.Atespace{ + Metadata: &ateapipb.ResourceMetadata{Name: actorRef.Atespace}, + } + if _, err := st.CreateAtespace(ctx, atespace); err != nil { + t.Fatalf("seed atespace: %v", err) + } + actor := &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, Status: &ateapipb.ActorStatus{State: f.state}, diff --git a/cmd/ateapi/internal/controlapi/actor.go b/cmd/ateapi/internal/controlapi/actor.go index f89bf52960..7ee51767aa 100644 --- a/cmd/ateapi/internal/controlapi/actor.go +++ b/cmd/ateapi/internal/controlapi/actor.go @@ -72,15 +72,6 @@ func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequ atespace := in.GetMetadata().GetAtespace() name := in.GetMetadata().GetName() - // The atespace must already exist. - exists, err := s.persistence.AtespaceExists(ctx, atespace) - if err != nil { - return nil, fmt.Errorf("while checking atespace: %w", err) - } - if !exists { - return nil, status.Errorf(codes.FailedPrecondition, "Atespace %s not found", atespace) - } - // Volume creation is completed asynchronously after the actor is recorded. initVols, err := initialActorVolumes(ctx, s.storageClassLister, template) if err != nil { diff --git a/cmd/ateapi/internal/controlapi/actor_template.go b/cmd/ateapi/internal/controlapi/actor_template.go index 2f4f122bbb..ba0668ef1a 100644 --- a/cmd/ateapi/internal/controlapi/actor_template.go +++ b/cmd/ateapi/internal/controlapi/actor_template.go @@ -36,14 +36,6 @@ func (s *Service) CreateActorTemplate(ctx context.Context, req *ateapipb.CreateA in := req.GetActorTemplate() templateRef := resources.ActorTemplateRefFromActorTemplate(in) - exists, err := s.persistence.AtespaceExists(ctx, templateRef.Atespace) - if err != nil { - return nil, fmt.Errorf("while checking atespace: %w", err) - } - if !exists { - return nil, status.Errorf(codes.FailedPrecondition, "Atespace %s not found", templateRef.Atespace) - } - // Rebuild the template from the client-owned fields only: status is // server-owned and ignored per the CreateActorTemplateRequest contract. template := &ateapipb.ActorTemplate{ diff --git a/cmd/ateapi/internal/controlapi/actor_test.go b/cmd/ateapi/internal/controlapi/actor_test.go index 173836b4e5..d4dac3d93a 100644 --- a/cmd/ateapi/internal/controlapi/actor_test.go +++ b/cmd/ateapi/internal/controlapi/actor_test.go @@ -391,6 +391,11 @@ func TestUpdateActor_DeleteRecreateRace(t *testing.T) { actorRef := resources.ActorRef{Atespace: testAtespace, Name: testActorID} + atespace := &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: actorRef.Atespace}} + if _, err := persistence.CreateAtespace(ctx, atespace); err != nil { + t.Fatalf("seed CreateAtespace: %v", err) + } + // Actor A: what the client reads, and what its uid precondition names. // Freshly created, so it sits at version 1. original, err := persistence.CreateActor(ctx, &ateapipb.Actor{ @@ -478,6 +483,10 @@ func TestUpdateActor_ConcurrentDisjointUpdates(t *testing.T) { actorRef := resources.ActorRef{Atespace: testAtespace, Name: testActorID} + atespace := &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: actorRef.Atespace}} + if _, err := persistence.CreateAtespace(ctx, atespace); err != nil { + t.Fatalf("seed CreateAtespace: %v", err) + } original, err := persistence.CreateActor(ctx, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: testActorID}, ActorTemplateNamespace: "ns1", @@ -568,6 +577,14 @@ func serviceWithActor(t *testing.T, actor *ateapipb.Actor) (*Service, *ateapipb. persistence, cleanup := storetest.SetupTestStore(t) t.Cleanup(cleanup) + atespace := &ateapipb.Atespace{ + Metadata: &ateapipb.ResourceMetadata{Name: actor.Metadata.Atespace}, + } + _, err := persistence.CreateAtespace(context.Background(), atespace) + if err != nil { + t.Fatalf("Failed to CreateAtespace: %v", err) + } + created, err := persistence.CreateActor(context.Background(), actor) if err != nil { t.Fatalf("Failed to CreateActor: %v", err) diff --git a/cmd/ateapi/internal/controlapi/crash_test.go b/cmd/ateapi/internal/controlapi/crash_test.go index dd9a8aba3f..f983338ace 100644 --- a/cmd/ateapi/internal/controlapi/crash_test.go +++ b/cmd/ateapi/internal/controlapi/crash_test.go @@ -36,6 +36,12 @@ import ( // tests can assert they are cleared when the actor crashes. func seedActor(t *testing.T, ctx context.Context, st store.Interface, actorRef resources.ActorRef) { t.Helper() + + atespace := &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: actorRef.Atespace}} + if _, err := st.CreateAtespace(ctx, atespace); err != nil { + t.Fatalf("Failed to CreateAtespace: %v", err) + } + if _, err := st.CreateActor(ctx, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Name: actorRef.Name, Atespace: actorRef.Atespace}, Status: &ateapipb.ActorStatus{ @@ -90,6 +96,9 @@ func seedWorker(t *testing.T, ctx context.Context, st store.Interface, actorRef // already cleared, e.g. by a prior release. func seedUnboundActor(t *testing.T, ctx context.Context, st store.Interface, actorRef resources.ActorRef) { t.Helper() + if _, err := st.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: actorRef.Atespace}}); err != nil { + t.Fatalf("CreateAtespace: %v", err) + } if _, err := st.CreateActor(ctx, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Name: actorRef.Name, Atespace: actorRef.Atespace}, Status: &ateapipb.ActorStatus{ @@ -458,6 +467,9 @@ func TestCrashActor_Metrics(t *testing.T) { }, }, } + if _, err := st.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: actor.Metadata.Atespace}}); err != nil { + t.Fatalf("CreateAtespace: %v", err) + } if _, err := st.CreateActor(ctx, actor); err != nil { t.Fatalf("CreateActor: %v", err) } diff --git a/cmd/ateapi/internal/controlapi/service.go b/cmd/ateapi/internal/controlapi/service.go index c50ccb755f..65d2b93564 100644 --- a/cmd/ateapi/internal/controlapi/service.go +++ b/cmd/ateapi/internal/controlapi/service.go @@ -93,7 +93,6 @@ type serviceStore interface { GetActorSnapshotTag(ctx context.Context, atespace, name string) (*ateapipb.ActorSnapshotTag, error) UpdateActorSnapshotTag(ctx context.Context, atespace, name string, precondition store.Precondition, mutate func(toUpdate *ateapipb.ActorSnapshotTag) error) (*ateapipb.ActorSnapshotTag, error) DeleteActorSnapshotTag(ctx context.Context, atespace, name string) (*ateapipb.ActorSnapshotTag, error) - AtespaceExists(ctx context.Context, name string) (bool, error) CreateAtespace(ctx context.Context, atespace *ateapipb.Atespace) (*ateapipb.Atespace, error) GetAtespace(ctx context.Context, name string) (*ateapipb.Atespace, error) ListAtespaces(ctx context.Context, opts store.ListOptions) (store.ListResponse[*ateapipb.Atespace], error) diff --git a/cmd/ateapi/internal/controlapi/syncer_test.go b/cmd/ateapi/internal/controlapi/syncer_test.go index 6d3f79c301..33c066592d 100644 --- a/cmd/ateapi/internal/controlapi/syncer_test.go +++ b/cmd/ateapi/internal/controlapi/syncer_test.go @@ -262,6 +262,9 @@ func TestSyncer_DeleteBoundWorker_ClearsActor(t *testing.T) { t.Fatalf("worker row not materialised: %v", err) } actorName := "actor-orphan" + if _, err := persistence.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: "team-orphan"}}); err != nil { + t.Fatalf("create atespace: %v", err) + } createdActor, err := persistence.CreateActor(ctx, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Name: actorName, Atespace: "team-orphan"}, ActorTemplateNamespace: ns, ActorTemplateName: "tmpl", Status: &ateapipb.ActorStatus{ @@ -582,6 +585,11 @@ func TestReconcileDeadWorker(t *testing.T) { ns, pool, pod := "ns-rdw", "pool1", "worker-rdw" atespace, actorID := "team-rdw", "actor-rdw" + + if _, err := persistence.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: atespace}}); err != nil { + t.Fatalf("CreateAtespace() failed: %v", err) + } + createdActor, err := persistence.CreateActor(ctx, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Name: actorID, Atespace: atespace}, ActorTemplateNamespace: ns, ActorTemplateName: "tmpl", Status: &ateapipb.ActorStatus{ @@ -637,6 +645,11 @@ func TestReconcileDeadWorker_IgnoresStaleIncarnationAssignment(t *testing.T) { ns, pool, pod, uid := "ns-rdw", "pool1", "worker-rdw", "uid-rdw" atespace, actorID := "team-rdw", "actor-rdw" + + if _, err := persistence.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: atespace}}); err != nil { + t.Fatalf("CreateAtespace() failed: %v", err) + } + createdActor, err := persistence.CreateActor(ctx, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Name: actorID, Atespace: atespace}, ActorTemplateNamespace: ns, ActorTemplateName: "tmpl", Status: &ateapipb.ActorStatus{ @@ -729,6 +742,11 @@ func TestSyncer_ReconcileOrphanedWorkers(t *testing.T) { // An orphan worker (no pod) whose actor is still RUNNING must be cleaned up. const orphanUID = "22222222-2222-2222-2222-222222222222" atespace, actorID := "team-recon", "actor-recon" + + if _, err := persistence.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: atespace}}); err != nil { + t.Fatalf("CreateAtespace() failed: %v", err) + } + createdActor, err := persistence.CreateActor(ctx, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Name: actorID, Atespace: atespace}, ActorTemplateNamespace: ns, ActorTemplateName: "tmpl", Status: &ateapipb.ActorStatus{ @@ -957,6 +975,11 @@ func TestReleaseActorOnDeadWorker_StateTransitions(t *testing.T) { s := &WorkerPoolSyncer{persistence: persistence} atespace, actorID := "team-status", "actor-status" + + if _, err := persistence.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: atespace}}); err != nil { + t.Fatalf("CreateAtespace() failed: %v", err) + } + createdActor, err := persistence.CreateActor(ctx, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Name: actorID, Atespace: atespace}, ActorTemplateNamespace: ns, diff --git a/cmd/ateapi/internal/controlapi/workflow_pause_test.go b/cmd/ateapi/internal/controlapi/workflow_pause_test.go index 67cf8387c9..f8386a40f3 100644 --- a/cmd/ateapi/internal/controlapi/workflow_pause_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_pause_test.go @@ -56,6 +56,9 @@ func TestEnsurePausedFinalized_WorkerGone(t *testing.T) { InProgressLocalSnapshotName: "local-snap-1", }, } + if _, err := st.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: actorRef.Atespace}}); err != nil { + t.Fatalf("CreateAtespace() failed: %v", err) + } if _, err := st.CreateActor(ctx, actor); err != nil { t.Fatalf("CreateActor: %v", err) } @@ -111,6 +114,10 @@ func TestEnsurePausedFinalized_RecordsContentScope(t *testing.T) { ctx := context.Background() actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"} + if _, err := st.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: actorRef.Atespace}}); err != nil { + t.Fatalf("CreateAtespace() failed: %v", err) + } + created, err := st.CreateActor(ctx, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, Status: &ateapipb.ActorStatus{ @@ -239,6 +246,11 @@ func TestEnsureMarkedPausing_StateMatrix(t *testing.T) { w := &ActorWorkflow{store: persistence} actorRef := resources.ActorRef{Atespace: "team-a", Name: "id1"} + + if _, err := persistence.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: actorRef.Atespace}}); err != nil { + t.Fatalf("CreateAtespace() failed: %v", err) + } + actor, err := persistence.CreateActor(ctx, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, Status: &ateapipb.ActorStatus{State: seedState}, @@ -275,6 +287,10 @@ func TestEnsureAteletPaused_DanglingWorkerDoesNotRecordPhantomSnapshot(t *testin ctx := context.Background() persistence := newTestPersistence(t) + if _, err := persistence.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: "team-a"}}); err != nil { + t.Fatalf("CreateAtespace() failed: %v", err) + } + actor := &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "actor-1"}, Status: &ateapipb.ActorStatus{ diff --git a/cmd/ateapi/internal/controlapi/workflow_resume_test.go b/cmd/ateapi/internal/controlapi/workflow_resume_test.go index 5b729e78bb..f13952838b 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume_test.go @@ -126,6 +126,9 @@ func TestAssignWorkerAttempt_ReleasesIneligibleStaleWorkerInBackground(t *testin ctx := context.Background() persistence := newTestPersistence(t) + if _, err := persistence.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: "team-a"}}); err != nil { + t.Fatalf("CreateAtespace: %v", err) + } actor, err := persistence.CreateActor(ctx, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "id1"}, Status: &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_SUSPENDED}, @@ -260,6 +263,9 @@ func TestAssignWorkerAttempt_RetryAfterConflictPicksFreshWorker(t *testing.T) { t.Fatalf("UpdateWorker (concurrent claim): %v", err) } + if _, err := persistence.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: "team-a"}}); err != nil { + t.Fatalf("CreateAtespace: %v", err) + } actor, err := persistence.CreateActor(ctx, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "id1"}, Status: &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_SUSPENDED}, @@ -350,6 +356,9 @@ func seedAssignFixture(t *testing.T, ctx context.Context, persistence store.Inte }); err != nil { t.Fatalf("CreateWorker: %v", err) } + if _, err := persistence.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: "team-a"}}); err != nil { + t.Fatalf("CreateAtespace: %v", err) + } actor, err := persistence.CreateActor(ctx, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "id1"}, Status: &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_SUSPENDED}, diff --git a/cmd/ateapi/internal/controlapi/workflow_suspend_test.go b/cmd/ateapi/internal/controlapi/workflow_suspend_test.go index 57c1ca3c25..5a8eff3582 100644 --- a/cmd/ateapi/internal/controlapi/workflow_suspend_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_suspend_test.go @@ -37,6 +37,9 @@ import ( func TestEnsureMarkedSuspending_SnapshotName(t *testing.T) { ctx := context.Background() persistence := newTestPersistence(t) + if _, err := persistence.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: "team-a"}}); err != nil { + t.Fatalf("CreateAtespace: %v", err) + } actor, err := persistence.CreateActor(ctx, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "actor-1"}, Status: &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_RUNNING}, @@ -77,6 +80,9 @@ func TestEnsureMarkedSuspending_SnapshotName(t *testing.T) { func TestEnsureMarkedSuspending_ReentryKeepsPersistedSnapshotLocation(t *testing.T) { ctx := context.Background() persistence := newTestPersistence(t) + if _, err := persistence.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: "team-a"}}); err != nil { + t.Fatalf("CreateAtespace: %v", err) + } actor, err := persistence.CreateActor(ctx, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "actor-1"}, Status: &ateapipb.ActorStatus{ @@ -175,6 +181,9 @@ func TestEnsureMarkedSuspending_StateMatrix(t *testing.T) { w := &ActorWorkflow{store: persistence} actorRef := resources.ActorRef{Atespace: "team-a", Name: "id1"} + if _, err := persistence.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: actorRef.Atespace}}); err != nil { + t.Fatalf("CreateAtespace: %v", err) + } actor, err := persistence.CreateActor(ctx, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, Status: &ateapipb.ActorStatus{State: seedState}, @@ -262,6 +271,10 @@ func TestEnsureAteletSuspended_DanglingWorkerDoesNotRecordPhantomSnapshot(t *tes ctx := context.Background() persistence := newTestPersistence(t) + if _, err := persistence.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: "team-a"}}); err != nil { + t.Fatalf("CreateAtespace: %v", err) + } + actor := &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "actor-1"}, Status: &ateapipb.ActorStatus{ @@ -316,6 +329,10 @@ func TestEnsureSuspendedFinalized_NoAssignment(t *testing.T) { ctx := context.Background() persistence := newTestPersistence(t) + if _, err := persistence.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: "team-a"}}); err != nil { + t.Fatalf("CreateAtespace: %v", err) + } + const snapshotName = "2026-01-01t00-00-00z-abc" actor := &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "actor-1"}, @@ -402,6 +419,10 @@ func TestEnsureSuspendedFinalized_ReleasesOnlyOwnWorker(t *testing.T) { ctx := context.Background() persistence := newTestPersistence(t) + if _, err := persistence.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: "team-a"}}); err != nil { + t.Fatalf("CreateAtespace: %v", err) + } + actor := &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "shared"}, Status: &ateapipb.ActorStatus{ @@ -467,6 +488,10 @@ func TestEnsureSuspendedFinalized_SnapshotSourceActorVersion(t *testing.T) { ctx := context.Background() persistence := newTestPersistence(t) + if _, err := persistence.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: "team-a"}}); err != nil { + t.Fatalf("CreateAtespace: %v", err) + } + const snapshotName = "2026-01-01t00-00-00z-abc" _, err := persistence.CreateActor(ctx, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "actor-1"}, @@ -594,6 +619,9 @@ func TestEnsureMarkedSuspending_PausedScopeRejection(t *testing.T) { w := &ActorWorkflow{store: persistence} actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"} + if _, err := persistence.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: actorRef.Atespace}}); err != nil { + t.Fatalf("CreateAtespace: %v", err) + } actor, err := persistence.CreateActor(ctx, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name}, Status: &ateapipb.ActorStatus{ @@ -628,6 +656,10 @@ func TestEnsurePausedSnapshotUploaded_Preconditions(t *testing.T) { persistence := newTestPersistence(t) w := &ActorWorkflow{store: persistence, dialer: newDanglingDialer()} + if _, err := persistence.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: "team-a"}}); err != nil { + t.Fatalf("CreateAtespace: %v", err) + } + created, err := persistence.CreateActor(ctx, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "actor-1"}, Status: &ateapipb.ActorStatus{ @@ -657,6 +689,10 @@ func TestEnsurePausedSnapshotUploaded_Preconditions(t *testing.T) { persistence := newTestPersistence(t) w := &ActorWorkflow{store: persistence, dialer: newDanglingDialer()} + if _, err := persistence.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: "team-a"}}); err != nil { + t.Fatalf("CreateAtespace: %v", err) + } + created, err := persistence.CreateActor(ctx, &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "actor-1"}, Status: &ateapipb.ActorStatus{ diff --git a/cmd/ateapi/internal/controlapi/workflow_testutil_test.go b/cmd/ateapi/internal/controlapi/workflow_testutil_test.go index c8da2fed8c..360affba0d 100644 --- a/cmd/ateapi/internal/controlapi/workflow_testutil_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_testutil_test.go @@ -57,6 +57,14 @@ func newTestActorWorkflow(t *testing.T, st store.Interface, tmplNamespace, tmplN // opts mutate the actor before it is stored. func seedWorkflowActor(t *testing.T, ctx context.Context, st store.Interface, actorRef resources.ActorRef, tmplNamespace, tmplName string, actorState ateapipb.ActorState, opts ...func(*ateapipb.Actor)) { t.Helper() + + atespace := &ateapipb.Atespace{ + Metadata: &ateapipb.ResourceMetadata{Name: actorRef.Atespace}, + } + if _, err := st.CreateAtespace(ctx, atespace); err != nil { + t.Fatalf("seed atespace: %v", err) + } + actor := &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Name: actorRef.Name, Atespace: actorRef.Atespace}, Status: &ateapipb.ActorStatus{State: actorState}, diff --git a/cmd/ateapi/internal/store/atepg/atepg.go b/cmd/ateapi/internal/store/atepg/atepg.go index 8343959cf0..e7ece7d06a 100644 --- a/cmd/ateapi/internal/store/atepg/atepg.go +++ b/cmd/ateapi/internal/store/atepg/atepg.go @@ -174,14 +174,6 @@ func (p *Persistence) GetAtespace(ctx context.Context, name string) (*ateapipb.A return getAtespaceRow(ctx, p.pool, name) } -func (p *Persistence) AtespaceExists(ctx context.Context, name string) (bool, error) { - var exists bool - if err := p.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM atespaces WHERE name = $1)`, name).Scan(&exists); err != nil { - return false, fmt.Errorf("checking atespace existence: %w", err) - } - return exists, nil -} - func (p *Persistence) ListAtespaces(ctx context.Context, opts store.ListOptions) (store.ListResponse[*ateapipb.Atespace], error) { pageSize, pageTokenStr := opts.PageSize, opts.PageToken token, err := decodePageToken(pageTokenStr, kindAtespace, "", 1) diff --git a/cmd/ateapi/internal/store/ateredis/ateredis.go b/cmd/ateapi/internal/store/ateredis/ateredis.go index 9e3df16399..e3594e0110 100644 --- a/cmd/ateapi/internal/store/ateredis/ateredis.go +++ b/cmd/ateapi/internal/store/ateredis/ateredis.go @@ -57,6 +57,8 @@ import ( "github.com/agent-substrate/substrate/pkg/proto/ateapipb" "github.com/google/uuid" "github.com/redis/go-redis/v9" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "google.golang.org/protobuf/encoding/protojson" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/timestamppb" @@ -173,9 +175,9 @@ func (s *Persistence) GetAtespace(ctx context.Context, name string) (*ateapipb.A return atespace, nil } -// AtespaceExists reports whether the atespace object exists. This is a plain +// atespaceExists reports whether the atespace object exists. This is a plain // EXISTS check and is NOT atomic with respect to a concurrent DeleteAtespace. -func (s *Persistence) AtespaceExists(ctx context.Context, name string) (bool, error) { +func (s *Persistence) atespaceExists(ctx context.Context, name string) (bool, error) { n, err := s.rdb.Exists(ctx, atespaceDBKey(name)).Result() if err != nil { return false, fmt.Errorf("while checking atespace existence: %w", err) @@ -282,6 +284,14 @@ func actorTemplateScanPattern(atespace string) string { } func (s *Persistence) CreateActorTemplate(ctx context.Context, template *ateapipb.ActorTemplate) (*ateapipb.ActorTemplate, error) { + // The atespace must already exist. This is non-atomic must be fixed if we + // keep the redis implementation. + if exists, err := s.atespaceExists(ctx, template.Metadata.Atespace); err != nil { + return nil, fmt.Errorf("while checking atespace: %w", err) + } else if !exists { + return nil, status.Errorf(codes.FailedPrecondition, "Atespace %s not found", template.Metadata.Atespace) + } + dbKey := actorTemplateDBKey(resources.ActorTemplateRefFromActorTemplate(template)) dbTemplate := proto.Clone(template).(*ateapipb.ActorTemplate) @@ -596,6 +606,14 @@ func (s *Persistence) GetActor(ctx context.Context, actorRef resources.ActorRef) } func (s *Persistence) CreateActor(ctx context.Context, actor *ateapipb.Actor) (*ateapipb.Actor, error) { + // The atespace must already exist. This is non-atomic must be fixed if we + // keep the redis implementation. + if exists, err := s.atespaceExists(ctx, actor.Metadata.Atespace); err != nil { + return nil, fmt.Errorf("while checking atespace: %w", err) + } else if !exists { + return nil, status.Errorf(codes.FailedPrecondition, "Atespace %s not found", actor.Metadata.Atespace) + } + dbKey := actorDBKey(resources.ActorRefFromActor(actor)) // Clone so we don't stomp the caller's copy, then attach fresh server-owned diff --git a/cmd/ateapi/internal/store/ateredis/ateredis_test.go b/cmd/ateapi/internal/store/ateredis/ateredis_test.go index fe7b26798c..b696d470c0 100644 --- a/cmd/ateapi/internal/store/ateredis/ateredis_test.go +++ b/cmd/ateapi/internal/store/ateredis/ateredis_test.go @@ -78,6 +78,10 @@ func TestGetActor_NotFound(t *testing.T) { func TestCreateActor_Success(t *testing.T) { _, s, ctx := setupTest(t) + if _, err := s.CreateAtespace(ctx, newTestAtespace(testAtespace)); err != nil { + t.Fatalf("CreateAtespace() failed: %v", err) + } + actor := &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Name: "actor-1", Atespace: testAtespace}, ActorTemplateNamespace: "default", @@ -133,6 +137,9 @@ func TestCreateActor_AlreadyExists(t *testing.T) { Status: &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_SUSPENDED}, } + if _, err := s.CreateAtespace(ctx, newTestAtespace(actor.Metadata.Atespace)); err != nil { + t.Fatalf("CreateAtespace() failed: %v", err) + } _, err := s.CreateActor(ctx, actor) if err != nil { t.Fatalf("CreateActor failed: %v", err) @@ -157,6 +164,9 @@ func newTestActor(name string) *ateapipb.Actor { func TestUpdateActor_Success(t *testing.T) { _, s, ctx := setupTest(t) actor := newTestActor("actor-1") + if _, err := s.CreateAtespace(ctx, newTestAtespace(actor.Metadata.Atespace)); err != nil { + t.Fatalf("CreateAtespace() failed: %v", err) + } created, err := s.CreateActor(ctx, actor) if err != nil { t.Fatalf("CreateActor failed: %v", err) @@ -199,6 +209,9 @@ func TestUpdateActor_Success(t *testing.T) { func TestUpdateActor_MutateErrorAreNotRetried(t *testing.T) { _, s, ctx := setupTest(t) actor := newTestActor("actor-1") + if _, err := s.CreateAtespace(ctx, newTestAtespace(actor.Metadata.Atespace)); err != nil { + t.Fatalf("CreateAtespace() failed: %v", err) + } created, err := s.CreateActor(ctx, actor) if err != nil { t.Fatalf("CreateActor failed: %v", err) @@ -235,6 +248,9 @@ func TestUpdateActor_DiscardsServerOwnedFieldsEdits(t *testing.T) { _, s, ctx := setupTest(t) actor := newTestActor("actor-1") + if _, err := s.CreateAtespace(ctx, newTestAtespace(actor.Metadata.Atespace)); err != nil { + t.Fatalf("CreateAtespace() failed: %v", err) + } created, err := s.CreateActor(ctx, actor) if err != nil { t.Fatalf("CreateActor failed: %v", err) @@ -303,6 +319,9 @@ func TestUpdateActor_RejectsImmutableFieldChange(t *testing.T) { t.Run(tt.name, func(t *testing.T) { _, s, ctx := setupTest(t) actor := newTestActor("actor-1") + if _, err := s.CreateAtespace(ctx, newTestAtespace(actor.Metadata.Atespace)); err != nil { + t.Fatalf("CreateAtespace() failed: %v", err) + } created, err := s.CreateActor(ctx, actor) if err != nil { t.Fatalf("CreateActor failed: %v", err) @@ -351,6 +370,9 @@ func (w *watchInterceptor) Watch(ctx context.Context, fn func(*redis.Tx) error, func TestUpdateActor_RetriesOnConcurrentWrite(t *testing.T) { mr, s, ctx := setupTest(t) actor := newTestActor("actor-1") + if _, err := s.CreateAtespace(ctx, newTestAtespace(actor.Metadata.Atespace)); err != nil { + t.Fatalf("CreateAtespace() failed: %v", err) + } created, err := s.CreateActor(ctx, actor) if err != nil { t.Fatalf("CreateActor failed: %v", err) @@ -423,7 +445,11 @@ func TestUpdateActor_NotFound(t *testing.T) { func TestUpdateActor_RejectsStaleUID(t *testing.T) { _, s, ctx := setupTest(t) - original, err := s.CreateActor(ctx, newTestActor("actor-1")) + a := newTestActor("actor-1") + if _, err := s.CreateAtespace(ctx, newTestAtespace(a.Metadata.Atespace)); err != nil { + t.Fatalf("CreateAtespace() failed: %v", err) + } + original, err := s.CreateActor(ctx, a) if err != nil { t.Fatalf("CreateActor failed: %v", err) } @@ -473,7 +499,11 @@ func TestUpdateActor_RejectsStaleUID(t *testing.T) { func TestUpdateActor_RejectsStaleVersion(t *testing.T) { _, s, ctx := setupTest(t) - created, err := s.CreateActor(ctx, newTestActor("actor-1")) + a := newTestActor("actor-1") + if _, err := s.CreateAtespace(ctx, newTestAtespace(a.Metadata.Atespace)); err != nil { + t.Fatalf("CreateAtespace() failed: %v", err) + } + created, err := s.CreateActor(ctx, a) if err != nil { t.Fatalf("CreateActor failed: %v", err) } @@ -844,6 +874,9 @@ func TestDeleteActor(t *testing.T) { Status: &ateapipb.ActorStatus{State: tt.state}, } + if _, err := s.CreateAtespace(ctx, newTestAtespace(testAtespace)); err != nil { + t.Fatalf("CreateAtespace(%s) failed: %v", testAtespace, err) + } if _, err := s.CreateActor(ctx, actor); err != nil { t.Fatalf("CreateActor failed: %v", err) } @@ -883,7 +916,6 @@ func TestListActors(t *testing.T) { _, s, ctx := setupTest(t) actor1 := &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Name: "id1", Atespace: testAtespace}, ActorTemplateNamespace: "ns1", ActorTemplateName: "tmpl1", @@ -901,7 +933,9 @@ func TestListActors(t *testing.T) { LatestSnapshot: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "snapshot-2"}, }, } - + if _, err := s.CreateAtespace(ctx, newTestAtespace(testAtespace)); err != nil { + t.Fatalf("CreateAtespace() failed: %v", err) + } if _, err := s.CreateActor(ctx, actor1); err != nil { t.Fatalf("failed to create actor1: %v", err) } @@ -1365,6 +1399,10 @@ func TestListActors_Empty(t *testing.T) { func TestListActors_Pagination(t *testing.T) { _, s, ctx := setupTest(t) + if _, err := s.CreateAtespace(ctx, newTestAtespace(testAtespace)); err != nil { + t.Fatalf("CreateAtespace() failed: %v", err) + } + for i := 0; i < 5; i++ { actor := &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Name: fmt.Sprintf("name%d", i), Atespace: testAtespace}, @@ -1764,6 +1802,13 @@ func receiveEvent(t *testing.T, ch <-chan store.WorkerEvent) store.WorkerEvent { func TestListActors_ScopedByAtespace(t *testing.T) { _, s, ctx := setupTest(t) + if _, err := s.CreateAtespace(ctx, newTestAtespace("team-a")); err != nil { + t.Fatalf("CreateAtespace(team-a) failed: %v", err) + } + if _, err := s.CreateAtespace(ctx, newTestAtespace("team-b")); err != nil { + t.Fatalf("CreateAtespace(team-b) failed: %v", err) + } + mkActor := func(atespace, name string) *ateapipb.Actor { return &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{Name: name, Atespace: atespace}, @@ -1886,14 +1931,14 @@ func TestGetAtespace_NotFound(t *testing.T) { func TestAtespaceExists(t *testing.T) { _, s, ctx := setupTest(t) - if ok, err := s.AtespaceExists(ctx, "team-a"); err != nil || ok { - t.Fatalf("AtespaceExists before create = (%v, %v), want (false, nil)", ok, err) + if ok, err := s.atespaceExists(ctx, "team-a"); err != nil || ok { + t.Fatalf("atespaceExists before create = (%v, %v), want (false, nil)", ok, err) } if _, err := s.CreateAtespace(ctx, newTestAtespace("team-a")); err != nil { t.Fatalf("CreateAtespace failed: %v", err) } - if ok, err := s.AtespaceExists(ctx, "team-a"); err != nil || !ok { - t.Fatalf("AtespaceExists after create = (%v, %v), want (true, nil)", ok, err) + if ok, err := s.atespaceExists(ctx, "team-a"); err != nil || !ok { + t.Fatalf("atespaceExists after create = (%v, %v), want (true, nil)", ok, err) } } @@ -2061,6 +2106,7 @@ func TestDeleteAtespace_EmptyWhileOtherAtespaceNonEmpty(t *testing.T) { if _, err := s.CreateAtespace(ctx, newTestAtespace("team-b")); err != nil { t.Fatalf("CreateAtespace(team-b) failed: %v", err) } + // Actor lives ONLY in team-b. if _, err := s.CreateActor(ctx, &ateapipb.Actor{Metadata: &ateapipb.ResourceMetadata{Name: "id1", Atespace: "team-b"}, Status: &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_SUSPENDED}}); err != nil { t.Fatalf("CreateActor failed: %v", err) @@ -2180,6 +2226,11 @@ func TestListActors_MultiMaster_Pagination(t *testing.T) { for shardIdx, sh := range shards { clients = append(clients, sh.client) tempS := &Persistence{rdb: sh.clusterClient} + + if _, err := tempS.CreateAtespace(ctx, newTestAtespace(testAtespace)); err != nil { + t.Fatalf("failed to seed atespace: %v", err) + } + for itemIdx := 0; itemIdx < 3; itemIdx++ { actor := &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{ @@ -2465,6 +2516,10 @@ func newTestActorTemplate(atespace, name string) *ateapipb.ActorTemplate { func TestActorTemplateLifecycle(t *testing.T) { _, s, ctx := setupTest(t) + if _, err := s.CreateAtespace(ctx, newTestAtespace("team-a")); err != nil { + t.Fatalf("CreateAtespace() failed: %v", err) + } + want := newTestActorTemplate("team-a", "tmpl-a") created, err := s.CreateActorTemplate(ctx, want) if err != nil { @@ -2516,6 +2571,10 @@ func TestActorTemplateLifecycle(t *testing.T) { func TestCreateActorTemplate_AlreadyExists(t *testing.T) { _, s, ctx := setupTest(t) + if _, err := s.CreateAtespace(ctx, newTestAtespace("team-a")); err != nil { + t.Fatalf("CreateAtespace() failed: %v", err) + } + if _, err := s.CreateActorTemplate(ctx, newTestActorTemplate("team-a", "tmpl-a")); err != nil { t.Fatalf("first CreateActorTemplate failed: %v", err) } @@ -2535,6 +2594,10 @@ func TestGetActorTemplate_NotFound(t *testing.T) { func TestActorTemplateExists(t *testing.T) { _, s, ctx := setupTest(t) + if _, err := s.CreateAtespace(ctx, newTestAtespace("team-a")); err != nil { + t.Fatalf("CreateAtespace() failed: %v", err) + } + if ok, err := s.ActorTemplateExists(ctx, resources.ActorTemplateRef{Atespace: "team-a", Name: "tmpl-a"}); err != nil || ok { t.Fatalf("ActorTemplateExists before create = (%v, %v), want (false, nil)", ok, err) } @@ -2571,6 +2634,10 @@ func TestDeleteActorTemplate_NotFound(t *testing.T) { func TestListActorTemplates_Pagination(t *testing.T) { _, s, ctx := setupTest(t) + if _, err := s.CreateAtespace(ctx, newTestAtespace("team-a")); err != nil { + t.Fatalf("CreateAtespace() failed: %v", err) + } + for i := 0; i < 5; i++ { if _, err := s.CreateActorTemplate(ctx, newTestActorTemplate("team-a", fmt.Sprintf("tmpl-%d", i))); err != nil { t.Fatalf("failed to create template %d: %v", i, err) @@ -2607,10 +2674,16 @@ func TestActorTemplates_AtespaceIsolation(t *testing.T) { _, s, ctx := setupTest(t) // The same name in two atespaces is two distinct resources. + if _, err := s.CreateAtespace(ctx, newTestAtespace("team-a")); err != nil { + t.Fatalf("CreateAtespace(%s) failed: %v", "team-a", err) + } inA, err := s.CreateActorTemplate(ctx, newTestActorTemplate("team-a", "tmpl")) if err != nil { t.Fatalf("CreateActorTemplate in team-a failed: %v", err) } + if _, err := s.CreateAtespace(ctx, newTestAtespace("team-b")); err != nil { + t.Fatalf("CreateAtespace(%s) failed: %v", "team-b", err) + } inB, err := s.CreateActorTemplate(ctx, newTestActorTemplate("team-b", "tmpl")) if err != nil { t.Fatalf("CreateActorTemplate in team-b = %v, want nil: the name is only taken in team-a", err) @@ -2647,6 +2720,13 @@ func TestActorTemplates_AtespaceIsolation(t *testing.T) { func TestListActorTemplates_AtespaceFilter(t *testing.T) { _, s, ctx := setupTest(t) + if _, err := s.CreateAtespace(ctx, newTestAtespace("team-a")); err != nil { + t.Fatalf("CreateAtespace(%s) failed: %v", "team-a", err) + } + if _, err := s.CreateAtespace(ctx, newTestAtespace("team-b")); err != nil { + t.Fatalf("CreateAtespace(%s) failed: %v", "team-b", err) + } + for _, tmpl := range []struct{ atespace, name string }{ {"team-a", "tmpl-1"}, {"team-a", "tmpl-2"}, {"team-b", "tmpl-3"}, } { diff --git a/cmd/ateapi/internal/store/store.go b/cmd/ateapi/internal/store/store.go index 554f34097c..064c3de024 100644 --- a/cmd/ateapi/internal/store/store.go +++ b/cmd/ateapi/internal/store/store.go @@ -139,9 +139,6 @@ type Interface interface { // Fetches an atespace by name. Returns ErrNotFound if missing. GetAtespace(ctx context.Context, name string) (*ateapipb.Atespace, error) - // AtespaceExists reports whether the atespace object exists. - AtespaceExists(ctx context.Context, name string) (bool, error) - // Lists atespaces. ListAtespaces(ctx context.Context, opts ListOptions) (ListResponse[*ateapipb.Atespace], error) diff --git a/cmd/ateapi/internal/store/storecontract/contract.go b/cmd/ateapi/internal/store/storecontract/contract.go index 60b4785e26..0d5f66bf6a 100644 --- a/cmd/ateapi/internal/store/storecontract/contract.go +++ b/cmd/ateapi/internal/store/storecontract/contract.go @@ -1396,21 +1396,6 @@ func runAtespaceContractTests(t *testing.T, setup func(t *testing.T) store.Inter } }) - t.Run("AtespaceExists", func(t *testing.T) { - s := setup(t) - ctx := context.Background() - - if ok, err := s.AtespaceExists(ctx, "team-a"); err != nil || ok { - t.Fatalf("AtespaceExists before create = (%v, %v), want (false, nil)", ok, err) - } - if _, err := s.CreateAtespace(ctx, newTestAtespace("team-a")); err != nil { - t.Fatalf("CreateAtespace failed: %v", err) - } - if ok, err := s.AtespaceExists(ctx, "team-a"); err != nil || !ok { - t.Fatalf("AtespaceExists after create = (%v, %v), want (true, nil)", ok, err) - } - }) - t.Run("ListAtespaces", func(t *testing.T) { s := setup(t) ctx := context.Background()