Fix/persistency ingest lock#193
Merged
Merged
Conversation
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 <noreply@anthropic.com>
Contributor
There was a problem hiding this comment.
can things can actually also get jumbled while data is exported?
import_state() uses saver and then does things with saver.locked() but export_state does plain persistency.save(),
should the export function also do
if saver is not None:
with saver.locked():
?
|
|
||
| data = data_structure.to_data(all_variables) | ||
| data_structure.add_data(data) | ||
| with self._lock: |
Contributor
There was a problem hiding this comment.
same lock as persistency and event count timer use , so ingest_event now has to wait periodically for the state serialization (necessary) AND file writes. maybe worth moving the file write out of the critical section?
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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
thorinaboenke
approved these changes
Jul 1, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Task
#188
Description
EventPersistency.ingest_event()held no lock, so it could run concurrently with_save()/_load()on other threads:_SaveTimerthread callssave()→_save()iteratesep.events_datawhile a concurrent ingest adds a newevent_id→RuntimeError: dictionary changed size during iteration. This is swallowed bysave()'stry/except, so the symptom is a silently-skipped save (logged as a warning), not a crash._load()rebindsevents_data/event_templates/events_seenwhile ingest reads-then-writes them → torn/lost state. Currently prevented operationally by stopping the engine before/load(the service-api 409 guard).Fix: give
EventPersistencya singlethreading.RLock, acquire it iningest_event(), and havePersistencySavershare the same lock so ingest / save / load are mutually exclusive.RLock(notLock) is required: ingest fires theon_ingestcallback →_check_event_count()→save()on the same thread, which re-acquires the lock. A plainLockself-deadlocks here.PersistencySaver._lockis now aliased topersistency._lock, so existingsave()/load()/locked()call sites are unchanged.Out of scope (intentionally kept):
/load409 guard stays. The lock makes load crash-safe, but load still wholesale-replaces live state — requiring the engine stopped is a semantics decision, not just a crash workaround.self._events_since_save += 1remains unsynchronized (cosmetic save-cadence drift only).Cost: one uncontended
RLockacquire per event — negligible vs the existing msgpack/parquet work inadd_data/dump.How Has This Been Tested?
TestPersistencySaverConcurrency::test_ingest_blocks_while_saver_lock_held(TDD): holds the saver lock on the main thread and asserts a concurrentingest_eventblocks until release. Fails on the old code (ingest_event ran while the saver lock was held), passes with the fix.test_events_until_save_triggers_saveguards the RLock reentrancy regression — it deadlocks under a plainLock, passes underRLock.ingest_event).Checklist