From 9dc82473936c273694c8d2aee4c8cdc42632aaed Mon Sep 17 00:00:00 2001 From: mesutoezdil Date: Sat, 25 Jul 2026 12:08:01 +0200 Subject: [PATCH] fix: cleanup uses fresh context after volume create failure Both cleanup paths for a half-created actor reused the caller's context. When a canceled or expired context is what made creation fail, every delete failed immediately and the external volumes leaked. Route both sites through cleanupActorVolumes, which detaches from the caller's context and applies its own 30s timeout. --- .../internal/controlapi/create_actor.go | 2 +- cmd/ateapi/internal/controlapi/volumes.go | 19 +++++- .../internal/controlapi/volumes_test.go | 67 +++++++++++++++++++ 3 files changed, 85 insertions(+), 3 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 d022b70d8..4cb7c999b 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) + } +}