Capacity Classification & Fallback

capacity

Resource unavailability as a state, not an exception.

The system has always reacted to "the cloud has no room": Dataproc walks zone candidates, Vertex Ray walks region candidates. It has never represented it. A reaction lives in a stack frame, shows up only in a log line, and when it runs out raises something indistinguishable from a ModuleNotFoundError. A state has a name, a counter, a deadline, and a row someone can query.

This module is the representation. It owns four things, in dependency order:

1. classify — one classifier, three verdicts. Previously there were two classifiers with opposite defaults (compute_fallback for Dataproc, private helpers in ray_cluster for Vertex) and no shared vocabulary. Two classifiers that disagree is worse than one that is wrong.
2. CapacityPolicy — how much patience to spend, bounded by attempts and wall-clock.
3. CapacityLedger — the attempt log that ends up in job_telemetry under $.capacity.
4. walk — the loop: try candidates, then back off and try them again, until one works or the budget is gone.

Everything here is pure and import-free (stdlib only) so the whole thing unit-tests offline with an injected clock. The cluster submitters drive it.

Why three verdicts and not a boolean. The old question was hop, or raise? The right question is why, because only one of the three answers is worth waiting on:

Verdict	Meaning	Action
TRANSIENT_CAPACITY	no room right now	hop and retry — time will plausibly fix it
HARD_CEILING	not allowed this much here	hop only; waiting cannot raise a quota
CONFIG_FAULT	the request is malformed	stop; hopping ends in the same error

The middle one was earned on 2026-09-02: us-east1 allows 2 training T4s and a smoke asked for 7, so no amount of retrying could ever succeed there — while us-central1's contentless error the same afternoon was worth retrying and succeeded untouched two hours later. Same accelerator, same day, opposite correct behaviour. A policy that treated both as transient would have burned the entire time budget on the region that was arithmetically impossible.

An unrecognised message is TRANSIENT_CAPACITY, not CONFIG_FAULT — see classify.

CapacityAttempt dataclass

CapacityAttempt(
    candidate, verdict, message, elapsed_seconds
)

One attempt at one candidate, and what the cloud said about it.

message is kept verbatim, truncated only to keep a pathological traceback out of a telemetry column. Three of this campaign's defects were classifiers failing on strings nobody had thought of, and the only durable fix is keeping the strings: the next unrecognised wording is discoverable from a query instead of from a live failure.

to_json

to_json()

The telemetry shape — plain JSON types only, so it merges into job_telemetry.

Source code in src/scale_forecasting/capacity.py

def to_json(self) -> dict[str, object]:
    """The telemetry shape — plain JSON types only, so it merges into ``job_telemetry``."""
    return {
        "candidate": self.candidate,
        "verdict": self.verdict,
        "message": self.message,
        "elapsed_seconds": round(self.elapsed_seconds, 3),
    }

CapacityExhausted

CapacityExhausted(message, ledger)

Bases: EngineError

Every candidate was tried until the attempt or time budget ran out.

Carries the ledger so the caller can write it to telemetry and stamp the job row FAILED with failure_reason: CAPACITY_EXHAUSTED — the one failure worth re-running unchanged, and the thing an operator could not previously tell apart from a broken import.

Source code in src/scale_forecasting/capacity.py

def __init__(self, message: str, ledger: CapacityLedger) -> None:
    super().__init__(message)
    self.ledger = ledger

CapacityLedger dataclass

CapacityLedger(
    service,
    attempts=list(),
    exhausted=False,
    preflight=list(),
)

Every attempt a walk made, in order — what lands under job_telemetry.$.capacity.

Mutable and append-only by design: walk writes to it as it goes, so a caller holding the ledger can publish partial state while the walk is still running. That is what makes AWAITING_CAPACITY observable rather than merely nameable — without it, the only view of a job in the middle of a three-region walk would be the same RUNNING row a computing job has.

dead_candidates property

dead_candidates

Candidates that answered HARD_CEILING — never worth another attempt this run.

A quota ceiling does not move while a run is in flight, so re-trying one is guaranteed waste. Dropping them is what keeps a long walk from spending its whole budget re-asking a region that has already said no on arithmetic grounds.

record

record(candidate, verdict, message, elapsed_seconds)

Append one attempt, clipping an over-long message.

Source code in src/scale_forecasting/capacity.py

def record(self, candidate: str, verdict: str, message: str, elapsed_seconds: float) -> None:
    """Append one attempt, clipping an over-long message."""
    self.attempts.append(
        CapacityAttempt(
            candidate=candidate,
            verdict=verdict,
            message=message[:_MESSAGE_LIMIT],
            elapsed_seconds=elapsed_seconds,
        )
    )

to_json

to_json()

The $.capacity payload: the service, the attempt list, and whether it gave up.

Source code in src/scale_forecasting/capacity.py

def to_json(self) -> dict[str, object]:
    """The ``$.capacity`` payload: the service, the attempt list, and whether it gave up."""
    payload: dict[str, object] = {
        "service": self.service,
        "exhausted": self.exhausted,
        "n_attempts": len(self.attempts),
        "attempts": [a.to_json() for a in self.attempts],
    }
    if self.preflight:
        payload["preflight"] = self.preflight
    return payload

CapacityPolicy dataclass

CapacityPolicy(
    max_attempts=6,
    max_wall_seconds=3600.0,
    max_passes=0,
    backoff_seconds=60.0,
    backoff_multiplier=2.0,
    backoff_max_seconds=600.0,
)

How much patience to spend looking for room, bounded two ways at once.

Both bounds, because attempt cost spans two orders of magnitude across services. A Ray GPU provision is ~12 minutes per attempt (measured 2026-09-02); a Serverless batch rejection comes back in seconds. max_attempts=5 therefore means an hour on one and half a minute on the other, so neither bound alone means anything. Whichever is reached first ends the walk.

max_attempts counts individual candidate attempts, not passes over the candidate list — a three-region config with max_attempts=6 gets two full passes. 0 disables the bound.

max_wall_seconds is measured from the first attempt. It is checked before starting an attempt and before sleeping, never mid-attempt: a create that is already running is allowed to finish, because killing it would leak the very resource ray_cluster._clear_stale_resource exists to clean up. So the real ceiling is this plus one attempt's duration, which is honest and documented rather than pretended away. 0.0 disables the bound.

max_passes bounds complete sweeps of the candidate list. 0 disables it; 1 reproduces the pre-retry behaviour exactly — try each place once, then give up — which is what config.CapacityConfig.enabled = false resolves to. It exists because "no retry" cannot be expressed with the other two bounds: the walk does not know how many candidates it will get.

Back-off applies only between passes, never between candidates within a pass — hopping is free and waiting is not, so the walk exhausts the places it can try before it spends any time.

backoff_for

backoff_for(completed_passes)

Seconds to wait after completed_passes exhausted passes (pure, capped).

Source code in src/scale_forecasting/capacity.py

def backoff_for(self, completed_passes: int) -> float:
    """Seconds to wait after ``completed_passes`` exhausted passes (pure, capped)."""
    if completed_passes < 1:
        return 0.0
    delay = self.backoff_seconds * (self.backoff_multiplier ** (completed_passes - 1))
    return min(delay, self.backoff_max_seconds)

classify

classify(message, exc=None)

Read a provisioning failure and return one of VERDICTS (pure).

message should be the richest text available. On Vertex that means the resource's own error.message concatenated with the raised exception — the SDK's exception is a generic "returned an error" while the "Resources are insufficient in region" / quota text lives only on the resource, so classifying on the exception alone would never detect a stockout. exc is consulted only for its class name (_CAPACITY_EXCEPTION_NAMES), which is how a Compute Engine stockout surfaces from create_cluster with no useful string at all.

Precedence, and every step of it is load-bearing:

1. Capacity wins outright. A message can name a machine type and say the zone ran out of it; that is about the place, and hopping is exactly right.
2. Quota, when it is not also capacity, is a HARD_CEILING. The region has room; this project is simply not allowed more. Hop, never wait.
3. A named region-invariant cause is a CONFIG_FAULT. Stop.
4. A regional prerequisite reported absent is a HARD_CEILING. Hop, never wait. This sits after step 3 on purpose: "service account … was not found" is a config fault, and a missing thing that is missing everywhere must keep stopping. Only a resource kind known to be regional reaches this step.
5. Only then, the exception's type. _CONFIG_FAULT_EXCEPTION_NAMES (a malformed request) stops; _HARD_CEILING_EXCEPTION_NAMES (a prerequisite that does not exist here) hops without waiting. Deliberately last, so it can only rescue cases that would otherwise have fallen through to step 6 — every classification the text already makes is left exactly as it was.
6. Everything else is TRANSIENT_CAPACITY.

Step 6 is the deliberate asymmetry: any unrecognised or contentless platform error (such as "An internal error occurred on your cluster" or "Unexpected response.") defaults to TRANSIENT_CAPACITY so the multi-region fallback walks the remaining candidate regions rather than aborting in the first region. The costs are not symmetric: retrying a config fault wastes minutes and still ends in an error that names every candidate tried, while refusing to retry an unrecognised stock-out loses the multi-region fallback silently. Give up immediately only when the message names a cause that travels with the request.

Source code in src/scale_forecasting/capacity.py

def classify(message: str, exc: BaseException | None = None) -> str:
    """Read a provisioning failure and return one of `VERDICTS` (pure).

    ``message`` should be the richest text available. On Vertex that means the *resource's* own
    ``error.message`` concatenated with the raised exception — the SDK's exception is a generic
    "returned an error" while the "Resources are insufficient in region" / quota text lives only on
    the resource, so classifying on the exception alone would never detect a stockout. ``exc`` is
    consulted only for its class name (`_CAPACITY_EXCEPTION_NAMES`), which is how a Compute Engine
    stockout surfaces from ``create_cluster`` with no useful string at all.

    **Precedence, and every step of it is load-bearing:**

    1. **Capacity wins outright.** A message can name a machine type *and* say the zone ran out of
       it; that is about the place, and hopping is exactly right.
    2. **Quota, when it is not also capacity, is a `HARD_CEILING`.** The region has room; this
       project is simply not allowed more. Hop, never wait.
    3. **A named region-invariant cause is a `CONFIG_FAULT`.** Stop.
    4. **A *regional* prerequisite reported absent is a `HARD_CEILING`.** Hop, never wait. This sits
       after step 3 on purpose: "service account … was not found" is a config fault, and a missing
       thing that is missing everywhere must keep stopping. Only a resource kind known to be
       regional reaches this step.
    5. **Only then, the exception's *type*.** `_CONFIG_FAULT_EXCEPTION_NAMES` (a malformed request)
       stops; `_HARD_CEILING_EXCEPTION_NAMES` (a prerequisite that does not exist here) hops without
       waiting. Deliberately last, so it can only rescue cases that would otherwise have fallen
       through to step 6 — every classification the text already makes is left exactly as it was.
    6. **Everything else is `TRANSIENT_CAPACITY`.**

    Step 6 is the deliberate asymmetry: any unrecognised or contentless platform error (such as
    ``"An internal error occurred on your cluster"`` or ``"Unexpected response."``) defaults to
    `TRANSIENT_CAPACITY` so the multi-region fallback walks the remaining candidate regions rather
    than aborting in the first region. The costs are not symmetric: retrying a config fault wastes
    minutes and still ends in an error that names every candidate tried, while refusing to retry an
    unrecognised stock-out loses the multi-region fallback silently. Give up immediately only when
    the message names a cause that travels with the request.
    """
    low = message.lower()
    if exc is not None and type(exc).__name__ in _CAPACITY_EXCEPTION_NAMES:
        return TRANSIENT_CAPACITY
    if any(marker in low for marker in _CAPACITY_MARKERS):
        return TRANSIENT_CAPACITY
    if any(q in low for q in _QUOTA_WORDS) and any(e in low for e in _EXHAUSTION_WORDS):
        return HARD_CEILING
    if any(marker in low for marker in _CONFIG_FAULT_MARKERS):
        return CONFIG_FAULT
    if any(kind in low for kind in _REGIONAL_PREREQUISITE_MARKERS) and any(
        absent in low for absent in _ABSENCE_MARKERS
    ):
        return HARD_CEILING
    if exc is not None:
        name = type(exc).__name__
        if name in _CONFIG_FAULT_EXCEPTION_NAMES:
            return CONFIG_FAULT
        if name in _HARD_CEILING_EXCEPTION_NAMES:
            return HARD_CEILING
    return TRANSIENT_CAPACITY

current_publisher

current_publisher()

The ambient publisher, or None when nothing installed one (a bare API call, a test).

Source code in src/scale_forecasting/capacity.py

def current_publisher() -> Callable[[CapacityLedger], None] | None:
    """The ambient publisher, or ``None`` when nothing installed one (a bare API call, a test)."""
    return _CURRENT_PUBLISHER.get()

publishing_to

publishing_to(on_state)

Install on_state as the ambient capacity publisher for the duration of the block.

Source code in src/scale_forecasting/capacity.py

@contextmanager
def publishing_to(on_state: Callable[[CapacityLedger], None] | None) -> Iterator[None]:
    """Install ``on_state`` as the ambient capacity publisher for the duration of the block."""
    token = _CURRENT_PUBLISHER.set(on_state)
    try:
        yield
    finally:
        _CURRENT_PUBLISHER.reset(token)

walk

walk(
    candidates,
    attempt,
    *,
    ledger,
    policy,
    label=str,
    describe_failure=None,
    on_state=None,
    now=monotonic,
    sleep=sleep,
)

Try candidates until one succeeds, retrying the list until the budget is gone.

The loop the design calls for, and the whole reason this module exists::

attempt 1 .......... TRANSIENT_CAPACITY in us-central1
  hop to the next candidate; no wait (another region may have room now)
candidates exhausted, budget remains
  back off, then restart the candidate walk from the top
budget exhausted (attempts OR wall-clock)
  terminal: CapacityExhausted, ledger intact

Returns whatever attempt returns for the candidate that worked.

Raises immediately on CONFIG_FAULT — re-raising the original exception, not a wrapper, so the caller's existing handling and the operator's traceback are both unchanged. Hopping on a malformed request wastes minutes and ends in the same error.

Raises CapacityExhausted when the bounds run out, carrying the ledger.

describe_failure lets a caller enrich the message before it is classified and recorded. The Vertex path needs it: the SDK raises a generic "returned an error" while the reason lives on the failed resource, so that path reads the resource and concatenates. Defaults to str(exc).

on_state is called after every recorded attempt, with the ledger, so a caller can publish AWAITING_CAPACITY and the partial ledger during the walk rather than only at the end. It is best-effort: a publish that raises must not sink a walk that might still succeed, so it is caught and logged.

now/sleep are injected so the whole loop — including the back-off schedule and both bounds — unit-tests offline in microseconds with no real clock.

Source code in src/scale_forecasting/capacity.py

def walk(
    candidates: Sequence[C],
    attempt: Callable[[C], T],
    *,
    ledger: CapacityLedger,
    policy: CapacityPolicy,
    label: Callable[[C], str] = str,
    describe_failure: Callable[[C, Exception], str] | None = None,
    on_state: Callable[[CapacityLedger], None] | None = None,
    now: Callable[[], float] = time.monotonic,
    sleep: Callable[[float], None] = time.sleep,
) -> T:
    """Try ``candidates`` until one succeeds, retrying the list until the budget is gone.

    The loop the design calls for, and the whole reason this module exists::

        attempt 1 .......... TRANSIENT_CAPACITY in us-central1
          hop to the next candidate; no wait (another region may have room now)
        candidates exhausted, budget remains
          back off, then restart the candidate walk from the top
        budget exhausted (attempts OR wall-clock)
          terminal: CapacityExhausted, ledger intact

    Returns whatever ``attempt`` returns for the candidate that worked.

    **Raises immediately on `CONFIG_FAULT`** — re-raising the original exception, not a wrapper, so
    the caller's existing handling and the operator's traceback are both unchanged. Hopping on a
    malformed request wastes minutes and ends in the same error.

    **Raises `CapacityExhausted`** when the bounds run out, carrying the ledger.

    ``describe_failure`` lets a caller enrich the message before it is classified and recorded. The
    Vertex path needs it: the SDK raises a generic "returned an error" while the reason lives on the
    failed resource, so that path reads the resource and concatenates. Defaults to ``str(exc)``.

    ``on_state`` is called after every recorded attempt, with the ledger, so a caller can publish
    ``AWAITING_CAPACITY`` and the partial ledger *during* the walk rather than only at the end. It
    is best-effort: a publish that raises must not sink a walk that might still succeed, so it is
    caught and logged.

    ``now``/``sleep`` are injected so the whole loop — including the back-off schedule and both
    bounds — unit-tests offline in microseconds with no real clock.
    """
    if not candidates:
        raise ValueError("walk needs at least one candidate")

    budget = _Budget(policy=policy, started=now())
    last_exc: Exception | None = None
    completed_passes = 0

    while True:
        live = _live_candidates(candidates, ledger, label)
        if not live:
            # Every candidate has answered HARD_CEILING. Waiting cannot raise a quota, so there is
            # nothing left to try and no point spending the remaining budget discovering that.
            break

        for candidate in live:
            if not budget.attempts_left(now()):
                break
            budget.attempts += 1
            started = now()
            try:
                return attempt(candidate)
            # `Exception`, not `BaseException`: a KeyboardInterrupt or SystemExit during a create is
            # an operator ending the run, not the cloud running out of room. Classifying one as
            # TRANSIENT_CAPACITY would make Ctrl-C start the next region.
            except Exception as exc:  # classified below; CONFIG_FAULT re-raises
                message = (
                    describe_failure(candidate, exc) if describe_failure else str(exc)
                ) or repr(exc)
                verdict = classify(message, exc)
                ledger.record(label(candidate), verdict, message, now() - started)
                _publish(on_state, ledger)
                _log.warning(
                    "capacity: %s attempt %d on %s -> %s (%s)",
                    ledger.service,
                    budget.attempts,
                    label(candidate),
                    verdict,
                    message.splitlines()[0][:200] if message else "no message",
                )
                if verdict == CONFIG_FAULT:
                    raise
                last_exc = exc

        completed_passes += 1
        if policy.max_passes and completed_passes >= policy.max_passes:
            break
        if not budget.attempts_left(now()):
            break
        # Re-check *before* sleeping, not just at the top of the next pass. A pass in which every
        # candidate answered HARD_CEILING has nothing left to come back to, so backing off first
        # would spend real minutes on the way to a conclusion already reached.
        if not _live_candidates(candidates, ledger, label):
            break
        delay = policy.backoff_for(completed_passes)
        if not budget.sleep_fits(now(), delay):
            break
        if delay:
            _log.warning(
                "capacity: %s exhausted %d candidate(s) on pass %d; backing off %.0fs",
                ledger.service,
                len(live),
                completed_passes,
                delay,
            )
            sleep(delay)

    ledger.exhausted = True
    _publish(on_state, ledger)
    raise CapacityExhausted(
        f"{ledger.service}: no capacity after {len(ledger.attempts)} attempt(s) across "
        f"{len({a.candidate for a in ledger.attempts})} candidate(s) "
        f"({', '.join(sorted({a.candidate for a in ledger.attempts})) or 'none'}); "
        f"last error {last_exc!r}",
        ledger,
    )