clover's git

the-knife

1	use serde::{Deserialize, Serialize};
2	use std::collections::{HashMap, HashSet, VecDeque};
3	use std::io::{self, BufRead, BufWriter, Write};
4	use std::sync::mpsc::{self, Receiver, RecvTimeoutError};
5	use std::thread;
6	use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
7	use sysinfo::{
8	MINIMUM_CPU_UPDATE_INTERVAL, Pid, ProcessRefreshKind, ProcessesToUpdate, System, UpdateKind,
9	};
10
11	const PROTOCOL_VERSION: u32 = 3;
12	const MIN_SAMPLE_INTERVAL_MS: u64 = 250;
13	const MAX_SAMPLE_INTERVAL_MS: u64 = 60_000;
14	const PROCESS_START_TIME_PRECISION_MS: u64 = 1_000;
15	const HISTORY_RETENTION_MS: u64 = 60 * 60_000;
16	const MAX_HISTORY_SNAPSHOTS: usize = 3_600;
17	const INPUT_QUEUE_CAPACITY: usize = 64;
18	const MAX_HISTORY_RETAINED_ENTRIES: usize = 20_000;
19	const MAX_HISTORY_RETAINED_BYTES: usize = 64 * 1024 * 1024;
20	const MAX_PROCESS_NAME_BYTES: usize = 1_024;
21	const MAX_PROCESS_COMMAND_BYTES: usize = 16 * 1_024;
22	const MAX_PROCESS_STATUS_BYTES: usize = 256;
23	const HISTORY_CHUNK_SNAPSHOTS: usize = 32;
24
25	#[derive(Debug, Clone, Deserialize, Serialize)]
26	#[serde(rename_all = "camelCase")]
27	struct ExternalProcess {
28	pid: u32,
29	#[serde(default)]
30	start_time_ms: Option<u64>,
31	}
32
33	impl ExternalProcess {
34	fn estimated_history_bytes(&self) -> usize {
35	std::mem::size_of::<Self>()
36	}
37	}
38
39	#[derive(Debug, Deserialize)]
40	#[serde(
41	tag = "type",
42	rename_all = "camelCase",
43	rename_all_fields = "camelCase"
44	)]
45	enum Command {
46	Configure {
47	version: u32,
48	root_pid: u32,
49	sample_interval_ms: u64,
50	#[serde(default)]
51	external_processes: Vec<ExternalProcess>,
52	},
53	SetExternalProcesses {
54	version: u32,
55	processes: Vec<ExternalProcess>,
56	},
57	SetSampleInterval {
58	version: u32,
59	sample_interval_ms: u64,
60	},
61	SetStreaming {
62	version: u32,
63	enabled: bool,
64	},
65	SampleNow {
66	version: u32,
67	request_id: String,
68	},
69	ProcessTable {
70	version: u32,
71	request_id: String,
72	},
73	ReadHistory {
74	version: u32,
75	request_id: String,
76	window_ms: u64,
77	},
78	Shutdown {
79	version: u32,
80	},
81	}
82
83	impl Command {
84	fn version(&self) -> u32 {
85	match self {
86	Self::Configure { version, .. }
87	| Self::SetExternalProcesses { version, .. }
88	| Self::SetSampleInterval { version, .. }
89	| Self::SetStreaming { version, .. }
90	| Self::SampleNow { version, .. }
91	| Self::ProcessTable { version, .. }
92	| Self::ReadHistory { version, .. }
93	| Self::Shutdown { version } => *version,
94	}
95	}
96	}
97
98	enum Input {
99	Command(Command),
100	Invalid(String),
101	}
102
103	#[derive(Debug, Serialize)]
104	#[serde(rename_all = "camelCase")]
105	struct Capabilities {
106	cumulative_cpu_time: bool,
107	current_cpu_percent: bool,
108	resident_memory: bool,
109	virtual_memory: bool,
110	io_bytes: bool,
111	process_start_time: bool,
112	process_tree: bool,
113	}
114
115	#[derive(Debug, Serialize)]
116	#[serde(rename_all = "camelCase")]
117	struct HelloEvent {
118	version: u32,
119	#[serde(rename = "type")]
120	event_type: &'static str,
121	sidecar_version: &'static str,
122	sidecar_pid: u32,
123	platform: &'static str,
124	arch: &'static str,
125	capabilities: Capabilities,
126	}
127
128	#[derive(Debug, Clone, Copy, Serialize)]
129	#[serde(rename_all = "kebab-case")]
130	enum IoSemantics {
131	Storage,
132	AllIo,
133	}
134
135	#[derive(Debug, Clone, Serialize)]
136	#[serde(rename_all = "camelCase")]
137	struct ProcessSample {
138	pid: u32,
139	ppid: u32,
140	start_time_ms: u64,
141	run_time_ms: u64,
142	name: String,
143	command: String,
144	status: String,
145	cpu_percent: f32,
146	cpu_time_ms: u64,
147	resident_bytes: u64,
148	virtual_bytes: u64,
149	io_read_bytes: u64,
150	io_write_bytes: u64,
151	io_semantics: IoSemantics,
152	}
153
154	#[derive(Debug, Serialize)]
155	#[serde(rename_all = "camelCase")]
156	struct ProcessTableEntry {
157	pid: u32,
158	ppid: u32,
159	name: String,
160	}
161
162	#[derive(Debug, Serialize)]
163	#[serde(rename_all = "camelCase")]
164	struct ProcessTableEvent<'a> {
165	version: u32,
166	#[serde(rename = "type")]
167	event_type: &'static str,
168	request_id: &'a str,
169	processes: Vec<ProcessTableEntry>,
170	}
171
172	impl ProcessSample {
173	fn estimated_history_bytes(&self) -> usize {
174	std::mem::size_of::<Self>()
175	.saturating_add(self.name.len())
176	.saturating_add(self.command.len())
177	.saturating_add(self.status.len())
178	}
179	}
180
181	#[derive(Debug, Clone, Serialize)]
182	#[serde(rename_all = "camelCase")]
183	struct SnapshotEvent {
184	version: u32,
185	#[serde(rename = "type")]
186	event_type: &'static str,
187	sequence: u64,
188	sampled_at_unix_ms: u64,
189	collection_duration_micros: u64,
190	scanned_process_count: usize,
191	retained_process_count: usize,
192	inaccessible_process_count: usize,
193	#[serde(skip_serializing_if = "Option::is_none")]
194	request_id: Option<String>,
195	external_processes: Vec<ExternalProcess>,
196	processes: Vec<ProcessSample>,
197	}
198
199	impl SnapshotEvent {
200	fn retained_entry_count(&self) -> usize {
201	self.processes
202	.len()
203	.saturating_add(self.external_processes.len())
204	}
205
206	fn estimated_history_bytes(&self) -> usize {
207	std::mem::size_of::<Self>()
208	.saturating_add(
209	self.processes
210	.iter()
211	.map(ProcessSample::estimated_history_bytes)
212	.sum::<usize>(),
213	)
214	.saturating_add(
215	self.external_processes
216	.iter()
217	.map(ExternalProcess::estimated_history_bytes)
218	.sum::<usize>(),
219	)
220	}
221	}
222
223	#[derive(Debug, Serialize)]
224	#[serde(rename_all = "camelCase")]
225	struct HistoryChunkEvent<'a> {
226	version: u32,
227	#[serde(rename = "type")]
228	event_type: &'static str,
229	request_id: &'a str,
230	done: bool,
231	snapshots: &'a [SnapshotEvent],
232	}
233
234	#[derive(Debug, Serialize)]
235	#[serde(rename_all = "camelCase")]
236	struct ErrorEvent {
237	version: u32,
238	#[serde(rename = "type")]
239	event_type: &'static str,
240	code: &'static str,
241	message: String,
242	recoverable: bool,
243	}
244
245	#[derive(Debug, Clone)]
246	struct CollectorConfig {
247	root_pid: u32,
248	sample_interval: Option<Duration>,
249	external_processes: HashMap<u32, Option<u64>>,
250	}
251
252	#[derive(Default)]
253	struct HistoryRecorder {
254	snapshots: VecDeque<SnapshotEvent>,
255	retained_entry_count: usize,
256	retained_bytes: usize,
257	}
258
259	impl HistoryRecorder {
260	fn record(&mut self, snapshot: &SnapshotEvent) {
261	self.record_with_limits(
262	snapshot,
263	MAX_HISTORY_SNAPSHOTS,
264	MAX_HISTORY_RETAINED_ENTRIES,
265	MAX_HISTORY_RETAINED_BYTES,
266	);
267	}
268
269	fn record_with_limits(
270	&mut self,
271	snapshot: &SnapshotEvent,
272	max_snapshots: usize,
273	max_retained_entries: usize,
274	max_retained_bytes: usize,
275	) {
276	let clock_moved_backward = self
277	.snapshots
278	.back()
279	.is_some_and(|previous| previous.sampled_at_unix_ms > snapshot.sampled_at_unix_ms);
280	let mut retained = snapshot.clone();
281	retained.request_id = None;
282	self.retained_entry_count = self
283	.retained_entry_count
284	.saturating_add(retained.retained_entry_count());
285	self.retained_bytes = self
286	.retained_bytes
287	.saturating_add(retained.estimated_history_bytes());
288	self.snapshots.push_back(retained);
289	self.trim_to_limits(
290	snapshot.sampled_at_unix_ms,
291	max_snapshots,
292	max_retained_entries,
293	max_retained_bytes,
294	clock_moved_backward,
295	);
296	}
297
298	fn trim_to_limits(
299	&mut self,
300	now_ms: u64,
301	max_snapshots: usize,
302	max_retained_entries: usize,
303	max_retained_bytes: usize,
304	clock_moved_backward: bool,
305	) {
306	if clock_moved_backward {
307	let mut future_entry_count = 0usize;
308	let mut future_bytes = 0usize;
309	self.snapshots.retain(|snapshot| {
310	let keep = snapshot.sampled_at_unix_ms <= now_ms;
311	if !keep {
312	future_entry_count =
313	future_entry_count.saturating_add(snapshot.retained_entry_count());
314	future_bytes = future_bytes.saturating_add(snapshot.estimated_history_bytes());
315	}
316	keep
317	});
318	self.retained_entry_count =
319	self.retained_entry_count.saturating_sub(future_entry_count);
320	self.retained_bytes = self.retained_bytes.saturating_sub(future_bytes);
321	}
322
323	while self.snapshots.front().is_some_and(|snapshot| {
324	snapshot.sampled_at_unix_ms < now_ms.saturating_sub(HISTORY_RETENTION_MS)
325	|| self.snapshots.len() > max_snapshots
326	|| self.retained_entry_count > max_retained_entries
327	|| self.retained_bytes > max_retained_bytes
328	}) {
329	if let Some(removed) = self.snapshots.pop_front() {
330	self.retained_entry_count = self
331	.retained_entry_count
332	.saturating_sub(removed.retained_entry_count());
333	self.retained_bytes = self
334	.retained_bytes
335	.saturating_sub(removed.estimated_history_bytes());
336	}
337	}
338	}
339
340	fn read(&self, window_ms: u64, now_ms: u64) -> Vec<SnapshotEvent> {
341	let started_at_ms = now_ms.saturating_sub(window_ms.min(HISTORY_RETENTION_MS));
342	self.snapshots
343	.iter()
344	.filter(|snapshot| {
345	snapshot.sampled_at_unix_ms >= started_at_ms
346	&& snapshot.sampled_at_unix_ms <= now_ms
347	})
348	.cloned()
349	.collect()
350	}
351	}
352
353	struct Collector {
354	system: System,
355	sequence: u64,
356	cpu_baseline_refreshed_at: Option<Instant>,
357	}
358
359	impl Collector {
360	fn new() -> Self {
361	Self {
362	system: System::new(),
363	sequence: 0,
364	cpu_baseline_refreshed_at: None,
365	}
366	}
367
368	fn prime_cpu_usage(&mut self) {
369	self.system.refresh_processes_specifics(
370	ProcessesToUpdate::All,
371	true,
372	process_discovery_refresh_kind(),
373	);
374	self.cpu_baseline_refreshed_at = Some(Instant::now());
375	}
376
377	fn process_table(&self) -> Vec<ProcessTableEntry> {
378	// Use a dedicated System so this refresh cannot reset the CPU
379	// baseline tracked by self.system for snapshots.
380	let mut process_table_system = System::new();
381	process_table_system.refresh_processes_specifics(
382	ProcessesToUpdate::All,
383	true,
384	ProcessRefreshKind::nothing().without_tasks(),
385	);
386	let mut processes = process_table_system
387	.processes()
388	.iter()
389	.filter_map(|(pid, process)| {
390	let pid = pid.as_u32();
391	// Pid 0 is the kernel idle process on some platforms. The
392	// processTable contract requires positive pids, and one zero
393	// would fail the whole event decode on the server, so drop it
394	// here. It can never be a terminal descendant.
395	if pid == 0 {
396	return None;
397	}
398	Some(ProcessTableEntry {
399	pid,
400	ppid: process.parent().map(Pid::as_u32).unwrap_or(0),
401	name: truncate_utf8(
402	process.name().to_string_lossy().into_owned(),
403	MAX_PROCESS_NAME_BYTES,
404	),
405	})
406	})
407	.collect::<Vec<_>>();
408	processes.sort_by_key(|process| process.pid);
409	processes
410	}
411
412	fn sample(&mut self, config: &CollectorConfig, request_id: Option<String>) -> SnapshotEvent {
413	if let Some(delay) =
414	remaining_cpu_measurement_delay(self.cpu_baseline_refreshed_at.take(), Instant::now())
415	{
416	thread::sleep(delay);
417	}
418	let collection_started = Instant::now();
419	self.system.refresh_processes_specifics(
420	ProcessesToUpdate::All,
421	true,
422	process_discovery_refresh_kind(),
423	);
424	self.cpu_baseline_refreshed_at = Some(Instant::now());
425
426	let rows = self
427	.system
428	.processes()
429	.iter()
430	.map(|(pid, process)| {
431	let pid = pid.as_u32();
432	let ppid = process.parent().map(Pid::as_u32).unwrap_or(0);
433	(pid, ppid, process.start_time().saturating_mul(1_000))
434	})
435	.collect::<Vec<_>>();
436	let external_processes = config
437	.external_processes
438	.iter()
439	.filter_map(|(pid, expected_start_time_ms)| {
440	let (_, _, actual_start_time_ms) = rows
441	.iter()
442	.find(|(candidate_pid, _, _)| candidate_pid == pid)?;
443	matches_external_identity(*actual_start_time_ms, *expected_start_time_ms).then_some(
444	ExternalProcess {
445	pid: *pid,
446	start_time_ms: Some(*actual_start_time_ms),
447	},
448	)
449	})
450	.collect::<Vec<_>>();
451	let mut roots = external_processes
452	.iter()
453	.map(|process| process.pid)
454	.collect::<HashSet<_>>();
455	roots.insert(config.root_pid);
456	let tracked = select_tracked_pids(&rows, &roots);
457	let tracked_process_count = tracked.len();
458	let process_details = if cfg!(target_os = "linux") && !tracked.is_empty() {
459	let monitor_pid = Pid::from_u32(std::process::id());
460	let mut detail_pids = tracked
461	.iter()
462	.copied()
463	.map(Pid::from_u32)
464	.collect::<Vec<_>>();
465	if !tracked.contains(&monitor_pid.as_u32()) {
466	detail_pids.push(monitor_pid);
467	}
468	// Detail fields need no baseline. Drop command data and OS handles after each sample.
469	let mut details = System::new();
470	details.refresh_processes_specifics(
471	ProcessesToUpdate::Some(&detail_pids),
472	true,
473	process_refresh_kind().without_cpu(),
474	);
475	// This process cannot be replaced during collection. Its start time
476	// exposes any boot-epoch shift between the two System instances.
477	let start_time_offset = self.system.process(monitor_pid).and_then(|process| {
478	details.process(monitor_pid).map(|detail| {
479	i128::from(detail.start_time()) - i128::from(process.start_time())
480	})
481	});
482	Some((details, start_time_offset))
483	} else {
484	None
485	};
486	let sample_details = process_details
487	.as_ref()
488	.map_or(&self.system, |(details, _)| details);
489	let start_time_offset = process_details
490	.as_ref()
491	.map_or(Some(0), |(_, offset)| *offset);
492	let mut processes = tracked
493	.into_iter()
494	.filter_map(|pid| {
495	let process = self.system.process(Pid::from_u32(pid))?;
496	let details = sample_details.process(Pid::from_u32(pid))?;
497	if !matches_process_start_time(
498	process.start_time(),
499	details.start_time(),
500	start_time_offset?,
501	) {
502	return None;
503	}
504	let disk_usage = details.disk_usage();
505	let command = if details.cmd().is_empty() {
506	process.name().to_string_lossy().into_owned()
507	} else {
508	details
509	.cmd()
510	.iter()
511	.map(|part| part.to_string_lossy())
512	.collect::<Vec<_>>()
513	.join(" ")
514	};
515
516	Some(ProcessSample {
517	pid,
518	ppid: process.parent().map(Pid::as_u32).unwrap_or(0),
519	start_time_ms: process.start_time().saturating_mul(1_000),
520	run_time_ms: process.run_time().saturating_mul(1_000),
521	name: truncate_utf8(
522	process.name().to_string_lossy().into_owned(),
523	MAX_PROCESS_NAME_BYTES,
524	),
525	command: truncate_utf8(command, MAX_PROCESS_COMMAND_BYTES),
526	status: truncate_utf8(
527	format!("{:?}", process.status()),
528	MAX_PROCESS_STATUS_BYTES,
529	),
530	cpu_percent: process.cpu_usage(),
531	cpu_time_ms: process.accumulated_cpu_time(),
532	resident_bytes: details.memory(),
533	virtual_bytes: details.virtual_memory(),
534	io_read_bytes: disk_usage.total_read_bytes,
535	io_write_bytes: disk_usage.total_written_bytes,
536	io_semantics: io_semantics(),
537	})
538	})
539	.collect::<Vec<_>>();
540	drop(process_details);
541	processes.sort_by_key(|process| process.pid);
542	self.sequence = self.sequence.saturating_add(1);
543
544	SnapshotEvent {
545	version: PROTOCOL_VERSION,
546	event_type: "snapshot",
547	sequence: self.sequence,
548	sampled_at_unix_ms: unix_time_ms(),
549	collection_duration_micros: collection_started.elapsed().as_micros() as u64,
550	scanned_process_count: self.system.processes().len(),
551	retained_process_count: processes.len(),
552	inaccessible_process_count: inaccessible_process_count(
553	tracked_process_count,
554	processes.len(),
555	),
556	request_id,
557	external_processes,
558	processes,
559	}
560	}
561	}
562
563	// Keep CPU baselines separate. Even a metadata refresh resets Linux process times.
564	fn process_discovery_refresh_kind() -> ProcessRefreshKind {
565	if cfg!(target_os = "linux") {
566	ProcessRefreshKind::nothing().with_cpu().without_tasks()
567	} else {
568	process_refresh_kind()
569	}
570	}
571
572	fn process_refresh_kind() -> ProcessRefreshKind {
573	ProcessRefreshKind::nothing()
574	.with_memory()
575	.with_cpu()
576	.with_disk_usage()
577	.with_cmd(UpdateKind::Always)
578	.without_tasks()
579	}
580
581	fn inaccessible_process_count(selected: usize, materialized: usize) -> usize {
582	selected.saturating_sub(materialized)
583	}
584
585	fn matches_process_start_time(discovered: u64, detail: u64, epoch_offset: i128) -> bool {
586	i128::from(detail) - epoch_offset == i128::from(discovered)
587	}
588
589	fn remaining_cpu_measurement_delay(
590	baseline_refreshed_at: Option<Instant>,
591	now: Instant,
592	) -> Option<Duration> {
593	baseline_refreshed_at
594	.and_then(|baseline| MINIMUM_CPU_UPDATE_INTERVAL.checked_sub(now.duration_since(baseline)))
595	.filter(|delay| !delay.is_zero())
596	}
597
598	fn matches_external_identity(
599	actual_start_time_ms: u64,
600	expected_start_time_ms: Option<u64>,
601	) -> bool {
602	// sysinfo reports process starts at whole-second precision. Normalize the
603	// higher-resolution Electron timestamp to that same bucket instead of
604	// accepting adjacent seconds, which could attach a quickly reused PID.
605	expected_start_time_ms.is_none_or(|expected| {
606	actual_start_time_ms == expected - (expected % PROCESS_START_TIME_PRECISION_MS)
607	})
608	}
609
610	fn select_tracked_pids(rows: &[(u32, u32, u64)], roots: &HashSet<u32>) -> HashSet<u32> {
611	let mut children_by_parent = HashMap::<u32, Vec<(u32, u64)>>::new();
612	let mut start_time_by_pid = HashMap::<u32, u64>::new();
613	for (pid, ppid, start_time_ms) in rows {
614	children_by_parent
615	.entry(*ppid)
616	.or_default()
617	.push((*pid, *start_time_ms));
618	start_time_by_pid.insert(*pid, *start_time_ms);
619	}
620
621	let mut tracked = HashSet::new();
622	let mut visited_identities = HashSet::new();
623	let mut queue = roots
624	.iter()
625	.filter_map(|pid| {
626	start_time_by_pid
627	.get(pid)
628	.map(|start_time_ms| (*pid, *start_time_ms))
629	})
630	.collect::<VecDeque<_>>();
631
632	while let Some((pid, start_time_ms)) = queue.pop_front() {
633	if !visited_identities.insert((pid, start_time_ms)) {
634	continue;
635	}
636	tracked.insert(pid);
637	if let Some(children) = children_by_parent.get(&pid) {
638	queue.extend(
639	children
640	.iter()
641	.copied()
642	.filter(|(_, child_start_time_ms)| *child_start_time_ms >= start_time_ms),
643	);
644	}
645	}
646
647	tracked
648	}
649
650	fn truncate_utf8(mut value: String, max_bytes: usize) -> String {
651	if value.len() <= max_bytes {
652	return value;
653	}
654	let mut boundary = max_bytes;
655	while !value.is_char_boundary(boundary) {
656	boundary = boundary.saturating_sub(1);
657	}
658	value.truncate(boundary);
659	value
660	}
661
662	fn io_semantics() -> IoSemantics {
663	if cfg!(target_os = "windows") {
664	IoSemantics::AllIo
665	} else {
666	IoSemantics::Storage
667	}
668	}
669
670	fn unix_time_ms() -> u64 {
671	SystemTime::now()
672	.duration_since(UNIX_EPOCH)
673	.unwrap_or_default()
674	.as_millis() as u64
675	}
676
677	fn clamp_sample_interval(sample_interval_ms: u64) -> Option<Duration> {
678	(sample_interval_ms > 0).then(|| {
679	Duration::from_millis(
680	sample_interval_ms.clamp(MIN_SAMPLE_INTERVAL_MS, MAX_SAMPLE_INTERVAL_MS),
681	)
682	})
683	}
684
685	fn spawn_input_reader() -> Receiver<Input> {
686	let (sender, receiver) = mpsc::sync_channel(INPUT_QUEUE_CAPACITY);
687	thread::spawn(move || {
688	let stdin = io::stdin();
689	for line in stdin.lock().lines() {
690	let line = match line {
691	Ok(line) => line,
692	Err(error) => {
693	let _ = sender.send(Input::Invalid(format!(
694	"failed reading command stream: {error}"
695	)));
696	return;
697	}
698	};
699	if line.trim().is_empty() {
700	continue;
701	}
702	match serde_json::from_str::<Command>(&line) {
703	Ok(command) => {
704	if sender.send(Input::Command(command)).is_err() {
705	return;
706	}
707	}
708	Err(error) => {
709	if sender
710	.send(Input::Invalid(format!("invalid command: {error}")))
711	.is_err()
712	{
713	return;
714	}
715	}
716	}
717	}
718	});
719	receiver
720	}
721
722	fn sample_now_deadline(
723	current: Option<Instant>,
724	interval: Option<Duration>,
725	now: Instant,
726	) -> Option<Instant> {
727	current.or_else(|| interval.map(|duration| now + duration))
728	}
729
730	fn write_event<T: Serialize>(writer: &mut impl Write, event: &T) -> io::Result<()> {
731	serde_json::to_writer(&mut *writer, event)?;
732	writer.write_all(b"\n")?;
733	writer.flush()
734	}
735
736	fn write_error(
737	writer: &mut impl Write,
738	code: &'static str,
739	message: impl Into<String>,
740	recoverable: bool,
741	) -> io::Result<()> {
742	write_event(
743	writer,
744	&ErrorEvent {
745	version: PROTOCOL_VERSION,
746	event_type: "error",
747	code,
748	message: message.into(),
749	recoverable,
750	},
751	)
752	}
753
754	fn write_history(
755	writer: &mut impl Write,
756	request_id: &str,
757	snapshots: &[SnapshotEvent],
758	) -> io::Result<()> {
759	if snapshots.is_empty() {
760	return write_event(
761	writer,
762	&HistoryChunkEvent {
763	version: PROTOCOL_VERSION,
764	event_type: "historyChunk",
765	request_id,
766	done: true,
767	snapshots,
768	},
769	);
770	}
771
772	let chunk_count = snapshots.len().div_ceil(HISTORY_CHUNK_SNAPSHOTS);
773	for (index, chunk) in snapshots.chunks(HISTORY_CHUNK_SNAPSHOTS).enumerate() {
774	write_event(
775	writer,
776	&HistoryChunkEvent {
777	version: PROTOCOL_VERSION,
778	event_type: "historyChunk",
779	request_id,
780	done: index + 1 == chunk_count,
781	snapshots: chunk,
782	},
783	)?;
784	}
785	Ok(())
786	}
787
788	fn main() -> io::Result<()> {
789	let mut writer = BufWriter::new(io::stdout().lock());
790	write_event(
791	&mut writer,
792	&HelloEvent {
793	version: PROTOCOL_VERSION,
794	event_type: "hello",
795	sidecar_version: env!("CARGO_PKG_VERSION"),
796	sidecar_pid: std::process::id(),
797	platform: std::env::consts::OS,
798	arch: std::env::consts::ARCH,
799	capabilities: Capabilities {
800	cumulative_cpu_time: true,
801	current_cpu_percent: true,
802	resident_memory: true,
803	virtual_memory: true,
804	io_bytes: true,
805	process_start_time: true,
806	process_tree: true,
807	},
808	},
809	)?;
810
811	let receiver = spawn_input_reader();
812	let mut collector = Collector::new();
813	let mut history = HistoryRecorder::default();
814	let mut config: Option<CollectorConfig> = None;
815	let mut next_sample_at: Option<Instant> = None;
816	let mut streaming_enabled = false;
817
818	loop {
819	if next_sample_at.is_some_and(|deadline| deadline <= Instant::now()) {
820	if let Some(current) = config.as_ref() {
821	if let Some(interval) = current.sample_interval {
822	let event = collector.sample(current, None);
823	history.record(&event);
824	if streaming_enabled {
825	write_event(&mut writer, &event)?;
826	}
827	next_sample_at = Some(Instant::now() + interval);
828	} else {
829	next_sample_at = None;
830	}
831	} else {
832	next_sample_at = None;
833	}
834	continue;
835	}
836
837	let timeout = next_sample_at
838	.map(|deadline| deadline.saturating_duration_since(Instant::now()))
839	.unwrap_or(Duration::from_secs(60));
840
841	match receiver.recv_timeout(timeout) {
842	Ok(Input::Invalid(message)) => {
843	write_error(&mut writer, "invalid-command", message, true)?;
844	}
845	Ok(Input::Command(command)) => {
846	if command.version() != PROTOCOL_VERSION {
847	write_error(
848	&mut writer,
849	"protocol-mismatch",
850	format!(
851	"unsupported protocol version {}; expected {PROTOCOL_VERSION}",
852	command.version()
853	),
854	false,
855	)?;
856	continue;
857	}
858
859	match command {
860	Command::Configure {
861	root_pid,
862	sample_interval_ms,
863	external_processes,
864	..
865	} => {
866	let sample_interval = clamp_sample_interval(sample_interval_ms);
867	config = Some(CollectorConfig {
868	root_pid,
869	sample_interval,
870	external_processes: external_processes
871	.into_iter()
872	.map(|process| (process.pid, process.start_time_ms))
873	.collect(),
874	});
875	collector.prime_cpu_usage();
876	next_sample_at = sample_interval.map(|_| Instant::now());
877	}
878	Command::SetExternalProcesses { processes, .. } => {
879	if let Some(current) = config.as_mut() {
880	current.external_processes = processes
881	.into_iter()
882	.map(|process| (process.pid, process.start_time_ms))
883	.collect();
884	} else {
885	write_error(
886	&mut writer,
887	"not-configured",
888	"configure must be sent before external processes",
889	true,
890	)?;
891	}
892	}
893	Command::SetSampleInterval {
894	sample_interval_ms, ..
895	} => {
896	if let Some(current) = config.as_mut() {
897	current.sample_interval = clamp_sample_interval(sample_interval_ms);
898	next_sample_at = current
899	.sample_interval
900	.map(|interval| Instant::now() + interval);
901	} else {
902	write_error(
903	&mut writer,
904	"not-configured",
905	"configure must be sent before changing the sample interval",
906	true,
907	)?;
908	}
909	}
910	Command::SetStreaming { enabled, .. } => {
911	streaming_enabled = enabled;
912	}
913	Command::SampleNow { request_id, .. } => {
914	if let Some(current) = config.as_ref() {
915	let event = collector.sample(current, Some(request_id));
916	history.record(&event);
917	write_event(&mut writer, &event)?;
918	next_sample_at = sample_now_deadline(
919	next_sample_at,
920	current.sample_interval,
921	Instant::now(),
922	);
923	} else {
924	write_error(
925	&mut writer,
926	"not-configured",
927	"configure must be sent before sampling",
928	true,
929	)?;
930	}
931	}
932	Command::ProcessTable { request_id, .. } => {
933	let event = ProcessTableEvent {
934	version: PROTOCOL_VERSION,
935	event_type: "processTable",
936	request_id: &request_id,
937	processes: collector.process_table(),
938	};
939	write_event(&mut writer, &event)?;
940	}
941	Command::ReadHistory {
942	request_id,
943	window_ms,
944	..
945	} => {
946	if config.is_some() {
947	let snapshots = history.read(window_ms, unix_time_ms());
948	write_history(&mut writer, &request_id, &snapshots)?;
949	} else {
950	write_error(
951	&mut writer,
952	"not-configured",
953	"configure must be sent before reading history",
954	true,
955	)?;
956	}
957	}
958	Command::Shutdown { .. } => return Ok(()),
959	}
960	}
961	Err(RecvTimeoutError::Timeout) => {}
962	Err(RecvTimeoutError::Disconnected) => return Ok(()),
963	}
964	}
965	}
966
967	#[cfg(test)]
968	mod tests {
969	use super::*;
970
971	#[test]
972	fn selects_roots_and_all_descendants() {
973	let rows = vec![
974	(10, 1, 1_000),
975	(11, 10, 1_100),
976	(12, 11, 1_200),
977	(20, 1, 2_000),
978	(21, 20, 2_100),
979	(30, 99, 3_000),
980	];
981	let tracked = select_tracked_pids(&rows, &HashSet::from([10, 20]));
982
983	assert_eq!(tracked, HashSet::from([10, 11, 12, 20, 21]));
984	}
985
986	#[test]
987	fn rejects_descendants_older_than_a_reused_parent_pid() {
988	let rows = vec![
989	(20, 1, 5_000),
990	(21, 20, 4_000),
991	(22, 20, 5_100),
992	(23, 21, 5_200),
993	];
994	let tracked = select_tracked_pids(&rows, &HashSet::from([20]));
995
996	assert_eq!(tracked, HashSet::from([20, 22]));
997	}
998
999	#[test]
1000	fn ignores_missing_roots() {
1001	let rows = vec![(10, 1, 1_000), (11, 10, 1_100)];
1002	let tracked = select_tracked_pids(&rows, &HashSet::from([99]));
1003
1004	assert!(tracked.is_empty());
1005	}
1006
1007	#[test]
1008	fn validates_external_process_start_identity() {
1009	assert!(matches_external_identity(10_000, None));
1010	assert!(matches_external_identity(10_000, Some(10_999)));
1011	assert!(!matches_external_identity(10_000, Some(11_000)));
1012	assert!(!matches_external_identity(10_000, Some(9_999)));
1013	}
1014
1015	#[test]
1016	fn loads_details_when_an_existing_process_becomes_selected() {
1017	let mut collector = Collector::new();
1018	let mut config = CollectorConfig {
1019	root_pid: u32::MAX,
1020	sample_interval: None,
1021	external_processes: HashMap::new(),
1022	};
1023	assert!(collector.sample(&config, None).processes.is_empty());
1024
1025	config.root_pid = std::process::id();
1026	let snapshot = collector.sample(&config, None);
1027	let process = snapshot
1028	.processes
1029	.iter()
1030	.find(|process| process.pid == config.root_pid)
1031	.expect("selected process");
1032
1033	assert!(!process.command.is_empty());
1034	assert!(process.resident_bytes > 0);
1035	assert!(process.cpu_percent.is_finite());
1036	}
1037
1038	#[test]
1039	fn accepts_clock_shifts_without_accepting_reused_process_starts() {
1040	assert!(matches_process_start_time(10_000, 13_600, 3_600));
1041	assert!(!matches_process_start_time(10_000, 13_601, 3_600));
1042	assert!(matches_process_start_time(10_000, 6_400, -3_600));
1043	assert!(!matches_process_start_time(10_000, 6_401, -3_600));
1044	assert!(matches_process_start_time(10_000, 10_000, 0));
1045	assert!(!matches_process_start_time(10_000, 10_001, 0));
1046	}
1047
1048	#[test]
1049	fn decodes_protocol_commands() {
1050	let configure = serde_json::from_str::<Command>(
1051	r#"{"version":3,"type":"configure","rootPid":42,"sampleIntervalMs":1000,"externalProcesses":[{"pid":7}]}"#,
1052	)
1053	.expect("configure command");
1054
1055	match configure {
1056	Command::Configure {
1057	root_pid,
1058	sample_interval_ms,
1059	external_processes,
1060	..
1061	} => {
1062	assert_eq!(root_pid, 42);
1063	assert_eq!(sample_interval_ms, 1_000);
1064	assert_eq!(external_processes[0].pid, 7);
1065	assert_eq!(external_processes[0].start_time_ms, None);
1066	}
1067	_ => panic!("unexpected command"),
1068	}
1069
1070	let read_history = serde_json::from_str::<Command>(
1071	r#"{"version":3,"type":"readHistory","requestId":"history-1","windowMs":60000}"#,
1072	)
1073	.expect("read history command");
1074	assert!(matches!(
1075	read_history,
1076	Command::ReadHistory {
1077	request_id,
1078	window_ms: 60_000,
1079	..
1080	} if request_id == "history-1"
1081	));
1082
1083	let process_table = serde_json::from_str::<Command>(
1084	r#"{"version":3,"type":"processTable","requestId":"processes-1"}"#,
1085	)
1086	.expect("process table command");
1087	assert!(matches!(
1088	process_table,
1089	Command::ProcessTable { request_id, .. } if request_id == "processes-1"
1090	));
1091	}
1092
1093	#[test]
1094	fn clamps_sample_interval() {
1095	assert_eq!(clamp_sample_interval(0), None);
1096	assert_eq!(clamp_sample_interval(1), Some(Duration::from_millis(250)));
1097	assert_eq!(
1098	clamp_sample_interval(100_000),
1099	Some(Duration::from_millis(60_000))
1100	);
1101	}
1102
1103	#[test]
1104	fn counts_selected_processes_that_could_not_be_materialized() {
1105	assert_eq!(inaccessible_process_count(5, 3), 2);
1106	assert_eq!(inaccessible_process_count(3, 5), 0);
1107	}
1108
1109	#[test]
1110	fn waits_for_a_cpu_measurement_window_after_priming() {
1111	let baseline = Instant::now();
1112
1113	assert_eq!(
1114	remaining_cpu_measurement_delay(Some(baseline), baseline),
1115	Some(MINIMUM_CPU_UPDATE_INTERVAL)
1116	);
1117	assert_eq!(
1118	remaining_cpu_measurement_delay(Some(baseline), baseline + MINIMUM_CPU_UPDATE_INTERVAL),
1119	None
1120	);
1121	assert_eq!(remaining_cpu_measurement_delay(None, baseline), None);
1122	}
1123
1124	#[test]
1125	fn retains_bounded_history_without_request_ids() {
1126	let mut history = HistoryRecorder::default();
1127	for sequence in 0..=MAX_HISTORY_SNAPSHOTS {
1128	history.record(&SnapshotEvent {
1129	version: PROTOCOL_VERSION,
1130	event_type: "snapshot",
1131	sequence: sequence as u64,
1132	sampled_at_unix_ms: sequence as u64 * 1_000,
1133	collection_duration_micros: 1,
1134	scanned_process_count: 0,
1135	retained_process_count: 0,
1136	inaccessible_process_count: 0,
1137	request_id: Some("request".to_owned()),
1138	external_processes: vec![ExternalProcess {
1139	pid: 7,
1140	start_time_ms: Some(1_000),
1141	}],
1142	processes: Vec::new(),
1143	});
1144	}
1145
1146	assert_eq!(history.snapshots.len(), MAX_HISTORY_SNAPSHOTS);
1147	assert!(
1148	history
1149	.snapshots
1150	.iter()
1151	.all(|snapshot| snapshot.request_id.is_none())
1152	);
1153	assert!(history.snapshots.iter().all(|snapshot| {
1154	snapshot.external_processes.len() == 1
1155	&& snapshot.external_processes[0].pid == 7
1156	&& snapshot.external_processes[0].start_time_ms == Some(1_000)
1157	}));
1158	assert_eq!(
1159	history
1160	.read(10_000, MAX_HISTORY_SNAPSHOTS as u64 * 1_000)
1161	.len(),
1162	11
1163	);
1164	}
1165
1166	#[test]
1167	fn excludes_and_trims_future_history_after_the_clock_moves_backward() {
1168	let mut history = HistoryRecorder::default();
1169	let snapshot = SnapshotEvent {
1170	version: PROTOCOL_VERSION,
1171	event_type: "snapshot",
1172	sequence: 1,
1173	sampled_at_unix_ms: 2_000,
1174	collection_duration_micros: 1,
1175	scanned_process_count: 0,
1176	retained_process_count: 0,
1177	inaccessible_process_count: 0,
1178	request_id: None,
1179	external_processes: Vec::new(),
1180	processes: Vec::new(),
1181	};
1182	history.record(&snapshot);
1183
1184	assert!(history.read(0, 1_000).is_empty());
1185
1186	history.record(&SnapshotEvent {
1187	sequence: 2,
1188	sampled_at_unix_ms: 1_000,
1189	..snapshot
1190	});
1191	assert_eq!(history.snapshots.len(), 1);
1192	assert_eq!(
1193	history.snapshots.front().map(|entry| entry.sequence),
1194	Some(2)
1195	);
1196	}
1197
1198	#[test]
1199	fn bounds_history_by_estimated_process_bytes() {
1200	let mut history = HistoryRecorder::default();
1201	let command = "x".repeat(128);
1202	let process = ProcessSample {
1203	pid: 1,
1204	ppid: 0,
1205	start_time_ms: 0,
1206	run_time_ms: 0,
1207	name: "process".to_owned(),
1208	command,
1209	status: "Run".to_owned(),
1210	cpu_percent: 0.0,
1211	cpu_time_ms: 0,
1212	resident_bytes: 0,
1213	virtual_bytes: 0,
1214	io_read_bytes: 0,
1215	io_write_bytes: 0,
1216	io_semantics: IoSemantics::Storage,
1217	};
1218	let snapshot_bytes =
1219	std::mem::size_of::<SnapshotEvent>() + process.estimated_history_bytes();
1220	for sequence in 0..3 {
1221	history.record_with_limits(
1222	&SnapshotEvent {
1223	version: PROTOCOL_VERSION,
1224	event_type: "snapshot",
1225	sequence,
1226	sampled_at_unix_ms: sequence * 1_000,
1227	collection_duration_micros: 1,
1228	scanned_process_count: 1,
1229	retained_process_count: 1,
1230	inaccessible_process_count: 0,
1231	request_id: None,
1232	external_processes: Vec::new(),
1233	processes: vec![ProcessSample {
1234	pid: sequence as u32 + 1,
1235	start_time_ms: sequence * 1_000,
1236	..process.clone()
1237	}],
1238	},
1239	3,
1240	3,
1241	snapshot_bytes * 2,
1242	);
1243	}
1244
1245	assert!(history.retained_bytes <= snapshot_bytes * 2);
1246	assert_eq!(history.snapshots.len(), 2);
1247	assert_eq!(
1248	history.snapshots.front().map(|snapshot| snapshot.sequence),
1249	Some(1)
1250	);
1251	}
1252
1253	#[test]
1254	fn counts_external_processes_toward_history_limits() {
1255	let mut history = HistoryRecorder::default();
1256	let external_processes = (1..=128)
1257	.map(|pid| ExternalProcess {
1258	pid,
1259	start_time_ms: Some(u64::from(pid) * 1_000),
1260	})
1261	.collect::<Vec<_>>();
1262	let snapshot = SnapshotEvent {
1263	version: PROTOCOL_VERSION,
1264	event_type: "snapshot",
1265	sequence: 0,
1266	sampled_at_unix_ms: 0,
1267	collection_duration_micros: 1,
1268	scanned_process_count: 0,
1269	retained_process_count: 0,
1270	inaccessible_process_count: 0,
1271	request_id: None,
1272	external_processes,
1273	processes: Vec::new(),
1274	};
1275	let snapshot_bytes = snapshot.estimated_history_bytes();
1276	let snapshot_entries = snapshot.retained_entry_count();
1277
1278	for sequence in 0..3 {
1279	history.record_with_limits(
1280	&SnapshotEvent {
1281	sequence,
1282	sampled_at_unix_ms: sequence * 1_000,
1283	..snapshot.clone()
1284	},
1285	3,
1286	snapshot_entries * 2,
1287	snapshot_bytes * 2,
1288	);
1289	}
1290
1291	assert_eq!(history.retained_entry_count, snapshot_entries * 2);
1292	assert!(history.retained_bytes <= snapshot_bytes * 2);
1293	assert_eq!(history.snapshots.len(), 2);
1294	assert_eq!(
1295	history.snapshots.front().map(|snapshot| snapshot.sequence),
1296	Some(1)
1297	);
1298	}
1299
1300	#[test]
1301	fn truncates_process_strings_at_utf8_boundaries() {
1302	let value = "é".repeat(MAX_PROCESS_NAME_BYTES);
1303	let truncated = truncate_utf8(value, MAX_PROCESS_NAME_BYTES - 1);
1304
1305	assert!(truncated.len() < MAX_PROCESS_NAME_BYTES);
1306	assert!(truncated.is_char_boundary(truncated.len()));
1307	}
1308
1309	#[test]
1310	fn refreshes_commands_without_enumerating_linux_tasks() {
1311	let refresh_kind = process_refresh_kind();
1312
1313	assert_eq!(refresh_kind.cmd(), UpdateKind::Always);
1314	assert!(!refresh_kind.tasks());
1315	assert!(refresh_kind.cpu());
1316	assert!(refresh_kind.memory());
1317	assert!(refresh_kind.disk_usage());
1318	}
1319
1320	#[test]
1321	fn sample_now_does_not_postpone_an_existing_periodic_deadline() {
1322	let now = Instant::now();
1323	let deadline = now + Duration::from_secs(1);
1324
1325	assert_eq!(
1326	sample_now_deadline(
1327	Some(deadline),
1328	Some(Duration::from_secs(5)),
1329	now + Duration::from_millis(100)
1330	),
1331	Some(deadline)
1332	);
1333	}
1334	}