Ensure the store key's constraint again when another manager made it meanwhile, and write again a stamp it refuses (#298)

drivine/store/store_identity_resolver.py

@@ -8,13 +8,16 @@
 
 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
 
 if TYPE_CHECKING:
     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)))

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)))