symfonic.services.privacy.saga_store

saga_store

The reference ErasureSagaStore: per-participant completion, durably.

start resumes rather than restarts, and that is the whole design. A handler-loop erasure that dies at store four of nine leaves nothing but an "initiated" audit row; this store leaves a row per participant saying exactly which four went, how many attempts each took, and what the last error said.

InMemoryErasureSagaStore

InMemoryErasureSagaStore(state: MutableMapping[str, SagaState] | None = None)

Reference implementation (CON-S-4); state is injectable for durability.

The lock is a thread lock, not a per-loop asyncio.Lock: an operator retry and a scheduled resume can land in different loops (or threads), and :meth:claim_generation_advance is only a compare-and-set if the compare and the set are indivisible for every caller, not just for co-scheduled ones. Nothing awaits inside a section, so a plain lock cannot deadlock.

Source code in src/symfonic/services/privacy/saga_store.py

def __init__(self, state: MutableMapping[str, SagaState] | None = None) -> None:
    self._state: MutableMapping[str, SagaState] = state if state is not None else {}
    self._lock = threading.Lock()

claim_generation_advance async

claim_generation_advance(scope_key: str) -> bool

Compare-and-set the completion transition; True for the winner only.

This is the only way to make the transition — there is deliberately no plain "mark it advanced" setter to reach for. The saga cannot do it as "read the flag, then advance": an operator retry racing a scheduled resume — a race PrivacyService.resume explicitly invites — would have both runs observe False across the await and both call complete_erasure, advancing the generation twice. The decision has to be one indivisible step in the store, which is the only place both runs meet.

Source code in src/symfonic/services/privacy/saga_store.py

async def claim_generation_advance(self, scope_key: str) -> bool:
    """Compare-and-set the completion transition; True for the winner only.

    This is the *only* way to make the transition — there is deliberately no
    plain "mark it advanced" setter to reach for. The saga cannot do it as
    "read the flag, then advance": an operator retry racing a scheduled
    ``resume`` — a race ``PrivacyService.resume`` explicitly invites — would
    have both runs observe ``False`` across the await and both call
    ``complete_erasure``, advancing the generation twice. The decision has to
    be one indivisible step *in the store*, which is the only place both runs
    meet.
    """
    with self._lock:
        state = self._require(scope_key)
        if state.generation_advanced:
            return False
        self._state[scope_key] = replace(state, generation_advanced=True)
        return True

incomplete_under async

incomplete_under(scope_key: str) -> SagaState | None

The descendant half of read suppression (see the port's docstring).

A scan, because the reference store is a mapping: a durable backend indexes the tenant segment and answers the same question with a query. Keys that are not scope keys are compared exactly rather than skipped — an unparseable key still names a saga somebody started, and dropping it would fail open for whoever recorded it.

Source code in src/symfonic/services/privacy/saga_store.py

async def incomplete_under(self, scope_key: str) -> SagaState | None:
    """The descendant half of read suppression (see the port's docstring).

    A scan, because the reference store is a mapping: a durable backend
    indexes the tenant segment and answers the same question with a query.
    Keys that are not scope keys are compared exactly rather than skipped —
    an unparseable key still names a saga somebody started, and dropping it
    would fail *open* for whoever recorded it.
    """
    try:
        reader = SubjectScope.from_key(scope_key)
    except SubjectScopeError:
        state = await self.read(scope_key)
        return state if state is not None and not state.complete else None
    with self._lock:
        for key in sorted(self._state):
            state = self._state[key]
            if state.complete:
                continue
            try:
                scope = SubjectScope.from_key(key)
            except SubjectScopeError:
                continue
            if scope.narrows(reader):
                return state
    return None

mark_exhausted async

mark_exhausted(scope_key: str, participant_id: str, *, error: str) -> SagaState

Bounded retry ran out. Recorded, never raised.

Raising here would abort the sweep and skip every participant sorted after the broken one — the exact failure mode the shipped legacy sweep already learned to avoid.

Source code in src/symfonic/services/privacy/saga_store.py

async def mark_exhausted(
    self, scope_key: str, participant_id: str, *, error: str
) -> SagaState:
    """Bounded retry ran out. Recorded, never raised.

    Raising here would abort the sweep and skip every participant sorted
    after the broken one — the exact failure mode the shipped legacy sweep
    already learned to avoid.
    """
    return await self._update(
        scope_key,
        participant_id,
        lambda p: replace(p, exhausted=True, last_error=error or p.last_error),
    )

start async

start(scope_key: str, *, participants: Sequence[str]) -> SagaState

Create, or resume and widen.

Widen, because PRIV-1's registry is live: a store registered while the saga was running still holds the subject's data, and a saga that captured its participant list at t=0 would leave it holding it.

Source code in src/symfonic/services/privacy/saga_store.py

async def start(self, scope_key: str, *, participants: Sequence[str]) -> SagaState:
    """Create, or resume and widen.

    Widen, because PRIV-1's registry is live: a store registered while the
    saga was running still holds the subject's data, and a saga that
    captured its participant list at t=0 would leave it holding it.
    """
    with self._lock:
        state = self._state.get(scope_key)
        if state is None:
            state = SagaState(
                scope_key=scope_key,
                started_at=time.time(),
                progress={
                    pid: ParticipantProgress(participant_id=pid) for pid in participants
                },
            )
        else:
            progress = dict(state.progress)
            for pid in participants:
                if pid not in progress:
                    progress[pid] = ParticipantProgress(participant_id=pid)
            state = replace(state, progress=progress)
        self._state[scope_key] = state
        return state