From a1a72cdd5053f08e94442176605f70d58383c45d Mon Sep 17 00:00:00 2001 From: logan-jl-cc <57258899+logan-jl-cc@users.noreply.github.com> Date: Tue, 25 Aug 2026 18:27:31 +0800 Subject: [PATCH] fix(triplet_store): OxigraphStore silently ignores storage_path; add_triplets skips flush (#970) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(triplet_store): OxigraphStore silently ignores storage_path and skips flush Two persistence bugs in OxigraphStore: 1. `storage_path=...` was silently swallowed by **config. The __init__ parameter is named `path`, so passing the project-conventional `storage_path` (used by ProvenanceManager and other stores) left self.path = None and the store silently degraded to in-memory — no error, no warning, data gone on exit. Accept `storage_path` as an alias for `path`. 2. add_triplets never called flush(). pyoxigraph auto-flushes via background threads but, per its docs, "might lag a little bit" — that lag is a race where reopening or crashing immediately after a write observes fewer triples. Call flush() explicitly for on-disk stores to close the window. Both verified: with the fix, `OxigraphStore(storage_path=...)` persists across reopen; without it, data is lost. * fix(triplet_store): improve oxigraph persistence * test(triplet_store): clarify oxigraph persistence test --------- Co-authored-by: administrator Co-authored-by: Sameer Kadam --- semantica/triplet_store/oxigraph_store.py | 59 +++++++++++++-- tests/triplet_store/test_oxigraph_store.py | 86 ++++++++++++++++++++++ 2 files changed, 138 insertions(+), 7 deletions(-) diff --git a/semantica/triplet_store/oxigraph_store.py b/semantica/triplet_store/oxigraph_store.py index 5262c00a..4ba94f03 100644 --- a/semantica/triplet_store/oxigraph_store.py +++ b/semantica/triplet_store/oxigraph_store.py @@ -42,6 +42,10 @@ class OxigraphStore: ProcessingError: If the store cannot be opened. """ self.logger = get_logger("oxigraph_store") + # Accept storage_path as an alias for path (matches the convention used + # by other Semantica stores). Pop it so it isn't left in self.config. + if path is None and "storage_path" in config: + path = config.pop("storage_path") self.config = config self.path = path if path is not None else config.get("path") @@ -74,24 +78,65 @@ class OxigraphStore: ) from exc def add_triplet(self, triplet: Triplet, **options) -> Dict[str, Any]: - """Add one triplet to the default graph or ``options['graph']``.""" - return self.add_triplets([triplet], **options) + """Add one triplet to the default graph or ``options['graph']``. - def add_triplets(self, triplets: List[Triplet], **options) -> Dict[str, Any]: - """Add triplets in one native Oxigraph batch.""" + The write is committed to the store's in-memory state immediately. + pyoxigraph's background threads will persist it to disk shortly + afterward; call :meth:`flush` explicitly if you need a synchronous + durability guarantee before reopening or crashing. + """ try: graph_name = self._graph_name(options.get("graph")) - quads = [self._to_quad(triplet, graph_name) for triplet in triplets] - self.store.extend(quads) + self.store.extend([self._to_quad(triplet, graph_name)]) return { "success": True, - "triplets_loaded": len(triplets), + "triplets_loaded": 1, "graph": options.get("graph"), } except Exception as exc: self.logger.error(f"Oxigraph load failed: {exc}") raise ProcessingError(f"Oxigraph load failed: {exc}") from exc + def add_triplets(self, triplets: List[Triplet], **options) -> Dict[str, Any]: + """Add triplets in one native Oxigraph batch. + + The batch is written transactionally and then explicitly flushed to + disk before returning. This makes the full batch durable without + requiring a separate :meth:`flush` call. In-memory stores skip the + flush (there is nothing to sync). + + For high-volume imports the :class:`~.bulk_loader.BulkLoader` splits + work into chunks and calls this method once per chunk, so each chunk + lands as one atomic, durable unit. + """ + try: + graph_name = self._graph_name(options.get("graph")) + quads = [self._to_quad(triplet, graph_name) for triplet in triplets] + self.store.extend(quads) + except Exception as exc: + self.logger.error(f"Oxigraph load failed: {exc}") + raise ProcessingError(f"Oxigraph load failed: {exc}") from exc + + # Flush is kept outside the write try/except so that a flush I/O error + # does not produce a misleading "load failed" message when extend() + # already committed the batch successfully. + if self.path is not None: + try: + self.flush() + except OSError as exc: + self.logger.warning( + f"Oxigraph flush failed after successful write: {exc}" + ) + raise ProcessingError( + f"Oxigraph flush failed after successful write: {exc}" + ) from exc + + return { + "success": True, + "triplets_loaded": len(triplets), + "graph": options.get("graph"), + } + def bulk_load(self, triplets: List[Triplet], **options) -> Dict[str, Any]: """Load a batch of triplets using Oxigraph's native bulk operation.""" return self.add_triplets(triplets, **options) diff --git a/tests/triplet_store/test_oxigraph_store.py b/tests/triplet_store/test_oxigraph_store.py index cbd3c979..9415795d 100644 --- a/tests/triplet_store/test_oxigraph_store.py +++ b/tests/triplet_store/test_oxigraph_store.py @@ -159,3 +159,89 @@ def test_missing_optional_dependency_has_install_hint(): ): with pytest.raises(ImportError, match="tripletstore-oxigraph"): _store() + + +def test_on_disk_add_triplets_calls_flush(tmp_path): + """add_triplets on a disk-backed store must flush once after the batch. + + The pyoxigraph background-thread flush "might lag a little bit"; an + explicit flush after the batch closes that race without fsyncing on + every individual write. This test verifies the contract directly + without relying on CPython destructor timing. + """ + store = OxigraphStore(path=tmp_path / "oxigraph") + with patch.object(store, "flush") as mock_flush: + store.add_triplets([ + Triplet(EX + "alice", EX + "knows", EX + "bob"), + Triplet(EX + "bob", EX + "knows", EX + "carol"), + ]) + mock_flush.assert_called_once() + + +def test_on_disk_add_triplet_does_not_flush(tmp_path): + """add_triplet (single write) must NOT flush on every call. + + Individual writes are committed to the store in memory; the caller is + responsible for calling flush() when a hard durability boundary is + needed. Flushing on every add_triplet() call would fsync on every + write, causing a severe throughput regression for workloads that write + triplets one at a time. + """ + store = OxigraphStore(path=tmp_path / "oxigraph") + with patch.object(store, "flush") as mock_flush: + store.add_triplet(Triplet(EX + "alice", EX + "knows", EX + "bob")) + mock_flush.assert_not_called() + + +def test_in_memory_add_triplets_does_not_flush(tmp_path): + """In-memory stores must not call flush() — there is nothing to flush.""" + store = OxigraphStore() # no path → in-memory + with patch.object(store, "flush") as mock_flush: + store.add_triplet(Triplet(EX + "alice", EX + "knows", EX + "bob")) + store.add_triplets([Triplet(EX + "bob", EX + "knows", EX + "carol")]) + mock_flush.assert_not_called() + + +def test_on_disk_add_triplets_is_durable_on_reopen(tmp_path): + """End-to-end durability: a batch written via add_triplets and closed + cleanly survives a reopen. + + This is an integration test for the full add_triplets → flush → close → + reopen lifecycle. The durability contract here is provided by the + explicit ``store.flush()`` call before deletion; the internal flush + inside add_triplets reduces (but does not eliminate) the crash-window + race. The authoritative unit test for the internal flush behaviour is + ``test_on_disk_add_triplets_calls_flush``. + """ + path = tmp_path / "oxigraph" + store = OxigraphStore(path=path) + store.add_triplets([ + Triplet(EX + "alice", EX + "knows", EX + "bob"), + Triplet(EX + "bob", EX + "knows", EX + "carol"), + ]) + store.flush() # belt-and-suspenders: ensures close is clean + del store + gc.collect() + + reopened = OxigraphStore(path=path) + assert len(reopened.get_triplets()) == 2 + + +def test_storage_path_is_accepted_as_alias_for_path(tmp_path): + """Regression: ``storage_path=...`` used to be silently swallowed by + ``**config`` (the __init__ parameter is named ``path``), so the store + silently degraded to in-memory with no warning. It must now be accepted + as an alias consistent with other Semantica stores (e.g. ProvenanceManager).""" + storage_path = tmp_path / "oxigraph" + + store = OxigraphStore(storage_path=str(storage_path)) + + assert store.path == str(storage_path) + # and it must actually persist (proves the alias wired through to the + # on-disk path, not just set the attribute) + store.add_triplet(Triplet(EX + "alice", EX + "knows", EX + "bob")) + del store + gc.collect() + + reopened = OxigraphStore(storage_path=str(storage_path)) + assert len(reopened.get_triplets()) == 1