fix: prevent watcher ingestion starvation #613
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -24,6 +24,7 @@ | |
| import tempfile | ||
| import threading | ||
| import time | ||
| from collections.abc import Set as AbstractSet | ||
| from dataclasses import dataclass | ||
| from datetime import datetime, timezone | ||
| from pathlib import Path | ||
|
|
@@ -44,6 +45,32 @@ | |
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
| _WATCH_MAX_FILE_BYTES_ENV = "BRAINLAYER_WATCH_MAX_FILE_BYTES" | ||
| _DEFAULT_WATCH_MAX_FILE_BYTES = 100 * 1024 * 1024 | ||
|
|
||
|
|
||
| def _watch_max_file_bytes() -> int: | ||
| raw_value = os.environ.get(_WATCH_MAX_FILE_BYTES_ENV, str(_DEFAULT_WATCH_MAX_FILE_BYTES)) | ||
| try: | ||
| parsed_value = int(raw_value) | ||
| except ValueError: | ||
| logger.warning( | ||
| "Invalid %s=%r; using default %d", | ||
| _WATCH_MAX_FILE_BYTES_ENV, | ||
| raw_value, | ||
| _DEFAULT_WATCH_MAX_FILE_BYTES, | ||
| ) | ||
| return _DEFAULT_WATCH_MAX_FILE_BYTES | ||
| if parsed_value < 0: | ||
| logger.warning( | ||
| "Invalid %s=%r; using default %d", | ||
| _WATCH_MAX_FILE_BYTES_ENV, | ||
| raw_value, | ||
| _DEFAULT_WATCH_MAX_FILE_BYTES, | ||
| ) | ||
| return _DEFAULT_WATCH_MAX_FILE_BYTES | ||
| return parsed_value | ||
|
|
||
|
|
||
| @dataclass(frozen=True) | ||
| class WatchRoot: | ||
|
|
@@ -499,15 +526,15 @@ def _has_unavailable_symlink_ancestor(candidate: Path, root: Path) -> bool: | |
| return False | ||
|
|
||
| @staticmethod | ||
| def _has_live_parent_evidence(candidate: Path, live_parent_dirs: AbstractSet[Path]) -> bool: | ||
| """Require a live transcript in the tracked file's containing directory.""" | ||
| parent = candidate.parent | ||
| try: | ||
| if not parent.is_dir(): | ||
| return False | ||
| except OSError: | ||
| return False | ||
| return parent in live_parent_dirs | ||
|
|
||
| @property | ||
| def last_prune_complete(self) -> bool: | ||
|
|
@@ -530,14 +557,13 @@ def prune_missing_files( | |
| live_files.append(candidate) | ||
| except OSError: | ||
| continue | ||
| live_parent_dirs = {parent for live_file in live_files for parent in live_file.parents} | ||
|
|
||
| root_availability: dict[Path, bool] = {} | ||
| for root in active_roots: | ||
| candidate_root = Path(os.path.abspath(os.path.expanduser(str(root)))) | ||
| try: | ||
| root_availability[candidate_root] = candidate_root.is_dir() and candidate_root in live_parent_dirs | ||
| except OSError: | ||
| root_availability[candidate_root] = False | ||
|
|
||
|
|
@@ -555,7 +581,7 @@ def prune_missing_files( | |
| if any(self._has_unavailable_symlink_ancestor(candidate, root) for root in most_specific_roots): | ||
| self._last_prune_complete = False | ||
| continue | ||
| if not self._has_live_parent_evidence(candidate, live_parent_dirs): | ||
| self._last_prune_complete = False | ||
| continue | ||
| try: | ||
|
|
@@ -632,6 +658,14 @@ def read_new_lines(self, max_lines: int | None = None) -> list[dict]: | |
|
|
||
| if new_data: | ||
| self._buffer += new_data | ||
| return self.read_buffered_lines(max_lines=max_lines) | ||
|
|
||
| def has_complete_buffered_line(self) -> bool: | ||
| """Return whether an already-read complete record is waiting to be emitted.""" | ||
| return b"\n" in self._buffer | ||
|
|
||
| def read_buffered_lines(self, max_lines: int | None = None) -> list[dict]: | ||
| """Parse complete buffered records without reading more bytes from disk.""" | ||
| lines = [] | ||
|
|
||
| while b"\n" in self._buffer: | ||
|
|
@@ -718,6 +752,11 @@ def flush(self): | |
| if self._buffer: | ||
| self._do_flush() | ||
|
|
||
| def has_buffered_source(self, filepath: str) -> bool: | ||
| """Return whether unconfirmed buffered entries came from filepath.""" | ||
| with self._lock: | ||
| return any(item.get("_source_file") == filepath for item in self._buffer) | ||
|
|
||
| def _do_flush(self): | ||
| """Internal flush — must be called with _lock held.""" | ||
| batch = self._buffer | ||
|
|
@@ -853,8 +892,10 @@ def __init__( | |
| self.poll_interval_s = poll_interval_s | ||
| self.registry_flush_interval_s = registry_flush_interval_s | ||
| self.max_lines_per_file = max(1, max_lines_per_file) | ||
| self.max_file_bytes = _watch_max_file_bytes() | ||
| self._tailers: dict[str, JSONLTailer] = {} | ||
| self._file_providers: dict[str, str] = {} | ||
| self._oversized_files: set[str] = set() | ||
| self._stop = threading.Event() | ||
| self._last_registry_flush = time.monotonic() | ||
| self.health_path = Path(health_path).expanduser() if health_path else None | ||
|
|
@@ -938,6 +979,27 @@ def _normalize_lines(self, filepath: str, new_lines: list[dict]) -> list[dict]: | |
| normalized.append(entry) | ||
| return normalized | ||
|
|
||
| def _checkpoint_discarded_progress( | ||
| self, | ||
| filepath: str, | ||
| read_start_offset: int, | ||
| read_end_offset: int, | ||
| normalized_lines: list[dict], | ||
| ) -> None: | ||
| """Confirm intentionally discarded bytes without crossing indexable work.""" | ||
| required_confirmed_offset = max( | ||
| (line["_line_end_offset"] for line in normalized_lines if isinstance(line.get("_line_end_offset"), int)), | ||
| default=read_start_offset, | ||
| ) | ||
| if read_end_offset <= required_confirmed_offset: | ||
| return | ||
| if self.indexer.has_buffered_source(filepath): | ||
| return | ||
| confirmed_offset, _confirmed_inode = self.registry.get(filepath) | ||
| if confirmed_offset < required_confirmed_offset: | ||
| return | ||
| self._advance_confirmed_offsets({filepath: read_end_offset}) | ||
|
|
||
| def _max_offset_lag_bytes(self, files: list[str]) -> int: | ||
| max_lag = 0 | ||
| for filepath in files: | ||
|
|
@@ -1098,6 +1160,110 @@ def _ensure_tailer(self, filepath: str) -> JSONLTailer: | |
| self._tailers[filepath] = tailer | ||
| return tailer | ||
|
|
||
| def _handle_rewind( | ||
| self, | ||
| filepath: str, | ||
| old_offset: int, | ||
| new_offset: int, | ||
| inode: int, | ||
| ) -> None: | ||
| """Persist a rewind and notify archival consumers.""" | ||
| session_id = Path(filepath).stem | ||
| self.registry.mark_rewind(filepath, inode) | ||
| logger.warning( | ||
| "Checkpoint restore: %s (offset %d → %d)", | ||
| session_id, | ||
| old_offset, | ||
| new_offset, | ||
| ) | ||
| try: | ||
| from .telemetry import emit | ||
|
|
||
| emit( | ||
| "brainlayer-watcher", | ||
| { | ||
| "_type": "rewind_detected", | ||
| "session_id": session_id, | ||
| "file_path": filepath, | ||
| "old_offset": old_offset, | ||
| "new_offset": new_offset, | ||
| }, | ||
| ) | ||
| except Exception: | ||
| pass | ||
|
|
||
| if self.on_rewind: | ||
| try: | ||
| self.on_rewind(filepath, session_id, old_offset, new_offset) | ||
| except Exception as e: | ||
| logger.error("Rewind callback failed: %s", e) | ||
|
|
||
| def _skip_oversized_file(self, filepath: str) -> bool: | ||
| if self.max_file_bytes <= 0: | ||
| self._oversized_files.discard(filepath) | ||
| return False | ||
|
|
||
| try: | ||
| file_stat = os.stat(filepath) | ||
| except OSError: | ||
| return False | ||
|
|
||
| tailer = self._tailers.get(filepath) | ||
| registry_offset, registry_inode = self.registry.get(filepath) | ||
| tailer_offset = tailer.offset if tailer else registry_offset | ||
| rewind_old_offset = tailer_offset | ||
| file_rewound = file_stat.st_size < tailer_offset | ||
| inode_changed = registry_inode != 0 and registry_inode != file_stat.st_ino | ||
| offset = 0 if inode_changed or file_rewound else registry_offset | ||
| pending_bytes = max(file_stat.st_size - offset, 0) | ||
|
macroscopeapp[bot] marked this conversation as resolved.
|
||
| if pending_bytes <= self.max_file_bytes: | ||
| self._oversized_files.discard(filepath) | ||
| return False | ||
|
|
||
| if tailer is not None and not inode_changed and not file_rewound and tailer_offset > registry_offset: | ||
| if self.indexer.has_buffered_source(filepath): | ||
| self.indexer.flush() | ||
| confirmed_offset, confirmed_inode = self.registry.get(filepath) | ||
| if confirmed_inode != file_stat.st_ino or confirmed_offset < tailer_offset: | ||
|
There was a problem hiding this comment.
When a tailer has advanced past records that Useful? React with 👍 / 👎.
Owner
Author
There was a problem hiding this comment. Fixed in 9be818a. The watcher now checkpoints intentionally discarded progress only when the registry has already confirmed every preceding indexable record and no source entry remains buffered. It therefore prevents the cap from deferring forever after dropped-only records without ever jumping past mixed-stream unconfirmed work. RED/GREEN regressions cover both cases; watcher suite 96 passed, full pre-push gate 3610 passed. |
||
| if filepath not in self._oversized_files: | ||
| logger.error( | ||
| "Oversized JSONL checkpoint deferred for unconfirmed entries: %s", | ||
| filepath, | ||
| ) | ||
| self._oversized_files.add(filepath) | ||
| return True | ||
| offset = confirmed_offset | ||
| pending_bytes = max(file_stat.st_size - offset, 0) | ||
| if pending_bytes <= self.max_file_bytes: | ||
| self._oversized_files.discard(filepath) | ||
| return False | ||
|
|
||
| self._tailers.pop(filepath, None) | ||
| if file_rewound: | ||
| self._handle_rewind( | ||
| filepath, | ||
| rewind_old_offset, | ||
| file_stat.st_size, | ||
| file_stat.st_ino, | ||
| ) | ||
| self.registry.set(filepath, file_stat.st_size, file_stat.st_ino) | ||
|
There was a problem hiding this comment.
If a previous poll read entries into Useful? React with 👍 / 👎.
Owner
Author
There was a problem hiding this comment. Fixed in 794d5ce. Before an oversized EOF checkpoint, the watcher now force-flushes buffered entries from that source and defers checkpointing while any remain retained/unconfirmed. Added a RED→GREEN flush-failure test proving the registry stays at its prior offset and the buffered item remains available. Watcher suite is 92 passed; full mandatory pre-push gate passed. There was a problem hiding this comment.
When a tailer has read past the last confirmed registry offset (for example after a partial flush or retained flush failure), this checkpoints the registry directly to EOF for an oversized append. That advances durable progress past bytes that were only parsed in memory and not confirmed by Useful? React with 👍 / 👎.
Owner
Author
There was a problem hiding this comment. Addressed in 794d5ce by the same pending-entry guard: if this source has buffered entries, the watcher force-flushes them and rechecks the buffer; while any remain retained after a full or partial flush failure, it returns without advancing the registry. The RED→GREEN regression asserts the prior registry offset and retained item are preserved. Successful production flushes return confirmed watermarks through
Comment on lines
+1242
to
+1249
There was a problem hiding this comment.
When a tracked transcript shrinks but the replacement content is still over Useful? React with 👍 / 👎.
Owner
Author
There was a problem hiding this comment. Fixed in 3c5f70a. Rewind handling is now centralized so both ordinary reads and oversized checkpoints execute the same registry generation, telemetry, and |
||
| if not self.registry.flush(): | ||
| logger.error( | ||
| "Oversized JSONL checkpoint could not be persisted immediately: %s", | ||
| filepath, | ||
| ) | ||
| if filepath not in self._oversized_files: | ||
| logger.warning( | ||
| "Oversized JSONL checkpointed and skipped: %s pending_bytes=%d max_file_bytes=%d offset=%d size=%d", | ||
| filepath, | ||
| pending_bytes, | ||
| self.max_file_bytes, | ||
| offset, | ||
| file_stat.st_size, | ||
| ) | ||
| self._oversized_files.add(filepath) | ||
| return True | ||
|
|
||
|
coderabbitai[bot] marked this conversation as resolved.
|
||
| def poll_once(self) -> int: | ||
| """Run one poll cycle. Returns number of new lines found.""" | ||
| total_new = 0 | ||
|
|
@@ -1106,6 +1272,7 @@ def poll_once(self) -> int: | |
|
|
||
| try: | ||
| files = self._discover_jsonl_files() | ||
| self._oversized_files.intersection_update(filepath for filepath in files if not is_denylisted(filepath)) | ||
| if not self._offset_prune_complete: | ||
| pruned = self.registry.prune_missing_files( | ||
| [root.resolved_path for root in self.watch_roots], | ||
|
|
@@ -1128,54 +1295,54 @@ def poll_once(self) -> int: | |
| self.registry.remove(filepath) | ||
| continue | ||
| try: | ||
| tailer = self._tailers.get(filepath) | ||
| drain_buffer = False | ||
| if tailer is not None and tailer.has_complete_buffered_line(): | ||
| _registry_offset, registry_inode = self.registry.get(filepath) | ||
| inode_changed = registry_inode != 0 and registry_inode != tailer.get_inode() | ||
| if not inode_changed and not tailer.check_rewind(): | ||
| drain_buffer = True | ||
| elif tailer.rewound: | ||
| self._handle_rewind( | ||
| filepath, | ||
| tailer.rewind_old_offset, | ||
| tailer.rewind_new_offset, | ||
| tailer.get_inode(), | ||
| ) | ||
| tailer.rewound = False | ||
|
|
||
| if drain_buffer: | ||
| read_start_offset = tailer.offset | ||
| new_lines = tailer.read_buffered_lines(max_lines=self.max_lines_per_file) | ||
| else: | ||
| if self._skip_oversized_file(filepath): | ||
|
There was a problem hiding this comment.
When a tailer has an unterminated buffered JSONL record, Useful? React with 👍 / 👎. |
||
| continue | ||
| tailer = self._ensure_tailer(filepath) | ||
| read_start_offset = tailer.offset | ||
| new_lines = tailer.read_new_lines(max_lines=self.max_lines_per_file) | ||
|
|
||
| # Handle rewind detection (checkpoint restore) | ||
| if tailer.rewound: | ||
| read_start_offset = 0 | ||
| self._handle_rewind( | ||
| filepath, | ||
| tailer.rewind_old_offset, | ||
| tailer.rewind_new_offset, | ||
| tailer.get_inode(), | ||
| ) | ||
| tailer.rewound = False # Reset flag | ||
|
|
||
| normalized_lines = self._normalize_lines(filepath, new_lines) if new_lines else [] | ||
| if normalized_lines: | ||
| self.indexer.add(normalized_lines) | ||
| self._health_entries_seen += len(normalized_lines) | ||
| total_new += len(normalized_lines) | ||
| self._checkpoint_discarded_progress( | ||
| filepath, | ||
| read_start_offset, | ||
| tailer.offset, | ||
| normalized_lines, | ||
| ) | ||
| except Exception: | ||
| logger.exception("Poll file error: %s", filepath) | ||
|
|
||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
When
max_lines_per_filelimits a read,JSONLTailer.read_new_lines()can leave complete, unflushed records intailer._bufferwhiletailer.offsetonly reflects the emitted lines. If that file then grows pastBRAINLAYER_WATCH_MAX_FILE_BYTES, this uses onlytailer.offset, sotailer_offset == registry_offsetand the oversized path below can pop the tailer and checkpoint tofile_stat.st_size, silently dropping buffered records that were never normalized or flushed. This affects hot transcripts that cross the byte cap between polls; the skip decision needs to account for buffered/unemitted bytes before advancing the registry to EOF.Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Fixed in 7890d18. The watcher now separates parsing already-buffered complete records from reading additional file bytes. When a tailer has buffered records, poll_once drains them under max_lines_per_file without calling f.read(), while preserving inode-replacement and rewind handling. RED reproduced the EOF checkpoint/data loss; GREEN: 94 watcher tests and the full pre-push gate (3608 passed, 9 skipped, 1 xfailed).