diff --git a/semantica/conflicts/source_tracker.py b/semantica/conflicts/source_tracker.py index 3a00567f..d2035c66 100644 --- a/semantica/conflicts/source_tracker.py +++ b/semantica/conflicts/source_tracker.py @@ -614,29 +614,44 @@ class SourceTracker: if item_type == "entity": entity_id = item.get("entity_id") if entity_id: - self.track_entity_source(entity_id, source_ref, **metadata) - stats["entities_tracked"] += 1 - stats["total_tracked"] += 1 + entry = self.track_entity_source(entity_id, source_ref, **metadata) + if entry is not None: + stats["entities_tracked"] += 1 + stats["total_tracked"] += 1 + else: + self.logger.warning( + f"Failed to track entity source for '{entity_id}' in item {i}" + ) elif item_type == "property": entity_id = item.get("entity_id") property_name = item.get("property_name") value = item.get("value") if entity_id and property_name is not None: - self.track_property_source( + entry = self.track_property_source( entity_id, property_name, value, source_ref, **metadata ) - stats["properties_tracked"] += 1 - stats["total_tracked"] += 1 + if entry is not None: + stats["properties_tracked"] += 1 + stats["total_tracked"] += 1 + else: + self.logger.warning( + f"Failed to track property source for '{entity_id}.{property_name}' in item {i}" + ) elif item_type == "relationship": relationship_id = item.get("relationship_id") if relationship_id: - self.track_relationship_source( + entry = self.track_relationship_source( relationship_id, source_ref, **metadata ) - stats["relationships_tracked"] += 1 - stats["total_tracked"] += 1 + if entry is not None: + stats["relationships_tracked"] += 1 + stats["total_tracked"] += 1 + else: + self.logger.warning( + f"Failed to track relationship source for '{relationship_id}' in item {i}" + ) else: self.logger.warning( diff --git a/semantica/provenance/manager.py b/semantica/provenance/manager.py index 2eafb937..b2e9874c 100644 --- a/semantica/provenance/manager.py +++ b/semantica/provenance/manager.py @@ -144,8 +144,8 @@ class ProvenanceManager: entry: ProvenanceEntry, _conn: Optional[Any] = None, _raise_on_error: bool = False, - ) -> ProvenanceEntry: - """Compute checksum, store entry persistently, and gracefully ignore storage errors.""" + ) -> Optional[ProvenanceEntry]: + """Compute checksum, store entry persistently, and log/handle storage errors (#783).""" entry.checksum = compute_checksum(entry) try: @@ -153,13 +153,19 @@ class ProvenanceManager: self.storage._store_with_conn(_conn, entry) else: self.storage.store(entry) - except Exception: + except Exception as e: + self.logger.error( + "Failed to save provenance entry for entity '%s': %s", + entry.entity_id, + e, + exc_info=True, + ) # 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 _raise_on_error: raise - pass # Graceful failure - don't break main functionality + return None # Graceful failure - don't return unpersisted entry (#783) return entry @@ -297,7 +303,7 @@ class ProvenanceManager: source: str, metadata: Optional[Dict[str, Any]] = None, **kwargs - ) -> ProvenanceEntry: + ) -> Optional[ProvenanceEntry]: """ Track relationship provenance (kg.ProvenanceTracker compatible). @@ -308,7 +314,7 @@ class ProvenanceManager: **kwargs: Additional fields Returns: - ProvenanceEntry object + Optional[ProvenanceEntry]: ProvenanceEntry on success, or None if storage fails Example: >>> prov_mgr.track_relationship( @@ -343,7 +349,7 @@ class ProvenanceManager: parent_chunk_id: Optional[str] = None, _conn: Optional[Any] = None, **metadata - ) -> ProvenanceEntry: + ) -> Optional[ProvenanceEntry]: """ Track chunk provenance (split.ProvenanceTracker compatible). @@ -357,7 +363,7 @@ class ProvenanceManager: **metadata: Additional metadata Returns: - ProvenanceEntry object + Optional[ProvenanceEntry]: ProvenanceEntry on success, or None if storage fails Example: >>> prov_mgr.track_chunk( @@ -392,7 +398,7 @@ class ProvenanceManager: value: Any, source: SourceReference, **metadata - ) -> ProvenanceEntry: + ) -> Optional[ProvenanceEntry]: """ Track property source (conflicts.SourceTracker compatible). @@ -404,7 +410,7 @@ class ProvenanceManager: **metadata: Additional metadata Returns: - ProvenanceEntry object + Optional[ProvenanceEntry]: ProvenanceEntry on success, or None if storage fails Example: >>> source = SourceReference( @@ -489,8 +495,12 @@ class ProvenanceManager: # Add batch_count to tracked_count only after the transaction context # exits successfully and commits (#807). tracked_count += batch_count - except Exception: - pass # Partial failure: if block-level storage transaction fails, continue to next block + except Exception as e: + self.logger.error( + "Block-level storage transaction failed in track_entities_batch: %s", + e, + exc_info=True, + ) return tracked_count @@ -544,8 +554,12 @@ class ProvenanceManager: # Add batch_count to tracked_count only after the transaction context # exits successfully and commits (#807). tracked_count += batch_count - except Exception: - pass + except Exception as e: + self.logger.error( + "Block-level storage transaction failed in track_chunks_batch: %s", + e, + exc_info=True, + ) return tracked_count diff --git a/tests/conflicts/test_conflicts.py b/tests/conflicts/test_conflicts.py index bee5c828..8018cb4c 100644 --- a/tests/conflicts/test_conflicts.py +++ b/tests/conflicts/test_conflicts.py @@ -84,6 +84,30 @@ class TestConflictsModule(unittest.TestCase): sources = tracker.get_entity_sources("e1") self.assertTrue(len(sources) >= 1) + def test_track_sources_batch_failure_not_counted(self): + """Test that track_sources_batch does not increment stats when tracking fails (#783).""" + tracker = SourceTracker() + source_data = [ + { + "type": "property", + "entity_id": "e1", + "property_name": "age", + "value": 30, + "source": self.source1, + }, + { + "type": "property", + "entity_id": "e2", + "property_name": "age", + "value": 25, + "source": self.source2, + }, + ] + with patch.object(tracker, "track_property_source", return_value=None): + stats = tracker.track_sources_batch(source_data) + self.assertEqual(stats["properties_tracked"], 0) + self.assertEqual(stats["total_tracked"], 0) + def test_conflict_detector(self): detector = ConflictDetector() diff --git a/tests/provenance/test_manager.py b/tests/provenance/test_manager.py index 87eab374..77f61b79 100644 --- a/tests/provenance/test_manager.py +++ b/tests/provenance/test_manager.py @@ -7,7 +7,8 @@ chunk tracking, source tracking, and lineage tracing. import pytest from unittest.mock import patch -from semantica.provenance import ProvenanceManager, SourceReference +from datetime import datetime +from semantica.provenance import ProvenanceManager, SourceReference, ProvenanceEntry from semantica.provenance.storage import InMemoryStorage, SQLiteStorage @@ -611,16 +612,18 @@ class TestProvenanceManager: assert entry is None def test_track_relationship_storage_error_swallowed(self): - """Test that track_relationship swallows storage.store() exceptions and still returns a ProvenanceEntry.""" + """Test that track_relationship returns None when storage.store() fails, + since nothing was persisted (#783).""" prov_mgr = ProvenanceManager() with patch.object(prov_mgr.storage, "store", side_effect=RuntimeError("storage error")): entry = prov_mgr.track_relationship("r_test", source="doc_1") - assert entry is not None - assert entry.entity_id == "r_test" - assert entry.checksum is not None + # Returning None is correct because nothing was actually persisted, + # and the old assertion was encoding the bug this issue was filed to fix. + assert entry is None def test_track_chunk_storage_error_swallowed(self): - """Test that track_chunk swallows storage.store() exceptions and still returns a ProvenanceEntry.""" + """Test that track_chunk returns None when storage.store() fails, + since nothing was persisted (#783).""" prov_mgr = ProvenanceManager() with patch.object(prov_mgr.storage, "store", side_effect=RuntimeError("storage error")): entry = prov_mgr.track_chunk( @@ -630,12 +633,13 @@ class TestProvenanceManager: start_index=0, end_index=100, ) - assert entry is not None - assert entry.entity_id == "c_test" - assert entry.checksum is not None + # Returning None is correct because nothing was actually persisted, + # and the old assertion was encoding the bug this issue was filed to fix. + assert entry is None def test_track_property_source_storage_error_swallowed(self): - """Test that track_property_source swallows storage.store() exceptions and still returns a ProvenanceEntry.""" + """Test that track_property_source returns None when storage.store() fails, + since nothing was persisted (#783).""" prov_mgr = ProvenanceManager() source = SourceReference(document="doc_1", page=1, confidence=0.9) with patch.object(prov_mgr.storage, "store", side_effect=RuntimeError("storage error")): @@ -645,9 +649,46 @@ class TestProvenanceManager: value="val", source=source, ) - assert entry is not None - assert entry.entity_id == "e_test_prop_test" - assert entry.checksum is not None + # Returning None is correct because nothing was actually persisted, + # and the old assertion was encoding the bug this issue was filed to fix. + assert entry is None + + def test_save_entry_logs_on_every_failure_path(self): + """Test that _save_entry logs on storage failures for both raising and swallowing paths (#783).""" + prov_mgr = ProvenanceManager() + entry = ProvenanceEntry( + entity_id="log_test_id", + entity_type="entity", + activity_id="test", + source_document="doc_1", + first_seen=datetime.utcnow().isoformat(), + last_updated=datetime.utcnow().isoformat(), + ) + with patch.object(prov_mgr.storage, "store", side_effect=RuntimeError("storage error")), \ + patch.object(prov_mgr.logger, "error") as mock_log_error: + # Swallowing branch (_raise_on_error=False) + res = prov_mgr._save_entry(entry, _raise_on_error=False) + assert res is None + mock_log_error.assert_called_once() + assert "log_test_id" in mock_log_error.call_args[0][1] + + mock_log_error.reset_mock() + + # Raising branch (_raise_on_error=True) + with pytest.raises(RuntimeError, match="storage error"): + prov_mgr._save_entry(entry, _raise_on_error=True) + mock_log_error.assert_called_once() + assert "log_test_id" in mock_log_error.call_args[0][1] + + def test_track_entities_batch_logs_per_item_failure(self): + """Test that track_entities_batch emits item-level logs via _save_entry when items fail (#783).""" + prov_mgr = ProvenanceManager() + entities = [{"id": "e_fail_1"}, {"id": "e_fail_2"}] + with patch.object(prov_mgr.storage, "_store_with_conn", side_effect=RuntimeError("batch store error")), \ + patch.object(prov_mgr.logger, "error") as mock_log_error: + count = prov_mgr.track_entities_batch(entities, "doc_1") + assert count == 0 + assert mock_log_error.call_count == 2 def test_track_entity_pre_build_failure_fallback_skips_store(self): """Test that when track_entity fails before the entry is built (e.g. a