diff --git a/detectmatelibrary_tests/test_detectors/test_persist_integration.py b/detectmatelibrary_tests/test_detectors/test_persist_integration.py index c39692c7..86ec7365 100644 --- a/detectmatelibrary_tests/test_detectors/test_persist_integration.py +++ b/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() diff --git a/detectmatelibrary_tests/test_utils/test_persistency_saver.py b/detectmatelibrary_tests/test_utils/test_persistency_saver.py index f01b8b5c..abcf9d98 100644 --- a/detectmatelibrary_tests/test_utils/test_persistency_saver.py +++ b/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() diff --git a/src/detectmatelibrary/common/core.py b/src/detectmatelibrary/common/core.py index 8284b346..b8aa5ed6 100644 --- a/src/detectmatelibrary/common/core.py +++ b/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( diff --git a/src/detectmatelibrary/utils/persistency/event_persistency.py b/src/detectmatelibrary/utils/persistency/event_persistency.py index 4b1a4137..ba9ae18c 100644 --- a/src/detectmatelibrary/utils/persistency/event_persistency.py +++ b/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: diff --git a/src/detectmatelibrary/utils/persistency/persistency_saver.py b/src/detectmatelibrary/utils/persistency/persistency_saver.py index aabf790a..6cc45e46 100644 --- a/src/detectmatelibrary/utils/persistency/persistency_saver.py +++ b/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