fix(triplet_store): OxigraphStore silently ignores storage_path; add_triplets skips flush (#970)

* 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 <administrator@administratordeMac-mini.local>
Co-authored-by: Sameer Kadam <sskadam6305@gmail.com>

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']``."""
+        """Add one triplet to the default graph or ``options['graph']``.
-        return self.add_triplets([triplet], **options)
 
-    def add_triplets(self, triplets: List[Triplet], **options) -> Dict[str, Any]:
+        The write is committed to the store's in-memory state immediately.
-        """Add triplets in one native Oxigraph batch."""
+        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([self._to_quad(triplet, graph_name)])
-            self.store.extend(quads)
             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)

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