async def discard_selected(domain, operation, *, journal, owner, revision, phase_key, ordinal):
"""Erase only current authenticated pending claims and exact-owner physical rows."""
domain._session_for_call() # inherited children cannot poison the actual owner
try:
from symfonic.capabilities.memory.graph_rows import node_scope
from symfonic.capabilities.memory.operation_receipt import MemoryOperationVerb
from symfonic.capabilities.memory.records import LifecycleReceipt
from symfonic.capabilities.memory.selected_records import (
CLAIM_KEY,
GENERATION_KEY,
OWNER_KEY,
owner_document,
)
from symfonic.memory.layers.procedural.lifecycle import procedure_is_sealed_terminal
from symfonic.memory.pending import PENDING_BATCH_KEY, PENDING_KEY, is_quarantined
from symfonic.memory.retraction import is_retracted
_owned_pair(domain.graph, domain.vector)
domain._require_selected_publication(operation)
if operation.verb is not MemoryOperationVerb.DISCARD or domain._read_only:
_reject()
scope = NativeStorageScope.from_scope(operation.scope)
domain._require_tenant(scope)
await _fence_owner(domain, operation, journal, owner, revision, phase_key, ordinal)
graph = domain.graph._backend
nodes, edges, vectors = graph._nodes_col(scope), graph._edges_col(scope), domain.vector._col
prepared = []
for claim in operation.selected_manifest.claims:
value = await domain.graph.get_node(scope, claim.record_id)
if (claim.owner != operation.scope or value is None
or str(value.id) != claim.record_id or node_scope(value) != claim.owner
or value.properties.get(PENDING_KEY) is not True
or value.properties.get("durability") != "transient"
or value.properties.get(PENDING_BATCH_KEY) != claim.pending_claim
or value.properties.get(OWNER_KEY) != owner_document(claim.owner)
or value.properties.get(CLAIM_KEY) != claim.pending_claim
or value.properties.get(GENERATION_KEY) != claim.publication_generation
or is_quarantined(value.properties) or is_retracted(value.properties)
or (value.layer.value == "procedural"
and procedure_is_sealed_terminal(value.properties))):
_reject()
vector = await vectors.find_one({"_id": claim.record_id, "tenant_id": scope.tenant_id})
if vector is not None and (
type(vector.get("tenant_id")) is not str or vector["tenant_id"] != scope.tenant_id
or type(vector.get("scope_path")) is not str
or vector["scope_path"] != scope.scope_path
or type(vector.get("metadata")) is not dict
or type(vector["metadata"].get("scope_path")) is not str
or vector["metadata"]["scope_path"] != scope.scope_path
):
_reject()
prepared.append((claim, vector is not None))
incident = {"$or": [
{"source": {"$in": list(operation.selected_manifest.record_ids)}},
{"target": {"$in": list(operation.selected_manifest.record_ids)}},
]}
selected_edges = []
async for edge in edges.find(incident).limit(1025):
if (len(selected_edges) >= 1024 or type(edge.get("scope_path")) is not str
or edge["scope_path"] != scope.scope_path
or type(edge.get("source")) is not str or type(edge.get("target")) is not str):
_reject()
selected_edges.append(edge["_id"])
# Every pending node, vector owner and incident edge is validated BEFORE effects.
if selected_edges:
result = await edges.delete_many({"_id": {"$in": selected_edges},
**_exact("scope_path", scope.scope_path)})
if type(result.deleted_count) is not int or result.deleted_count != len(selected_edges):
_reject()
for claim, has_vector in prepared:
result = await vectors.delete_many({"$and": [
{"_id": claim.record_id}, _exact("tenant_id", scope.tenant_id),
_exact("scope_path", scope.scope_path),
_exact("metadata.scope_path", scope.scope_path),
]})
if type(result.deleted_count) is not int or result.deleted_count != int(has_vector):
_reject()
result = await nodes.delete_one({
"_id": claim.record_id, **_exact("properties.scope_path", scope.scope_path),
"properties." + PENDING_KEY: True,
"properties." + PENDING_BATCH_KEY: claim.pending_claim,
"properties." + CLAIM_KEY: claim.pending_claim,
"properties." + GENERATION_KEY: claim.publication_generation,
"properties." + OWNER_KEY: owner_document(claim.owner),
"properties.durability": "transient",
})
if type(result.deleted_count) is not int or result.deleted_count != 1:
_reject()
domain._complete_selected_publication(operation)
return LifecycleReceipt(scope_path=operation.scope.path,
discarded=operation.selected_manifest.record_ids)
except BaseException:
domain._poison_receipt_mutation()
raise