diff --git a/arborist/qa/canonical_cache.py b/arborist/qa/canonical_cache.py index f5ed76b..96736cb 100644 --- a/arborist/qa/canonical_cache.py +++ b/arborist/qa/canonical_cache.py @@ -248,6 +248,10 @@ def persist_canonical( ts=now, ) qa_conn.execute( + # ON CONFLICT(cache_key) DO NOTHING: canonical-projection answers + # are deterministic, so a concurrent peer persisting the same + # cache_key writes an identical row — the loser no-ops instead of + # raising UNIQUE constraint failed. "INSERT INTO providence_cache " "(cache_key, source_root, document_uri, question_hash, question_text, " " answer_text, merkle_proof, model_profile_hash, conversation_hash, " @@ -256,7 +260,8 @@ def persist_canonical( " created_at, hit_count, audit_mode, n_quotes, n_verified, " " unverified_quotes, verifier_method, run_dag_root, run_dag_blob) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'live', ?, ?, ?, 0, " - " ?, ?, ?, ?, ?, ?, ?)", + " ?, ?, ?, ?, ?, ?, ?) " + "ON CONFLICT(cache_key) DO NOTHING", ( cache_key_value, src, diff --git a/arborist/qa/query.py b/arborist/qa/query.py index d957e83..5eabb89 100644 --- a/arborist/qa/query.py +++ b/arborist/qa/query.py @@ -3403,6 +3403,10 @@ def query( ts=now, ) qa_conn.execute( + # ON CONFLICT(cache_key) DO NOTHING: the cache lookup runs + # outside this transaction, so two concurrent queries on the + # same cache_key can both miss and both reach this INSERT — the + # loser no-ops instead of raising UNIQUE constraint failed. "INSERT INTO providence_cache " "(cache_key, source_root, document_uri, question_hash, question_text, " " answer_text, merkle_proof, model_profile_hash, conversation_hash, " @@ -3411,7 +3415,8 @@ def query( " created_at, hit_count, audit_mode, n_quotes, n_verified, " " unverified_quotes, verifier_method, run_dag_root, run_dag_blob) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'live', ?, ?, ?, 0, " - " ?, ?, ?, ?, ?, ?, ?)", + " ?, ?, ?, ?, ?, ?, ?) " + "ON CONFLICT(cache_key) DO NOTHING", ( ckey, context_root, diff --git a/arborist/qa/runner.py b/arborist/qa/runner.py index 961d420..58d4c77 100644 --- a/arborist/qa/runner.py +++ b/arborist/qa/runner.py @@ -1128,6 +1128,11 @@ def ask( ts=now, ) conn.execute( + # ON CONFLICT(cache_key) DO NOTHING: the cache lookup above runs + # outside this transaction, so two concurrent ask()s on the same + # cache_key can both miss and both reach this INSERT — the loser + # no-ops instead of raising UNIQUE constraint failed (its answer is + # equivalent: same question/model/policy ⇒ same cache_key). "INSERT INTO providence_cache " "(cache_key, source_root, document_uri, question_hash, question_text, " " answer_text, merkle_proof, model_profile_hash, conversation_hash, " @@ -1136,7 +1141,8 @@ def ask( " created_at, hit_count, audit_mode, n_quotes, n_verified, " " unverified_quotes, verifier_method, run_dag_root, run_dag_blob) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'live', ?, ?, ?, 0, " - " ?, ?, ?, ?, ?, ?, ?)", + " ?, ?, ?, ?, ?, ?, ?) " + "ON CONFLICT(cache_key) DO NOTHING", ( ckey, document_root, diff --git a/tests/test_qa.py b/tests/test_qa.py index 681d6f1..171d111 100644 --- a/tests/test_qa.py +++ b/tests/test_qa.py @@ -233,3 +233,61 @@ def test_cache_key_is_pure_function(): "|".join([src_root, qh, mh, ch, gh, "v1", "v1", "v1"]).encode() ).hexdigest() assert k1 == expected + + +def test_concurrent_ask_same_cache_key_no_unique_crash(tmp_path): + """Two threads run ``ask()`` on the same question/model concurrently. + + The cache lookup happens *outside* the write transaction, so both miss + and both reach the ``providence_cache`` INSERT — ``ON CONFLICT(cache_key) + DO NOTHING`` makes the loser no-op instead of raising + ``UNIQUE constraint failed: providence_cache.cache_key``. Exactly one + cache row lands. (Without the fix the losers raise; with ``busy_timeout`` + on every connection the loser waits on the writer's lock then no-ops.) + """ + import threading + + db = tmp_path / "race-qa.db" + conn = connect(db) + try: + root = _ingest_one(conn) + finally: + conn.close() + + n_threads = 6 + barrier = threading.Barrier(n_threads) + errors: list[BaseException] = [] + lock = threading.Lock() + + def worker() -> None: + try: + c = connect(db) + try: + barrier.wait(timeout=20) + ask( + c, + document_root=root, + question="What are the forms of capital?", + client=StubClient(answer="The forms include living and social."), + model_id="test-model", + ) + finally: + c.close() + except BaseException as exc: # noqa: BLE001 — surface it + with lock: + errors.append(exc) + + threads = [threading.Thread(target=worker) for _ in range(n_threads)] + for t in threads: + t.start() + for t in threads: + t.join(timeout=40) + + assert not errors, f"concurrent ask() raised: {errors!r}" + + conn = connect(db) + try: + n_rows = conn.execute("SELECT COUNT(*) FROM providence_cache").fetchone()[0] + finally: + conn.close() + assert n_rows == 1, f"expected exactly one providence_cache row, got {n_rows}"