Store the in-progress snapshot as a URI instead of a name

cmd/ateapi/internal/controlapi/actor_test.go

@@ -400,15 +400,19 @@ func TestValidateActorUpdate(t *testing.T) {
 			field.Invalid(field.NewPath("status", "worker_assignment", "worker_pod_ip"), nil, "").WithOrigin("format=ip-strict"),
 		},
 	}, {
-		"valid actor.status.in_progress_snapshot_name",
+		"valid actor.status.in_progress_snapshot_uri",
 		validInput(),
-		validOutput(withStatus(func(s *ateapipb.ActorStatus) { s.InProgressSnapshotName = "snap-1" })),
+		validOutput(withStatus(func(s *ateapipb.ActorStatus) {
+			s.InProgressSnapshotUri = "gs://private/atespaces/as/actors/" + someActorUID + "/snapshots/snap-1"
+		})),
 		nil,
 	}, {
-		"invalid actor.status.in_progress_snapshot_name",
+		"invalid actor.status.in_progress_snapshot_uri: too long",
 		validInput(),
-		validOutput(withStatus(func(s *ateapipb.ActorStatus) { s.InProgressSnapshotName = "SNAP 1" })),
-		field.ErrorList{field.Invalid(field.NewPath("status", "in_progress_snapshot_name"), nil, "").WithOrigin("format=k8s-short-name")},
+		validOutput(withStatus(func(s *ateapipb.ActorStatus) {
+			s.InProgressSnapshotUri = "gs://" + strings.Repeat("x", 2044)
+		})),
+		field.ErrorList{field.TooLong(field.NewPath("status", "in_progress_snapshot_uri"), nil, 2048).WithOrigin("maxLength")},
 	}, {
 		"valid actor.status.external_snapshot",
 		validInput(),

cmd/ateapi/internal/controlapi/crash.go

@@ -94,7 +94,7 @@ func crashActor(ctx context.Context, st crashActorStore, actorRef resources.Acto
 	_, err = st.UpdateActor(ctx, actorRef, store.PreconditionFrom(actor), func(toUpdate *ateapipb.Actor) error {
 		toUpdate.Status.State = ateapipb.ActorState_ACTOR_STATE_CRASHED
 
-		// InProgressSnapshotName and InProgressLocalSnapshotName are kept for
+		// InProgressSnapshotUri and InProgressLocalSnapshotName are kept for
 		// debugging; failed workflow steps must never promote either of them to an
 		// ActorSnapshot or to LocalSnapshotInfo.
 		toUpdate.Status.WorkerAssignment = nil

cmd/ateapi/internal/controlapi/crash_test.go

@@ -51,7 +51,7 @@ func seedActor(t *testing.T, ctx context.Context, st store.Interface, actorRef r
 				WorkerPodUid:    "uid",
 				WorkerPodIp:     "1.2.3.4",
 			},
-			InProgressSnapshotName: "reserved-snapshot",
+			InProgressSnapshotUri: "gs://bucket/atespaces/as/actors/uid/snapshots/reserved-snapshot",
 		},
 	})
 }
@@ -94,8 +94,8 @@ func seedUnboundActor(t *testing.T, ctx context.Context, st store.Interface, act
 	storetest.MustCreateActor(t, ctx, st, &ateapipb.Actor{
 		Metadata: &ateapipb.ResourceMetadata{Name: actorRef.Name, Atespace: actorRef.Atespace},
 		Status: &ateapipb.ActorStatus{
-			State:                  ateapipb.ActorState_ACTOR_STATE_RUNNING,
-			InProgressSnapshotName: "reserved-snapshot",
+			State:                 ateapipb.ActorState_ACTOR_STATE_RUNNING,
+			InProgressSnapshotUri: "gs://bucket/atespaces/as/actors/uid/snapshots/reserved-snapshot",
 		},
 	})
 }
@@ -112,8 +112,8 @@ func assertCrashed(t *testing.T, ctx context.Context, st store.Interface, actorR
 		t.Errorf("status = %v, want %v", got.GetStatus().GetState(), ateapipb.ActorState_ACTOR_STATE_CRASHED)
 	}
 	// Keep the snapshot uri for debugging.
-	if got.GetStatus().GetInProgressSnapshotName() == "" {
-		t.Error(`InProgressSnapshotName = "", want preserved`)
+	if got.GetStatus().GetInProgressSnapshotUri() == "" {
+		t.Error(`InProgressSnapshotUri = "", want preserved`)
 	}
 	if got.GetStatus().GetWorkerAssignment() != nil {
 		t.Errorf("WorkerAssignment = %v, want cleared", got.GetStatus().GetWorkerAssignment())

cmd/ateapi/internal/controlapi/workflow_delete.go

@@ -87,7 +87,7 @@ func (w *ActorWorkflow) DeleteActor(ctx context.Context, actorRef resources.Acto
 		errs = append(errs, fmt.Errorf("while deleting volumes: %w", err))
 	}
 
-	if err := w.ensureExternalSnapshotsReleased(ctx, actor, actorTemplate); err != nil {
+	if err := w.ensureExternalSnapshotsReleased(ctx, actor); err != nil {
 		errs = append(errs, fmt.Errorf("while releasing external snapshots: %w", err))
 	}
 
@@ -372,7 +372,7 @@ func (w *ActorWorkflow) ensureVolumesDeleted(ctx context.Context, actor *ateapip
 //
 // A snapshot borrowed from a tag lives under the tag's prefix, so it survives:
 // the tag owns it and outlives the actor.
-func (w *ActorWorkflow) ensureExternalSnapshotsReleased(ctx context.Context, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate) (err error) {
+func (w *ActorWorkflow) ensureExternalSnapshotsReleased(ctx context.Context, actor *ateapipb.Actor) (err error) {
 	ctx, done := stepSpan(ctx, "ReleaseExternalSnapshots")
 	defer func() { err = done(err) }()
 
@@ -381,7 +381,7 @@ func (w *ActorWorkflow) ensureExternalSnapshotsReleased(ctx context.Context, act
 		return nil
 	}
 
-	prefix, err := actorSnapshotStoragePrefix(ctx, actor, actorTemplate)
+	prefix, err := actorSnapshotStoragePrefix(actor)
 	if err != nil {
 		return err
 	}
@@ -396,38 +396,37 @@ func (w *ActorWorkflow) ensureExternalSnapshotsReleased(ctx context.Context, act
 // the snapshot it last took, the one a suspend was in the middle of taking, and
 // anything a crashed suspend stranded. A zero prefix means the actor never
 // wrote anything.
-func actorSnapshotStoragePrefix(ctx context.Context, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate) (resources.StoragePrefix, error) {
-	owner := actorSnapshotOwner(actor)
+func actorSnapshotStoragePrefix(actor *ateapipb.Actor) (resources.StoragePrefix, error) {
+	actorOwner := actorSnapshotOwner(actor)
 	if snapshotURI := actor.GetStatus().GetExternalSnapshot().GetSnapshotUri(); snapshotURI != "" {
 		uri, err := resources.ParseSnapshotURI(snapshotURI)
 		if err != nil {
 			return resources.StoragePrefix{}, fmt.Errorf("while parsing the external snapshot %q: %w", snapshotURI, err)
 		}
-		// The recorded URI names the actor's prefix, so no template lookup is
-		// needed. A URI the actor does not own is a tag's, borrowed until the
-		// actor's first suspend completes, which means it has written nothing of
-		// its own yet.
-		if uri.OwnedBy(owner) {
+		// A URI the actor does not own is a tag's snapshot, borrowed until the actor's
+		// first suspend completes, which means it has written nothing of its
+		// own yet.
+		if uri.OwnedBy(actorOwner) {
 			return uri.OwnerPrefix(), nil
 		}
 	}
 	// Nothing of the actor's own is recorded. Unless a suspend died partway,
-	// nothing was ever written under its prefix: the in-progress name is
+	// nothing was ever written under its prefix: the in-progress URI is
 	// recorded before atelet uploads the first object.
-	name := actor.GetStatus().GetInProgressSnapshotName()
-	if name == "" {
+	inProgress := actor.GetStatus().GetInProgressSnapshotUri()
+	if inProgress == "" {
 		return resources.StoragePrefix{}, nil
 	}
-	// The template's storage location is the only place the actor's prefix can
-	// be derived from now. Without it the actor would be stuck DELETING forever.
-	// TODO: prevent this from leaking objects in the external storage.
-	if actorTemplate == nil {
-		slog.WarnContext(ctx, "Leaking an in-progress external snapshot, the actor's template no longer resolves",
-			slog.String("actor", actor.GetMetadata().GetName()),
-			slog.String("in_progress_snapshot_name", name))
-		return resources.StoragePrefix{}, nil
+	uri, err := resources.ParseSnapshotURI(inProgress)
+	if err != nil {
+		return resources.StoragePrefix{}, fmt.Errorf("while parsing the in-progress snapshot %q: %w", inProgress, err)
+	}
+	if !uri.OwnedBy(actorOwner) {
+		// Corrupted record. This should never happen. An in-progress snapshot should
+		// always be owned by the actor that holds it.
+		return resources.StoragePrefix{}, fmt.Errorf("the in-progress snapshot %q is not owned by actor %s", inProgress, actorOwner)
 	}
-	return owner.Prefix(actorTemplate.GetSnapshotsConfig().GetStorageLocation())
+	return uri.OwnerPrefix(), nil
 }
 
 // finalizeDeleted removes the actor from the store and returns the deleted

cmd/ateapi/internal/controlapi/workflow_delete_test.go

@@ -269,10 +269,10 @@ func TestEnsureExternalSnapshotsReleased(t *testing.T) {
 			}
 			actor = mustUpdateActorStatus(t, ctx, persistence, actor, func(s *ateapipb.ActorStatus) {
 				s.ExternalSnapshot = &ateapipb.ExternalSnapshot{SnapshotUri: current.String()}
-				s.InProgressSnapshotName = tt.inFlight
+				s.InProgressSnapshotUri = inFlight.String()
 			})
 
-			if err := w.ensureExternalSnapshotsReleased(ctx, actor, template); err != nil {
+			if err := w.ensureExternalSnapshotsReleased(ctx, actor); err != nil {
 				t.Fatalf("ensureExternalSnapshotsReleased: %v", err)
 			}
 			if released := len(objects.Snapshot(t, current)) == 0; released != tt.wantCurrentReleased {
@@ -317,7 +317,7 @@ func TestEnsureExternalSnapshotsReleased_CollectsStrandedSnapshots(t *testing.T)
 		s.ExternalSnapshot = &ateapipb.ExternalSnapshot{SnapshotUri: currentSnapshot.String()}
 	})
 
-	if err := w.ensureExternalSnapshotsReleased(ctx, actor, template); err != nil {
+	if err := w.ensureExternalSnapshotsReleased(ctx, actor); err != nil {
 		t.Fatalf("ensureExternalSnapshotsReleased: %v", err)
 	}
 	if remaining := objects.Snapshot(t, strandedSnapshot); len(remaining) != 0 {
@@ -329,6 +329,39 @@ func TestEnsureExternalSnapshotsReleased_CollectsStrandedSnapshots(t *testing.T)
 	}
 }
 
+// TestDeleteActor_CollectsSnapshotWithoutActorTemplate checks that snapshots
+// are collected even if the actor template is gone.
+func TestDeleteActor_CollectsInFlightSnapshotWithoutTemplate(t *testing.T) {
+	ctx := context.Background()
+	persistence := newTestPersistence(t)
+	objects := objectstoretest.New()
+	w := NewActorWorkflow(persistence, nil, nil, nil, nil, nil, "", nil, objects)
+
+	actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"}
+	actor := storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{
+		Metadata:      &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name},
+		ActorTemplate: &ateapipb.ObjectRef{Atespace: "team-a", Name: "gone-tmpl"},
+		Status:        &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_CRASHED},
+	})
+
+	// The template that holds the storage location is gone (was never written to storage).
+	// We should still be able to access/delete the current snapshot for this actor.
+	inFlight := mustActorSnapshotURI(t, &ateapipb.ActorTemplate{
+		SnapshotsConfig: &ateapipb.SnapshotsConfig{StorageLocation: testStorageLocation},
+	}, actor, "abandoned")
+	objects.PutSnapshot(t, inFlight, "manifest.json")
+	mustUpdateActorStatus(t, ctx, persistence, actor, func(s *ateapipb.ActorStatus) {
+		s.InProgressSnapshotUri = inFlight.String()
+	})
+
+	if _, err := w.DeleteActor(ctx, actorRef, true); err != nil {
+		t.Fatalf("DeleteActor: %v", err)
+	}
+	if left := objects.Prefix(t, inFlight.OwnerPrefix()); len(left) != 0 {
+		t.Errorf("Deleting the actor left %v under its own prefix, want the in-flight snapshot collected", left)
+	}
+}
+
 // TestDeleteActor_CollectsSnapshotsAfterWorkerDelete verifies that
 // deleting an actor whose suspend a worker delete crashed mid-finalize reclaims
 // every object that suspend wrote. CRASHED is terminal, so the actor delete is
@@ -400,12 +433,12 @@ func TestDeleteActor_CollectsSnapshotsAfterWorkerDelete(t *testing.T) {
 
 			actorWorkflow := NewActorWorkflow(persistence, nil, nil, nil, nil, nil, "", nil, objects)
 			// Suspend the actor as far as it gets: MarkSuspending mints the
-			// in-progress name, and the checkpoint writes under it
+			// in-progress URI, and the checkpoint writes under it
 			actor, err := actorWorkflow.ensureMarkedSuspending(ctx, actorRef, actor, template)
 			if err != nil {
 				t.Fatalf("ensureMarkedSuspending: %v", err)
 			}
-			fresh := mustActorSnapshotURI(t, template, actor, actor.GetStatus().GetInProgressSnapshotName())
+			fresh := mustParseSnapshotURI(t, actor.GetStatus().GetInProgressSnapshotUri())
 			objects.PutSnapshot(t, fresh, "manifest.json")
 
 			// The worker's pod goes away with the commit still outstanding, so

cmd/ateapi/internal/controlapi/workflow_suspend.go

@@ -143,15 +143,15 @@ func (w *ActorWorkflow) ensureMarkedSuspending(ctx context.Context, actorRef res
 		return nil, status.Errorf(codes.FailedPrecondition, "actor %s paused with a Data snapshot; the template commits Full, which a paused-origin suspend cannot produce", actorRef)
 	}
 
-	name := resources.NewSnapshotName()
 	// Fail here rather than at checkpoint time if the template's location
 	// cannot produce a usable URI: nothing has been written yet.
-	if _, err := inProgressSnapshotURI(actorTemplate, actor, name); err != nil {
+	uri, err := newInProgressSnapshotURI(actorTemplate, actor)
+	if err != nil {
 		return nil, err
 	}
 	storedActor, err := w.store.UpdateActor(ctx, actorRef, store.PreconditionFrom(actor), func(toUpdate *ateapipb.Actor) error {
 		toUpdate.Status.State = ateapipb.ActorState_ACTOR_STATE_SUSPENDING
-		toUpdate.Status.InProgressSnapshotName = name
+		toUpdate.Status.InProgressSnapshotUri = uri.String()
 		return nil
 	})
 	if err != nil {
@@ -221,7 +221,7 @@ func (w *ActorWorkflow) ensureAteletSuspended(ctx context.Context, actorRef reso
 	ateletConn, err := w.dialer.DialForWorker(assignment.GetWorkerNamespace(), assignment.GetWorkerPod())
 	if err != nil {
 		if errors.Is(err, ErrWorkerPodNotFound) {
-			slog.ErrorContext(ctx, "Worker pod gone before checkpoint, crashing actor", "namespace", assignment.GetWorkerNamespace(), "pod", assignment.GetWorkerPod(), "in_progress_snapshot_name", actor.GetStatus().GetInProgressSnapshotName())
+			slog.ErrorContext(ctx, "Worker pod gone before checkpoint, crashing actor", "namespace", assignment.GetWorkerNamespace(), "pod", assignment.GetWorkerPod(), "in_progress_snapshot_uri", actor.GetStatus().GetInProgressSnapshotUri())
 			if err := crashActor(ctx, w.store, actorRef, ateattr.OperationSuspend, ateattr.ReasonWorkerPodGone); err != nil {
 				slog.ErrorContext(ctx, "Failed to crash actor", slog.String("err", err.Error()))
 			}
@@ -236,11 +236,6 @@ func (w *ActorWorkflow) ensureAteletSuspended(ctx context.Context, actorRef reso
 		return "", err
 	}
 
-	snapshotURI, err := inProgressSnapshotURI(actorTemplate, actor, actor.GetStatus().GetInProgressSnapshotName())
-	if err != nil {
-		return "", err
-	}
-
 	// Checkpoint does not carry the sandbox config: atelet uses the version the
 	// actor is currently running (recorded on-node at Run/Restore) and pins it
 	// into the snapshot manifest.
@@ -254,7 +249,7 @@ func (w *ActorWorkflow) ensureAteletSuspended(ctx context.Context, actorRef reso
 		Type:                  ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL,
 		Config: &ateletpb.CheckpointRequest_ExternalConfig{
 			ExternalConfig: &ateletpb.ExternalCheckpointConfiguration{
-				SnapshotUri: snapshotURI.String(),
+				SnapshotUri: actor.GetStatus().GetInProgressSnapshotUri(),
 			},
 		},
 		Scope:    actorSnapshotContentScopeToAtelet(commitSnapshotScope(actor.GetMetadata().GetAtespace(), actorTemplate)),
@@ -295,19 +290,14 @@ func (w *ActorWorkflow) ensurePausedSnapshotUploaded(ctx context.Context, actorR
 	}
 	client := ateletpb.NewAteomHerderClient(ateletConn)
 
-	snapshotURI, err := inProgressSnapshotURI(actorTemplate, actor, actor.GetStatus().GetInProgressSnapshotName())
-	if err != nil {
-		return "", err
-	}
-
 	req := &ateletpb.UploadPausedCheckpointRequest{
 		Atespace:               actor.GetMetadata().GetAtespace(),
 		ActorName:              actor.GetMetadata().GetName(),
 		ActorUid:               actor.GetMetadata().GetUid(),
 		ActorTemplateAtespace:  actor.GetActorTemplate().GetAtespace(),
 		ActorTemplateName:      actor.GetActorTemplate().GetName(),
 		LocalSnapshotName:      local.GetSnapshotName(),
-		DestinationSnapshotUri: snapshotURI.String(),
+		DestinationSnapshotUri: actor.GetStatus().GetInProgressSnapshotUri(),
 		// The commit scope, like a running-origin suspend; atelet converts
 		// from the captured scope in the snapshot's manifest where possible.
 		DesiredScope: actorSnapshotContentScopeToAtelet(commitSnapshotScope(actor.GetMetadata().GetAtespace(), actorTemplate)),
@@ -318,11 +308,11 @@ func (w *ActorWorkflow) ensurePausedSnapshotUploaded(ctx context.Context, actorR
 	return wireSnapshotScope, maybeCrashActor(ctx, w.store, actorRef, err, "while uploading paused snapshot", ateattr.OperationSuspend)
 }
 
-// inProgressSnapshotURI is where the snapshot an actor is currently taking is
+// newInProgressSnapshotURI is where the snapshot an actor is currently taking is
 // written: under the actor's own prefix, so the objects name their owner.
-func inProgressSnapshotURI(actorTemplate *ateapipb.ActorTemplate, actor *ateapipb.Actor, name string) (resources.SnapshotURI, error) {
+func newInProgressSnapshotURI(actorTemplate *ateapipb.ActorTemplate, actor *ateapipb.Actor) (resources.SnapshotURI, error) {
 	atespace := actor.GetMetadata().GetAtespace()
-	uri, err := resources.NewActorSnapshotURI(actorTemplate.GetSnapshotsConfig().GetStorageLocation(), atespace, actor.GetMetadata().GetUid(), name)
+	uri, err := resources.NewActorSnapshotURI(actorTemplate.GetSnapshotsConfig().GetStorageLocation(), atespace, actor.GetMetadata().GetUid(), resources.NewSnapshotName())
 	if err != nil {
 		return resources.SnapshotURI{}, fmt.Errorf("while building the snapshot URI for actor %s/%s: %w", atespace, actor.GetMetadata().GetName(), err)
 	}
@@ -396,17 +386,11 @@ func (w *ActorWorkflow) ensureSuspendedFinalized(ctx context.Context, actorRef r
 	// 2. Finalize the actor: record its new external snapshot and mark it SUSPENDED. This
 	// must run even with no worker assignment (nothing to free), or the actor
 	// would be left SUSPENDING forever with the workflow reporting success.
-	snapshotName := latestActor.GetStatus().GetInProgressSnapshotName()
+	inProgressSnapshotURI := latestActor.GetStatus().GetInProgressSnapshotUri()
 	externalSnapshot := latestActor.GetStatus().GetExternalSnapshot()
-	if snapshotName != "" {
-		// The same inputs CallAteletSuspend used, so the recorded URI is
-		// where the external snapshot was actually written.
-		uri, err := inProgressSnapshotURI(actorTemplate, latestActor, snapshotName)
-		if err != nil {
-			return nil, err
-		}
+	if inProgressSnapshotURI != "" {
 		externalSnapshot = &ateapipb.ExternalSnapshot{
-			SnapshotUri:  uri.String(),
+			SnapshotUri:  inProgressSnapshotURI,
 			ContentScope: commitSnapshotScope(actorRef.Atespace, actorTemplate),
 		}
 	}
@@ -424,12 +408,12 @@ func (w *ActorWorkflow) ensureSuspendedFinalized(ctx context.Context, actorRef r
 	t = time.Now()
 	storedActor, err := w.store.UpdateActor(ctx, actorRef, store.PreconditionFrom(latestActor), func(toUpdate *ateapipb.Actor) error {
 		toUpdate.Status.State = ateapipb.ActorState_ACTOR_STATE_SUSPENDED
-		if snapshotName != "" {
+		if inProgressSnapshotURI != "" {
 			// The recorded URI is under the actor's own prefix, so the actor now
 			// owns its external snapshot rather than borrowing the tag's it may
 			// have been created from.
 			toUpdate.Status.ExternalSnapshot = proto.CloneOf(externalSnapshot)
-			toUpdate.Status.InProgressSnapshotName = ""
+			toUpdate.Status.InProgressSnapshotUri = ""
 		}
 		toUpdate.Status.WorkerAssignment = nil
 		toUpdate.Status.LocalSnapshotInfo = nil