coordinator: hold dispatch until a matching worker arrives

Merged
YO
6d ago
+1078-49

What

A dispatch whose selector matches no healthy worker is now held for a worker to register instead of being refused.

  • Dispatch accepts the task, records the run as Queued with the same two conditions the scheduler's queue writes in that situation (Runnable=False, WorkerReady=False, reason NoMatchingWorker or NoAvailableWorker), and stores the task with a HoldUntil. The first matching worker to poll takes it.
  • coordinator.worker_wait_timeout (default 2m, env DAGU_COORDINATOR_WORKER_WAIT_TIMEOUT) bounds the wait. 0 restores the immediate refusal. The zombie detector fails runs whose wait has passed, so the failure follows the deadline by at most one detection interval; the run's error names what it waited for: no worker matched the selector {zone: b} within 2m0s.
  • RequestCancel aborts a held run and drops its task, so a worker registering later does not start a run its owner already cancelled.
  • Held tasks outlive the dispatch store's reservation TTL and are never recycled by it; once their wait has passed they stop being claimable and wait for the coordinator to fail the run and drop them, so a coordinator that dies between the two steps finds the task again.
  • Events: the DAG-run status snapshot in events carries conditions, and event data carries worker_selector, so a controller that opens worker capacity on demand can read from dag.run.queued which labels a waiting run needs.

Why

Workers that scale from zero. Today a start, a retry, a resume, or a sub-DAG that reaches the coordinator before its worker exists fails at once with no workers match the required selector, so anything that creates workers on demand has to arrange them before dispatching. The scheduler's queue already holds such runs; this gives the direct-dispatch paths the same behaviour, with the same conditions, so a waiting run reads the same however it was dispatched.

What does not change

  • The scheduler's re-dispatch of a queued run keeps today's refusal: the queue already holds that run and reads the refusal to keep it queued (queueDispatchStatusForTask tells the two apart).
  • The in-memory dispatch path (no shared store) keeps refusing; it has no durable task to hold.
  • HandlerConfig.WorkerWaitTimeout defaults to 0 in tests, so existing expectations stand.
  • No proto change: HoldUntil lives in the JSON task payload.

Tests

  • internal/service/coordinator: held until a matching worker polls; held with no worker at all; deadline fails the run naming the selector and the task is gone; a task claimed in time is left alone; queue re-dispatch still refused; disabled refuses; cancel aborts and drops.
  • internal/persis/store: held pending outlives the TTL while an ordinary one is recycled; expired held is listed, not claimed, until deleted; delete leaves claims alone; the hold survives an index rebuild.
  • internal/intg/distr: a direct dispatch with zone: b waits as Queued with WorkerReady=False/NoMatchingWorker, then runs on worker-b when it registers; with a 2s wait it fails naming the selector.
  • internal/cmn/config: default, set, "0", invalid, env, negative rejected.
  • internal/eventstore: conditions round-trip through the snapshot.

Notes for reviewers

  • Stacked on #3197 (FormatWorkerSelector, applyTaskWorkerSelector); the diff against main shrinks once that merges.
  • The Helm chart gains no value for the new key on purpose; it is reachable through the config file and the environment like the other coordinator settings.

Summary by cubic

Direct dispatches (start, retry, resume, sub-DAG) whose selector matches no healthy worker are now held as Queued runs instead of failing immediately; the first worker that matches the selector takes the run up. New coordinator.worker_wait_timeout config (default 2m, env DAGU_COORDINATOR_WORKER_WAIT_TIMEOUT) bounds the wait; 0 restores the immediate refusal.

The zombie detector fails runs whose wait passed, naming the selector in the error, and drops the task; a run that cannot be read keeps its task for the next pass, and only a run that no longer exists loses it. Cancelling a held run aborts it and drops the task. Held tasks outlive the dispatch store's reservation TTL and are never recycled by it. Events now carry the run's conditions in the status snapshot and worker_selector in the event data, so a controller that opens worker capacity on demand can react to dag.run.queued.

The scheduler's re-dispatch of a queued run keeps today's refusal, and the in-memory dispatch path keeps refusing since it has no durable task to hold.

Summary by CodeRabbit

  • New Features
    • Dispatches can wait for a healthy worker matching their requirements to become available, rather than failing immediately. If no matching worker arrives before the configured timeout, the run fails.
    • Configure the wait duration with coordinator.worker_wait_timeout or COORDINATOR_WORKER_WAIT_TIMEOUT. The default is 2 minutes; setting it to 0 disables waiting.
    • Canceling a queued dispatch removes it from pending delivery.