From e5ebaaaf543fc55fad65a920c4af40ec276b209e Mon Sep 17 00:00:00 2001 From: mesutoezdil Date: Wed, 29 Jul 2026 23:24:28 +0200 Subject: [PATCH 1/2] ateapi: backfill unset worker state during pod sync The scheduler now skips any worker whose state is not STATE_ACTIVE, but the syncer only sets State when it creates a worker record. The update path diffs Ip, SandboxClass and Labels and never touches State, and the Redis store outlives ate-api-server while worker pods are not recreated on upgrade. Every worker record written before the State field existed therefore stays at STATE_UNSPECIFIED and is permanently unschedulable, so actor placement fails with ErrNoCapacity until each worker pod is deleted and recreated. Backfill the unset state to STATE_ACTIVE when syncing a running pod, which converges legacy records on the first startup sync. Only the unset state is backfilled, so a DRAINING worker whose pod name is reused is not resurrected. ValidateWorker already documents STATE_UNSPECIFIED as tolerated for backward compatibility, and workflow_resume rejects only DRAINING, so the scheduler's strict check stays as is. --- cmd/ateapi/internal/controlapi/syncer.go | 12 +++ cmd/ateapi/internal/controlapi/syncer_test.go | 79 +++++++++++++++++++ 2 files changed, 91 insertions(+) diff --git a/cmd/ateapi/internal/controlapi/syncer.go b/cmd/ateapi/internal/controlapi/syncer.go index c5756e57f..ca1e3ac2b 100644 --- a/cmd/ateapi/internal/controlapi/syncer.go +++ b/cmd/ateapi/internal/controlapi/syncer.go @@ -181,6 +181,18 @@ func (s *WorkerPoolSyncer) syncWorkerToStore(ctx context.Context, pod *corev1.Po w.Labels = pool.GetLabels() changed = true } + if w.State == ateapipb.Worker_STATE_UNSPECIFIED { + // A worker record written before the State field existed. The store + // outlives ate-api-server and worker pods are not recreated on upgrade, + // so backfill it here: the scheduler places only onto STATE_ACTIVE + // workers, and nothing else writes State onto an existing record. The pod + // is running (a Terminating pod returned above), so ACTIVE is correct. + // Only the unset state is backfilled, so a DRAINING worker whose pod name + // gets reused is never resurrected. + slog.InfoContext(ctx, "Syncer: updating worker in store (backfilling unset state)", slog.String("worker", pod.Namespace+"/"+pod.Name)) + w.State = ateapipb.Worker_STATE_ACTIVE + changed = true + } if changed { if err = s.persistence.UpdateWorker(ctx, w, w.Version); err != nil { diff --git a/cmd/ateapi/internal/controlapi/syncer_test.go b/cmd/ateapi/internal/controlapi/syncer_test.go index fa3da123c..49fc8f8c0 100644 --- a/cmd/ateapi/internal/controlapi/syncer_test.go +++ b/cmd/ateapi/internal/controlapi/syncer_test.go @@ -373,6 +373,85 @@ func TestSyncer_OmittedFields(t *testing.T) { } } +// TestSyncer_BackfillsUnsetWorkerState covers the two live-pod cases for the +// State field. A record left at STATE_UNSPECIFIED (written before the field +// existed) is backfilled to STATE_ACTIVE: the store outlives ate-api-server, the +// scheduler places only onto STATE_ACTIVE workers, and nothing else ever writes +// State, so without the backfill such a record stays unschedulable for the life +// of its pod. A DRAINING record is left alone, so a pod reusing a dead worker's +// name cannot resurrect it. +func TestSyncer_BackfillsUnsetWorkerState(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + ns, poolName, ip := "ns-backfill", "pool1", "10.0.0.6" + pool := &atev1alpha1.WorkerPool{ + ObjectMeta: metav1.ObjectMeta{Name: poolName, Namespace: ns, Labels: map[string]string{"foo": "bar"}}, + Spec: atev1alpha1.WorkerPoolSpec{SandboxClass: "gvisor"}, + } + + //nolint:staticcheck // NewSimpleClientset is the only available fake clientset for versioned CRDs. + fakeAte := atefake.NewSimpleClientset(pool) + ateInformerFactory := externalversions.NewSharedInformerFactory(fakeAte, 0) + workerPoolLister := ateInformerFactory.Api().V1alpha1().WorkerPools().Lister() + ateInformerFactory.Start(ctx.Done()) + ateInformerFactory.WaitForCacheSync(ctx.Done()) + + // Every field except State already matches the pod, so the state is the only + // thing left for the sync to reconcile. + newWorker := func(podName string, state ateapipb.Worker_State) *ateapipb.Worker { + return &ateapipb.Worker{ + WorkerNamespace: ns, WorkerPool: poolName, WorkerPod: podName, Ip: ip, + WorkerPodUid: "08675309-4a65-6e6e-7973-6e756d626572", NodeName: "node1", + SandboxClass: "gvisor", Labels: map[string]string{"foo": "bar"}, + State: state, + } + } + runningPod := func(podName string) *corev1.Pod { + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: podName, + Namespace: ns, + UID: "08675309-4a65-6e6e-7973-6e756d626572", + Labels: map[string]string{workerPodLabel: poolName}, + }, + Spec: corev1.PodSpec{NodeName: "node1"}, + Status: corev1.PodStatus{Phase: corev1.PodRunning, PodIP: ip, PodIPs: []corev1.PodIP{{IP: ip}}}, + } + } + + tests := []struct { + name string + podName string + stored ateapipb.Worker_State + want ateapipb.Worker_State + }{ + {"unset state backfilled", "worker-legacy", ateapipb.Worker_STATE_UNSPECIFIED, ateapipb.Worker_STATE_ACTIVE}, + {"draining state preserved", "worker-draining", ateapipb.Worker_STATE_DRAINING, ateapipb.Worker_STATE_DRAINING}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + persistence, cleanup := storetest.SetupTestStore(t) + defer cleanup() + + if err := persistence.CreateWorker(ctx, newWorker(tc.podName, tc.stored)); err != nil { + t.Fatalf("create worker: %v", err) + } + + s := &WorkerPoolSyncer{persistence: persistence, workerPoolLister: workerPoolLister} + s.syncWorkerToStore(ctx, runningPod(tc.podName)) + + w, err := persistence.GetWorker(ctx, ns, poolName, tc.podName) + if err != nil { + t.Fatalf("get worker: %v", err) + } + if w.GetState() != tc.want { + t.Errorf("worker state = %v, want %v", w.GetState(), tc.want) + } + }) + } +} + // TestSyncer_SoftDelete_MarksDraining verifies that a pod entering Terminating // (DeletionTimestamp set) flips its worker to STATE_DRAINING without deleting the // worker record or touching the bound actor — the actor is still gracefully From e66351699e5b79773064896dade68f92644430ab Mon Sep 17 00:00:00 2001 From: mesutoezdil Date: Sat, 1 Aug 2026 07:49:28 +0200 Subject: [PATCH 2/2] ateapi: drop unsafe unset-state backfill in syncer Blindly promoting STATE_UNSPECIFIED to STATE_ACTIVE could resurrect a worker whose state was unset by an unrelated bug, not just a legacy pre-field record. Backward-incompatible changes are fine at this stage, so handle stale legacy records operationally instead. --- cmd/ateapi/internal/controlapi/syncer.go | 19 +---- cmd/ateapi/internal/controlapi/syncer_test.go | 81 +------------------ 2 files changed, 4 insertions(+), 96 deletions(-) diff --git a/cmd/ateapi/internal/controlapi/syncer.go b/cmd/ateapi/internal/controlapi/syncer.go index 9dd86729e..b584a7e92 100644 --- a/cmd/ateapi/internal/controlapi/syncer.go +++ b/cmd/ateapi/internal/controlapi/syncer.go @@ -101,7 +101,7 @@ func (s *WorkerPoolSyncer) Start(ctx context.Context) { // Reconcile the other direction: clean up stored workers whose pods no // longer exist. This recovers delete events missed while ate-api-server - // was down — neither the watch relist nor the resync period can replay a + // was down, neither the watch relist nor the resync period can replay a // delete across a process restart, because the informer cache starts empty. s.reconcileOrphanedWorkers(ctx) }() @@ -115,7 +115,7 @@ func (s *WorkerPoolSyncer) syncWorkerToStore(ctx context.Context, pod *corev1.Po if pod.DeletionTimestamp != nil { // The pod has entered Terminating: mark the worker DRAINING so the // scheduler stops routing new actors to it. We deliberately do NOT touch - // the bound actor here — inside the pod ateom has received SIGTERM and is + // the bound actor here, inside the pod ateom has received SIGTERM and is // gracefully shutting the actor down. Actor cleanup happens on the Pod // Deleted event. if err := s.markWorkerDraining(ctx, pod.Namespace, pod.Labels[workerPodLabel], pod.Name); err != nil { @@ -185,19 +185,6 @@ func (s *WorkerPoolSyncer) syncWorkerToStore(ctx context.Context, pod *corev1.Po w.Labels = pool.GetLabels() changed = true } - if w.State == ateapipb.Worker_STATE_UNSPECIFIED { - // A worker record written before the State field existed. The store - // outlives ate-api-server and worker pods are not recreated on upgrade, - // so backfill it here: the scheduler places only onto STATE_ACTIVE - // workers, and nothing else writes State onto an existing record. The pod - // is running (a Terminating pod returned above), so ACTIVE is correct. - // Only the unset state is backfilled, so a DRAINING worker whose pod name - // gets reused is never resurrected. - slog.InfoContext(ctx, "Syncer: updating worker in store (backfilling unset state)", slog.String("worker", pod.Namespace+"/"+pod.Name)) - w.State = ateapipb.Worker_STATE_ACTIVE - changed = true - } - if changed { if err = s.persistence.UpdateWorker(ctx, w, w.Version); err != nil { slog.ErrorContext(ctx, "Failed to update worker in store", slog.Any("err", err)) @@ -212,7 +199,7 @@ func isWorkerEligible(pod *corev1.Pod) bool { // markWorkerDraining transitions a worker to STATE_DRAINING so the scheduler // stops routing new actors to it while its pod is Terminating. Best-effort: if // the worker is already gone or already draining, or a concurrent update wins, -// there is nothing more to do — the Pod Deleted event will clean up the record. +// there is nothing more to do, the Pod Deleted event will clean up the record. func (s *WorkerPoolSyncer) markWorkerDraining(ctx context.Context, namespace, pool, podName string) error { worker, err := s.persistence.GetWorker(ctx, namespace, pool, podName) if err != nil { diff --git a/cmd/ateapi/internal/controlapi/syncer_test.go b/cmd/ateapi/internal/controlapi/syncer_test.go index 6da51186b..df621d512 100644 --- a/cmd/ateapi/internal/controlapi/syncer_test.go +++ b/cmd/ateapi/internal/controlapi/syncer_test.go @@ -367,88 +367,9 @@ func TestSyncer_OmittedFields(t *testing.T) { } } -// TestSyncer_BackfillsUnsetWorkerState covers the two live-pod cases for the -// State field. A record left at STATE_UNSPECIFIED (written before the field -// existed) is backfilled to STATE_ACTIVE: the store outlives ate-api-server, the -// scheduler places only onto STATE_ACTIVE workers, and nothing else ever writes -// State, so without the backfill such a record stays unschedulable for the life -// of its pod. A DRAINING record is left alone, so a pod reusing a dead worker's -// name cannot resurrect it. -func TestSyncer_BackfillsUnsetWorkerState(t *testing.T) { - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - - ns, poolName, ip := "ns-backfill", "pool1", "10.0.0.6" - pool := &atev1alpha1.WorkerPool{ - ObjectMeta: metav1.ObjectMeta{Name: poolName, Namespace: ns, Labels: map[string]string{"foo": "bar"}}, - Spec: atev1alpha1.WorkerPoolSpec{SandboxClass: "gvisor"}, - } - - //nolint:staticcheck // NewSimpleClientset is the only available fake clientset for versioned CRDs. - fakeAte := atefake.NewSimpleClientset(pool) - ateInformerFactory := externalversions.NewSharedInformerFactory(fakeAte, 0) - workerPoolLister := ateInformerFactory.Api().V1alpha1().WorkerPools().Lister() - ateInformerFactory.Start(ctx.Done()) - ateInformerFactory.WaitForCacheSync(ctx.Done()) - - // Every field except State already matches the pod, so the state is the only - // thing left for the sync to reconcile. - newWorker := func(podName string, state ateapipb.Worker_State) *ateapipb.Worker { - return &ateapipb.Worker{ - WorkerNamespace: ns, WorkerPool: poolName, WorkerPod: podName, Ip: ip, - WorkerPodUid: "08675309-4a65-6e6e-7973-6e756d626572", NodeName: "node1", - SandboxClass: "gvisor", Labels: map[string]string{"foo": "bar"}, - State: state, - } - } - runningPod := func(podName string) *corev1.Pod { - return &corev1.Pod{ - ObjectMeta: metav1.ObjectMeta{ - Name: podName, - Namespace: ns, - UID: "08675309-4a65-6e6e-7973-6e756d626572", - Labels: map[string]string{workerPodLabel: poolName}, - }, - Spec: corev1.PodSpec{NodeName: "node1"}, - Status: corev1.PodStatus{Phase: corev1.PodRunning, PodIP: ip, PodIPs: []corev1.PodIP{{IP: ip}}}, - } - } - - tests := []struct { - name string - podName string - stored ateapipb.Worker_State - want ateapipb.Worker_State - }{ - {"unset state backfilled", "worker-legacy", ateapipb.Worker_STATE_UNSPECIFIED, ateapipb.Worker_STATE_ACTIVE}, - {"draining state preserved", "worker-draining", ateapipb.Worker_STATE_DRAINING, ateapipb.Worker_STATE_DRAINING}, - } - for _, tc := range tests { - t.Run(tc.name, func(t *testing.T) { - persistence, cleanup := storetest.SetupTestStore(t) - defer cleanup() - - if err := persistence.CreateWorker(ctx, newWorker(tc.podName, tc.stored)); err != nil { - t.Fatalf("create worker: %v", err) - } - - s := &WorkerPoolSyncer{persistence: persistence, workerPoolLister: workerPoolLister} - s.syncWorkerToStore(ctx, runningPod(tc.podName)) - - w, err := persistence.GetWorker(ctx, ns, poolName, tc.podName) - if err != nil { - t.Fatalf("get worker: %v", err) - } - if w.GetState() != tc.want { - t.Errorf("worker state = %v, want %v", w.GetState(), tc.want) - } - }) - } -} - // TestSyncer_SoftDelete_MarksDraining verifies that a pod entering Terminating // (DeletionTimestamp set) flips its worker to STATE_DRAINING without deleting the -// worker record or touching the bound actor — the actor is still gracefully +// worker record or touching the bound actor, the actor is still gracefully // shutting down inside the pod. func TestSyncer_SoftDelete_MarksDraining(t *testing.T) { ctx := context.Background()