`symfonic.services.effects`¶

effects ¶

EFX — the effect-aware security-revert fence (T2.3.8).

Atomic, generation-scoped effect admission at every port T2.3.7 classified, and a security revert that is honest about its own reach: it closes new admissions for the rejected generation, blocks fenced commits, requests cancellation, quarantines late results, drains what it can within a bounded budget, and records as exposed everything it could not stop — irreversible effects that had already left, and anything an opaque in-process extension did outside the ports entirely.

The load-bearing decision is that the fencing claim is derived rather than asserted (see :mod:exposure and :mod:incident). A revert cannot be configured to report success it did not achieve.

BoundedDrain ¶

BoundedDrain(tracker: InFlightTracker, *, clock: Callable[[], float] = time.time, sleep: Callable[[float], Awaitable[None]] = _default_sleep, interval: float = 0.05, budget: float = 5.0)

Waits, briefly, for in-flight work to reach a terminal state.

Source code in src/symfonic/services/effects/drain.py

def __init__(
    self,
    tracker: InFlightTracker,
    *,
    clock: Callable[[], float] = time.time,
    sleep: Callable[[float], Awaitable[None]] = _default_sleep,
    interval: float = 0.05,
    budget: float = 5.0,
) -> None:
    self._tracker = tracker
    self._clock = clock
    self._sleep = sleep
    self._interval = interval
    self._budget = budget

CommitDecision ¶

Bases: StrEnum

EFX-F-3 — what the commit-time conditional check decided.

SUPPRESSED deliberately does not say "cancelled" or "undone". The write did not land; whether an external effect already left the process is a separate question the exposure ledger answers.

CompensationOutcome `dataclass` ¶

CompensationOutcome(port_id: str, invocation_id: str, invoked: bool, credentials_rotated: bool, note: str, rule_id: str | None = None)

What containment actually did. invoked is never assumed true.

CompensationRegistry ¶

CompensationRegistry(ports: FencedPortRegistry | None = None)

Which irreversible ports have an approved remedy, and which block cutover.

Source code in src/symfonic/services/effects/compensation.py

def __init__(self, ports: FencedPortRegistry | None = None) -> None:
    self._rules: dict[str, CompensationRule] = {}
    self._ports = ports if ports is not None else FencedPortRegistry()

cutover_blocked_ports ¶

cutover_blocked_ports(port_ids: Iterable[str]) -> tuple[str, ...]

Irreversible ports among these that carry no approved rule.

Source code in src/symfonic/services/effects/compensation.py

def cutover_blocked_ports(self, port_ids: Iterable[str]) -> tuple[str, ...]:
    """Irreversible ports among these that carry no approved rule."""
    blocked = {
        port_id
        for port_id in port_ids
        if self._ports.knows(port_id)
        and self._ports.port(port_id).irreversible
        and port_id not in self._rules
    }
    return tuple(sorted(blocked))

CompensationRule `dataclass` ¶

CompensationRule(rule_id: str, port_id: str, procedure: str, tested: bool, approved_by: str, rotates_credentials: bool = False)

One documented, tested procedure for undoing what can be undone.

ContainmentCoordinator ¶

ContainmentCoordinator(registry: CompensationRegistry, *, rotator: Callable[[str], None] | None = None)

Runs the approved remedy, or records precisely why it could not.

Source code in src/symfonic/services/effects/compensation.py

def __init__(
    self,
    registry: CompensationRegistry,
    *,
    rotator: Callable[[str], None] | None = None,
) -> None:
    self._registry = registry
    self._rotator = rotator

registry `property` ¶

registry: CompensationRegistry

Exposed so an incident can ask which ports are still blocking cutover.

contain_all ¶

contain_all(exposures: Iterable[ExposureRecord]) -> tuple[CompensationOutcome, ...]

Only irreversible-effect exposures have a port-keyed remedy.

Source code in src/symfonic/services/effects/compensation.py

def contain_all(
    self, exposures: Iterable[ExposureRecord]
) -> tuple[CompensationOutcome, ...]:
    """Only irreversible-effect exposures have a port-keyed remedy."""
    return tuple(
        self.contain(exposure)
        for exposure in exposures
        if exposure.kind is ExposureKind.IRREVERSIBLE_EFFECT
    )

DrainReport `dataclass` ¶

DrainReport(drained: tuple[str, ...], contained: tuple[str, ...], budget: float, elapsed: float)

What the drain observed. complete is false if anything was contained.