Skip to content

symfonic.capabilities.memory.write_pending

write_pending

Unpublished coordinator entries and their accepted native PG claims.

retire_snapshot 

retire_snapshot(entries: dict[str, _Pending], snapshots: dict[str, PendingSnapshot], confirmed: set[str]) -> tuple[Claim, ...]

Prune confirmed snapshot tokens, never a later accepted version.

Source code in src/symfonic/capabilities/memory/write_pending.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def retire_snapshot(entries: dict[str, _Pending], snapshots: dict[str, PendingSnapshot],
                    confirmed: set[str]) -> tuple[Claim, ...]:
    """Prune confirmed snapshot tokens, never a later accepted version."""
    retired: list[Claim] = []
    for path, snap in snapshots.items():
        current = entries.get(path)
        if current is not snap.pending:
            continue
        if current.generation == snap.generation and set(snap.ids) <= confirmed:
            retired.extend(snap.claims)
            del entries[path]
            continue
        safe = {
            claim for claim, generation in snap.claim_generations
            if claim[1] in confirmed and current.claim_generations.get(claim) == generation
        }
        if not safe:
            continue
        retired.extend(safe)
        current.claims = [claim for claim in current.claims if claim not in safe]
        for claim in safe:
            del current.claim_generations[claim]
        remaining = {claim[1] for claim in current.claims}
        retired_ids = {claim[1] for claim in safe}
        current.ids = [record_id for record_id in current.ids
                       if record_id not in retired_ids or record_id in remaining]
        if not current.ids and not current.claims:
            del entries[path]
    return tuple(retired)

settle async 

settle(request: WriteRequest, task: Any) -> WriteOutcome

Classify a background write's cancellation without swallowing ours.

Source code in src/symfonic/capabilities/memory/write_pending.py 
86
87
88
89
90
91
92
93
94
95
96
97
async def settle(request: WriteRequest, task: Any) -> WriteOutcome:
    """Classify a background write's cancellation without swallowing ours."""
    try:
        receipt = await task
    except asyncio.CancelledError:
        current = asyncio.current_task()
        if current is not None and current.cancelling() > 0:
            raise
        return WriteOutcome(request=request, error="CancelledError: write cancelled")
    except Exception as exc:  # noqa: BLE001 - the outcome carries the failure
        return WriteOutcome(request=request, error=f"{type(exc).__name__}: {exc}")
    return WriteOutcome(request=request, receipt=receipt)

snapshot_pending 

snapshot_pending(entries: dict[str, _Pending], paths: tuple[str, ...]) -> dict[str, PendingSnapshot]

Capture exact accepted claim versions before an asynchronous lifecycle.

Source code in src/symfonic/capabilities/memory/write_pending.py 
41
42
43
44
45
46
47
48
49
50
51
52
def snapshot_pending(
    entries: dict[str, _Pending], paths: tuple[str, ...],
) -> dict[str, PendingSnapshot]:
    """Capture exact accepted claim versions before an asynchronous lifecycle."""
    return {
        path: PendingSnapshot(
            pending, pending.generation, tuple(pending.ids), tuple(pending.claims),
            tuple((claim, pending.claim_generations[claim]) for claim in pending.claims
                  if claim in pending.claim_generations),
        )
        for path in paths if (pending := entries.get(path)) is not None
    }