symfonic.services.effects.compensation

compensation

Compensation and containment for already-dispatched irreversible effects.

Cancellation does not un-send a request. Once an irreversible effect has left the process, the honest sequence is: record the exposure, run the tested compensation procedure, rotate or revoke whatever credential the effect may have disclosed, and — where no approved rule exists — block cutover for that capability rather than describe the gap as handled.

That last clause is why registration is strict. A rule that was never tested is a plan, not a remedy, and a rule nobody named an approver for is not a decision anybody owns.

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
    )