Skip to content

symfonic.services.effects.store

store

EFX-L-6 — the lease backend, and the linearization point of a revert.

Every method that establishes a fact holds lock. That is deliberate and it is the whole mechanism: EFX-F-3 forbids closing the check-to-use gap "by ordering conventions", so acquisition, renewal, revocation, and the fence-raise of a security revert all serialize against one another. A lease granted while a fence is landing is exactly the invocation that would keep running after the revert said stop.

Operated platforms replace this with the switch-store family's CAS backend (CUT-SS-1); the conditional-write shape is the same, and unavailable is here so a suite can prove what the fence does when the backend cannot be reached at all.

InMemoryLeaseStore 

InMemoryLeaseStore(*, fences: FenceLedger, clock: Callable[[], float] = time.time)

A single-process lease table with conditional writes (DMC-3).

Source code in src/symfonic/services/effects/store.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
def __init__(
    self,
    *,
    fences: FenceLedger,
    clock: Callable[[], float] = time.time,
) -> None:
    self._leases: dict[str, EffectLease] = {}
    self._children: dict[str, list[str]] = {}
    self._epochs: dict[str, int] = {}
    self._fences = fences
    self._clock = clock
    self._lock = asyncio.Lock()
    self.unavailable = False

lock property 

lock: Lock

Held across every conditional write, and by a landing revert.

acquire_child async 

acquire_child(parent_lease_id: str, *, invocation_id: str, ttl: float) -> EffectLease

EFX-L-2 — a child narrows its parent's lease and never outlives it.

Source code in src/symfonic/services/effects/store.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
async def acquire_child(
    self, parent_lease_id: str, *, invocation_id: str, ttl: float
) -> EffectLease:
    """EFX-L-2 — a child narrows its parent's lease and never outlives it."""
    async with self._lock:
        self._require_reachable()
        parent = self._require_live_locked(parent_lease_id)
        now = self._clock()
        return self._acquire_locked(
            invocation_id=invocation_id,
            bundle_id=parent.bundle_id,
            admitted_epoch=parent.admitted_epoch,
            generation_vector_hash=parent.generation_vector_hash,
            tenant_scope_hash=parent.tenant_scope_hash,
            ttl=ttl,
            stale_binding=parent.stale_binding,
            parent_lease_id=parent.lease_id,
            expires_at=min(now + ttl, parent.expires_at),
        )

check async 

check(lease_id: str) -> EffectLease

One atomic verification that this lease may still perform effects.

Source code in src/symfonic/services/effects/store.py 
168
169
170
171
async def check(self, lease_id: str) -> EffectLease:
    """One atomic verification that this lease may still perform effects."""
    async with self._lock:
        return self.check_locked(lease_id)

check_locked 

check_locked(lease_id: str) -> EffectLease

Fence first, then lease state. Caller holds lock.

The fence is consulted before liveness because a revert revokes what it fences, so both are true afterwards and the more specific answer is the useful one.

Source code in src/symfonic/services/effects/store.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def check_locked(self, lease_id: str) -> EffectLease:
    """Fence first, then lease state. Caller holds ``lock``.

    The fence is consulted before liveness because a revert revokes what it
    fences, so both are true afterwards and the more specific answer is the
    useful one.
    """
    self._require_reachable()
    lease = self._leases.get(lease_id)
    if lease is not None:
        fence = self._fences.covering(lease)
        if fence is not None:
            raise FencedEffectError(
                f"a {fence.kind.value} fence ({fence.fence_id}: {fence.reason}) "
                f"covers invocation {lease.invocation_id!r}; no effect is "
                "admitted, dispatched, or committed under a rejected "
                "generation or a fenced subject"
            )
    return self._require_live_locked(lease_id)

expire_due 

expire_due(now: float) -> tuple[str, ...]

EFX-L-5 — a crashed worker's lease expires and stops holding the drain.

Source code in src/symfonic/services/effects/store.py 
218
219
220
221
222
223
224
def expire_due(self, now: float) -> tuple[str, ...]:
    """EFX-L-5 — a crashed worker's lease expires and stops holding the drain."""
    expired: list[str] = []
    for lease in tuple(self._leases.values()):
        if lease.revoked_reason is None and now >= lease.expires_at:
            expired.extend(self._revoke_locked(lease.lease_id, "lease expired"))
    return tuple(dict.fromkeys(expired))

revoke_covered 

revoke_covered(fence: Fence, reason: str) -> tuple[str, ...]

Close every live lease this fence covers. Caller holds lock.

Source code in src/symfonic/services/effects/store.py 
198
199
200
201
202
203
204
205
206
def revoke_covered(self, fence: Fence, reason: str) -> tuple[str, ...]:
    """Close every live lease this fence covers. Caller holds ``lock``."""
    self._require_reachable()
    revoked: list[str] = []
    now = self._clock()
    for lease in tuple(self._leases.values()):
        if lease.live(now) and fence.covers(lease):
            revoked.extend(self._revoke_locked(lease.lease_id, reason))
    return tuple(dict.fromkeys(revoked))