symfonic.memory.backends.postgres_bulk_dedup

postgres_bulk_dedup

Private PostgreSQL BulkDedupPlan apply; never a public admission proof.

Both graph tables take SHARE ROW EXCLUSIVE locks for this short transaction. That blocks all tenants' writers, including v0 edge inserts that do not lock endpoint rows. It does not protect the earlier read/planning pass or later result read, so the public PostgreSQL bulk gate must stay closed.

apply async

apply(backend: PostgresGraphBackend, scope: TenantScope, plan: BulkDedupPlan) -> int

Own a short transaction for standalone private apply.

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

async def apply(backend: PostgresGraphBackend, scope: TenantScope, plan: BulkDedupPlan) -> int:
    """Own a short transaction for standalone private apply."""
    from symfonic.memory.janitor_bulk_plan import BulkDedupPlan

    if not isinstance(scope, TenantScope) or not isinstance(plan, BulkDedupPlan):
        raise TypeError("Legacy TenantScope and BulkDedupPlan required")
    if bound_connection(backend.pool) is not None:
        raise RuntimeError("Private PostgreSQL bulk apply requires its own transaction")
    _validate_plan(scope, plan)
    if not plan.expected_preimages and not plan.actions:
        return 0
    async with backend.pool.transaction():
        conn = await backend.pool.acquire()
        await lock_graph_tables(conn)
        return await apply_locked(conn, scope, plan)

apply_locked async

apply_locked(conn: Any, scope: TenantScope, plan: BulkDedupPlan) -> int

Apply a validated plan while the caller holds both table locks.

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

async def apply_locked(conn: Any, scope: TenantScope, plan: BulkDedupPlan) -> int:
    """Apply a validated plan while the caller holds both table locks."""
    from symfonic.memory.backends.postgres_erase_jobs import refuse_if_erasing
    from symfonic.memory.backends.postgres_graph import _node_properties_jsonb
    from symfonic.memory.scope_isolation import _is_native_scope

    _validate_plan(scope, plan)
    if not plan.expected_preimages and not plan.actions:
        return 0
    if _is_native_scope(scope):  # CORE-199: under the caller's table locks
        await refuse_if_erasing(conn, scope.tenant_id, (scope,))
    nodes = await _locked_rows(conn, scope, plan)
    edges = await _locked_edges(conn, scope, plan)
    removed_edges: set[str] = set()
    for action in plan.actions:
        if action.kind == "update":
            canonical = action.canonical_update
            assert canonical is not None
            status = await conn.execute(
                "UPDATE memory_nodes SET label=$4, importance=$5, "
                "properties=$6::jsonb, updated_at=$7 "
                "WHERE tenant_id=$1 AND id=$2 "
                'AND tenant_id COLLATE "C"=$1 AND id COLLATE "C"=$2 '
                "AND scope_path IS NOT DISTINCT FROM $3",
                scope.tenant_id,
                action.node_id,
                nodes[action.node_id]["scope_path"],
                canonical.label,
                canonical.importance,
                _node_properties_jsonb(canonical),
                datetime.now(UTC),
            )
            if status != "UPDATE 1":
                raise ValueError("Bulk dedup node preimage or owner changed")
        else:
            edge_ids = [edge for edge in edges[action.node_id] if edge not in removed_edges]
            if edge_ids:
                status = await conn.execute(
                    "DELETE FROM memory_edges WHERE tenant_id=$1 "
                    'AND tenant_id COLLATE "C"=$1 AND id=ANY($2::text[]) '
                    'AND id COLLATE "C"=ANY($2::text[])',
                    scope.tenant_id,
                    edge_ids,
                )
                if status != f"DELETE {len(edge_ids)}":
                    raise ValueError("Bulk dedup incident edge changed")
                removed_edges.update(edge_ids)
            status = await conn.execute(
                "DELETE FROM memory_nodes WHERE tenant_id=$1 AND id=$2 "
                'AND tenant_id COLLATE "C"=$1 AND id COLLATE "C"=$2 '
                "AND scope_path IS NOT DISTINCT FROM $3",
                scope.tenant_id,
                action.node_id,
                nodes[action.node_id]["scope_path"],
            )
            if status != "DELETE 1":
                raise ValueError("Bulk dedup node preimage or owner changed")
    return plan.merged_count

lock_graph_tables async

lock_graph_tables(conn: Any) -> None

Exclude node and edge writers before a private graph read.

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

async def lock_graph_tables(conn: Any) -> None:
    """Exclude node and edge writers before a private graph read."""
    await conn.execute("SET LOCAL lock_timeout = '5s'")
    await conn.execute("LOCK TABLE memory_nodes, memory_edges IN SHARE ROW EXCLUSIVE MODE")