From bbfd2ddc1744d4a7b5e5b0ec7e65de1bf5f57b49 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Mon, 11 May 2026 17:29:12 -0400 Subject: [PATCH] =?UTF-8?q?qa:=20providence=5Fcache=20INSERT=20=E2=80=94?= =?UTF-8?q?=20ON=20CONFLICT(cache=5Fkey)=20DO=20NOTHING?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The cache lookup in ask()/query()/canonical-persist runs outside the write transaction, so two concurrent callers on the same cache_key can both miss and both reach the providence_cache INSERT — the loser raised `UNIQUE constraint failed: providence_cache.cache_key` and ask() crashed (MOAD-0005 / TOCTOU; sibling of the af870bb append_audit fix). All three write sites (qa/runner.py, qa/query.py, qa/canonical_cache.py) now end in `ON CONFLICT(cache_key) DO NOTHING`, so the loser no-ops (its answer is equivalent — same question/model/policy ⇒ same cache_key; canonical answers are deterministic). With busy_timeout on every connection (af870bb) the loser waits on the writer's lock then no-ops. test_qa.py::test_concurrent_ask_same_cache_key_no_unique_crash — 6 threads run ask() on the same question concurrently; must not raise; exactly one cache row lands. Verified it fails without the fix (5 of 6 threads raise IntegrityError). --- arborist/qa/canonical_cache.py | 7 +++- arborist/qa/query.py | 7 +++- arborist/qa/runner.py | 8 ++++- tests/test_qa.py | 58 ++++++++++++++++++++++++++++++++++ 4 files changed, 77 insertions(+), 3 deletions(-) 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}"