diff --git a/cmd/ateapi/internal/controlapi/syncer.go b/cmd/ateapi/internal/controlapi/syncer.go index 77919b439..f7493e8a9 100644 --- a/cmd/ateapi/internal/controlapi/syncer.go +++ b/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 } diff --git a/cmd/ateapi/internal/controlapi/syncer_test.go b/cmd/ateapi/internal/controlapi/syncer_test.go index 1569d5f10..a842778ee 100644 --- a/cmd/ateapi/internal/controlapi/syncer_test.go +++ b/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).