Skip to content

symfonic.services.shadow.replay

replay

Replay — re-run a recorded invocation against stubbed external effects.

A replay is a shadow run whose stub responder answers from a recording instead of from a hash. That is the whole design: replay reuses the harness, so it inherits deny-or-stub, the fail-closed unclassified-port rule, and the opaque extension abort, and there is no second code path where those could drift.

Two refusals are specific to replay. A recording that names an opaque extension is non-replayable and aborts before the body runs. A recorded answer that is missing for a request the replay makes raises through the gateway rather than falling back to the real port — a "helpful" fallback is exactly how a replay would duplicate an externally visible effect (TM-29d).

RecordedResponder 

RecordedResponder(recording: Recording)

Answers a stubbed port call from the recording, in recorded order.

Answers for one key are consumed FIFO rather than looked up, because the recorded ports are non-idempotent by definition: the same request made twice may have produced two different answers, and replaying the second answer to the first call is a divergence the comparator cannot catch. Once a key's recorded answers are exhausted the responder refuses, so a replay that calls a port more times than the original is a refusal — never a silently repeated answer.

Source code in src/symfonic/services/shadow/replay.py 
46
47
48
49
def __init__(self, recording: Recording) -> None:
    self._queues = recording.response_queues()
    self._cursor: dict[tuple[str, str, str], int] = {}
    self._served: list[tuple[str, str, str]] = []

unserved 

unserved() -> tuple[tuple[str, str, str], ...]

Every recorded answer the replay never asked for, repeats included.

Source code in src/symfonic/services/shadow/replay.py 
55
56
57
58
59
60
61
def unserved(self) -> tuple[tuple[str, str, str], ...]:
    """Every recorded answer the replay never asked for, repeats included."""
    return tuple(
        key
        for key, answers in self._queues.items()
        for _ in range(len(answers) - self._cursor.get(key, 0))
    )

ReplayRunner 

ReplayRunner(harness: ShadowHarness)

Replays one recording through the shadow harness.

Source code in src/symfonic/services/shadow/replay.py 
80
81
def __init__(self, harness: ShadowHarness) -> None:
    self._harness = harness

replayable staticmethod 

replayable(result: ShadowRunResult) -> bool

A replay is usable as comparison input only when nothing leaked.

Source code in src/symfonic/services/shadow/replay.py 
105
106
107
108
@staticmethod
def replayable(result: ShadowRunResult) -> bool:
    """A replay is usable as comparison input only when nothing leaked."""
    return result.status is ShadowStatus.COMPLETED and result.suppression_claim