diff --git a/cmd/ateapi/internal/controlapi/create_actor.go b/cmd/ateapi/internal/controlapi/create_actor.go index f19ee0d1b..996d671e0 100644 --- a/cmd/ateapi/internal/controlapi/create_actor.go +++ b/cmd/ateapi/internal/controlapi/create_actor.go @@ -83,7 +83,7 @@ func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequ stored, err := s.persistence.CreateActor(ctx, actor) if err != nil { // Cleanup created volumes if DB write fails - _ = s.deleteActorVolumes(ctx, actorRef, volumes) + s.cleanupActorVolumes(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/volumes.go b/cmd/ateapi/internal/controlapi/volumes.go index 161c74e8c..5db8f6c28 100644 --- a/cmd/ateapi/internal/controlapi/volumes.go +++ b/cmd/ateapi/internal/controlapi/volumes.go @@ -19,6 +19,7 @@ import ( "errors" "fmt" "log/slog" + "time" "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" "github.com/agent-substrate/substrate/internal/volume" @@ -29,9 +30,13 @@ import ( ) var ( - globalVolumePlugin = volume.NewMockVolumePlugin() + globalVolumePlugin volume.VolumePluginControlPlane = volume.NewMockVolumePlugin() ) +// volumeCleanupTimeout bounds the best-effort deletion of volumes created for +// an actor that failed to come up. +const volumeCleanupTimeout = 30 * time.Second + // TODO: Replace with actual volume plugin search func getVolumePlugin() volume.VolumePluginControlPlane { return globalVolumePlugin @@ -52,7 +57,7 @@ func (s *Service) createActorVolumes(ctx context.Context, ref *ateapipb.ObjectRe 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) + s.cleanupActorVolumes(ctx, ref, volumes) return nil, status.Errorf(codes.Internal, "failed to create volume %q: %v", vol.Name, err) } volumes = append(volumes, &ateapipb.ExternalVolume{ @@ -66,6 +71,16 @@ func (s *Service) createActorVolumes(ctx context.Context, ref *ateapipb.ObjectRe return volumes, nil } +// cleanupActorVolumes deletes, on a best-effort basis, volumes created for an +// actor that failed to come up. It runs on a fresh timeout detached from ctx, +// because a canceled or expired ctx is itself a common reason for that failure: +// reusing it would make every delete fail immediately and leak the volumes. +func (s *Service) cleanupActorVolumes(ctx context.Context, ref *ateapipb.ObjectRef, volumes []*ateapipb.ExternalVolume) { + cleanupCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), volumeCleanupTimeout) + defer cancel() + _ = s.deleteActorVolumes(cleanupCtx, ref, volumes) +} + // deleteActorVolumes deletes all external volumes in the list. func (s *Service) deleteActorVolumes(ctx context.Context, ref *ateapipb.ObjectRef, volumes []*ateapipb.ExternalVolume) error { var errs []error diff --git a/cmd/ateapi/internal/controlapi/volumes_test.go b/cmd/ateapi/internal/controlapi/volumes_test.go new file mode 100644 index 000000000..48bd21a48 --- /dev/null +++ b/cmd/ateapi/internal/controlapi/volumes_test.go @@ -0,0 +1,67 @@ +// 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/internal/volume" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" +) + +// ctxCapturingPlugin records the context each DeleteVolume call is made with, +// so a test can assert cleanup does not inherit a canceled context. +type ctxCapturingPlugin struct { + *volume.MockVolumePlugin + + deleteCtxErr error + deleted []string +} + +func (p *ctxCapturingPlugin) DeleteVolume(ctx context.Context, volumeID string) error { + p.deleteCtxErr = ctx.Err() + p.deleted = append(p.deleted, volumeID) + return p.MockVolumePlugin.DeleteVolume(ctx, volumeID) +} + +// TestCleanupActorVolumesDetachesFromCanceledContext covers the case where the +// caller's context is itself why actor creation failed: the volumes must still +// be deleted rather than leaked. +func TestCleanupActorVolumesDetachesFromCanceledContext(t *testing.T) { + plugin := &ctxCapturingPlugin{MockVolumePlugin: volume.NewMockVolumePlugin()} + prev := globalVolumePlugin + globalVolumePlugin = plugin + t.Cleanup(func() { globalVolumePlugin = prev }) + + volumeID, err := plugin.CreateVolume(context.Background(), "atespace-actor-data", "1Gi", "standard") + if err != nil { + t.Fatalf("CreateVolume() failed: %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + s := &Service{} + s.cleanupActorVolumes(ctx, &ateapipb.ObjectRef{Atespace: "atespace", Name: "actor"}, + []*ateapipb.ExternalVolume{{StorageVolumeId: volumeID}}) + + if got, want := len(plugin.deleted), 1; got != want { + t.Fatalf("DeleteVolume called %d times, want %d", got, want) + } + if plugin.deleteCtxErr != nil { + t.Errorf("DeleteVolume ran with a done context: %v", plugin.deleteCtxErr) + } +}