api: add actor_uid to Assignment alongside actor

cmd/ateapi/internal/actoridentity/actoridentity.go

@@ -384,9 +384,9 @@ func (s *Server) authorizeActor(ctx context.Context, caller *ateletCaller, actor
 	}
 
 	// The worker must still agree that it is hosting this actor.
-	if assigned := worker.GetAssignment().GetActor(); resources.ActorRefFromObjectRef(assigned) != actorRef {
+	if assignedActorUID := worker.GetAssignment().GetActorUid(); assignedActorUID != actor.GetMetadata().GetUid() {
 		return nil, deny("worker is no longer assigned to the actor",
-			slog.String("workerAssignment", assigned.GetAtespace()+"/"+assigned.GetName()))
+			slog.String("assignedActorUID", assignedActorUID))
 	}
 
 	return actor, nil

cmd/ateapi/internal/actoridentity/actoridentity_test.go

@@ -180,6 +180,8 @@ type actorFixture struct {
 	noPlacement bool
 	// noWorker skips seeding the worker record entirely.
 	noWorker bool
+	// mismatchedUID simulates a worker assigned to an actor with the same name/atespace but a different UID.
+	mismatchedUID bool
 }
 
 // seedActor writes an actor, and normally its hosting worker, into st.
@@ -201,7 +203,8 @@ func seedActor(t *testing.T, ctx context.Context, st store.Interface, f actorFix
 			WorkerPodUid:    "worker-uid",
 		}
 	}
-	if _, err := st.CreateActor(ctx, actor); err != nil {
+	created, err := st.CreateActor(ctx, actor)
+	if err != nil {
 		t.Fatalf("seed actor: %v", err)
 	}
 
@@ -212,14 +215,21 @@ func seedActor(t *testing.T, ctx context.Context, st store.Interface, f actorFix
 	if assigned == (resources.ActorRef{}) {
 		assigned = actorRef
 	}
+	assignedActorUID := created.GetMetadata().GetUid()
+	if f.mismatchedUID || assigned != actorRef {
+		assignedActorUID = "other-actor-uid"
+	}
 	worker := &ateapipb.Worker{
 		WorkerNamespace: testPodNS,
 		WorkerPool:      testPool,
 		WorkerPod:       testWorkerPod,
 		WorkerPodUid:    "worker-uid",
 		NodeName:        f.workerNode,
 		State:           ateapipb.Worker_STATE_ACTIVE,
-		Assignment:      &ateapipb.Assignment{Actor: assigned.ToObjectRef()},
+		Assignment: &ateapipb.Assignment{
+			Actor:    assigned.ToObjectRef(),
+			ActorUid: assignedActorUID,
+		},
 	}
 	if f.unassigned {
 		worker.Assignment = nil
@@ -319,6 +329,14 @@ func TestMintCertAuthorization(t *testing.T) {
 			},
 			wantCode: codes.PermissionDenied,
 		},
+		"worker is assigned to an actor with same name and atespace but different UID": {
+			fixture: actorFixture{
+				status:        ateapipb.Actor_STATUS_RUNNING,
+				workerNode:    testNode,
+				mismatchedUID: true,
+			},
+			wantCode: codes.PermissionDenied,
+		},
 		"hosting worker record is missing": {
 			fixture: actorFixture{
 				status:     ateapipb.Actor_STATUS_RUNNING,

cmd/ateapi/internal/controlapi/crash.go

@@ -123,7 +123,7 @@ func releaseWorker(ctx context.Context, st store.Interface, actor *ateapipb.Acto
 		return sandboxClass, nil
 	}
 	// Only free it if it still belongs to us
-	if resources.ActorRefFromObjectRef(wass.GetActor()) != resources.ActorRefFromActor(actor) {
+	if wass.GetActorUid() != actor.GetMetadata().GetUid() {
 		slog.WarnContext(ctx, "Worker already assigned to another Actor", slog.String("worker", podUid))
 		return sandboxClass, nil
 	}

cmd/ateapi/internal/controlapi/crash_test.go

@@ -62,8 +62,17 @@ func seedWorker(t *testing.T, ctx context.Context, st store.Interface, actorRef
 		WorkerPod:       "pod",
 	}
 	if actorRef != (resources.ActorRef{}) {
-		worker.Assignment = &ateapipb.Assignment{
-			Actor: actorRef.ToObjectRef(),
+		actor, err := st.GetActor(ctx, actorRef)
+		if err != nil {
+			worker.Assignment = &ateapipb.Assignment{
+				Actor:    &ateapipb.ObjectRef{Atespace: actorRef.Atespace, Name: actorRef.Name},
+				ActorUid: "synthetic-" + actorRef.Name,
+			}
+		} else {
+			worker.Assignment = &ateapipb.Assignment{
+				Actor:    &ateapipb.ObjectRef{Atespace: actor.GetMetadata().GetAtespace(), Name: actor.GetMetadata().GetName()},
+				ActorUid: actor.GetMetadata().GetUid(),
+			}
 		}
 	}
 	if err := st.CreateWorker(ctx, worker); err != nil {
@@ -161,7 +170,45 @@ func TestCrashActor(t *testing.T) {
 					t.Fatalf("GetWorker() = %v, want nil", gerr)
 				}
 				if got := worker.GetAssignment().GetActor().GetName(); got != "actor-2" {
-					t.Errorf("worker assigned actor = %q, want %q", got, "actor-2")
+					t.Errorf("worker assigned actor name = %q, want %q", got, "actor-2")
+				}
+				if got := worker.GetAssignment().GetActorUid(); got != "synthetic-actor-2" {
+					t.Errorf("worker assigned actor uid = %q, want %q", got, "synthetic-actor-2")
+				}
+			},
+		},
+		{
+			name: "keeps worker assigned to previous incarnation of same actor",
+			seed: true,
+			setup: func(t *testing.T, ctx context.Context, st store.Interface) {
+				// Create a worker assigned to the same actorRef, but with a stale UID
+				worker := &ateapipb.Worker{
+					WorkerNamespace: "ns",
+					WorkerPool:      "pool",
+					WorkerPod:       "pod",
+					Assignment: &ateapipb.Assignment{
+						Actor:    &ateapipb.ObjectRef{Atespace: actorRef.Atespace, Name: actorRef.Name},
+						ActorUid: "stale-incarnation-uid",
+					},
+				}
+				if err := st.CreateWorker(ctx, worker); err != nil {
+					t.Fatalf("CreateWorker: %v", err)
+				}
+			},
+			check: func(t *testing.T, ctx context.Context, st store.Interface, err error) {
+				if err != nil {
+					t.Fatalf("crashActor() = %v, want nil", err)
+				}
+				assertCrashed(t, ctx, st, actorRef)
+				worker, gerr := st.GetWorker(ctx, "ns", "pool", "pod")
+				if gerr != nil {
+					t.Fatalf("GetWorker() = %v, want nil", gerr)
+				}
+				if got := worker.GetAssignment().GetActor().GetName(); got != actorRef.Name {
+					t.Errorf("worker assigned actor name = %q, want %q", got, actorRef.Name)
+				}
+				if got := worker.GetAssignment().GetActorUid(); got != "stale-incarnation-uid" {
+					t.Errorf("worker assigned actor uid = %q, want %q", got, "stale-incarnation-uid")
 				}
 			},
 		},

cmd/ateapi/internal/controlapi/functional_test.go

@@ -1710,6 +1710,7 @@ func TestResumeActor(t *testing.T) {
 				Name:     name,
 				Atespace: testAtespace,
 			},
+			ActorUid: getResp.GetMetadata().GetUid(),
 		},
 		Ip:           "127.0.0.1",
 		NodeName:     "node1",

cmd/ateapi/internal/controlapi/syncer.go

@@ -345,16 +345,20 @@ func (s *WorkerPoolSyncer) releaseActorOnDeadWorker(ctx context.Context, namespa
 		}
 		return err
 	}
-	if worker.Assignment == nil {
+	if worker.Assignment == nil || worker.Assignment.GetActor() == nil {
 		return nil
 	}
-	actor, err := s.persistence.GetActor(ctx, resources.ActorRefFromObjectRef(worker.Assignment.Actor))
+	actorRef := resources.ActorRefFromObjectRef(worker.Assignment.GetActor())
+	actor, err := s.persistence.GetActor(ctx, actorRef)
 	if err != nil {
 		if errors.Is(err, store.ErrNotFound) {
 			return nil
 		}
 		return err
 	}
+	if actor.GetMetadata().GetUid() != worker.Assignment.GetActorUid() {
+		return nil
+	}
 	// Skip if a concurrent SuspendActor already cleared the pointer.
 	assignment := actor.GetWorkerAssignment()
 	if assignment.GetWorkerNamespace() != namespace || assignment.GetWorkerPod() != podName {

cmd/ateapi/internal/controlapi/syncer_test.go

@@ -256,15 +256,16 @@ func TestSyncer_DeleteBoundWorker_ClearsActor(t *testing.T) {
 		t.Fatalf("worker row not materialised: %v", err)
 	}
 	actorName := "actor-orphan"
-	if _, err := persistence.CreateActor(ctx, &ateapipb.Actor{
+	createdActor, err := persistence.CreateActor(ctx, &ateapipb.Actor{
 		Metadata: &ateapipb.ResourceMetadata{Name: actorName, Atespace: "team-orphan"}, ActorTemplateNamespace: ns, ActorTemplateName: "tmpl",
 		Status: ateapipb.Actor_STATUS_RUNNING,
 		WorkerAssignment: &ateapipb.WorkerAssignment{
 			WorkerNamespace: ns, WorkerPool: pool, WorkerPod: pod, WorkerPodIp: ip,
 		},
 		InProgressSnapshot: "gs://snapshots/partial",
 		LatestSnapshot:     &ateapipb.ObjectRef{Atespace: "team-orphan", Name: "last"},
-	}); err != nil {
+	})
+	if err != nil {
 		t.Fatalf("create actor: %v", err)
 	}
 	w, _ := persistence.GetWorker(ctx, ns, pool, pod)
@@ -273,10 +274,8 @@ func TestSyncer_DeleteBoundWorker_ClearsActor(t *testing.T) {
 			Namespace: ns,
 			Name:      "tmpl",
 		},
-		Actor: &ateapipb.ObjectRef{
-			Name:     actorName,
-			Atespace: "team-orphan",
-		},
+		Actor:    &ateapipb.ObjectRef{Atespace: createdActor.GetMetadata().GetAtespace(), Name: createdActor.GetMetadata().GetName()},
+		ActorUid: createdActor.GetMetadata().GetUid(),
 	}
 	if err := persistence.UpdateWorker(ctx, w, w.Version); err != nil {
 		t.Fatalf("update worker: %v", err)
@@ -560,13 +559,14 @@ func TestReconcileDeadWorker(t *testing.T) {
 
 	ns, pool, pod := "ns-rdw", "pool1", "worker-rdw"
 	atespace, actorID := "team-rdw", "actor-rdw"
-	if _, err := persistence.CreateActor(ctx, &ateapipb.Actor{
+	createdActor, err := persistence.CreateActor(ctx, &ateapipb.Actor{
 		Metadata: &ateapipb.ResourceMetadata{Name: actorID, Atespace: atespace}, ActorTemplateNamespace: ns, ActorTemplateName: "tmpl",
 		Status: ateapipb.Actor_STATUS_RUNNING,
 		WorkerAssignment: &ateapipb.WorkerAssignment{
 			WorkerNamespace: ns, WorkerPool: pool, WorkerPod: pod, WorkerPodIp: "10.0.0.5",
 		},
-	}); err != nil {
+	})
+	if err != nil {
 		t.Fatalf("create actor: %v", err)
 	}
 	if err := persistence.CreateWorker(ctx, &ateapipb.Worker{
@@ -575,7 +575,8 @@ func TestReconcileDeadWorker(t *testing.T) {
 		State: ateapipb.Worker_STATE_DRAINING,
 		Assignment: &ateapipb.Assignment{
 			ActorTemplate: &ateapipb.KubeNamespacedObjectRef{Namespace: ns, Name: "tmpl"},
-			Actor:         &ateapipb.ObjectRef{Name: actorID, Atespace: atespace},
+			Actor:         &ateapipb.ObjectRef{Atespace: createdActor.GetMetadata().GetAtespace(), Name: createdActor.GetMetadata().GetName()},
+			ActorUid:      createdActor.GetMetadata().GetUid(),
 		},
 	}); err != nil {
 		t.Fatalf("create worker: %v", err)
@@ -596,6 +597,57 @@ func TestReconcileDeadWorker(t *testing.T) {
 	}
 }
 
+func TestReconcileDeadWorker_IgnoresStaleIncarnationAssignment(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	persistence, cleanup := storetest.SetupTestStore(t)
+	defer cleanup()
+
+	s := &WorkerPoolSyncer{persistence: persistence}
+
+	ns, pool, pod := "ns-rdw", "pool1", "worker-rdw"
+	atespace, actorID := "team-rdw", "actor-rdw"
+	createdActor, err := persistence.CreateActor(ctx, &ateapipb.Actor{
+		Metadata: &ateapipb.ResourceMetadata{Name: actorID, Atespace: atespace}, ActorTemplateNamespace: ns, ActorTemplateName: "tmpl",
+		Status: ateapipb.Actor_STATUS_RUNNING,
+		WorkerAssignment: &ateapipb.WorkerAssignment{
+			WorkerNamespace: ns, WorkerPool: pool, WorkerPod: pod, WorkerPodIp: "10.0.0.5",
+		},
+	})
+	if err != nil {
+		t.Fatalf("create actor: %v", err)
+	}
+	if err := persistence.CreateWorker(ctx, &ateapipb.Worker{
+		WorkerNamespace: ns, WorkerPool: pool, WorkerPod: pod,
+		WorkerPodUid: "uid-rdw",
+		State:        ateapipb.Worker_STATE_DRAINING,
+		Assignment: &ateapipb.Assignment{
+			ActorTemplate: &ateapipb.KubeNamespacedObjectRef{Namespace: ns, Name: "tmpl"},
+			Actor:         &ateapipb.ObjectRef{Atespace: createdActor.GetMetadata().GetAtespace(), Name: createdActor.GetMetadata().GetName()},
+			ActorUid:      "old-incarnation-uid",
+		},
+	}); err != nil {
+		t.Fatalf("create worker: %v", err)
+	}
+
+	if err := s.reconcileDeadWorker(ctx, ns, pool, pod); err != nil {
+		t.Fatalf("reconcileDeadWorker = %v, want nil", err)
+	}
+	// The dead worker should be deleted.
+	if _, err := persistence.GetWorker(ctx, ns, pool, pod); !errors.Is(err, store.ErrNotFound) {
+		t.Errorf("worker not deleted: err=%v", err)
+	}
+	// Because ActorUid did not match, the new actor must remain RUNNING.
+	got, err := persistence.GetActor(ctx, resources.ActorRef{Name: actorID, Atespace: atespace})
+	if err != nil {
+		t.Fatalf("get actor: %v", err)
+	}
+	if got.GetStatus() != ateapipb.Actor_STATUS_RUNNING {
+		t.Errorf("actor status = %v, want RUNNING (should ignore dead worker assigned to stale incarnation)", got.GetStatus())
+	}
+}
+
 // TestSyncer_ReconcileOrphanedWorkers verifies the startup reconcile: a stored
 // worker whose pod no longer exists is cleaned up (and its actor released), while
 // a worker whose pod is still live is preserved. This is the durability backstop
@@ -640,13 +692,14 @@ func TestSyncer_ReconcileOrphanedWorkers(t *testing.T) {
 
 	// An orphan worker (no pod) whose actor is still RUNNING must be cleaned up.
 	atespace, actorID := "team-recon", "actor-recon"
-	if _, err := persistence.CreateActor(ctx, &ateapipb.Actor{
+	createdActor, err := persistence.CreateActor(ctx, &ateapipb.Actor{
 		Metadata: &ateapipb.ResourceMetadata{Name: actorID, Atespace: atespace}, ActorTemplateNamespace: ns, ActorTemplateName: "tmpl",
 		Status: ateapipb.Actor_STATUS_RUNNING,
 		WorkerAssignment: &ateapipb.WorkerAssignment{
 			WorkerNamespace: ns, WorkerPool: pool, WorkerPod: "worker-orphan", WorkerPodIp: "10.0.0.10",
 		},
-	}); err != nil {
+	})
+	if err != nil {
 		t.Fatalf("create actor: %v", err)
 	}
 	if err := persistence.CreateWorker(ctx, &ateapipb.Worker{
@@ -655,7 +708,8 @@ func TestSyncer_ReconcileOrphanedWorkers(t *testing.T) {
 		State: ateapipb.Worker_STATE_DRAINING,
 		Assignment: &ateapipb.Assignment{
 			ActorTemplate: &ateapipb.KubeNamespacedObjectRef{Namespace: ns, Name: "tmpl"},
-			Actor:         &ateapipb.ObjectRef{Name: actorID, Atespace: atespace},
+			Actor:         &ateapipb.ObjectRef{Atespace: createdActor.GetMetadata().GetAtespace(), Name: createdActor.GetMetadata().GetName()},
+			ActorUid:      createdActor.GetMetadata().GetUid(),
 		},
 	}); err != nil {
 		t.Fatalf("create orphan worker: %v", err)
@@ -719,15 +773,16 @@ func TestReleaseActorOnDeadWorker_StatusTransitions(t *testing.T) {
 			s := &WorkerPoolSyncer{persistence: persistence}
 
 			atespace, actorID := "team-status", "actor-status"
-			if _, err := persistence.CreateActor(ctx, &ateapipb.Actor{
+			createdActor, err := persistence.CreateActor(ctx, &ateapipb.Actor{
 				Metadata:               &ateapipb.ResourceMetadata{Name: actorID, Atespace: atespace},
 				ActorTemplateNamespace: ns,
 				ActorTemplateName:      "tmpl",
 				Status:                 tc.start,
 				WorkerAssignment: &ateapipb.WorkerAssignment{
 					WorkerNamespace: ns, WorkerPool: pool, WorkerPod: pod, WorkerPodIp: ip, WorkerPodUid: "uid",
 				},
-			}); err != nil {
+			})
+			if err != nil {
 				t.Fatalf("create actor: %v", err)
 			}
 			if err := persistence.CreateWorker(ctx, &ateapipb.Worker{
@@ -737,7 +792,8 @@ func TestReleaseActorOnDeadWorker_StatusTransitions(t *testing.T) {
 				State:        ateapipb.Worker_STATE_ACTIVE,
 				Assignment: &ateapipb.Assignment{
 					ActorTemplate: &ateapipb.KubeNamespacedObjectRef{Namespace: ns, Name: "tmpl"},
-					Actor:         &ateapipb.ObjectRef{Name: actorID, Atespace: atespace},
+					Actor:         &ateapipb.ObjectRef{Atespace: createdActor.GetMetadata().GetAtespace(), Name: createdActor.GetMetadata().GetName()},
+					ActorUid:      createdActor.GetMetadata().GetUid(),
 				},
 			}); err != nil {
 				t.Fatalf("create worker: %v", err)

cmd/ateapi/internal/controlapi/workflow_pause.go

@@ -233,12 +233,11 @@ func (s *FinalizePausedStep) Execute(ctx context.Context, input *PauseInput, sta
 			}
 			slog.Warn("Worker already gone during finalize pause, skipping release", "worker", assignment.GetWorkerPod())
 		} else {
-			// TODO(dberkov) - what if worker does not belong to this actor?
 			nodeName = worker.GetNodeName()
 			// Only free it if it still belongs to us
 
 			if wass := worker.Assignment; wass != nil {
-				if resources.ActorRefFromObjectRef(wass.Actor) == input.ActorRef {
+				if wass.GetActorUid() == latestActor.GetMetadata().GetUid() {
 					worker.Assignment = nil
 					err = s.store.UpdateWorker(ctx, worker, worker.Version)
 					if err != nil {

cmd/ateapi/internal/controlapi/workflow_resume.go

@@ -319,7 +319,7 @@ func (s *AssignWorkerStep) Execute(ctx context.Context, input *ResumeInput, stat
 		if worker.Assignment == nil {
 			continue
 		}
-		if resources.ActorRefFromObjectRef(worker.Assignment.Actor) != input.ActorRef {
+		if worker.Assignment.GetActorUid() != state.Actor.GetMetadata().GetUid() {
 			continue
 		}
 		if s.scheduler.Applies(worker, constraints) {
@@ -366,7 +366,11 @@ func (s *AssignWorkerStep) Execute(ctx context.Context, input *ResumeInput, stat
 			Namespace: state.Actor.GetActorTemplateNamespace(),
 			Name:      state.Actor.GetActorTemplateName(),
 		},
-		Actor: input.ActorRef.ToObjectRef(),
+		Actor: &ateapipb.ObjectRef{
+			Atespace: state.Actor.GetMetadata().GetAtespace(),
+			Name:     state.Actor.GetMetadata().GetName(),
+		},
+		ActorUid: state.Actor.GetMetadata().GetUid(),
 	}
 
 	if err := s.store.UpdateWorker(ctx, assignedWorker, assignedWorker.Version); err != nil {
@@ -504,8 +508,8 @@ func (s *CallAteletRestoreStep) CheckPrerequisite(ctx context.Context, input *Re
 		return status.Errorf(codes.FailedPrecondition, "Assigned worker is nil")
 	}
 	// Verify if the worker is still assigned to the same Actor.
-	assigned := state.Worker.GetAssignment().GetActor()
-	if resources.ActorRefFromObjectRef(assigned) != input.ActorRef {
+	assignedActorUID := state.Worker.GetAssignment().GetActorUid()
+	if assignedActorUID != state.Actor.GetMetadata().GetUid() {
 		slog.ErrorContext(ctx, "crashing actor because its assigned worker no longer belongs to it",
 			slog.String("worker", state.Worker.GetWorkerPod()),
 			slog.Any("assignment", state.Worker.GetAssignment()))