fix(ateapi): retry stored-worker scan in syncer startup

cmd/ateapi/internal/controlapi/syncer.go

@@ -310,19 +310,77 @@ func (s *WorkerPoolSyncer) reconcileDeadWorker(ctx context.Context, namespace, p
 // enqueueStoredWorkers enqueues a key for every worker record in the store.
 // Records whose pods are live and unchanged reconcile to a no-op; orphaned
 // records (pod gone, or its name reused by a new pod UID) get cleaned up.
+//
+// It retries the paginated scan with bounded backoff on a store error. Returning
+// on the first page error would skip the rest of the stored workers until the
+// next ate-api-server restart, leaving ghost workers behind: the per-key
+// workqueue retries reconciles, but nothing retries this initial enqueue scan.
 func (s *WorkerPoolSyncer) enqueueStoredWorkers(ctx context.Context) {
+	workers, err := s.listAllStoredWorkers(ctx)
+	if err != nil {
+		slog.ErrorContext(ctx, "Syncer: giving up enqueue of stored workers after repeated list failures; ghost workers will be retried at the next startup", slog.Any("err", err))
+		return
+	}
+	for _, w := range workers {
+		s.queue.Add(workerKey{namespace: w.GetWorkerNamespace(), pool: w.GetWorkerPool(), name: w.GetWorkerPod()})
+	}
+}
+
+// storedWorkerListBackoff is the exponential backoff schedule for retrying the
+// startup stored-worker scan when the store errors. It ramps then plateaus at
+// Cap; listAllStoredWorkers retries until the scan succeeds or the context is
+// cancelled (ate-api-server shutdown). Duration is a var so tests can shrink it.
+var (
+	storedWorkerListBackoff = 500 * time.Millisecond
+	storedWorkerListCap     = 30 * time.Second
+)
+
+// listAllStoredWorkers paginates every worker record, retrying the whole scan
+// with capped exponential backoff until it succeeds or ctx is cancelled. A scan
+// that gave up after a fixed number of attempts could still skip the orphan
+// cleanup until the next restart if the store stayed briefly unavailable; the
+// retry runs in the background startup goroutine (off the serving path) and, on
+// a lasting outage, issues at most one cheap failing list per Cap interval.
+func (s *WorkerPoolSyncer) listAllStoredWorkers(ctx context.Context) ([]*ateapipb.Worker, error) {
+	backoff := wait.Backoff{
+		Duration: storedWorkerListBackoff,
+		Factor:   2.0,
+		Jitter:   0.1,
+		// Steps must be large enough for the ramp (Duration*Factor^n) to reach
+		// Cap, or Cap never triggers and the plateau sits at the last ramp step.
+		// With Duration=500ms, Factor=2, the ramp hits Cap=30s at step 6
+		// (0.5,1,2,4,8,16,30,30...).
+		Steps: 6,
+		Cap:   storedWorkerListCap,
+	}
+	var lastErr error
+	for {
+		workers, err := listAllWorkersPaged(ctx, s.persistence)
+		if err == nil {
+			return workers, nil
+		}
+		lastErr = err
+		slog.WarnContext(ctx, "Syncer: failed to list stored workers for orphan cleanup, retrying", slog.Any("err", err))
+		select {
+		case <-ctx.Done():
+			return nil, fmt.Errorf("listing stored workers aborted: %w; last store error: %v", ctx.Err(), lastErr)
+		case <-time.After(backoff.Step()):
+		}
+	}
+}
+
+// listAllWorkersPaged paginates all worker records in a single pass.
+func listAllWorkersPaged(ctx context.Context, persistence store.Interface) ([]*ateapipb.Worker, error) {
+	var workers []*ateapipb.Worker
 	var pageToken string
 	for {
-		workers, nextToken, err := s.persistence.ListWorkers(ctx, 1000, pageToken)
+		wPage, nextToken, err := persistence.ListWorkers(ctx, 1000, pageToken)
 		if err != nil {
-			slog.ErrorContext(ctx, "Syncer: failed to list workers for orphan reconcile", slog.Any("err", err))
-			return
-		}
-		for _, w := range workers {
-			s.queue.Add(workerKey{namespace: w.GetWorkerNamespace(), pool: w.GetWorkerPool(), name: w.GetWorkerPod()})
+			return nil, err
 		}
+		workers = append(workers, wPage...)
 		if nextToken == "" {
-			return
+			return workers, nil
 		}
 		pageToken = nextToken
 	}

cmd/ateapi/internal/controlapi/syncer_test.go

@@ -686,6 +686,88 @@ func TestSyncer_ReconcileOrphanedWorkers(t *testing.T) {
 	}
 }
 
+// flakyListWorkersStore wraps a store and fails the first failsLeft ListWorkers
+// calls, then delegates, simulating a transient store error mid-scan.
+type flakyListWorkersStore struct {
+	store.Interface
+	failsLeft int
+}
+
+func (f *flakyListWorkersStore) ListWorkers(ctx context.Context, pageSize int32, pageToken string) ([]*ateapipb.Worker, string, error) {
+	if f.failsLeft > 0 {
+		f.failsLeft--
+		return nil, "", errors.New("transient store error")
+	}
+	return f.Interface.ListWorkers(ctx, pageSize, pageToken)
+}
+
+// A transient store error mid-scan must not abandon the startup enqueue of
+// stored workers: enqueueStoredWorkers retries the worker list so a blip does
+// not skip workers (leaving ghost records) until the next restart. The per-key
+// workqueue retries reconciles, but not this initial scan.
+func TestSyncer_EnqueueStoredWorkers_RetriesTransientListError(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	// Shrink the retry backoff so the test's single retry is fast.
+	prev := storedWorkerListBackoff
+	storedWorkerListBackoff = time.Millisecond
+	defer func() { storedWorkerListBackoff = prev }()
+
+	persistence, cleanup := storetest.SetupTestStore(t)
+	defer cleanup()
+
+	ns, pool := "ns-enq-retry", "pool1"
+	if err := persistence.CreateWorker(ctx, &ateapipb.Worker{
+		WorkerNamespace: ns, WorkerPool: pool, WorkerPod: "worker-1", Ip: "10.0.0.10",
+		WorkerPodUid: "22222222-2222-2222-2222-222222222222", NodeName: "node1",
+		State: ateapipb.Worker_STATE_ACTIVE,
+	}); err != nil {
+		t.Fatalf("create worker: %v", err)
+	}
+
+	// The store errors on the first ListWorkers call; enqueue must retry rather
+	// than abandon the scan and skip the worker.
+	flaky := &flakyListWorkersStore{Interface: persistence, failsLeft: 1}
+	s := NewWorkerPoolSyncer(flaky, nil, nil)
+
+	s.enqueueStoredWorkers(ctx)
+
+	if flaky.failsLeft != 0 {
+		t.Errorf("flaky store failsLeft = %d, want 0 (ListWorkers should have been retried)", flaky.failsLeft)
+	}
+	if got := s.queue.Len(); got != 1 {
+		t.Errorf("queue length = %d, want 1 (the worker must be enqueued after the retry)", got)
+	}
+}
+
+// listAllStoredWorkers must stop retrying and return once the context is
+// cancelled, rather than spinning forever, when the store stays unavailable.
+func TestSyncer_ListAllStoredWorkers_StopsOnContextCancel(t *testing.T) {
+	prev := storedWorkerListBackoff
+	storedWorkerListBackoff = time.Millisecond
+	defer func() { storedWorkerListBackoff = prev }()
+
+	persistence, cleanup := storetest.SetupTestStore(t)
+	defer cleanup()
+
+	// Fails far more times than the test allows, so only ctx cancellation ends
+	// the loop.
+	flaky := &flakyListWorkersStore{Interface: persistence, failsLeft: 1 << 30}
+	s := NewWorkerPoolSyncer(flaky, nil, nil)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+
+	_, err := s.listAllStoredWorkers(ctx)
+	if err == nil {
+		t.Fatal("listAllStoredWorkers() = nil error, want a context error")
+	}
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Errorf("listAllStoredWorkers() error = %v, want it to wrap context.DeadlineExceeded", err)
+	}
+}
+
 // TestReleaseActorOnDeadWorker_StatusTransitions verifies that a running actor on
 // a deleted worker becomes CRASHED, while an actor that had already suspended
 // cleanly stays SUSPENDED (resumable).