Plan, apply, roll back and clean up a CORE-171 positional row migration.
Explicit and adopter-invoked: no read path calls this, and nothing here
dual-reads a positional key as native. plan mutates no row. apply
re-plans inside the backend's write/forget serialization and refuses unless
the result matches the reviewed plan's digest, then re-keys every owner copy
of every row it re-read -- never the caller's rows -- in that same
transaction: all or none. A plan whose rows do not hash to its digest
refuses before any read, so an edited plan authorizes nothing. A row whose
target a recorded native forget already erased is quarantined, never
migrated. A backend that cannot prove which owners a forget erased refuses
any plan that would move a row: PostgreSQL keeps no record of a forget that
ran before memory_erase_generations existed, so there only quarantine-only
plans run. rollback restores quarantined rows only, each to the v0 owner
its quarantine key records, and refuses a plan that moved any row to a native
owner (ROLLBACK_UNPROVABLE): no store or audit record proves a row there
is the one apply moved rather than one written since. It refuses a row no
plan could have quarantined (NOT_RESTORABLE), a restore the store cannot
prove came after no forget of its origin (every PostgreSQL and Mongo
rollback; on the in-process store, any forget or delete in the tenant since
apply), and refuses while any of them holds a staged (unflushed) write.
delete_quarantined_rows is the only way a quarantined row leaves the
store, because no scoped read or forget
in either version matches its key. Apply, rollback and deletion each record
their rows through an audit logger before any row changes, or refuse
(:mod:symfonic.capabilities.memory.positional_audit).
A migrated row leaves every positional route -- v0 HTTP, explicit v0
composition and the legacy engine's reads -- in exchange for its complete
native owner; no supported step undoes that. A restored quarantined row is
served by every positional route again and leaves list_quarantined_rows.
Which stores are admitted, and how each serializes, is
:mod:symfonic.capabilities.memory.positional_store.
PlannedRow
dataclass
PlannedRow(kind: str, row_id: str, source: str | bool | int | float | None, target: str | None, disposition: RowDisposition, reason: str)
One row's audited move; owner keys only, never memory content.
PositionalMigrationPlan
dataclass
PositionalMigrationPlan(tenant_id: str, rows: tuple[PlannedRow, ...], digest: str)
A reviewable migration of one tenant, bound to the rows it read.
counts
counts() -> dict[str, int]
Rows per disposition, for an operator's review.
Source code in src/symfonic/capabilities/memory/positional_plan.py
| def counts(self) -> dict[str, int]:
"""Rows per disposition, for an operator's review."""
return dict(Counter(row.disposition.value for row in self.rows))
|
moved
moved() -> tuple[PlannedRow, ...]
The rows apply re-keys and rollback restores.
Source code in src/symfonic/capabilities/memory/positional_plan.py
| def moved(self) -> tuple[PlannedRow, ...]:
"""The rows apply re-keys and rollback restores."""
return tuple(row for row in self.rows if row.disposition is not RowDisposition.UNTOUCHED)
|
apply_positional_migration
async
apply_positional_migration(store: Any, plan: PositionalMigrationPlan, declaration: PositionalOwnerDeclaration, *, audit: Any = None, actor: str | None = None, delete_quarantined_edges: tuple[str, ...] = ()) -> PositionalMigrationPlan
Re-key exactly the reviewed plan, atomically, or refuse.
Every moved row -- migrated or quarantined -- is recorded with its prior
owner and reason before the re-key; see positional_audit.
delete_quarantined_edges names edges this plan quarantines that the
operator approves deleting in the same transaction, after the re-key: an
edge not yet quarantined is outside delete_quarantined_rows, and on
Mongo apply refuses while one would touch a node a native forget erases.
Source code in src/symfonic/capabilities/memory/positional_migration.py
| async def apply_positional_migration(
store: Any, plan: PositionalMigrationPlan, declaration: PositionalOwnerDeclaration,
*, audit: Any = None, actor: str | None = None,
delete_quarantined_edges: tuple[str, ...] = (),
) -> PositionalMigrationPlan:
"""Re-key exactly the reviewed plan, atomically, or refuse.
Every moved row -- migrated or quarantined -- is recorded with its prior
owner and reason before the re-key; see ``positional_audit``.
``delete_quarantined_edges`` names edges this plan quarantines that the
operator approves deleting in the same transaction, after the re-key: an
edge not yet quarantined is outside ``delete_quarantined_rows``, and on
Mongo apply refuses while one would touch a node a native forget erases.
"""
adapter = adapter_for(store)
if type(plan) is not PositionalMigrationPlan:
raise MemoryContractError("apply needs the reviewed PositionalMigrationPlan")
if adapter.apply_refusal is not None:
raise MemoryContractError(adapter.apply_refusal)
tenant = _tenant(plan.tenant_id)
_approved_edges(bound_plan(plan), delete_quarantined_edges)
trail = PositionalAudit(audit, tenant, actor, "positional_migration_apply")
adapter.refuse_caller_transaction()
try:
async with adapter.session(tenant, write=True) as session:
current = await plan_rows(session, tenant, _declared(declaration))
if current != plan:
raise MemoryContractError("the store changed since this migration was planned")
# From here every row, approval and audit entry is the re-read one.
plan = current
doomed = _approved_edges(plan, delete_quarantined_edges)
changes = tuple((row.kind, row.row_id, row.source, row.target)
for row in plan.moved())
await session.refuse_unerasable(changes, frozenset(row.row_id for row in doomed))
await trail.initiated(_evidence(plan.moved()) + [
[row.kind, row.row_id, prior_owner(row.stored), None, _APPROVED] for row in doomed
], plan_digest=plan.digest, counts=plan.counts(),
deleted_edges=[row.row_id for row in doomed])
await session.rekey(changes, require_all=True)
await session.delete(doomed)
except BaseException:
await trail.settled("failed")
raise
await trail.settled("completed")
return plan
|
delete_quarantined_rows
async
delete_quarantined_rows(store: Any, tenant_id: str, row_ids: tuple[str, ...], *, audit: Any = None, actor: str | None = None) -> tuple[tuple[str, str], ...]
Delete named quarantined rows and their quarantined dependents, or refuse all.
A named node takes its quarantined edges and quarantined claims with it,
and on Mongo its quarantined vector copy, which shares its id. PostgreSQL
never re-keys or quarantines a vector row and the in-process store keeps
none, so no vector is deleted there. The same refusal answers "missing"
and "not quarantined", so this cannot probe which rows exist. More
dependents than one deletion holds refuse before any row is deleted. The
rows are recorded through the audit logger before they are deleted, and
the returned pairs repeat what was deleted.
Source code in src/symfonic/capabilities/memory/positional_migration.py
| async def delete_quarantined_rows(
store: Any, tenant_id: str, row_ids: tuple[str, ...],
*, audit: Any = None, actor: str | None = None,
) -> tuple[tuple[str, str], ...]:
"""Delete named quarantined rows and their quarantined dependents, or refuse all.
A named node takes its quarantined edges and quarantined claims with it,
and on Mongo its quarantined vector copy, which shares its id. PostgreSQL
never re-keys or quarantines a vector row and the in-process store keeps
none, so no vector is deleted there. The same refusal answers "missing"
and "not quarantined", so this cannot probe which rows exist. More
dependents than one deletion holds refuse before any row is deleted. The
rows are recorded through the audit logger before they are deleted, and
the returned pairs repeat what was deleted.
"""
adapter, tenant = adapter_for(store), _tenant(tenant_id)
if type(row_ids) is not tuple or not row_ids or any(
type(row_id) is not str or not row_id for row_id in row_ids
):
raise MemoryContractError("name the quarantined rows to delete")
if len(set(row_ids)) > DELETE_BUDGET:
raise MemoryContractError(f"delete at most {DELETE_BUDGET} quarantined rows at once")
named, ceiling = frozenset(row_ids), 4 * DELETE_BUDGET
trail = PositionalAudit(audit, tenant, actor, "positional_quarantine_delete")
try:
async with adapter.session(tenant, write=True) as session:
found = await session.quarantined(named, ceiling + 1)
if len(found) > ceiling:
raise MemoryContractError(
f"the named rows carry more than {ceiling} quarantined rows to delete; "
"delete their quarantined edges by name first, in smaller batches")
if any(not is_quarantined_owner(row.stored) for row in found):
raise MemoryContractError(_NOT_QUARANTINED)
if not named <= {row.row_id for row in found} | {row.record for row in found}:
raise MemoryContractError(_NOT_QUARANTINED)
doomed = tuple(sorted(found, key=lambda row: (row.kind, row.row_id)))
await trail.initiated([[row.kind, row.row_id, prior_owner(row.stored), None,
"explicit deletion of a quarantined positional row"]
for row in doomed], named=sorted(named),
counts=dict(Counter(row.kind for row in doomed)))
await session.delete(doomed)
except BaseException:
await trail.settled("failed")
raise
await trail.settled("completed")
return tuple((row.kind, row.row_id) for row in doomed)
|
list_quarantined_rows
async
list_quarantined_rows(store: Any, tenant_id: str, *, limit: int = DELETE_BUDGET) -> tuple[tuple[str, str], ...]
Up to limit quarantined rows of one tenant, for a tenant erasure.
No scoped forget reaches a quarantined row, so an erasure of the tenant
must page this list into :func:delete_quarantined_rows until it is empty.
Source code in src/symfonic/capabilities/memory/positional_migration.py
| async def list_quarantined_rows(
store: Any, tenant_id: str, *, limit: int = DELETE_BUDGET,
) -> tuple[tuple[str, str], ...]:
"""Up to ``limit`` quarantined rows of one tenant, for a tenant erasure.
No scoped forget reaches a quarantined row, so an erasure of the tenant
must page this list into :func:`delete_quarantined_rows` until it is empty.
"""
adapter, tenant = adapter_for(store), _tenant(tenant_id)
if type(limit) is not int or not 0 < limit <= DELETE_BUDGET:
raise MemoryContractError(f"list at most {DELETE_BUDGET} quarantined rows at once")
async with adapter.session(tenant, write=False) as session:
rows = await session.quarantined(frozenset(), limit)
return tuple(sorted((row.kind, row.row_id) for row in rows))[:limit]
|
plan_positional_migration
async
plan_positional_migration(store: Any, tenant_id: str, declaration: PositionalOwnerDeclaration) -> PositionalMigrationPlan
Classify every candidate row of one tenant; moves nothing.
Source code in src/symfonic/capabilities/memory/positional_migration.py
| async def plan_positional_migration(
store: Any, tenant_id: str, declaration: PositionalOwnerDeclaration,
) -> PositionalMigrationPlan:
"""Classify every candidate row of one tenant; moves nothing."""
adapter, tenant = adapter_for(store), _tenant(tenant_id)
_declared(declaration)
async with adapter.session(tenant, write=False) as session:
return await plan_rows(session, tenant, declaration)
|
rollback_positional_migration
async
rollback_positional_migration(store: Any, plan: PositionalMigrationPlan, *, audit: Any = None, actor: str | None = None) -> tuple[tuple[str, str], ...]
Restore a quarantine-only plan's rows where the store proves it safe.
Returns the rows deleted since. Refuses before any read a plan that is
not exact values hashing to its digest, any row moved to a native owner
(ROLLBACK_UNPROVABLE), and any row whose target is not the quarantine
key of its source (NOT_RESTORABLE). A backend that cannot prove a
restored row's origin was not forgotten since apply refuses before its
session (restore_refusal: PostgreSQL and Mongo); the in-process store
refuses, before the audit trail, a row its own apply did not record or
that a later forget or delete may have covered. It also refuses while any
row holds a staged write. All refusals move nothing.
Source code in src/symfonic/capabilities/memory/positional_migration.py
| async def rollback_positional_migration(
store: Any, plan: PositionalMigrationPlan, *, audit: Any = None, actor: str | None = None,
) -> tuple[tuple[str, str], ...]:
"""Restore a quarantine-only plan's rows where the store proves it safe.
Returns the rows deleted since. Refuses before any read a plan that is
not exact values hashing to its digest, any row moved to a native owner
(``ROLLBACK_UNPROVABLE``), and any row whose target is not the quarantine
key of its source (``NOT_RESTORABLE``). A backend that cannot prove a
restored row's origin was not forgotten since apply refuses before its
session (``restore_refusal``: PostgreSQL and Mongo); the in-process store
refuses, before the audit trail, a row its own apply did not record or
that a later forget or delete may have covered. It also refuses while any
row holds a staged write. All refusals move nothing.
"""
adapter = adapter_for(store)
if type(plan) is not PositionalMigrationPlan:
raise MemoryContractError("rollback needs the applied PositionalMigrationPlan")
tenant = _tenant(plan.tenant_id)
trail = PositionalAudit(audit, tenant, actor, "positional_migration_rollback")
moved = restorable_plan(plan)
if adapter.restore_refusal is not None:
raise MemoryContractError(adapter.restore_refusal)
adapter.refuse_caller_transaction()
try:
async with adapter.session(tenant, write=True) as session:
await session.refuse_unproven_restore(tuple(
(row.kind, row.row_id, row.source, row.target) for row in moved))
await session.refuse_pending(tuple((row.kind, row.row_id) for row in moved))
await trail.initiated([[row.kind, row.row_id, row.target, row.source, row.reason]
for row in moved], plan_digest=plan.digest)
missing = await session.rekey(tuple(
(row.kind, row.row_id, row.target, row.source) for row in moved
), require_all=False)
except BaseException:
await trail.settled("failed")
raise
await trail.settled("completed", gone_since=sorted([list(row) for row in missing]))
return tuple(sorted(missing))
|