Skip to content

symfonic.services.switching.registry

registry

CUT-AIR — the generation-indexed active-invocation registry.

The registry exists to answer one question that nothing else can answer honestly: is any live invocation still running under the old generation? Log absence does not answer it. Traffic graphs do not answer it. An open record with an admitted epoch does.

ActiveInvocationRegistry 

ActiveInvocationRegistry(*, clock: Callable[[], float] = time.time)

Append-only from the data plane; indexed by bundle and by generation.

Source code in src/symfonic/services/switching/registry.py 
81
82
83
84
85
86
def __init__(self, *, clock: Callable[[], float] = time.time) -> None:
    self._rows: dict[str, InvocationRecord] = {}
    self._by_bundle: dict[str, list[str]] = {}
    self._by_generation: dict[str, list[str]] = {}
    self._clock = clock
    self._lock = asyncio.Lock()

admission_lock property 

admission_lock: Lock

Held across bind-and-register so the pair is one atomic admission.

expire_lease 

expire_lease(invocation_id: str, *, reason: str = 'lease expired') -> None

EFX-L-5 — a crashed worker's record is closed by lease expiry.

Source code in src/symfonic/services/switching/registry.py 
111
112
113
def expire_lease(self, invocation_id: str, *, reason: str = "lease expired") -> None:
    """EFX-L-5 — a crashed worker's record is closed by lease expiry."""
    self._close(invocation_id, reason)

invalidate 

invalidate(invocation_id: str, *, actor_role: str) -> None

CUT-AIR-4 — only the control plane, and only via SCP-REV-4.

Source code in src/symfonic/services/switching/registry.py 
123
124
125
126
127
128
129
130
131
132
def invalidate(self, invocation_id: str, *, actor_role: str) -> None:
    """CUT-AIR-4 — only the control plane, and only via SCP-REV-4."""
    if actor_role != CONTROL_PLANE_ROLE:
        raise PermissionError(
            f"role {actor_role!r} may not invalidate registry rows; registry "
            "writes are append-only from the data plane and only the control "
            "plane marks rows invalidated."
        )
    row = self._rows[invocation_id]
    self._rows[invocation_id] = replace(row, invalidated_by=CONTROL_PLANE_ROLE)

quiescent_below 

quiescent_below(bundle_id: str, epoch: int) -> bool

CUT-AIR-3 — no open record on this bundle below epoch.

Source code in src/symfonic/services/switching/registry.py 
153
154
155
156
157
def quiescent_below(self, bundle_id: str, epoch: int) -> bool:
    """CUT-AIR-3 — no open record on this bundle below ``epoch``."""
    return not any(
        row.admitted_epoch < epoch for row in self.open_records(bundle_id)
    )

InvocationRecord dataclass 

InvocationRecord(invocation_id: str, tenant_scope_hash: str, bundle_id: str, admitted_epoch: int, generation_vector_hash: str, admitted_at: float, stale_binding: bool = False, completed_at: float | None = None, closed_reason: str | None = None, invalidated_by: str | None = None)

One admitted invocation. Carries a scope hash, never tenant content.

QuiescenceBarrier 

QuiescenceBarrier(workers: Sequence[str] = ())

CUT-BR-6 — the worker-acknowledged half of the drain proof.

A freeze is only propagated when every worker says so. Waiting out T_stale + T_outage and hoping is the alternative this replaces: it cannot distinguish "every worker saw the freeze" from "every worker is wedged", and those two states need opposite responses.

Source code in src/symfonic/services/switching/registry.py 
53
54
55
def __init__(self, workers: Sequence[str] = ()) -> None:
    self._workers = frozenset(workers)
    self._acks: dict[str, set[str]] = {}