From 4ef2c634dae48e995faecb840e61704c9ddd4e6a Mon Sep 17 00:00:00 2001 From: Luiz Oliveira Date: Thu, 17 Sep 2026 16:57:15 -0400 Subject: [PATCH] Move DeleteTag orchestration to a workflow The tag deletion logic is getting too complex, so let's move it to a workflow, following what we do for other resources. --- cmd/ateapi/internal/controlapi/tag.go | 64 +---------- cmd/ateapi/internal/controlapi/tag_test.go | 93 --------------- cmd/ateapi/internal/controlapi/workflow.go | 1 + .../internal/controlapi/workflow_tag.go | 93 +++++++++++++++ .../internal/controlapi/workflow_tag_test.go | 106 +++++++++++++++++- 5 files changed, 198 insertions(+), 159 deletions(-) diff --git a/cmd/ateapi/internal/controlapi/tag.go b/cmd/ateapi/internal/controlapi/tag.go index 13037e1602..93928e3977 100644 --- a/cmd/ateapi/internal/controlapi/tag.go +++ b/cmd/ateapi/internal/controlapi/tag.go @@ -20,7 +20,6 @@ import ( "fmt" "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" - "github.com/agent-substrate/substrate/internal/objectstore" "github.com/agent-substrate/substrate/internal/resources" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" "google.golang.org/grpc/codes" @@ -245,71 +244,12 @@ func ValidateCustom_UpdateTagRequest_Tag(ctx context.Context, op operation.Opera return errs } -// DeleteTag releases the external snapshot the tag owns and then -// removes the row, in that order: the row is the only handle on that snapshot, -// so dropping it first would leak. A failure at any point fails the whole RPC; -// the client retries the same delete, which rediscovers the work from the row -// and resumes over whatever is left. -// -// The tag stays resolvable while its snapshot is being collected, so a -// CreateActor racing this delete can seed an Actor from content that is going -// away. That race is accepted for now. -// -// Note that this destroys the external snapshot: an Actor created from the tag -// and never suspended is still borrowing it and becomes unrecoverable. Do not -// delete a tag while clones of it exist. +// DeleteTag removes the tag and collects the external snapshot it owns. func (s *RPCService) DeleteTag(ctx context.Context, req *ateapipb.DeleteTagRequest) (*ateapipb.Tag, error) { - // TODO: mode delete orchestration to a workflow. if errs := validateDeleteTagRequest(ctx, req); len(errs) > 0 { return nil, toGRPCStatusError(errs) } - tagRef := resources.TagRefFromObjectRef(req.GetTag()) - - // Serializes against a create of the same tag, whose copy would otherwise - // keep writing into the prefix this is collecting. - ctx, lease, err := acquireTagLease(ctx, s.impl, tagRef) - if err != nil { - return nil, err - } - defer lease.Close() - - stored, err := s.impl.GetTag(ctx, tagRef) - if err != nil { - if errors.Is(err, store.ErrNotFound) { - return nil, status.Errorf(codes.NotFound, "Tag %s/%s not found", tagRef.Atespace, tagRef.Name) - } - return nil, fmt.Errorf("while getting tag: %w", err) - } - if err := s.releaseTagSnapshot(ctx, stored); err != nil { - return nil, err - } - - tag, err := s.impl.DeleteTag(ctx, tagRef) - if errors.Is(err, store.ErrNotFound) { - return nil, status.Errorf(codes.NotFound, "Tag %s/%s not found", tagRef.Atespace, tagRef.Name) - } - if err != nil { - return nil, fmt.Errorf("while deleting tag: %w", err) - } - return tag, nil -} - -// releaseTagSnapshot deletes the objects the tag's external snapshot is made -// of. It tolerates a partly-collected snapshot, so a retry finishes cleanly. -// It collects the in-progress snapshot too. -func (s *RPCService) releaseTagSnapshot(ctx context.Context, tag *ateapipb.Tag) error { - if s.objectStore == nil { - return nil - } - atespace, name := tag.GetMetadata().GetAtespace(), tag.GetMetadata().GetName() - uri, err := resources.NewTagSnapshotURI(tag.GetStatus().GetStorageLocation(), atespace, tag.GetMetadata().GetUid()) - if err != nil { - return fmt.Errorf("while resolving the external snapshot of tag %s/%s: %w", atespace, name, err) - } - if err := objectstore.DeletePrefix(ctx, s.objectStore, uri.Prefix()); err != nil { - return fmt.Errorf("while releasing the external snapshot %q of tag %s/%s: %w", uri, atespace, name, err) - } - return nil + return s.actorWorkflow.DeleteTag(ctx, resources.TagRefFromObjectRef(req.GetTag())) } func (s *ServiceImpl) DeleteTag(ctx context.Context, tagRef resources.TagRef) (*ateapipb.Tag, error) { diff --git a/cmd/ateapi/internal/controlapi/tag_test.go b/cmd/ateapi/internal/controlapi/tag_test.go index 69c6ad52d3..2433f8b0a4 100644 --- a/cmd/ateapi/internal/controlapi/tag_test.go +++ b/cmd/ateapi/internal/controlapi/tag_test.go @@ -16,7 +16,6 @@ package controlapi import ( "context" - "errors" "strings" "testing" @@ -28,7 +27,6 @@ import ( "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" "github.com/agent-substrate/substrate/cmd/ateapi/internal/store/storetest" - "github.com/agent-substrate/substrate/internal/objectstore/objectstoretest" "github.com/agent-substrate/substrate/internal/resources" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" ) @@ -938,97 +936,6 @@ func TestUpdateTag_ConcurrentUpdate(t *testing.T) { } } -// TestDeleteTag_ReleasesExternalSnapshot verifies the delete -// collects the external snapshot the tag owns before dropping the row that -// names it, and that a failure to collect leaves the row intact so a retry can -// finish the job. -func TestDeleteTag_ReleasesExternalSnapshot(t *testing.T) { - ctx := context.Background() - persistence := newTestPersistence(t) - template := seedSubstrateTemplate(t, ctx, persistence, "sub-tmpl") - w, objects := newFinalizeWorkflow(persistence) - actor, _ := seedTagSource(t, ctx, persistence, objects, template, "actor-1", "manifest.json", "memory.zst") - tag, err := w.TagActorSnapshot(ctx, tagToCreate(resources.ActorRefFromActor(actor), "v1")) - if err != nil { - t.Fatalf("TagActorSnapshot: %v", err) - } - tagRef := resources.TagRefFromTag(tag) - uri := mustReservedTagSnapshotURI(t, tag) - svc := &RPCService{impl: newServiceImpl(persistence, nil), objectStore: objects} - - // A delete that cannot reach object storage must not drop the row: it is - // the only handle left on the snapshot. - objects.OnDelete = func(string, string) error { return errObjectStore } - req := &ateapipb.DeleteTagRequest{Tag: tagRef.ToObjectRef()} - if _, err := svc.DeleteTag(ctx, req); !errors.Is(err, errObjectStore) { - t.Fatalf("DeleteTag = %v, want an error wrapping %v", err, errObjectStore) - } - if _, err := persistence.GetTag(ctx, tagRef); err != nil { - t.Fatalf("GetTag after the failure: %v", err) - } - - // Simulates a retried deletion. Now, the object deletion succeeds, - // so we can remove the row from the DB. - objects.OnDelete = nil - if _, err := svc.DeleteTag(ctx, req); err != nil { - t.Fatalf("retried DeleteTag: %v", err) - } - if got := objects.Snapshot(t, uri); len(got) != 0 { - t.Errorf("the tag's external snapshot still holds %v, want it collected", got) - } - if _, err := persistence.GetTag(ctx, tagRef); !errors.Is(err, store.ErrNotFound) { - t.Errorf("GetTag after the delete = %v, want ErrNotFound", err) - } -} - -// TestDeleteTag_ReleasesPendingSnapshot verifies that deleting a -// tag whose create never finished collects what that create stranded. The -// pending row names the prefix the copy was writing into, and it is the only -// handle left on those objects. -func TestDeleteTag_ReleasesPendingSnapshot(t *testing.T) { - ctx := context.Background() - persistence, cleanup := storetest.SetupTestStore(t) - t.Cleanup(cleanup) - - actor := newTestSuspendedActor(t, ctx, persistence, testAtespace, "actor-1") - tag := storetest.MustCreateTag(t, ctx, persistence, newPendingTestTag(t, "v1", actor)) - tagRef := resources.TagRefFromTag(tag) - - objects := objectstoretest.New() - uri, err := resources.NewTagSnapshotURI(tag.GetStatus().GetStorageLocation(), tag.GetMetadata().GetAtespace(), tag.GetMetadata().GetUid()) - if err != nil { - t.Fatalf("NewTagSnapshotURI: %v", err) - } - // What a copy that died halfway through left behind. - objects.PutSnapshot(t, uri, "manifest.json") - svc := &RPCService{impl: newServiceImpl(persistence, nil), objectStore: objects} - - // Cleanup must work without the source actor or its template. - mustUpdateActorStatus(t, ctx, persistence, actor, func(s *ateapipb.ActorStatus) { - s.State = ateapipb.ActorState_ACTOR_STATE_DELETING - }) - if _, err := persistence.DeleteActor(ctx, resources.ActorRefFromActor(actor)); err != nil { - t.Fatalf("DeleteActor: %v", err) - } - objects.OnDelete = func(string, string) error { return errObjectStore } - if _, err := svc.DeleteTag(ctx, &ateapipb.DeleteTagRequest{Tag: tagRef.ToObjectRef()}); !errors.Is(err, errObjectStore) { - t.Fatalf("DeleteTag = %v, want an error wrapping %v", err, errObjectStore) - } - if _, err := persistence.GetTag(ctx, tagRef); err != nil { - t.Fatalf("GetTag after failed cleanup: %v", err) - } - objects.OnDelete = nil - if _, err := svc.DeleteTag(ctx, &ateapipb.DeleteTagRequest{Tag: tagRef.ToObjectRef()}); err != nil { - t.Fatalf("DeleteTag: %v", err) - } - if got := objects.Snapshot(t, uri); len(got) != 0 { - t.Errorf("the pending tag's stranded objects are still %v, want them collected", got) - } - if _, err := persistence.GetTag(ctx, tagRef); !errors.Is(err, store.ErrNotFound) { - t.Errorf("GetTag after the delete = %v, want ErrNotFound", err) - } -} - // TestUpdateTag_PendingTag verifies a tag whose create never // finished cannot be published: it names a copy that may be partial, so it must // not become usable until the create completes. diff --git a/cmd/ateapi/internal/controlapi/workflow.go b/cmd/ateapi/internal/controlapi/workflow.go index 1057169ba3..4474119f3b 100644 --- a/cmd/ateapi/internal/controlapi/workflow.go +++ b/cmd/ateapi/internal/controlapi/workflow.go @@ -160,6 +160,7 @@ type actorWorkflowStore interface { CreateTag(ctx context.Context, tag *ateapipb.Tag) (*ateapipb.Tag, error) GetTag(ctx context.Context, tagRef resources.TagRef) (*ateapipb.Tag, error) UpdateTag(ctx context.Context, tagRef resources.TagRef, precondition store.Precondition, mutate func(toUpdate *ateapipb.Tag) error) (*ateapipb.Tag, error) + DeleteTag(ctx context.Context, tagRef resources.TagRef) (*ateapipb.Tag, error) GetActorTemplate(ctx context.Context, templateRef resources.ActorTemplateRef) (*ateapipb.ActorTemplate, error) AcquireLease(ctx context.Context, key string) (*store.Lease, error) } diff --git a/cmd/ateapi/internal/controlapi/workflow_tag.go b/cmd/ateapi/internal/controlapi/workflow_tag.go index 880ea9c46d..a36a168c34 100644 --- a/cmd/ateapi/internal/controlapi/workflow_tag.go +++ b/cmd/ateapi/internal/controlapi/workflow_tag.go @@ -85,6 +85,99 @@ func (w *ActorWorkflow) TagActorSnapshot(ctx context.Context, tag *ateapipb.Tag) return w.ensureTagFinalized(leaseCtx, reserved, snapshot, dst) } +// DeleteTag releases the external snapshot the tag owns and then removes the +// row, in that order: the row is the only handle on that snapshot, so dropping +// it first would leak. +// +// The workflow is built in 3 phases: +// 1. Load the tag (which names the snapshot to collect). +// 2. Release that snapshot, tolerating a previous attempt partly collected. +// 3. Finalize: drop the row. +// +// Idempotent: a failure at any phase leaves the row in place, so the same +// delete run again rediscovers the work from it and resumes over whatever is +// left. +// +// The tag stays resolvable while its snapshot is being collected, so a +// CreateActor racing this delete can seed an Actor from content that is going +// away. That race is accepted for now. +// +// Note that this destroys the external snapshot: an Actor created from the tag +// and never suspended is still borrowing it and becomes unrecoverable. Do not +// delete a tag while clones of it exist. +func (w *ActorWorkflow) DeleteTag(ctx context.Context, tagRef resources.TagRef) (*ateapipb.Tag, error) { + // Serializes against a create of the same tag, whose copy would otherwise + // keep writing into the prefix this is collecting. + ctx, lease, err := acquireTagLease(ctx, w.store, tagRef) + if err != nil { + return nil, err + } + defer lease.Close() + + tag, err := w.loadTagForDelete(ctx, tagRef) + if err != nil { + return nil, err + } + if err := w.ensureTagSnapshotReleased(ctx, tag); err != nil { + return nil, err + } + return w.finalizeTagDeleted(ctx, tagRef) +} + +// loadTagForDelete fetches the row the delete works from. The row records where +// the snapshot lives, so the work is rediscovered from it rather than rebuilt +// from the source actor, which may be long gone. +func (w *ActorWorkflow) loadTagForDelete(ctx context.Context, tagRef resources.TagRef) (_ *ateapipb.Tag, err error) { + ctx, done := stepSpan(ctx, "LoadTagForDelete") + defer func() { err = done(err) }() + + tag, err := w.store.GetTag(ctx, tagRef) + if err != nil { + if errors.Is(err, store.ErrNotFound) { + return nil, status.Errorf(codes.NotFound, "Tag %s not found", tagRef) + } + return nil, fmt.Errorf("while getting tag %s: %w", tagRef, err) + } + return tag, nil +} + +// ensureTagSnapshotReleased deletes the objects the tag's external snapshot is +// made of. It tolerates a partly-collected snapshot, so a retry finishes +// cleanly. It collects the in-progress snapshot of a pending tag too. +func (w *ActorWorkflow) ensureTagSnapshotReleased(ctx context.Context, tag *ateapipb.Tag) (err error) { + ctx, done := stepSpan(ctx, "ReleaseTagSnapshot") + defer func() { err = done(err) }() + + if w.objectStore == nil { + markSkipped(ctx, "no object store configured") + return nil + } + tagRef := resources.TagRefFromTag(tag) + uri, err := resources.NewTagSnapshotURI(tag.GetStatus().GetStorageLocation(), tagRef.Atespace, tag.GetMetadata().GetUid()) + if err != nil { + return fmt.Errorf("while resolving the external snapshot of tag %s: %w", tagRef, err) + } + if err := objectstore.DeletePrefix(ctx, w.objectStore, uri.Prefix()); err != nil { + return fmt.Errorf("while releasing the external snapshot %q of tag %s: %w", uri, tagRef, err) + } + return nil +} + +// finalizeTagDeleted drops the row, once nothing it names is left behind. +func (w *ActorWorkflow) finalizeTagDeleted(ctx context.Context, tagRef resources.TagRef) (_ *ateapipb.Tag, err error) { + ctx, done := stepSpan(ctx, "FinalizeTagDeleted") + defer func() { err = done(err) }() + + tag, err := w.store.DeleteTag(ctx, tagRef) + if err != nil { + if errors.Is(err, store.ErrNotFound) { + return nil, status.Errorf(codes.NotFound, "Tag %s not found", tagRef) + } + return nil, fmt.Errorf("while deleting tag %s: %w", tagRef, err) + } + return tag, nil +} + // loadActorForTag fetches the actor to tag and its template, and checks that // the actor holds an external snapshot a tag can be made from. func (w *ActorWorkflow) loadActorForTag(ctx context.Context, actorRef resources.ActorRef) (_ *ateapipb.Actor, _ *ateapipb.ActorTemplate, err error) { diff --git a/cmd/ateapi/internal/controlapi/workflow_tag_test.go b/cmd/ateapi/internal/controlapi/workflow_tag_test.go index 3e239860bc..668dcd1033 100644 --- a/cmd/ateapi/internal/controlapi/workflow_tag_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_tag_test.go @@ -226,7 +226,6 @@ func TestTagActorSnapshot_RecreateAfterCopyFailure(t *testing.T) { persistence := newTestPersistence(t) template := seedSubstrateTemplate(t, ctx, persistence, "sub-tmpl") w, objects := newFinalizeWorkflow(persistence) - svc := &RPCService{impl: persistence, objectStore: objects} actor, _ := seedTagSource(t, ctx, persistence, objects, template, "actor-1", "manifest.json", "memory.zst") actorRef := resources.ActorRefFromActor(actor) @@ -275,7 +274,7 @@ func TestTagActorSnapshot_RecreateAfterCopyFailure(t *testing.T) { } // Deleting the tag collects the partial copy and frees the name. - if _, err := svc.DeleteTag(ctx, &ateapipb.DeleteTagRequest{Tag: tagRef.ToObjectRef()}); err != nil { + if _, err := w.DeleteTag(ctx, tagRef); err != nil { t.Fatalf("DeleteTag on the pending tag: %v", err) } if got := objects.Snapshot(t, strandedURI); len(got) != 0 { @@ -312,7 +311,6 @@ func TestTagActorSnapshot_RacesDelete(t *testing.T) { persistence := newTestPersistence(t) template := seedSubstrateTemplate(t, ctx, persistence, "sub-tmpl") w, objects := newFinalizeWorkflow(persistence) - svc := &RPCService{impl: persistence, objectStore: objects} actor, _ := seedTagSource(t, ctx, persistence, objects, template, "actor-1", "manifest.json", "memory.zst") actorRef := resources.ActorRefFromActor(actor) @@ -331,7 +329,7 @@ func TestTagActorSnapshot_RacesDelete(t *testing.T) { return } pendingURI = mustReservedTagSnapshotURI(t, reserved).String() - _, deleteErr = svc.DeleteTag(ctx, &ateapipb.DeleteTagRequest{Tag: tagRef.ToObjectRef()}) + _, deleteErr = w.DeleteTag(ctx, tagRef) }) return nil } @@ -415,6 +413,106 @@ func TestTagActorSnapshot_NameTakenByAnotherActor(t *testing.T) { } } +// TestDeleteTag_ReleasesExternalSnapshot verifies the delete collects the +// external snapshot the tag owns before dropping the row that names it, and +// that a failure to collect leaves the row intact so a retry can finish the job. +func TestDeleteTag_ReleasesExternalSnapshot(t *testing.T) { + ctx := context.Background() + persistence := newTestPersistence(t) + template := seedSubstrateTemplate(t, ctx, persistence, "sub-tmpl") + w, objects := newFinalizeWorkflow(persistence) + actor, _ := seedTagSource(t, ctx, persistence, objects, template, "actor-1", "manifest.json", "memory.zst") + tag, err := w.TagActorSnapshot(ctx, tagToCreate(resources.ActorRefFromActor(actor), "v1")) + if err != nil { + t.Fatalf("TagActorSnapshot: %v", err) + } + tagRef := resources.TagRefFromTag(tag) + uri := mustReservedTagSnapshotURI(t, tag) + + // A delete that cannot reach object storage must not drop the row: it is + // the only handle left on the snapshot. + objects.OnDelete = func(string, string) error { return errObjectStore } + if _, err := w.DeleteTag(ctx, tagRef); !errors.Is(err, errObjectStore) { + t.Fatalf("DeleteTag = %v, want an error wrapping %v", err, errObjectStore) + } + if _, err := persistence.GetTag(ctx, tagRef); err != nil { + t.Fatalf("GetTag after the failure: %v", err) + } + + // Simulates a retried deletion. Now, the object deletion succeeds, + // so we can remove the row from the DB. + objects.OnDelete = nil + if _, err := w.DeleteTag(ctx, tagRef); err != nil { + t.Fatalf("retried DeleteTag: %v", err) + } + if got := objects.Snapshot(t, uri); len(got) != 0 { + t.Errorf("the tag's external snapshot still holds %v, want it collected", got) + } + if _, err := persistence.GetTag(ctx, tagRef); !errors.Is(err, store.ErrNotFound) { + t.Errorf("GetTag after the delete = %v, want ErrNotFound", err) + } +} + +// TestDeleteTag_ReleasesPendingSnapshot verifies that deleting a tag whose +// create never finished collects what that create stranded. The pending row +// names the prefix the copy was writing into, and it is the only handle left on +// those objects. +func TestDeleteTag_ReleasesPendingSnapshot(t *testing.T) { + ctx := context.Background() + persistence, cleanup := storetest.SetupTestStore(t) + t.Cleanup(cleanup) + + actor := newTestSuspendedActor(t, ctx, persistence, testAtespace, "actor-1") + tag := storetest.MustCreateTag(t, ctx, persistence, newPendingTestTag(t, "v1", actor)) + tagRef := resources.TagRefFromTag(tag) + + w, objects := newFinalizeWorkflow(persistence) + uri := mustReservedTagSnapshotURI(t, tag) + // What a copy that died halfway through left behind. + objects.PutSnapshot(t, uri, "manifest.json") + + // Cleanup must work without the source actor or its template. + mustUpdateActorStatus(t, ctx, persistence, actor, func(s *ateapipb.ActorStatus) { + s.State = ateapipb.ActorState_ACTOR_STATE_DELETING + }) + if _, err := persistence.DeleteActor(ctx, resources.ActorRefFromActor(actor)); err != nil { + t.Fatalf("DeleteActor: %v", err) + } + objects.OnDelete = func(string, string) error { return errObjectStore } + if _, err := w.DeleteTag(ctx, tagRef); !errors.Is(err, errObjectStore) { + t.Fatalf("DeleteTag = %v, want an error wrapping %v", err, errObjectStore) + } + if _, err := persistence.GetTag(ctx, tagRef); err != nil { + t.Fatalf("GetTag after failed cleanup: %v", err) + } + objects.OnDelete = nil + if _, err := w.DeleteTag(ctx, tagRef); err != nil { + t.Fatalf("DeleteTag: %v", err) + } + if got := objects.Snapshot(t, uri); len(got) != 0 { + t.Errorf("the pending tag's stranded objects are still %v, want them collected", got) + } + if _, err := persistence.GetTag(ctx, tagRef); !errors.Is(err, store.ErrNotFound) { + t.Errorf("GetTag after the delete = %v, want ErrNotFound", err) + } +} + +// TestDeleteTag_NotFound verifies a delete of a tag that is not there is +// NotFound rather than a silent success: the client asked to collect a snapshot +// no row names. +func TestDeleteTag_NotFound(t *testing.T) { + ctx := context.Background() + persistence, cleanup := storetest.SetupTestStore(t) + t.Cleanup(cleanup) + storetest.MustCreateAtespace(t, ctx, persistence, testAtespace) + w, _ := newFinalizeWorkflow(persistence) + + _, err := w.DeleteTag(ctx, resources.TagRef{Atespace: testAtespace, Name: "missing"}) + if code := status.Code(err); code != codes.NotFound { + t.Errorf("DeleteTag = %v (code %v), want code NotFound", err, code) + } +} + func mustReservedTagSnapshotURI(t *testing.T, tag *ateapipb.Tag) resources.SnapshotURI { t.Helper() uri, err := resources.NewTagSnapshotURI(tag.GetStatus().GetStorageLocation(), tag.GetMetadata().GetAtespace(), tag.GetMetadata().GetUid())