Add new validator to fix a bug in ResumeActor

Author: minhngo248

cmd/ateapi/internal/controlapi/functional_test.go

@@ -1374,6 +1374,63 @@ func TestCreateActor_DefaultsRequiredWorkerPoolFromTemplate(t *testing.T) {
 	}
 }
 
+func TestResumeActor_LegacyActorUsesTemplateRequiredWorkerPool(t *testing.T) {
+	ns := namespaceForTest("ns-legacy-template-required-pool")
+	tc := setupTest(t, ns)
+	defer tc.cleanup()
+
+	createWorkerPool(t, tc, ns, "pool-a", map[string]string{"group": ns})
+	createWorkerPool(t, tc, ns, "pool-b", map[string]string{"group": ns})
+
+	actorTemplate := &atev1alpha1.ActorTemplate{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "tmpl-required-pool",
+			Namespace: ns,
+		},
+		Spec: atev1alpha1.ActorTemplateSpec{
+			PauseImage: "pause@sha256:abc",
+			SnapshotsConfig: atev1alpha1.SnapshotsConfig{
+				Location: "gs://fake-fake-fake",
+			},
+			Containers: []atev1alpha1.Container{
+				{Name: "main", Image: "main@sha256:abc", Command: []string{"/main"}},
+			},
+			RequiredWorkerPoolName: "pool-a",
+		},
+	}
+	if _, err := tc.substrateClient.ApiV1alpha1().ActorTemplates(ns).Create(context.Background(), actorTemplate, metav1.CreateOptions{}); err != nil {
+		t.Fatalf("failed to create actor template: %v", err)
+	}
+
+	createWorkerPod(t, tc, ns, "worker-b", "node1", "pool-b")
+
+	_, err := tc.persistence.CreateActor(context.Background(), &ateapipb.Actor{
+		Metadata: &ateapipb.ResourceMetadata{
+			Atespace: testAtespace,
+			Name:     "legacy-id1",
+		},
+		Status:                 ateapipb.Actor_STATUS_SUSPENDED,
+		ActorTemplateNamespace: ns,
+		ActorTemplateName:      "tmpl-required-pool",
+	})
+	if err != nil {
+		t.Fatalf("CreateActor in persistence failed: %v", err)
+	}
+
+	_, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{
+		Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "legacy-id1"},
+	})
+	assertGrpcError(t, err, codes.FailedPrecondition, "no free workers available")
+
+	getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "legacy-id1"}})
+	if err != nil {
+		t.Fatalf("GetActor failed: %v", err)
+	}
+	if got := getResp.GetWorkerPoolName(); got != "" {
+		t.Errorf("expected actor to remain unassigned, got worker_pool_name=%q", got)
+	}
+}
+
 func TestResumeActor_RequiredWorkerPoolDoesNotFallback(t *testing.T) {
 	ns := namespaceForTest("ns-required-pool-no-fallback")
 	tc := setupTest(t, ns)

cmd/ateapi/internal/controlapi/workflow_resume.go

@@ -170,6 +170,10 @@ func (s *AssignWorkerStep) Execute(ctx context.Context, input *ResumeInput, stat
 	}
 
 	var assignedWorker *ateapipb.Worker
+	requiredWorkerPool := state.Actor.GetRequiredWorkerPoolName()
+	if requiredWorkerPool == "" {
+		requiredWorkerPool = state.ActorTemplate.Spec.RequiredWorkerPoolName
+	}
 
 	// Check if we already have a worker assigned from a previous failed attempt.
 	// This can happen if ateapi crashed after updating worker with actor assignment,
@@ -181,7 +185,7 @@ func (s *AssignWorkerStep) Execute(ctx context.Context, input *ResumeInput, stat
 		if worker.Assignment.Actor.Atespace != input.Atespace || worker.Assignment.Actor.Name != input.ActorName {
 			continue
 		}
-		eligible, err := isWorkerEligibleForActor(worker, state.ActorTemplate.Spec.SandboxClass, state.ActorTemplate.Spec.WorkerSelector, state.Actor.GetWorkerSelector(), state.Actor.GetRequiredWorkerPoolName())
+		eligible, err := isWorkerEligibleForActor(worker, state.ActorTemplate.Spec.SandboxClass, state.ActorTemplate.Spec.WorkerSelector, state.Actor.GetWorkerSelector(), requiredWorkerPool)
 		if err != nil {
 			return fmt.Errorf("while checking worker eligibility: %w", err)
 		}
@@ -208,7 +212,7 @@ func (s *AssignWorkerStep) Execute(ctx context.Context, input *ResumeInput, stat
 		}(releaseWorker)
 	}
 	if assignedWorker == nil {
-		pickedWorker, err := s.findFreeWorker(workers, state.ActorTemplate.Spec.SandboxClass, state.ActorTemplate.Spec.WorkerSelector, state.Actor.GetWorkerSelector(), state.Actor.GetRequiredWorkerPoolName(), state.Actor.GetLatestSnapshotInfo().GetLocal().GetNodeVmsWithLocalSnapshots())
+		pickedWorker, err := s.findFreeWorker(workers, state.ActorTemplate.Spec.SandboxClass, state.ActorTemplate.Spec.WorkerSelector, state.Actor.GetWorkerSelector(), requiredWorkerPool, state.Actor.GetLatestSnapshotInfo().GetLocal().GetNodeVmsWithLocalSnapshots())
 		if err != nil {
 			return err
 		}