Use Tag UIDs in snapshot storage paths

cmd/ateapi/internal/controlapi/functionaltest/actor_test.go

@@ -356,7 +356,7 @@ func TestCreateActor_PendingTag(t *testing.T) {
 	// Simulates a tag creation that failed in while writing the snapshot to
 	// external storage.
 	pending := seedTag(t, tc, "pending-source", "pending", func(tag *ateapipb.Tag) {
-		tag.Status.InProgressSnapshotUri = tag.GetStatus().GetSnapshot().GetSnapshotUri()
+		tag.Status.StorageLocation = testStorageLocation
 		tag.Status.Snapshot = nil
 		tag.Status.ActorTemplateUid = tmpl.GetMetadata().GetUid()
 	})
@@ -371,15 +371,17 @@ func TestCreateActor_PendingTag(t *testing.T) {
 	assertGrpcError(t, err, codes.FailedPrecondition, "source Tag is still being created or failed creation")
 
 	// Finishing the tag creation, so now the tag is qualified to be a tag source.
-	snapshotURI := pending.GetStatus().GetInProgressSnapshotUri()
+	snapshotURI, err := resources.NewTagSnapshotURI(pending.GetStatus().GetStorageLocation(), pending.GetMetadata().GetAtespace(), pending.GetMetadata().GetUid())
+	if err != nil {
+		t.Fatalf("NewTagSnapshotURI: %v", err)
+	}
 	if _, err := tc.persistence.UpdateTag(ctx,
 		resources.TagRefFromTag(pending), store.PreconditionFrom(pending),
 		func(toUpdate *ateapipb.Tag) error {
 			toUpdate.Status.Snapshot = &ateapipb.ExternalSnapshot{
-				SnapshotUri:  snapshotURI,
+				SnapshotUri:  snapshotURI.String(),
 				ContentScope: ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL,
 			}
-			toUpdate.Status.InProgressSnapshotUri = ""
 			return nil
 		}); err != nil {
 		t.Fatalf("finalizing the tag: %v", err)
@@ -396,7 +398,7 @@ func TestCreateActor_PendingTag(t *testing.T) {
 	}
 	// The clone points at the tag's snapshot, under the tag's own prefix: the
 	// tag still owns those objects.
-	if got := clone.GetStatus().GetExternalSnapshot().GetSnapshotUri(); got != snapshotURI {
+	if got := clone.GetStatus().GetExternalSnapshot().GetSnapshotUri(); got != snapshotURI.String() {
 		t.Errorf("clone external snapshot = %q, want the tag's %q", got, snapshotURI)
 	}
 }
@@ -1375,7 +1377,7 @@ func TestActorLifecycle_WithExternalVolumes(t *testing.T) {
 		},
 	}
 	createTemplateWithVolumes(t, tc, ns, volumes, mounts)
-	createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1")
+	workerName := createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1")
 
 	// 1. CreateActor
 	createResp, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
@@ -1423,6 +1425,7 @@ func TestActorLifecycle_WithExternalVolumes(t *testing.T) {
 	}
 
 	// 4. ResumeActor from paused
+	waitForWorkerAvailable(t, tc, workerName)
 	resumeResp2, err := tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{
 		Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "actor-vol-lc"},
 	})
@@ -2213,6 +2216,7 @@ func TestSuspendActor(t *testing.T) {
 			Snapshot:         &ateapipb.ExternalSnapshot{SnapshotUri: tagSnapshotURI, ContentScope: sourceActor.GetStatus().GetExternalSnapshot().GetContentScope()},
 			ActorTemplateUid: tmpl.GetMetadata().GetUid(),
 			SourceActorUid:   sourceActor.GetMetadata().GetUid(),
+			StorageLocation:  tmpl.GetSnapshotsConfig().GetStorageLocation(),
 		},
 	}
 	stored, err := tc.client.GetTag(context.Background(), &ateapipb.GetTagRequest{Tag: tagRef})

cmd/ateapi/internal/controlapi/functionaltest/tag_test.go

@@ -53,24 +53,38 @@ func seedTag(t *testing.T, tc *testContext, actorName, tagName string, opts ...f
 	if err != nil {
 		t.Fatalf("recording the actor's external snapshot: %v", err)
 	}
-	// The tag owns its own copy, under the tag prefix rather than the actor's.
-	tagSnapshotURI, err := resources.NewTagSnapshotURI(testStorageLocation, testAtespace, tagName)
-	if err != nil {
-		t.Fatalf("NewTagSnapshotURI: %v", err)
-	}
 	tag := &ateapipb.Tag{
 		Metadata:    &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: tagName},
 		Scope:       ateapipb.TagScope_TAG_SCOPE_ATESPACE,
 		SourceActor: resources.ActorRefFromActor(actor).ToObjectRef(),
 		Status: &ateapipb.TagStatus{
-			Snapshot:       &ateapipb.ExternalSnapshot{SnapshotUri: tagSnapshotURI.String(), ContentScope: actor.GetStatus().GetExternalSnapshot().GetContentScope()},
-			SourceActorUid: actor.GetMetadata().GetUid(),
+			Snapshot:        &ateapipb.ExternalSnapshot{ContentScope: actor.GetStatus().GetExternalSnapshot().GetContentScope()},
+			StorageLocation: testStorageLocation,
+			SourceActorUid:  actor.GetMetadata().GetUid(),
 		},
 	}
 	for _, opt := range opts {
 		opt(tag)
 	}
-	return storetest.MustCreateTag(t, ctx, tc.persistence, tag)
+	snapshot := tag.Status.Snapshot
+	tag.Status.Snapshot = nil
+	tag = storetest.MustCreateTag(t, ctx, tc.persistence, tag)
+	if snapshot == nil {
+		return tag
+	}
+	uri, err := resources.NewTagSnapshotURI(testStorageLocation, testAtespace, tag.GetMetadata().GetUid())
+	if err != nil {
+		t.Fatalf("NewTagSnapshotURI: %v", err)
+	}
+	snapshot.SnapshotUri = uri.String()
+	tag, err = tc.persistence.UpdateTag(ctx, resources.TagRefFromTag(tag), store.PreconditionFrom(tag), func(toUpdate *ateapipb.Tag) error {
+		toUpdate.Status.Snapshot = snapshot
+		return nil
+	})
+	if err != nil {
+		t.Fatalf("finalizing tag: %v", err)
+	}
+	return tag
 }
 
 // TestCreateTag_ReusedTagName tests that a tag does not move

cmd/ateapi/internal/controlapi/tag.go

@@ -299,20 +299,12 @@ func (s *RPCService) releaseTagSnapshot(ctx context.Context, tag *ateapipb.Tag)
 		return nil
 	}
 	atespace, name := tag.GetMetadata().GetAtespace(), tag.GetMetadata().GetName()
-	for _, snapshotURI := range []string{
-		tag.GetStatus().GetSnapshot().GetSnapshotUri(),
-		tag.GetStatus().GetInProgressSnapshotUri(),
-	} {
-		if snapshotURI == "" {
-			continue
-		}
-		uri, err := resources.ParseSnapshotURI(snapshotURI)
-		if err != nil {
-			return fmt.Errorf("while parsing the external snapshot %q of tag %s/%s: %w", snapshotURI, 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", snapshotURI, atespace, name, err)
-		}
+	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
 }

cmd/ateapi/internal/controlapi/tag_test.go

@@ -814,7 +814,7 @@ func newTestTag(t *testing.T, name string, actor *ateapipb.Actor) *ateapipb.Tag
 func newPendingTestTag(t *testing.T, name string, actor *ateapipb.Actor) *ateapipb.Tag {
 	t.Helper()
 	tag := newTestTag(t, name, actor)
-	tag.Status.InProgressSnapshotUri = tag.GetStatus().GetSnapshot().GetSnapshotUri()
+	tag.Status.StorageLocation = testStorageLocation
 	tag.Status.Snapshot = nil
 	return tag
 }
@@ -944,19 +944,16 @@ func TestUpdateTag_ConcurrentUpdate(t *testing.T) {
 // finish the job.
 func TestDeleteTag_ReleasesExternalSnapshot(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, newTestTag(t, "v1", actor))
-	tagRef := resources.TagRefFromTag(tag)
-
-	objects := objectstoretest.New()
-	uri, err := resources.ParseSnapshotURI(tag.GetStatus().GetSnapshot().GetSnapshotUri())
+	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("ParseSnapshotURI: %v", err)
+		t.Fatalf("TagActorSnapshot: %v", err)
 	}
-	objects.PutSnapshot(t, uri, "manifest.json", "memory.zst")
+	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
@@ -998,14 +995,29 @@ func TestDeleteTag_ReleasesPendingSnapshot(t *testing.T) {
 	tagRef := resources.TagRefFromTag(tag)
 
 	objects := objectstoretest.New()
-	uri, err := resources.ParseSnapshotURI(tag.GetStatus().GetInProgressSnapshotUri())
+	uri, err := resources.NewTagSnapshotURI(tag.GetStatus().GetStorageLocation(), tag.GetMetadata().GetAtespace(), tag.GetMetadata().GetUid())
 	if err != nil {
-		t.Fatalf("ParseSnapshotURI: %v", err)
+		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)
 	}

cmd/ateapi/internal/controlapi/workflow_tag.go

@@ -33,10 +33,9 @@ import (
 // or deleting the actor does not garbage collect the tag's snapshot.
 //
 // The tag is built in 3 phases:
-//  1. Record intent: in status.in_progress_snapshot_uri
-//  2. Write snapshot to the remote storage indicated in status.in_progress_snapshot_uri
-//  3. Finalize: clear status.in_progress_snapshot_uri and write the final
-//     snapshot object to the tag.
+//  1. Reserve the tag and record its storage location.
+//  2. Copy the snapshot under the reserved tag's UID.
+//  3. Finalize: write the completed snapshot object to the tag.
 //
 // The tag captures whichever snapshot the actor holds when the workflow runs.
 // An actor keeps no snapshot history, so a suspend that lands first moves what
@@ -76,10 +75,14 @@ func (w *ActorWorkflow) TagActorSnapshot(ctx context.Context, tag *ateapipb.Tag)
 	if err != nil {
 		return nil, err
 	}
-	if err := w.ensureTagSnapshotCopied(leaseCtx, reserved, snapshot); err != nil {
+	dst, err := resources.NewTagSnapshotURI(reserved.GetStatus().GetStorageLocation(), tagRef.Atespace, reserved.GetMetadata().GetUid())
+	if err != nil {
+		return nil, fmt.Errorf("while building the snapshot URI for tag %s: %w", tagRef, err)
+	}
+	if err := w.ensureTagSnapshotCopied(leaseCtx, reserved, snapshot, dst); err != nil {
 		return nil, err
 	}
-	return w.ensureTagFinalized(leaseCtx, reserved, snapshot)
+	return w.ensureTagFinalized(leaseCtx, reserved, snapshot, dst)
 }
 
 // loadActorForTag fetches the actor to tag and its template, and checks that
@@ -115,8 +118,7 @@ func (w *ActorWorkflow) loadActorForTag(ctx context.Context, actorRef resources.
 	return actor, actorTemplate, nil
 }
 
-// ensureTagReserved takes the tag's name and sets status.in_progress_snapshot_uri
-// to express the intent to create a new tag.
+// ensureTagReserved takes the tag's name and records the storage location.
 //
 // A name already taken is AlreadyExists, whether the tag holding it is finished
 // or was left pending by a create that died. Resuming a pending row would mean
@@ -127,9 +129,9 @@ func (w *ActorWorkflow) ensureTagReserved(ctx context.Context, tagRef resources.
 	ctx, done := stepSpan(ctx, "ReserveTag")
 	defer func() { err = done(err) }()
 
-	dst, err := resources.NewTagSnapshotURI(actorTemplate.GetSnapshotsConfig().GetStorageLocation(), tagRef.Atespace, resources.NewSnapshotName())
-	if err != nil {
-		return nil, fmt.Errorf("while building the snapshot URI for tag %s: %w", tagRef, err)
+	location := actorTemplate.GetSnapshotsConfig().GetStorageLocation()
+	if err := resources.ValidateSnapshotLocation(location); err != nil {
+		return nil, fmt.Errorf("invalid storage location for tag %s: %w", tagRef, err)
 	}
 	tagToCreate := &ateapipb.Tag{
 		Metadata:    &ateapipb.ResourceMetadata{Atespace: tagRef.Atespace, Name: tagRef.Name},
@@ -141,9 +143,9 @@ func (w *ActorWorkflow) ensureTagReserved(ctx context.Context, tagRef resources.
 			// and a tag that claimed the new template would hand clones the old template's
 			// memory under the new one's identity, past the data-only downgrade a resume of
 			// the actor itself would take.
-			ActorTemplateUid:      actor.GetStatus().GetCurrentActorTemplateUid(),
-			InProgressSnapshotUri: dst.String(),
-			SourceActorUid:        actor.GetMetadata().GetUid(),
+			ActorTemplateUid: actor.GetStatus().GetCurrentActorTemplateUid(),
+			StorageLocation:  location,
+			SourceActorUid:   actor.GetMetadata().GetUid(),
 		},
 	}
 
@@ -160,10 +162,9 @@ func (w *ActorWorkflow) ensureTagReserved(ctx context.Context, tagRef resources.
 }
 
 // ensureTagSnapshotCopied copies the actor's external snapshot to the tag's own
-// prefix, the one the reserved row names. That prefix is freshly minted, so it
-// is empty by construction and the copy never blends with objects some other
-// attempt left.
-func (w *ActorWorkflow) ensureTagSnapshotCopied(ctx context.Context, tag *ateapipb.Tag, snapshot *ateapipb.ExternalSnapshot) (err error) {
+// prefix, derived from the reserved row's freshly minted UID. The prefix is
+// empty by construction, so the copy never blends with another attempt's objects.
+func (w *ActorWorkflow) ensureTagSnapshotCopied(ctx context.Context, tag *ateapipb.Tag, snapshot *ateapipb.ExternalSnapshot, dst resources.SnapshotURI) (err error) {
 	ctx, done := stepSpan(ctx, "CopyTagSnapshot")
 	defer func() { err = done(err) }()
 
@@ -176,33 +177,26 @@ func (w *ActorWorkflow) ensureTagSnapshotCopied(ctx context.Context, tag *ateapi
 	if err != nil {
 		return fmt.Errorf("while parsing the external snapshot %q of the source actor: %w", snapshot.GetSnapshotUri(), err)
 	}
-	dst, err := resources.ParseSnapshotURI(tag.GetStatus().GetInProgressSnapshotUri())
-	if err != nil {
-		return fmt.Errorf("while parsing the in-progress snapshot %q of tag %s: %w", tag.GetStatus().GetInProgressSnapshotUri(), tagRef, err)
-	}
 	if err := objectstore.CopyPrefix(ctx, w.objectStore, src.Prefix(), dst.Prefix()); err != nil {
 		return fmt.Errorf("while copying the external snapshot for tag %s: %w", tagRef, err)
 	}
 	return nil
 }
 
-// ensureTagFinalized publishes the copy: it sets status.snapshot to the URI the
-// row already names and clears status.in_progress_snapshot_uri. Until this
-// lands the tag is pending — visible, unusable, and naming exactly the objects
-// an unfinished create stranded, so deleting it collects them.
-func (w *ActorWorkflow) ensureTagFinalized(ctx context.Context, tag *ateapipb.Tag, snapshot *ateapipb.ExternalSnapshot) (_ *ateapipb.Tag, err error) {
+// ensureTagFinalized publishes the copy by setting status.snapshot. Until this
+// lands the tag is pending and unusable; deleting it collects any partial copy.
+func (w *ActorWorkflow) ensureTagFinalized(ctx context.Context, tag *ateapipb.Tag, snapshot *ateapipb.ExternalSnapshot, dst resources.SnapshotURI) (_ *ateapipb.Tag, err error) {
 	ctx, done := stepSpan(ctx, "FinalizeTag")
 	defer func() { err = done(err) }()
 
 	tagRef := resources.TagRefFromTag(tag)
 	// The copy is byte-identical to the source, so it carries the same content.
 	finalSnapshot := &ateapipb.ExternalSnapshot{
-		SnapshotUri:  tag.GetStatus().GetInProgressSnapshotUri(),
+		SnapshotUri:  dst.String(),
 		ContentScope: snapshot.GetContentScope(),
 	}
 	stored, err := w.store.UpdateTag(ctx, tagRef, store.PreconditionFrom(tag), func(toUpdate *ateapipb.Tag) error {
 		toUpdate.Status.Snapshot = finalSnapshot
-		toUpdate.Status.InProgressSnapshotUri = ""
 		return nil
 	})
 	if err != nil {