Lightweight, fail-open in-memory log recorder. Safety properties: - bounded memory via deque(maxlen=...) - no file/network I/O - swallow-all-errors in emit() - does not log from inside itself - stores only small text snapshots
| 89 | |
| 90 | |
| 91 | class InMemoryDebugRecorder(logging.Handler): |
| 92 | """Lightweight, fail-open in-memory log recorder. |
| 93 | |
| 94 | Safety properties: |
| 95 | - bounded memory via deque(maxlen=...) |
| 96 | - no file/network I/O |
| 97 | - swallow-all-errors in emit() |
| 98 | - does not log from inside itself |
| 99 | - stores only small text snapshots |
| 100 | """ |
| 101 | |
| 102 | def __init__(self, *, capacity: int = LOG_QUEUE_MAXLEN, level: int = logging.DEBUG): |
| 103 | super().__init__(level=level) |
| 104 | self._records: deque[RecordedLog] = deque(maxlen=max(1, int(capacity))) |
| 105 | self._lock = threading.Lock() |
| 106 | self._dropped = 0 |
| 107 | |
| 108 | @property |
| 109 | def dropped_count(self) -> int: |
| 110 | with self._lock: |
| 111 | return self._dropped |
| 112 | |
| 113 | def emit(self, record: logging.LogRecord) -> None: |
| 114 | try: |
| 115 | # Never call logging from here. |
| 116 | # Never inspect application objects. |
| 117 | msg = self._safe_message(record) |
| 118 | exc_text = self._safe_exception_text(record) |
| 119 | |
| 120 | snap = RecordedLog( |
| 121 | created=float(getattr(record, "created", 0.0) or 0.0), |
| 122 | level=str(getattr(record, "levelname", "UNKNOWN")), |
| 123 | logger_name=str(getattr(record, "name", "")), |
| 124 | message=msg, |
| 125 | exc_text=exc_text, |
| 126 | ) |
| 127 | |
| 128 | with self._lock: |
| 129 | self._records.append(snap) |
| 130 | |
| 131 | except Exception: |
| 132 | # Fail open: never let diagnostics interfere with runtime behavior. |
| 133 | try: |
| 134 | self._dropped += 1 |
| 135 | except Exception: |
| 136 | pass |
| 137 | |
| 138 | def clear(self) -> None: |
| 139 | try: |
| 140 | with self._lock: |
| 141 | self._records.clear() |
| 142 | self._dropped = 0 |
| 143 | except Exception: |
| 144 | pass |
| 145 | |
| 146 | def snapshot(self) -> list[RecordedLog]: |
| 147 | try: |
| 148 | with self._lock: |