symfonic.kernel.deadline

deadline

The run's G8 budget, owned by one clock (CXL-6, BP-6).

Its own module because a deadline answers a different question from a teardown stack. :class:~symfonic.kernel.lifecycle.RunLifecycle decides what a run owns and in what order it is released; this class knows only how much time is left, and how to spend a bounded amount of it without the two situations CXL-6 separates — "the backend hung" and "the caller stopped caring" — collapsing into one exception on the way out.

Two ways to spend the budget are offered, and they are not interchangeable:

• :meth:scope wraps a whole call in one timeout. It is legal only where one task both enters and leaves the scope — the blocking entry point.
• :meth:bounded wraps one step of forward progress. It is what a streaming entry point uses, because an async generator may be resumed from a different task than the one that suspended it, and a scope spanning that hop cancels a task that never entered it.

Both convert only their own expiry. A TimeoutError the body raised on its own travels out untouched, which is the same rule pointed the other way.

RunDeadline

RunDeadline(run_id: str, seconds: float | None = None)

One run's remaining time, and the only place it becomes an error.

Source code in src/symfonic/kernel/deadline.py

def __init__(self, run_id: str, seconds: float | None = None) -> None:
    self.run_id = run_id
    self.seconds = seconds
    self._at = time.monotonic() + seconds if seconds is not None else None
    self._error: ServiceTimeoutError | None = None

remaining property

remaining: float | None

Seconds left, or None when this run declared no deadline.

Measured from the moment the run began, never from the moment a step began: a budget that restarted on every await would bound one provider call rather than one run.

bounded async

bounded(step: Awaitable[Any], *, doing: str) -> Any

Await one step of forward progress inside the remaining budget.

Source code in src/symfonic/kernel/deadline.py

async def bounded(self, step: Awaitable[Any], *, doing: str) -> Any:
    """Await one step of forward progress inside the remaining budget."""
    remaining = self.remaining
    if remaining is None:
        return await step
    if remaining <= 0:
        _discard(step)
        raise self.expired(doing)
    budget = asyncio.timeout(remaining)
    try:
        async with budget:
            return await step
    except TimeoutError as exc:
        if not budget.expired():
            raise  # the step's own timeout, not this run's
        raise self.expired(doing) from exc

expired

expired(doing: str) -> ServiceTimeoutError

Mint — and remember — the error that says this run ran out of time.

Source code in src/symfonic/kernel/deadline.py

def expired(self, doing: str) -> ServiceTimeoutError:
    """Mint — and remember — the error that says this run ran out of time."""
    error = ServiceTimeoutError(
        f"run {self.run_id} exceeded its {self.seconds}s deadline "
        f"while {doing} (CXL-6)."
    )
    if self._error is None:
        self._error = error
    return error

owns

owns(error: BaseException) -> bool

True only for the timeout this deadline raised.

Identity, not type: ServiceTimeoutError is also what an adopter's tool and every inner bounded operation raises, and reporting one of those as the run's deadline sends the reader to tune a budget nothing touched.

Source code in src/symfonic/kernel/deadline.py

def owns(self, error: BaseException) -> bool:
    """True only for the timeout *this* deadline raised.

    Identity, not type: ``ServiceTimeoutError`` is also what an adopter's
    tool and every inner bounded operation raises, and reporting one of
    those as the run's deadline sends the reader to tune a budget nothing
    touched.
    """
    return self._error is not None and error is self._error

scope async

scope() -> AsyncIterator[None]

Bound a whole call, translating expiry into an error not a cancel.

Source code in src/symfonic/kernel/deadline.py

@asynccontextmanager
async def scope(self) -> AsyncIterator[None]:
    """Bound a whole call, translating expiry into an error not a cancel."""
    if self.seconds is None:
        yield
        return
    budget = asyncio.timeout(self.seconds)
    try:
        async with budget:
            yield
    except TimeoutError as exc:
        if not budget.expired():
            raise
        raise self.expired("awaiting its entry point") from exc