ResumeActor can require an expected snapshot

api/ateapi/v1/ateapi.proto

@@ -1845,6 +1845,24 @@ message ResumeActorRequest {
   // +k8s:required
   // +k8s:subfield(atespace)=+k8s:required
   ObjectRef actor = 1;
+
+  // Resume only if this is still the actor and checkpoint the caller observed.
+  // This guards the current restore source; it does not select an older one.
+  // Restoring requires SUSPENDED or retryable RESUMING and a matching completed
+  // durable checkpoint, even when restoring a local copy. If its node was
+  // deleted, resume attempts that checkpoint's upload even if completion was
+  // not reported. A missing upload moves the actor to CRASHED and returns
+  // FAILED_PRECONDITION; call RevertActor before resuming again.
+  // A RUNNING actor with a matching UID and newest held checkpoint returns
+  // resumed=false.
+  // A UID or checkpoint mismatch returns FAILED_PRECONDITION: the caller's
+  // view is stale, so retrying the same request won't help. An invalid state
+  // also returns FAILED_PRECONDITION.
+  // A concurrent template change returns ABORTED and can be retried; a matching
+  // pending upload on a node not known to be deleted returns UNAVAILABLE.
+  // Omit for unconditional resume.
+  // +k8s:optional
+  ExpectedSnapshot expected = 2;
 }
 
 message ResumeActorResponse {
@@ -3266,3 +3284,21 @@ extend google.protobuf.MessageOptions {
 extend google.protobuf.MethodOptions {
   MethodAnnotation method = 50000;
 }
+
+// ExpectedSnapshot identifies the actor incarnation and checkpoint to resume.
+message ExpectedSnapshot {
+  // UID of the actor the caller observed.
+  // +k8s:required
+  // +k8s:format=k8s-uuid
+  string actor_uid = 1;
+
+  // UUID of the checkpoint the caller observed.
+  // +k8s:required
+  // +k8s:format=k8s-short-name
+  string snapshot_uuid = 2;
+
+  // Generation of the checkpoint the caller observed.
+  // +k8s:required
+  // +k8s:minimum=1
+  int32 generation = 3;
+}

cmd/ateapi/internal/apivalidation/actor_test.go

@@ -924,11 +924,41 @@ func TestValidateDeleteActorRequest(t *testing.T) {
 }
 
 func TestValidateResumeActorRequest(t *testing.T) {
+	withExpected := func(expected *ateapi.ExpectedSnapshot) *ateapi.ResumeActorRequest {
+		return &ateapi.ResumeActorRequest{
+			Actor:    &ateapi.ObjectRef{Atespace: "ns1", Name: "id1"},
+			Expected: expected,
+		}
+	}
 	tests := []struct {
 		name string
 		req  *ateapi.ResumeActorRequest
 		want field.ErrorList
 	}{{
+		"valid expected",
+		withExpected(&ateapi.ExpectedSnapshot{ActorUid: "c4652e88-7079-448b-94ce-55b45fa12c43", SnapshotUuid: "snap-1", Generation: 1}),
+		nil,
+	}, {
+		"empty expected",
+		withExpected(&ateapi.ExpectedSnapshot{}),
+		field.ErrorList{
+			field.Required(field.NewPath("expected", "actor_uid"), ""),
+			field.Required(field.NewPath("expected", "snapshot_uuid"), ""),
+			field.Required(field.NewPath("expected", "generation"), ""),
+		},
+	}, {
+		"expected generation zero",
+		withExpected(&ateapi.ExpectedSnapshot{ActorUid: "c4652e88-7079-448b-94ce-55b45fa12c43", SnapshotUuid: "snap-1", Generation: 0}),
+		field.ErrorList{field.Required(field.NewPath("expected", "generation"), "")},
+	}, {
+		"negative expected generation",
+		withExpected(&ateapi.ExpectedSnapshot{ActorUid: "c4652e88-7079-448b-94ce-55b45fa12c43", SnapshotUuid: "snap-1", Generation: -1}),
+		field.ErrorList{field.Invalid(field.NewPath("expected", "generation"), nil, "").WithOrigin("minimum")},
+	}, {
+		"malformed expected snapshot_uuid",
+		withExpected(&ateapi.ExpectedSnapshot{ActorUid: "c4652e88-7079-448b-94ce-55b45fa12c43", SnapshotUuid: "Bad/Snapshot", Generation: 1}),
+		field.ErrorList{field.Invalid(field.NewPath("expected", "snapshot_uuid"), "Bad/Snapshot", "").WithOrigin("format=k8s-short-name")},
+	}, {
 		"valid",
 		&ateapi.ResumeActorRequest{Actor: &ateapi.ObjectRef{Atespace: "ns1", Name: "id1"}},
 		nil,

cmd/ateapi/internal/controlapi/actor.go

@@ -488,7 +488,7 @@ func (s *RPCService) ResumeActor(ctx context.Context, req *ateapi.ResumeActorReq
 	actorRef := resources.ActorRefFromObjectRef(req.GetActor())
 	setSpanActorRefAttributes(ctx, actorRef)
 
-	actor, resumed, err := s.actorWorkflow.ResumeActor(ctx, actorRef)
+	actor, resumed, err := s.actorWorkflow.ResumeActor(ctx, actorRef, req.GetExpected())
 	if err != nil {
 		if errors.Is(err, store.ErrVersionConflict) {
 			return nil, apierror.Aborted("concurrent update conflict, please retry")

cmd/ateapi/internal/controlapi/actor_test.go

@@ -1053,7 +1053,7 @@ func TestCreateActor_GoldenTagDefault(t *testing.T) {
 			}
 			workflow := &ActorWorkflow{store: persistence}
 			actorRef := resources.ActorRefFromActor(created)
-			loaded, _, err := workflow.loadActorForResume(ctx, actorRef)
+			loaded, _, err := workflow.loadActorForResume(ctx, actorRef, nil)
 			if err != nil {
 				t.Fatal(err)
 			}

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

@@ -3258,17 +3258,24 @@ func TestResumeActor(t *testing.T) {
 	podUID := createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1")
 
 	name := "id1"
-	_, err := tc.client.CreateActor(context.Background(), &ateapi.CreateActorRequest{Actor: &ateapi.Actor{
+	created, err := tc.client.CreateActor(context.Background(), &ateapi.CreateActorRequest{Actor: &ateapi.Actor{
 		Metadata:      &ateapi.ResourceMetadata{Atespace: testAtespace, Name: name},
 		ActorTemplate: &ateapi.ObjectRef{Atespace: testAtespace, Name: "tmpl1"},
 	}})
 	if err != nil {
 		t.Fatalf("CreateActor failed: %v", err)
 	}
 
-	_, err = tc.client.ResumeActor(context.Background(), &ateapi.ResumeActorRequest{
-		Actor: &ateapi.ObjectRef{Atespace: testAtespace, Name: name},
-	})
+	expected := &ateapi.ExpectedSnapshot{ActorUid: created.GetMetadata().GetUid(), SnapshotUuid: goldenSnapshotName, Generation: 2}
+	request := &ateapi.ResumeActorRequest{Actor: &ateapi.ObjectRef{Atespace: testAtespace, Name: name}, Expected: expected}
+	if _, err := tc.client.ResumeActor(context.Background(), request); status.Code(err) != codes.FailedPrecondition {
+		t.Fatalf("ResumeActor with wrong generation = %v, want FailedPrecondition", err)
+	}
+	if tc.fakeAtelet.RestoreCalled {
+		t.Fatal("rejected resume called Restore")
+	}
+	expected.Generation = 1
+	_, err = tc.client.ResumeActor(context.Background(), request)
 	if err != nil {
 		t.Fatalf("ResumeActor failed: %v", err)
 	}

cmd/ateapi/internal/controlapi/hosting_test.go

@@ -190,7 +190,7 @@ func TestHostingNodeAfterResumeOntoNewWorker(t *testing.T) {
 	tmpl := &ateapi.ActorTemplate{
 		WorkerSelector: &ateapi.Selector{MatchLabels: map[string]string{"pool": "b"}},
 	}
-	resuming, _, err := w.assignWorkerAttempt(ctx, actorRef, actor, tmpl)
+	resuming, _, err := w.assignWorkerAttempt(ctx, actorRef, actor, tmpl, nil)
 	if err != nil {
 		t.Fatalf("assignWorkerAttempt: %v", err)
 	}

cmd/ateapi/internal/controlapi/workflow_reconcile_assignments_test.go

@@ -453,7 +453,7 @@ func TestAssignWorkerAttempt_LeavesWorkerEpochUnset(t *testing.T) {
 
 	w := &ActorWorkflow{store: persistence, workerCache: wc, scheduler: scheduling.New(wc)}
 	tmpl := &ateapi.ActorTemplate{}
-	stored, _, err := w.assignWorkerAttempt(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, tmpl)
+	stored, _, err := w.assignWorkerAttempt(ctx, resources.ActorRef{Atespace: "team-a", Name: "id1"}, actor, tmpl, nil)
 	if err != nil {
 		t.Fatalf("assignWorkerAttempt: %v", err)
 	}