From 890b520b1caa915a9a4a765414eef6d2766ebe0c Mon Sep 17 00:00:00 2001 From: mountain Date: Tue, 7 Jul 2026 18:23:01 +0800 Subject: [PATCH] fix(sqlite): make concurrent indexing writes robust (no "database is locked") MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Real concurrency e2e (8 threads adding the same file) surfaced a bug the mocked unit tests missed: concurrent add_document calls failed with sqlite3.OperationalError "database is locked". Root cause — under WAL the dedup SELECT (find_document_by_hash) left a read snapshot on the connection, and the subsequent INSERT on that stale snapshot raised SQLITE_BUSY_SNAPSHOT, which busy_timeout does not retry. Fixes: - open connections in autocommit (isolation_level=None) so a SELECT never leaves a lingering read snapshot and each write is its own transaction - PRAGMA busy_timeout=10000 so concurrent writers wait for the WAL single-writer lock instead of failing immediately - an instance-level write lock serializing the fast write methods within the process (the expensive LLM indexing stays parallel) Now 8 concurrent adds of one file -> a single doc_id, zero errors. Adds a real-thread regression test. Claude-Session: https://claude.ai/code/session_01Kx5DgKbhK1N8autqXH8SmS --- pageindex/storage/sqlite.py | 73 +++++++++++++++++++++++------------- tests/test_sqlite_storage.py | 27 +++++++++++++ 2 files changed, 73 insertions(+), 27 deletions(-) diff --git a/pageindex/storage/sqlite.py b/pageindex/storage/sqlite.py index ff4ab8c..2ed902f 100644 --- a/pageindex/storage/sqlite.py +++ b/pageindex/storage/sqlite.py @@ -11,6 +11,12 @@ class SQLiteStorage: self._local = threading.local() self._connections: list[sqlite3.Connection] = [] self._conn_lock = threading.Lock() + # Serializes the (fast) write operations within this process so + # concurrent indexing threads don't collide on WAL's single writer + # ("database is locked"). Reads stay concurrent; the expensive LLM + # indexing runs outside this lock. busy_timeout above covers the + # cross-process case. + self._write_lock = threading.Lock() self._init_schema() def _get_conn(self) -> sqlite3.Connection: @@ -21,9 +27,17 @@ class SQLiteStorage: # close() can close every tracked connection from whichever thread # calls it — with the default True those closes raise # ProgrammingError and the connections leak. - conn = sqlite3.connect(str(self._db_path), check_same_thread=False) + # isolation_level=None -> autocommit: a plain SELECT (e.g. the + # dedup hash lookup) never leaves a lingering read snapshot that a + # later write on the same connection would conflict with + # (SQLITE_BUSY_SNAPSHOT, which busy_timeout can't retry). Each + # statement is its own transaction, so busy_timeout can actually + # wait for the WAL single-writer lock under concurrency. + conn = sqlite3.connect(str(self._db_path), check_same_thread=False, + isolation_level=None) conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA foreign_keys=ON") + conn.execute("PRAGMA busy_timeout=10000") self._local.conn = conn with self._conn_lock: self._connections.append(conn) @@ -56,14 +70,16 @@ class SQLiteStorage: conn.commit() def create_collection(self, name: str) -> None: - conn = self._get_conn() - conn.execute("INSERT INTO collections (name) VALUES (?)", (name,)) - conn.commit() + with self._write_lock: + conn = self._get_conn() + conn.execute("INSERT INTO collections (name) VALUES (?)", (name,)) + conn.commit() def get_or_create_collection(self, name: str) -> None: - conn = self._get_conn() - conn.execute("INSERT OR IGNORE INTO collections (name) VALUES (?)", (name,)) - conn.commit() + with self._write_lock: + conn = self._get_conn() + conn.execute("INSERT OR IGNORE INTO collections (name) VALUES (?)", (name,)) + conn.commit() def list_collections(self) -> list[str]: conn = self._get_conn() @@ -71,25 +87,27 @@ class SQLiteStorage: return [r[0] for r in rows] def delete_collection(self, name: str) -> None: - conn = self._get_conn() - conn.execute("DELETE FROM collections WHERE name = ?", (name,)) - conn.commit() + with self._write_lock: + conn = self._get_conn() + conn.execute("DELETE FROM collections WHERE name = ?", (name,)) + conn.commit() def save_document(self, collection: str, doc_id: str, doc: dict) -> None: - conn = self._get_conn() # Plain INSERT (doc_id is a fresh uuid, never pre-existing). A duplicate # (collection_name, file_hash) raises sqlite3.IntegrityError, which the # caller uses to resolve a concurrent add-of-same-file race. - conn.execute( - """INSERT INTO documents - (doc_id, collection_name, doc_name, doc_description, file_path, file_hash, doc_type, structure, pages) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", - (doc_id, collection, doc.get("doc_name"), doc.get("doc_description"), - doc.get("file_path"), doc.get("file_hash"), doc["doc_type"], - json.dumps(doc.get("structure", [])), - json.dumps(doc.get("pages")) if doc.get("pages") else None), - ) - conn.commit() + with self._write_lock: + conn = self._get_conn() + conn.execute( + """INSERT INTO documents + (doc_id, collection_name, doc_name, doc_description, file_path, file_hash, doc_type, structure, pages) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", + (doc_id, collection, doc.get("doc_name"), doc.get("doc_description"), + doc.get("file_path"), doc.get("file_hash"), doc["doc_type"], + json.dumps(doc.get("structure", [])), + json.dumps(doc.get("pages")) if doc.get("pages") else None), + ) + conn.commit() def find_document_by_hash(self, collection: str, file_hash: str) -> str | None: conn = self._get_conn() @@ -140,12 +158,13 @@ class SQLiteStorage: return [{"doc_id": r[0], "doc_name": r[1], "doc_description": r[2] or "", "doc_type": r[3]} for r in rows] def delete_document(self, collection: str, doc_id: str) -> None: - conn = self._get_conn() - conn.execute( - "DELETE FROM documents WHERE doc_id = ? AND collection_name = ?", - (doc_id, collection), - ) - conn.commit() + with self._write_lock: + conn = self._get_conn() + conn.execute( + "DELETE FROM documents WHERE doc_id = ? AND collection_name = ?", + (doc_id, collection), + ) + conn.commit() def __enter__(self): return self diff --git a/tests/test_sqlite_storage.py b/tests/test_sqlite_storage.py index 751fcf7..c21da94 100644 --- a/tests/test_sqlite_storage.py +++ b/tests/test_sqlite_storage.py @@ -92,3 +92,30 @@ def test_duplicate_file_hash_in_collection_raises(storage): # same hash in a DIFFERENT collection is fine storage.create_collection("other") storage.save_document("other", "doc-3", {**doc}) + + +def test_concurrent_read_then_write_no_database_locked(storage): + """Regression: concurrent add (read hash -> write) hit 'database is locked' + under WAL. Fixed via autocommit + busy_timeout + write lock. All writers + must succeed (dedup via UNIQUE), none raise OperationalError.""" + import sqlite3, threading, uuid, time + storage.create_collection("c") + errs = [] + + def worker(): + try: + storage.list_collections() + storage.find_document_by_hash("c", "SAME") # read snapshot + time.sleep(0.001) # widen the window + try: + storage.save_document("c", str(uuid.uuid4()), + {"doc_name": "d", "doc_type": "pdf", "file_hash": "SAME", "structure": []}) + except sqlite3.IntegrityError: + pass # expected: lost the dedup race + except Exception as e: + errs.append(f"{type(e).__name__}: {e}") + + threads = [threading.Thread(target=worker) for _ in range(12)] + [t.start() for t in threads]; [t.join() for t in threads] + assert not errs, f"concurrent write errored: {errs}" + assert len(storage.list_documents("c")) == 1 # dedup held