Skip to content

symfonic.services.effects.ports

ports

EFX-F-2 — the fenced-port table, derived from T2.3.7's classification.

T2.3.7 answered "what does shadow mode do with this port?". The fence needs a second, independent verdict for every one of those rows: if this effect already left, can the framework undo it? The two questions have different answers — cache.write is denied in shadow and compensable in production — so the verdict is its own table rather than an inference from the disposition.

Coverage is exhaustive by check. FencedPortRegistry refuses to construct if any classified port has no verdict, so a port added to T2.3.7's table tomorrow breaks the build until somebody decides whether it can be undone.

FencedPortRegistry 

FencedPortRegistry(classification: EffectPortClassification = DEFAULT_EFFECT_CLASSIFICATION)

Every classified effect port, with its reversibility verdict.

Source code in src/symfonic/services/effects/ports.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def __init__(
    self, classification: EffectPortClassification = DEFAULT_EFFECT_CLASSIFICATION
) -> None:
    rows: dict[str, FencedPort] = {}
    missing: list[str] = []
    for port_id, port in classification.by_id.items():
        verdict = _verdict(port_id)
        if verdict is None:
            missing.append(port_id)
            continue
        reversibility, rationale = verdict
        rows[port_id] = FencedPort(
            port_id=port_id,
            family=port.family,
            reversibility=reversibility,
            externally_visible=port.externally_visible,
            rationale=rationale,
        )
    if missing:
        raise IncompleteFenceCoverageError(
            "these classified effect ports carry no reversibility verdict, so "
            "the fence cannot say whether a revert may claim to have stopped "
            f"them: {sorted(missing)}. Add a row to ports.py — an unclassified "
            "port is a hole in the evidence, not a permission."
        )
    self._by_id = MappingProxyType(rows)

assert_covers_classification 

assert_covers_classification(classification: EffectPortClassification = DEFAULT_EFFECT_CLASSIFICATION) -> None

The fence's table and T2.3.7's table name the same ports, or raise.

Source code in src/symfonic/services/effects/ports.py 
116
117
118
119
120
121
122
123
124
def assert_covers_classification(
    self, classification: EffectPortClassification = DEFAULT_EFFECT_CLASSIFICATION
) -> None:
    """The fence's table and T2.3.7's table name the same ports, or raise."""
    gap = set(classification.by_id) - set(self._by_id)
    if gap:
        raise IncompleteFenceCoverageError(
            f"the fence does not cover classified ports {sorted(gap)}"
        )