From 59c5c32b5b016f2f163703a0b17c5a202c486e79 Mon Sep 17 00:00:00 2001 From: Jeroen Schweitzer Date: Mon, 15 Jun 2026 10:15:49 +0200 Subject: [PATCH] feat(watchdog): dedicated-isolate heartbeat + resource sampler (T-435) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A main-isolate Timer would freeze WITH the main isolate and tell us nothing, so the watchdog runs in its own isolate: it fsyncs a heartbeat every ~500ms (so the last on-disk heartbeat bounds a freeze to ~500ms) and every ~2s samples this process's thread / handle-or-fd / child-host / RSS counts. A monotonically climbing child or thread count is the leak signature the soak couldn't reproduce on CI but a real freeze would show. Output is JSON-lines in clide-watchdog.log, bounded by the same truncate-on-cap scheme as the crumb files. - watchdog.dart (Flutter-free, tested): ResourceSample, ResourceSampler (forPlatform), PosixResourceSampler (/proc/self: Threads, fd count, task children, ProcessInfo.currentRss), WatchdogFile (bounded fsynced JSON-lines), runWatchdog (the loop, bounded by maxTicks for tests), watchdogEntry (the sendable Isolate.spawn entry). - watchdog_windows.dart (coverage:ignore — Win32 FFI, validated only at runtime on Windows): one CreateToolhelp32Snapshot for thread + conhost/OpenConsole child count, GetProcessHandleCount, ProcessInfo.currentRss. Exhaustively defensive: any failure yields a -1 field, snapshot handle always closed, never throws. - main.dart: spawn the watchdog at boot (desktop only), non-fatal. Per-line fsync means the OS reaping the isolate at exit loses nothing. Tests: ResourceSample.toJson, Posix sampler against real /proc, WatchdogFile (JSON shape, cap, disabled), runWatchdog (immediate baseline tick). Coverage gate 95.08%. Co-Authored-By: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 5 + lib/kernel/kernel.dart | 1 + lib/kernel/src/watchdog.dart | 201 +++++++++++++++++++++++++++ lib/kernel/src/watchdog_windows.dart | 141 +++++++++++++++++++ lib/main.dart | 9 ++ test/kernel/watchdog_test.dart | 102 ++++++++++++++ 6 files changed, 459 insertions(+) create mode 100644 lib/kernel/src/watchdog.dart create mode 100644 lib/kernel/src/watchdog_windows.dart create mode 100644 test/kernel/watchdog_test.dart diff --git a/CHANGELOG.md b/CHANGELOG.md index 5170ed8a..d88d7a85 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,11 @@ heading, and (b) bumping `pubspec.yaml` `version:` in the same commit. so a wedged isolate's last crumb survives a freeze that also froze the main isolate — naming the wedge after the fact. Per-syscall crumbs at debug level. (T-434) +- **Crash-diagnostic watchdog.** A dedicated isolate fsyncs a heartbeat every + ~500ms (bounding a freeze to ~500ms) and every ~2s samples this process's + thread / handle / child-host / RSS counts to `clide-watchdog.log` — a climbing + child or thread count is the leak signature. Survives a frozen main isolate; + spawn failure is non-fatal. (T-435) ## [2.5.0] — 2026-06-14 diff --git a/lib/kernel/kernel.dart b/lib/kernel/kernel.dart index 36aec01b..6cdd67ff 100644 --- a/lib/kernel/kernel.dart +++ b/lib/kernel/kernel.dart @@ -18,6 +18,7 @@ export 'src/events/types.dart'; export 'src/ipc/client.dart'; export 'src/log.dart'; export 'src/file_log_sink.dart'; +export 'src/watchdog.dart'; export 'src/settings.dart'; export 'src/facade.dart'; export 'src/clipboard.dart'; diff --git a/lib/kernel/src/watchdog.dart b/lib/kernel/src/watchdog.dart new file mode 100644 index 00000000..a76b7df7 --- /dev/null +++ b/lib/kernel/src/watchdog.dart @@ -0,0 +1,201 @@ +/// Crash-diagnostic watchdog (T-435, under the T-425 observability epic). +/// +/// The Windows freeze leaves no evidence partly because it is a *whole-process* +/// stall: a main-isolate `Timer` heartbeat would freeze WITH the main isolate +/// and tell us nothing. So the watchdog runs in a DEDICATED isolate that: +/// +/// - appends + fsyncs a heartbeat every ~500ms, so the last heartbeat on disk +/// bounds a freeze to ~500ms ("it was alive at T, dead by T+0.5s"); and +/// - every ~2s samples this process's resource counts — threads, open +/// handles/fds, child/ConPTY-host processes, RSS — and appends + fsyncs +/// them. A monotonically climbing child/thread/handle count is the leak +/// signature the soak couldn't reproduce on CI but a real freeze would show. +/// +/// Output is JSON-lines in `logDirectory()/clide-watchdog.log`, matching +/// [FileLogSink] so it greps/parses the same way, bounded by the same +/// truncate-on-cap scheme as [IsolateCrumbFile]. Everything is synchronous and +/// swallows its own errors — the watchdog must never add a second hang or take +/// the app down. +/// +/// Flutter-free (dart:io / dart:isolate / dart:convert + a thin FFI sampler on +/// Windows) so the entry point is `Isolate.spawn`-able and it unit-tests under +/// `dart test`. +library; + +import 'dart:convert'; +import 'dart:io'; + +import 'watchdog_windows.dart'; + +/// One resource sample of the current process. A field of `-1` means "not +/// available on this platform or the probe failed" — never an error. +class ResourceSample { + const ResourceSample({this.threads = -1, this.handles = -1, this.children = -1, this.rssBytes = -1}); + + /// Live OS thread count (culprit #2: blocked-FFI isolate threads piling up). + final int threads; + + /// Open handle count (Windows) / open fd count (POSIX). + final int handles; + + /// Child / ConPTY-host process count (the orphan-accumulation leak signature). + final int children; + + /// Resident set size in bytes. + final int rssBytes; + + Map toJson() => { + if (threads >= 0) 'threads': threads, + if (handles >= 0) 'handles': handles, + if (children >= 0) 'children': children, + if (rssBytes >= 0) 'rssMB': (rssBytes / (1024 * 1024)).round(), + }; +} + +/// Samples the CURRENT process's resource counts. Cheap, synchronous, never +/// throws (returns `-1` fields on failure). +abstract class ResourceSampler { + ResourceSample sample(); + + /// The backend for the running OS — `/proc` on POSIX, a thin Win32 FFI + /// snapshot on Windows. + static ResourceSampler forPlatform() => Platform.isWindows ? WindowsResourceSampler() : PosixResourceSampler(); +} + +/// POSIX sampler — reads `/proc/self`. Runs (and is tested) on the Linux CI box. +class PosixResourceSampler implements ResourceSampler { + @override + ResourceSample sample() => ResourceSample(threads: _threads(), handles: _fdCount(), children: _childCount(), rssBytes: _rss()); + + int _threads() { + try { + for (final line in File('/proc/self/status').readAsLinesSync()) { + if (line.startsWith('Threads:')) return int.parse(line.split(RegExp(r'\s+'))[1]); + } + } catch (_) {} + return -1; + } + + int _fdCount() { + try { + return Directory('/proc/self/fd').listSync().length; + } catch (_) { + return -1; + } + } + + int _childCount() { + // Sum each thread's direct-children list (`/proc//task//children`, + // Linux 5.3+). Best-effort: absent file / old kernel → 0 from that thread. + try { + var n = 0; + for (final task in Directory('/proc/self/task').listSync()) { + final f = File('${task.path}/children'); + if (!f.existsSync()) continue; + final s = f.readAsStringSync().trim(); + if (s.isNotEmpty) n += s.split(RegExp(r'\s+')).length; + } + return n; + } catch (_) { + return -1; + } + } + + int _rss() { + try { + return ProcessInfo.currentRss; + } catch (_) { + return -1; + } + } +} + +/// The watchdog's on-disk writer: bounded, synchronously-fsynced JSON-lines +/// (heartbeats + samples). Separated from the loop so it unit-tests in +/// isolation. Mirrors [IsolateCrumbFile]'s truncate-on-cap bound. +class WatchdogFile { + WatchdogFile(String? path, {int capBytes = 256 * 1024}) : _capBytes = capBytes { + if (path == null) return; + try { + final f = File(path); + f.parent.createSync(recursive: true); + final raf = f.openSync(mode: FileMode.append); + _raf = raf; + _size = raf.lengthSync(); + } catch (_) { + _raf = null; + } + } + + final int _capBytes; + RandomAccessFile? _raf; + int _size = 0; + + bool get enabled => _raf != null; + + void heartbeat() => _write({'ts': _now(), 'evt': 'hb'}); + + void sample(ResourceSample s) => _write({'ts': _now(), 'evt': 'sample', 'pid': pid, ...s.toJson()}); + + String _now() => DateTime.now().toUtc().toIso8601String(); + + void _write(Map json) { + final raf = _raf; + if (raf == null) return; + try { + if (_size >= _capBytes) { + raf.truncateSync(0); + raf.setPositionSync(0); + _size = 0; + } + final bytes = utf8.encode('${jsonEncode(json)}\n'); + raf.writeFromSync(bytes); + raf.flushSync(); + _size += bytes.length; + } catch (_) {} + } + + void close() { + try { + _raf?.flushSync(); + _raf?.closeSync(); + } catch (_) {} + _raf = null; + } +} + +/// The watchdog loop. Extracted from [watchdogEntry] so a test can bound it +/// with [maxTicks]; production passes null and the loop runs until the isolate +/// is killed at shutdown. Heartbeats fire every [hbIntervalMs], samples every +/// [sampleIntervalMs]; a short sleep between keeps the cadence without spinning. +void runWatchdog(WatchdogFile file, ResourceSampler sampler, {required int hbIntervalMs, required int sampleIntervalMs, int? maxTicks}) { + if (!file.enabled) return; + final sw = Stopwatch()..start(); + // Seed both "last" markers a full interval in the past so the first tick + // emits an immediate heartbeat + sample (a baseline at startup). + var lastHb = -hbIntervalMs; + var lastSample = -sampleIntervalMs; + var ticks = 0; + while (maxTicks == null || ticks < maxTicks) { + final e = sw.elapsedMilliseconds; + if (e - lastHb >= hbIntervalMs) { + file.heartbeat(); + lastHb = e; + } + if (e - lastSample >= sampleIntervalMs) { + file.sample(sampler.sample()); + lastSample = e; + } + ticks++; + if (maxTicks != null && ticks >= maxTicks) break; + sleep(const Duration(milliseconds: 25)); + } + file.close(); +} + +/// Top-level entry for `Isolate.spawn`. Args are a sendable tuple — the log +/// path (not a Logger; isolates can't share one) and the two intervals in ms. +void watchdogEntry((String, int, int) msg) { + final (logPath, hbMs, sampleMs) = msg; + runWatchdog(WatchdogFile(logPath), ResourceSampler.forPlatform(), hbIntervalMs: hbMs, sampleIntervalMs: sampleMs); +} diff --git a/lib/kernel/src/watchdog_windows.dart b/lib/kernel/src/watchdog_windows.dart new file mode 100644 index 00000000..d2c98dff --- /dev/null +++ b/lib/kernel/src/watchdog_windows.dart @@ -0,0 +1,141 @@ +// coverage:ignore-file +// +// Windows-only resource sampler for the watchdog (T-435). All of it is Win32 +// FFI through kernel32/psapi, so it cannot execute on the Linux CI runner that +// produces the coverage report (PosixResourceSampler is used there). It is +// validated only when the app actually runs on Windows — which is acceptable +// because it is a DIAGNOSTIC that reads, never mutates, and is exhaustively +// defensive: every probe is wrapped so any failure yields a `-1` field rather +// than an exception, and the toolhelp snapshot handle is always closed. A +// missing sample is fine; a sampler that throws or leaks would not be. +library; + +import 'dart:ffi' as ffi; +import 'dart:io' show ProcessInfo; + +import 'package:ffi/ffi.dart'; + +import 'watchdog.dart'; + +const int _kTh32csSnapprocess = 0x00000002; + +final ffi.DynamicLibrary _k32 = ffi.DynamicLibrary.open('kernel32.dll'); +final ffi.DynamicLibrary _psapi = ffi.DynamicLibrary.open('psapi.dll'); + +final _getCurrentProcess = _k32.lookupFunction Function(), ffi.Pointer Function()>('GetCurrentProcess'); +final _getCurrentProcessId = _k32.lookupFunction('GetCurrentProcessId'); +final _createToolhelp32Snapshot = _k32.lookupFunction Function(ffi.Uint32, ffi.Uint32), ffi.Pointer Function(int, int)>( + 'CreateToolhelp32Snapshot', +); +final _process32First = _k32 + .lookupFunction, ffi.Pointer<_ProcessEntry32>), int Function(ffi.Pointer, ffi.Pointer<_ProcessEntry32>)>( + 'Process32First', + ); +final _process32Next = _k32 + .lookupFunction, ffi.Pointer<_ProcessEntry32>), int Function(ffi.Pointer, ffi.Pointer<_ProcessEntry32>)>( + 'Process32Next', + ); +final _closeHandle = _k32.lookupFunction), int Function(ffi.Pointer)>('CloseHandle'); +final _getProcessHandleCount = _psapi + .lookupFunction, ffi.Pointer), int Function(ffi.Pointer, ffi.Pointer)>( + 'GetProcessHandleCount', + ); + +/// Win32 `PROCESSENTRY32` (ANSI). szExeFile is `CHAR[MAX_PATH]`. +final class _ProcessEntry32 extends ffi.Struct { + @ffi.Uint32() + external int dwSize; + @ffi.Uint32() + external int cntUsage; + @ffi.Uint32() + external int th32ProcessID; + @ffi.IntPtr() + external int th32DefaultHeapID; + @ffi.Uint32() + external int th32ModuleID; + @ffi.Uint32() + external int cntThreads; + @ffi.Uint32() + external int th32ParentProcessID; + @ffi.Int32() + external int pcPriClassBase; + @ffi.Uint32() + external int dwFlags; + @ffi.Array(260) + external ffi.Array szExeFile; +} + +/// Samples the current process via a single toolhelp snapshot (thread count + +/// ConPTY-host children) plus GetProcessHandleCount and ProcessInfo.currentRss. +class WindowsResourceSampler implements ResourceSampler { + @override + ResourceSample sample() { + final (threads, children) = _snapshotThreadsAndHosts(); + return ResourceSample(threads: threads, children: children, handles: _handleCount(), rssBytes: _rss()); + } + + /// One toolhelp snapshot → (this process's thread count, count of its direct + /// conhost/OpenConsole children). Both `-1`/unavailable on any failure. + (int, int) _snapshotThreadsAndHosts() { + var threads = -1; + var conhosts = 0; + var sawAny = false; + ffi.Pointer? snap; + final entry = calloc<_ProcessEntry32>(); + try { + final myPid = _getCurrentProcessId(); + snap = _createToolhelp32Snapshot(_kTh32csSnapprocess, 0); + entry.ref.dwSize = ffi.sizeOf<_ProcessEntry32>(); + var ok = _process32First(snap, entry); + while (ok != 0) { + sawAny = true; + if (entry.ref.th32ProcessID == myPid) threads = entry.ref.cntThreads; + if (entry.ref.th32ParentProcessID == myPid) { + final name = _exeName(entry.ref.szExeFile).toLowerCase(); + if (name == 'conhost.exe' || name == 'openconsole.exe') conhosts++; + } + ok = _process32Next(snap, entry); + } + } catch (_) { + // any FFI failure → unavailable, not a crash + } finally { + if (snap != null) { + try { + _closeHandle(snap); + } catch (_) {} + } + calloc.free(entry); + } + return (threads, sawAny ? conhosts : -1); + } + + int _handleCount() { + final out = calloc(); + try { + final ok = _getProcessHandleCount(_getCurrentProcess(), out); + return ok != 0 ? out.value : -1; + } catch (_) { + return -1; + } finally { + calloc.free(out); + } + } + + int _rss() { + try { + return ProcessInfo.currentRss; + } catch (_) { + return -1; + } + } + + String _exeName(ffi.Array arr) { + final bytes = []; + for (var i = 0; i < 260; i++) { + final b = arr[i]; + if (b == 0) break; + bytes.add(b); + } + return String.fromCharCodes(bytes); + } +} diff --git a/lib/main.dart b/lib/main.dart index 57ed9520..8ca4cea1 100644 --- a/lib/main.dart +++ b/lib/main.dart @@ -1,4 +1,5 @@ import 'dart:async'; +import 'dart:isolate'; import 'package:clide/app.dart'; import 'package:clide/test_app.dart'; @@ -110,6 +111,14 @@ Future main() async { settingValue: bootSettings.get('app.log.level'), ); bootLogSinks = [FileLogSink(dir: Directory(logDirectory())).call]; + // Crash-diagnostic watchdog in its own isolate (T-435): heartbeats + + // resource samples that survive a frozen main isolate. Non-fatal — a + // leak-detector that breaks startup is worse than a missing one. The OS + // reaps the isolate on exit; every line is fsynced, so abrupt death loses + // nothing. + try { + await Isolate.spawn(watchdogEntry, ('${logDirectory()}/clide-watchdog.log', 500, 2000)); + } catch (_) {} } // Resolve toolchain + boot daemon inline — same as Linux. diff --git a/test/kernel/watchdog_test.dart b/test/kernel/watchdog_test.dart new file mode 100644 index 00000000..1b00ace6 --- /dev/null +++ b/test/kernel/watchdog_test.dart @@ -0,0 +1,102 @@ +import 'dart:convert'; +import 'dart:io'; + +import 'package:clide/kernel/kernel.dart'; +import 'package:flutter_test/flutter_test.dart'; + +class _FakeSampler implements ResourceSampler { + const _FakeSampler(); + @override + ResourceSample sample() => const ResourceSample(threads: 7, handles: 42, children: 1, rssBytes: 100 * 1024 * 1024); +} + +void main() { + late Directory dir; + setUp(() => dir = Directory.systemTemp.createTempSync('clide-wd-')); + tearDown(() { + if (dir.existsSync()) dir.deleteSync(recursive: true); + }); + String path() => '${dir.path}${Platform.pathSeparator}wd.log'; + + group('ResourceSample', () { + test('toJson omits unavailable (-1) fields and converts RSS to MB', () { + expect(const ResourceSample(threads: 5, rssBytes: 2 * 1024 * 1024).toJson(), {'threads': 5, 'rssMB': 2}); + expect(const ResourceSample().toJson(), isEmpty); + expect(const ResourceSample(handles: 9, children: 0).toJson(), {'handles': 9, 'children': 0}); + }); + }); + + group('ResourceSampler.forPlatform', () { + test('returns the POSIX sampler off Windows', () { + if (Platform.isWindows) return; + expect(ResourceSampler.forPlatform(), isA()); + }); + }); + + group('PosixResourceSampler', () { + test('reads real /proc counts for this process', () { + if (!Platform.isLinux) return; + final s = PosixResourceSampler().sample(); + expect(s.threads, greaterThan(0)); + expect(s.handles, greaterThan(0)); // at least stdio fds + expect(s.rssBytes, greaterThan(0)); + expect(s.children, greaterThanOrEqualTo(0)); + }); + }); + + group('WatchdogFile', () { + test('heartbeat + sample write tagged JSON lines', () { + WatchdogFile(path()) + ..heartbeat() + ..sample(const ResourceSample(threads: 12, handles: 200, children: 0, rssBytes: 50 * 1024 * 1024)) + ..close(); + + final lines = File(path()).readAsLinesSync(); + expect(lines, hasLength(2)); + expect((jsonDecode(lines[0]) as Map)['evt'], 'hb'); + final s = jsonDecode(lines[1]) as Map; + expect(s['evt'], 'sample'); + expect(s['threads'], 12); + expect(s['rssMB'], 50); + expect(s['pid'], isA()); + }); + + test('null path → disabled, writes are no-ops', () { + final f = WatchdogFile(null); + expect(f.enabled, isFalse); + f + ..heartbeat() + ..sample(const ResourceSample()) + ..close(); + }); + + test('bounded by the size cap', () { + final f = WatchdogFile(path(), capBytes: 200); + for (var i = 0; i < 100; i++) { + f.heartbeat(); + } + f.close(); + expect(File(path()).lengthSync(), lessThan(400)); + }); + }); + + group('runWatchdog', () { + test('emits an immediate heartbeat + sample on the first tick', () { + runWatchdog(WatchdogFile(path()), const _FakeSampler(), hbIntervalMs: 0, sampleIntervalMs: 0, maxTicks: 1); + + final lines = File(path()).readAsLinesSync(); + expect(lines, hasLength(2)); + expect((jsonDecode(lines[0]) as Map)['evt'], 'hb'); + final s = jsonDecode(lines[1]) as Map; + expect(s['evt'], 'sample'); + expect(s['threads'], 7); + expect(s['handles'], 42); + expect(s['children'], 1); + expect(s['rssMB'], 100); + }); + + test('is a no-op when the file is disabled', () { + expect(() => runWatchdog(WatchdogFile(null), const _FakeSampler(), hbIntervalMs: 0, sampleIntervalMs: 0, maxTicks: 5), returnsNormally); + }); + }); +}