diff --git a/drivine/store/store_identity_resolver.py b/drivine/store/store_identity_resolver.py index c7c3df3..ccff2a7 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,64 @@ _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" +_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: + """ + 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: _sqlstate(it) is not None): + return any_cause( + failure, + lambda it: ( + _sqlstate(it) == _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 +126,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 +161,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 +204,21 @@ 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 (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 _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").""" + 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..2482e93 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,64 @@ _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" +_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: + """ + 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: _sqlstate(it) is not None): + return any_cause( + failure, + lambda it: ( + _sqlstate(it) == _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 +128,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 +163,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 +206,21 @@ 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 (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 _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").""" + 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..3facda3 --- /dev/null +++ b/tests/unit/test_store_identity_race.py @@ -0,0 +1,369 @@ +""" +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 + + +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. + store = Store(0, [violated()]) + + with pytest.raises(DrivineError): + await resolve(store, api, schema_management=False) + + assert store.merges == 1 and store.creates == 0