From 192814a97651c8be9335aa4843f5e2d003b94586 Mon Sep 17 00:00:00 2001 From: Rafly Hanggaraksa Date: Tue, 30 Jun 2026 08:42:05 -0500 Subject: [PATCH 1/3] fix: eliminate three thread-safety bugs in telemetry delta reporting - active_duration_s was incremented outside any lock in _activity_detector, letting _post_telemetry_delta read it twice with a torn value and permanently lose increments; move the increment inside _counter_lock and snapshot the field once under that lock before computing the delta - dict(rows_written) could raise RuntimeError when concurrent stream-flush threads added new keys mid-iteration; add _rows_written_lock to WriteManager and hold it at both the mutation site and every call site that copies the dict (telemetry delta + manifest snapshot) - skip post_telemetry network call when both duration_delta and events_delta are zero; defer advancing _last_posted_* until we actually post so no interval is silently dropped from the accumulator Co-Authored-By: Claude Sonnet 4.6 --- src/tracer/IOTracer.py | 25 ++++++++++++++++++------- src/tracer/WriterManager.py | 8 +++++--- 2 files changed, 23 insertions(+), 10 deletions(-) diff --git a/src/tracer/IOTracer.py b/src/tracer/IOTracer.py index 921e83f..020e8ab 100644 --- a/src/tracer/IOTracer.py +++ b/src/tracer/IOTracer.py @@ -168,7 +168,7 @@ def __init__( self.automatic_upload = False - self.writer = WriteManager(output_dir, self.upload_manager, automatic_upload) + self.writer = WriteManager(output_dir, self.upload_manager, self.automatic_upload) if self.automatic_upload: self.upload_manager.on_upload_success = self._post_telemetry_delta self.fs_snapper = FilesystemSnapper(self.writer, anonymous) @@ -294,20 +294,29 @@ def _activity_detector(self) -> None: with self._counter_lock: count = self._disk_event_counter self._disk_event_counter = 0 - if count > _ACTIVITY_THRESHOLD: - self.active_duration_s += 1 + if count > _ACTIVITY_THRESHOLD: + self.active_duration_s += 1 def _post_telemetry_delta(self) -> None: + with self._counter_lock: + current_active_s = self.active_duration_s + with self._telemetry_lock: - duration_delta = self.active_duration_s - self._last_posted_active_s - self._last_posted_active_s = self.active_duration_s + duration_delta = current_active_s - self._last_posted_active_s + + with self.writer._rows_written_lock: + current_rows = dict(self.writer.rows_written) - current_rows = dict(self.writer.rows_written) events_delta = { k.lower().replace(" ", "_"): current_rows.get(k, 0) - self._last_posted_rows.get(k, 0) for k in current_rows if current_rows.get(k, 0) > self._last_posted_rows.get(k, 0) } + + if duration_delta == 0 and not events_delta: + return + + self._last_posted_active_s = current_active_s self._last_posted_rows = current_rows self.upload_manager.post_telemetry( @@ -1351,6 +1360,8 @@ def _write_manifest(self, started_at, stopped_at=None): """ manifest = schema.schema_for_manifest() duration = (stopped_at - started_at).total_seconds() if stopped_at else None + with self.writer._rows_written_lock: + rows_written_snap = dict(self.writer.rows_written) manifest.update({ "tracer": {"version": self.version}, "machine_id": capture_machine_id(), @@ -1375,7 +1386,7 @@ def _write_manifest(self, started_at, stopped_at=None): "diagnostics": { "attached_probes": self._attached_probes(), "lost_events": dict(self._lost_counts), - "rows_written": dict(self.writer.rows_written), + "rows_written": rows_written_snap, "write_dropped": dict(getattr(self.writer, "write_dropped", {})), "block": self._block_stats(), }, diff --git a/src/tracer/WriterManager.py b/src/tracer/WriterManager.py index 3ad0df5..7874001 100644 --- a/src/tracer/WriterManager.py +++ b/src/tracer/WriterManager.py @@ -84,6 +84,7 @@ def __init__(self, output_dir: str, upload_manager: ObjectStorageManager, automa # Total rows written to disk per stream (keyed by buffer label), for the # session manifest — lets a consumer spot a stream whose probes attached # but produced no events. + self._rows_written_lock = threading.Lock() self.rows_written = {} # Rows that could NOT be persisted (write/flush raised — e.g. ENOSPC / # EDQUOT / closed handle) and were dropped after exceeding the on-error @@ -958,9 +959,10 @@ def _write_buffer_to_file(self, buffer, file_handle, buffer_name: str): + (f" — dropped {dropped} rows past retry cap" if dropped else "")) return - self.rows_written[buffer_name] = ( - self.rows_written.get(buffer_name, 0) + len(drained) - ) + with self._rows_written_lock: + self.rows_written[buffer_name] = ( + self.rows_written.get(buffer_name, 0) + len(drained) + ) def write_to_disk(self): """Write all buffered data to disk using parallel threads.""" From 36359a7d7cf2d3b6c81dcdfc8c0f616b17a52463 Mon Sep 17 00:00:00 2001 From: Rafly Hanggaraksa Date: Tue, 30 Jun 2026 08:56:20 -0500 Subject: [PATCH 2/3] fix: extend _rows_written_lock to cover write_dropped dict write_dropped was mutated on error paths without any lock, meaning two streams failing simultaneously could concurrently add new keys while _write_manifest iterated the dict via dict(), raising RuntimeError. Reuse the existing _rows_written_lock at the mutation site in _write_buffer_to_file and expand the snapshot block in _write_manifest to capture write_dropped alongside rows_written under the same lock. Co-Authored-By: Claude Sonnet 4.6 --- src/tracer/IOTracer.py | 3 ++- src/tracer/WriterManager.py | 7 ++++--- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/src/tracer/IOTracer.py b/src/tracer/IOTracer.py index 020e8ab..b91cf11 100644 --- a/src/tracer/IOTracer.py +++ b/src/tracer/IOTracer.py @@ -1362,6 +1362,7 @@ def _write_manifest(self, started_at, stopped_at=None): duration = (stopped_at - started_at).total_seconds() if stopped_at else None with self.writer._rows_written_lock: rows_written_snap = dict(self.writer.rows_written) + write_dropped_snap = dict(getattr(self.writer, "write_dropped", {})) manifest.update({ "tracer": {"version": self.version}, "machine_id": capture_machine_id(), @@ -1387,7 +1388,7 @@ def _write_manifest(self, started_at, stopped_at=None): "attached_probes": self._attached_probes(), "lost_events": dict(self._lost_counts), "rows_written": rows_written_snap, - "write_dropped": dict(getattr(self.writer, "write_dropped", {})), + "write_dropped": write_dropped_snap, "block": self._block_stats(), }, }) diff --git a/src/tracer/WriterManager.py b/src/tracer/WriterManager.py index 7874001..5cb5d6b 100644 --- a/src/tracer/WriterManager.py +++ b/src/tracer/WriterManager.py @@ -952,9 +952,10 @@ def _write_buffer_to_file(self, buffer, file_handle, buffer_name: str): buffer.popleft() dropped += 1 if dropped: - self.write_dropped[buffer_name] = ( - self.write_dropped.get(buffer_name, 0) + dropped - ) + with self._rows_written_lock: + self.write_dropped[buffer_name] = ( + self.write_dropped.get(buffer_name, 0) + dropped + ) logger("error", f"Error writing {buffer_name} buffer: {e}" + (f" — dropped {dropped} rows past retry cap" if dropped else "")) return From 15f5d220fd5b176f435cda6bb24c5b618448f710 Mon Sep 17 00:00:00 2001 From: Rafly Hanggaraksa Date: Tue, 30 Jun 2026 09:02:18 -0500 Subject: [PATCH 3/3] refactor: encapsulate rows_written/write_dropped snapshots in WriteManager IOTracer was reaching directly into _rows_written_lock and rows_written to build thread-safe snapshots, duplicating the lock pattern in two places. Add get_rows_written_snapshot() and get_counters_snapshot() to WriteManager so callers never touch the internal lock directly. Co-Authored-By: Claude Sonnet 4.6 --- src/tracer/IOTracer.py | 7 ++----- src/tracer/WriterManager.py | 9 +++++++++ 2 files changed, 11 insertions(+), 5 deletions(-) diff --git a/src/tracer/IOTracer.py b/src/tracer/IOTracer.py index b91cf11..d236c2c 100644 --- a/src/tracer/IOTracer.py +++ b/src/tracer/IOTracer.py @@ -304,8 +304,7 @@ def _post_telemetry_delta(self) -> None: with self._telemetry_lock: duration_delta = current_active_s - self._last_posted_active_s - with self.writer._rows_written_lock: - current_rows = dict(self.writer.rows_written) + current_rows = self.writer.get_rows_written_snapshot() events_delta = { k.lower().replace(" ", "_"): current_rows.get(k, 0) - self._last_posted_rows.get(k, 0) @@ -1360,9 +1359,7 @@ def _write_manifest(self, started_at, stopped_at=None): """ manifest = schema.schema_for_manifest() duration = (stopped_at - started_at).total_seconds() if stopped_at else None - with self.writer._rows_written_lock: - rows_written_snap = dict(self.writer.rows_written) - write_dropped_snap = dict(getattr(self.writer, "write_dropped", {})) + rows_written_snap, write_dropped_snap = self.writer.get_counters_snapshot() manifest.update({ "tracer": {"version": self.version}, "machine_id": capture_machine_id(), diff --git a/src/tracer/WriterManager.py b/src/tracer/WriterManager.py index 5cb5d6b..8a06f70 100644 --- a/src/tracer/WriterManager.py +++ b/src/tracer/WriterManager.py @@ -965,6 +965,15 @@ def _write_buffer_to_file(self, buffer, file_handle, buffer_name: str): self.rows_written.get(buffer_name, 0) + len(drained) ) + def get_rows_written_snapshot(self) -> dict[str, int]: + with self._rows_written_lock: + return dict(self.rows_written) + + def get_counters_snapshot(self) -> tuple[dict[str, int], dict[str, int]]: + """Return (rows_written, write_dropped) snapshots under one lock.""" + with self._rows_written_lock: + return dict(self.rows_written), dict(self.write_dropped) + def write_to_disk(self): """Write all buffered data to disk using parallel threads.""" def write_vfs():