From 273ab9f0c604b941b68035f9a29020be26f03871 Mon Sep 17 00:00:00 2001 From: viktorbeck98 Date: Tue, 30 Jun 2026 15:05:43 +0200 Subject: [PATCH 1/3] Serialize ingest_event with save/load via shared RLock ingest_event held no lock, so the background save timer could iterate events_data while ingest mutated it (silently-skipped saves), and /load could replace state mid-ingest (torn state). Give EventPersistency an RLock, acquire it in ingest_event, and have PersistencySaver share it so ingest/save/load are mutually exclusive. RLock (not Lock) because ingest can re-enter save() via the events_until_save callback on the same thread. Co-Authored-By: Claude Opus 4.8 --- .../test_utils/test_persistency_saver.py | 23 +++++++++++ .../utils/persistency/event_persistency.py | 38 +++++++++++-------- .../utils/persistency/persistency_saver.py | 4 +- 3 files changed, 48 insertions(+), 17 deletions(-) diff --git a/detectmatelibrary_tests/test_utils/test_persistency_saver.py b/detectmatelibrary_tests/test_utils/test_persistency_saver.py index f01b8b5c..71197ef6 100644 --- a/detectmatelibrary_tests/test_utils/test_persistency_saver.py +++ b/detectmatelibrary_tests/test_utils/test_persistency_saver.py @@ -351,6 +351,29 @@ 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() + + class TestStandaloneSaveLoad: def test_save_creates_metadata(self): p = _make_persistency_with_data() diff --git a/src/detectmatelibrary/utils/persistency/event_persistency.py b/src/detectmatelibrary/utils/persistency/event_persistency.py index 4b1a4137..083c9274 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 (not Lock) — ingest fires the on-ingest callback, + # which may re-enter via PersistencySaver.save() on the same thread. + # Shared with PersistencySaver so ingest/save/load are mutually exclusive. + self._lock = threading.RLock() def ingest_event( self, @@ -42,22 +47,23 @@ def ingest_event( named_variables: Dict[str, Any] = {} ) -> None: """Ingest event data into the appropriate EventData store.""" - self._events_since_save += 1 - 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) + with self._lock: + self._events_since_save += 1 + 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..dd60e719 100644 --- a/src/detectmatelibrary/utils/persistency/persistency_saver.py +++ b/src/detectmatelibrary/utils/persistency/persistency_saver.py @@ -193,7 +193,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: From 020dc8fb090da2e5f8b3152cbd7d687169d46af8 Mon Sep 17 00:00:00 2001 From: viktorbeck98 Date: Wed, 1 Jul 2026 15:05:02 +0200 Subject: [PATCH 2/3] Keep save file I/O out of the ingest lock; lock export_state Addresses PR review: ingest_event held the shared lock across save serialization AND file writes, and export_state ran unlocked. - ingest_event fires on-ingest callbacks outside the lock, so a count-triggered save no longer holds the ingest lock during I/O. - Split the save path into _serialize (CPU, under lock) and _write (I/O, no lock); PersistencySaver.save() serializes+resets the counter under the lock, then writes outside it. - export_state acquires saver.locked(), symmetric with import_state. Counter reset moved into PersistencySaver.save(), so export no longer zeroes events_since_save. Co-Authored-By: Claude Opus 4.8 --- .../test_persist_integration.py | 43 ++++++++++++++ .../test_utils/test_persistency_saver.py | 45 +++++++++++++- src/detectmatelibrary/common/core.py | 9 ++- .../utils/persistency/event_persistency.py | 35 +++++------ .../utils/persistency/persistency_saver.py | 58 ++++++++++++------- 5 files changed, 149 insertions(+), 41 deletions(-) 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 71197ef6..abcf9d98 100644 --- a/detectmatelibrary_tests/test_utils/test_persistency_saver.py +++ b/detectmatelibrary_tests/test_utils/test_persistency_saver.py @@ -373,6 +373,45 @@ def worker(): 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): @@ -388,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 083c9274..ba9ae18c 100644 --- a/src/detectmatelibrary/utils/persistency/event_persistency.py +++ b/src/detectmatelibrary/utils/persistency/event_persistency.py @@ -34,9 +34,9 @@ def __init__( self.event_templates: Dict[int | str, str] = {} self._events_since_save: int = 0 self._on_ingest_callbacks: list[Callable[[], None]] = [] - # ponytail: RLock (not Lock) — ingest fires the on-ingest callback, - # which may re-enter via PersistencySaver.save() on the same thread. - # Shared with PersistencySaver so ingest/save/load are mutually exclusive. + # 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( @@ -49,21 +49,22 @@ def ingest_event( """Ingest event data into the appropriate EventData store.""" with self._lock: self._events_since_save += 1 - 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) + 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() @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 dd60e719..f2053f0c 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() @@ -210,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. From 2510b4557fe9b37b1000e9196dc21aaa291fb9ea Mon Sep 17 00:00:00 2001 From: viktorbeck98 Date: Wed, 1 Jul 2026 15:32:52 +0200 Subject: [PATCH 3/3] Document why _check_event_count is intentionally lock-free Code review flagged the unsynchronized read of _events_since_save as looking like a bug. It is intentional: firing the on-ingest callback outside the ingest lock is what keeps save()'s file I/O off that lock. Comment names the TOCTOU (late/redundant save, never a missed save or lost event) and warns against re-adding a lock. Co-Authored-By: Claude Opus 4.8 --- .../utils/persistency/persistency_saver.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/detectmatelibrary/utils/persistency/persistency_saver.py b/src/detectmatelibrary/utils/persistency/persistency_saver.py index f2053f0c..6cc45e46 100644 --- a/src/detectmatelibrary/utils/persistency/persistency_saver.py +++ b/src/detectmatelibrary/utils/persistency/persistency_saver.py @@ -304,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