Skip to content
FastVideo
Esc
↑↓navigate↵open⌘Jpreview
On this page

registry

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) -&gt; str:
    return self.endpoint.url

@property
def primary(self) -&gt; bool:
    return self.endpoint.primary

@property
def is_healthy(self) -&gt; bool:
    return self.health.status is ReplicaStatus.HEALTHY</code></pre></details>


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``.
&quot;&quot;&quot;

def __init__(self, replicas: list&#91;ReplicaEndpoint&#93;) -&gt; None:
    if not replicas:
        raise ValueError(&quot;ReplicaRegistry requires at least one replica&quot;)
    self._replicas: dict&#91;str, Replica&#93; = &#123;endpoint.url: Replica(endpoint=endpoint) for endpoint in replicas&#125;
    self._lock = asyncio.Lock()

def all(self) -&gt; list&#91;Replica&#93;:
    return list(self._replicas.values())

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

def primaries(self) -&gt; list&#91;Replica&#93;:
    return &#91;r for r in self._replicas.values() if r.primary&#93;

def select(self) -&gt; Replica | None:
    &quot;&quot;&quot;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.
    &quot;&quot;&quot;
    healthy_primaries = &#91;r for r in self._replicas.values() if r.primary and r.is_healthy&#93;
    if healthy_primaries:
        return healthy_primaries&#91;0&#93;
    healthy = &#91;r for r in self._replicas.values() if r.is_healthy&#93;
    if healthy:
        return healthy&#91;0&#93;
    return None

async def record_success(
    self,
    replica: Replica,
    *,
    recovery_threshold: int,
    latency_ms: float,
) -&gt; 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 -&gt; HEALTHY is immediate; only the
        # UNHEALTHY -&gt; HEALTHY transition is gated by recovery_threshold.
        if h.status is ReplicaStatus.UNKNOWN:
            logger.info(&quot;router: replica %s initial probe ok, marking HEALTHY&quot;, replica.url)
            h.status = ReplicaStatus.HEALTHY
            h.consecutive_successes = 0
        elif (h.status is ReplicaStatus.UNHEALTHY and h.consecutive_successes &gt;= recovery_threshold):
            logger.info(&quot;router: replica %s recovered to HEALTHY after %d successes&quot;, 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,
) -&gt; 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 &gt;= failure_threshold):
            logger.warning(&quot;router: replica %s marked UNHEALTHY after %d failures: %s&quot;, replica.url,
                           h.consecutive_failures, reason)
            h.status = ReplicaStatus.UNHEALTHY</code></pre></details>


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:


    

  1. The first healthy primary (insertion order).
    2. 

  2. The first healthy non-primary (insertion order).
    3. 

  3. None when nothing is healthy.
    4. 



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.
&quot;&quot;&quot;
healthy_primaries = &#91;r for r in self._replicas.values() if r.primary and r.is_healthy&#93;
if healthy_primaries:
    return healthy_primaries&#91;0&#93;
healthy = &#91;r for r in self._replicas.values() if r.is_healthy&#93;
if healthy:
    return healthy&#91;0&#93;
return None</code></pre></details>


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&#x27;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&#x27;t push the cycle past
``health_check_interval_seconds``.
&quot;&quot;&quot;
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)</code></pre></details>

Was this page helpful?