Move DeleteTag orchestration to a workflow (#1720)

The tag deletion logic is getting too complex, so let's move it to a
workflow, following what we do for other resources.

#1510 #664

- [x] Tests pass
- [x] Appropriate changes to documentation are included in the PR

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) {

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.

cmd/ateapi/internal/controlapi/workflow.go

@@ -162,6 +162,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)
 }

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) {

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())