From 821acc0a2208a415f670012f6a84a63ba08983f7 Mon Sep 17 00:00:00 2001 From: msau42 Date: Fri, 24 Jul 2026 02:50:42 +0000 Subject: [PATCH 1/5] Handle partial creation failures during CreateActor. 1. During CreateActor, the volumes to be created get instantiated in the Actor API in CREATING status. The Actor is still instantiated as SUSPENDED. However, the CreateVolume operation is moved to the beginning of ResumeActor. This is because CreateActor is intentionally designed to not be idempotent. The downside is that the user won't know dependent resource creation failed until the first resume. 2. Changes volume id to use actor uid instead of atespace+actorname, and add "substrate" prefix 3. Adds a new DELETING status that gets persisted before we start to delete volumes. This shifts the existing DB DELETE precondition checks to the new DELETING status. And now you can only delete the actor in DB from DELETING status. In the future, we will also add a workflow that will do mandatory node cleanup of volumes before starting to delete the volumes. --- .../internal/controlapi/create_actor.go | 16 +- .../internal/controlapi/delete_actor.go | 27 +- .../internal/controlapi/delete_actor_test.go | 118 ++++ .../internal/controlapi/functional_test.go | 502 +++++++++++++++++- cmd/ateapi/internal/controlapi/service.go | 1 - cmd/ateapi/internal/controlapi/volumes.go | 120 +++-- .../internal/controlapi/volumes_test.go | 304 +++++++++++ cmd/ateapi/internal/controlapi/workflow.go | 1 + .../internal/controlapi/workflow_resume.go | 48 ++ .../internal/controlapi/workload_spec.go | 3 +- .../internal/controlapi/workload_spec_test.go | 2 +- .../internal/store/ateredis/ateredis.go | 3 +- .../internal/store/ateredis/ateredis_test.go | 7 +- pkg/proto/ateapipb/ateapi.pb.go | 65 ++- pkg/proto/ateapipb/ateapi.proto | 20 +- 15 files changed, 1143 insertions(+), 94 deletions(-) create mode 100644 cmd/ateapi/internal/controlapi/volumes_test.go diff --git a/cmd/ateapi/internal/controlapi/create_actor.go b/cmd/ateapi/internal/controlapi/create_actor.go index f19ee0d1b..d6b9e9bea 100644 --- a/cmd/ateapi/internal/controlapi/create_actor.go +++ b/cmd/ateapi/internal/controlapi/create_actor.go @@ -59,15 +59,9 @@ func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequ return nil, status.Errorf(codes.FailedPrecondition, "Atespace %s not found", atespace) } - actorRef := &ateapipb.ObjectRef{ - Atespace: atespace, - Name: name, - } - - volumes, err := s.createActorVolumes(ctx, actorRef, template) - if err != nil { - return nil, err - } + // Assuming CreateActor() will not become idempotent so we won't override volume state + // if the actor already exists. + initVols := initialActorVolumes(template) actor := &ateapipb.Actor{ Metadata: &ateapipb.ResourceMetadata{ @@ -78,12 +72,10 @@ func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequ ActorTemplateNamespace: templateNamespace, ActorTemplateName: templateName, WorkerSelector: in.GetWorkerSelector(), - ActorVolumes: volumes, + ActorVolumes: initVols, } stored, err := s.persistence.CreateActor(ctx, actor) if err != nil { - // Cleanup created volumes if DB write fails - _ = s.deleteActorVolumes(ctx, actorRef, volumes) if errors.Is(err, store.ErrAlreadyExists) { return nil, status.Errorf(codes.AlreadyExists, "Actor %s already exists", name) } diff --git a/cmd/ateapi/internal/controlapi/delete_actor.go b/cmd/ateapi/internal/controlapi/delete_actor.go index 1dc2a1c01..53ea38f02 100644 --- a/cmd/ateapi/internal/controlapi/delete_actor.go +++ b/cmd/ateapi/internal/controlapi/delete_actor.go @@ -42,8 +42,29 @@ func (s *Service) DeleteActor(ctx context.Context, req *ateapipb.DeleteActorRequ return nil, fmt.Errorf("while fetching actor: %w", err) } + if actor.GetStatus() != ateapipb.Actor_STATUS_SUSPENDED && + actor.GetStatus() != ateapipb.Actor_STATUS_CRASHED && + actor.GetStatus() != ateapipb.Actor_STATUS_DELETING { + return nil, status.Errorf(codes.FailedPrecondition, "Actor %s is not in a deletable status (status: %v)", name, actor.GetStatus()) + } + + if actor.GetStatus() != ateapipb.Actor_STATUS_DELETING { + actor.Status = ateapipb.Actor_STATUS_DELETING + for _, vol := range actor.GetActorVolumes() { + vol.Status = ateapipb.ExternalVolume_DELETING + } + updated, err := s.persistence.UpdateActor(ctx, actor, actor.GetMetadata().GetVersion()) + if err != nil { + if errors.Is(err, store.ErrPersistenceRetry) { + return nil, status.Error(codes.Aborted, "concurrent update conflict, please retry") + } + return nil, fmt.Errorf("while setting actor status to DELETING: %w", err) + } + actor = updated + } + // Delete associated volumes - if err := s.deleteActorVolumes(ctx, req.GetActor(), actor.GetActorVolumes()); err != nil { + if err := s.deleteActorVolumes(ctx, actor.GetMetadata().GetUid(), actor.GetActorVolumes()); err != nil { return nil, status.Errorf(codes.Internal, "while deleting actor volumes: %v", err) } @@ -55,9 +76,9 @@ func (s *Service) DeleteActor(ctx context.Context, req *ateapipb.DeleteActorRequ if errors.Is(err, store.ErrFailedPrecondition) { current, getErr := s.persistence.GetActor(ctx, actorRef) if getErr == nil { - return nil, status.Errorf(codes.FailedPrecondition, "Actor %s is not suspended (status: %v)", actorRef, current.GetStatus()) + return nil, status.Errorf(codes.FailedPrecondition, "Actor %s is not in a deletable status (status: %v)", actorRef, current.GetStatus()) } - return nil, status.Errorf(codes.FailedPrecondition, "Actor %s is not suspended", actorRef) + return nil, status.Errorf(codes.FailedPrecondition, "Actor %s is not in a deletable status", actorRef) } if errors.Is(err, store.ErrVersionConflict) { return nil, status.Error(codes.Aborted, "concurrent update conflict, please retry") diff --git a/cmd/ateapi/internal/controlapi/delete_actor_test.go b/cmd/ateapi/internal/controlapi/delete_actor_test.go index 9e78f676e..00d294c36 100644 --- a/cmd/ateapi/internal/controlapi/delete_actor_test.go +++ b/cmd/ateapi/internal/controlapi/delete_actor_test.go @@ -16,9 +16,14 @@ package controlapi import ( "context" + "fmt" + "strings" "testing" + "github.com/google/go-cmp/cmp" + "github.com/agent-substrate/substrate/internal/ateattr" + "github.com/agent-substrate/substrate/internal/volume" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" "k8s.io/apimachinery/pkg/util/validation/field" ) @@ -88,3 +93,116 @@ func TestValidateDeleteActorRequest(t *testing.T) { }) } } + +func TestDeleteActor_StatusDeleting(t *testing.T) { + ns := namespaceForTest("ns-delete-deleting") + tc := setupTest(t, ns) + defer tc.cleanup() + createTemplate(t, tc, ns) + + deletingActor := &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{ + Atespace: testAtespace, + Name: "deleting-actor", + }, + Status: ateapipb.Actor_STATUS_DELETING, + ActorTemplateNamespace: ns, + ActorTemplateName: "tmpl1", + } + if _, err := tc.persistence.CreateActor(context.Background(), deletingActor); err != nil { + t.Fatalf("CreateActor: %v", err) + } + + if _, err := tc.service.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "deleting-actor"}, + }); err != nil { + t.Fatalf("DeleteActor on STATUS_DELETING actor failed: %v", err) + } + + if _, err := tc.persistence.GetActor(context.Background(), resources.ActorRef{Atespace: testAtespace, Name: "deleting-actor"}); err == nil { + t.Errorf("expected actor to be deleted, but it still exists") + } +} + +func TestDeleteActor_WrongStatus(t *testing.T) { + ns := namespaceForTest("ns-delete-wrong-status") + tc := setupTest(t, ns) + defer tc.cleanup() + createTemplate(t, tc, ns) + + runningActor := &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{ + Atespace: testAtespace, + Name: "running-actor", + }, + Status: ateapipb.Actor_STATUS_RUNNING, + ActorTemplateNamespace: ns, + ActorTemplateName: "tmpl1", + } + if _, err := tc.persistence.CreateActor(context.Background(), runningActor); err != nil { + t.Fatalf("CreateActor: %v", err) + } + + _, err := tc.service.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "running-actor"}, + }) + if err == nil { + t.Fatalf("expected DeleteActor on STATUS_RUNNING actor to fail, but it succeeded") + } +} + +type failingVolumePlugin struct { + volume.VolumePluginControlPlane + deletedIDs []string +} + +func (f *failingVolumePlugin) DeleteVolume(ctx context.Context, volumeID string) error { + f.deletedIDs = append(f.deletedIDs, volumeID) + return fmt.Errorf("simulated delete error for %s", volumeID) +} + +func TestDeleteActor_MultipleVolumeDeletionFailures(t *testing.T) { + ns := namespaceForTest("ns-delete-multivol-fail") + tc := setupTest(t, ns) + defer tc.cleanup() + createTemplate(t, tc, ns) + + plugin := &failingVolumePlugin{} + oldGlobalPlugin := globalVolumePlugin + globalVolumePlugin = plugin + defer func() { globalVolumePlugin = oldGlobalPlugin }() + + actor := &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{ + Atespace: testAtespace, + Name: "multi-vol-actor", + }, + Status: ateapipb.Actor_STATUS_SUSPENDED, + ActorTemplateNamespace: ns, + ActorTemplateName: "tmpl1", + ActorVolumes: []*ateapipb.ExternalVolume{ + {VolumeName: "vol1", StorageVolumeId: "storage-vol-1", Status: ateapipb.ExternalVolume_CREATED}, + {VolumeName: "vol2", StorageVolumeId: "storage-vol-2", Status: ateapipb.ExternalVolume_CREATED}, + }, + } + if _, err := tc.persistence.CreateActor(context.Background(), actor); err != nil { + t.Fatalf("CreateActor: %v", err) + } + + _, err := tc.service.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "multi-vol-actor"}, + }) + if err == nil { + t.Fatalf("expected DeleteActor to fail when volume deletion fails, but it succeeded") + } + + wantDeleted := []string{"storage-vol-1", "storage-vol-2"} + if diff := cmp.Diff(wantDeleted, plugin.deletedIDs); diff != "" { + t.Errorf("deletedIDs mismatch (-want +got):\n%s", diff) + } + + errMsg := err.Error() + if !strings.Contains(errMsg, "storage-vol-1") || !strings.Contains(errMsg, "storage-vol-2") { + t.Errorf("expected error message to contain both volume failure details, got: %v", errMsg) + } +} diff --git a/cmd/ateapi/internal/controlapi/functional_test.go b/cmd/ateapi/internal/controlapi/functional_test.go index 4b914d4fd..8793a82cb 100644 --- a/cmd/ateapi/internal/controlapi/functional_test.go +++ b/cmd/ateapi/internal/controlapi/functional_test.go @@ -32,6 +32,7 @@ import ( "github.com/agent-substrate/substrate/internal/envtestbins" "github.com/agent-substrate/substrate/internal/proto/ateletpb" "github.com/agent-substrate/substrate/internal/resources" + "github.com/agent-substrate/substrate/internal/volume" atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" "github.com/agent-substrate/substrate/pkg/client/clientset/versioned" "github.com/agent-substrate/substrate/pkg/client/informers/externalversions" @@ -51,6 +52,7 @@ import ( "google.golang.org/protobuf/types/known/timestamppb" corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/util/validation/field" "k8s.io/apimachinery/pkg/util/wait" @@ -433,6 +435,21 @@ func createAtespace(t *testing.T, tc *testContext, name string) { const poolLabelKey = "pool" func createTemplateWithContainers(t *testing.T, tc *testContext, ns string, containers []atev1alpha1.Container) { + createTemplateWithContainersAndVolumes(t, tc, ns, containers, nil) +} + +func createTemplateWithVolumes(t *testing.T, tc *testContext, ns string, volumes []atev1alpha1.Volume, mounts []atev1alpha1.VolumeMount) { + createTemplateWithContainersAndVolumes(t, tc, ns, []atev1alpha1.Container{ + { + Name: "main", + Image: "main@sha256:abc", + Command: []string{"/main"}, + VolumeMounts: mounts, + }, + }, volumes) +} + +func createTemplateWithContainersAndVolumes(t *testing.T, tc *testContext, ns string, containers []atev1alpha1.Container, volumes []atev1alpha1.Volume) { t.Helper() // Sandbox binaries now live on a (cluster-scoped) SandboxConfig resolved via @@ -452,6 +469,7 @@ func createTemplateWithContainers(t *testing.T, tc *testContext, ns string, cont Location: "gs://fake-fake-fake", }, Containers: containers, + Volumes: volumes, WorkerSelector: &metav1.LabelSelector{ MatchLabels: map[string]string{poolLabelKey: ns}, }, @@ -764,6 +782,484 @@ func TestCreateActor_Success(t *testing.T) { } } +func TestCreateActor_WithExternalVolumes(t *testing.T) { + ns := namespaceForTest("ns-create-ext-vols") + tc := setupTest(t, ns) + defer tc.cleanup() + + volumes := []atev1alpha1.Volume{ + { + Name: "ext-vol-1", + VolumeSource: atev1alpha1.VolumeSource{ + ExternalVolumeTemplate: &atev1alpha1.ExternalVolumeTemplate{ + StorageClassName: "standard", + Capacity: resource.MustParse("10Gi"), + }, + }, + }, + } + mounts := []atev1alpha1.VolumeMount{ + { + Name: "ext-vol-1", + MountPath: "/data", + }, + } + createTemplateWithVolumes(t, tc, ns, volumes, mounts) + + createResp, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{ + Actor: &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "vol-actor-1"}, + ActorTemplateNamespace: ns, + ActorTemplateName: "tmpl1", + }, + }) + if err != nil { + t.Fatalf("CreateActor failed: %v", err) + } + + if len(createResp.GetActorVolumes()) != 1 { + t.Fatalf("expected 1 volume in CreateActor response, got %d", len(createResp.GetActorVolumes())) + } + vol := createResp.GetActorVolumes()[0] + if vol.GetVolumeName() != "ext-vol-1" { + t.Errorf("volume name = %q, want %q", vol.GetVolumeName(), "ext-vol-1") + } + if vol.GetStatus() != ateapipb.ExternalVolume_CREATING { + t.Errorf("volume status = %v, want %v", vol.GetStatus(), ateapipb.ExternalVolume_CREATING) + } + if vol.GetStorageVolumeId() != "" { + t.Errorf("expected empty storageVolumeId before resume, got %q", vol.GetStorageVolumeId()) + } + + // Verify GetActor returns the same external volume state + getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "vol-actor-1"}, + }) + if err != nil { + t.Fatalf("GetActor failed: %v", err) + } + if len(getResp.GetActorVolumes()) != 1 { + t.Fatalf("expected 1 volume in GetActor response, got %d", len(getResp.GetActorVolumes())) + } + if getResp.GetActorVolumes()[0].GetStatus() != ateapipb.ExternalVolume_CREATING { + t.Errorf("GetActor status = %v, want %v", getResp.GetActorVolumes()[0].GetStatus(), ateapipb.ExternalVolume_CREATING) + } +} + +func TestActorLifecycle_WithExternalVolumes(t *testing.T) { + ns := namespaceForTest("ns-lifecycle-ext-vols") + tc := setupTest(t, ns) + defer tc.cleanup() + + volumes := []atev1alpha1.Volume{ + { + Name: "data-vol", + VolumeSource: atev1alpha1.VolumeSource{ + ExternalVolumeTemplate: &atev1alpha1.ExternalVolumeTemplate{ + StorageClassName: "fast", + Capacity: resource.MustParse("20Gi"), + }, + }, + }, + } + mounts := []atev1alpha1.VolumeMount{ + { + Name: "data-vol", + MountPath: "/mnt/data", + }, + } + createTemplateWithVolumes(t, tc, ns, volumes, mounts) + createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1") + + // 1. CreateActor + createResp, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{ + Actor: &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "actor-vol-lc"}, + ActorTemplateNamespace: ns, + ActorTemplateName: "tmpl1", + }, + }) + if err != nil { + t.Fatalf("CreateActor failed: %v", err) + } + if createResp.GetStatus() != ateapipb.Actor_STATUS_SUSPENDED { + t.Fatalf("expected initial status STATUS_SUSPENDED, got %v", createResp.GetStatus()) + } + if len(createResp.GetActorVolumes()) != 1 || createResp.GetActorVolumes()[0].GetStatus() != ateapipb.ExternalVolume_CREATING { + t.Fatalf("expected 1 creating volume after CreateActor, got %v", createResp.GetActorVolumes()) + } + + // 2. ResumeActor + resumeResp, err := tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "actor-vol-lc"}, + }) + if err != nil { + t.Fatalf("ResumeActor failed: %v", err) + } + if resumeResp.GetActor().GetStatus() != ateapipb.Actor_STATUS_RUNNING { + t.Fatalf("expected status STATUS_RUNNING after resume, got %v", resumeResp.GetActor().GetStatus()) + } + if len(resumeResp.GetActor().GetActorVolumes()) != 1 || resumeResp.GetActor().GetActorVolumes()[0].GetStatus() != ateapipb.ExternalVolume_CREATED { + t.Fatalf("expected 1 created volume after ResumeActor, got %v", resumeResp.GetActor().GetActorVolumes()) + } + if resumeResp.GetActor().GetActorVolumes()[0].GetStorageVolumeId() == "" { + t.Fatalf("expected non-empty storageVolumeId after ResumeActor") + } + + // 3. PauseActor + pauseResp, err := tc.client.PauseActor(context.Background(), &ateapipb.PauseActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "actor-vol-lc"}, + }) + if err != nil { + t.Fatalf("PauseActor failed: %v", err) + } + if pauseResp.GetActor().GetStatus() != ateapipb.Actor_STATUS_PAUSED { + t.Fatalf("expected status STATUS_PAUSED after pause, got %v", pauseResp.GetActor().GetStatus()) + } + + // 4. ResumeActor from paused + resumeResp2, err := tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "actor-vol-lc"}, + }) + if err != nil { + t.Fatalf("ResumeActor from paused failed: %v", err) + } + if resumeResp2.GetActor().GetStatus() != ateapipb.Actor_STATUS_RUNNING { + t.Fatalf("expected status STATUS_RUNNING after second resume, got %v", resumeResp2.GetActor().GetStatus()) + } + + // 5. SuspendActor + suspendResp, err := tc.client.SuspendActor(context.Background(), &ateapipb.SuspendActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "actor-vol-lc"}, + }) + if err != nil { + t.Fatalf("SuspendActor failed: %v", err) + } + if suspendResp.GetActor().GetStatus() != ateapipb.Actor_STATUS_SUSPENDED { + t.Fatalf("expected status STATUS_SUSPENDED after suspend, got %v", suspendResp.GetActor().GetStatus()) + } + + // 6. DeleteActor + deleteResp, err := tc.client.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "actor-vol-lc"}, + }) + if err != nil { + t.Fatalf("DeleteActor failed: %v", err) + } + if deleteResp.GetMetadata().GetName() != "actor-vol-lc" { + t.Errorf("deleted actor name = %q, want %q", deleteResp.GetMetadata().GetName(), "actor-vol-lc") + } + + // Confirm GetActor returns NotFound after deletion + _, err = tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "actor-vol-lc"}, + }) + if status.Code(err) != codes.NotFound { + t.Errorf("GetActor after delete err = %v, want NotFound", err) + } +} + +type partialFailVolumePlugin struct { + volume.VolumePluginControlPlane + deleted []string +} + +func (f *partialFailVolumePlugin) CreateVolume(ctx context.Context, name, capacity, storageClass string) (string, error) { + if strings.HasSuffix(name, "fail-vol2") { + return "", fmt.Errorf("simulated volume creation failure") + } + return "storage-" + name, nil +} + +func (f *partialFailVolumePlugin) AttachVolume(ctx context.Context, volumeID, node string) error { + return nil +} + +func (f *partialFailVolumePlugin) DetachVolume(ctx context.Context, volumeID, node string) error { + return nil +} + +func (f *partialFailVolumePlugin) DeleteVolume(ctx context.Context, volumeID string) error { + f.deleted = append(f.deleted, volumeID) + return nil +} + +// TestResumeActor_VolumeCreationFailure tests that when volume provisioning fails during ResumeActor, +// successfully created volumes are saved, the actor remains in STATUS_SUSPENDED, +// and that calling DeleteActor on the suspended actor cleans up all partially created volumes. +func TestResumeActor_VolumeCreationFailure(t *testing.T) { + ns := namespaceForTest("ns-resume-vol-fail") + tc := setupTest(t, ns) + defer tc.cleanup() + + volumes := []atev1alpha1.Volume{ + { + Name: "succ-vol1", + VolumeSource: atev1alpha1.VolumeSource{ + ExternalVolumeTemplate: &atev1alpha1.ExternalVolumeTemplate{ + StorageClassName: "standard", + Capacity: resource.MustParse("10Gi"), + }, + }, + }, + { + Name: "fail-vol2", + VolumeSource: atev1alpha1.VolumeSource{ + ExternalVolumeTemplate: &atev1alpha1.ExternalVolumeTemplate{ + StorageClassName: "standard", + Capacity: resource.MustParse("10Gi"), + }, + }, + }, + } + mounts := []atev1alpha1.VolumeMount{ + {Name: "succ-vol1", MountPath: "/mnt/vol1"}, + {Name: "fail-vol2", MountPath: "/mnt/vol2"}, + } + createTemplateWithVolumes(t, tc, ns, volumes, mounts) + + // Inject a custom partial-failing VolumePlugin into global scope + // TODO this doesn't support parallelism of test cases + plugin := &partialFailVolumePlugin{} + oldGlobalPlugin := globalVolumePlugin + globalVolumePlugin = plugin + defer func() { + globalVolumePlugin = oldGlobalPlugin + }() + + // Call CreateActor RPC directly + _, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{ + Actor: &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "fail-actor"}, + ActorTemplateNamespace: ns, + ActorTemplateName: "tmpl1", + }, + }) + if err != nil { + t.Fatalf("expected CreateActor to succeed, got: %v", err) + } + + // Call ResumeActor RPC, which should trigger volume provisioning and fail on fail-vol2 + _, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "fail-actor"}, + }) + if err == nil { + t.Fatalf("expected ResumeActor to fail due to volume creation error, but it succeeded") + } + + // Verify GetActor returns the actor in STATUS_SUSPENDED status + getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "fail-actor"}, + }) + if err != nil { + t.Fatalf("GetActor failed: %v", err) + } + if getResp.GetStatus() != ateapipb.Actor_STATUS_SUSPENDED { + t.Errorf("actor status = %v, want %v", getResp.GetStatus(), ateapipb.Actor_STATUS_SUSPENDED) + } + + actorUID := getResp.GetMetadata().GetUid() + if actorUID == "" { + t.Fatalf("expected non-empty UID on actor") + } + + // Verify that succ-vol1 was updated to CREATED with a storageVolumeId, and fail-vol2 is still CREATING + if len(getResp.GetActorVolumes()) != 2 { + t.Fatalf("expected 2 volumes on actor, got %d", len(getResp.GetActorVolumes())) + } + volsByName := make(map[string]*ateapipb.ExternalVolume) + for _, v := range getResp.GetActorVolumes() { + volsByName[v.GetVolumeName()] = v + } + if v1, ok := volsByName["succ-vol1"]; !ok || v1.GetStatus() != ateapipb.ExternalVolume_CREATED || v1.GetStorageVolumeId() == "" { + t.Errorf("succ-vol1 unexpected state: %v", v1) + } + if v2, ok := volsByName["fail-vol2"]; !ok || v2.GetStatus() != ateapipb.ExternalVolume_CREATING { + t.Errorf("fail-vol2 unexpected state: %v", v2) + } + + // Call DeleteActor on the actor in STATUS_SUSPENDED + _, err = tc.client.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "fail-actor"}, + }) + if err != nil { + t.Fatalf("DeleteActor failed: %v", err) + } + + // Verify both volumes were deleted (succ-vol1 via storageID, fail-vol2 via fallback actorVolumeID) + wantDeleted := []string{ + "storage-substrate-" + actorUID + "-succ-vol1", + "substrate-" + actorUID + "-fail-vol2", + } + if diff := cmp.Diff(wantDeleted, plugin.deleted); diff != "" { + t.Errorf("deleted volume IDs mismatch (-want +got):\n%s", diff) + } + + // Confirm GetActor returns NotFound after deletion + _, err = tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "fail-actor"}, + }) + if status.Code(err) != codes.NotFound { + t.Errorf("GetActor after DeleteActor err = %v, want NotFound", err) + } +} + +type retrySuccessVolumePlugin struct { + volume.VolumePluginControlPlane + mu sync.Mutex + attempts int + deleted []string +} + +func (r *retrySuccessVolumePlugin) CreateVolume(ctx context.Context, name, capacity, storageClass string) (string, error) { + r.mu.Lock() + defer r.mu.Unlock() + if strings.HasSuffix(name, "retry-vol2") { + r.attempts++ + if r.attempts == 1 { + return "", fmt.Errorf("simulated temporary volume creation failure") + } + } + return "storage-" + name, nil +} + +func (r *retrySuccessVolumePlugin) AttachVolume(ctx context.Context, volumeID, node string) error { + return nil +} + +func (r *retrySuccessVolumePlugin) DetachVolume(ctx context.Context, volumeID, node string) error { + return nil +} + +func (r *retrySuccessVolumePlugin) DeleteVolume(ctx context.Context, volumeID string) error { + r.mu.Lock() + defer r.mu.Unlock() + r.deleted = append(r.deleted, volumeID) + return nil +} + +// TestResumeActor_VolumeCreationRetrySuccess tests that when volume provisioning fails on the first ResumeActor call, +// a subsequent call to ResumeActor retries provisioning only the pending volumes and succeeds. +func TestResumeActor_VolumeCreationRetrySuccess(t *testing.T) { + ns := namespaceForTest("ns-resume-vol-retry") + tc := setupTest(t, ns) + defer tc.cleanup() + + volumes := []atev1alpha1.Volume{ + { + Name: "succ-vol1", + VolumeSource: atev1alpha1.VolumeSource{ + ExternalVolumeTemplate: &atev1alpha1.ExternalVolumeTemplate{ + StorageClassName: "standard", + Capacity: resource.MustParse("10Gi"), + }, + }, + }, + { + Name: "retry-vol2", + VolumeSource: atev1alpha1.VolumeSource{ + ExternalVolumeTemplate: &atev1alpha1.ExternalVolumeTemplate{ + StorageClassName: "standard", + Capacity: resource.MustParse("10Gi"), + }, + }, + }, + } + retryMounts := []atev1alpha1.VolumeMount{ + {Name: "succ-vol1", MountPath: "/mnt/vol1"}, + {Name: "retry-vol2", MountPath: "/mnt/vol2"}, + } + createTemplateWithVolumes(t, tc, ns, volumes, retryMounts) + createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1") + + plugin := &retrySuccessVolumePlugin{} + oldGlobalPlugin := globalVolumePlugin + globalVolumePlugin = plugin + defer func() { + globalVolumePlugin = oldGlobalPlugin + }() + + // Call CreateActor RPC directly + _, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{ + Actor: &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "retry-actor"}, + ActorTemplateNamespace: ns, + ActorTemplateName: "tmpl1", + }, + }) + if err != nil { + t.Fatalf("expected CreateActor to succeed, got: %v", err) + } + + // First call to ResumeActor RPC, which should fail on retry-vol2 (attempt 1) + _, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "retry-actor"}, + }) + if err == nil { + t.Fatalf("expected first ResumeActor to fail due to temporary volume creation error, but it succeeded") + } + + // Verify GetActor returns the actor in STATUS_SUSPENDED status with succ-vol1 created and retry-vol2 creating + getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "retry-actor"}, + }) + if err != nil { + t.Fatalf("GetActor after first resume failed: %v", err) + } + if getResp.GetStatus() != ateapipb.Actor_STATUS_SUSPENDED { + t.Errorf("actor status after first resume = %v, want %v", getResp.GetStatus(), ateapipb.Actor_STATUS_SUSPENDED) + } + + volsByName := make(map[string]*ateapipb.ExternalVolume) + for _, v := range getResp.GetActorVolumes() { + volsByName[v.GetVolumeName()] = v + } + if v1, ok := volsByName["succ-vol1"]; !ok || v1.GetStatus() != ateapipb.ExternalVolume_CREATED || v1.GetStorageVolumeId() == "" { + t.Errorf("succ-vol1 unexpected state after first resume: %v", v1) + } + if v2, ok := volsByName["retry-vol2"]; !ok || v2.GetStatus() != ateapipb.ExternalVolume_CREATING { + t.Errorf("retry-vol2 unexpected state after first resume: %v", v2) + } + + // Second call to ResumeActor RPC, which should succeed on retry-vol2 (attempt 2) + _, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "retry-actor"}, + }) + if err != nil { + t.Fatalf("expected second ResumeActor to succeed, got: %v", err) + } + + // Verify GetActor returns the actor in STATUS_RUNNING status with both volumes CREATED + getResp, err = tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "retry-actor"}, + }) + if err != nil { + t.Fatalf("GetActor after second resume failed: %v", err) + } + if getResp.GetStatus() != ateapipb.Actor_STATUS_RUNNING { + t.Errorf("actor status after second resume = %v, want %v", getResp.GetStatus(), ateapipb.Actor_STATUS_RUNNING) + } + for _, v := range getResp.GetActorVolumes() { + if v.GetStatus() != ateapipb.ExternalVolume_CREATED || v.GetStorageVolumeId() == "" { + t.Errorf("volume %s unexpected state after second resume: %v", v.GetVolumeName(), v) + } + } + + // Clean up by suspending and deleting the actor + _, err = tc.client.SuspendActor(context.Background(), &ateapipb.SuspendActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "retry-actor"}, + }) + if err != nil { + t.Fatalf("SuspendActor failed: %v", err) + } + _, err = tc.client.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "retry-actor"}, + }) + if err != nil { + t.Fatalf("DeleteActor failed: %v", err) + } +} + // TestCreateActor_TemplateNotFound tests that creating an actor with a non-existent template fails with FailedPrecondition. func TestCreateActor_TemplateNotFound(t *testing.T) { ns := namespaceForTest("ns-create-notfound") @@ -2328,7 +2824,7 @@ func TestDeleteActor_NotSuspended(t *testing.T) { _, err = tc.client.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{ Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "id1"}, }) - assertGrpcError(t, err, codes.FailedPrecondition, "Actor test-atespace/id1 is not suspended (status: STATUS_RUNNING)") + assertGrpcError(t, err, codes.FailedPrecondition, "Actor test-atespace/id1 is not in a deletable status (status: STATUS_RUNNING)") } func TestDeleteActor_Crashed(t *testing.T) { @@ -2362,8 +2858,8 @@ func TestDeleteActor_Crashed(t *testing.T) { if err != nil { t.Fatalf("DeleteActor of crashed actor failed: %v", err) } - if got := deleted.GetStatus(); got != ateapipb.Actor_STATUS_CRASHED { - t.Errorf("deleted actor status = %v, want %v", got, ateapipb.Actor_STATUS_CRASHED) + if got := deleted.GetStatus(); got != ateapipb.Actor_STATUS_DELETING { + t.Errorf("deleted actor status = %v, want %v", got, ateapipb.Actor_STATUS_DELETING) } _, err = tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ diff --git a/cmd/ateapi/internal/controlapi/service.go b/cmd/ateapi/internal/controlapi/service.go index 7ec695b00..944ae441f 100644 --- a/cmd/ateapi/internal/controlapi/service.go +++ b/cmd/ateapi/internal/controlapi/service.go @@ -51,6 +51,5 @@ func NewService( dialer: dialer, actorWorkflow: NewActorWorkflow(persistence, workerCache, dialer, actorTemplateLister, workerPoolLister, sandboxConfigLister, kubeClient), } - return s } diff --git a/cmd/ateapi/internal/controlapi/volumes.go b/cmd/ateapi/internal/controlapi/volumes.go index 161c74e8c..302d0ae28 100644 --- a/cmd/ateapi/internal/controlapi/volumes.go +++ b/cmd/ateapi/internal/controlapi/volumes.go @@ -29,7 +29,7 @@ import ( ) var ( - globalVolumePlugin = volume.NewMockVolumePlugin() + globalVolumePlugin volume.VolumePluginControlPlane = volume.NewMockVolumePlugin() ) // TODO: Replace with actual volume plugin search @@ -37,46 +37,100 @@ func getVolumePlugin() volume.VolumePluginControlPlane { return globalVolumePlugin } -// TODO: we should persist creation first so that we can handle background cleanup. -// this probably requires us to add a PROVISIONING actor state. - -// createActorVolumes provisions external volumes specified in the actor template. -// It returns the list of created external volumes, or an error if any creation fails. -// If any volume creation fails, it cleans up any volumes created in this call on a best-effort basis. -func (s *Service) createActorVolumes(ctx context.Context, ref *ateapipb.ObjectRef, template *atev1alpha1.ActorTemplate) ([]*ateapipb.ExternalVolume, error) { +// initialActorVolumes constructs initial volume objects in CREATING state before volume creation. +func initialActorVolumes(template *atev1alpha1.ActorTemplate) []*ateapipb.ExternalVolume { var volumes []*ateapipb.ExternalVolume for _, vol := range template.Spec.Volumes { if vol.ExternalVolumeTemplate != nil { - // Use a unique name for the volume to ensure idempotency - uniqueVolName := actorVolumeID(ref, vol.Name) - storageVolumeID, err := getVolumePlugin().CreateVolume(ctx, uniqueVolName, vol.ExternalVolumeTemplate.Capacity.String(), vol.ExternalVolumeTemplate.StorageClassName) - if err != nil { - // TODO: need better system - best effort cleanup of already created volumes - _ = s.deleteActorVolumes(ctx, ref, volumes) - return nil, status.Errorf(codes.Internal, "failed to create volume %q: %v", vol.Name, err) - } volumes = append(volumes, &ateapipb.ExternalVolume{ - ActorVolumeId: uniqueVolName, - StorageVolumeId: storageVolumeID, - VolumeType: "mock", // TODO fix when we support multiple plugins - Status: ateapipb.ExternalVolume_CREATED, + VolumeName: vol.Name, + VolumeType: "mock", // TODO fix when we support multiple plugins + Status: ateapipb.ExternalVolume_CREATING, }) } } - return volumes, nil + return volumes +} + +// createActorVolumes provisions external volumes specified in volumesToCreate using the provided volume plugin. +// It returns the list of external volumes (with updated status and storage IDs), or an error if any creation fails. +// Any volumes processed before or during a failure are returned alongside the error so they can be persisted on the actor. +func createActorVolumes(ctx context.Context, actorUID string, template *atev1alpha1.ActorTemplate, volumesToCreate []*ateapipb.ExternalVolume) (resultVolumes []*ateapipb.ExternalVolume, err error) { + resultVolumes = make([]*ateapipb.ExternalVolume, 0, len(volumesToCreate)) + + var currentIdx int + defer func() { + if err != nil { + // If we encounter an error, append the rest of the volumes to the result with the last + // known state. This allows the caller to persist the state of the volumes. + resultVolumes = append(resultVolumes, volumesToCreate[currentIdx:]...) + } + }() + + for idx, vol := range volumesToCreate { + currentIdx = idx + + var specVol *atev1alpha1.Volume + volName := vol.GetVolumeName() + for i := range template.Spec.Volumes { + if template.Spec.Volumes[i].Name == volName { + specVol = &template.Spec.Volumes[i] + break + } + } + if specVol == nil || specVol.ExternalVolumeTemplate == nil { + return resultVolumes, status.Errorf(codes.NotFound, "volume %q not found in template", volName) + } + + switch vol.GetStatus() { + case ateapipb.ExternalVolume_CREATING: + // proceed with volume creation + case ateapipb.ExternalVolume_CREATED: + resultVolumes = append(resultVolumes, vol) + continue + case ateapipb.ExternalVolume_DELETING: + return resultVolumes, status.Errorf(codes.FailedPrecondition, "cannot create volume %q in DELETING status", volName) + default: + return resultVolumes, status.Errorf(codes.Internal, "unexpected status %s for volume %q", vol.GetStatus(), volName) + } + + actVolID := actorVolumeID(actorUID, volName) + + plugin := getVolumePlugin() + if plugin == nil { + return resultVolumes, status.Errorf(codes.Internal, "volume plugin is not configured for creating volume %q", volName) + } + storageVolumeID, volErr := plugin.CreateVolume(ctx, actVolID, specVol.ExternalVolumeTemplate.Capacity.String(), specVol.ExternalVolumeTemplate.StorageClassName) + if volErr != nil { + return resultVolumes, status.Errorf(codes.Internal, "failed to create volume %q: %v", specVol.Name, volErr) + } + + resultVolumes = append(resultVolumes, &ateapipb.ExternalVolume{ + VolumeName: volName, + StorageVolumeId: storageVolumeID, + VolumeType: vol.GetVolumeType(), + Status: ateapipb.ExternalVolume_CREATED, + }) + } + return resultVolumes, nil } // deleteActorVolumes deletes all external volumes in the list. -func (s *Service) deleteActorVolumes(ctx context.Context, ref *ateapipb.ObjectRef, volumes []*ateapipb.ExternalVolume) error { +func (s *Service) deleteActorVolumes(ctx context.Context, actorUID string, volumes []*ateapipb.ExternalVolume) error { + if actorUID == "" { + return errors.New("actorUID is required") + } var errs []error for _, vol := range volumes { - if err := getVolumePlugin().DeleteVolume(ctx, vol.GetStorageVolumeId()); err != nil { - slog.ErrorContext(ctx, "failed to delete volume", - slog.String("atespace", ref.GetAtespace()), - slog.String("actor_id", ref.GetName()), - slog.String("volume_id", vol.GetStorageVolumeId()), - slog.Any("error", err)) - errs = append(errs, fmt.Errorf("failed to delete volume %q: %w", vol.GetStorageVolumeId(), err)) + volID := vol.GetStorageVolumeId() + if volID == "" { + // If the volume hasn't been successfully created yet, it's possible + // that it doesn't have a storage volume ID. In that case, fallback + // to the original requested volID. + volID = actorVolumeID(actorUID, vol.GetVolumeName()) + } + if err := getVolumePlugin().DeleteVolume(ctx, volID); err != nil { + errs = append(errs, fmt.Errorf("failed to delete volume %q: %w", volID, err)) } } return errors.Join(errs...) @@ -89,8 +143,7 @@ func getMountedActorVolumes(ctx context.Context, ref *ateapipb.ObjectRef, volume // Find the corresponding volume in the ActorTemplate to check if it's mounted var matchedTemplateVol *atev1alpha1.Volume for _, tVol := range template.Spec.Volumes { - expectedID := actorVolumeID(ref, tVol.Name) - if vol.GetActorVolumeId() == expectedID { + if vol.GetVolumeName() == tVol.Name { matchedTemplateVol = &tVol break } @@ -110,9 +163,8 @@ func getMountedActorVolumes(ctx context.Context, ref *ateapipb.ObjectRef, volume return mounted } -func actorVolumeID(ref *ateapipb.ObjectRef, volumeName string) string { - // TODO consider if this should be actor UUID - return fmt.Sprintf("%s-%s-%s", ref.GetAtespace(), ref.GetName(), volumeName) +func actorVolumeID(actorUID string, volumeName string) string { + return fmt.Sprintf("substrate-%s-%s", actorUID, volumeName) } // detachActorVolumes detaches all mounted external volumes for an actor from its worker node. diff --git a/cmd/ateapi/internal/controlapi/volumes_test.go b/cmd/ateapi/internal/controlapi/volumes_test.go new file mode 100644 index 000000000..de2f8562b --- /dev/null +++ b/cmd/ateapi/internal/controlapi/volumes_test.go @@ -0,0 +1,304 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package controlapi + +import ( + "context" + "testing" + + "github.com/google/go-cmp/cmp" + "google.golang.org/protobuf/testing/protocmp" + + "github.com/agent-substrate/substrate/internal/volume" + atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" +) + +func TestInitialActorVolumes_CreatingState(t *testing.T) { + tmpl := &atev1alpha1.ActorTemplate{ + Spec: atev1alpha1.ActorTemplateSpec{ + Volumes: []atev1alpha1.Volume{ + { + Name: "data-vol-1", + VolumeSource: atev1alpha1.VolumeSource{ + ExternalVolumeTemplate: &atev1alpha1.ExternalVolumeTemplate{ + StorageClassName: "standard", + }, + }, + }, + { + Name: "scratch-vol", + }, + { + Name: "durable-vol", + VolumeSource: atev1alpha1.VolumeSource{ + DurableDir: &atev1alpha1.DurableDirVolumeSource{}, + }, + }, + { + Name: "data-vol-2", + VolumeSource: atev1alpha1.VolumeSource{ + ExternalVolumeTemplate: &atev1alpha1.ExternalVolumeTemplate{ + StorageClassName: "fast", + }, + }, + }, + }, + }, + } + + want := []*ateapipb.ExternalVolume{ + { + VolumeName: "data-vol-1", + VolumeType: "mock", + Status: ateapipb.ExternalVolume_CREATING, + }, + { + VolumeName: "data-vol-2", + VolumeType: "mock", + Status: ateapipb.ExternalVolume_CREATING, + }, + } + + initVols := initialActorVolumes(tmpl) + if diff := cmp.Diff(want, initVols, protocmp.Transform()); diff != "" { + t.Errorf("initialActorVolumes mismatch (-want +got):\n%s", diff) + } +} + +func TestCreateActorVolumes(t *testing.T) { + ctx := context.Background() + + standardTmpl := &atev1alpha1.ActorTemplate{ + Spec: atev1alpha1.ActorTemplateSpec{ + Volumes: []atev1alpha1.Volume{ + { + Name: "data-vol", + VolumeSource: atev1alpha1.VolumeSource{ + ExternalVolumeTemplate: &atev1alpha1.ExternalVolumeTemplate{ + StorageClassName: "standard", + }, + }, + }, + }, + }, + } + + multiVolTmpl := &atev1alpha1.ActorTemplate{ + Spec: atev1alpha1.ActorTemplateSpec{ + Volumes: []atev1alpha1.Volume{ + { + Name: "vol1", + VolumeSource: atev1alpha1.VolumeSource{ + ExternalVolumeTemplate: &atev1alpha1.ExternalVolumeTemplate{ + StorageClassName: "standard", + }, + }, + }, + { + Name: "vol2", + VolumeSource: atev1alpha1.VolumeSource{ + ExternalVolumeTemplate: &atev1alpha1.ExternalVolumeTemplate{ + StorageClassName: "standard", + }, + }, + }, + { + Name: "vol3", + VolumeSource: atev1alpha1.VolumeSource{ + ExternalVolumeTemplate: &atev1alpha1.ExternalVolumeTemplate{ + StorageClassName: "standard", + }, + }, + }, + }, + }, + } + + tests := []struct { + name string + tmpl *atev1alpha1.ActorTemplate + inputVolumes []*ateapipb.ExternalVolume + wantErr bool + wantRes []*ateapipb.ExternalVolume + }{ + { + name: "partial failure returns error and preserves succeeded, failed, and remaining volumes", + tmpl: multiVolTmpl, + inputVolumes: []*ateapipb.ExternalVolume{ + { + VolumeName: "vol1", + VolumeType: "mock", + Status: ateapipb.ExternalVolume_CREATING, + }, + { + VolumeName: "vol2", + Status: ateapipb.ExternalVolume_DELETING, + }, + { + VolumeName: "vol3", + Status: ateapipb.ExternalVolume_CREATING, + }, + }, + wantErr: true, + wantRes: []*ateapipb.ExternalVolume{ + { + VolumeName: "vol1", + StorageVolumeId: "mock-vol-1", + VolumeType: "mock", + Status: ateapipb.ExternalVolume_CREATED, + }, + { + VolumeName: "vol2", + Status: ateapipb.ExternalVolume_DELETING, + }, + { + VolumeName: "vol3", + Status: ateapipb.ExternalVolume_CREATING, + }, + }, + }, + { + name: "created volume status succeeds", + tmpl: standardTmpl, + inputVolumes: []*ateapipb.ExternalVolume{ + { + VolumeName: "data-vol", + StorageVolumeId: "existing-vol-id", + Status: ateapipb.ExternalVolume_CREATED, + }, + }, + wantErr: false, + wantRes: []*ateapipb.ExternalVolume{ + { + VolumeName: "data-vol", + StorageVolumeId: "existing-vol-id", + Status: ateapipb.ExternalVolume_CREATED, + }, + }, + }, + { + name: "unknown volume status returns error", + tmpl: standardTmpl, + inputVolumes: []*ateapipb.ExternalVolume{ + { + VolumeName: "data-vol", + Status: ateapipb.ExternalVolume_STATUS_UNKNOWN, + }, + }, + wantErr: true, + wantRes: []*ateapipb.ExternalVolume{ + { + VolumeName: "data-vol", + Status: ateapipb.ExternalVolume_STATUS_UNKNOWN, + }, + }, + }, + { + name: "volume not found in template returns error", + tmpl: &atev1alpha1.ActorTemplate{ + Spec: atev1alpha1.ActorTemplateSpec{ + Volumes: []atev1alpha1.Volume{}, + }, + }, + inputVolumes: []*ateapipb.ExternalVolume{ + { + VolumeName: "missing-vol", + Status: ateapipb.ExternalVolume_CREATING, + }, + }, + wantErr: true, + wantRes: []*ateapipb.ExternalVolume{ + { + VolumeName: "missing-vol", + Status: ateapipb.ExternalVolume_CREATING, + }, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + globalVolumePlugin = volume.NewMockVolumePlugin() + res, err := createActorVolumes(ctx, "actor-uid-123", tt.tmpl, tt.inputVolumes) + if (err != nil) != tt.wantErr { + t.Errorf("createActorVolumes() error = %v, wantErr %v", err, tt.wantErr) + } + if diff := cmp.Diff(tt.wantRes, res, protocmp.Transform()); diff != "" { + t.Errorf("createActorVolumes() mismatch (-want +got):\n%s", diff) + } + }) + } +} + +type trackingVolumePlugin struct { + volume.VolumePluginControlPlane + deletedIDs []string +} + +func (t *trackingVolumePlugin) DeleteVolume(ctx context.Context, volumeID string) error { + t.deletedIDs = append(t.deletedIDs, volumeID) + return nil +} + +func TestDeleteActorVolumes(t *testing.T) { + ctx := context.Background() + + tests := []struct { + name string + actorUID string + volumes []*ateapipb.ExternalVolume + wantDeleted []string + wantErr bool + }{ + { + name: "uses storage volume ID when present", + actorUID: "uid-abc", + volumes: []*ateapipb.ExternalVolume{ + {VolumeName: "vol1", StorageVolumeId: "storage-vol-123"}, + }, + wantDeleted: []string{"storage-vol-123"}, + wantErr: false, + }, + { + name: "falls back to actorVolumeID when storage volume ID is empty regardless of status", + actorUID: "uid-abc", + volumes: []*ateapipb.ExternalVolume{ + {VolumeName: "vol1", StorageVolumeId: "", Status: ateapipb.ExternalVolume_CREATED}, + }, + wantDeleted: []string{"substrate-uid-abc-vol1"}, + wantErr: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + plugin := &trackingVolumePlugin{} + oldGlobalPlugin := globalVolumePlugin + globalVolumePlugin = plugin + defer func() { globalVolumePlugin = oldGlobalPlugin }() + svc := &Service{} + + err := svc.deleteActorVolumes(ctx, tt.actorUID, tt.volumes) + if (err != nil) != tt.wantErr { + t.Fatalf("deleteActorVolumes() error = %v, wantErr %v", err, tt.wantErr) + } + + if diff := cmp.Diff(tt.wantDeleted, plugin.deletedIDs); diff != "" { + t.Errorf("deletedIDs mismatch (-want +got):\n%s", diff) + } + }) + } +} diff --git a/cmd/ateapi/internal/controlapi/workflow.go b/cmd/ateapi/internal/controlapi/workflow.go index f06efa7b6..14c1df9dc 100644 --- a/cmd/ateapi/internal/controlapi/workflow.go +++ b/cmd/ateapi/internal/controlapi/workflow.go @@ -180,6 +180,7 @@ func (w *ActorWorkflow) ResumeActor(ctx context.Context, actorRef resources.Acto steps := []WorkflowStep[*ResumeInput, *ResumeState]{ &LoadActorForResumeStep{store: w.store, actorTemplateLister: w.actorTemplateLister}, + &CreateVolumesStep{store: w.store}, &AssignWorkerStep{store: w.store, workerCache: w.workerCache, scheduler: w.scheduler}, &AttachVolumesStep{store: w.store}, &CallAteletRestoreStep{store: w.store, dialer: w.dialer, kubeClient: w.kubeClient, secretCache: w.secretCache, workerPoolLister: w.workerPoolLister, sandboxConfigLister: w.sandboxConfigLister, scheduler: w.scheduler}, diff --git a/cmd/ateapi/internal/controlapi/workflow_resume.go b/cmd/ateapi/internal/controlapi/workflow_resume.go index ebb955a37..22d44e5a5 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume.go @@ -124,6 +124,54 @@ func (s *LoadActorForResumeStep) Execute(ctx context.Context, input *ResumeInput func (s *LoadActorForResumeStep) RetryBackoff() *wait.Backoff { return nil } +// CreateVolumesStep provisions any initial actor volumes that are in CREATING state. +type CreateVolumesStep struct { + store store.Interface +} + +func (s *CreateVolumesStep) Name() string { return "CreateVolumes" } + +func (s *CreateVolumesStep) IsComplete(ctx context.Context, input *ResumeInput, state *ResumeState) (bool, error) { + for _, vol := range state.Actor.GetActorVolumes() { + if vol.GetStatus() == ateapipb.ExternalVolume_CREATING { + return false, nil + } + } + return true, nil +} + +func (s *CreateVolumesStep) CheckPrerequisite(ctx context.Context, input *ResumeInput, state *ResumeState) error { + if state.Actor == nil { + return fmt.Errorf("actor is required for CreateVolumesStep") + } + if state.ActorTemplate == nil { + return fmt.Errorf("actorTemplate is required for CreateVolumesStep") + } + return nil +} + +func (s *CreateVolumesStep) Execute(ctx context.Context, input *ResumeInput, state *ResumeState) error { + volumes, err := createActorVolumes(ctx, state.Actor.GetMetadata().GetUid(), state.ActorTemplate, state.Actor.GetActorVolumes()) + state.Actor.ActorVolumes = volumes + if err != nil { + // Even if volume creation failed, we still want to persist any updated volume state. + if updated, updateErr := s.store.UpdateActor(ctx, state.Actor, state.Actor.GetMetadata().GetVersion()); updateErr != nil { + slog.ErrorContext(ctx, "failed to update actor volumes on volume creation failure in resume", slog.Any("error", updateErr)) + } else { + state.Actor = updated + } + return err + } + updated, updateErr := s.store.UpdateActor(ctx, state.Actor, state.Actor.GetMetadata().GetVersion()) + if updateErr != nil { + return fmt.Errorf("while updating actor after volume creation: %w", updateErr) + } + state.Actor = updated + return nil +} + +func (s *CreateVolumesStep) RetryBackoff() *wait.Backoff { return nil } + type AssignWorkerStep struct { store store.Interface workerCache *workercache.Cache diff --git a/cmd/ateapi/internal/controlapi/workload_spec.go b/cmd/ateapi/internal/controlapi/workload_spec.go index f57bb2c2f..cce9e02bf 100644 --- a/cmd/ateapi/internal/controlapi/workload_spec.go +++ b/cmd/ateapi/internal/controlapi/workload_spec.go @@ -128,9 +128,8 @@ func appendExternalVolumes(workloadSpec *ateletpb.WorkloadSpec, template *atev1a var storageVolID string var volType string - expectedID := actorVolumeID(&ateapipb.ObjectRef{Atespace: actor.GetMetadata().GetAtespace(), Name: actor.GetMetadata().GetName()}, vol.Name) for _, dbVol := range actor.GetActorVolumes() { - if dbVol.GetActorVolumeId() == expectedID { + if dbVol.GetVolumeName() == vol.Name { storageVolID = dbVol.GetStorageVolumeId() volType = dbVol.GetVolumeType() break diff --git a/cmd/ateapi/internal/controlapi/workload_spec_test.go b/cmd/ateapi/internal/controlapi/workload_spec_test.go index a41efa984..d55d44562 100644 --- a/cmd/ateapi/internal/controlapi/workload_spec_test.go +++ b/cmd/ateapi/internal/controlapi/workload_spec_test.go @@ -516,7 +516,7 @@ func TestAppendExternalVolumes(t *testing.T) { }, ActorVolumes: []*ateapipb.ExternalVolume{ { - ActorVolumeId: "space-abc-actor-123-vol-1", + VolumeName: "vol-1", StorageVolumeId: "vol-gce-pd-123", VolumeType: "pd-standard", }, diff --git a/cmd/ateapi/internal/store/ateredis/ateredis.go b/cmd/ateapi/internal/store/ateredis/ateredis.go index e5ada1ff8..ae08e54bc 100644 --- a/cmd/ateapi/internal/store/ateredis/ateredis.go +++ b/cmd/ateapi/internal/store/ateredis/ateredis.go @@ -504,8 +504,7 @@ func (s *Persistence) DeleteActor(ctx context.Context, actorRef resources.ActorR return fmt.Errorf("in protojson.Unmarshal: %w", err) } - if currentActor.GetStatus() != ateapipb.Actor_STATUS_SUSPENDED && - currentActor.GetStatus() != ateapipb.Actor_STATUS_CRASHED { + if currentActor.GetStatus() != ateapipb.Actor_STATUS_DELETING { return store.ErrFailedPrecondition } diff --git a/cmd/ateapi/internal/store/ateredis/ateredis_test.go b/cmd/ateapi/internal/store/ateredis/ateredis_test.go index 931949f94..ee00d8b61 100644 --- a/cmd/ateapi/internal/store/ateredis/ateredis_test.go +++ b/cmd/ateapi/internal/store/ateredis/ateredis_test.go @@ -414,8 +414,9 @@ func TestDeleteActor(t *testing.T) { status ateapipb.Actor_Status wantErr error }{ - {name: "suspended", status: ateapipb.Actor_STATUS_SUSPENDED}, - {name: "crashed", status: ateapipb.Actor_STATUS_CRASHED}, + {name: "suspended", status: ateapipb.Actor_STATUS_SUSPENDED, wantErr: store.ErrFailedPrecondition}, + {name: "crashed", status: ateapipb.Actor_STATUS_CRASHED, wantErr: store.ErrFailedPrecondition}, + {name: "deleting", status: ateapipb.Actor_STATUS_DELETING}, {name: "running", status: ateapipb.Actor_STATUS_RUNNING, wantErr: store.ErrFailedPrecondition}, {name: "paused", status: ateapipb.Actor_STATUS_PAUSED, wantErr: store.ErrFailedPrecondition}, } @@ -1365,7 +1366,7 @@ func TestDeleteAtespace_EmptyAfterActorsRemoved(t *testing.T) { if _, err := s.CreateAtespace(ctx, newTestAtespace("team-a")); err != nil { t.Fatalf("CreateAtespace failed: %v", err) } - if _, err := s.CreateActor(ctx, &ateapipb.Actor{Metadata: &ateapipb.ResourceMetadata{Name: "id1", Atespace: "team-a"}, Status: ateapipb.Actor_STATUS_SUSPENDED}); err != nil { + if _, err := s.CreateActor(ctx, &ateapipb.Actor{Metadata: &ateapipb.ResourceMetadata{Name: "id1", Atespace: "team-a"}, Status: ateapipb.Actor_STATUS_DELETING}); err != nil { t.Fatalf("CreateActor failed: %v", err) } if _, err := s.DeleteAtespace(ctx, "team-a"); !errors.Is(err, store.ErrFailedPrecondition) { diff --git a/pkg/proto/ateapipb/ateapi.pb.go b/pkg/proto/ateapipb/ateapi.pb.go index ae7c136a4..802cf2aed 100644 --- a/pkg/proto/ateapipb/ateapi.pb.go +++ b/pkg/proto/ateapipb/ateapi.pb.go @@ -41,22 +41,28 @@ const ( type ExternalVolume_Status int32 const ( - ExternalVolume_PROVISIONING ExternalVolume_Status = 0 - ExternalVolume_CREATED ExternalVolume_Status = 1 - ExternalVolume_DELETING ExternalVolume_Status = 2 + ExternalVolume_STATUS_UNKNOWN ExternalVolume_Status = 0 + // Volume being created in the storage system. + ExternalVolume_CREATING ExternalVolume_Status = 1 + // Volume successfully created in the storage system. + ExternalVolume_CREATED ExternalVolume_Status = 2 + // Volume being deleted from the storage system. + ExternalVolume_DELETING ExternalVolume_Status = 3 ) // Enum value maps for ExternalVolume_Status. var ( ExternalVolume_Status_name = map[int32]string{ - 0: "PROVISIONING", - 1: "CREATED", - 2: "DELETING", + 0: "STATUS_UNKNOWN", + 1: "CREATING", + 2: "CREATED", + 3: "DELETING", } ExternalVolume_Status_value = map[string]int32{ - "PROVISIONING": 0, - "CREATED": 1, - "DELETING": 2, + "STATUS_UNKNOWN": 0, + "CREATING": 1, + "CREATED": 2, + "DELETING": 3, } ) @@ -98,6 +104,7 @@ const ( Actor_STATUS_PAUSING Actor_Status = 5 Actor_STATUS_PAUSED Actor_Status = 6 Actor_STATUS_CRASHED Actor_Status = 7 + Actor_STATUS_DELETING Actor_Status = 8 ) // Enum value maps for Actor_Status. @@ -111,6 +118,7 @@ var ( 5: "STATUS_PAUSING", 6: "STATUS_PAUSED", 7: "STATUS_CRASHED", + 8: "STATUS_DELETING", } Actor_Status_value = map[string]int32{ "STATUS_UNSPECIFIED": 0, @@ -121,6 +129,7 @@ var ( "STATUS_PAUSING": 5, "STATUS_PAUSED": 6, "STATUS_CRASHED": 7, + "STATUS_DELETING": 8, } ) @@ -521,9 +530,10 @@ func (x *ResourceMetadata) GetUpdateTime() *timestamppb.Timestamp { type ExternalVolume struct { state protoimpl.MessageState `protogen:"open.v1"` - // actor_id + the volume name specified in the actor template. - ActorVolumeId string `protobuf:"bytes,1,opt,name=actor_volume_id,json=actorVolumeId,proto3" json:"actor_volume_id,omitempty"` - // The volume_id returned from the storage system. + // Name of the volume specified in the actor template. + VolumeName string `protobuf:"bytes,1,opt,name=volume_name,json=volumeName,proto3" json:"volume_name,omitempty"` + // The globally unique volume_id returned from the storage system. + // This will be initially empty during volume creation StorageVolumeId string `protobuf:"bytes,2,opt,name=storage_volume_id,json=storageVolumeId,proto3" json:"storage_volume_id,omitempty"` // Internal volume plugin name or CSI driver name. VolumeType string `protobuf:"bytes,3,opt,name=volume_type,json=volumeType,proto3" json:"volume_type,omitempty"` @@ -562,9 +572,9 @@ func (*ExternalVolume) Descriptor() ([]byte, []int) { return file_ateapi_proto_rawDescGZIP(), []int{5} } -func (x *ExternalVolume) GetActorVolumeId() string { +func (x *ExternalVolume) GetVolumeName() string { if x != nil { - return x.ActorVolumeId + return x.VolumeName } return "" } @@ -587,7 +597,7 @@ func (x *ExternalVolume) GetStatus() ExternalVolume_Status { if x != nil { return x.Status } - return ExternalVolume_PROVISIONING + return ExternalVolume_STATUS_UNKNOWN } type Actor struct { @@ -617,7 +627,7 @@ type Actor struct { WorkerPoolName string `protobuf:"bytes,12,opt,name=worker_pool_name,json=workerPoolName,proto3" json:"worker_pool_name,omitempty"` // Volumes attached to the actor. These volumes only live as long as the actor. // They are deleted when the actor is deleted. - ActorVolumes []*ExternalVolume `protobuf:"bytes,16,rep,name=actor_volumes,json=actorVolumes,proto3" json:"actor_volumes,omitempty"` + ActorVolumes []*ExternalVolume `protobuf:"bytes,13,rep,name=actor_volumes,json=actorVolumes,proto3" json:"actor_volumes,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -2408,17 +2418,19 @@ const file_ateapi_proto_rawDesc = "" + "\vcreate_time\x18\x05 \x01(\v2\x1a.google.protobuf.TimestampR\n" + "createTime\x12;\n" + "\vupdate_time\x18\x06 \x01(\v2\x1a.google.protobuf.TimestampR\n" + - "updateTime\"\xf3\x01\n" + - "\x0eExternalVolume\x12&\n" + - "\x0factor_volume_id\x18\x01 \x01(\tR\ractorVolumeId\x12*\n" + + "updateTime\"\xfc\x01\n" + + "\x0eExternalVolume\x12\x1f\n" + + "\vvolume_name\x18\x01 \x01(\tR\n" + + "volumeName\x12*\n" + "\x11storage_volume_id\x18\x02 \x01(\tR\x0fstorageVolumeId\x12\x1f\n" + "\vvolume_type\x18\x03 \x01(\tR\n" + "volumeType\x125\n" + - "\x06status\x18\x04 \x01(\x0e2\x1d.ateapi.ExternalVolume.StatusR\x06status\"5\n" + - "\x06Status\x12\x10\n" + - "\fPROVISIONING\x10\x00\x12\v\n" + - "\aCREATED\x10\x01\x12\f\n" + - "\bDELETING\x10\x02\"\xc1\x06\n" + + "\x06status\x18\x04 \x01(\x0e2\x1d.ateapi.ExternalVolume.StatusR\x06status\"E\n" + + "\x06Status\x12\x12\n" + + "\x0eSTATUS_UNKNOWN\x10\x00\x12\f\n" + + "\bCREATING\x10\x01\x12\v\n" + + "\aCREATED\x10\x02\x12\f\n" + + "\bDELETING\x10\x03\"\xd6\x06\n" + "\x05Actor\x124\n" + "\bmetadata\x18\x01 \x01(\v2\x18.ateapi.ResourceMetadataR\bmetadata\x128\n" + "\x18actor_template_namespace\x18\x02 \x01(\tR\x16actorTemplateNamespace\x12.\n" + @@ -2434,7 +2446,7 @@ const file_ateapi_proto_rawDesc = "" + " \x01(\v2\x14.ateapi.SnapshotInfoR\x12latestSnapshotInfo\x129\n" + "\x0fworker_selector\x18\v \x01(\v2\x10.ateapi.SelectorR\x0eworkerSelector\x12(\n" + "\x10worker_pool_name\x18\f \x01(\tR\x0eworkerPoolName\x12;\n" + - "\ractor_volumes\x18\x10 \x03(\v2\x16.ateapi.ExternalVolumeR\factorVolumes\"\xb1\x01\n" + + "\ractor_volumes\x18\r \x03(\v2\x16.ateapi.ExternalVolumeR\factorVolumes\"\xc6\x01\n" + "\x06Status\x12\x16\n" + "\x12STATUS_UNSPECIFIED\x10\x00\x12\x13\n" + "\x0fSTATUS_RESUMING\x10\x01\x12\x12\n" + @@ -2443,7 +2455,8 @@ const file_ateapi_proto_rawDesc = "" + "\x10STATUS_SUSPENDED\x10\x04\x12\x12\n" + "\x0eSTATUS_PAUSING\x10\x05\x12\x11\n" + "\rSTATUS_PAUSED\x10\x06\x12\x12\n" + - "\x0eSTATUS_CRASHED\x10\a\"@\n" + + "\x0eSTATUS_CRASHED\x10\a\x12\x13\n" + + "\x0fSTATUS_DELETING\x10\b\"@\n" + "\bAtespace\x124\n" + "\bmetadata\x18\x01 \x01(\v2\x18.ateapi.ResourceMetadataR\bmetadata\";\n" + "\tObjectRef\x12\x1a\n" + diff --git a/pkg/proto/ateapipb/ateapi.proto b/pkg/proto/ateapipb/ateapi.proto index 873f994e9..5cb5b4dc7 100644 --- a/pkg/proto/ateapipb/ateapi.proto +++ b/pkg/proto/ateapipb/ateapi.proto @@ -115,19 +115,24 @@ message ResourceMetadata { } message ExternalVolume { - // actor_id + the volume name specified in the actor template. - string actor_volume_id = 1; + // Name of the volume specified in the actor template. + string volume_name = 1; - // The volume_id returned from the storage system. + // The globally unique volume_id returned from the storage system. + // This will be initially empty during volume creation string storage_volume_id = 2; // Internal volume plugin name or CSI driver name. string volume_type = 3; enum Status { - PROVISIONING = 0; - CREATED = 1; - DELETING = 2; + STATUS_UNKNOWN = 0; + // Volume being created in the storage system. + CREATING = 1; + // Volume successfully created in the storage system. + CREATED = 2; + // Volume being deleted from the storage system. + DELETING = 3; } Status status = 4; } @@ -148,6 +153,7 @@ message Actor { STATUS_PAUSING = 5; STATUS_PAUSED = 6; STATUS_CRASHED = 7; + STATUS_DELETING = 8; } Status status = 4; @@ -175,7 +181,7 @@ message Actor { // Volumes attached to the actor. These volumes only live as long as the actor. // They are deleted when the actor is deleted. - repeated ExternalVolume actor_volumes = 16; + repeated ExternalVolume actor_volumes = 13; } // Atespace is the isolation boundary an Actor is created into. Global-scoped: From 50c3aa13385371fc576bf6d1dd6c2ee7bded9733 Mon Sep 17 00:00:00 2001 From: msau42 Date: Tue, 28 Jul 2026 20:16:02 +0000 Subject: [PATCH 2/5] Rename ExternalVolume_CREATING to ExternalVolume_PENDING --- .../internal/controlapi/functional_test.go | 20 +++++++++---------- cmd/ateapi/internal/controlapi/volumes.go | 6 +++--- .../internal/controlapi/volumes_test.go | 16 +++++++-------- .../internal/controlapi/workflow_resume.go | 4 ++-- pkg/proto/ateapipb/ateapi.pb.go | 16 +++++++-------- pkg/proto/ateapipb/ateapi.proto | 4 ++-- 6 files changed, 33 insertions(+), 33 deletions(-) diff --git a/cmd/ateapi/internal/controlapi/functional_test.go b/cmd/ateapi/internal/controlapi/functional_test.go index 8793a82cb..3e4f05622 100644 --- a/cmd/ateapi/internal/controlapi/functional_test.go +++ b/cmd/ateapi/internal/controlapi/functional_test.go @@ -824,8 +824,8 @@ func TestCreateActor_WithExternalVolumes(t *testing.T) { if vol.GetVolumeName() != "ext-vol-1" { t.Errorf("volume name = %q, want %q", vol.GetVolumeName(), "ext-vol-1") } - if vol.GetStatus() != ateapipb.ExternalVolume_CREATING { - t.Errorf("volume status = %v, want %v", vol.GetStatus(), ateapipb.ExternalVolume_CREATING) + if vol.GetStatus() != ateapipb.ExternalVolume_PENDING { + t.Errorf("volume status = %v, want %v", vol.GetStatus(), ateapipb.ExternalVolume_PENDING) } if vol.GetStorageVolumeId() != "" { t.Errorf("expected empty storageVolumeId before resume, got %q", vol.GetStorageVolumeId()) @@ -841,8 +841,8 @@ func TestCreateActor_WithExternalVolumes(t *testing.T) { if len(getResp.GetActorVolumes()) != 1 { t.Fatalf("expected 1 volume in GetActor response, got %d", len(getResp.GetActorVolumes())) } - if getResp.GetActorVolumes()[0].GetStatus() != ateapipb.ExternalVolume_CREATING { - t.Errorf("GetActor status = %v, want %v", getResp.GetActorVolumes()[0].GetStatus(), ateapipb.ExternalVolume_CREATING) + if getResp.GetActorVolumes()[0].GetStatus() != ateapipb.ExternalVolume_PENDING { + t.Errorf("GetActor status = %v, want %v", getResp.GetActorVolumes()[0].GetStatus(), ateapipb.ExternalVolume_PENDING) } } @@ -885,8 +885,8 @@ func TestActorLifecycle_WithExternalVolumes(t *testing.T) { if createResp.GetStatus() != ateapipb.Actor_STATUS_SUSPENDED { t.Fatalf("expected initial status STATUS_SUSPENDED, got %v", createResp.GetStatus()) } - if len(createResp.GetActorVolumes()) != 1 || createResp.GetActorVolumes()[0].GetStatus() != ateapipb.ExternalVolume_CREATING { - t.Fatalf("expected 1 creating volume after CreateActor, got %v", createResp.GetActorVolumes()) + if len(createResp.GetActorVolumes()) != 1 || createResp.GetActorVolumes()[0].GetStatus() != ateapipb.ExternalVolume_PENDING { + t.Fatalf("expected 1 pending volume after CreateActor, got %v", createResp.GetActorVolumes()) } // 2. ResumeActor @@ -1063,7 +1063,7 @@ func TestResumeActor_VolumeCreationFailure(t *testing.T) { t.Fatalf("expected non-empty UID on actor") } - // Verify that succ-vol1 was updated to CREATED with a storageVolumeId, and fail-vol2 is still CREATING + // Verify that succ-vol1 was updated to CREATED with a storageVolumeId, and fail-vol2 is still PENDING if len(getResp.GetActorVolumes()) != 2 { t.Fatalf("expected 2 volumes on actor, got %d", len(getResp.GetActorVolumes())) } @@ -1074,7 +1074,7 @@ func TestResumeActor_VolumeCreationFailure(t *testing.T) { if v1, ok := volsByName["succ-vol1"]; !ok || v1.GetStatus() != ateapipb.ExternalVolume_CREATED || v1.GetStorageVolumeId() == "" { t.Errorf("succ-vol1 unexpected state: %v", v1) } - if v2, ok := volsByName["fail-vol2"]; !ok || v2.GetStatus() != ateapipb.ExternalVolume_CREATING { + if v2, ok := volsByName["fail-vol2"]; !ok || v2.GetStatus() != ateapipb.ExternalVolume_PENDING { t.Errorf("fail-vol2 unexpected state: %v", v2) } @@ -1199,7 +1199,7 @@ func TestResumeActor_VolumeCreationRetrySuccess(t *testing.T) { t.Fatalf("expected first ResumeActor to fail due to temporary volume creation error, but it succeeded") } - // Verify GetActor returns the actor in STATUS_SUSPENDED status with succ-vol1 created and retry-vol2 creating + // Verify GetActor returns the actor in STATUS_SUSPENDED status with succ-vol1 created and retry-vol2 pending getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "retry-actor"}, }) @@ -1217,7 +1217,7 @@ func TestResumeActor_VolumeCreationRetrySuccess(t *testing.T) { if v1, ok := volsByName["succ-vol1"]; !ok || v1.GetStatus() != ateapipb.ExternalVolume_CREATED || v1.GetStorageVolumeId() == "" { t.Errorf("succ-vol1 unexpected state after first resume: %v", v1) } - if v2, ok := volsByName["retry-vol2"]; !ok || v2.GetStatus() != ateapipb.ExternalVolume_CREATING { + if v2, ok := volsByName["retry-vol2"]; !ok || v2.GetStatus() != ateapipb.ExternalVolume_PENDING { t.Errorf("retry-vol2 unexpected state after first resume: %v", v2) } diff --git a/cmd/ateapi/internal/controlapi/volumes.go b/cmd/ateapi/internal/controlapi/volumes.go index 302d0ae28..0b99038e0 100644 --- a/cmd/ateapi/internal/controlapi/volumes.go +++ b/cmd/ateapi/internal/controlapi/volumes.go @@ -37,7 +37,7 @@ func getVolumePlugin() volume.VolumePluginControlPlane { return globalVolumePlugin } -// initialActorVolumes constructs initial volume objects in CREATING state before volume creation. +// initialActorVolumes constructs initial volume objects in PENDING state before volume creation. func initialActorVolumes(template *atev1alpha1.ActorTemplate) []*ateapipb.ExternalVolume { var volumes []*ateapipb.ExternalVolume for _, vol := range template.Spec.Volumes { @@ -45,7 +45,7 @@ func initialActorVolumes(template *atev1alpha1.ActorTemplate) []*ateapipb.Extern volumes = append(volumes, &ateapipb.ExternalVolume{ VolumeName: vol.Name, VolumeType: "mock", // TODO fix when we support multiple plugins - Status: ateapipb.ExternalVolume_CREATING, + Status: ateapipb.ExternalVolume_PENDING, }) } } @@ -83,7 +83,7 @@ func createActorVolumes(ctx context.Context, actorUID string, template *atev1alp } switch vol.GetStatus() { - case ateapipb.ExternalVolume_CREATING: + case ateapipb.ExternalVolume_PENDING: // proceed with volume creation case ateapipb.ExternalVolume_CREATED: resultVolumes = append(resultVolumes, vol) diff --git a/cmd/ateapi/internal/controlapi/volumes_test.go b/cmd/ateapi/internal/controlapi/volumes_test.go index de2f8562b..c70b37c46 100644 --- a/cmd/ateapi/internal/controlapi/volumes_test.go +++ b/cmd/ateapi/internal/controlapi/volumes_test.go @@ -26,7 +26,7 @@ import ( "github.com/agent-substrate/substrate/pkg/proto/ateapipb" ) -func TestInitialActorVolumes_CreatingState(t *testing.T) { +func TestInitialActorVolumes_PendingState(t *testing.T) { tmpl := &atev1alpha1.ActorTemplate{ Spec: atev1alpha1.ActorTemplateSpec{ Volumes: []atev1alpha1.Volume{ @@ -63,12 +63,12 @@ func TestInitialActorVolumes_CreatingState(t *testing.T) { { VolumeName: "data-vol-1", VolumeType: "mock", - Status: ateapipb.ExternalVolume_CREATING, + Status: ateapipb.ExternalVolume_PENDING, }, { VolumeName: "data-vol-2", VolumeType: "mock", - Status: ateapipb.ExternalVolume_CREATING, + Status: ateapipb.ExternalVolume_PENDING, }, } @@ -141,7 +141,7 @@ func TestCreateActorVolumes(t *testing.T) { { VolumeName: "vol1", VolumeType: "mock", - Status: ateapipb.ExternalVolume_CREATING, + Status: ateapipb.ExternalVolume_PENDING, }, { VolumeName: "vol2", @@ -149,7 +149,7 @@ func TestCreateActorVolumes(t *testing.T) { }, { VolumeName: "vol3", - Status: ateapipb.ExternalVolume_CREATING, + Status: ateapipb.ExternalVolume_PENDING, }, }, wantErr: true, @@ -166,7 +166,7 @@ func TestCreateActorVolumes(t *testing.T) { }, { VolumeName: "vol3", - Status: ateapipb.ExternalVolume_CREATING, + Status: ateapipb.ExternalVolume_PENDING, }, }, }, @@ -216,14 +216,14 @@ func TestCreateActorVolumes(t *testing.T) { inputVolumes: []*ateapipb.ExternalVolume{ { VolumeName: "missing-vol", - Status: ateapipb.ExternalVolume_CREATING, + Status: ateapipb.ExternalVolume_PENDING, }, }, wantErr: true, wantRes: []*ateapipb.ExternalVolume{ { VolumeName: "missing-vol", - Status: ateapipb.ExternalVolume_CREATING, + Status: ateapipb.ExternalVolume_PENDING, }, }, }, diff --git a/cmd/ateapi/internal/controlapi/workflow_resume.go b/cmd/ateapi/internal/controlapi/workflow_resume.go index 22d44e5a5..6948aaf9a 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume.go @@ -124,7 +124,7 @@ func (s *LoadActorForResumeStep) Execute(ctx context.Context, input *ResumeInput func (s *LoadActorForResumeStep) RetryBackoff() *wait.Backoff { return nil } -// CreateVolumesStep provisions any initial actor volumes that are in CREATING state. +// CreateVolumesStep provisions any initial actor volumes that are in PENDING state. type CreateVolumesStep struct { store store.Interface } @@ -133,7 +133,7 @@ func (s *CreateVolumesStep) Name() string { return "CreateVolumes" } func (s *CreateVolumesStep) IsComplete(ctx context.Context, input *ResumeInput, state *ResumeState) (bool, error) { for _, vol := range state.Actor.GetActorVolumes() { - if vol.GetStatus() == ateapipb.ExternalVolume_CREATING { + if vol.GetStatus() == ateapipb.ExternalVolume_PENDING { return false, nil } } diff --git a/pkg/proto/ateapipb/ateapi.pb.go b/pkg/proto/ateapipb/ateapi.pb.go index 802cf2aed..659a9a95c 100644 --- a/pkg/proto/ateapipb/ateapi.pb.go +++ b/pkg/proto/ateapipb/ateapi.pb.go @@ -42,8 +42,8 @@ type ExternalVolume_Status int32 const ( ExternalVolume_STATUS_UNKNOWN ExternalVolume_Status = 0 - // Volume being created in the storage system. - ExternalVolume_CREATING ExternalVolume_Status = 1 + // Volume creation pending in the storage system. + ExternalVolume_PENDING ExternalVolume_Status = 1 // Volume successfully created in the storage system. ExternalVolume_CREATED ExternalVolume_Status = 2 // Volume being deleted from the storage system. @@ -54,13 +54,13 @@ const ( var ( ExternalVolume_Status_name = map[int32]string{ 0: "STATUS_UNKNOWN", - 1: "CREATING", + 1: "PENDING", 2: "CREATED", 3: "DELETING", } ExternalVolume_Status_value = map[string]int32{ "STATUS_UNKNOWN": 0, - "CREATING": 1, + "PENDING": 1, "CREATED": 2, "DELETING": 3, } @@ -2418,17 +2418,17 @@ const file_ateapi_proto_rawDesc = "" + "\vcreate_time\x18\x05 \x01(\v2\x1a.google.protobuf.TimestampR\n" + "createTime\x12;\n" + "\vupdate_time\x18\x06 \x01(\v2\x1a.google.protobuf.TimestampR\n" + - "updateTime\"\xfc\x01\n" + + "updateTime\"\xfb\x01\n" + "\x0eExternalVolume\x12\x1f\n" + "\vvolume_name\x18\x01 \x01(\tR\n" + "volumeName\x12*\n" + "\x11storage_volume_id\x18\x02 \x01(\tR\x0fstorageVolumeId\x12\x1f\n" + "\vvolume_type\x18\x03 \x01(\tR\n" + "volumeType\x125\n" + - "\x06status\x18\x04 \x01(\x0e2\x1d.ateapi.ExternalVolume.StatusR\x06status\"E\n" + + "\x06status\x18\x04 \x01(\x0e2\x1d.ateapi.ExternalVolume.StatusR\x06status\"D\n" + "\x06Status\x12\x12\n" + - "\x0eSTATUS_UNKNOWN\x10\x00\x12\f\n" + - "\bCREATING\x10\x01\x12\v\n" + + "\x0eSTATUS_UNKNOWN\x10\x00\x12\v\n" + + "\aPENDING\x10\x01\x12\v\n" + "\aCREATED\x10\x02\x12\f\n" + "\bDELETING\x10\x03\"\xd6\x06\n" + "\x05Actor\x124\n" + diff --git a/pkg/proto/ateapipb/ateapi.proto b/pkg/proto/ateapipb/ateapi.proto index 5cb5b4dc7..19dc7af18 100644 --- a/pkg/proto/ateapipb/ateapi.proto +++ b/pkg/proto/ateapipb/ateapi.proto @@ -127,8 +127,8 @@ message ExternalVolume { enum Status { STATUS_UNKNOWN = 0; - // Volume being created in the storage system. - CREATING = 1; + // Volume creation pending in the storage system. + PENDING = 1; // Volume successfully created in the storage system. CREATED = 2; // Volume being deleted from the storage system. From c7122b93919122ebbdb1e145cf9be68073134dec Mon Sep 17 00:00:00 2001 From: msau42 Date: Tue, 28 Jul 2026 20:26:39 +0000 Subject: [PATCH 3/5] Refactor DeleteActor to use a workflow. The steps are: - LoadActorForDeleteStep: Loads the actor state from store. - MarkDeletingStep: Validates state prerequisites (STATUS_SUSPENDED, STATUS_CRASHED, STATUS_DELETING) and updates actor and external volume statuses to STATUS_DELETING / ExternalVolume_DELETING. - DeleteVolumesStep: Deletes associated external volumes via deleteActorVolumes. - FinalizeDeletedStep: Deletes the actor from store and returns the deleted object. --- .../internal/controlapi/delete_actor.go | 54 +----- .../internal/controlapi/delete_actor_test.go | 1 + cmd/ateapi/internal/controlapi/volumes.go | 2 +- .../internal/controlapi/volumes_test.go | 6 +- cmd/ateapi/internal/controlapi/workflow.go | 28 +++ .../internal/controlapi/workflow_delete.go | 160 ++++++++++++++++++ .../controlapi/workflow_delete_test.go | 140 +++++++++++++++ 7 files changed, 334 insertions(+), 57 deletions(-) create mode 100644 cmd/ateapi/internal/controlapi/workflow_delete.go create mode 100644 cmd/ateapi/internal/controlapi/workflow_delete_test.go diff --git a/cmd/ateapi/internal/controlapi/delete_actor.go b/cmd/ateapi/internal/controlapi/delete_actor.go index 53ea38f02..9b1b37644 100644 --- a/cmd/ateapi/internal/controlapi/delete_actor.go +++ b/cmd/ateapi/internal/controlapi/delete_actor.go @@ -16,10 +16,7 @@ package controlapi import ( "context" - "errors" - "fmt" - "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" "github.com/agent-substrate/substrate/internal/resources" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" "google.golang.org/grpc/codes" @@ -34,56 +31,9 @@ func (s *Service) DeleteActor(ctx context.Context, req *ateapipb.DeleteActorRequ actorRef := resources.ActorRefFromObjectRef(req.GetActor()) setSpanActorRefAttributes(ctx, actorRef) - actor, err := s.persistence.GetActor(ctx, actorRef) + deleted, err := s.actorWorkflow.DeleteActor(ctx, req.GetActor().GetAtespace(), req.GetActor().GetName()) if err != nil { - if errors.Is(err, store.ErrNotFound) { - return nil, status.Errorf(codes.NotFound, "Actor %s not found", actorRef) - } - return nil, fmt.Errorf("while fetching actor: %w", err) - } - - if actor.GetStatus() != ateapipb.Actor_STATUS_SUSPENDED && - actor.GetStatus() != ateapipb.Actor_STATUS_CRASHED && - actor.GetStatus() != ateapipb.Actor_STATUS_DELETING { - return nil, status.Errorf(codes.FailedPrecondition, "Actor %s is not in a deletable status (status: %v)", name, actor.GetStatus()) - } - - if actor.GetStatus() != ateapipb.Actor_STATUS_DELETING { - actor.Status = ateapipb.Actor_STATUS_DELETING - for _, vol := range actor.GetActorVolumes() { - vol.Status = ateapipb.ExternalVolume_DELETING - } - updated, err := s.persistence.UpdateActor(ctx, actor, actor.GetMetadata().GetVersion()) - if err != nil { - if errors.Is(err, store.ErrPersistenceRetry) { - return nil, status.Error(codes.Aborted, "concurrent update conflict, please retry") - } - return nil, fmt.Errorf("while setting actor status to DELETING: %w", err) - } - actor = updated - } - - // Delete associated volumes - if err := s.deleteActorVolumes(ctx, actor.GetMetadata().GetUid(), actor.GetActorVolumes()); err != nil { - return nil, status.Errorf(codes.Internal, "while deleting actor volumes: %v", err) - } - - deleted, err := s.persistence.DeleteActor(ctx, actorRef) - if err != nil { - if errors.Is(err, store.ErrNotFound) { - return nil, status.Errorf(codes.NotFound, "Actor %s not found", actorRef) - } - if errors.Is(err, store.ErrFailedPrecondition) { - current, getErr := s.persistence.GetActor(ctx, actorRef) - if getErr == nil { - return nil, status.Errorf(codes.FailedPrecondition, "Actor %s is not in a deletable status (status: %v)", actorRef, current.GetStatus()) - } - return nil, status.Errorf(codes.FailedPrecondition, "Actor %s is not in a deletable status", actorRef) - } - 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) + return nil, err } return deleted, nil diff --git a/cmd/ateapi/internal/controlapi/delete_actor_test.go b/cmd/ateapi/internal/controlapi/delete_actor_test.go index 00d294c36..d228bc1a8 100644 --- a/cmd/ateapi/internal/controlapi/delete_actor_test.go +++ b/cmd/ateapi/internal/controlapi/delete_actor_test.go @@ -23,6 +23,7 @@ import ( "github.com/google/go-cmp/cmp" "github.com/agent-substrate/substrate/internal/ateattr" + "github.com/agent-substrate/substrate/internal/resources" "github.com/agent-substrate/substrate/internal/volume" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" "k8s.io/apimachinery/pkg/util/validation/field" diff --git a/cmd/ateapi/internal/controlapi/volumes.go b/cmd/ateapi/internal/controlapi/volumes.go index 0b99038e0..f2520a182 100644 --- a/cmd/ateapi/internal/controlapi/volumes.go +++ b/cmd/ateapi/internal/controlapi/volumes.go @@ -116,7 +116,7 @@ func createActorVolumes(ctx context.Context, actorUID string, template *atev1alp } // deleteActorVolumes deletes all external volumes in the list. -func (s *Service) deleteActorVolumes(ctx context.Context, actorUID string, volumes []*ateapipb.ExternalVolume) error { +func deleteActorVolumes(ctx context.Context, actorUID string, volumes []*ateapipb.ExternalVolume) error { if actorUID == "" { return errors.New("actorUID is required") } diff --git a/cmd/ateapi/internal/controlapi/volumes_test.go b/cmd/ateapi/internal/controlapi/volumes_test.go index c70b37c46..2601eb0be 100644 --- a/cmd/ateapi/internal/controlapi/volumes_test.go +++ b/cmd/ateapi/internal/controlapi/volumes_test.go @@ -156,7 +156,7 @@ func TestCreateActorVolumes(t *testing.T) { wantRes: []*ateapipb.ExternalVolume{ { VolumeName: "vol1", - StorageVolumeId: "mock-vol-1", + StorageVolumeId: "mock-vol-substrate-actor-uid-123-vol1", VolumeType: "mock", Status: ateapipb.ExternalVolume_CREATED, }, @@ -289,9 +289,7 @@ func TestDeleteActorVolumes(t *testing.T) { oldGlobalPlugin := globalVolumePlugin globalVolumePlugin = plugin defer func() { globalVolumePlugin = oldGlobalPlugin }() - svc := &Service{} - - err := svc.deleteActorVolumes(ctx, tt.actorUID, tt.volumes) + err := deleteActorVolumes(ctx, tt.actorUID, tt.volumes) if (err != nil) != tt.wantErr { t.Fatalf("deleteActorVolumes() error = %v, wantErr %v", err, tt.wantErr) } diff --git a/cmd/ateapi/internal/controlapi/workflow.go b/cmd/ateapi/internal/controlapi/workflow.go index 14c1df9dc..5289670d2 100644 --- a/cmd/ateapi/internal/controlapi/workflow.go +++ b/cmd/ateapi/internal/controlapi/workflow.go @@ -250,6 +250,34 @@ func (w *ActorWorkflow) PauseActor(ctx context.Context, actorRef resources.Actor return state.Actor, nil } +// DeleteActor executes the workflow to delete an actor. Idempotent. +func (w *ActorWorkflow) DeleteActor(ctx context.Context, atespace, name string) (*ateapipb.Actor, error) { + actorRef := resources.ActorRef{Atespace: atespace, Name: name} + input := &DeleteInput{ + ActorRef: actorRef, + } + state := &DeleteState{} + + ctx, lock, err := w.acquireActorLock(ctx, actorRef) + if err != nil { + return nil, err + } + defer lock.Close() + + steps := []WorkflowStep[*DeleteInput, *DeleteState]{ + &LoadActorForDeleteStep{store: w.store}, + &MarkDeletingStep{store: w.store}, + &DeleteVolumesStep{store: w.store}, + &FinalizeDeletedStep{store: w.store}, + } + + if err := RunWorkflow(ctx, input, state, steps); err != nil { + return nil, err + } + + return state.DeletedActor, nil +} + func (w *ActorWorkflow) acquireActorLock(ctx context.Context, actorRef resources.ActorRef) (context.Context, *store.Lock, error) { lockKey := "lock:actor:" + actorRef.Atespace + ":" + actorRef.Name diff --git a/cmd/ateapi/internal/controlapi/workflow_delete.go b/cmd/ateapi/internal/controlapi/workflow_delete.go new file mode 100644 index 000000000..21a31c9e5 --- /dev/null +++ b/cmd/ateapi/internal/controlapi/workflow_delete.go @@ -0,0 +1,160 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package controlapi + +import ( + "context" + "errors" + "fmt" + + "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" + "github.com/agent-substrate/substrate/internal/resources" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "k8s.io/apimachinery/pkg/util/wait" +) + +// DeleteInput holds the immutable parameters requested by the client. +type DeleteInput struct { + ActorRef resources.ActorRef +} + +// DeleteState holds the mutable state loaded and modified during execution. +type DeleteState struct { + Actor *ateapipb.Actor + DeletedActor *ateapipb.Actor +} + +type LoadActorForDeleteStep struct { + store store.Interface +} + +func (s *LoadActorForDeleteStep) Name() string { return "LoadActorForDelete" } +func (s *LoadActorForDeleteStep) IsComplete(ctx context.Context, input *DeleteInput, state *DeleteState) (bool, error) { + // Always run to get the freshest state + return false, nil +} +func (s *LoadActorForDeleteStep) CheckPrerequisite(ctx context.Context, input *DeleteInput, state *DeleteState) error { + return nil +} +func (s *LoadActorForDeleteStep) Execute(ctx context.Context, input *DeleteInput, state *DeleteState) error { + actor, err := s.store.GetActor(ctx, input.ActorRef) + if err != nil { + if errors.Is(err, store.ErrNotFound) { + return status.Errorf(codes.NotFound, "Actor %s not found", input.ActorRef) + } + return fmt.Errorf("while fetching actor: %w", err) + } + state.Actor = actor + return nil +} + +func (s *LoadActorForDeleteStep) RetryBackoff() *wait.Backoff { return nil } + +type MarkDeletingStep struct { + store store.Interface +} + +func (s *MarkDeletingStep) Name() string { return "MarkDeleting" } +func (s *MarkDeletingStep) IsComplete(ctx context.Context, input *DeleteInput, state *DeleteState) (bool, error) { + return state.Actor.GetStatus() == ateapipb.Actor_STATUS_DELETING, nil +} +func (s *MarkDeletingStep) CheckPrerequisite(ctx context.Context, input *DeleteInput, state *DeleteState) error { + if state.Actor.GetStatus() != ateapipb.Actor_STATUS_SUSPENDED && + state.Actor.GetStatus() != ateapipb.Actor_STATUS_CRASHED && + state.Actor.GetStatus() != ateapipb.Actor_STATUS_DELETING { + return status.Errorf(codes.FailedPrecondition, "Actor %s is not in a deletable status (status: %v)", input.ActorRef, state.Actor.GetStatus()) + } + return nil +} +func (s *MarkDeletingStep) Execute(ctx context.Context, input *DeleteInput, state *DeleteState) error { + state.Actor.Status = ateapipb.Actor_STATUS_DELETING + for _, vol := range state.Actor.GetActorVolumes() { + vol.Status = ateapipb.ExternalVolume_DELETING + } + updated, err := s.store.UpdateActor(ctx, state.Actor, state.Actor.GetMetadata().GetVersion()) + if err != nil { + if errors.Is(err, store.ErrVersionConflict) { + return status.Error(codes.Aborted, "concurrent update conflict, please retry") + } + return fmt.Errorf("while setting actor status to DELETING: %w", err) + } + state.Actor = updated + return nil +} + +func (s *MarkDeletingStep) RetryBackoff() *wait.Backoff { return nil } + +type DeleteVolumesStep struct { + store store.Interface +} + +func (s *DeleteVolumesStep) Name() string { return "DeleteVolumes" } +func (s *DeleteVolumesStep) IsComplete(ctx context.Context, input *DeleteInput, state *DeleteState) (bool, error) { + return false, nil +} +func (s *DeleteVolumesStep) CheckPrerequisite(ctx context.Context, input *DeleteInput, state *DeleteState) error { + if state.Actor.GetStatus() != ateapipb.Actor_STATUS_DELETING { + return status.Errorf(codes.FailedPrecondition, "DeleteVolumesStep prerequisite not met for Actor: %s (got: %v, want %s)", input.ActorRef, state.Actor.GetStatus(), ateapipb.Actor_STATUS_DELETING) + } + return nil +} +func (s *DeleteVolumesStep) Execute(ctx context.Context, input *DeleteInput, state *DeleteState) error { + if err := deleteActorVolumes(ctx, state.Actor.GetMetadata().GetUid(), state.Actor.GetActorVolumes()); err != nil { + return status.Errorf(codes.Internal, "while deleting actor volumes: %v", err) + } + return nil +} + +func (s *DeleteVolumesStep) RetryBackoff() *wait.Backoff { return nil } + +type FinalizeDeletedStep struct { + store store.Interface +} + +func (s *FinalizeDeletedStep) Name() string { return "FinalizeDeleted" } +func (s *FinalizeDeletedStep) IsComplete(ctx context.Context, input *DeleteInput, state *DeleteState) (bool, error) { + return state.DeletedActor != nil, nil +} +func (s *FinalizeDeletedStep) CheckPrerequisite(ctx context.Context, input *DeleteInput, state *DeleteState) error { + if state.Actor.GetStatus() != ateapipb.Actor_STATUS_DELETING { + return status.Errorf(codes.FailedPrecondition, "FinalizeDeletedStep prerequisite not met for Actor: %s (got: %v, want %s)", input.ActorRef, state.Actor.GetStatus(), ateapipb.Actor_STATUS_DELETING) + } + return nil +} +func (s *FinalizeDeletedStep) Execute(ctx context.Context, input *DeleteInput, state *DeleteState) error { + deleted, err := s.store.DeleteActor(ctx, input.ActorRef) + if err != nil { + if errors.Is(err, store.ErrNotFound) { + return status.Errorf(codes.NotFound, "Actor %s not found", input.ActorRef) + } + if errors.Is(err, store.ErrFailedPrecondition) { + current, getErr := s.store.GetActor(ctx, input.ActorRef) + if getErr == nil { + return status.Errorf(codes.FailedPrecondition, "Actor %s is not in a deletable status (status: %v)", input.ActorRef, current.GetStatus()) + } + return status.Errorf(codes.FailedPrecondition, "Actor %s is not in a deletable status", input.ActorRef) + } + if errors.Is(err, store.ErrVersionConflict) { + return status.Error(codes.Aborted, "concurrent update conflict, please retry") + } + return fmt.Errorf("while deleting actor from DB: %w", err) + } + state.DeletedActor = deleted + return nil +} + +func (s *FinalizeDeletedStep) RetryBackoff() *wait.Backoff { return nil } diff --git a/cmd/ateapi/internal/controlapi/workflow_delete_test.go b/cmd/ateapi/internal/controlapi/workflow_delete_test.go new file mode 100644 index 000000000..2751470fa --- /dev/null +++ b/cmd/ateapi/internal/controlapi/workflow_delete_test.go @@ -0,0 +1,140 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package controlapi + +import ( + "context" + "testing" + + "github.com/agent-substrate/substrate/cmd/ateapi/internal/store/storetest" + "github.com/agent-substrate/substrate/internal/resources" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +func TestDeleteActorWorkflow_ExecutionPaths(t *testing.T) { + tests := []struct { + name string + seedStatus ateapipb.Actor_Status + wantErr bool + wantCode codes.Code + }{ + { + name: "delete suspended actor succeeds", + seedStatus: ateapipb.Actor_STATUS_SUSPENDED, + wantErr: false, + }, + { + name: "delete crashed actor succeeds", + seedStatus: ateapipb.Actor_STATUS_CRASHED, + wantErr: false, + }, + { + name: "delete deleting actor succeeds", + seedStatus: ateapipb.Actor_STATUS_DELETING, + wantErr: false, + }, + { + name: "delete running actor rejected", + seedStatus: ateapipb.Actor_STATUS_RUNNING, + wantErr: true, + wantCode: codes.FailedPrecondition, + }, + { + name: "delete paused actor rejected", + seedStatus: ateapipb.Actor_STATUS_PAUSED, + wantErr: true, + wantCode: codes.FailedPrecondition, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + ctx := context.Background() + st, cleanup := storetest.SetupTestStore(t) + defer cleanup() + w := newTestActorWorkflow(t, st, "ns", "tmpl1") + + actorRef := resources.ActorRef{Atespace: "team-a", Name: "id1"} + seedWorkflowActor(t, ctx, st, actorRef, "ns", "tmpl1", tc.seedStatus) + + deleted, err := w.DeleteActor(ctx, "team-a", "id1") + if tc.wantErr { + if got := status.Code(err); got != tc.wantCode { + t.Fatalf("status.Code(err) = %v, want %v (err: %v)", got, tc.wantCode, err) + } + } else { + if err != nil { + t.Fatalf("DeleteActor failed: %v", err) + } + if deleted == nil { + t.Fatalf("expected non-nil deleted actor") + } + if _, err := st.GetActor(ctx, actorRef); err == nil { + t.Errorf("expected actor to be deleted from store, but it still exists") + } + } + }) + } +} + +func TestDeleteSteps_CheckPrerequisite(t *testing.T) { + tests := []struct { + name string + step WorkflowStep[*DeleteInput, *DeleteState] + allowed map[ateapipb.Actor_Status]bool + }{ + { + name: "LoadActorForDeleteStep", + step: &LoadActorForDeleteStep{}, + allowed: nil, + }, + { + name: "MarkDeletingStep", + step: &MarkDeletingStep{}, + allowed: map[ateapipb.Actor_Status]bool{ + ateapipb.Actor_STATUS_SUSPENDED: true, + ateapipb.Actor_STATUS_CRASHED: true, + ateapipb.Actor_STATUS_DELETING: true, + }, + }, + { + name: "DeleteVolumesStep", + step: &DeleteVolumesStep{}, + allowed: map[ateapipb.Actor_Status]bool{ + ateapipb.Actor_STATUS_DELETING: true, + }, + }, + { + name: "FinalizeDeletedStep", + step: &FinalizeDeletedStep{}, + allowed: map[ateapipb.Actor_Status]bool{ + ateapipb.Actor_STATUS_DELETING: true, + }, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + ctx := context.Background() + actorRef := resources.ActorRef{Atespace: "team-a", Name: "id1"} + for _, st := range allActorStatuses { + err := tc.step.CheckPrerequisite(ctx, &DeleteInput{ActorRef: actorRef}, &DeleteState{Actor: &ateapipb.Actor{Status: st}}) + assertPrerequisiteResult(t, st, err, tc.allowed == nil || tc.allowed[st]) + } + }) + } +} From 5a3e47a091500ba4607598fe423daaca65eebe0e Mon Sep 17 00:00:00 2001 From: msau42 Date: Wed, 29 Jul 2026 21:42:47 +0000 Subject: [PATCH 4/5] Add STATUS prefix to ExternalVolume Status fields --- .../internal/controlapi/create_actor.go | 2 +- .../internal/controlapi/delete_actor_test.go | 4 +- .../internal/controlapi/functional_test.go | 22 +++++----- cmd/ateapi/internal/controlapi/volumes.go | 10 ++--- .../internal/controlapi/volumes_test.go | 32 +++++++-------- .../internal/controlapi/workflow_delete.go | 2 +- .../internal/controlapi/workflow_resume.go | 2 +- pkg/proto/ateapipb/ateapi.pb.go | 40 +++++++++---------- pkg/proto/ateapipb/ateapi.proto | 8 ++-- 9 files changed, 61 insertions(+), 61 deletions(-) diff --git a/cmd/ateapi/internal/controlapi/create_actor.go b/cmd/ateapi/internal/controlapi/create_actor.go index d6b9e9bea..475134b60 100644 --- a/cmd/ateapi/internal/controlapi/create_actor.go +++ b/cmd/ateapi/internal/controlapi/create_actor.go @@ -72,7 +72,7 @@ func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequ ActorTemplateNamespace: templateNamespace, ActorTemplateName: templateName, WorkerSelector: in.GetWorkerSelector(), - ActorVolumes: initVols, + ActorVolumes: initVols, // TODO: validate external volume fields } stored, err := s.persistence.CreateActor(ctx, actor) if err != nil { diff --git a/cmd/ateapi/internal/controlapi/delete_actor_test.go b/cmd/ateapi/internal/controlapi/delete_actor_test.go index d228bc1a8..fdd9f05ec 100644 --- a/cmd/ateapi/internal/controlapi/delete_actor_test.go +++ b/cmd/ateapi/internal/controlapi/delete_actor_test.go @@ -182,8 +182,8 @@ func TestDeleteActor_MultipleVolumeDeletionFailures(t *testing.T) { ActorTemplateNamespace: ns, ActorTemplateName: "tmpl1", ActorVolumes: []*ateapipb.ExternalVolume{ - {VolumeName: "vol1", StorageVolumeId: "storage-vol-1", Status: ateapipb.ExternalVolume_CREATED}, - {VolumeName: "vol2", StorageVolumeId: "storage-vol-2", Status: ateapipb.ExternalVolume_CREATED}, + {VolumeName: "vol1", StorageVolumeId: "storage-vol-1", Status: ateapipb.ExternalVolume_STATUS_CREATED}, + {VolumeName: "vol2", StorageVolumeId: "storage-vol-2", Status: ateapipb.ExternalVolume_STATUS_CREATED}, }, } if _, err := tc.persistence.CreateActor(context.Background(), actor); err != nil { diff --git a/cmd/ateapi/internal/controlapi/functional_test.go b/cmd/ateapi/internal/controlapi/functional_test.go index 3e4f05622..b23d8ce76 100644 --- a/cmd/ateapi/internal/controlapi/functional_test.go +++ b/cmd/ateapi/internal/controlapi/functional_test.go @@ -824,8 +824,8 @@ func TestCreateActor_WithExternalVolumes(t *testing.T) { if vol.GetVolumeName() != "ext-vol-1" { t.Errorf("volume name = %q, want %q", vol.GetVolumeName(), "ext-vol-1") } - if vol.GetStatus() != ateapipb.ExternalVolume_PENDING { - t.Errorf("volume status = %v, want %v", vol.GetStatus(), ateapipb.ExternalVolume_PENDING) + if vol.GetStatus() != ateapipb.ExternalVolume_STATUS_PENDING { + t.Errorf("volume status = %v, want %v", vol.GetStatus(), ateapipb.ExternalVolume_STATUS_PENDING) } if vol.GetStorageVolumeId() != "" { t.Errorf("expected empty storageVolumeId before resume, got %q", vol.GetStorageVolumeId()) @@ -841,8 +841,8 @@ func TestCreateActor_WithExternalVolumes(t *testing.T) { if len(getResp.GetActorVolumes()) != 1 { t.Fatalf("expected 1 volume in GetActor response, got %d", len(getResp.GetActorVolumes())) } - if getResp.GetActorVolumes()[0].GetStatus() != ateapipb.ExternalVolume_PENDING { - t.Errorf("GetActor status = %v, want %v", getResp.GetActorVolumes()[0].GetStatus(), ateapipb.ExternalVolume_PENDING) + if getResp.GetActorVolumes()[0].GetStatus() != ateapipb.ExternalVolume_STATUS_PENDING { + t.Errorf("GetActor status = %v, want %v", getResp.GetActorVolumes()[0].GetStatus(), ateapipb.ExternalVolume_STATUS_PENDING) } } @@ -885,7 +885,7 @@ func TestActorLifecycle_WithExternalVolumes(t *testing.T) { if createResp.GetStatus() != ateapipb.Actor_STATUS_SUSPENDED { t.Fatalf("expected initial status STATUS_SUSPENDED, got %v", createResp.GetStatus()) } - if len(createResp.GetActorVolumes()) != 1 || createResp.GetActorVolumes()[0].GetStatus() != ateapipb.ExternalVolume_PENDING { + if len(createResp.GetActorVolumes()) != 1 || createResp.GetActorVolumes()[0].GetStatus() != ateapipb.ExternalVolume_STATUS_PENDING { t.Fatalf("expected 1 pending volume after CreateActor, got %v", createResp.GetActorVolumes()) } @@ -899,7 +899,7 @@ func TestActorLifecycle_WithExternalVolumes(t *testing.T) { if resumeResp.GetActor().GetStatus() != ateapipb.Actor_STATUS_RUNNING { t.Fatalf("expected status STATUS_RUNNING after resume, got %v", resumeResp.GetActor().GetStatus()) } - if len(resumeResp.GetActor().GetActorVolumes()) != 1 || resumeResp.GetActor().GetActorVolumes()[0].GetStatus() != ateapipb.ExternalVolume_CREATED { + if len(resumeResp.GetActor().GetActorVolumes()) != 1 || resumeResp.GetActor().GetActorVolumes()[0].GetStatus() != ateapipb.ExternalVolume_STATUS_CREATED { t.Fatalf("expected 1 created volume after ResumeActor, got %v", resumeResp.GetActor().GetActorVolumes()) } if resumeResp.GetActor().GetActorVolumes()[0].GetStorageVolumeId() == "" { @@ -1071,10 +1071,10 @@ func TestResumeActor_VolumeCreationFailure(t *testing.T) { for _, v := range getResp.GetActorVolumes() { volsByName[v.GetVolumeName()] = v } - if v1, ok := volsByName["succ-vol1"]; !ok || v1.GetStatus() != ateapipb.ExternalVolume_CREATED || v1.GetStorageVolumeId() == "" { + if v1, ok := volsByName["succ-vol1"]; !ok || v1.GetStatus() != ateapipb.ExternalVolume_STATUS_CREATED || v1.GetStorageVolumeId() == "" { t.Errorf("succ-vol1 unexpected state: %v", v1) } - if v2, ok := volsByName["fail-vol2"]; !ok || v2.GetStatus() != ateapipb.ExternalVolume_PENDING { + if v2, ok := volsByName["fail-vol2"]; !ok || v2.GetStatus() != ateapipb.ExternalVolume_STATUS_PENDING { t.Errorf("fail-vol2 unexpected state: %v", v2) } @@ -1214,10 +1214,10 @@ func TestResumeActor_VolumeCreationRetrySuccess(t *testing.T) { for _, v := range getResp.GetActorVolumes() { volsByName[v.GetVolumeName()] = v } - if v1, ok := volsByName["succ-vol1"]; !ok || v1.GetStatus() != ateapipb.ExternalVolume_CREATED || v1.GetStorageVolumeId() == "" { + if v1, ok := volsByName["succ-vol1"]; !ok || v1.GetStatus() != ateapipb.ExternalVolume_STATUS_CREATED || v1.GetStorageVolumeId() == "" { t.Errorf("succ-vol1 unexpected state after first resume: %v", v1) } - if v2, ok := volsByName["retry-vol2"]; !ok || v2.GetStatus() != ateapipb.ExternalVolume_PENDING { + if v2, ok := volsByName["retry-vol2"]; !ok || v2.GetStatus() != ateapipb.ExternalVolume_STATUS_PENDING { t.Errorf("retry-vol2 unexpected state after first resume: %v", v2) } @@ -1240,7 +1240,7 @@ func TestResumeActor_VolumeCreationRetrySuccess(t *testing.T) { t.Errorf("actor status after second resume = %v, want %v", getResp.GetStatus(), ateapipb.Actor_STATUS_RUNNING) } for _, v := range getResp.GetActorVolumes() { - if v.GetStatus() != ateapipb.ExternalVolume_CREATED || v.GetStorageVolumeId() == "" { + if v.GetStatus() != ateapipb.ExternalVolume_STATUS_CREATED || v.GetStorageVolumeId() == "" { t.Errorf("volume %s unexpected state after second resume: %v", v.GetVolumeName(), v) } } diff --git a/cmd/ateapi/internal/controlapi/volumes.go b/cmd/ateapi/internal/controlapi/volumes.go index f2520a182..c14b87996 100644 --- a/cmd/ateapi/internal/controlapi/volumes.go +++ b/cmd/ateapi/internal/controlapi/volumes.go @@ -45,7 +45,7 @@ func initialActorVolumes(template *atev1alpha1.ActorTemplate) []*ateapipb.Extern volumes = append(volumes, &ateapipb.ExternalVolume{ VolumeName: vol.Name, VolumeType: "mock", // TODO fix when we support multiple plugins - Status: ateapipb.ExternalVolume_PENDING, + Status: ateapipb.ExternalVolume_STATUS_PENDING, }) } } @@ -83,12 +83,12 @@ func createActorVolumes(ctx context.Context, actorUID string, template *atev1alp } switch vol.GetStatus() { - case ateapipb.ExternalVolume_PENDING: + case ateapipb.ExternalVolume_STATUS_PENDING: // proceed with volume creation - case ateapipb.ExternalVolume_CREATED: + case ateapipb.ExternalVolume_STATUS_CREATED: resultVolumes = append(resultVolumes, vol) continue - case ateapipb.ExternalVolume_DELETING: + case ateapipb.ExternalVolume_STATUS_DELETING: return resultVolumes, status.Errorf(codes.FailedPrecondition, "cannot create volume %q in DELETING status", volName) default: return resultVolumes, status.Errorf(codes.Internal, "unexpected status %s for volume %q", vol.GetStatus(), volName) @@ -109,7 +109,7 @@ func createActorVolumes(ctx context.Context, actorUID string, template *atev1alp VolumeName: volName, StorageVolumeId: storageVolumeID, VolumeType: vol.GetVolumeType(), - Status: ateapipb.ExternalVolume_CREATED, + Status: ateapipb.ExternalVolume_STATUS_CREATED, }) } return resultVolumes, nil diff --git a/cmd/ateapi/internal/controlapi/volumes_test.go b/cmd/ateapi/internal/controlapi/volumes_test.go index 2601eb0be..50643c48f 100644 --- a/cmd/ateapi/internal/controlapi/volumes_test.go +++ b/cmd/ateapi/internal/controlapi/volumes_test.go @@ -63,12 +63,12 @@ func TestInitialActorVolumes_PendingState(t *testing.T) { { VolumeName: "data-vol-1", VolumeType: "mock", - Status: ateapipb.ExternalVolume_PENDING, + Status: ateapipb.ExternalVolume_STATUS_PENDING, }, { VolumeName: "data-vol-2", VolumeType: "mock", - Status: ateapipb.ExternalVolume_PENDING, + Status: ateapipb.ExternalVolume_STATUS_PENDING, }, } @@ -141,15 +141,15 @@ func TestCreateActorVolumes(t *testing.T) { { VolumeName: "vol1", VolumeType: "mock", - Status: ateapipb.ExternalVolume_PENDING, + Status: ateapipb.ExternalVolume_STATUS_PENDING, }, { VolumeName: "vol2", - Status: ateapipb.ExternalVolume_DELETING, + Status: ateapipb.ExternalVolume_STATUS_DELETING, }, { VolumeName: "vol3", - Status: ateapipb.ExternalVolume_PENDING, + Status: ateapipb.ExternalVolume_STATUS_PENDING, }, }, wantErr: true, @@ -158,15 +158,15 @@ func TestCreateActorVolumes(t *testing.T) { VolumeName: "vol1", StorageVolumeId: "mock-vol-substrate-actor-uid-123-vol1", VolumeType: "mock", - Status: ateapipb.ExternalVolume_CREATED, + Status: ateapipb.ExternalVolume_STATUS_CREATED, }, { VolumeName: "vol2", - Status: ateapipb.ExternalVolume_DELETING, + Status: ateapipb.ExternalVolume_STATUS_DELETING, }, { VolumeName: "vol3", - Status: ateapipb.ExternalVolume_PENDING, + Status: ateapipb.ExternalVolume_STATUS_PENDING, }, }, }, @@ -177,7 +177,7 @@ func TestCreateActorVolumes(t *testing.T) { { VolumeName: "data-vol", StorageVolumeId: "existing-vol-id", - Status: ateapipb.ExternalVolume_CREATED, + Status: ateapipb.ExternalVolume_STATUS_CREATED, }, }, wantErr: false, @@ -185,24 +185,24 @@ func TestCreateActorVolumes(t *testing.T) { { VolumeName: "data-vol", StorageVolumeId: "existing-vol-id", - Status: ateapipb.ExternalVolume_CREATED, + Status: ateapipb.ExternalVolume_STATUS_CREATED, }, }, }, { - name: "unknown volume status returns error", + name: "unspecified volume status returns error", tmpl: standardTmpl, inputVolumes: []*ateapipb.ExternalVolume{ { VolumeName: "data-vol", - Status: ateapipb.ExternalVolume_STATUS_UNKNOWN, + Status: ateapipb.ExternalVolume_STATUS_UNSPECIFIED, }, }, wantErr: true, wantRes: []*ateapipb.ExternalVolume{ { VolumeName: "data-vol", - Status: ateapipb.ExternalVolume_STATUS_UNKNOWN, + Status: ateapipb.ExternalVolume_STATUS_UNSPECIFIED, }, }, }, @@ -216,14 +216,14 @@ func TestCreateActorVolumes(t *testing.T) { inputVolumes: []*ateapipb.ExternalVolume{ { VolumeName: "missing-vol", - Status: ateapipb.ExternalVolume_PENDING, + Status: ateapipb.ExternalVolume_STATUS_PENDING, }, }, wantErr: true, wantRes: []*ateapipb.ExternalVolume{ { VolumeName: "missing-vol", - Status: ateapipb.ExternalVolume_PENDING, + Status: ateapipb.ExternalVolume_STATUS_PENDING, }, }, }, @@ -276,7 +276,7 @@ func TestDeleteActorVolumes(t *testing.T) { name: "falls back to actorVolumeID when storage volume ID is empty regardless of status", actorUID: "uid-abc", volumes: []*ateapipb.ExternalVolume{ - {VolumeName: "vol1", StorageVolumeId: "", Status: ateapipb.ExternalVolume_CREATED}, + {VolumeName: "vol1", StorageVolumeId: "", Status: ateapipb.ExternalVolume_STATUS_CREATED}, }, wantDeleted: []string{"substrate-uid-abc-vol1"}, wantErr: false, diff --git a/cmd/ateapi/internal/controlapi/workflow_delete.go b/cmd/ateapi/internal/controlapi/workflow_delete.go index 21a31c9e5..761f46c78 100644 --- a/cmd/ateapi/internal/controlapi/workflow_delete.go +++ b/cmd/ateapi/internal/controlapi/workflow_delete.go @@ -83,7 +83,7 @@ func (s *MarkDeletingStep) CheckPrerequisite(ctx context.Context, input *DeleteI func (s *MarkDeletingStep) Execute(ctx context.Context, input *DeleteInput, state *DeleteState) error { state.Actor.Status = ateapipb.Actor_STATUS_DELETING for _, vol := range state.Actor.GetActorVolumes() { - vol.Status = ateapipb.ExternalVolume_DELETING + vol.Status = ateapipb.ExternalVolume_STATUS_DELETING } updated, err := s.store.UpdateActor(ctx, state.Actor, state.Actor.GetMetadata().GetVersion()) if err != nil { diff --git a/cmd/ateapi/internal/controlapi/workflow_resume.go b/cmd/ateapi/internal/controlapi/workflow_resume.go index 6948aaf9a..aee116890 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume.go @@ -133,7 +133,7 @@ func (s *CreateVolumesStep) Name() string { return "CreateVolumes" } func (s *CreateVolumesStep) IsComplete(ctx context.Context, input *ResumeInput, state *ResumeState) (bool, error) { for _, vol := range state.Actor.GetActorVolumes() { - if vol.GetStatus() == ateapipb.ExternalVolume_PENDING { + if vol.GetStatus() == ateapipb.ExternalVolume_STATUS_PENDING { return false, nil } } diff --git a/pkg/proto/ateapipb/ateapi.pb.go b/pkg/proto/ateapipb/ateapi.pb.go index 659a9a95c..674895f0e 100644 --- a/pkg/proto/ateapipb/ateapi.pb.go +++ b/pkg/proto/ateapipb/ateapi.pb.go @@ -41,28 +41,28 @@ const ( type ExternalVolume_Status int32 const ( - ExternalVolume_STATUS_UNKNOWN ExternalVolume_Status = 0 + ExternalVolume_STATUS_UNSPECIFIED ExternalVolume_Status = 0 // Volume creation pending in the storage system. - ExternalVolume_PENDING ExternalVolume_Status = 1 + ExternalVolume_STATUS_PENDING ExternalVolume_Status = 1 // Volume successfully created in the storage system. - ExternalVolume_CREATED ExternalVolume_Status = 2 + ExternalVolume_STATUS_CREATED ExternalVolume_Status = 2 // Volume being deleted from the storage system. - ExternalVolume_DELETING ExternalVolume_Status = 3 + ExternalVolume_STATUS_DELETING ExternalVolume_Status = 3 ) // Enum value maps for ExternalVolume_Status. var ( ExternalVolume_Status_name = map[int32]string{ - 0: "STATUS_UNKNOWN", - 1: "PENDING", - 2: "CREATED", - 3: "DELETING", + 0: "STATUS_UNSPECIFIED", + 1: "STATUS_PENDING", + 2: "STATUS_CREATED", + 3: "STATUS_DELETING", } ExternalVolume_Status_value = map[string]int32{ - "STATUS_UNKNOWN": 0, - "PENDING": 1, - "CREATED": 2, - "DELETING": 3, + "STATUS_UNSPECIFIED": 0, + "STATUS_PENDING": 1, + "STATUS_CREATED": 2, + "STATUS_DELETING": 3, } ) @@ -597,7 +597,7 @@ func (x *ExternalVolume) GetStatus() ExternalVolume_Status { if x != nil { return x.Status } - return ExternalVolume_STATUS_UNKNOWN + return ExternalVolume_STATUS_UNSPECIFIED } type Actor struct { @@ -2418,19 +2418,19 @@ const file_ateapi_proto_rawDesc = "" + "\vcreate_time\x18\x05 \x01(\v2\x1a.google.protobuf.TimestampR\n" + "createTime\x12;\n" + "\vupdate_time\x18\x06 \x01(\v2\x1a.google.protobuf.TimestampR\n" + - "updateTime\"\xfb\x01\n" + + "updateTime\"\x94\x02\n" + "\x0eExternalVolume\x12\x1f\n" + "\vvolume_name\x18\x01 \x01(\tR\n" + "volumeName\x12*\n" + "\x11storage_volume_id\x18\x02 \x01(\tR\x0fstorageVolumeId\x12\x1f\n" + "\vvolume_type\x18\x03 \x01(\tR\n" + "volumeType\x125\n" + - "\x06status\x18\x04 \x01(\x0e2\x1d.ateapi.ExternalVolume.StatusR\x06status\"D\n" + - "\x06Status\x12\x12\n" + - "\x0eSTATUS_UNKNOWN\x10\x00\x12\v\n" + - "\aPENDING\x10\x01\x12\v\n" + - "\aCREATED\x10\x02\x12\f\n" + - "\bDELETING\x10\x03\"\xd6\x06\n" + + "\x06status\x18\x04 \x01(\x0e2\x1d.ateapi.ExternalVolume.StatusR\x06status\"]\n" + + "\x06Status\x12\x16\n" + + "\x12STATUS_UNSPECIFIED\x10\x00\x12\x12\n" + + "\x0eSTATUS_PENDING\x10\x01\x12\x12\n" + + "\x0eSTATUS_CREATED\x10\x02\x12\x13\n" + + "\x0fSTATUS_DELETING\x10\x03\"\xd6\x06\n" + "\x05Actor\x124\n" + "\bmetadata\x18\x01 \x01(\v2\x18.ateapi.ResourceMetadataR\bmetadata\x128\n" + "\x18actor_template_namespace\x18\x02 \x01(\tR\x16actorTemplateNamespace\x12.\n" + diff --git a/pkg/proto/ateapipb/ateapi.proto b/pkg/proto/ateapipb/ateapi.proto index 19dc7af18..0384bec60 100644 --- a/pkg/proto/ateapipb/ateapi.proto +++ b/pkg/proto/ateapipb/ateapi.proto @@ -126,13 +126,13 @@ message ExternalVolume { string volume_type = 3; enum Status { - STATUS_UNKNOWN = 0; + STATUS_UNSPECIFIED = 0; // Volume creation pending in the storage system. - PENDING = 1; + STATUS_PENDING = 1; // Volume successfully created in the storage system. - CREATED = 2; + STATUS_CREATED = 2; // Volume being deleted from the storage system. - DELETING = 3; + STATUS_DELETING = 3; } Status status = 4; } From df90729eeae600e98f189ec96fbc0486edc4e4ed Mon Sep 17 00:00:00 2001 From: msau42 Date: Wed, 29 Jul 2026 22:09:58 +0000 Subject: [PATCH 5/5] Fix unused imports in delete_actor.go --- cmd/ateapi/internal/controlapi/delete_actor.go | 2 -- 1 file changed, 2 deletions(-) diff --git a/cmd/ateapi/internal/controlapi/delete_actor.go b/cmd/ateapi/internal/controlapi/delete_actor.go index 9b1b37644..2f9b5f546 100644 --- a/cmd/ateapi/internal/controlapi/delete_actor.go +++ b/cmd/ateapi/internal/controlapi/delete_actor.go @@ -19,8 +19,6 @@ import ( "github.com/agent-substrate/substrate/internal/resources" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" - "google.golang.org/grpc/codes" - "google.golang.org/grpc/status" "k8s.io/apimachinery/pkg/util/validation/field" )