From 2af29553ca18fa61b5b4b196985e292b8952da54 Mon Sep 17 00:00:00 2001 From: jasper blues Date: Sat, 31 Oct 2026 18:29:58 +1000 Subject: [PATCH 1/2] Ensure the store key's constraint again when another manager made it meanwhile, and write again a stamp it refuses (#298) Port of liberation-data/drivine4j#473 (drivine4j PR #477), following drivine PR #489. Managers resolving the store identity at once on a store without the DrivineStore(key) constraint race to make it. On FalkorDB and Postgres the losers' statements were refused for what the winner made meanwhile ("Constraint already exists.", a 23505 on pg_class_relname_nsp_index), so they went on without the constraint, warning that identity was best-effort, and a stamp the winner's constraint then refused was raised out of resolve(). - _ensure_unique_key ensures the constraint again, at most 3 times, when its making was refused as made meanwhile. A Postgres failure counts only as a 23505 on a pg_ catalog constraint, judged by SQLSTATE and constraint name, since psycopg's message carries "already exists" for any violation. Deadlocks are left to the schema executor's contention retry, which already runs Neo4j's DeadlockDetected again. - _assign_if_absent merges again on a uniqueness violation even when the constraint wasn't confirmed, on a manager with schema management only. A write conflict is still merged again only under a confirmed constraint. Tests: unit tests with a fake Neo4j store through the real ConstraintManager and SchemaExecutor, async and sync; and 16 managers racing per round on every store (5 rounds; 3 on Postgres, 2 on Oracle), with tasks and with threads. Fixes #298. --- drivine/store/store_identity_resolver.py | 88 ++++- drivine/sync/store/store_identity_resolver.py | 88 ++++- tests/integration/test_schema.py | 30 ++ tests/sync/test_store_identity_race.py | 29 ++ tests/sync_integration/test_schema.py | 11 + tests/unit/test_store_identity_race.py | 307 ++++++++++++++++++ 6 files changed, 545 insertions(+), 8 deletions(-) create mode 100644 tests/sync/test_store_identity_race.py create mode 100644 tests/unit/test_store_identity_race.py diff --git a/drivine/store/store_identity_resolver.py b/drivine/store/store_identity_resolver.py index c7c3df3..ebbacf9 100644 --- a/drivine/store/store_identity_resolver.py +++ b/drivine/store/store_identity_resolver.py @@ -8,6 +8,7 @@ from drivine import concurrency from drivine.drivine_error import DrivineError +from drivine.manager.contention import any_cause, contended_pause from drivine.schema.schema_specs import uniqueness_constraint from drivine.store.store_identity import StoreIdentity @@ -15,6 +16,8 @@ from drivine.backend.graph_backend import GraphBackend from drivine.manager.persistence_manager import PersistenceManager from drivine.schema.constraint_manager import ConstraintManager + from drivine.schema.ensure_result import EnsureResult + from drivine.schema.uniqueness_constraint_spec import UniquenessConstraintSpec logger = logging.getLogger("drivine.store") @@ -25,7 +28,33 @@ _ATTEMPTS = 3 _MERGE_ATTEMPTS = 5 _MERGE_BACKOFF_SECONDS = 0.05 -_LOST_RACE = re.compile(r"constraint|already exists|conflict|another write", re.IGNORECASE) +_ENSURE_ATTEMPTS = 3 +_VIOLATED_THE_KEY = re.compile(r"unique constraint|already exists", re.IGNORECASE) +_WRITE_CONFLICT = re.compile(r"conflict|another write", re.IGNORECASE) +_MADE_MEANWHILE = re.compile(r"already (exists|indexed)|ORA-00955\b", re.IGNORECASE) +_UNIQUE_VIOLATION = "23505" + + +def _made_meanwhile(failure: BaseException) -> bool: + """ + Whether a schema statement was refused for what another session made meanwhile: a constraint or index that already + exists (FalkorDB's "Constraint already exists", or its "Attribute 'key' is already indexed" for the index the + constraint needs), or the index's name already taken: on Oracle ORA-00955, and on Postgres a unique violation in + its catalog, `pg_class` (23505 on a `pg_` index), as two sessions running `CREATE ... IF NOT EXISTS` at once get. + Not one in a table of the user's, which nothing made meanwhile: a Postgres failure is judged by its SQLSTATE and + constraint alone, as psycopg's message carries the DETAIL, "Key (...)=(...) already exists.", of any violation. The + failure and each cause under it are looked at. A refusal for something else that already exists is ensured again + too, in vain: it is raised after _ENSURE_ATTEMPTS, and `_ensure_unique_key` warns of it. + """ + if any_cause(failure, lambda it: getattr(it, "sqlstate", None) is not None): + return any_cause( + failure, + lambda it: ( + getattr(it, "sqlstate", None) == _UNIQUE_VIOLATION + and str(getattr(getattr(it, "diag", None), "constraint_name", None) or "").startswith("pg_") + ), + ) + return any_cause(failure, lambda it: bool(_MADE_MEANWHILE.search(str(it)))) class StoreIdentityResolver: @@ -66,11 +95,21 @@ async def resolve(self) -> StoreIdentity: ) async def _ensure_unique_key(self) -> bool: + """ + Ensures the uniqueness constraint on the key. Returns whether it is in force; False degrades `resolve` to + best-effort convergence rather than failing boot over an identity that can still be read. + + Managers making the constraint at once get in each other's way. A deadlock (Neo4j's, on the label's lock) or a + write turned away is already run again by the schema executor (see `when_not_contended`). One manager's + statement can also be refused for what another's has just made: the constraint or the key's index already there + (FalkorDB), or the index's name taken in the catalog (Postgres's `pg_class`, 23505). The constraint is then + ensured again, which finds the other's. Going on without it would let a stamp be refused by the other's + constraint, or two stamps be written before it is in force (#298, as drivine4j#473). + """ if self.constraints is None: return False try: - # The label is Drivine's, so a plain index on the key (Neo4j) is replaced by the constraint. - result = await self.constraints.ensure(_KEY_CONSTRAINT, replace_blocking_index=True) + result = await self._ensured_again_when_made_meanwhile(self.constraints) except Exception as error: logger.warning("Could not ensure the DrivineStore key constraint: %s. Identity is best-effort.", error) return False @@ -91,7 +130,33 @@ async def _ensure_unique_key(self) -> bool: return False return True + async def _ensured_again_when_made_meanwhile( + self, constraints: ConstraintManager + ) -> EnsureResult[UniquenessConstraintSpec]: + """`ensure`s the key's constraint, again when another manager made it, or what it needs, meanwhile.""" + attempt = 1 + while True: + try: + # The label is Drivine's, so a plain index on the key (Neo4j) is replaced by the constraint. + return await constraints.ensure(_KEY_CONSTRAINT, replace_blocking_index=True) + except Exception as error: + # Not a deadlock: the schema executor has run that again already, as many times as it runs one. + if attempt >= _ENSURE_ATTEMPTS or not _made_meanwhile(error): + raise + await concurrency.sleep(contended_pause(attempt - 1)) + attempt += 1 + async def _assign_if_absent(self, guarded: bool) -> None: + """ + `MERGE`s the stamp. Under the constraint, a concurrent first writer can make this one fail with a violation or a + write conflict rather than block; merging again, after a pause, then matches the winner's node. + + A violation is merged again even when `_ensure_unique_key` could not confirm the constraint, for the violation + itself proves one is in force: managers making it at once can be turned away making it, and a loser's stamp is + then refused by the winner's constraint (#298, as drivine4j#473). A conflict proves nothing, so it is merged + again only under a constraint confirmed. A store without schema management (no `constraints`) has no + constraint to have refused it, and raises at once. + """ attempt = 1 while True: try: @@ -108,7 +173,22 @@ async def _assign_if_absent(self, guarded: bool) -> None: return except Exception as error: # Lost the race to a peer's first boot: its stamp is there, or about to be. Back off and merge onto it. - if not guarded or attempt >= _MERGE_ATTEMPTS or not _LOST_RACE.search(str(error)): + if attempt >= _MERGE_ATTEMPTS or not ( + self._lost_the_race(error) if guarded else self._violated_the_key(error) + ): raise await concurrency.sleep(_MERGE_BACKOFF_SECONDS * 2 ** (attempt - 1)) attempt += 1 + + def _violated_the_key(self, error: BaseException) -> bool: + """ + Whether a uniqueness constraint refused the stamp: Neo4j's "already exists with label", Memgraph's and + FalkorDB's "unique constraint violation", Oracle's ORA-00001 "unique constraint violated", Postgres's "violates + unique constraint". Not another constraint, such as a not-null one, which would refuse it each time it was + written again. Only a manager with schema management can have made one. + """ + return self.constraints is not None and bool(_VIOLATED_THE_KEY.search(str(error))) + + def _lost_the_race(self, error: BaseException) -> bool: + """Whether a peer's boot won: a violation, or a write conflict (Memgraph's, or FalkorDB's "another write").""" + return self._violated_the_key(error) or bool(_WRITE_CONFLICT.search(str(error))) diff --git a/drivine/sync/store/store_identity_resolver.py b/drivine/sync/store/store_identity_resolver.py index d2ab3a8..765057f 100644 --- a/drivine/sync/store/store_identity_resolver.py +++ b/drivine/sync/store/store_identity_resolver.py @@ -12,9 +12,12 @@ from drivine.schema.schema_specs import uniqueness_constraint from drivine.store.store_identity import StoreIdentity from drivine.sync import concurrency +from drivine.sync.manager.contention import any_cause, contended_pause if TYPE_CHECKING: from drivine.backend.graph_backend import GraphBackend + from drivine.schema.ensure_result import EnsureResult + from drivine.schema.uniqueness_constraint_spec import UniquenessConstraintSpec from drivine.sync.manager.persistence_manager import PersistenceManager from drivine.sync.schema.constraint_manager import ConstraintManager @@ -27,7 +30,33 @@ _ATTEMPTS = 3 _MERGE_ATTEMPTS = 5 _MERGE_BACKOFF_SECONDS = 0.05 -_LOST_RACE = re.compile(r"constraint|already exists|conflict|another write", re.IGNORECASE) +_ENSURE_ATTEMPTS = 3 +_VIOLATED_THE_KEY = re.compile(r"unique constraint|already exists", re.IGNORECASE) +_WRITE_CONFLICT = re.compile(r"conflict|another write", re.IGNORECASE) +_MADE_MEANWHILE = re.compile(r"already (exists|indexed)|ORA-00955\b", re.IGNORECASE) +_UNIQUE_VIOLATION = "23505" + + +def _made_meanwhile(failure: BaseException) -> bool: + """ + Whether a schema statement was refused for what another session made meanwhile: a constraint or index that already + exists (FalkorDB's "Constraint already exists", or its "Attribute 'key' is already indexed" for the index the + constraint needs), or the index's name already taken: on Oracle ORA-00955, and on Postgres a unique violation in + its catalog, `pg_class` (23505 on a `pg_` index), as two sessions running `CREATE ... IF NOT EXISTS` at once get. + Not one in a table of the user's, which nothing made meanwhile: a Postgres failure is judged by its SQLSTATE and + constraint alone, as psycopg's message carries the DETAIL, "Key (...)=(...) already exists.", of any violation. The + failure and each cause under it are looked at. A refusal for something else that already exists is ensured again + too, in vain: it is raised after _ENSURE_ATTEMPTS, and `_ensure_unique_key` warns of it. + """ + if any_cause(failure, lambda it: getattr(it, "sqlstate", None) is not None): + return any_cause( + failure, + lambda it: ( + getattr(it, "sqlstate", None) == _UNIQUE_VIOLATION + and str(getattr(getattr(it, "diag", None), "constraint_name", None) or "").startswith("pg_") + ), + ) + return any_cause(failure, lambda it: bool(_MADE_MEANWHILE.search(str(it)))) class StoreIdentityResolver: @@ -68,11 +97,21 @@ def resolve(self) -> StoreIdentity: ) def _ensure_unique_key(self) -> bool: + """ + Ensures the uniqueness constraint on the key. Returns whether it is in force; False degrades `resolve` to + best-effort convergence rather than failing boot over an identity that can still be read. + + Managers making the constraint at once get in each other's way. A deadlock (Neo4j's, on the label's lock) or a + write turned away is already run again by the schema executor (see `when_not_contended`). One manager's + statement can also be refused for what another's has just made: the constraint or the key's index already there + (FalkorDB), or the index's name taken in the catalog (Postgres's `pg_class`, 23505). The constraint is then + ensured again, which finds the other's. Going on without it would let a stamp be refused by the other's + constraint, or two stamps be written before it is in force (#298, as drivine4j#473). + """ if self.constraints is None: return False try: - # The label is Drivine's, so a plain index on the key (Neo4j) is replaced by the constraint. - result = self.constraints.ensure(_KEY_CONSTRAINT, replace_blocking_index=True) + result = self._ensured_again_when_made_meanwhile(self.constraints) except Exception as error: logger.warning("Could not ensure the DrivineStore key constraint: %s. Identity is best-effort.", error) return False @@ -93,7 +132,33 @@ def _ensure_unique_key(self) -> bool: return False return True + def _ensured_again_when_made_meanwhile( + self, constraints: ConstraintManager + ) -> EnsureResult[UniquenessConstraintSpec]: + """`ensure`s the key's constraint, again when another manager made it, or what it needs, meanwhile.""" + attempt = 1 + while True: + try: + # The label is Drivine's, so a plain index on the key (Neo4j) is replaced by the constraint. + return constraints.ensure(_KEY_CONSTRAINT, replace_blocking_index=True) + except Exception as error: + # Not a deadlock: the schema executor has run that again already, as many times as it runs one. + if attempt >= _ENSURE_ATTEMPTS or not _made_meanwhile(error): + raise + concurrency.sleep(contended_pause(attempt - 1)) + attempt += 1 + def _assign_if_absent(self, guarded: bool) -> None: + """ + `MERGE`s the stamp. Under the constraint, a concurrent first writer can make this one fail with a violation or a + write conflict rather than block; merging again, after a pause, then matches the winner's node. + + A violation is merged again even when `_ensure_unique_key` could not confirm the constraint, for the violation + itself proves one is in force: managers making it at once can be turned away making it, and a loser's stamp is + then refused by the winner's constraint (#298, as drivine4j#473). A conflict proves nothing, so it is merged + again only under a constraint confirmed. A store without schema management (no `constraints`) has no + constraint to have refused it, and raises at once. + """ attempt = 1 while True: try: @@ -110,7 +175,22 @@ def _assign_if_absent(self, guarded: bool) -> None: return except Exception as error: # Lost the race to a peer's first boot: its stamp is there, or about to be. Back off and merge onto it. - if not guarded or attempt >= _MERGE_ATTEMPTS or not _LOST_RACE.search(str(error)): + if attempt >= _MERGE_ATTEMPTS or not ( + self._lost_the_race(error) if guarded else self._violated_the_key(error) + ): raise concurrency.sleep(_MERGE_BACKOFF_SECONDS * 2 ** (attempt - 1)) attempt += 1 + + def _violated_the_key(self, error: BaseException) -> bool: + """ + Whether a uniqueness constraint refused the stamp: Neo4j's "already exists with label", Memgraph's and + FalkorDB's "unique constraint violation", Oracle's ORA-00001 "unique constraint violated", Postgres's "violates + unique constraint". Not another constraint, such as a not-null one, which would refuse it each time it was + written again. Only a manager with schema management can have made one. + """ + return self.constraints is not None and bool(_VIOLATED_THE_KEY.search(str(error))) + + def _lost_the_race(self, error: BaseException) -> bool: + """Whether a peer's boot won: a violation, or a write conflict (Memgraph's, or FalkorDB's "another write").""" + return self._violated_the_key(error) or bool(_WRITE_CONFLICT.search(str(error))) diff --git a/tests/integration/test_schema.py b/tests/integration/test_schema.py index 6a6827b..2ef0a11 100644 --- a/tests/integration/test_schema.py +++ b/tests/integration/test_schema.py @@ -1,3 +1,4 @@ +import asyncio import logging import re import uuid @@ -283,6 +284,35 @@ async def test_store_identity_is_stable(factory, pm): assert one.id == two.id and one.engine == pm.type +RACE_ROUNDS = {DatabaseType.ORACLE: 2, DatabaseType.POSTGRES: 3} +"""How many rounds of managers racing to make the store key's constraint a store runs: fewer on the SQL stores.""" + + +async def wipe_store_stamps(pm: PersistenceManager) -> None: + if pm.type in (DatabaseType.ORACLE, DatabaseType.POSTGRES): + await pm.execute(QuerySpecification("DELETE FROM drivine_node")) + else: + await pm.execute(QuerySpecification("MATCH (s:DrivineStore) DETACH DELETE s")) + + +@pytest.mark.async_only +async def test_store_identity_is_agreed_under_the_key_constraint_by_managers_racing_to_make_it(pm, caplog): + # #298, as drivine4j#473: on a store that has never had the key's constraint, the managers race to make it. On + # Neo4j they can deadlock doing so; elsewhere a loser's statement is refused for what the winner made meanwhile. + # Going on without the constraint, a loser's stamp was refused by the winner's and raised out of resolve(), or + # written before it was in force, leaving two stamps. Each manager must instead find the winner's constraint. + stamps = StatelessGraphObjectManager(pm).backend.store_stamps().spec() + with caplog.at_level(logging.WARNING, "drivine.store"): + for _ in range(RACE_ROUNDS.get(pm.type, 5)): + await StatelessGraphObjectManager(pm).constraints.drop(uniqueness_constraint("DrivineStore", "key")) + await wipe_store_stamps(pm) + managers = [StatelessGraphObjectManager(pm) for _ in range(16)] + identities = await asyncio.gather(*(it.store_identity() for it in managers)) + assert len({it.id for it in identities}) == 1 + assert len(await pm.query(stamps)) == 1 + assert [it.getMessage() for it in caplog.records if "best-effort" in it.getMessage()] == [] + + def neo4j_only(pm: PersistenceManager) -> None: if pm.type is not DatabaseType.NEO4J: pytest.skip("Only Neo4j's constraint brings its own index, so only there is a plain one in the way of it.") diff --git a/tests/sync/test_store_identity_race.py b/tests/sync/test_store_identity_race.py new file mode 100644 index 0000000..b7e0bcf --- /dev/null +++ b/tests/sync/test_store_identity_race.py @@ -0,0 +1,29 @@ +import logging +from concurrent.futures import ThreadPoolExecutor + +import pytest + +from drivine import uniqueness_constraint +from drivine.store.store_identity import StoreIdentity +from drivine.sync import PersistenceManager, StatelessGraphObjectManager +from tests.sync_integration.test_schema import RACE_ROUNDS, wipe_store_stamps + + +def test_store_identity_is_agreed_under_the_key_constraint_by_managers_racing_to_make_it( + pm: PersistenceManager, caplog: pytest.LogCaptureFixture +): + # #298, as drivine4j#473, as tests/integration/test_schema.py runs it with tasks: here each manager on a thread. + stamps = StatelessGraphObjectManager(pm).backend.store_stamps().spec() + + def resolve(manager: StatelessGraphObjectManager) -> StoreIdentity: + return manager.store_identity() + + with caplog.at_level(logging.WARNING, "drivine.store"), ThreadPoolExecutor(16) as threads: + for _ in range(RACE_ROUNDS.get(pm.type, 5)): + StatelessGraphObjectManager(pm).constraints.drop(uniqueness_constraint("DrivineStore", "key")) + wipe_store_stamps(pm) + managers = [StatelessGraphObjectManager(pm) for _ in range(16)] + identities = list(threads.map(resolve, managers)) + assert len({it.id for it in identities}) == 1 + assert len(pm.query(stamps)) == 1 + assert [it.getMessage() for it in caplog.records if "best-effort" in it.getMessage()] == [] diff --git a/tests/sync_integration/test_schema.py b/tests/sync_integration/test_schema.py index 26354e5..5c73690 100644 --- a/tests/sync_integration/test_schema.py +++ b/tests/sync_integration/test_schema.py @@ -283,6 +283,17 @@ def test_store_identity_is_stable(factory, pm): assert one.id == two.id and one.engine == pm.type +RACE_ROUNDS = {DatabaseType.ORACLE: 2, DatabaseType.POSTGRES: 3} +"""How many rounds of managers racing to make the store key's constraint a store runs: fewer on the SQL stores.""" + + +def wipe_store_stamps(pm: PersistenceManager) -> None: + if pm.type in (DatabaseType.ORACLE, DatabaseType.POSTGRES): + pm.execute(QuerySpecification("DELETE FROM drivine_node")) + else: + pm.execute(QuerySpecification("MATCH (s:DrivineStore) DETACH DELETE s")) + + def neo4j_only(pm: PersistenceManager) -> None: if pm.type is not DatabaseType.NEO4J: pytest.skip("Only Neo4j's constraint brings its own index, so only there is a plain one in the way of it.") diff --git a/tests/unit/test_store_identity_race.py b/tests/unit/test_store_identity_race.py new file mode 100644 index 0000000..1ad9cc1 --- /dev/null +++ b/tests/unit/test_store_identity_race.py @@ -0,0 +1,307 @@ +""" +Managers resolving the store identity at once, each making the key's constraint: #298, as drivine4j#473. The fake store +runs its schema statements through the real ConstraintManager and SchemaExecutor, async and sync, so a deadlock making +the constraint is run again as it would be. +""" + +import logging +from types import SimpleNamespace +from typing import Any, Literal + +import pytest +from neo4j.exceptions import Neo4jError + +from drivine.backend.backends import backend_for +from drivine.connection.database_type import DatabaseType +from drivine.drivine_error import DrivineError +from drivine.query.query_specification import QuerySpecification +from drivine.schema.constraint_manager import ConstraintManager +from drivine.schema.schema_executor import SchemaExecutor +from drivine.schema.schema_grammar import NEO4J_SCHEMA +from drivine.store.store_identity_resolver import StoreIdentityResolver +from drivine.sync.schema.constraint_manager import ConstraintManager as SyncConstraintManager +from drivine.sync.schema.schema_executor import SchemaExecutor as SyncSchemaExecutor +from drivine.sync.store.store_identity_resolver import StoreIdentityResolver as SyncStoreIdentityResolver + +Api = Literal["async", "sync"] +APIS: list[Api] = ["async", "sync"] +EVERY_TIME = 1_000_000 + +DEADLOCK = "Neo.TransientError.Transaction.DeadlockDetected" +CONFLICT = "Cannot resolve conflicting transactions. You can retry this transaction." + + +def bolt(code: str, message: str | None = None) -> Neo4jError: + """An error as the driver makes one from the server's failure.""" + return Neo4jError._hydrate_neo4j(code=code, message=message or code) # pyright: ignore[reportPrivateUsage] + + +def violated() -> Neo4jError: + return bolt( + "Neo.ClientError.Schema.ConstraintValidationFailed", + "Node(0) already exists with label `DrivineStore` and property `key` = 'singleton'", + ) + + +class PostgresUniqueViolation(Exception): + """ + A unique violation as psycopg raises one: its SQLSTATE, the constraint it names in its diagnostics, and a message + that carries the DETAIL, which says "already exists" of any violation. + """ + + def __init__(self, constraint: str) -> None: + super().__init__( + f'duplicate key value violates unique constraint "{constraint}"\nDETAIL: Key (id)=(7) already exists.' + ) + self.sqlstate = "23505" + self.diag = SimpleNamespace(constraint_name=constraint) + + +class Store: + """ + A Neo4j store whose stamp MERGE fails with the next of `merge_failures`, if any. The first `create_failures` + CREATE CONSTRAINTs deadlock, or, given `made_meanwhile`, are refused with it, another session having made the + constraint, unless `found_after` is False: then nothing made it, and the refusal is for something else. + """ + + def __init__( + self, + create_failures: int, + merge_failures: list[BaseException], + made_meanwhile: BaseException | None = None, + found_after: bool = True, + ) -> None: + self.create_failures = create_failures + self.merge_failures = merge_failures + self.made_meanwhile = made_meanwhile + self.found_after = found_after + self.created = False + self.creates = 0 + self.merges = 0 + + def query(self, spec: QuerySpecification[Any]) -> list[Any]: + assert spec.statement is not None + statement = spec.statement.text + if statement.startswith("CREATE CONSTRAINT"): + self.creates += 1 + if self.creates <= self.create_failures: + self.created = self.made_meanwhile is not None and self.found_after + raise self.made_meanwhile or bolt(DEADLOCK, "Deadlock wait cycle.") + self.created = True + if statement.startswith("SHOW CONSTRAINTS") and self.created: + return [{"name": "DrivineStore_key", "labels": ["DrivineStore"], "properties": ["key"]}] + return [] + + def execute(self) -> None: + self.merges += 1 + if self.merge_failures: + failure = self.merge_failures.pop(0) + raise DrivineError.with_root_cause(failure) from failure + + def stamps(self) -> list[dict[str, str]]: + return [{"id": "store-8", "assignedAt": "2026-01-01T00:00:00Z"}] + + +class AsyncConnection: + def __init__(self, store: Store) -> None: + self.store = store + + async def query(self, spec: QuerySpecification[Any]) -> list[Any]: + return self.store.query(spec) + + async def release(self, error: BaseException | None = None) -> None: + pass + + +class SyncConnection: + def __init__(self, store: Store) -> None: + self.store = store + + def query(self, spec: QuerySpecification[Any]) -> list[Any]: + return self.store.query(spec) + + def release(self, error: BaseException | None = None) -> None: + pass + + +class AsyncProvider: + name = "neo4j" + type = DatabaseType.NEO4J + + def __init__(self, store: Store) -> None: + self.store = store + + async def connect(self) -> AsyncConnection: + return AsyncConnection(self.store) + + +class SyncProvider: + name = "neo4j" + type = DatabaseType.NEO4J + + def __init__(self, store: Store) -> None: + self.store = store + + def connect(self) -> SyncConnection: + return SyncConnection(self.store) + + +class AsyncPm: + database = "neo4j" + type = DatabaseType.NEO4J + + def __init__(self, store: Store) -> None: + self.store = store + + async def execute(self, spec: QuerySpecification[Any]) -> None: + self.store.execute() + + async def query(self, spec: QuerySpecification[Any]) -> list[Any]: + return self.store.stamps() + + +class SyncPm: + database = "neo4j" + type = DatabaseType.NEO4J + + def __init__(self, store: Store) -> None: + self.store = store + + def execute(self, spec: QuerySpecification[Any]) -> None: + self.store.execute() + + def query(self, spec: QuerySpecification[Any]) -> list[Any]: + return self.store.stamps() + + +async def resolve(store: Store, api: Api, schema_management: bool = True) -> str: + """The id the API's resolver gives over the store, with schema management or (as on Neptune) without it.""" + backend = backend_for(DatabaseType.NEO4J) + if api == "async": + constraints = ConstraintManager(SchemaExecutor(AsyncProvider(store)), NEO4J_SCHEMA) # pyright: ignore[reportArgumentType] + resolver = StoreIdentityResolver(AsyncPm(store), backend, constraints if schema_management else None) # pyright: ignore[reportArgumentType] + return (await resolver.resolve()).id + sync_constraints = SyncConstraintManager(SyncSchemaExecutor(SyncProvider(store)), NEO4J_SCHEMA) # pyright: ignore[reportArgumentType] + sync_resolver = SyncStoreIdentityResolver(SyncPm(store), backend, sync_constraints if schema_management else None) # pyright: ignore[reportArgumentType] + return sync_resolver.resolve().id + + +def best_effort(caplog: pytest.LogCaptureFixture) -> list[str]: + return [it.getMessage() for it in caplog.records if "best-effort" in it.getMessage()] + + +@pytest.fixture(autouse=True) +def warnings(caplog: pytest.LogCaptureFixture) -> None: + caplog.set_level(logging.WARNING, "drivine.store") + + +@pytest.mark.parametrize("api", APIS) +async def test_a_stamp_the_constraint_refuses_is_written_again_though_its_making_deadlocked(api: Api): + # The refusal itself proves a constraint in force, so it is taken for a lost race though none was confirmed. + store = Store(EVERY_TIME, [violated()]) + + assert await resolve(store, api) == "store-8" + assert store.creates > 1, "the schema executor ran the deadlocked statement again" + assert store.merges == 2 + + +@pytest.mark.parametrize("api", APIS) +async def test_a_constraint_whose_making_deadlocked_once_is_made_again_so_the_stamp_is_written_under_it(api: Api): + # Only under a constraint confirmed is a conflict taken for a lost race, and the stamp written again. + store = Store(1, [Exception(CONFLICT)]) + + assert await resolve(store, api) == "store-8" + assert store.creates == 2 and store.merges == 2 + + +@pytest.mark.parametrize("api", APIS) +async def test_the_constraint_is_ensured_again_when_another_manager_made_it_meanwhile(api: Api, caplog): + # As FalkorDB refuses a second constraint on the key. + store = Store(1, [Exception(CONFLICT)], made_meanwhile=Exception("Constraint already exists.")) + + assert await resolve(store, api) == "store-8" + # The second ensure finds the other's constraint, so it isn't made again, and the stamp is written under it. + assert store.creates == 1 and store.merges == 2 + assert best_effort(caplog) == [] + + +@pytest.mark.parametrize("api", APIS) +async def test_the_constraint_is_ensured_again_when_another_manager_made_the_index_it_needs_meanwhile(api: Api, caplog): + store = Store(1, [], made_meanwhile=Exception("Attribute 'key' is already indexed")) + + assert await resolve(store, api) == "store-8" + assert store.creates == 1 and best_effort(caplog) == [] + + +@pytest.mark.parametrize("api", APIS) +async def test_the_constraint_is_ensured_again_when_postgres_refused_its_index_for_a_name_taken_in_pg_class( + api: Api, caplog +): + raced = PostgresUniqueViolation("pg_class_relname_nsp_index") + store = Store(1, [], made_meanwhile=DrivineError.with_root_cause(raced)) + + assert await resolve(store, api) == "store-8" + assert store.creates == 1 and best_effort(caplog) == [] + + +@pytest.mark.parametrize("api", APIS) +async def test_a_unique_violation_in_a_users_table_is_not_taken_for_one_made_meanwhile(api: Api, caplog): + violation = PostgresUniqueViolation("orders_pkey") + store = Store(1, [], made_meanwhile=DrivineError.with_root_cause(violation)) + + assert await resolve(store, api) == "store-8" + assert store.creates == 1 + assert len(best_effort(caplog)) == 1 and "orders_pkey" in best_effort(caplog)[0] + + +@pytest.mark.parametrize("api", APIS) +async def test_identity_is_best_effort_once_its_making_is_refused_as_made_meanwhile_three_times(api: Api, caplog): + # A real misconfiguration: something in the way that ensure can't find, so the constraint is never there. + refusal = Exception("There already exists an index (:DrivineStore {key}).") + store = Store(EVERY_TIME, [], made_meanwhile=refusal, found_after=False) + + assert await resolve(store, api) == "store-8" + assert store.creates == 3 and store.merges == 1 + assert len(best_effort(caplog)) == 1 and "There already exists an index" in best_effort(caplog)[0] + + +@pytest.mark.parametrize("api", APIS) +async def test_a_constraint_refusal_of_another_kind_is_not_ensured_again(api: Api, caplog): + store = Store(EVERY_TIME, [], made_meanwhile=bolt("Neo.ClientError.Security.Forbidden", "Schema operation denied")) + + assert await resolve(store, api) == "store-8" + assert store.creates == 1 and len(best_effort(caplog)) == 1 + + +@pytest.mark.parametrize("api", APIS) +async def test_a_stamp_refused_for_another_reason_after_the_deadlock_is_raised_at_once(api: Api): + refused = bolt("Neo.ClientError.Statement.SyntaxError", "Invalid input") + store = Store(EVERY_TIME, [refused]) + + with pytest.raises(DrivineError) as raised: + await resolve(store, api) + + assert raised.value.root_cause is refused and store.merges == 1 + + +@pytest.mark.parametrize("api", APIS) +async def test_a_stamp_refused_by_a_constraint_that_is_not_a_uniqueness_one_is_raised_at_once(api: Api): + # Only a uniqueness violation proves the key's constraint in force; a not-null one fails the same way each time. + refused = Exception('null value in column "props" of relation "drivine_node" violates not-null constraint') + store = Store(EVERY_TIME, [refused]) + + with pytest.raises(DrivineError) as raised: + await resolve(store, api) + + assert raised.value.root_cause is refused and store.merges == 1 + + +@pytest.mark.parametrize("api", APIS) +async def test_a_store_without_schema_management_raises_a_refused_stamp_at_once(api: Api): + # No constraint of its own can have refused it. + store = Store(0, [violated()]) + + with pytest.raises(DrivineError): + await resolve(store, api, schema_management=False) + + assert store.merges == 1 and store.creates == 0 From 2ab22b2c755092ca043354ac7fcc9d396907585d Mon Sep 17 00:00:00 2001 From: jasper blues Date: Sat, 31 Oct 2026 19:28:25 +1000 Subject: [PATCH 2/2] Judge a SQL store's refused stamp by its code, not its message, and read Oracle's code from the driver's error --- drivine/store/store_identity_resolver.py | 44 ++++++++++--- drivine/sync/store/store_identity_resolver.py | 44 ++++++++++--- tests/unit/test_store_identity_race.py | 62 +++++++++++++++++++ 3 files changed, 136 insertions(+), 14 deletions(-) diff --git a/drivine/store/store_identity_resolver.py b/drivine/store/store_identity_resolver.py index ebbacf9..ccff2a7 100644 --- a/drivine/store/store_identity_resolver.py +++ b/drivine/store/store_identity_resolver.py @@ -33,6 +33,37 @@ _WRITE_CONFLICT = re.compile(r"conflict|another write", re.IGNORECASE) _MADE_MEANWHILE = re.compile(r"already (exists|indexed)|ORA-00955\b", re.IGNORECASE) _UNIQUE_VIOLATION = "23505" +_ORACLE_UNIQUE_VIOLATION = "ORA-00001" + + +def _sqlstate(failure: BaseException) -> str | None: + """The SQLSTATE psycopg gives a Postgres failure; None for another store's.""" + return getattr(failure, "sqlstate", None) + + +def _oracle_code(failure: BaseException) -> str | None: + """ + The code python-oracledb gives an Oracle failure, its error's `full_code` ("ORA-00001"): the code that leads the + message, not another quoted further on, such as a stacked error's. None for another store's. + """ + detail = failure.args[0] if failure.args else None + code = getattr(detail, "full_code", None) + return code if isinstance(code, str) and code.startswith("ORA-") else None + + +def _unique_violation(failure: BaseException) -> bool: + """ + Whether a uniqueness constraint refused a write. A store that gives a code is judged by it alone: Postgres's + SQLSTATE 23505, whose message (psycopg's carries the DETAIL) says "already exists" as other failures' do, such as + 42P07's "relation ... already exists"; Oracle's ORA-00001 "unique constraint violated", not ORA-02261's "such unique + or primary key already exists". The graph engines are judged by their message: Neo4j's "already exists with + label", Memgraph's and FalkorDB's "unique constraint violation". The failure and each cause under it are looked at. + """ + if any_cause(failure, lambda it: _sqlstate(it) is not None): + return any_cause(failure, lambda it: _sqlstate(it) == _UNIQUE_VIOLATION) + if any_cause(failure, lambda it: _oracle_code(it) is not None): + return any_cause(failure, lambda it: _oracle_code(it) == _ORACLE_UNIQUE_VIOLATION) + return any_cause(failure, lambda it: bool(_VIOLATED_THE_KEY.search(str(it)))) def _made_meanwhile(failure: BaseException) -> bool: @@ -46,11 +77,11 @@ def _made_meanwhile(failure: BaseException) -> bool: failure and each cause under it are looked at. A refusal for something else that already exists is ensured again too, in vain: it is raised after _ENSURE_ATTEMPTS, and `_ensure_unique_key` warns of it. """ - if any_cause(failure, lambda it: getattr(it, "sqlstate", None) is not None): + if any_cause(failure, lambda it: _sqlstate(it) is not None): return any_cause( failure, lambda it: ( - getattr(it, "sqlstate", None) == _UNIQUE_VIOLATION + _sqlstate(it) == _UNIQUE_VIOLATION and str(getattr(getattr(it, "diag", None), "constraint_name", None) or "").startswith("pg_") ), ) @@ -182,12 +213,11 @@ async def _assign_if_absent(self, guarded: bool) -> None: def _violated_the_key(self, error: BaseException) -> bool: """ - Whether a uniqueness constraint refused the stamp: Neo4j's "already exists with label", Memgraph's and - FalkorDB's "unique constraint violation", Oracle's ORA-00001 "unique constraint violated", Postgres's "violates - unique constraint". Not another constraint, such as a not-null one, which would refuse it each time it was - written again. Only a manager with schema management can have made one. + Whether a uniqueness constraint refused the stamp (see `_unique_violation`). Not another constraint, such as a + not-null one, nor another failure that says "already exists", which would refuse it each time it was written + again. Only a manager with schema management can have made one. """ - return self.constraints is not None and bool(_VIOLATED_THE_KEY.search(str(error))) + return self.constraints is not None and _unique_violation(error) def _lost_the_race(self, error: BaseException) -> bool: """Whether a peer's boot won: a violation, or a write conflict (Memgraph's, or FalkorDB's "another write").""" diff --git a/drivine/sync/store/store_identity_resolver.py b/drivine/sync/store/store_identity_resolver.py index 765057f..2482e93 100644 --- a/drivine/sync/store/store_identity_resolver.py +++ b/drivine/sync/store/store_identity_resolver.py @@ -35,6 +35,37 @@ _WRITE_CONFLICT = re.compile(r"conflict|another write", re.IGNORECASE) _MADE_MEANWHILE = re.compile(r"already (exists|indexed)|ORA-00955\b", re.IGNORECASE) _UNIQUE_VIOLATION = "23505" +_ORACLE_UNIQUE_VIOLATION = "ORA-00001" + + +def _sqlstate(failure: BaseException) -> str | None: + """The SQLSTATE psycopg gives a Postgres failure; None for another store's.""" + return getattr(failure, "sqlstate", None) + + +def _oracle_code(failure: BaseException) -> str | None: + """ + The code python-oracledb gives an Oracle failure, its error's `full_code` ("ORA-00001"): the code that leads the + message, not another quoted further on, such as a stacked error's. None for another store's. + """ + detail = failure.args[0] if failure.args else None + code = getattr(detail, "full_code", None) + return code if isinstance(code, str) and code.startswith("ORA-") else None + + +def _unique_violation(failure: BaseException) -> bool: + """ + Whether a uniqueness constraint refused a write. A store that gives a code is judged by it alone: Postgres's + SQLSTATE 23505, whose message (psycopg's carries the DETAIL) says "already exists" as other failures' do, such as + 42P07's "relation ... already exists"; Oracle's ORA-00001 "unique constraint violated", not ORA-02261's "such unique + or primary key already exists". The graph engines are judged by their message: Neo4j's "already exists with + label", Memgraph's and FalkorDB's "unique constraint violation". The failure and each cause under it are looked at. + """ + if any_cause(failure, lambda it: _sqlstate(it) is not None): + return any_cause(failure, lambda it: _sqlstate(it) == _UNIQUE_VIOLATION) + if any_cause(failure, lambda it: _oracle_code(it) is not None): + return any_cause(failure, lambda it: _oracle_code(it) == _ORACLE_UNIQUE_VIOLATION) + return any_cause(failure, lambda it: bool(_VIOLATED_THE_KEY.search(str(it)))) def _made_meanwhile(failure: BaseException) -> bool: @@ -48,11 +79,11 @@ def _made_meanwhile(failure: BaseException) -> bool: failure and each cause under it are looked at. A refusal for something else that already exists is ensured again too, in vain: it is raised after _ENSURE_ATTEMPTS, and `_ensure_unique_key` warns of it. """ - if any_cause(failure, lambda it: getattr(it, "sqlstate", None) is not None): + if any_cause(failure, lambda it: _sqlstate(it) is not None): return any_cause( failure, lambda it: ( - getattr(it, "sqlstate", None) == _UNIQUE_VIOLATION + _sqlstate(it) == _UNIQUE_VIOLATION and str(getattr(getattr(it, "diag", None), "constraint_name", None) or "").startswith("pg_") ), ) @@ -184,12 +215,11 @@ def _assign_if_absent(self, guarded: bool) -> None: def _violated_the_key(self, error: BaseException) -> bool: """ - Whether a uniqueness constraint refused the stamp: Neo4j's "already exists with label", Memgraph's and - FalkorDB's "unique constraint violation", Oracle's ORA-00001 "unique constraint violated", Postgres's "violates - unique constraint". Not another constraint, such as a not-null one, which would refuse it each time it was - written again. Only a manager with schema management can have made one. + Whether a uniqueness constraint refused the stamp (see `_unique_violation`). Not another constraint, such as a + not-null one, nor another failure that says "already exists", which would refuse it each time it was written + again. Only a manager with schema management can have made one. """ - return self.constraints is not None and bool(_VIOLATED_THE_KEY.search(str(error))) + return self.constraints is not None and _unique_violation(error) def _lost_the_race(self, error: BaseException) -> bool: """Whether a peer's boot won: a violation, or a write conflict (Memgraph's, or FalkorDB's "another write").""" diff --git a/tests/unit/test_store_identity_race.py b/tests/unit/test_store_identity_race.py index 1ad9cc1..3facda3 100644 --- a/tests/unit/test_store_identity_race.py +++ b/tests/unit/test_store_identity_race.py @@ -296,6 +296,68 @@ async def test_a_stamp_refused_by_a_constraint_that_is_not_a_uniqueness_one_is_r assert raised.value.root_cause is refused and store.merges == 1 +class PostgresFailure(Exception): + """A failure as psycopg raises one: its SQLSTATE beside its message.""" + + def __init__(self, sqlstate: str, message: str) -> None: + super().__init__(message) + self.sqlstate = sqlstate + + +class OracleFailure(Exception): + """A failure as python-oracledb raises one: its error, which carries the code that leads the message.""" + + def __init__(self, message: str) -> None: + super().__init__(SimpleNamespace(full_code=message.split(":")[0], message=message)) + self.message = message + + def __str__(self) -> str: + return self.message + + +@pytest.mark.parametrize("api", APIS) +@pytest.mark.parametrize( + "refused", + [ + pytest.param(PostgresFailure("42P07", 'relation "drivine_node" already exists'), id="postgres-42P07"), + pytest.param( + OracleFailure("ORA-02261: such unique or primary key already exists in the table"), id="oracle-ORA-02261" + ), + pytest.param( + OracleFailure("ORA-02261: such unique or primary key already exists\nORA-06512: after ORA-00001"), + id="oracle-ORA-02261-quoting-ORA-00001", + ), + ], +) +async def test_a_stamp_refused_for_something_else_that_already_exists_is_raised_at_once( + api: Api, refused: BaseException +): + # Where the store gives a code, only its uniqueness violation's (23505, ORA-00001) proves the key's constraint. + store = Store(EVERY_TIME, [refused]) + + with pytest.raises(DrivineError) as raised: + await resolve(store, api) + + assert raised.value.root_cause is refused and store.merges == 1 + + +@pytest.mark.parametrize("api", APIS) +@pytest.mark.parametrize( + "refused", + [ + pytest.param(PostgresUniqueViolation("drivine_node_store_key"), id="postgres-23505"), + pytest.param( + OracleFailure("ORA-00001: unique constraint (DRIVINE.DRIVINE_STORE_KEY) violated"), id="oracle-ORA-00001" + ), + ], +) +async def test_a_stamp_the_sql_stores_key_constraint_refuses_is_written_again(api: Api, refused: BaseException): + store = Store(EVERY_TIME, [refused]) + + assert await resolve(store, api) == "store-8" + assert store.merges == 2 + + @pytest.mark.parametrize("api", APIS) async def test_a_store_without_schema_management_raises_a_refused_stamp_at_once(api: Api): # No constraint of its own can have refused it.