symfonic.services.effects.tracker

tracker

In-flight effect attempts — the set a revert has to reason about.

Without this, "operations admitted before the revert linearization point" is a phrase rather than a set, and the bounded drain of EFX-F-4 would have nothing to bound. Every ticket records when it dispatched, because the only defensible statement a revert can make about an already-dispatched effect is whether it left before or after the fence landed.

InFlightTracker

InFlightTracker()

Ticket lifecycle, cancellation requests, and dispatch timestamps.

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

def __init__(self) -> None:
    self._tickets: dict[str, EffectTicket] = {}
    self._states: dict[str, TicketState] = {}
    self._dispatched_at: dict[str, float] = {}
    self._cancelled: dict[str, str] = {}

dispatched_before

dispatched_before(invocation_id: str, moment: float) -> bool

Did every dispatched effect for this invocation leave before moment?

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

def dispatched_before(self, invocation_id: str, moment: float) -> bool:
    """Did every dispatched effect for this invocation leave before ``moment``?"""
    stamps = [
        at
        for ticket_id, at in self._dispatched_at.items()
        if self._tickets[ticket_id].invocation_id == invocation_id
    ]
    return all(at <= moment for at in stamps)

mark

mark(ticket_id: str, state: TicketState) -> None

Terminal states are sticky: a contained ticket never becomes drained.

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

def mark(self, ticket_id: str, state: TicketState) -> None:
    """Terminal states are sticky: a contained ticket never becomes drained."""
    current = self._states.get(ticket_id)
    if current is not None and current.terminal:
        return
    self._states[ticket_id] = state