Reverting an actor keeps a snapshot atelet uploaded but had not yet reported

cmd/ateapi/internal/controlapi/snapshot_report.go

@@ -37,9 +37,9 @@ func (s *RPCService) ApplyActorSnapshotState(ctx context.Context, actorRef resou
 }
 
 // ApplyReportedSnapshotState applies a report from the atelet on node to the
-// Actor. It holds the Actor's lease and applies reports only to a SUSPENDED or
-// RUNNING Actor, so a report never interleaves with a workflow. A report that
-// changes nothing writes nothing.
+// Actor. It holds the Actor's lease and applies reports only to a SUSPENDED,
+// RUNNING, or CRASHED Actor, so a report never interleaves with a workflow. A
+// report that changes nothing writes nothing.
 func (w *ActorWorkflow) ApplyReportedSnapshotState(ctx context.Context, actorRef resources.ActorRef, actorUID, node string, state *ateapipb.ActorSnapshotState) error {
 	ctx = preview.ContextWithOptIn(ctx)
 	// Check before taking the lease, so an atelet can't hold the lease of an
@@ -57,23 +57,26 @@ func (w *ActorWorkflow) ApplyReportedSnapshotState(ctx context.Context, actorRef
 		return err
 	}
 
+	applied := proto.Clone(before).(*ateapipb.Actor)
 	// A workflow owns a transitioning Actor's snapshots, even after its lease
-	// lapses. A crashed Actor's snapshots wait for a revert.
+	// lapses.
 	switch st := before.GetStatus().GetState(); st {
 	case ateapipb.ActorState_ACTOR_STATE_SUSPENDED, ateapipb.ActorState_ACTOR_STATE_RUNNING:
+		if err := applyActorSnapshotState(leaseCtx, applied.GetStatus(), node, state); err != nil {
+			return err
+		}
 	case ateapipb.ActorState_ACTOR_STATE_CRASHED:
-		return apierror.FailedPrecondition("actor %s is %s; revert it first", actorRef, st)
+		// An Actor whose later generation crashed may have another async
+		// upload that uploaded its snapshot.
+		if err := applyUploadOnNode(leaseCtx, actorRef, applied.GetStatus(), node, state.GetLatestDurable()); err != nil {
+			return err
+		}
 	case ateapipb.ActorState_ACTOR_STATE_SUSPENDING, ateapipb.ActorState_ACTOR_STATE_RESUMING,
 		ateapipb.ActorState_ACTOR_STATE_DELETING, ateapipb.ActorState_ACTOR_STATE_REVERTING:
 		return apierror.Aborted("actor %s is %s; retry once that finishes", actorRef, st)
 	default:
 		return apierror.Internal("actor %s has unexpected state %s", actorRef, st)
 	}
-
-	applied := proto.Clone(before).(*ateapipb.Actor)
-	if err := applyActorSnapshotState(leaseCtx, applied.GetStatus(), node, state); err != nil {
-		return err
-	}
 	if proto.Equal(applied.GetStatus(), before.GetStatus()) {
 		return nil
 	}
@@ -162,6 +165,28 @@ func applyAteletResponseState(ctx context.Context, status *ateapipb.ActorStatus,
 	return applyLatestDurable(ctx, status, snapshotReportFromAtelet(state.GetLatestDurable()))
 }
 
+// applyUploadOnNode marks the reported durable snapshot COMPLETED, as
+// applyLatestDurable does, if the record places that snapshot on node. Only a
+// finalized suspend places a snapshot on a node, so any other snapshot, such
+// as one a crashed suspend created, stays PENDING and a revert deletes it with
+// the rest of the discarded execution.
+func applyUploadOnNode(ctx context.Context, actorRef resources.ActorRef, status *ateapipb.ActorStatus, node string, reported *ateapipb.SnapshotReport) error {
+	if reported == nil {
+		return nil
+	}
+	snap := findSnapshotByGeneration(status.GetSnapshots(), reported.GetGeneration())
+	if got := snap.GetLocality(); got != node {
+		slog.InfoContext(ctx, "Discarding an upload reported for a snapshot not on its node",
+			slog.Any("actor", actorRef),
+			slog.String("node", node),
+			slog.String("snapshot_locality", got),
+			slog.Int("generation", int(reported.GetGeneration())),
+			slog.String("uuid", reported.GetUuid()))
+		return nil
+	}
+	return applyLatestDurable(ctx, status, reported)
+}
+
 // applyLatestDurable marks the reported durable snapshot COMPLETED, and records
 // its fidelity. Its uuid and upload URI must match the recorded ones, and its
 // fidelity must be set. The snapshot keeps its locality, so an Actor with a

cmd/ateapi/internal/controlapi/snapshot_report_test.go

@@ -468,7 +468,7 @@
 			assertUnwritten(t)
 		})
 	}
-	// A transitioning actor gets Aborted, and a crashed one FailedPrecondition.
+	// A transitioning actor gets Aborted.
 	for _, tc := range []struct {
 		state ateapipb.ActorState
 		want  codes.Code
@@ -477,7 +477,6 @@
 		{ateapipb.ActorState_ACTOR_STATE_RESUMING, codes.Aborted},
 		{ateapipb.ActorState_ACTOR_STATE_DELETING, codes.Aborted},
 		{ateapipb.ActorState_ACTOR_STATE_REVERTING, codes.Aborted},
-		{ateapipb.ActorState_ACTOR_STATE_CRASHED, codes.FailedPrecondition},
 	} {
 		t.Run("refused for a "+tc.state.String()+" actor", func(t *testing.T) {
 			actor = mustUpdateActorStatus(t, ctx, persistence, actor, func(s *ateapipb.ActorStatus) { s.State = tc.state })
@@ -519,6 +518,73 @@
 	assertUnwritten(t)
 }
 
+// A report for a CRASHED actor records only the upload, and only of a snapshot
+// the record places on the reporting node. atelet forgets the actor when a
+// revert or delete terminates it, so the report is the upload's last chance
+// to be recorded before the revert deletes it as in progress.
+func TestApplyReportedSnapshotState_Crashed(t *testing.T) {
+	completed := ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_COMPLETED
+	pending := ateapipb.SnapshotStorageStatus_SNAPSHOT_STORAGE_STATUS_PENDING
+	for _, tc := range []struct {
+		name string
+		// locality is the snapshot's node. A suspend that crashed before it
+		// committed leaves its snapshot on no node.
+		locality string
+		// detach sends the report atelet sends once it detached the actor:
+		// no local snapshot and no upload in progress.
+		detach      bool
+		wantDurable ateapipb.SnapshotStorageStatus
+	}{
+		{name: "records the upload", locality: reportNode, wantDurable: completed},
+		{name: "does not apply a detach", locality: reportNode, detach: true, wantDurable: completed},
+		{name: "leaves a crashed suspend's upload pending", wantDurable: pending},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.Background()
+			persistence := newTestPersistence(t)
+			w := NewActorWorkflow(persistence, nil, nil, nil, nil, "", nil, nil)
+			actorRef := resources.ActorRef{Atespace: reportAtespace, Name: "actor-1"}
+			actor := storetest.MustCreateActor(t, ctx, persistence, &ateapipb.Actor{
+				Metadata:      &ateapipb.ResourceMetadata{Atespace: actorRef.Atespace, Name: actorRef.Name},
+				ActorTemplate: &ateapipb.ObjectRef{Atespace: reportAtespace, Name: "tmpl"},
+				Status:        &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_CRASHED},
+			})
+			uid := actor.GetMetadata().GetUid()
+			uri := ownedURI(t, uid, 2)
+			actor = mustUpdateActorStatus(t, ctx, persistence, actor, func(s *ateapipb.ActorStatus) {
+				s.AssignedNode = reportNode
+				s.LastAssignedGeneration = 2
+				snap := newDurableSnapshot(2, ateapipb.SnapshotOwner_SNAPSHOT_OWNER_ACTOR, ateapipb.SnapshotFidelity_SNAPSHOT_FIDELITY_MEMORY, "", snapName(2), uri, pending)
+				snap.Locality = tc.locality
+				s.Snapshots = []*ateapipb.Snapshot{snap}
+			})
+			uploaded := reportedAt(t, 2, uri)
+			state := &ateapipb.ActorSnapshotState{LatestLocal: uploaded, LatestDurable: uploaded}
+			if tc.detach {
+				state.LatestLocal = nil
+			}
+
+			if err := w.ApplyReportedSnapshotState(ctx, actorRef, uid, reportNode, state); err != nil {
+				t.Fatalf("ApplyReportedSnapshotState() failed: %v", err)
+			}
+			got, err := persistence.GetActor(ctx, actorRef)
+			if err != nil {
+				t.Fatal(err)
+			}
+			snap := got.GetStatus().GetSnapshots()[0]
+			if st := snap.GetDurableSnapshot().GetStatus(); st != tc.wantDurable {
+				t.Errorf("durable snapshot status = %v, want %v", st, tc.wantDurable)
+			}
+			if snap.GetLocality() != tc.locality || got.GetStatus().GetAssignedNode() != reportNode {
+				t.Errorf("snapshot locality %q, assigned node %q, want %q and %q", snap.GetLocality(), got.GetStatus().GetAssignedNode(), tc.locality, reportNode)
+			}
+			if got.GetStatus().GetState() != ateapipb.ActorState_ACTOR_STATE_CRASHED {
+				t.Errorf("state = %v, want CRASHED", got.GetStatus().GetState())
+			}
+		})
+	}
+}
+
 func TestActorOnNode(t *testing.T) {
 	actor := func(assigned string, localities ...string) *ateapipb.Actor {
 		status := &ateapipb.ActorStatus{AssignedNode: assigned}

cmd/ateapi/internal/controlapi/workflow_delete.go

@@ -61,7 +61,7 @@ func (w *ActorWorkflow) DeleteActor(ctx context.Context, actorRef resources.Acto
 	}
 
 	var atletTerminatedErr, volumesDetachedErr error
-	if err := w.ensureAteletTerminated(ctx, actorRef, actor, actorTemplate, ateattr.OperationDelete); err != nil {
+	if _, err := w.ensureAteletTerminated(ctx, actorRef, actor, actorTemplate, ateattr.OperationDelete); err != nil {
 		atletTerminatedErr = fmt.Errorf("while terminating atelet: %w", err)
 		errs = append(errs, atletTerminatedErr)
 	}
@@ -96,15 +96,17 @@ func (w *ActorWorkflow) DeleteActor(ctx context.Context, actorRef resources.Acto
 	return w.finalizeDeleted(ctx, actor)
 }
 
-// ensureAteletTerminated calls atelet to terminate the workload for opName.
-func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate, opName string) (err error) {
+// ensureAteletTerminated calls atelet to terminate the workload for opName. It
+// returns the actor's state as atelet forgot it, or nil when it skipped the
+// call.
+func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef resources.ActorRef, actor *ateapipb.Actor, actorTemplate *ateapipb.ActorTemplate, opName string) (_ *ateletpb.ActorSnapshotState, err error) {
 	ctx, done := stepSpan(ctx, "CallAteletTerminate")
 	defer func() { err = done(err) }()
 
 	nodeName := actor.GetStatus().GetAssignedNode()
 	if nodeName == "" {
 		slog.InfoContext(ctx, "actor has no assigned node, skipping atelet terminate request", slog.Any("actor", actorRef))
-		return nil
+		return nil, nil
 	}
 
 	workerPodUID := ""
@@ -116,7 +118,7 @@ func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef res
 			// ordinary case, and the others are none of this delete's business.
 			hosted, err := workerHostsActor(ctx, w.store, workerName, actor.GetMetadata().GetUid())
 			if err != nil {
-				return err
+				return nil, err
 			}
 			if !hosted {
 				slog.InfoContext(ctx, "worker is no longer assigned to this actor, skipping worker workload termination",
@@ -131,18 +133,18 @@ func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef res
 	if errors.Is(err, ErrNoAteletOnNode) {
 		gone, nodeErr := w.dialer.NodeGone(ctx, nodeName)
 		if nodeErr != nil {
-			return fmt.Errorf("while connecting to atelet on node %q: %w (and while checking the node: %w)", nodeName, err, nodeErr)
+			return nil, fmt.Errorf("while connecting to atelet on node %q: %w (and while checking the node: %w)", nodeName, err, nodeErr)
 		}
 		if gone {
 			// A deleted Node object counts as the node being gone for good. If the
 			// machine still runs, whatever atelet holds for the actor stays there.
 			slog.WarnContext(ctx, "assigned node no longer exists, skipping atelet terminate request",
 				slog.String("node", nodeName), slog.Any("actor", actorRef))
-			return nil
+			return nil, nil
 		}
 	}
 	if err != nil {
-		return fmt.Errorf("while connecting to atelet on node %q: %w", nodeName, err)
+		return nil, fmt.Errorf("while connecting to atelet on node %q: %w", nodeName, err)
 	}
 
 	client := ateletpb.NewAteletClient(conn)
@@ -151,7 +153,7 @@ func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef res
 	if actorTemplate != nil {
 		spec, err := workloadSpecFromActorTemplate(actorTemplate, actor, nil)
 		if err != nil {
-			return err
+			return nil, err
 		}
 		workloadSpec = spec
 	} else {
@@ -174,15 +176,15 @@ func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef res
 		Spec:                  workloadSpec,
 	}
 
-	if _, err := client.Terminate(ctx, req); err != nil {
+	resp, err := client.Terminate(ctx, req)
+	if err != nil {
 		if status.Code(err) == codes.NotFound {
 			slog.InfoContext(ctx, "workload already terminated on atelet", slog.Any("actor", actorRef))
-			return nil
+			return nil, nil
 		}
-		return handleAteletError(ctx, w.store, actorRef, opName, "Terminate", true, err)
+		return nil, handleAteletError(ctx, w.store, actorRef, opName, "Terminate", true, err)
 	}
-
-	return nil
+	return resp.GetState(), nil
 }
 
 // fallbackWorkloadSpec is the spec to terminate an actor with when its template

cmd/ateapi/internal/controlapi/workflow_delete_test.go

@@ -124,7 +124,7 @@ func TestEnsureAteletTerminated_NodeGone(t *testing.T) {
 			w := newTestActorWorkflow(t, newTestPersistence(t), "team-a", "tmpl")
 			w.dialer = NewAteletDialer(newTestAteletIndexer(t), installdefaults.AteletSPIFFEID(installdefaults.SystemNamespace), "", "",
 				WithNodeGetter(nodeGetter(tc.lookupErr, tc.nodes...)))
-			err := w.ensureAteletTerminated(context.Background(), actorRef, actor, nil, "revert")
+			_, err := w.ensureAteletTerminated(context.Background(), actorRef, actor, nil, "revert")
 			if tc.wantErr != (err != nil) {
 				t.Fatalf("ensureAteletTerminated = %v, want error %v", err, tc.wantErr)
 			}

cmd/ateapi/internal/controlapi/workflow_reconcile_assignments.go

@@ -194,7 +194,7 @@ func (w *WorkerWorkflow) releaseEarlierAssignment(ctx context.Context, worker *a
 		return nil
 	}
 	if actor.GetStatus().GetState() != ateapipb.ActorState_ACTOR_STATE_SUSPENDED {
-		if err := w.crashBoundActor(ctx, worker, actorRef, actor, "Releasing actor from a restarted worker", crashMessageWorkerRestarted); err != nil {
+		if err := w.crashBoundActor(ctx, worker, actorRef, actor, nil, "Releasing actor from a restarted worker", crashMessageWorkerRestarted); err != nil {
 			return err
 		}
 	}

cmd/ateapi/internal/controlapi/workflow_resume_detach.go

@@ -139,9 +139,6 @@ func (w *ActorWorkflow) recordDetach(ctx context.Context, actorRef resources.Act
 	if err := applyDetach(applied, node, applied.GetLastAssignedGeneration()); err != nil {
 		return nil, err
 	}
-	if applied.GetState() == ateapipb.ActorState_ACTOR_STATE_REVERTING && applied.GetWorkerAssignment() == nil {
-		applied.AssignedNode = ""
-	}
 	stored, err := w.store.UpdateActor(ctx, actorRef, store.PreconditionFrom(actor), func(toUpdate *ateapipb.Actor) error {
 		// The precondition makes toUpdate the version applied was built from.
 		toUpdate.Status.Snapshots = applied.GetSnapshots()

cmd/ateapi/internal/controlapi/workflow_resume_test.go

@@ -1150,7 +1150,9 @@ type capturingAtelet struct {
 	// onDetach, when set, runs before DetachActor returns.
 	onDetach     func()
 	terminateErr error
-	onTerminate  func()
+	// terminateState, when set, is the snapshot state Terminate returns.
+	// Terminate is unimplemented otherwise.
+	terminateState *ateletpb.ActorSnapshotState
 }
 
 func (f *capturingAtelet) Restore(ctx context.Context, req *ateletpb.RestoreRequest) (*ateletpb.RestoreResponse, error) {
@@ -1212,17 +1214,14 @@ func (f *capturingAtelet) DetachActor(ctx context.Context, req *ateletpb.DetachA
 
 func (f *capturingAtelet) Terminate(ctx context.Context, req *ateletpb.TerminateRequest) (*ateletpb.TerminateResponse, error) {
 	f.mu.Lock()
-	hook := f.onTerminate
-	f.mu.Unlock()
-	if hook != nil {
-		hook()
-	}
-	f.mu.Lock()
-	err := f.terminateErr
+	err, state := f.terminateErr, proto.CloneOf(f.terminateState)
 	f.mu.Unlock()
 	if err != nil {
 		return nil, err
 	}
+	if state != nil {
+		return &ateletpb.TerminateResponse{State: state}, nil
+	}
 	return f.UnimplementedAteletServer.Terminate(ctx, req)
 }