symfonic.services.effects.revert

revert

The security revert itself — one linearization point, then honest accounting.

The ordering below is not stylistic. Raising the fence and revoking the leases it covers happen inside a single hold of the store lock, because that is what makes "admitted before the revert" a decidable question rather than a race: every checkpoint in :mod:admission takes the same lock, so an admission is either wholly before the fence or refused by it.

Everything after that hold is accounting, and it is deliberately pessimistic. Cancellation is requested; the drain is bounded; what does not drain is contained; what already left through an irreversible port is exposed and compensated; and an invocation that ran an opaque extension takes the fencing claim down with it. A revert that could not reach the lease store reports itself degraded rather than reporting success it cannot substantiate.

SecurityRevertCoordinator

SecurityRevertCoordinator(*, store: InMemoryLeaseStore, fences: FenceLedger, tracker: InFlightTracker, drain: BoundedDrain, quarantine: ResultQuarantine, accountant: OpaqueExposureAccountant, containment: ContainmentCoordinator, ports: FencedPortRegistry, clock: Callable[[], float] = time.time)

Closes admissions, blocks commits, contains what it cannot stop.

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

def __init__(
    self,
    *,
    store: InMemoryLeaseStore,
    fences: FenceLedger,
    tracker: InFlightTracker,
    drain: BoundedDrain,
    quarantine: ResultQuarantine,
    accountant: OpaqueExposureAccountant,
    containment: ContainmentCoordinator,
    ports: FencedPortRegistry,
    clock: Callable[[], float] = time.time,
) -> None:
    self._store = store
    self._fences = fences
    self._tracker = tracker
    self._drain = drain
    self._quarantine = quarantine
    self._books = RevertAccounting(
        accountant=accountant, containment=containment, ports=ports
    )
    self._clock = clock
    self.last_linearized_at = 0.0

refresh

refresh(incident: SecurityRevertIncident) -> SecurityRevertIncident

Re-read the suppression and quarantine ledgers for this incident.

A provider or tool result dispatched before the fence landed arrives after revert returns, so the frozen record cannot contain it — the late-result evidence would otherwise be structurally unreachable from the incident. This re-reads both ledgers, still scoped to the tickets this revert accounted for, and returns a new record with the same incident_id. Nothing else is recomputed: the fence, the drain, and the claims are findings about a moment, not a running total.

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

def refresh(self, incident: SecurityRevertIncident) -> SecurityRevertIncident:
    """Re-read the suppression and quarantine ledgers for this incident.

    A provider or tool result dispatched before the fence landed arrives
    *after* ``revert`` returns, so the frozen record cannot contain it — the
    late-result evidence would otherwise be structurally unreachable from
    the incident. This re-reads both ledgers, still scoped to the tickets
    this revert accounted for, and returns a new record with the same
    ``incident_id``. Nothing else is recomputed: the fence, the drain, and
    the claims are findings about a moment, not a running total.
    """
    suppressed, quarantined = self._ledgers(incident.affected_tickets)
    return replace(
        incident, suppressed_commits=suppressed, quarantined=quarantined
    )

revert async

revert(*, bundle_id: str, reason: str, actor: str, rejected_vector_hash: str | None = None, epoch_ceiling: int | None = None, tenant_scope_hash: str | None = None) -> SecurityRevertIncident

Raise the fence, then account for everything it could not stop.

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

async def revert(
    self,
    *,
    bundle_id: str,
    reason: str,
    actor: str,
    rejected_vector_hash: str | None = None,
    epoch_ceiling: int | None = None,
    tenant_scope_hash: str | None = None,
) -> SecurityRevertIncident:
    """Raise the fence, then account for everything it could not stop."""
    fence, revoked, covered, degraded_reason = await self._linearize(
        bundle_id=bundle_id,
        reason=reason,
        actor=actor,
        rejected_vector_hash=rejected_vector_hash,
        epoch_ceiling=epoch_ceiling,
        tenant_scope_hash=tenant_scope_hash,
    )
    linearized_at = self.last_linearized_at

    affected = self._tracker.tickets_for_leases(
        lease.lease_id for lease in covered
    )
    invocations = self._books.invocations(covered, affected)
    open_tickets = [t for t in affected if not self._state(t).terminal]
    self._tracker.request_cancellation(
        (t.ticket_id for t in open_tickets), f"security revert: {reason}"
    )

    cancelled = self._cancel_undispatched(open_tickets)
    dispatched = [t for t in affected if self._tracker.dispatched_at(t.ticket_id)]
    report = await self._drain.drain([t.ticket_id for t in dispatched])

    exposures = self._books.exposures(dispatched, invocations)
    compensations = self._books.contain_all(exposures)
    claims = self._books.claims(invocations, exposures)
    scope = tuple(t.ticket_id for t in affected)
    suppressed, quarantined = self._ledgers(scope)
    return SecurityRevertIncident(
        incident_id=f"inc-{uuid.uuid4().hex[:12]}",
        bundle_id=bundle_id,
        reason=reason,
        actor=actor,
        linearized_at=linearized_at,
        fence=fence,
        revoked_leases=revoked,
        stale_admissions=self._stale(revoked),
        cancelled=cancelled,
        affected_tickets=scope,
        suppressed_commits=suppressed,
        quarantined=quarantined,
        exposures=exposures,
        compensations=compensations,
        claims=claims,
        cutover_blocked=self._books.blocked(exposures),
        degraded=bool(degraded_reason),
        degraded_reason=degraded_reason,
        drain=report,
    )