orchestrata_ultimate.core.graph.retry

Reference › orchestrata_ultimate › core › graph › retry

A bounded, jittered retry for write conflicts in the operational graph. Neo4j blocks writers on locks, but Memgraph's MVCC engine aborts one of two conflicting writers with a transient error, so code that worked unchanged on Neo4j needs a retry against orchgraph (migration plan phase 2, step 2.1). Both Bolt adapters (bolt_sync and bolt_async) use it, and the knowledge-graph client folds is_mvcc_conflict_message into its own transient-error check.

is_mvcc_conflict_message and is_mvcc_conflict recognise a conflict by substrings of the message. The marker list deliberately includes a unique-constraint commit failure, because a concurrent MERGE race can fail that way and re-running the statement is correct. A permanent-failure guard keeps genuinely permanent errors out of the retry budget. retry_on_mvcc_conflict and retry_on_mvcc_conflict_async call a function up to MAX_ATTEMPTS (8, raised from 5 after a live concurrent-writer test exhausted five attempts) times, sleeping an exponential backoff with full jitter (_backoff_delay, base BASE_DELAY_SECONDS of 0.05). An error that is not a conflict propagates at once; when every attempt fails, the last conflict error is raised.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/graph/retry.py

"""MVCC write-conflict retry for the operational-graph repository seam.

Neo4j's write transactions block on locks; Memgraph's MVCC storage engine
instead ABORTS one of two concurrently-conflicting writers with a transient
error (``Cannot resolve conflicting transactions`` / a ``TransientError``
exception class). Code that ran unmodified against the graph (which simply
serializes the writers) needs a bounded, jittered retry to keep working
unchanged against orchgraph. See
``plans/neo4j-to-orchgraph-migration-plan.md`` Phase 2 Step 2.1.

Used by ``core.graph.adapters.{bolt_sync,bolt_async}`` and by
``knowledge_graph.graph_client.GraphClient._is_transient_error`` (which folds
``is_mvcc_conflict_message`` into its own existing transient-error check
rather than duplicating the marker list).
"""

from __future__ import annotations

import asyncio
import random
import time
from collections.abc import Awaitable, Callable

#: Bounded retry budget. Was 5 (the migration plan's worked example: "5
#: tries, 50ms base, jitter") until 2026-08-05: under the 14-worker gate a
#: live concurrent-writer proof (tests/core/graph/test_mvcc_retry.py)
#: measured a real conflict burning ALL 5 attempts and leaking a
#: TransientError — the absorption contract ("execute_write must absorb
#: every MVCC conflict") is the requirement, the worked example was only a
#: starting point. 8 attempts adds three more jittered sleep tiers —
#: attempts 4/5/6, ceilings 0.8s/1.6s/3.2s (0.05 × 2^n; +5.6s max added
#: wait) — that are paid ONLY while conflicts persist; the
#: _PERMANENT_MARKERS guard below keeps genuinely-permanent failures from
#: ever entering this budget.
MAX_ATTEMPTS = 8
BASE_DELAY_SECONDS = 0.05

# Substrings that identify a Memgraph MVCC serialization abort. Plain
# substring checks (not a regex) mirror the existing
# ``GraphClient._is_transient_error`` convention in this codebase.
#
# "constraint violation" is in this list DELIBERATELY, not an oversight:
# measured live 2026-07-26 with a real uniqueness constraint in place
# (every canonical-schema label carries one), a concurrent MERGE race on
# Memgraph does NOT always surface as "Cannot resolve conflicting
# transactions" — it can equally commit-fail as "Unable to commit due to
# unique constraint violation on :Label(prop)", a ClientError, not a
# TransientError. Retrying is still correct: the losing transaction just
# needs to re-run MERGE and observe the winner's now-committed row. Mirrors
# ``GraphClient._is_transient_error``'s existing "already exists" marker,
# which retries the SAME race's Neo4j-side error shape.
_TRANSIENT_MARKERS = (
    "Cannot resolve conflicting transactions",
    "TransientError",
    "SerializationError",
    "constraint violation",
    "already exists",
)


# Errors orchgraph/Memgraph reports with a TRANSIENT class that are in fact
# PERMANENT — retrying them can never succeed, it only converts a clear failure
# into a long hang. Measured live 2026-07-26: naming a database that does not
# exist answers `Memgraph.TransientError.MemgraphError` ("Tried to retrieve an
# unknown database"), so the driver's own managed-transaction retry (30s
# default) and this helper's 5 attempts NESTED, turning one misconfigured
# `database=` into minutes of silence per call instead of an immediate error.
_PERMANENT_MARKERS = (
    "unknown database",
    "Database does not exist",
)


def is_mvcc_conflict_message(msg: str) -> bool:
    """Return True if *msg* looks like a Memgraph MVCC write-conflict abort."""
    lowered = msg.lower()
    if any(marker.lower() in lowered for marker in _PERMANENT_MARKERS):
        return False
    return any(marker in msg for marker in _TRANSIENT_MARKERS)


def is_mvcc_conflict(exc: BaseException) -> bool:
    """Return True if *exc* looks like a Memgraph MVCC write-conflict abort.

    Checks both the exception's type name (``neo4j.exceptions.TransientError``
    and Memgraph-side ``Memgraph.TransientError`` both contain "Transient")
    and its message text (covers driver-wrapped ``ClientError`` shapes that
    carry the Memgraph error text but not the transient type).
    """
    message = str(exc)
    if any(marker.lower() in message.lower() for marker in _PERMANENT_MARKERS):
        # A transient-CLASSED but permanent failure (see _PERMANENT_MARKERS):
        # the type name check below would otherwise retry it forever.
        return False
    if "Transient" in type(exc).__name__:
        return True
    return is_mvcc_conflict_message(message)


def _backoff_delay(attempt: int) -> float:
    """Exponential backoff with full jitter. ``attempt`` is 0-indexed."""
    ceiling = BASE_DELAY_SECONDS * (2**attempt)
    return random.uniform(0, ceiling)


def retry_on_mvcc_conflict[T](fn: Callable[[], T]) -> T:
    """Call ``fn()``, retrying up to ``MAX_ATTEMPTS`` times on an MVCC abort.

    Non-MVCC exceptions propagate immediately (no retry). Re-raises the last
    MVCC exception if every attempt is exhausted.
    """
    last_exc: BaseException | None = None
    for attempt in range(MAX_ATTEMPTS):
        try:
            return fn()
        except Exception as exc:
            if not is_mvcc_conflict(exc):
                raise
            last_exc = exc
            if attempt < MAX_ATTEMPTS - 1:
                time.sleep(_backoff_delay(attempt))
    assert last_exc is not None
    raise last_exc


async def retry_on_mvcc_conflict_async[T](fn: Callable[[], Awaitable[T]]) -> T:
    """Await ``fn()``, retrying up to ``MAX_ATTEMPTS`` times on an MVCC abort."""
    last_exc: BaseException | None = None
    for attempt in range(MAX_ATTEMPTS):
        try:
            return await fn()
        except Exception as exc:
            if not is_mvcc_conflict(exc):
                raise
            last_exc = exc
            if attempt < MAX_ATTEMPTS - 1:
                await asyncio.sleep(_backoff_delay(attempt))
    assert last_exc is not None
    raise last_exc