symfonic.memory.backends.mongo_session

mongo_session

Opt-in shared-session Mongo backend composition; not a public mutation fence.

One domain owns its newly constructed client. Transactions are single-task, never replay their body, and refuse unsupported topology and operations. Unknown commit outcomes block subsequent transactions pending reconciliation.

MongoSessionDomain

MongoSessionDomain(uri: str, database: str = 'symfonic', *, collection_prefix: str = 'graph', collection: str = 'vectors')

Own a graph/vector pair whose collection operations share one session.

Creation is lazy. The extra lazy vector client is newly created here and closed before any use; existing externally owned clients are never rebound. This domain is an explicit backend prerequisite, not HMS admission.

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

def __init__(self, uri: str, database: str = "symfonic", *,
             collection_prefix: str = "graph", collection: str = "vectors") -> None:
    from symfonic.memory.backends.mongodb import MongoGraphBackend, MongoVectorBackend

    self.graph = MongoGraphBackend(uri, database, collection_prefix)
    try:
        self.vector = MongoVectorBackend(uri, database, collection)
    except BaseException:
        self.graph._client.close()
        raise
    self.vector._client.close()
    self._client = self.graph._client
    self.vector._client = self._client
    self._owner = None
    self._session = None
    self._closed = False
    self._uncertain = False
    shared_database = self._client[database]
    self.graph._db = _Database(shared_database, self)
    self.vector._db = self.graph._db
    self.vector._col = self.vector._db[collection]
    self._database = self.graph._db
    self._vector_collection = self.vector._col

close

close() -> None

Close only this domain's client; active transactions must exit first.

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

def close(self) -> None:
    """Close only this domain's client; active transactions must exit first."""
    if self._owner is not None:
        raise MongoSessionError("Cannot close an active transaction")
    self._closed = True
    self._client.close()

transaction async

transaction()

Settle I/O before abort; commit cancellation requires reconciliation.

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

@asynccontextmanager
async def transaction(self):
    """Settle I/O before abort; commit cancellation requires reconciliation."""
    try:
        from pymongo.read_concern import ReadConcern
        from pymongo.write_concern import WriteConcern
    except ImportError as exc:
        raise ImportError(
            "Mongo sessions require pymongo. Install: pip install symfonic-core[mongodb]"
        ) from exc

    if self._uncertain:
        raise MongoSessionError("Unknown commit outcome; reconcile before using a new domain")
    if self._closed or self._owner is not None:
        raise MongoSessionError("Closed, nested, or concurrent transaction")
    if self.graph._client is not self._client or self.vector._client is not self._client:
        raise MongoSessionError("Backends no longer share the owned client")
    if (self.graph._db is not self._database or self.vector._db is not self._database
            or self.vector._col is not self._vector_collection):
        raise MongoSessionError("Backends no longer use enlisted collection boundaries")
    self._owner = asyncio.current_task()
    try:
        hello = await self._client.admin.command("hello")
        if (not hello.get("setName") and hello.get("msg") != "isdbgrid") or (
            hello.get("logicalSessionTimeoutMinutes") is None
        ):
            raise MongoSessionError("Mongo transactions require replica set or mongos sessions")
        opened = await _settle(
            self._client.start_session(),
            cancelled_result=lambda session: session.end_session(),
        )
        async with opened as session:
            self._session = session
            session.start_transaction(
                read_concern=ReadConcern("snapshot"), write_concern=WriteConcern("majority"),
            )
            try:
                yield self
            except BaseException:
                try:
                    await _settle(session.abort_transaction())
                except BaseException:
                    self._uncertain = True
                    raise
                raise
            else:
                try:
                    await _settle(session.commit_transaction())
                except BaseException:
                    self._uncertain = True
                    raise
    except Exception as exc:
        label = getattr(exc, "has_error_label", None)
        if callable(label) and label("UnknownTransactionCommitResult"):
            self._uncertain = True
        raise
    finally:
        self._session = None
        self._owner = None

MongoSessionError

Bases: RuntimeError

An operation cannot participate in the current Mongo transaction.