symfonic.memory.backends.mongo_native_erased

mongo_native_erased

Which native owners a Mongo v1 forget erased, recorded durably (CORE-171).

A forget upserts one document per erased scope, keyed by that scope's canonical storage key, in the forget's own guarded transaction: it commits with the erasure or not at all. A document is only ever advanced, never removed. The positional migration reads these documents inside its own guarded transaction of the same tenant, so a forget either committed before the plan, and quarantines the v0 rows that would move under it, or waits on the tenant guard and erases them after the re-key.

An Agent invocation reads the same documents once, when its journal row is claimed, and every WRITE and FLUSH compares them again inside the tenant guard's transaction: a forget of the turn's scope or an ancestor committed after admission refuses the publication instead of resurrecting the scope.

erased async

erased(collection: Any, tenant_id: str, targets: tuple[str, ...]) -> frozenset[str]

The targets whose own key or an ancestor key a forget erased.

Source code in src/symfonic/memory/backends/mongo_native_erased.py

async def erased(collection: Any, tenant_id: str, targets: tuple[str, ...]) -> frozenset[str]:
    """The ``targets`` whose own key or an ancestor key a forget erased."""
    if not targets:
        return frozenset()
    chains = {target: covering_keys(target) for target in targets}
    keys = sorted({key for chain in chains.values() for key in chain})
    found = await collection.find(
        {"_id": {"$in": keys}, "tenant_id": tenant_id}, {"_id": 1}).to_list(None)
    recorded = {doc["_id"] for doc in found}
    return frozenset(target for target, chain in chains.items() if recorded.intersection(chain))

record async

record(collection: Any, scope: Any) -> None

Advance scope's erase document inside the forget's transaction.

Source code in src/symfonic/memory/backends/mongo_native_erased.py

async def record(collection: Any, scope: Any) -> None:
    """Advance ``scope``'s erase document inside the forget's transaction."""
    result = await collection.update_one(
        {"_id": scope.scope_path, "tenant_id": scope.tenant_id},
        {"$inc": {"generation": 1}},
        upsert=True,
    )
    if result.matched_count != 1 and result.upserted_id is None:
        raise MemoryContractError("native erase record was not written")

require_admitted async

require_admitted(database: Any, expected: dict, scope: Any) -> None

Refuse a publication whose admission witness a later forget advanced.

Runs inside the publication's tenant-guarded transaction, so a forget either committed before this read and refuses it, or waits on the guard.

Source code in src/symfonic/memory/backends/mongo_native_erased.py

async def require_admitted(database: Any, expected: dict, scope: Any) -> None:
    """Refuse a publication whose admission witness a later forget advanced.

    Runs inside the publication's tenant-guarded transaction, so a forget
    either committed before this read and refuses it, or waits on the guard.
    """
    row = await database["__symfonic_invocation_journal"].find_one(
        {**expected, "completion": {"$exists": False}}, {WITNESS_FIELD: 1})
    captured = None if row is None else row.get(WITNESS_FIELD)
    if type(captured) is not dict or any(type(v) is not int for v in captured.values()):
        raise MemoryContractError(STALE_PUBLICATION)
    current = await witness(database[ERASED_COLLECTION], scope)
    if captured != current:
        raise MemoryContractError(STALE_PUBLICATION)

witness async

witness(collection: Any, scope: Any) -> dict[str, int]

Every covering key's erase generation for scope now; absent is 0.

Source code in src/symfonic/memory/backends/mongo_native_erased.py

async def witness(collection: Any, scope: Any) -> dict[str, int]:
    """Every covering key's erase generation for ``scope`` now; absent is 0."""
    from symfonic.memory.storage_scope import NativeStorageScope

    storage = NativeStorageScope.from_scope(scope)
    keys = covering_keys(storage.scope_path)
    found = await collection.find(
        {"_id": {"$in": list(keys)}, "tenant_id": storage.tenant_id},
        {"_id": 1, "generation": 1}).to_list(None)
    generations = {doc["_id"]: doc.get("generation") for doc in found}
    return {key: generations.get(key, 0) for key in keys}