Dynamic routing

Resource requests and worker affinity are hard constraints: they filter the workers that can run a workflow. Dynamic routing adds a soft preference on top — a scoring policy that ranks the workers that survived the filter, by queue depth, event-loop lag, payload locality, live utilization, or anything else you can express as a number. Without a policy, Flux picks the least-loaded eligible worker.

Declaring a policy

Pass routing=score(...) to workflow.with_options. The DSL lives in flux.routing:

from flux import ExecutionContext, workflow
from flux.routing import score, prefer, least, most, sticky, label, metric, resource, load, input


@workflow.with_options(
    routing=score(
        prefer(label("region") == input("region"), weight=10),  # payload locality
        prefer(metric("temp") < 60, weight=2),                   # threshold preference
        least(metric("queue_depth"), weight=5),                  # minimize a worker metric
        most(resource("memory_available")),                      # maximize a resource field
        sticky(weight=3),                                        # opt the relay hint into the score
        least(load()),                                           # built-in: active executions
    ),
)
async def train(ctx: ExecutionContext[dict]):
    ...

A policy is a weighted combination of terms over selectors.

Term types

• prefer(condition, weight=...) — scores 1 when the comparison holds. Conditions use ordinary Python operators (==, !=, <, <=, >, >=) between a selector and a constant or input(...).
• least(selector, weight=...) / most(selector, weight=...) — prefer the lowest / highest numeric value.
• sticky(weight=...) — opts the relayed-call() worker hint into the score. A workflow with a policy owns its score stage entirely: the hint participates only through this term (see Relationship to sticky routing).

Selectors

input("path") resolves against the execution’s input at dispatch time — dotted paths (input("customer.region")) descend nested dictionaries. This is how payload-driven locality works: the same workflow routes each execution by its own data.

How scoring works

1. Hard constraints filter first — a policy can never route to a worker that fails requests/affinity/runner matching, is unhealthy, or has no free capacity slot.
2. Each term is normalized to 0–1 across the eligible workers (so an unbounded load term cannot drown a boolean prefer), multiplied by its weight, and summed.
3. The highest total wins; ties break deterministically (lower load, then name).

Degradation is deliberate: a worker missing a metric scores 0 for that term; a metric absent on every worker makes the term a no-op; a malformed policy falls back to least-loaded selection. A routing policy can never strand an execution.

Policies are data, not code

The score(...) expression compiles to a JSON spec that is extracted statically at registration — the same AST mechanism as requests — and evaluated natively by the server. Nothing user-supplied executes in the dispatcher. The flip side: the policy must be declared with literal values (or input(...)); a policy the parser cannot extract fails registration with a clear error rather than silently routing differently than written.

Built-in worker metrics

Every worker publishes a standard metric set under the reserved flux. prefix on its heartbeat — no configuration needed ([flux.workers] builtin_metrics = true by default):

So these work with zero setup:

# Steer latency-sensitive work away from degraded-but-not-unhealthy workers
routing=score(least(metric("flux.loop_lag_p95_seconds"), weight=5), least(load()))

# Quarantine workers that accept work and fail it (full disk, sick GPU, ...)
routing=score(prefer(metric("flux.crash_rate") < 0.1, weight=10), least(load()))

Aggregates are computed on the worker over fixed windows and published as single scalars — the server stores only the latest snapshot per worker, never a time series. For history and trending, use the observability pipeline (see Worker observability).

Custom metrics providers

For anything the built-ins don’t cover, point the worker at your own callable (sync or async) returning dict[str, float]:

# myapp/routing.py — runs inside the worker process
import psutil


async def collect() -> dict[str, float]:
    return {
        "gpu_queue_depth": gpu_queue.qsize(),
        "shard_latency_ms": await probe_local_shard(),
        "scratch_free_gb": psutil.disk_usage("/scratch").free / 1e9,
    }

[flux.workers]
metrics_provider = "myapp.routing:collect"
metrics_interval = 10.0

The worker refreshes the provider on that cadence (sync providers run in a thread; a failure keeps the previous snapshot), merges the result with the built-ins, and advertises the snapshot on its heartbeat pong. This is the intended home for arbitrary routing logic: measure anything worker-side — including windowed aggregates like a rolling p95 you compute yourself — and publish it as a number the server can rank on declaratively.

Guardrails: a provider may publish up to 32 metrics (string keys ≤ 64 characters, finite numbers; the server caps the merged set at 64); invalid payloads are dropped with a warning, never an error. Provider keys under the reserved flux. prefix are stripped, so user values can never impersonate a built-in signal.

Observing routing decisions

• flux worker list / flux worker show <name> (and GET /workers) show each worker’s latest advertised metrics — the values the last dispatch decision actually saw.
• flux workflow show <name> includes the registered routing policy in the workflow’s metadata.

Relationship to sticky routing

Relayed call()s tag their child executions with the calling worker’s name (the X-Flux-Preferred-Worker hint), and workflows without a policy prefer that worker when eligible — keeping mesh hops on warm module caches. A workflow with a policy takes full ownership of the score stage: include sticky(weight=...) to blend the hint into your ranking, or omit it to override the hint entirely.

What’s next

• Worker affinity — the hard label constraint that filters before scoring runs
• Resource requests — capacity thresholds, the other hard constraint
• Dispatch modes — enabling the event dispatcher that evaluates policies
• Worker observability — inspecting worker state, heartbeats, and metrics in production