From 56220e896eb181c8041341fad6aac935ec5de5be Mon Sep 17 00:00:00 2001 From: jasper blues Date: Sun, 1 Nov 2026 06:15:52 +1000 Subject: [PATCH 1/2] Run an Oracle index over the node table again while another session's lock turns it away, so racing managers keep the store key's constraint Managers making the DrivineStore(key) constraint together on Oracle were turned away by each other's lock on the node table (ORA-00054), as Oracle's DDL doesn't wait for a lock, and once by a library cache deadlock (ORA-04020). The schema executor's six attempts, about a second in all, ran out under CI's load, and a manager went on without the constraint ("Identity is best-effort"), failing tests/sync/test_store_identity_race.py on main (run 41190521832). The block that makes an index over the node table now runs its statement again a tenth of a second after a busy lock, for up to 300 tries, as drivine4j's OracleGraph.ensuring waits out a busy lock, and then finds what the other session made. The goldens are copied from Drivine for TypeScript, which makes the same change. --- .../backend/oracle/oracle_schema_grammar.py | 31 ++++++++++++++++--- tests/golden/VERSION | 2 +- tests/golden/statements.json | 6 ++-- tests/unit/test_oracle_schema_grammar.py | 22 +++++++++++-- 4 files changed, 51 insertions(+), 10 deletions(-) diff --git a/drivine/backend/oracle/oracle_schema_grammar.py b/drivine/backend/oracle/oracle_schema_grammar.py index 775bc25..8f5244b 100644 --- a/drivine/backend/oracle/oracle_schema_grammar.py +++ b/drivine/backend/oracle/oracle_schema_grammar.py @@ -52,15 +52,38 @@ def recorded_as(kind: CatalogKind) -> list[str]: """ +BUSY_ATTEMPTS = 300 +""" +How many times `_made` runs an index's statement that another session's lock on the node table turned away (ORA-00054), +a tenth of a second apart, before it raises: as drivine4j's `OracleGraph.BUSY_ATTEMPTS`. +""" + + def _made(ddl: str) -> str: - """A block that runs `ddl`, an object another session made meanwhile (ORA-00955) taken as made.""" + """ + A block that runs `ddl`, an index over the node table, an object another session made meanwhile (ORA-00955) taken + as made. A statement turned away because another session holds the node table's lock (ORA-00054), as one making the + same index, or writing a node, does, is run again a tenth of a second later, for up to `BUSY_ATTEMPTS` tries, and + then finds what the other made: managers making the store key's constraint together otherwise ran out of attempts + on the busy lock, and went on without the constraint. drivine4j's `OracleGraph.ensuring` waits out a busy lock so. + """ return f"""DECLARE e_exists EXCEPTION; PRAGMA EXCEPTION_INIT(e_exists, -955); + e_busy EXCEPTION; + PRAGMA EXCEPTION_INIT(e_busy, -54); BEGIN - EXECUTE IMMEDIATE {sql_text(ddl)}; -EXCEPTION - WHEN e_exists THEN NULL; + FOR attempt IN 1 .. {BUSY_ATTEMPTS} LOOP + BEGIN + EXECUTE IMMEDIATE {sql_text(ddl)}; + RETURN; + EXCEPTION + WHEN e_exists THEN RETURN; + WHEN e_busy THEN + IF attempt = {BUSY_ATTEMPTS} THEN RAISE; END IF; + DBMS_SESSION.SLEEP(0.1); + END; + END LOOP; END;""" diff --git a/tests/golden/VERSION b/tests/golden/VERSION index 3c91767..e7bfdc8 100644 --- a/tests/golden/VERSION +++ b/tests/golden/VERSION @@ -1 +1 @@ -32b7497ad14ce26d40058037da16b89542562540 +c34cf55178cf5971c4963e5a0f2445b581d469a7 diff --git a/tests/golden/statements.json b/tests/golden/statements.json index 50f1b11..485b5b2 100644 --- a/tests/golden/statements.json +++ b/tests/golden/statements.json @@ -333,10 +333,10 @@ "schema.listUniques": "SELECT JSON_OBJECT('label' VALUE c.label, 'properties' VALUE JSON(c.properties), 'name' VALUE c.name, 'details' VALUE c.details, 'incomplete' VALUE CASE WHEN c.kind = 'RANGE_INDEX' AND NOT EXISTS (SELECT 1 FROM user_indexes l WHERE l.index_name = UPPER(JSON_VALUE(c.details, '$.labelled'))) AND NOT EXISTS (SELECT 1 FROM drivine_schema u WHERE u.label = c.label AND u.properties = c.properties AND u.kind IN ('UNIQUENESS_CONSTRAINT', 'UNIQUE') AND EXISTS (SELECT 1 FROM user_indexes q WHERE q.index_name = UPPER(u.object_name))) THEN 'true' END RETURNING JSON) AS \"item\" FROM drivine_schema c WHERE c.kind IN ('UNIQUENESS_CONSTRAINT', 'UNIQUE') AND (c.object_name IS NULL OR EXISTS (SELECT 1 FROM user_indexes i WHERE i.index_name = UPPER(c.object_name)) OR EXISTS (SELECT 1 FROM user_tables t WHERE t.table_name = UPPER(c.object_name))) ORDER BY c.label, c.properties", "schema.listVectors": "SELECT JSON_OBJECT('label' VALUE c.label, 'properties' VALUE JSON(c.properties), 'name' VALUE c.name, 'details' VALUE c.details, 'incomplete' VALUE CASE WHEN c.kind = 'RANGE_INDEX' AND NOT EXISTS (SELECT 1 FROM user_indexes l WHERE l.index_name = UPPER(JSON_VALUE(c.details, '$.labelled'))) AND NOT EXISTS (SELECT 1 FROM drivine_schema u WHERE u.label = c.label AND u.properties = c.properties AND u.kind IN ('UNIQUENESS_CONSTRAINT', 'UNIQUE') AND EXISTS (SELECT 1 FROM user_indexes q WHERE q.index_name = UPPER(u.object_name))) THEN 'true' END RETURNING JSON) AS \"item\" FROM drivine_schema c WHERE c.kind IN ('VECTOR_INDEX', 'VECTOR') AND (c.object_name IS NULL OR EXISTS (SELECT 1 FROM user_indexes i WHERE i.index_name = UPPER(c.object_name)) OR EXISTS (SELECT 1 FROM user_tables t WHERE t.table_name = UPPER(c.object_name))) ORDER BY c.label, c.properties", "schema.listFullTexts": "SELECT JSON_OBJECT('label' VALUE c.label, 'properties' VALUE JSON(c.properties), 'name' VALUE c.name, 'details' VALUE c.details, 'incomplete' VALUE CASE WHEN c.kind = 'RANGE_INDEX' AND NOT EXISTS (SELECT 1 FROM user_indexes l WHERE l.index_name = UPPER(JSON_VALUE(c.details, '$.labelled'))) AND NOT EXISTS (SELECT 1 FROM drivine_schema u WHERE u.label = c.label AND u.properties = c.properties AND u.kind IN ('UNIQUENESS_CONSTRAINT', 'UNIQUE') AND EXISTS (SELECT 1 FROM user_indexes q WHERE q.index_name = UPPER(u.object_name))) THEN 'true' END RETURNING JSON) AS \"item\" FROM drivine_schema c WHERE c.kind IN ('FULLTEXT_INDEX', 'FULLTEXT') AND (c.object_name IS NULL OR EXISTS (SELECT 1 FROM user_indexes i WHERE i.index_name = UPPER(c.object_name)) OR EXISTS (SELECT 1 FROM user_tables t WHERE t.table_name = UPPER(c.object_name))) ORDER BY c.label, c.properties", - "schema.createRange": "DECLARE\n e_exists EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_exists, -955);\nBEGIN\n EXECUTE IMMEDIATE 'CREATE INDEX IF NOT EXISTS drivine_ix_lastActivityAtsessionId_8a35af29 ON drivine_node (JSON_VALUE(props, ''$.lastActivityAt''), JSON_VALUE(props, ''$.sessionId''))';\nEXCEPTION\n WHEN e_exists THEN NULL;\nEND;\nDECLARE\n v_unique NUMBER;\n e_exists EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_exists, -955);\nBEGIN\n SELECT COUNT(*) INTO v_unique FROM drivine_schema u WHERE u.label = 'Session' AND u.properties = '[\"lastActivityAt\",\"sessionId\"]' AND u.kind IN ('UNIQUENESS_CONSTRAINT', 'UNIQUE') AND EXISTS (SELECT 1 FROM user_indexes q WHERE q.index_name = UPPER(u.object_name));\n IF v_unique = 0 THEN\n EXECUTE IMMEDIATE 'CREATE INDEX IF NOT EXISTS drivine_lx_SessionlastActivityAtsessionId_396a2573 ON drivine_node n (CASE WHEN JSON_EXISTS(n.labels, ''$?(@ == \"Session\")'') THEN JSON_VALUE(n.props, ''$.\"lastActivityAt\"'' RETURNING VARCHAR2(4000) NULL ON ERROR) END, CASE WHEN JSON_EXISTS(n.labels, ''$?(@ == \"Session\")'') THEN JSON_VALUE(n.props, ''$.\"sessionId\"'' RETURNING VARCHAR2(4000) NULL ON ERROR) END)';\n END IF;\nEXCEPTION\n WHEN e_exists THEN NULL;\nEND;\nBEGIN\n MERGE INTO drivine_schema c\n USING (SELECT 'RANGE_INDEX' AS kind, 'Session' AS label, '[\"lastActivityAt\",\"sessionId\"]' AS properties FROM dual) k\n ON (c.kind = k.kind AND c.label = k.label AND c.properties = k.properties)\n WHEN MATCHED THEN UPDATE SET c.details = '{\"labelled\": \"drivine_lx_SessionlastActivityAtsessionId_396a2573\"}'\n WHEN NOT MATCHED THEN INSERT (kind, label, properties, name, object_name, details)\n VALUES (k.kind, k.label, k.properties, 'Session_lastActivityAt_sessionId_range', 'drivine_ix_lastActivityAtsessionId_8a35af29', '{\"labelled\": \"drivine_lx_SessionlastActivityAtsessionId_396a2573\"}');\nEXCEPTION\n WHEN DUP_VAL_ON_INDEX THEN NULL;\nEND;", + "schema.createRange": "DECLARE\n e_exists EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_exists, -955);\n e_busy EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_busy, -54);\nBEGIN\n FOR attempt IN 1 .. 300 LOOP\n BEGIN\n EXECUTE IMMEDIATE 'CREATE INDEX IF NOT EXISTS drivine_ix_lastActivityAtsessionId_8a35af29 ON drivine_node (JSON_VALUE(props, ''$.lastActivityAt''), JSON_VALUE(props, ''$.sessionId''))';\n RETURN;\n EXCEPTION\n WHEN e_exists THEN RETURN;\n WHEN e_busy THEN\n IF attempt = 300 THEN RAISE; END IF;\n DBMS_SESSION.SLEEP(0.1);\n END;\n END LOOP;\nEND;\nDECLARE\n v_unique NUMBER;\n e_exists EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_exists, -955);\nBEGIN\n SELECT COUNT(*) INTO v_unique FROM drivine_schema u WHERE u.label = 'Session' AND u.properties = '[\"lastActivityAt\",\"sessionId\"]' AND u.kind IN ('UNIQUENESS_CONSTRAINT', 'UNIQUE') AND EXISTS (SELECT 1 FROM user_indexes q WHERE q.index_name = UPPER(u.object_name));\n IF v_unique = 0 THEN\n EXECUTE IMMEDIATE 'CREATE INDEX IF NOT EXISTS drivine_lx_SessionlastActivityAtsessionId_396a2573 ON drivine_node n (CASE WHEN JSON_EXISTS(n.labels, ''$?(@ == \"Session\")'') THEN JSON_VALUE(n.props, ''$.\"lastActivityAt\"'' RETURNING VARCHAR2(4000) NULL ON ERROR) END, CASE WHEN JSON_EXISTS(n.labels, ''$?(@ == \"Session\")'') THEN JSON_VALUE(n.props, ''$.\"sessionId\"'' RETURNING VARCHAR2(4000) NULL ON ERROR) END)';\n END IF;\nEXCEPTION\n WHEN e_exists THEN NULL;\nEND;\nBEGIN\n MERGE INTO drivine_schema c\n USING (SELECT 'RANGE_INDEX' AS kind, 'Session' AS label, '[\"lastActivityAt\",\"sessionId\"]' AS properties FROM dual) k\n ON (c.kind = k.kind AND c.label = k.label AND c.properties = k.properties)\n WHEN MATCHED THEN UPDATE SET c.details = '{\"labelled\": \"drivine_lx_SessionlastActivityAtsessionId_396a2573\"}'\n WHEN NOT MATCHED THEN INSERT (kind, label, properties, name, object_name, details)\n VALUES (k.kind, k.label, k.properties, 'Session_lastActivityAt_sessionId_range', 'drivine_ix_lastActivityAtsessionId_8a35af29', '{\"labelled\": \"drivine_lx_SessionlastActivityAtsessionId_396a2573\"}');\nEXCEPTION\n WHEN DUP_VAL_ON_INDEX THEN NULL;\nEND;", "schema.dropRange": "DECLARE\n v_others NUMBER;\nBEGIN\n FOR it IN (SELECT object_name, JSON_VALUE(details, '$.labelled') AS labelled FROM drivine_schema WHERE kind IN ('RANGE_INDEX', 'RANGE') AND label = 'Session' AND properties = '[\"lastActivityAt\",\"sessionId\"]') LOOP\n SELECT COUNT(*) INTO v_others FROM drivine_schema\n WHERE UPPER(object_name) = UPPER(it.object_name) AND NOT (kind IN ('RANGE_INDEX', 'RANGE') AND label = 'Session' AND properties = '[\"lastActivityAt\",\"sessionId\"]');\n IF it.labelled IS NOT NULL THEN\n EXECUTE IMMEDIATE 'DROP INDEX IF EXISTS ' || it.labelled;\n END IF;\n IF it.object_name IS NOT NULL AND v_others = 0 THEN\n EXECUTE IMMEDIATE 'DROP INDEX IF EXISTS ' || it.object_name;\n END IF;\n END LOOP;\n DELETE FROM drivine_schema WHERE kind IN ('RANGE_INDEX', 'RANGE') AND label = 'Session' AND properties = '[\"lastActivityAt\",\"sessionId\"]';\n COMMIT;\nEND;", - "schema.createUnique": "DECLARE\n e_exists EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_exists, -955);\nBEGIN\n EXECUTE IMMEDIATE 'CREATE INDEX IF NOT EXISTS drivine_ix_uuid_36f3bb ON drivine_node (JSON_VALUE(props, ''$.uuid''))';\nEXCEPTION\n WHEN e_exists THEN NULL;\nEND;\nBEGIN\n MERGE INTO drivine_schema c\n USING (SELECT 'RANGE_INDEX' AS kind, 'Person' AS label, '[\"uuid\"]' AS properties FROM dual) k\n ON (c.kind = k.kind AND c.label = k.label AND c.properties = k.properties)\n WHEN NOT MATCHED THEN INSERT (kind, label, properties, name, object_name, details)\n VALUES (k.kind, k.label, k.properties, 'Person_uuid_range', 'drivine_ix_uuid_36f3bb', '{\"unique\": \"drivine_uq_Personuuid_44d1cd66\"}');\nEXCEPTION\n WHEN DUP_VAL_ON_INDEX THEN NULL;\nEND;\nDECLARE\n v_labelled VARCHAR2(128);\nBEGIN\n SELECT MAX(JSON_VALUE(details, '$.labelled')) INTO v_labelled FROM drivine_schema WHERE kind = 'RANGE_INDEX' AND label = 'Person' AND properties = '[\"uuid\"]';\n IF v_labelled IS NOT NULL THEN\n EXECUTE IMMEDIATE 'DROP INDEX IF EXISTS ' || v_labelled;\n END IF;\n FOR it IN (SELECT object_name FROM drivine_schema WHERE kind = 'RANGE' AND label = 'Person' AND properties = '[\"uuid\"]') LOOP\n EXECUTE IMMEDIATE 'DROP INDEX IF EXISTS ' || it.object_name;\n END LOOP;\n DELETE FROM drivine_schema WHERE kind = 'RANGE' AND label = 'Person' AND properties = '[\"uuid\"]';\n UPDATE drivine_schema SET details = '{\"unique\": \"drivine_uq_Personuuid_44d1cd66\"}' WHERE kind = 'RANGE_INDEX' AND label = 'Person' AND properties = '[\"uuid\"]';\n COMMIT;\nEND;\nDECLARE\n e_exists EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_exists, -955);\nBEGIN\n EXECUTE IMMEDIATE 'CREATE UNIQUE INDEX IF NOT EXISTS drivine_uq_Personuuid_44d1cd66 ON drivine_node n (CASE WHEN JSON_EXISTS(n.labels, ''$?(@ == \"Person\")'') THEN JSON_VALUE(n.props, ''$.\"uuid\"'' RETURNING VARCHAR2(4000) NULL ON ERROR) END)';\nEXCEPTION\n WHEN e_exists THEN NULL;\nEND;\nBEGIN\n MERGE INTO drivine_schema c\n USING (SELECT 'UNIQUENESS_CONSTRAINT' AS kind, 'Person' AS label, '[\"uuid\"]' AS properties FROM dual) k\n ON (c.kind = k.kind AND c.label = k.label AND c.properties = k.properties)\n WHEN NOT MATCHED THEN INSERT (kind, label, properties, name, object_name, details)\n VALUES (k.kind, k.label, k.properties, 'Person_uuid_unique', 'drivine_uq_Personuuid_44d1cd66', '{}');\nEXCEPTION\n WHEN DUP_VAL_ON_INDEX THEN NULL;\nEND;", - "schema.createUnique.pair": "DECLARE\n e_exists EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_exists, -955);\nBEGIN\n EXECUTE IMMEDIATE 'CREATE INDEX IF NOT EXISTS drivine_ix_emailtenant_6d0f958e ON drivine_node (JSON_VALUE(props, ''$.email''), JSON_VALUE(props, ''$.tenant''))';\nEXCEPTION\n WHEN e_exists THEN NULL;\nEND;\nBEGIN\n MERGE INTO drivine_schema c\n USING (SELECT 'RANGE_INDEX' AS kind, 'Person' AS label, '[\"email\",\"tenant\"]' AS properties FROM dual) k\n ON (c.kind = k.kind AND c.label = k.label AND c.properties = k.properties)\n WHEN NOT MATCHED THEN INSERT (kind, label, properties, name, object_name, details)\n VALUES (k.kind, k.label, k.properties, 'Person_email_tenant_range', 'drivine_ix_emailtenant_6d0f958e', '{\"unique\": \"drivine_uq_Personemailtenant_83779439\"}');\nEXCEPTION\n WHEN DUP_VAL_ON_INDEX THEN NULL;\nEND;\nDECLARE\n v_labelled VARCHAR2(128);\nBEGIN\n SELECT MAX(JSON_VALUE(details, '$.labelled')) INTO v_labelled FROM drivine_schema WHERE kind = 'RANGE_INDEX' AND label = 'Person' AND properties = '[\"email\",\"tenant\"]';\n IF v_labelled IS NOT NULL THEN\n EXECUTE IMMEDIATE 'DROP INDEX IF EXISTS ' || v_labelled;\n END IF;\n FOR it IN (SELECT object_name FROM drivine_schema WHERE kind = 'RANGE' AND label = 'Person' AND properties = '[\"email\",\"tenant\"]') LOOP\n EXECUTE IMMEDIATE 'DROP INDEX IF EXISTS ' || it.object_name;\n END LOOP;\n DELETE FROM drivine_schema WHERE kind = 'RANGE' AND label = 'Person' AND properties = '[\"email\",\"tenant\"]';\n UPDATE drivine_schema SET details = '{\"unique\": \"drivine_uq_Personemailtenant_83779439\"}' WHERE kind = 'RANGE_INDEX' AND label = 'Person' AND properties = '[\"email\",\"tenant\"]';\n COMMIT;\nEND;\nDECLARE\n e_exists EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_exists, -955);\nBEGIN\n EXECUTE IMMEDIATE 'CREATE UNIQUE INDEX IF NOT EXISTS drivine_uq_Personemailtenant_83779439 ON drivine_node n (CASE WHEN JSON_EXISTS(n.labels, ''$?(@ == \"Person\")'') THEN JSON_VALUE(n.props, ''$.\"email\"'' RETURNING VARCHAR2(4000) NULL ON ERROR) END, CASE WHEN JSON_EXISTS(n.labels, ''$?(@ == \"Person\")'') THEN JSON_VALUE(n.props, ''$.\"tenant\"'' RETURNING VARCHAR2(4000) NULL ON ERROR) END)';\nEXCEPTION\n WHEN e_exists THEN NULL;\nEND;\nBEGIN\n MERGE INTO drivine_schema c\n USING (SELECT 'UNIQUENESS_CONSTRAINT' AS kind, 'Person' AS label, '[\"email\",\"tenant\"]' AS properties FROM dual) k\n ON (c.kind = k.kind AND c.label = k.label AND c.properties = k.properties)\n WHEN NOT MATCHED THEN INSERT (kind, label, properties, name, object_name, details)\n VALUES (k.kind, k.label, k.properties, 'Person_email_tenant_unique', 'drivine_uq_Personemailtenant_83779439', '{}');\nEXCEPTION\n WHEN DUP_VAL_ON_INDEX THEN NULL;\nEND;", + "schema.createUnique": "DECLARE\n e_exists EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_exists, -955);\n e_busy EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_busy, -54);\nBEGIN\n FOR attempt IN 1 .. 300 LOOP\n BEGIN\n EXECUTE IMMEDIATE 'CREATE INDEX IF NOT EXISTS drivine_ix_uuid_36f3bb ON drivine_node (JSON_VALUE(props, ''$.uuid''))';\n RETURN;\n EXCEPTION\n WHEN e_exists THEN RETURN;\n WHEN e_busy THEN\n IF attempt = 300 THEN RAISE; END IF;\n DBMS_SESSION.SLEEP(0.1);\n END;\n END LOOP;\nEND;\nBEGIN\n MERGE INTO drivine_schema c\n USING (SELECT 'RANGE_INDEX' AS kind, 'Person' AS label, '[\"uuid\"]' AS properties FROM dual) k\n ON (c.kind = k.kind AND c.label = k.label AND c.properties = k.properties)\n WHEN NOT MATCHED THEN INSERT (kind, label, properties, name, object_name, details)\n VALUES (k.kind, k.label, k.properties, 'Person_uuid_range', 'drivine_ix_uuid_36f3bb', '{\"unique\": \"drivine_uq_Personuuid_44d1cd66\"}');\nEXCEPTION\n WHEN DUP_VAL_ON_INDEX THEN NULL;\nEND;\nDECLARE\n v_labelled VARCHAR2(128);\nBEGIN\n SELECT MAX(JSON_VALUE(details, '$.labelled')) INTO v_labelled FROM drivine_schema WHERE kind = 'RANGE_INDEX' AND label = 'Person' AND properties = '[\"uuid\"]';\n IF v_labelled IS NOT NULL THEN\n EXECUTE IMMEDIATE 'DROP INDEX IF EXISTS ' || v_labelled;\n END IF;\n FOR it IN (SELECT object_name FROM drivine_schema WHERE kind = 'RANGE' AND label = 'Person' AND properties = '[\"uuid\"]') LOOP\n EXECUTE IMMEDIATE 'DROP INDEX IF EXISTS ' || it.object_name;\n END LOOP;\n DELETE FROM drivine_schema WHERE kind = 'RANGE' AND label = 'Person' AND properties = '[\"uuid\"]';\n UPDATE drivine_schema SET details = '{\"unique\": \"drivine_uq_Personuuid_44d1cd66\"}' WHERE kind = 'RANGE_INDEX' AND label = 'Person' AND properties = '[\"uuid\"]';\n COMMIT;\nEND;\nDECLARE\n e_exists EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_exists, -955);\n e_busy EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_busy, -54);\nBEGIN\n FOR attempt IN 1 .. 300 LOOP\n BEGIN\n EXECUTE IMMEDIATE 'CREATE UNIQUE INDEX IF NOT EXISTS drivine_uq_Personuuid_44d1cd66 ON drivine_node n (CASE WHEN JSON_EXISTS(n.labels, ''$?(@ == \"Person\")'') THEN JSON_VALUE(n.props, ''$.\"uuid\"'' RETURNING VARCHAR2(4000) NULL ON ERROR) END)';\n RETURN;\n EXCEPTION\n WHEN e_exists THEN RETURN;\n WHEN e_busy THEN\n IF attempt = 300 THEN RAISE; END IF;\n DBMS_SESSION.SLEEP(0.1);\n END;\n END LOOP;\nEND;\nBEGIN\n MERGE INTO drivine_schema c\n USING (SELECT 'UNIQUENESS_CONSTRAINT' AS kind, 'Person' AS label, '[\"uuid\"]' AS properties FROM dual) k\n ON (c.kind = k.kind AND c.label = k.label AND c.properties = k.properties)\n WHEN NOT MATCHED THEN INSERT (kind, label, properties, name, object_name, details)\n VALUES (k.kind, k.label, k.properties, 'Person_uuid_unique', 'drivine_uq_Personuuid_44d1cd66', '{}');\nEXCEPTION\n WHEN DUP_VAL_ON_INDEX THEN NULL;\nEND;", + "schema.createUnique.pair": "DECLARE\n e_exists EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_exists, -955);\n e_busy EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_busy, -54);\nBEGIN\n FOR attempt IN 1 .. 300 LOOP\n BEGIN\n EXECUTE IMMEDIATE 'CREATE INDEX IF NOT EXISTS drivine_ix_emailtenant_6d0f958e ON drivine_node (JSON_VALUE(props, ''$.email''), JSON_VALUE(props, ''$.tenant''))';\n RETURN;\n EXCEPTION\n WHEN e_exists THEN RETURN;\n WHEN e_busy THEN\n IF attempt = 300 THEN RAISE; END IF;\n DBMS_SESSION.SLEEP(0.1);\n END;\n END LOOP;\nEND;\nBEGIN\n MERGE INTO drivine_schema c\n USING (SELECT 'RANGE_INDEX' AS kind, 'Person' AS label, '[\"email\",\"tenant\"]' AS properties FROM dual) k\n ON (c.kind = k.kind AND c.label = k.label AND c.properties = k.properties)\n WHEN NOT MATCHED THEN INSERT (kind, label, properties, name, object_name, details)\n VALUES (k.kind, k.label, k.properties, 'Person_email_tenant_range', 'drivine_ix_emailtenant_6d0f958e', '{\"unique\": \"drivine_uq_Personemailtenant_83779439\"}');\nEXCEPTION\n WHEN DUP_VAL_ON_INDEX THEN NULL;\nEND;\nDECLARE\n v_labelled VARCHAR2(128);\nBEGIN\n SELECT MAX(JSON_VALUE(details, '$.labelled')) INTO v_labelled FROM drivine_schema WHERE kind = 'RANGE_INDEX' AND label = 'Person' AND properties = '[\"email\",\"tenant\"]';\n IF v_labelled IS NOT NULL THEN\n EXECUTE IMMEDIATE 'DROP INDEX IF EXISTS ' || v_labelled;\n END IF;\n FOR it IN (SELECT object_name FROM drivine_schema WHERE kind = 'RANGE' AND label = 'Person' AND properties = '[\"email\",\"tenant\"]') LOOP\n EXECUTE IMMEDIATE 'DROP INDEX IF EXISTS ' || it.object_name;\n END LOOP;\n DELETE FROM drivine_schema WHERE kind = 'RANGE' AND label = 'Person' AND properties = '[\"email\",\"tenant\"]';\n UPDATE drivine_schema SET details = '{\"unique\": \"drivine_uq_Personemailtenant_83779439\"}' WHERE kind = 'RANGE_INDEX' AND label = 'Person' AND properties = '[\"email\",\"tenant\"]';\n COMMIT;\nEND;\nDECLARE\n e_exists EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_exists, -955);\n e_busy EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_busy, -54);\nBEGIN\n FOR attempt IN 1 .. 300 LOOP\n BEGIN\n EXECUTE IMMEDIATE 'CREATE UNIQUE INDEX IF NOT EXISTS drivine_uq_Personemailtenant_83779439 ON drivine_node n (CASE WHEN JSON_EXISTS(n.labels, ''$?(@ == \"Person\")'') THEN JSON_VALUE(n.props, ''$.\"email\"'' RETURNING VARCHAR2(4000) NULL ON ERROR) END, CASE WHEN JSON_EXISTS(n.labels, ''$?(@ == \"Person\")'') THEN JSON_VALUE(n.props, ''$.\"tenant\"'' RETURNING VARCHAR2(4000) NULL ON ERROR) END)';\n RETURN;\n EXCEPTION\n WHEN e_exists THEN RETURN;\n WHEN e_busy THEN\n IF attempt = 300 THEN RAISE; END IF;\n DBMS_SESSION.SLEEP(0.1);\n END;\n END LOOP;\nEND;\nBEGIN\n MERGE INTO drivine_schema c\n USING (SELECT 'UNIQUENESS_CONSTRAINT' AS kind, 'Person' AS label, '[\"email\",\"tenant\"]' AS properties FROM dual) k\n ON (c.kind = k.kind AND c.label = k.label AND c.properties = k.properties)\n WHEN NOT MATCHED THEN INSERT (kind, label, properties, name, object_name, details)\n VALUES (k.kind, k.label, k.properties, 'Person_email_tenant_unique', 'drivine_uq_Personemailtenant_83779439', '{}');\nEXCEPTION\n WHEN DUP_VAL_ON_INDEX THEN NULL;\nEND;", "schema.dropUnique": "DECLARE\n v_others NUMBER;\nBEGIN\n FOR it IN (SELECT object_name, JSON_VALUE(details, '$.labelled') AS labelled FROM drivine_schema WHERE kind IN ('UNIQUENESS_CONSTRAINT', 'UNIQUE') AND label = 'Person' AND properties = '[\"email\",\"tenant\"]') LOOP\n SELECT COUNT(*) INTO v_others FROM drivine_schema\n WHERE UPPER(object_name) = UPPER(it.object_name) AND NOT (kind IN ('UNIQUENESS_CONSTRAINT', 'UNIQUE') AND label = 'Person' AND properties = '[\"email\",\"tenant\"]');\n IF it.object_name IS NOT NULL AND v_others = 0 THEN\n EXECUTE IMMEDIATE 'DROP INDEX IF EXISTS ' || it.object_name;\n END IF;\n END LOOP;\n DELETE FROM drivine_schema WHERE kind IN ('UNIQUENESS_CONSTRAINT', 'UNIQUE') AND label = 'Person' AND properties = '[\"email\",\"tenant\"]';\n COMMIT;\nEND;", "schema.createVector": "CREATE TABLE IF NOT EXISTS drivine_v_Chunkembedding_7e60bd64 (nid NUMBER PRIMARY KEY REFERENCES drivine_node (nid) ON DELETE CASCADE, embedding VECTOR(1536, FLOAT32) NOT NULL)\nBEGIN EXECUTE IMMEDIATE 'CREATE OR REPLACE TRIGGER drivine_v_Chunkembedding_7e60bd64_t AFTER INSERT OR UPDATE OF labels, props ON drivine_node FOR EACH ROW BEGIN IF JSON_EXISTS(:NEW.labels, ''$?(@ == \"Chunk\")'') AND JSON_EXISTS(:NEW.props, ''$.\"embedding\"'') THEN BEGIN MERGE INTO drivine_v_Chunkembedding_7e60bd64 s USING (SELECT :NEW.nid AS nid, TO_VECTOR(JSON_QUERY(:NEW.props, ''$.\"embedding\"'' RETURNING CLOB)) AS embedding FROM dual) n ON (s.nid = n.nid) WHEN MATCHED THEN UPDATE SET s.embedding = n.embedding WHEN NOT MATCHED THEN INSERT (nid, embedding) VALUES (n.nid, n.embedding); EXCEPTION WHEN OTHERS THEN DELETE FROM drivine_v_Chunkembedding_7e60bd64 WHERE nid = :NEW.nid; END; ELSE DELETE FROM drivine_v_Chunkembedding_7e60bd64 WHERE nid = :NEW.nid; END IF; END;'; END;\nINSERT INTO drivine_v_Chunkembedding_7e60bd64 (nid, embedding) SELECT n.nid, TO_VECTOR(JSON_QUERY(n.props, '$.\"embedding\"' RETURNING CLOB)) FROM drivine_node n WHERE JSON_EXISTS(n.labels, '$?(@ == \"Chunk\")') AND JSON_EXISTS(n.props, '$.\"embedding\"') AND VECTOR_DIMENSION_COUNT(TO_VECTOR(JSON_QUERY(n.props, '$.\"embedding\"' RETURNING CLOB))) = 1536 AND NOT EXISTS (SELECT 1 FROM drivine_v_Chunkembedding_7e60bd64 s WHERE s.nid = n.nid)\nCREATE VECTOR INDEX IF NOT EXISTS drivine_v_Chunkembedding_7e60bd64_i ON drivine_v_Chunkembedding_7e60bd64 (embedding) ORGANIZATION NEIGHBOR PARTITIONS DISTANCE COSINE\nBEGIN\n MERGE INTO drivine_schema c\n USING (SELECT 'VECTOR_INDEX' AS kind, 'Chunk' AS label, '[\"embedding\"]' AS properties FROM dual) k\n ON (c.kind = k.kind AND c.label = k.label AND c.properties = k.properties)\n WHEN NOT MATCHED THEN INSERT (kind, label, properties, name, object_name, details)\n VALUES (k.kind, k.label, k.properties, 'Chunk_embedding_vector', 'drivine_v_Chunkembedding_7e60bd64', '{\"dimensions\": 1536, \"similarity\": \"COSINE\"}');\nEXCEPTION\n WHEN DUP_VAL_ON_INDEX THEN NULL;\nEND;", "schema.createVector.plain": "CREATE TABLE IF NOT EXISTS drivine_v_Chunksummary_1a6b3cf3 (nid NUMBER PRIMARY KEY REFERENCES drivine_node (nid) ON DELETE CASCADE, embedding VECTOR(384, FLOAT32) NOT NULL)\nBEGIN EXECUTE IMMEDIATE 'CREATE OR REPLACE TRIGGER drivine_v_Chunksummary_1a6b3cf3_t AFTER INSERT OR UPDATE OF labels, props ON drivine_node FOR EACH ROW BEGIN IF JSON_EXISTS(:NEW.labels, ''$?(@ == \"Chunk\")'') AND JSON_EXISTS(:NEW.props, ''$.\"summary\"'') THEN BEGIN MERGE INTO drivine_v_Chunksummary_1a6b3cf3 s USING (SELECT :NEW.nid AS nid, TO_VECTOR(JSON_QUERY(:NEW.props, ''$.\"summary\"'' RETURNING CLOB)) AS embedding FROM dual) n ON (s.nid = n.nid) WHEN MATCHED THEN UPDATE SET s.embedding = n.embedding WHEN NOT MATCHED THEN INSERT (nid, embedding) VALUES (n.nid, n.embedding); EXCEPTION WHEN OTHERS THEN DELETE FROM drivine_v_Chunksummary_1a6b3cf3 WHERE nid = :NEW.nid; END; ELSE DELETE FROM drivine_v_Chunksummary_1a6b3cf3 WHERE nid = :NEW.nid; END IF; END;'; END;\nINSERT INTO drivine_v_Chunksummary_1a6b3cf3 (nid, embedding) SELECT n.nid, TO_VECTOR(JSON_QUERY(n.props, '$.\"summary\"' RETURNING CLOB)) FROM drivine_node n WHERE JSON_EXISTS(n.labels, '$?(@ == \"Chunk\")') AND JSON_EXISTS(n.props, '$.\"summary\"') AND VECTOR_DIMENSION_COUNT(TO_VECTOR(JSON_QUERY(n.props, '$.\"summary\"' RETURNING CLOB))) = 384 AND NOT EXISTS (SELECT 1 FROM drivine_v_Chunksummary_1a6b3cf3 s WHERE s.nid = n.nid)\nCREATE VECTOR INDEX IF NOT EXISTS drivine_v_Chunksummary_1a6b3cf3_i ON drivine_v_Chunksummary_1a6b3cf3 (embedding) ORGANIZATION NEIGHBOR PARTITIONS DISTANCE EUCLIDEAN_SQUARED\nBEGIN\n MERGE INTO drivine_schema c\n USING (SELECT 'VECTOR_INDEX' AS kind, 'Chunk' AS label, '[\"summary\"]' AS properties FROM dual) k\n ON (c.kind = k.kind AND c.label = k.label AND c.properties = k.properties)\n WHEN NOT MATCHED THEN INSERT (kind, label, properties, name, object_name, details)\n VALUES (k.kind, k.label, k.properties, 'Chunk_summary_vector', 'drivine_v_Chunksummary_1a6b3cf3', '{\"dimensions\": 384, \"similarity\": \"EUCLIDEAN\"}');\nEXCEPTION\n WHEN DUP_VAL_ON_INDEX THEN NULL;\nEND;", diff --git a/tests/unit/test_oracle_schema_grammar.py b/tests/unit/test_oracle_schema_grammar.py index 0e0cb1d..9d86da4 100644 --- a/tests/unit/test_oracle_schema_grammar.py +++ b/tests/unit/test_oracle_schema_grammar.py @@ -1,12 +1,13 @@ """Oracle's schema statements where the golden statements hold none of the case, the catalog's key, and its items.""" +import re from typing import Any import pytest -from drivine import range_index +from drivine import range_index, uniqueness_constraint from drivine.backend.oracle.oracle_graph import OracleGraph -from drivine.backend.oracle.oracle_schema_grammar import oracle_schema_grammar +from drivine.backend.oracle.oracle_schema_grammar import BUSY_ATTEMPTS, oracle_schema_grammar from drivine.backend.sql.sql_catalog import catalog_properties, catalog_record from drivine.connection.database_type import DatabaseType from drivine.drivine_error import DrivineError @@ -37,6 +38,23 @@ def test_quotes_a_key_drivine4j_quotes_one_a_json_path_reads_as_a_value_or_that_ assert "(JSON_VALUE(props, ''$.\"a\\\"b\"''))" in shared_index('a"b') +def test_runs_an_index_over_the_node_table_again_while_another_sessions_lock_turns_it_away_up_to_a_bound(): + # Managers making the store key's constraint together were turned away at once by each other's lock on the node + # table (ORA-00054), ran out of attempts, and went on without it. + unique = next( + it + for it in GRAMMAR.create_unique(uniqueness_constraint("DrivineStore", "key")) + if isinstance(it, str) and "CREATE UNIQUE INDEX" in it + ) + for block in (shared_index("name"), unique): + assert "PRAGMA EXCEPTION_INIT(e_busy, -54);" in block + assert f"FOR attempt IN 1 .. {BUSY_ATTEMPTS} LOOP" in block + assert re.search( + r"WHEN e_busy THEN\s+IF attempt = \d+ THEN RAISE; END IF;\s+DBMS_SESSION\.SLEEP\(0\.1\);", block + ) + assert re.search(r"EXECUTE IMMEDIATE 'CREATE[^;]*;\s+RETURN;", block) + + class _Failing: """A provider whose connections fail each statement with the next of `failures`, then return no rows.""" From c0c0d34d120650d7f08ce02f4963addba3982a25 Mon Sep 17 00:00:00 2001 From: jasper blues Date: Sun, 1 Nov 2026 10:40:18 +1000 Subject: [PATCH 2/2] Retry a block that waits out a busy lock only for a deadlock --- .../backend/oracle/oracle_schema_grammar.py | 14 +++++- drivine/schema/schema_executor.py | 16 ++++-- drivine/sync/schema/schema_executor.py | 16 ++++-- tests/unit/test_contention.py | 49 +++++++++++++++++++ 4 files changed, 84 insertions(+), 11 deletions(-) diff --git a/drivine/backend/oracle/oracle_schema_grammar.py b/drivine/backend/oracle/oracle_schema_grammar.py index 8f5244b..3ca05a0 100644 --- a/drivine/backend/oracle/oracle_schema_grammar.py +++ b/drivine/backend/oracle/oracle_schema_grammar.py @@ -58,6 +58,18 @@ def recorded_as(kind: CatalogKind) -> list[str]: a tenth of a second apart, before it raises: as drivine4j's `OracleGraph.BUSY_ATTEMPTS`. """ +_WAITS_OUT_BUSY = "PRAGMA EXCEPTION_INIT(e_busy, -54);" +"""What declares the busy lock `_made` waits out: a block that holds it has waited before it fails for that lock.""" + + +def waits_out_busy_lock(statement: str) -> bool: + """ + Whether `statement` waited out another session's busy lock (ORA-00054) itself before it failed for it, as a block of + `_made` does for thirty seconds: the schema executor doesn't run it again for that lock, or each run would wait as + long again. + """ + return _WAITS_OUT_BUSY in statement + def _made(ddl: str) -> str: """ @@ -71,7 +83,7 @@ def _made(ddl: str) -> str: e_exists EXCEPTION; PRAGMA EXCEPTION_INIT(e_exists, -955); e_busy EXCEPTION; - PRAGMA EXCEPTION_INIT(e_busy, -54); + {_WAITS_OUT_BUSY} BEGIN FOR attempt IN 1 .. {BUSY_ATTEMPTS} LOOP BEGIN diff --git a/drivine/schema/schema_executor.py b/drivine/schema/schema_executor.py index 99a6e8b..39d284d 100644 --- a/drivine/schema/schema_executor.py +++ b/drivine/schema/schema_executor.py @@ -5,24 +5,27 @@ from typing import Any from drivine.backend.cypher_grammar import FALKORDB_GRAMMAR, NEO4J_GRAMMAR +from drivine.backend.oracle.oracle_schema_grammar import waits_out_busy_lock from drivine.backend.postgres.postgres_statement import value_of from drivine.connection.connection_provider import ConnectionProvider from drivine.connection.database_type import DatabaseType from drivine.drivine_error import DrivineError -from drivine.manager.contention import any_cause, ddl_contended, when_not_contended +from drivine.manager.contention import any_cause, ddl_contended, ddl_deadlocked, when_not_contended from drivine.query.query_specification import QuerySpecification from drivine.schema.native_operation import NativeOperation _ORACLE_SERIAL = re.compile(r"ORA-08177\b") -def _contended(failure: BaseException) -> bool: +def _contended(failure: BaseException, waited_out_busy: bool) -> bool: """ A transient error over Bolt, FalkorDB turning a write away while another is in progress, an Oracle serialization failure, or a SQL store turning DDL away because of another session's (see ddl_contended), as when managers that - start together each make the indexes of the store's key. + start together each make the indexes of the store's key. A statement that `waited_out_busy`, as an Oracle index's + block does (see waits_out_busy_lock), is taken as turned away only for a deadlock (see ddl_deadlocked). """ - return ddl_contended(failure) or any_cause( + ddl = ddl_deadlocked(failure) if waited_out_busy else ddl_contended(failure) + return ddl or any_cause( failure, lambda it: ( NEO4J_GRAMMAR.contended(it) or FALKORDB_GRAMMAR.contended(it) or bool(_ORACLE_SERIAL.search(str(it))) @@ -42,7 +45,10 @@ async def run(self, statement: str | NativeOperation, bindings: Mapping[str, Any """ Runs a statement, again after a pause when the engine turned it away because another write was in progress. """ - return await when_not_contended(lambda: self._run_once(statement, bindings or {}), _contended) + waited_out_busy = isinstance(statement, str) and waits_out_busy_lock(statement) + return await when_not_contended( + lambda: self._run_once(statement, bindings or {}), lambda it: _contended(it, waited_out_busy) + ) async def _run_once(self, statement: str | NativeOperation, bindings: Mapping[str, Any]) -> list[Any]: if isinstance(statement, NativeOperation): diff --git a/drivine/sync/schema/schema_executor.py b/drivine/sync/schema/schema_executor.py index c060fc7..149ac82 100644 --- a/drivine/sync/schema/schema_executor.py +++ b/drivine/sync/schema/schema_executor.py @@ -7,24 +7,27 @@ from typing import Any from drivine.backend.cypher_grammar import FALKORDB_GRAMMAR, NEO4J_GRAMMAR +from drivine.backend.oracle.oracle_schema_grammar import waits_out_busy_lock from drivine.backend.postgres.postgres_statement import value_of from drivine.connection.database_type import DatabaseType from drivine.drivine_error import DrivineError from drivine.query.query_specification import QuerySpecification from drivine.schema.native_operation import NativeOperation from drivine.sync.connection.connection_provider import ConnectionProvider -from drivine.sync.manager.contention import any_cause, ddl_contended, when_not_contended +from drivine.sync.manager.contention import any_cause, ddl_contended, ddl_deadlocked, when_not_contended _ORACLE_SERIAL = re.compile(r"ORA-08177\b") -def _contended(failure: BaseException) -> bool: +def _contended(failure: BaseException, waited_out_busy: bool) -> bool: """ A transient error over Bolt, FalkorDB turning a write away while another is in progress, an Oracle serialization failure, or a SQL store turning DDL away because of another session's (see ddl_contended), as when managers that - start together each make the indexes of the store's key. + start together each make the indexes of the store's key. A statement that `waited_out_busy`, as an Oracle index's + block does (see waits_out_busy_lock), is taken as turned away only for a deadlock (see ddl_deadlocked). """ - return ddl_contended(failure) or any_cause( + ddl = ddl_deadlocked(failure) if waited_out_busy else ddl_contended(failure) + return ddl or any_cause( failure, lambda it: ( NEO4J_GRAMMAR.contended(it) or FALKORDB_GRAMMAR.contended(it) or bool(_ORACLE_SERIAL.search(str(it))) @@ -44,7 +47,10 @@ def run(self, statement: str | NativeOperation, bindings: Mapping[str, Any] | No """ Runs a statement, again after a pause when the engine turned it away because another write was in progress. """ - return when_not_contended(lambda: self._run_once(statement, bindings or {}), _contended) + waited_out_busy = isinstance(statement, str) and waits_out_busy_lock(statement) + return when_not_contended( + lambda: self._run_once(statement, bindings or {}), lambda it: _contended(it, waited_out_busy) + ) def _run_once(self, statement: str | NativeOperation, bindings: Mapping[str, Any]) -> list[Any]: if isinstance(statement, NativeOperation): diff --git a/tests/unit/test_contention.py b/tests/unit/test_contention.py index 58527f8..b01692e 100644 --- a/tests/unit/test_contention.py +++ b/tests/unit/test_contention.py @@ -10,10 +10,13 @@ from drivine.backend.cypher_backend import CypherBackend from drivine.backend.cypher_grammar import FALKORDB_GRAMMAR, MEMGRAPH_GRAMMAR, NEO4J_GRAMMAR +from drivine.backend.oracle.oracle_graph import OracleGraph +from drivine.backend.oracle.oracle_schema_grammar import oracle_schema_grammar from drivine.connection.database_type import DatabaseType from drivine.drivine_error import DrivineError from drivine.manager.contention import CONTENDED_ATTEMPTS, ddl_contended, ddl_deadlocked, when_not_contended from drivine.schema.schema_executor import SchemaExecutor +from drivine.schema.schema_specs import uniqueness_constraint def bolt(code: str) -> Neo4jError: @@ -191,6 +194,52 @@ async def connect(self) -> Connection: assert ran == runs +@pytest.mark.parametrize( + ("message", "runs"), + [ + ("ORA-00054: resource busy and acquire with NOWAIT specified or timeout expired", 1), + ("ORA-04020: deadlock detected while trying to lock object JVM_NODE", 2), + ], +) +async def test_an_oracle_indexs_block_that_waited_out_a_busy_lock_itself_is_not_run_again_for_it_but_is_for_a_deadlock( + message: str, runs: int +): + # An Oracle index's block has waited thirty seconds on the busy lock before it gives up: run again, as other DDL + # is, it would wait as long again each time, three minutes in all. + block = next( + it + for it in oracle_schema_grammar(OracleGraph("jvm")).create_unique(uniqueness_constraint("DrivineStore", "key")) + if isinstance(it, str) and "CREATE UNIQUE INDEX" in it + ) + ran = 0 + + class Connection: + async def query(self, _spec: object) -> list[object]: + nonlocal ran + ran += 1 + if ran == 1: + raise oracledb.DatabaseError(message) + return [] + + async def release(self, _error: BaseException | None = None) -> None: + return None + + class Provider: + name = "oracle" + type = DatabaseType.ORACLE + + async def connect(self) -> Connection: + return Connection() + + run = SchemaExecutor(Provider()).run(block) # pyright: ignore[reportArgumentType] + if runs == 2: + assert await run == [] + else: + with pytest.raises(DrivineError, match="ORA-00054"): + await run + assert ran == runs + + @pytest.mark.parametrize( "failure", [