diff --git a/cmd/ateapi/internal/controlapi/syncer.go b/cmd/ateapi/internal/controlapi/syncer.go index 93f3d0507..984cfb801 100644 --- a/cmd/ateapi/internal/controlapi/syncer.go +++ b/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 diff --git a/cmd/ateapi/internal/controlapi/syncer_test.go b/cmd/ateapi/internal/controlapi/syncer_test.go index 917ef0299..1db18baa4 100644 --- a/cmd/ateapi/internal/controlapi/syncer_test.go +++ b/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).