async def flush_selected(
domain, operation, recall, *, journal, owner, revision, phase_key, ordinal,
):
"""Publish only the armed authenticated manifest, with no raw/HMS fallback."""
domain._session_for_call() # inherited children cannot poison the actual owner
try:
from symfonic.capabilities.memory.compat import record_from_legacy_node
from symfonic.capabilities.memory.graph_rows import node_payload, 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.pending import (
PENDING_BATCH_KEY,
PENDING_KEY,
is_pending,
is_quarantined,
published_properties,
)
from symfonic.memory.retraction import is_retracted
_owned_pair(domain.graph, domain.vector)
domain._require_selected_publication(operation)
if operation.verb is not MemoryOperationVerb.FLUSH or domain._read_only:
raise MongoSessionError("Native selected publication requires armed private FLUSH")
scope = NativeStorageScope.from_scope(operation.scope)
domain._require_tenant(scope)
expected = await _fence_owner(
domain, operation, journal, owner, revision, phase_key, ordinal)
await require_admitted(domain._database, expected, operation.scope)
prepared = []
for claim in operation.selected_manifest.claims:
if claim.owner != operation.scope:
_reject()
value = await domain.graph.get_node(scope, claim.record_id)
if (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 is_quarantined(value.properties) or is_retracted(value.properties)
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):
_reject()
record = record_from_legacy_node(node_payload(value))
if record.scope != claim.owner or record.record_id != claim.record_id:
_reject()
prepared.append((claim, value, record))
for claim, value, record in prepared:
await domain._remember_strict(scope, recall, (record,))
updated = await _Backend.__getattr__(domain.graph, "update_node")(
scope, claim.record_id, {"properties": published_properties(value.properties)},
)
if (updated is None or str(updated.id) != claim.record_id
or node_scope(updated) != claim.owner or is_pending(updated.properties)
or updated.properties.get(OWNER_KEY) != owner_document(claim.owner)
or updated.properties.get(CLAIM_KEY) != claim.pending_claim
or updated.properties.get(GENERATION_KEY) != claim.publication_generation):
_reject()
domain._complete_selected_publication(operation)
return LifecycleReceipt(scope_path=operation.scope.path,
committed=operation.selected_manifest.record_ids)
except BaseException:
domain._poison_receipt_mutation()
raise