mengzhx/semantica
1
0
Fork 0
mirror of https://github.com/semantica-agi/semantica.git synced 2026-08-29 04:26:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(provenance): log tracking failures and return None on storage error (closes #783)

Browse Source
This commit is contained in:
Sameer6305
2026-07-31 15:00:23 +05:30
parent 7cab35bbc0
commit 495e29d543
4 changed files with 130 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
semantica/conflicts/source_tracker.py
+24 -9
View File
@@ -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(
semantica/provenance/manager.py
+28 -14
View File
@@ -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
tests/conflicts/test_conflicts.py
+24
View File
@@ -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()
tests/provenance/test_manager.py
+54 -13
View File
@@ -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