Open navigation menu

Compose Timeouts With A Shared Deadline

A deadline helper that derives per-step timeouts from one shared budget so a slow first step cannot make later steps exceed the total.

PythonResilience

Code

import time
from typing import Callable


class DeadlineExceededError(RuntimeError):
    """Raised when the shared budget no longer covers the requested step."""


class Deadline:
    """A single budget shared by every step of an operation."""

    def __init__(self, budget_s: float, clock: Callable[[], float] = time.monotonic) -> None:
        if budget_s <= 0:
            raise ValueError("budget_s must be positive")
        self._budget_s = budget_s
        self._clock = clock
        self._start = clock()

    def remaining(self) -> float:
        return max(0.0, self._budget_s - (self._clock() - self._start))

    def expired(self) -> bool:
        return self.remaining() <= 0.0

    def require(self, needed_s: float) -> float:
        """Return the time granted for a step, or raise if it cannot fit."""
        if needed_s <= 0:
            raise ValueError("needed_s must be positive")
        left = self.remaining()
        if left < needed_s:
            raise DeadlineExceededError(
                "budget exhausted: %.3fs left, %.3fs needed" % (left, needed_s)
            )
        return needed_s

Related Examples