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

cmd/ateapi/internal/controlapi/syncer.go

@@ -340,15 +340,33 @@ func (s *WorkerPoolSyncer) reconcileDeadWorker(ctx context.Context, namespace, p
 	return s.persistence.DeleteWorker(ctx, namespace, pool, podName)
 }
 
+// storedWorkerListBackoff and storedWorkerListCap are the exponential backoff
+// schedule for retrying a failed page of the startup stored-worker scan. They
+// are vars so tests can shrink them.
+var (
+	storedWorkerListBackoff = 500 * time.Millisecond
+	storedWorkerListCap     = 30 * time.Second
+)
+
 // 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.
+//
+// Each page's ListWorkers call is retried with capped backoff until it succeeds
+// or ctx is cancelled, so a transient store error does not abandon the scan and
+// leave ghost workers behind until the next restart (the per-key workqueue
+// retries reconciles, but nothing retries this initial enqueue scan). Pages are
+// enqueued as they are read, so the whole worker set is never held in memory at
+// once and a late failure does not re-scan the pages already enqueued.
 func (s *WorkerPoolSyncer) enqueueStoredWorkers(ctx context.Context) {
 	var pageToken string
 	for {
-		page, err := s.persistence.ListWorkers(ctx, store.ListOptions{PageSize: 1000, PageToken: pageToken})
+		page, err := s.listWorkersPageWithRetry(ctx, pageToken)
 		if err != nil {
-			slog.ErrorContext(ctx, "Syncer: failed to list workers for orphan reconcile", slog.Any("err", err))
+			// Only ctx cancellation (ate-api-server shutdown) ends the retry
+			// loop. Pages read so far are already enqueued (partial progress);
+			// the rest are recovered by the next startup scan.
+			slog.ErrorContext(ctx, "Syncer: stopped enqueue of stored workers before completing the scan; remaining workers will be retried at the next startup", slog.Any("err", err))
 			return
 		}
 		for _, w := range page.Items {
@@ -361,6 +379,38 @@ func (s *WorkerPoolSyncer) enqueueStoredWorkers(ctx context.Context) {
 	}
 }
 
+// listWorkersPageWithRetry reads one page of workers, retrying the store call
+// with capped exponential backoff until it succeeds or ctx is cancelled. The
+// page token is a stateless cursor, so retrying the failed call with the same
+// token resumes from the same position. A fresh backoff per page means only
+// consecutive failures of the same call accumulate delay; a page that succeeds
+// resets it.
+func (s *WorkerPoolSyncer) listWorkersPageWithRetry(ctx context.Context, pageToken string) (store.ListResponse[*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,
+	}
+	for {
+		page, err := s.persistence.ListWorkers(ctx, store.ListOptions{PageSize: 1000, PageToken: pageToken})
+		if err == nil {
+			return page, nil
+		}
+		slog.WarnContext(ctx, "Syncer: failed to list a page of stored workers for orphan cleanup, retrying", slog.Any("err", err))
+		select {
+		case <-ctx.Done():
+			return store.ListResponse[*ateapipb.Worker]{}, fmt.Errorf("listing stored workers aborted: %w", ctx.Err())
+		case <-time.After(backoff.Step()):
+		}
+	}
+}
+
 // releaseActorOnDeadWorker resets the actor bound to a vanishing worker pod. An
 // actor that already reached STATUS_SUSPENDED (it saved its state cleanly during
 // graceful termination) is left untouched and remains resumable. An actor that

cmd/ateapi/internal/controlapi/syncer_test.go

@@ -19,6 +19,7 @@ import (
 	"errors"
 	"fmt"
 	"maps"
+	"strconv"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -743,6 +744,145 @@ func TestSyncer_ReconcileOrphanedWorkers(t *testing.T) {
 	}
 }
 
+// flakyListWorkersStore wraps a store and fails the first failsLeft ListWorkers
+// calls, then delegates, simulating a transient store error.
+type flakyListWorkersStore struct {
+	store.Interface
+	failsLeft int
+}
+
+func (f *flakyListWorkersStore) ListWorkers(ctx context.Context, opts store.ListOptions) (store.ListResponse[*ateapipb.Worker], error) {
+	if f.failsLeft > 0 {
+		f.failsLeft--
+		return store.ListResponse[*ateapipb.Worker]{}, errors.New("transient store error")
+	}
+	return f.Interface.ListWorkers(ctx, opts)
+}
+
+// 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)
+	}
+}
+
+// listWorkersPageWithRetry must stop retrying and return once the context is
+// cancelled, rather than spinning forever, when the store stays unavailable.
+func TestSyncer_ListWorkersPageWithRetry_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.listWorkersPageWithRetry(ctx, "")
+	if err == nil {
+		t.Fatal("listWorkersPageWithRetry() = nil error, want a context error")
+	}
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Errorf("listWorkersPageWithRetry() error = %v, want it to wrap context.DeadlineExceeded", err)
+	}
+}
+
+// pagedListWorkersStore serves workers in fixed pages keyed by a numeric page
+// token, and fails the page at failOnPage exactly once before serving it, to
+// exercise per-page streaming plus retry of a late page.
+type pagedListWorkersStore struct {
+	store.Interface
+	pages      [][]*ateapipb.Worker
+	failOnPage int
+	failed     bool
+}
+
+func (p *pagedListWorkersStore) ListWorkers(_ context.Context, opts store.ListOptions) (store.ListResponse[*ateapipb.Worker], error) {
+	idx := 0
+	if opts.PageToken != "" {
+		var err error
+		if idx, err = strconv.Atoi(opts.PageToken); err != nil {
+			return store.ListResponse[*ateapipb.Worker]{}, err
+		}
+	}
+	if idx == p.failOnPage && !p.failed {
+		p.failed = true
+		return store.ListResponse[*ateapipb.Worker]{}, errors.New("transient store error on page")
+	}
+	resp := store.ListResponse[*ateapipb.Worker]{Items: p.pages[idx]}
+	if idx+1 < len(p.pages) {
+		resp.NextPageToken = strconv.Itoa(idx + 1)
+	}
+	return resp, nil
+}
+
+// enqueueStoredWorkers streams pages: a transient error on a later page must
+// retry just that page (the earlier pages stay enqueued) and the scan must
+// still enumerate every worker across all pages.
+func TestSyncer_EnqueueStoredWorkers_StreamsPagesAndRetriesLatePage(t *testing.T) {
+	prev := storedWorkerListBackoff
+	storedWorkerListBackoff = time.Millisecond
+	defer func() { storedWorkerListBackoff = prev }()
+
+	paged := &pagedListWorkersStore{
+		pages: [][]*ateapipb.Worker{
+			{{WorkerNamespace: "ns", WorkerPool: "p", WorkerPod: "w0"}},
+			{{WorkerNamespace: "ns", WorkerPool: "p", WorkerPod: "w1"}},
+			{{WorkerNamespace: "ns", WorkerPool: "p", WorkerPod: "w2"}},
+		},
+		failOnPage: 1, // second page fails once before succeeding
+	}
+	s := NewWorkerPoolSyncer(paged, nil, nil)
+
+	s.enqueueStoredWorkers(context.Background())
+
+	if !paged.failed {
+		t.Error("expected the injected page-2 failure to be exercised")
+	}
+	if got := s.queue.Len(); got != 3 {
+		t.Errorf("queue length = %d, want 3 (every worker across all pages must be enqueued despite the transient page error)", got)
+	}
+}
+
 // TestReleaseActorOnDeadWorker_StatusTransitions verifies that a running actor on
 // a deleted worker becomes CRASHED, while an actor that had already suspended
 // cleanly stays SUSPENDED (resumable).