Replica registry + health-check loop.

The registry tracks the set of known backend replicas and their live health. The router consults it for "pick a backend for this session" decisions and a background task updates it from periodic HTTP probes.

State machine per replica::

HEALTHY ──(N consecutive failures)──▶ UNHEALTHY
   ▲                                     │
   └──────(M consecutive successes)──────┘

Where N = :attr:RouterConfig.failure_threshold and M = :attr:RouterConfig.recovery_threshold.

`fastvideo.entrypoints.streaming.router.registry.ReplicaEndpoint`

ReplicaEndpoint(url: str, name: str | None = None, primary: bool = False, weight: float = 1.0)

One backend replica the router can route to.

`fastvideo.entrypoints.streaming.router.registry.ReplicaEndpoint.url`

url: str

HTTP base URL, e.g. http://host:8000. WebSocket URL is derived automatically by replacing the scheme.

`fastvideo.entrypoints.streaming.router.registry.ReplicaEndpoint.primary`

primary: bool = False

True = prefer this replica over others in steady state.

`fastvideo.entrypoints.streaming.router.registry.RouterConfig`

RouterConfig(host: str = '0.0.0.0', port: int = 9000, replicas: list[ReplicaEndpoint] = list(), health_check_path: str = '/health', health_check_interval_seconds: float = 5.0, health_check_timeout_seconds: float = 2.0, failure_threshold: int = 3, recovery_threshold: int = 2)

Typed router config loaded from a YAML file.

Example::

router:
  host: 0.0.0.0
  port: 9000
  replicas:
    - url: http://streamer-a:8000
      primary: true
    - url: http://streamer-b:8000
  health_check:
    path: /health
    interval_seconds: 5
    failure_threshold: 3

Validation runs in __post_init__: empty replicas, non-positive intervals/timeouts, thresholds < 1, non-http(s) URLs, and more than one primary all raise ValueError so misconfigurations surface at load time rather than as confusing runtime failures.

`fastvideo.entrypoints.streaming.router.registry.init_logger`

init_logger(name: str) -> _FastvideoLogger

The main purpose of this function is to ensure that loggers are retrieved in such a way that we can be sure the root fastvideo logger has already been configured.

`fastvideo.entrypoints.streaming.router.registry.HttpProbe`

HttpProbe = Any

Structural alias for health-probe callables. Concrete signature is async def __call__(url: str, *, timeout: float) -> tuple[float, str | None]; typing.Callable cannot express keyword-only parameters, so duck-typing is the pragmatic compromise.

Source

HttpProbe = Any

`fastvideo.entrypoints.streaming.router.registry.ReplicaStatus`

ReplicaStatus()

Source

class ReplicaStatus(enum.Enum):
    UNKNOWN = "unknown"
    HEALTHY = "healthy"
    UNHEALTHY = "unhealthy"

`fastvideo.entrypoints.streaming.router.registry.ReplicaHealth`

ReplicaHealth(status: ReplicaStatus = ReplicaStatus.UNKNOWN, last_ok_at: float | None = None, last_failure_at: float | None = None, consecutive_failures: int = 0, consecutive_successes: int = 0, last_latency_ms: float | None = None)

Source

@dataclass
class ReplicaHealth:
    status: ReplicaStatus = ReplicaStatus.UNKNOWN
    last_ok_at: float | None = None
    last_failure_at: float | None = None
    consecutive_failures: int = 0
    consecutive_successes: int = 0
    last_latency_ms: float | None = None

`fastvideo.entrypoints.streaming.router.registry.Replica`

Replica(endpoint: ReplicaEndpoint, health: ReplicaHealth = ReplicaHealth())

Source

@dataclass
class Replica:
    endpoint: ReplicaEndpoint
    health: ReplicaHealth = field(default_factory=ReplicaHealth)

    @property
    def url(self) -> str:
        return self.endpoint.url

    @property
    def primary(self) -> bool:
        return self.endpoint.primary

    @property
    def is_healthy(self) -> bool:
        return self.health.status is ReplicaStatus.HEALTHY

`fastvideo.entrypoints.streaming.router.registry.ReplicaRegistry`

ReplicaRegistry(replicas: list[ReplicaEndpoint])

Stateful map of replica URL → :class:Replica.

Selection favors primary replicas when healthy; otherwise the first healthy non-primary is returned. When none are healthy, the registry returns None so the router can reject incoming sessions with gpu_unavailable.

Source

class ReplicaRegistry:
    """Stateful map of replica URL → :class:`Replica`.

    Selection favors primary replicas when healthy; otherwise the first
    healthy non-primary is returned. When none are healthy, the
    registry returns ``None`` so the router can reject incoming
    sessions with ``gpu_unavailable``.
    """

    def __init__(self, replicas: list[ReplicaEndpoint]) -> None:
        if not replicas:
            raise ValueError("ReplicaRegistry requires at least one replica")
        self._replicas: dict[str, Replica] = {endpoint.url: Replica(endpoint=endpoint) for endpoint in replicas}
        self._lock = asyncio.Lock()

    def all(self) -> list[Replica]:
        return list(self._replicas.values())

    def get(self, url: str) -> Replica | None:
        return self._replicas.get(url)

    def primaries(self) -> list[Replica]:
        return [r for r in self._replicas.values() if r.primary]

    def select(self) -> Replica | None:
        """Pick the best healthy replica.

        Priority order:

        1. The first healthy primary (insertion order).
        2. The first healthy non-primary (insertion order).
        3. ``None`` when nothing is healthy.

        This MVP picks the first match within each tier; it does NOT
        load-balance across multiple healthy replicas of the same tier.
        Round-robin and weighted distribution are deferred until a real
        N-way active deployment exists.
        """
        healthy_primaries = [r for r in self._replicas.values() if r.primary and r.is_healthy]
        if healthy_primaries:
            return healthy_primaries[0]
        healthy = [r for r in self._replicas.values() if r.is_healthy]
        if healthy:
            return healthy[0]
        return None

    async def record_success(
        self,
        replica: Replica,
        *,
        recovery_threshold: int,
        latency_ms: float,
    ) -> None:
        async with self._lock:
            h = replica.health
            h.last_ok_at = time.time()
            h.last_latency_ms = latency_ms
            h.consecutive_failures = 0
            h.consecutive_successes += 1
            # State machine: UNKNOWN -> HEALTHY is immediate; only the
            # UNHEALTHY -> HEALTHY transition is gated by recovery_threshold.
            if h.status is ReplicaStatus.UNKNOWN:
                logger.info("router: replica %s initial probe ok, marking HEALTHY", replica.url)
                h.status = ReplicaStatus.HEALTHY
                h.consecutive_successes = 0
            elif (h.status is ReplicaStatus.UNHEALTHY and h.consecutive_successes >= recovery_threshold):
                logger.info("router: replica %s recovered to HEALTHY after %d successes", replica.url,
                            h.consecutive_successes)
                h.status = ReplicaStatus.HEALTHY
                h.consecutive_successes = 0

    async def record_failure(
        self,
        replica: Replica,
        *,
        failure_threshold: int,
        reason: str,
    ) -> None:
        async with self._lock:
            h = replica.health
            h.last_failure_at = time.time()
            h.consecutive_successes = 0
            h.consecutive_failures += 1
            if (h.status is not ReplicaStatus.UNHEALTHY and h.consecutive_failures >= failure_threshold):
                logger.warning("router: replica %s marked UNHEALTHY after %d failures: %s", replica.url,
                               h.consecutive_failures, reason)
                h.status = ReplicaStatus.UNHEALTHY

`fastvideo.entrypoints.streaming.router.registry.ReplicaRegistry.all`

all() -> list[Replica]

Source

def all(self) -> list[Replica]:
    return list(self._replicas.values())

`fastvideo.entrypoints.streaming.router.registry.ReplicaRegistry.get`

get(url: str) -> Replica | None

Source

def get(self, url: str) -> Replica | None:
    return self._replicas.get(url)

`fastvideo.entrypoints.streaming.router.registry.ReplicaRegistry.primaries`

primaries() -> list[Replica]

Source

def primaries(self) -> list[Replica]:
    return [r for r in self._replicas.values() if r.primary]

`fastvideo.entrypoints.streaming.router.registry.ReplicaRegistry.select`

select() -> Replica | None

Pick the best healthy replica.

Priority order:

The first healthy primary (insertion order).
The first healthy non-primary (insertion order).
None when nothing is healthy.

This MVP picks the first match within each tier; it does NOT load-balance across multiple healthy replicas of the same tier. Round-robin and weighted distribution are deferred until a real N-way active deployment exists.

Source

def select(self) -> Replica | None:
    """Pick the best healthy replica.

    Priority order:

    1. The first healthy primary (insertion order).
    2. The first healthy non-primary (insertion order).
    3. ``None`` when nothing is healthy.

    This MVP picks the first match within each tier; it does NOT
    load-balance across multiple healthy replicas of the same tier.
    Round-robin and weighted distribution are deferred until a real
    N-way active deployment exists.
    """
    healthy_primaries = [r for r in self._replicas.values() if r.primary and r.is_healthy]
    if healthy_primaries:
        return healthy_primaries[0]
    healthy = [r for r in self._replicas.values() if r.is_healthy]
    if healthy:
        return healthy[0]
    return None

`fastvideo.entrypoints.streaming.router.registry.ReplicaRegistry.record_success`

record_success(replica: Replica, *, recovery_threshold: int, latency_ms: float) -> None

Source

async def record_success(
    self,
    replica: Replica,
    *,
    recovery_threshold: int,
    latency_ms: float,
) -> None:
    async with self._lock:
        h = replica.health
        h.last_ok_at = time.time()
        h.last_latency_ms = latency_ms
        h.consecutive_failures = 0
        h.consecutive_successes += 1
        # State machine: UNKNOWN -> HEALTHY is immediate; only the
        # UNHEALTHY -> HEALTHY transition is gated by recovery_threshold.
        if h.status is ReplicaStatus.UNKNOWN:
            logger.info("router: replica %s initial probe ok, marking HEALTHY", replica.url)
            h.status = ReplicaStatus.HEALTHY
            h.consecutive_successes = 0
        elif (h.status is ReplicaStatus.UNHEALTHY and h.consecutive_successes >= recovery_threshold):
            logger.info("router: replica %s recovered to HEALTHY after %d successes", replica.url,
                        h.consecutive_successes)
            h.status = ReplicaStatus.HEALTHY
            h.consecutive_successes = 0

`fastvideo.entrypoints.streaming.router.registry.ReplicaRegistry.record_failure`

record_failure(replica: Replica, *, failure_threshold: int, reason: str) -> None

Source

async def record_failure(
    self,
    replica: Replica,
    *,
    failure_threshold: int,
    reason: str,
) -> None:
    async with self._lock:
        h = replica.health
        h.last_failure_at = time.time()
        h.consecutive_successes = 0
        h.consecutive_failures += 1
        if (h.status is not ReplicaStatus.UNHEALTHY and h.consecutive_failures >= failure_threshold):
            logger.warning("router: replica %s marked UNHEALTHY after %d failures: %s", replica.url,
                           h.consecutive_failures, reason)
            h.status = ReplicaStatus.UNHEALTHY

`fastvideo.entrypoints.streaming.router.registry.run_health_check_loop`

run_health_check_loop(registry: ReplicaRegistry, config: RouterConfig, *, stop_event: asyncio.Event, http_get: HttpProbe | None = None) -> None

Poll all replicas' health endpoints in parallel on a fixed interval.

http_get is pluggable so unit tests can inject a deterministic probe without hitting the network. The default builds a single httpx.AsyncClient shared across the loop's lifetime so the common case (steady polling against a stable replica set) reuses TCP/TLS connections instead of paying handshake cost per probe.

Probes within one polling cycle run concurrently via asyncio.gather so a slow replica doesn't push the cycle past health_check_interval_seconds.

Source

async def run_health_check_loop(
    registry: ReplicaRegistry,
    config: RouterConfig,
    *,
    stop_event: asyncio.Event,
    http_get: HttpProbe | None = None,
) -> None:
    """Poll all replicas' health endpoints in parallel on a fixed interval.

    ``http_get`` is pluggable so unit tests can inject a deterministic
    probe without hitting the network. The default builds a single
    ``httpx.AsyncClient`` shared across the loop's lifetime so the
    common case (steady polling against a stable replica set) reuses
    TCP/TLS connections instead of paying handshake cost per probe.

    Probes within one polling cycle run concurrently via ``asyncio.gather``
    so a slow replica doesn't push the cycle past
    ``health_check_interval_seconds``.
    """
    if http_get is not None:
        await _run_loop(registry, config, stop_event, http_get)
        return
    async with _build_default_probe(config) as probe:
        await _run_loop(registry, config, stop_event, probe)