# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-happens-when-an-instance-forgets-a-memory-that-was-never-propagated-to-a-replica-the-provenance-erasure-divergence-invariant.md
# Remnant Warrant:    https://corpusfetch.com/c/what-happens-when-an-instance-forgets-a-memory-that-was-never-propagated-to-a-replica-the-provenance-erasure-divergence-invariant.warrant.json
# Markdown SHA-256:   a0ea7e1eb32ec76f01dbf1bbdf6a61de4c15b2a5be4ce858a793ae6a82820933
# Code AST SHA-256:   72a8a21a8b9e63cd61393fe4abe4b3c4b75b710aca0fa8328c924f39ad98a676
# Verified Classes:   ErasureDivergenceError, TombstoneIndex, ReplicaWriteValidator
# Defensive Guards:   3 fail-closed invariants (zero external deps)

class ErasureDivergenceError(RuntimeError):
    """Raised when a replica is about to apply a write that would resurrect
    a record already subject to an applied forget tombstone."""
    pass


class TombstoneIndex:
    def __init__(self):
        # mem_id -> highest primary sequence at which a forget was applied
        self._applied = {}

    def apply_tombstone(self, mem_id: str, primary_seq: int) -> None:
        prev = self._applied.get(mem_id)
        if prev is None or primary_seq > prev:
            self._applied[mem_id] = primary_seq

    def tombstone_seq(self, mem_id: str):
        return self._applied.get(mem_id)


class ReplicaWriteValidator:
    def __init__(self, tombstones: TombstoneIndex):
        self.tombstones = tombstones

    def validate_and_apply(self, mem_id: str, write_primary_seq: int,
                           client_provenance_ts: int, payload: bytes) -> None:
        # Ordering authority is the primary sequence, NOT the client provenance
        # timestamp. Client provenance is advisory audit metadata only.
        tomb_seq = self.tombstones.tombstone_seq(mem_id)
        if tomb_seq is not None and write_primary_seq < tomb_seq:
            raise ErasureDivergenceError(
                f"write for {mem_id!r} at primary_seq={write_primary_seq} "
                f"is older than applied tombstone at primary_seq={tomb_seq}; "
                f"applying it would resurrect an erased record "
                f"(client_provenance_ts={client_provenance_ts} is ignored for ordering)"
            )
        if tomb_seq is not None and write_primary_seq == tomb_seq:
            # Same sequence cannot be both a write and a forget; this indicates
            # a primary-side sequencing bug and must fail closed.
            raise ErasureDivergenceError(
                f"write and tombstone share primary_seq={write_primary_seq} "
                f"for {mem_id!r}; refusing to apply"
            )
        # Safe to apply: write is strictly newer than any applied tombstone.
        # (Payload application elided; the invariant check above is the point.)
        _ = payload


def forget_with_drain_barrier(primary, mem_id: str, basis: str,
                              requestor: str, replicas, timeout_ms: int = 2000):
    """Forget path: drain in-flight writes for mem_id before enqueuing the
    tombstone, so no replica can observe a write enqueued after the tombstone
    but carrying a higher client provenance timestamp."""
    pending = primary.pending_writes_for(mem_id)
    if pending:
        drained = primary.wait_for_acks(pending, timeout_ms=timeout_ms)
        if not drained:
            degraded = primary.replicas_with_pending(pending)
            for r in degraded:
                r.mark_degraded_for(mem_id)
            # Fail closed on the forget: do not acknowledge erasure while a
            # replica could still resurrect the record.
            raise ErasureDivergenceError(
                f"cannot guarantee erasure of {mem_id!r}: replicas {degraded} "
                f"still hold in-flight writes after {timeout_ms}ms; "
                f"forget NOT acknowledged (basis={basis}, requestor={requestor})"
            )
    return primary.enqueue_tombstone(mem_id, basis=basis, requestor=requestor)