Fix/persistency ingest lock

detectmatelibrary_tests/test_detectors/test_persist_integration.py

@@ -1,3 +1,5 @@
+import threading
+
 import fsspec
 
 from detectmatelibrary.detectors.new_value_detector import NewValueDetector, NewValueDetectorConfig
@@ -224,3 +226,44 @@ def test_import_state_accepts_bytes(self):
         det2 = NewValueDetector(name="BytesDst", config=NewValueDetectorConfig(auto_config=False))
         det2.import_state(data)
         assert 5 in det2.persistency.get_events_seen()
+
+    def test_export_does_not_reset_events_since_save(self):
+        # export_state is a snapshot, not a persistency save: it must not reset
+        # the counter that governs the background saver's cadence.
+        det = NewValueDetector(name="ExportNoReset", config=NewValueDetectorConfig(auto_config=False))
+        det.persistency.ingest_event(event_id=1, event_template="login <*>", named_variables={"user": "a"})
+        det.persistency.ingest_event(event_id=1, event_template="login <*>", named_variables={"user": "b"})
+        assert det.persistency._events_since_save == 2
+        det.export_state("memory://export_noreset/state")
+        assert det.persistency._events_since_save == 2
+
+    def test_export_state_with_running_saver_does_not_raise(self):
+        # export must acquire the saver lock: a concurrent ingest + background
+        # timer save would otherwise race the state snapshot.
+        det = NewValueDetector(
+            name="ExportSaverConcurrent",
+            config=NewValueDetectorConfig(
+                auto_config=False,
+                persist=PersistConfig(path="memory://export_saver_dst/state", interval_seconds=0),
+            ),
+        )
+        stop = threading.Event()
+
+        def ingest_loop():
+            i = 0
+            while not stop.is_set():
+                det.persistency.ingest_event(
+                    event_id=i % 5, event_template="e <*>", named_variables={"n": str(i)}
+                )
+                i += 1
+
+        t = threading.Thread(target=ingest_loop)
+        t.start()
+        try:
+            for _ in range(20):
+                data = det.export_state()  # concurrent with timer save + ingest
+                assert isinstance(data, bytes) and len(data) > 0
+        finally:
+            stop.set()
+            t.join(timeout=2.0)
+            det.saver.stop()

detectmatelibrary_tests/test_utils/test_persistency_saver.py

@@ -351,6 +351,68 @@ def test_full_cycle_tracker_backend(self):
             assert rest.unique_set == orig.unique_set
 
 
+class TestPersistencySaverConcurrency:
+    def test_ingest_blocks_while_saver_lock_held(self):
+        """ingest_event must serialize on the saver's lock so save/load can't
+        race with a concurrent ingest."""
+        p = EventPersistency(event_data_class=EventDataFrame)
+        saver = PersistencySaver(p, PersistencySaverConfig(path="memory://concurrency/state"))
+        done = threading.Event()
+
+        def worker():
+            p.ingest_event(event_id="E1", event_template="T", variables=["x"], named_variables={})
+            done.set()
+
+        with saver.locked():
+            t = threading.Thread(target=worker)
+            t.start()
+            time.sleep(0.1)  # ample time for an unguarded ingest to complete
+            assert not done.is_set(), "ingest_event ran while the saver lock was held"
+            assert "E1" not in p.get_events_seen()
+        t.join(timeout=1.0)
+        assert done.is_set()
+        assert "E1" in p.get_events_seen()
+
+    def test_ingest_not_blocked_by_save_write(self, monkeypatch):
+        """The file write must run OUTSIDE the lock: a blocked save write must
+        not block a concurrent ingest_event (only serialization is guarded)."""
+        import detectmatelibrary.utils.persistency.persistency_saver as ps
+
+        p = EventPersistency(event_data_class=EventDataFrame)
+        saver = PersistencySaver(p, PersistencySaverConfig(path="memory://write_block/state"))
+
+        write_started = threading.Event()
+        release_write = threading.Event()
+        real_write = ps._write
+
+        def blocking_write(fs, root, files):
+            write_started.set()
+            release_write.wait(timeout=2.0)
+            real_write(fs, root, files)
+
+        monkeypatch.setattr(ps, "_write", blocking_write)
+
+        saver_thread = threading.Thread(target=saver.save)
+        saver_thread.start()
+        assert write_started.wait(timeout=1.0), "save write never started"
+
+        # Write is in progress with the lock released — ingest must proceed.
+        ingested = threading.Event()
+
+        def worker():
+            p.ingest_event(event_id="E2", event_template="T", variables=["y"], named_variables={})
+            ingested.set()
+
+        t = threading.Thread(target=worker)
+        t.start()
+        assert ingested.wait(timeout=1.0), "ingest_event blocked while save write was in progress"
+
+        release_write.set()
+        saver_thread.join(timeout=2.0)
+        t.join(timeout=1.0)
+        assert "E2" in p.get_events_seen()
+
+
 class TestStandaloneSaveLoad:
     def test_save_creates_metadata(self):
         p = _make_persistency_with_data()
@@ -365,11 +427,13 @@ def test_save_creates_event_files(self):
         assert fs.exists("standalone_save2/state/events/E1.parquet")
         assert fs.exists("standalone_save2/state/events/E2.parquet")
 
-    def test_save_resets_events_since_save(self):
+    def test_save_does_not_reset_events_since_save(self):
+        # Module-level save() (used by export_state) is a plain snapshot and
+        # must NOT touch the save counter — only PersistencySaver.save() does.
         p = _make_persistency_with_data()
         assert p._events_since_save == 3
         standalone_save(p, "memory://standalone_save3/state")
-        assert p._events_since_save == 0
+        assert p._events_since_save == 3
 
     def test_load_restores_events_seen(self):
         p = _make_persistency_with_data()

src/detectmatelibrary/common/core.py

@@ -137,8 +137,9 @@ def export_state(
 
         When path is None, returns the state as bytes (zip archive).
         When path is given, writes to that fsspec URI and returns None.
-        Returns None if no persistency is configured. Not thread-safe
-        when a PersistencySaver is running on this component.
+        Returns None if no persistency is configured. Thread-safe when a
+        PersistencySaver is running: acquires the saver lock before saving
+        (guards against the background save timer and concurrent ingest).
         """
         # ponytail: local import keeps common.core free of a persistency
         # package import at module load (see _Stoppable).
@@ -148,6 +149,10 @@ def export_state(
         if ep is None:
             logger.debug(f"{self.name}: no persistency configured, nothing to export")
             return None
+        saver = getattr(self, "saver", None)
+        if saver is not None:
+            with saver.locked():
+                return persistency.save(ep, path, storage_options)
         return persistency.save(ep, path, storage_options)
 
     def import_state(

src/detectmatelibrary/utils/persistency/event_persistency.py

@@ -1,3 +1,4 @@
+import threading
 from typing import Any, Callable, Dict, List, Optional, Type
 
 from .event_data_structures.base import EventDataStructure
@@ -33,6 +34,10 @@ def __init__(
         self.event_templates: Dict[int | str, str] = {}
         self._events_since_save: int = 0
         self._on_ingest_callbacks: list[Callable[[], None]] = []
+        # ponytail: RLock shared with PersistencySaver so ingest/save/load are
+        # mutually exclusive. On-ingest callbacks fire outside this lock, so
+        # re-entrancy is no longer required — kept as RLock (harmless, safer).
+        self._lock = threading.RLock()
 
     def ingest_event(
         self,
@@ -42,22 +47,24 @@ def ingest_event(
         named_variables: Dict[str, Any] = {}
     ) -> None:
         """Ingest event data into the appropriate EventData store."""
-        self._events_since_save += 1
+        with self._lock:
+            self._events_since_save += 1
+            self.events_seen.add(event_id)
+            if variables or named_variables:
+                self.event_templates[event_id] = event_template
+                all_variables = self.get_all_variables(variables, named_variables)
+
+                data_structure = self.events_data.get(event_id)
+                if data_structure is None:
+                    data_structure = self.event_data_class(**self.event_data_kwargs)
+                    self.events_data[event_id] = data_structure
+
+                data = data_structure.to_data(all_variables)
+                data_structure.add_data(data)
+        # ponytail: fire callbacks outside the lock so a count-triggered save
+        # doesn't hold the ingest lock across serialize + file I/O.
         for _cb in self._on_ingest_callbacks:
             _cb()
-        self.events_seen.add(event_id)
-        if not variables and not named_variables:
-            return
-        self.event_templates[event_id] = event_template
-        all_variables = self.get_all_variables(variables, named_variables)
-
-        data_structure = self.events_data.get(event_id)
-        if data_structure is None:
-            data_structure = self.event_data_class(**self.event_data_kwargs)
-            self.events_data[event_id] = data_structure
-
-        data = data_structure.to_data(all_variables)
-        data_structure.add_data(data)
 
     @property
     def events_since_save(self) -> int:

src/detectmatelibrary/utils/persistency/persistency_saver.py

@@ -56,8 +56,14 @@ def _safe_event_data_kwargs(ep: EventPersistency) -> dict[str, Any]:
     return safe
 
 
-def _save(ep: EventPersistency, fs: Any, root: str) -> None:
-    fs.makedirs(f"{root}/events", exist_ok=True)
+def _serialize(ep: EventPersistency) -> dict[str, bytes]:
+    """Serialize EP state to {relative_path: bytes}.
+
+    Pure CPU — reads live ep state, does no I/O and does NOT reset the
+    events-since-save counter. Callers that persist a save reset the
+    counter themselves (see PersistencySaver.save).
+    """
+    files: dict[str, bytes] = {}
     event_backends: dict[str, str] = {}
     event_extensions: dict[str, str] = {}
 
@@ -66,8 +72,7 @@ def _save(ep: EventPersistency, fs: Any, root: str) -> None:
         ext = _EXTENSION_MAP.get(backend_name, "bin")
         event_backends[str(event_id)] = backend_name
         event_extensions[str(event_id)] = ext
-        file_path = f"{root}/events/{event_id}.{ext}"
-        fs.pipe(file_path, data_structure.dump())
+        files[f"events/{event_id}.{ext}"] = data_structure.dump()
 
     metadata = {
         "version": 1,
@@ -79,9 +84,23 @@ def _save(ep: EventPersistency, fs: Any, root: str) -> None:
         "event_data_kwargs": _safe_event_data_kwargs(ep),
         "event_data_class": ep.event_data_class.__name__,  # read back by _load
     }
-    fs.pipe(f"{root}/metadata.json", json.dumps(metadata, indent=2).encode())
+    files["metadata.json"] = json.dumps(metadata, indent=2).encode()
+    return files
+
+
+def _write(fs: Any, root: str, files: dict[str, bytes]) -> None:
+    """Write serialized files to storage.
+
+    Pure I/O.
+    """
+    fs.makedirs(f"{root}/events", exist_ok=True)
+    for rel, data in files.items():
+        fs.pipe(f"{root}/{rel}", data)
 
-    ep.reset_events_since_save()
+
+def _save(ep: EventPersistency, fs: Any, root: str) -> None:
+    # No counter reset here — that is a PersistencySaver concern (see save()).
+    _write(fs, root, _serialize(ep))
 
 
 class PersistencyLoadError(Exception):
@@ -132,17 +151,10 @@ def _load(ep: EventPersistency, fs: Any, root: str) -> None:
 
 def _save_to_bytes(ep: EventPersistency) -> bytes:
     """Serialize EP state to a zip archive in memory."""
-    # reusing the same _save logic with an in-memory filesystem to avoid code duplication
-    from fsspec.implementations.memory import MemoryFileSystem
-    mem_fs = MemoryFileSystem()
-    root = "s"
-    _save(ep, mem_fs, root)
     buf = io.BytesIO()
     with zipfile.ZipFile(buf, "w", compression=zipfile.ZIP_DEFLATED) as zf:
-        for file_path in mem_fs.find(root):
-            if mem_fs.isfile(file_path):
-                rel = file_path[len(root) + 1:]
-                zf.writestr(rel, mem_fs.cat(file_path))
+        for rel, data in _serialize(ep).items():
+            zf.writestr(rel, data)
     return buf.getvalue()
 
 
@@ -193,7 +205,9 @@ def __init__(self, persistency: EventPersistency, config: PersistencySaverConfig
         self._persistency = persistency
         self._config = config
         self._fs, self._root = fsspec.url_to_fs(config.path, **config.storage_options)
-        self._lock = threading.Lock()
+        # Share the EventPersistency lock so save/load are mutually exclusive
+        # with ingest_event (which holds the same lock).
+        self._lock = persistency._lock
         self._timer: _SaveTimer | None = None
 
         if config.events_until_save is not None:
@@ -208,13 +222,19 @@ def __init__(self, persistency: EventPersistency, config: PersistencySaverConfig
     def save(self) -> None:
         """Write full EventPersistency state to storage.
 
-        Thread-safe.
+        Thread-safe. Serialization runs under the shared lock (reads
+        live state); the file write runs outside it so ingest_event
+        isn't blocked on I/O.
         """
         with self._lock:
-            try:
-                _save(self._persistency, self._fs, self._root)
-            except Exception as e:
-                logger.warning(f"PersistencySaver: save failed — {e}")
+            files = _serialize(self._persistency)
+            # Reset here (atomically with the snapshot) not in _serialize:
+            # module-level save()/export must not touch the save counter.
+            self._persistency.reset_events_since_save()
+        try:
+            _write(self._fs, self._root, files)
+        except Exception as e:
+            logger.warning(f"PersistencySaver: save failed — {e}")
 
     def load(self) -> None:
         """Restore EventPersistency state from storage.
@@ -284,6 +304,12 @@ def locked(self) -> Any:
     def _check_event_count(self) -> None:
         """Trigger a save when ingested-event count reaches
         events_until_save."""
+        # ponytail: this runs OUTSIDE the ingest lock (callbacks fire after the
+        # lock releases) so save()'s file I/O never stalls ingest. The counter
+        # read is intentionally lock-free — a TOCTOU race here only ever causes a
+        # slightly-late or redundant save, never a missed save or lost event
+        # (events stay in memory until persisted). Do NOT re-add a lock around
+        # this: it would put save's I/O back under the ingest lock.
         if (
             self._config.events_until_save is not None
             and self._persistency._events_since_save >= self._config.events_until_save