Native endpoint locks coordinate edge insertion with node cascades.
V0 writes retain their SQL and endpoint behavior. This prerequisite does not
declare PostgreSQL native backend admission or migrate stored identities.
lock_native_endpoints
async
lock_native_endpoints(conn: Any, scope: StorageScope, edge: MemoryEdge) -> None
Hold live visible endpoints FOR SHARE until the enclosing write commits.
Source code in src/symfonic/memory/backends/postgres_native_edge_write.py
| async def lock_native_endpoints(conn: Any, scope: StorageScope, edge: MemoryEdge) -> None:
"""Hold live visible endpoints FOR SHARE until the enclosing write commits."""
from symfonic.memory.backends.postgres_erase_jobs import refuse_if_erasing
from symfonic.memory.pending import is_pending
from symfonic.memory.retraction import is_retracted
identifiers = sorted({str(edge.source), str(edge.target)})
rows = await conn.fetch(
'SELECT id, properties FROM memory_nodes WHERE tenant_id=$1 '
'AND tenant_id COLLATE "C"=$1 AND id=ANY($2::text[]) '
'AND id COLLATE "C"=ANY($2::text[]) '
'AND scope_path COLLATE "C"=ANY($3::text[]) '
'ORDER BY id COLLATE "C" FOR SHARE',
scope.tenant_id, identifiers, scope.ancestor_prefix_paths(),
)
if len(rows) != len(identifiers):
raise ValueError("Native edge endpoints must be live and visible in this scope")
for row in rows:
properties = row["properties"]
if isinstance(properties, str):
properties = json.loads(properties)
if not isinstance(properties, dict) or is_pending(properties) or is_retracted(properties):
raise ValueError("Native edge endpoints must be live and visible in this scope")
# FOR SHARE holds ROW SHARE on memory_nodes, which the erase sweep waits on.
await refuse_if_erasing(conn, scope.tenant_id, (scope,))
|