From f99241ca88ebb2877a4b1b9350a282cba5f7cb68 Mon Sep 17 00:00:00 2001 From: KaifAhmad1 Date: Wed, 29 Jul 2026 12:54:43 +0530 Subject: [PATCH] fix(provenance): stop reads from taking the writer lock, fix batch count inflation Two review findings on #807/#812: - retrieve() and trace_lineage() were routed through transaction()'s BEGIN IMMEDIATE, so plain reads took SQLite's writer lock and serialized behind every other read/write, defeating the WAL concurrency this PR was meant to add. They now use a dedicated _read_connection() (configured, no explicit BEGIN). - track_entity()/track_chunk() swallowed all internal storage exceptions unconditionally, so a single item's failure inside track_entities_batch()/track_chunks_batch()'s shared transaction never reached the batch loop's per-item except, inflating tracked_count for entries that were never persisted. Both now re-raise when called with a shared _conn (batch context) while still degrading gracefully on standalone calls. Added regression tests for both, corrected the CHANGELOG entry and docs that described the prior (overly broad) behavior. --- CHANGELOG.md | 2 + docs/guides/provenance.md | 2 +- docs/reference/provenance.md | 2 +- semantica/provenance/manager.py | 15 +++- semantica/provenance/storage.py | 26 +++++- .../test_sqlite_storage_performance_807.py | 84 +++++++++++++++++++ 6 files changed, 123 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4ee1e02a..221f314c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -40,6 +40,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Implemented the `SQLiteStorage.transaction()` context manager with Write-Ahead Logging (`PRAGMA journal_mode=WAL`), `busy_timeout=5000`, `synchronous=NORMAL`, and immediate write transactions (`BEGIN IMMEDIATE`), ensuring concurrent read-modify-write sequences (including history version ID generation) are serialized without lock contention or data loss - Added block-level transaction sharing to `track_entities_batch()` and `track_chunks_batch()`, reducing SQLite commit overhead by ~99.9% for large batches and deferring `tracked_count` increments until successful commit so rolled-back items are never reported as successes - Preserved 100% backward compatibility for custom storage backends overriding `trace_lineage(self, entity_id)` by inspecting signatures dynamically before passing `max_depth`, and optimized BFS lineage queries with batched IN-clause lookups per frontier level + - **Follow-up fix**: `retrieve()` and `trace_lineage()` were initially routed through `transaction()` too, so plain reads took the same `BEGIN IMMEDIATE` writer lock as read-modify-write calls, serializing every read behind every other read/write and defeating the WAL concurrency this PR was meant to add. They now use a dedicated `_read_connection()` (configured, no explicit `BEGIN`) so reads no longer contend for the writer lock + - **Follow-up fix**: `track_entity()`/`track_chunk()` caught all internal storage exceptions unconditionally, so when called from `track_entities_batch()`/`track_chunks_batch()`'s shared per-block transaction, a single item's storage failure (e.g. non-JSON-serializable metadata) was swallowed inside the call and never surfaced to the batch loop's per-item `except`, inflating `tracked_count` for entries that were never persisted. Both methods now re-raise when invoked with a shared `_conn` (batch context) while still degrading gracefully on standalone calls, so batch counts match what's actually committed - Added 8 dedicated regression tests in `tests/provenance/test_sqlite_storage_performance_807.py` covering PRAGMA configuration, Windows unlink safety, batch transaction sharing, BFS `max_depth`, rollback count accuracy, custom storage backward compatibility, concurrent read-modify-write serialization, and connection cleanup guards on configuration error - **Explorer's Provenance UI used a naive 2-hop graph traversal instead of the audit-grade `ProvenanceManager` backend** (#792, #809) by @Sameer6305 diff --git a/docs/guides/provenance.md b/docs/guides/provenance.md index 16baa401..7b536b4e 100644 --- a/docs/guides/provenance.md +++ b/docs/guides/provenance.md @@ -84,7 +84,7 @@ prov = ProvenanceManager(storage=SQLiteStorage("audit.db")) For any regulated deployment — security operations, clinical data, financial risk — use `storage_path`. A SQLite file can be backed up, versioned, and queried with standard tools without requiring a server. - `SQLiteStorage` automatically configures Write-Ahead Logging (`WAL`), `busy_timeout=5000`, and `synchronous=NORMAL`, and executes operations in atomic immediate transactions (`BEGIN IMMEDIATE`). Furthermore, `ProvenanceManager` automatically supports custom storage backends overriding only `trace_lineage(self, entity_id)` without requiring `max_depth` in their signature. + `SQLiteStorage` automatically configures Write-Ahead Logging (`WAL`), `busy_timeout=5000`, and `synchronous=NORMAL`, and executes read-modify-write operations (like `track_entity()`) in atomic immediate transactions (`BEGIN IMMEDIATE`); plain reads (`retrieve()`, `trace_lineage()`) use a separate connection without an explicit write lock so they don't serialize behind writers. Furthermore, `ProvenanceManager` automatically supports custom storage backends overriding only `trace_lineage(self, entity_id)` without requiring `max_depth` in their signature. ## Recording provenance when ingesting data diff --git a/docs/reference/provenance.md b/docs/reference/provenance.md index 65f2c898..d162c680 100644 --- a/docs/reference/provenance.md +++ b/docs/reference/provenance.md @@ -324,7 +324,7 @@ manager = ProvenanceManager(storage_path="provenance.db") `SQLiteStorage` creates the database and indexes automatically on first use. -- **Atomicity & Concurrency**: Configures Write-Ahead Logging (`PRAGMA journal_mode=WAL`), `PRAGMA busy_timeout=5000`, and `PRAGMA synchronous=NORMAL`. Each public method call opens a single connection and executes inside an immediate write transaction (`BEGIN IMMEDIATE`), ensuring read-modify-write sequences are serialized across concurrent connections without leaving open file handles across calls. +- **Atomicity & Concurrency**: Configures Write-Ahead Logging (`PRAGMA journal_mode=WAL`), `PRAGMA busy_timeout=5000`, and `PRAGMA synchronous=NORMAL`. Read-modify-write methods (`track_entity()`, `store()`) open a single connection and execute inside an immediate write transaction (`BEGIN IMMEDIATE`), ensuring these sequences are serialized across concurrent connections without leaving open file handles across calls. Plain reads (`retrieve()`, `trace_lineage()`) use a separate connection with no explicit write lock, so concurrent reads don't serialize behind writers or each other. - **Backward Compatibility**: Custom storage subclasses overriding `trace_lineage(self, entity_id)` remain backward compatible; `ProvenanceManager` inspects the override signature and automatically calls it with one argument if `max_depth` is unsupported. ## Tamper-Evident Checksums diff --git a/semantica/provenance/manager.py b/semantica/provenance/manager.py index 19a6a5ec..3ac3036d 100644 --- a/semantica/provenance/manager.py +++ b/semantica/provenance/manager.py @@ -246,6 +246,11 @@ class ProvenanceManager: self.storage._store_with_conn(conn, entry) except Exception: + # When called from a batch's shared transaction (_conn is not None), + # propagate so the caller's per-item try/except can skip counting + # this item instead of reporting an unpersisted entry as tracked (#807). + if _conn is not None: + raise if entry is None: entry = ProvenanceEntry( entity_id=entity_id, @@ -262,7 +267,7 @@ class ProvenanceManager: used_entities=list(kwargs.get("used_entities", [])), ) entry.checksum = compute_checksum(entry) - + return entry def track_relationship( @@ -370,8 +375,12 @@ class ProvenanceManager: else: self.storage.store(entry) except Exception: - pass - + # Propagate when called from a batch's shared transaction so the + # caller's per-item try/except can skip counting this item instead + # of reporting an unpersisted entry as tracked (#807). + if _conn is not None: + raise + return entry # === Source Tracking (from conflicts.SourceTracker) === diff --git a/semantica/provenance/storage.py b/semantica/provenance/storage.py index d64608bb..5d9481d2 100644 --- a/semantica/provenance/storage.py +++ b/semantica/provenance/storage.py @@ -347,7 +347,27 @@ class SQLiteStorage(ProvenanceStorage): raise finally: conn.close() - + + @contextmanager + def _read_connection(self): + """ + Context manager providing a configured SQLite connection for read-only + operations, without taking SQLite's writer lock. + + `transaction()` uses BEGIN IMMEDIATE, which acquires the single RESERVED + (writer) lock so read-modify-write sequences serialize correctly. That + lock is unnecessary for plain reads (retrieve, trace_lineage) and, if + used there, would serialize every read behind every other read/write — + defeating WAL's readers-don't-block-writers concurrency. Reads use a + connection with no explicit BEGIN instead. + """ + conn = sqlite3.connect(self.db_path) + try: + self._configure_connection(conn) + yield conn + finally: + conn.close() + def _init_db(self) -> None: """Create tables with W3C PROV-O compliant schema.""" conn = sqlite3.connect(self.db_path) @@ -448,7 +468,7 @@ class SQLiteStorage(ProvenanceStorage): Returns: ProvenanceEntry if found, None otherwise """ - with self.transaction() as conn: + with self._read_connection() as conn: return self._retrieve_with_conn(conn, entity_id) def _retrieve_with_conn(self, conn: sqlite3.Connection, entity_id: str) -> Optional[ProvenanceEntry]: @@ -510,7 +530,7 @@ class SQLiteStorage(ProvenanceStorage): frontier = [entity_id] depth = 0 - with self.transaction() as conn: + with self._read_connection() as conn: cursor = conn.cursor() while frontier and (max_depth is None or depth < max_depth): # Remove duplicates while preserving order, and filter out already visited IDs diff --git a/tests/provenance/test_sqlite_storage_performance_807.py b/tests/provenance/test_sqlite_storage_performance_807.py index 05505eb0..d9af2bcf 100644 --- a/tests/provenance/test_sqlite_storage_performance_807.py +++ b/tests/provenance/test_sqlite_storage_performance_807.py @@ -203,3 +203,87 @@ def test_sqlite_storage_cleanup_guard_on_configure_error(tmp_path): assert os.path.exists(db_path) os.unlink(db_path) assert not os.path.exists(db_path) + + +def test_batch_does_not_count_individually_failed_items(tmp_path): + """Test that a single item's storage failure inside an otherwise-successful + shared batch transaction is not counted in tracked_count, even though the + rest of the block commits (#807 follow-up).""" + db_path = str(tmp_path / "test_batch_partial_failure.db") + mgr = ProvenanceManager(storage_path=db_path) + + entities = [{"id": f"ent_{i}", "metadata": {"index": i}} for i in range(5)] + # A set() is not JSON-serializable, so json.dumps(entry.metadata) raises + # inside _store_with_conn for this one item, without aborting the shared + # transaction the other items are committed under. + entities[2]["metadata"] = {"bad": {1, 2, 3}} + + count = mgr.track_entities_batch(entities, source="doc_1") + stored = mgr.storage.retrieve_all() + + assert count == len(stored) == 4 + assert "ent_2" not in [e.entity_id for e in stored] + + +def test_chunks_batch_does_not_count_individually_failed_items(tmp_path): + """Same as above for track_chunks_batch/track_chunk (#807 follow-up).""" + db_path = str(tmp_path / "test_chunks_batch_partial_failure.db") + mgr = ProvenanceManager(storage_path=db_path) + + chunks = [ + {"id": f"chk_{i}", "start_index": 0, "end_index": 10} + for i in range(5) + ] + chunks[2]["metadata"] = {"bad": {1, 2, 3}} + + count = mgr.track_chunks_batch(chunks, source_document="doc_1") + stored = mgr.storage.retrieve_all() + + assert count == len(stored) == 4 + assert "chk_2" not in [e.entity_id for e in stored] + + +def test_track_entity_standalone_call_still_degrades_gracefully(tmp_path): + """Test that a direct (non-batch) track_entity() call still returns an + entry on storage failure instead of raising, preserving the public API's + existing graceful-degradation contract.""" + db_path = str(tmp_path / "test_standalone_degrade.db") + mgr = ProvenanceManager(storage_path=db_path) + + entry = mgr.track_entity("entity_1", source="doc_1", metadata={"bad": {1, 2, 3}}) + + assert entry.entity_id == "entity_1" + assert mgr.storage.retrieve("entity_1") is None + + +def test_retrieve_and_trace_lineage_do_not_block_on_writer_lock(tmp_path): + """Test that retrieve() and trace_lineage() no longer take BEGIN IMMEDIATE, + so they don't serialize behind a connection holding the writer lock + (#807 follow-up: this used to hang until busy_timeout elapsed).""" + import sqlite3 + import threading + + db_path = str(tmp_path / "test_read_concurrency.db") + storage = SQLiteStorage(db_path) + storage.store(ProvenanceEntry(entity_id="entity_1", entity_type="entity", activity_id="test")) + + writer = sqlite3.connect(db_path) + writer.execute("PRAGMA journal_mode=WAL") + writer.execute("BEGIN IMMEDIATE") + try: + result = {} + + def do_read(): + result["entry"] = storage.retrieve("entity_1") + result["lineage"] = storage.trace_lineage("entity_1") + + t = threading.Thread(target=do_read) + t.start() + t.join(timeout=2) + + assert not t.is_alive(), "retrieve()/trace_lineage() blocked behind the writer lock" + assert result["entry"].entity_id == "entity_1" + assert len(result["lineage"]) == 1 + finally: + writer.rollback() + writer.close()