Refactor routing phases and bound subprocess output

src/process.rs

@@ -2,7 +2,7 @@
 
 //! Bounded subprocess execution and process-group cleanup.
 
-use std::io::Read;
+use std::io::{self, Read};
 use std::os::unix::process::CommandExt;
 use std::process::{Child, Command, ExitStatus, Stdio};
 use std::sync::atomic::{AtomicBool, Ordering};
@@ -11,6 +11,30 @@ use std::time::{Duration, Instant};
 
 pub(crate) const COMMAND_TIMEOUT: Duration = Duration::from_secs(15);
 const CHILD_POLL_INTERVAL: Duration = Duration::from_millis(20);
+const MAX_CAPTURED_OUTPUT: usize = 1024 * 1024;
+
+fn read_bounded(mut reader: impl Read, limit: usize) -> io::Result<Vec<u8>> {
+	let mut captured = Vec::with_capacity(limit.min(8192));
+	let mut buffer = [0; 8192];
+	let mut exceeded_limit = false;
+	loop {
+		let count = reader.read(&mut buffer)?;
+		if count == 0 {
+			break;
+		}
+		let remaining = limit.saturating_sub(captured.len());
+		captured.extend_from_slice(&buffer[..count.min(remaining)]);
+		exceeded_limit |= count > remaining;
+	}
+	if exceeded_limit {
+		Err(io::Error::new(
+			io::ErrorKind::InvalidData,
+			format!("captured output exceeded the {limit}-byte limit"),
+		))
+	} else {
+		Ok(captured)
+	}
+}
 
 /// Runs a command with captured output and a deadline.
 pub(crate) fn command(args: &[&str], timeout: Duration) -> Result<String, String> {
@@ -110,26 +134,20 @@ pub(crate) fn command_cancellable(
 		let _ = child.wait();
 		return Err(format!("{name}: failed to capture command output"));
 	};
-	let output = thread::spawn(move || {
-		let mut bytes = Vec::new();
-		stdout.read_to_end(&mut bytes).map(|_| bytes)
-	});
-	let error = thread::spawn(move || {
-		let mut bytes = Vec::new();
-		stderr.read_to_end(&mut bytes).map(|_| bytes)
-	});
+	let output = thread::spawn(move || read_bounded(&mut stdout, MAX_CAPTURED_OUTPUT));
+	let error = thread::spawn(move || read_bounded(&mut stderr, MAX_CAPTURED_OUTPUT));
 	let status = wait_child(&mut child, name, timeout, stopped, KillMode::ProcessGroup)
 		.inspect_err(|error| {
 			log::debug!("{name} stopped after {:?}: {error}", started.elapsed())
 		})?;
 	let stdout = output
 		.join()
 		.map_err(|_| "stdout reader panicked".to_string())?
-		.map_err(|e| e.to_string())?;
+		.map_err(|e| format!("stdout: {e}"))?;
 	let stderr = error
 		.join()
 		.map_err(|_| "stderr reader panicked".to_string())?
-		.map_err(|e| e.to_string())?;
+		.map_err(|e| format!("stderr: {e}"))?;
 	log::debug!(
 		"{name} exited with {status} after {:?} ({} stdout bytes, {} stderr bytes)",
 		started.elapsed(),
@@ -180,6 +198,29 @@ mod tests {
 		);
 	}
 
+	#[test]
+	fn output_readers_drain_beyond_the_capture_limit() {
+		let bytes = vec![0; 1025];
+		let mut reader = std::io::Cursor::new(bytes);
+		assert!(read_bounded(&mut reader, 1024).is_err());
+		assert_eq!(reader.position(), 1025);
+	}
+
+	#[test]
+	fn commands_reject_stdout_and_stderr_over_the_capture_limit() {
+		let size = (MAX_CAPTURED_OUTPUT + 1).to_string();
+		let stdout_error =
+			command(&["head", "-c", &size, "/dev/zero"], Duration::from_secs(5)).unwrap_err();
+		assert!(stdout_error.contains("stdout: captured output exceeded"));
+
+		let stderr_error = command(
+			&["sh", "-c", &format!("head -c {size} /dev/zero >&2")],
+			Duration::from_secs(5),
+		)
+		.unwrap_err();
+		assert!(stderr_error.contains("stderr: captured output exceeded"));
+	}
+
 	#[test]
 	fn commands_reject_missing_program_names() {
 		assert!(command(&[], Duration::from_secs(1)).is_err());

src/session.rs

@@ -346,20 +346,9 @@ impl Session {
 	}
 
 	fn update_routes(&mut self) -> Result<(), String> {
-		let config_error = match groups::load(&self.config.groups_path) {
-			Ok(groups) => {
-				self.transmit.set_groups(&groups);
-				self.config.groups = groups;
-				None
-			}
-			Err(error) => Some(error),
-		};
+		let config_error = self.reload_groups();
 		self.refresh_topology()?;
-		let sources = if self.input.is_some() {
-			self.transmit.sources().clone()
-		} else {
-			self.router.allowed.clone()
-		};
+		let sources = self.routing_sources();
 		let prepared = self.router.prepare_transport(&sources);
 		guard_half_update(self.config.mode, &prepared, || self.router.close())?;
 		let transport_ready = prepared?;
@@ -371,17 +360,7 @@ impl Session {
 			self.last_update = None;
 			return Ok(());
 		}
-		let update = if !transport_ready {
-			self.router.inspect().map(|(headsets, _)| headsets)
-		} else if self.config.groups.is_empty() && self.input.is_none() {
-			self.router.update(true)
-		} else if self.config.mode == Mode::HalfDuplex && !self.config.groups.is_empty() {
-			self.router
-				.update_group_sources_in_groups(self.transmit.group_sources(), &self.config.groups)
-		} else {
-			self.router
-				.update_sources_in_groups(&sources, &self.config.groups)
-		};
+		let update = self.update_route_set(&sources, transport_ready);
 		guard_half_update(self.config.mode, &update, || self.router.close())?;
 		let (changed, inspect_error) = self.inspect_and_confirm()?;
 		if changed {
@@ -393,6 +372,43 @@ impl Session {
 		Ok(())
 	}
 
+	fn reload_groups(&mut self) -> Option<String> {
+		match groups::load(&self.config.groups_path) {
+			Ok(groups) => {
+				self.transmit.set_groups(&groups);
+				self.config.groups = groups;
+				None
+			}
+			Err(error) => Some(error),
+		}
+	}
+
+	fn routing_sources(&self) -> BTreeSet<String> {
+		if self.input.is_some() {
+			self.transmit.sources().clone()
+		} else {
+			self.router.allowed.clone()
+		}
+	}
+
+	fn update_route_set(
+		&mut self,
+		sources: &BTreeSet<String>,
+		transport_ready: bool,
+	) -> Result<BTreeMap<String, Headset>, String> {
+		if !transport_ready {
+			self.router.inspect().map(|(headsets, _)| headsets)
+		} else if self.config.groups.is_empty() && self.input.is_none() {
+			self.router.update(true)
+		} else if self.config.mode == Mode::HalfDuplex && !self.config.groups.is_empty() {
+			self.router
+				.update_group_sources_in_groups(self.transmit.group_sources(), &self.config.groups)
+		} else {
+			self.router
+				.update_sources_in_groups(sources, &self.config.groups)
+		}
+	}
+
 	fn inspect_and_confirm(&mut self) -> Result<(bool, Option<String>), String> {
 		let (current, active) = match self.router.inspect_owned() {
 			Ok(snapshot) => snapshot,

src/transport.rs

@@ -262,6 +262,19 @@ impl Coordinator {
 			return Ok(false);
 		}
 		let snapshot = self.snapshot(io.allowed, io.execute)?;
+		self.prepare_snapshot(sources, io, snapshot)
+	}
+
+	fn prepare_snapshot<F, C>(
+		&mut self,
+		sources: &BTreeSet<String>,
+		io: &mut TransitionIo<'_, F, C>,
+		snapshot: Snapshot,
+	) -> Result<bool, String>
+	where
+		F: FnMut(&[&str]) -> Result<String, String>,
+		C: FnMut(),
+	{
 		let devices = snapshot.transport_devices(io.allowed)?;
 		self.profile_changes
 			.retain(|address, (expected, index, _)| {
@@ -294,8 +307,27 @@ impl Coordinator {
 		if !self.profile_changes.is_empty() {
 			return Ok(false);
 		}
+
+		if self.demote_receivers(sources, io, &snapshot, &devices)? {
+			return Ok(false);
+		}
+		self.promote_sources(sources, io, &devices)?;
+		Ok(self.sources_ready(sources, &headsets, &devices))
+	}
+
+	fn demote_receivers<F, C>(
+		&mut self,
+		sources: &BTreeSet<String>,
+		io: &mut TransitionIo<'_, F, C>,
+		snapshot: &Snapshot,
+		devices: &BTreeMap<String, Device>,
+	) -> Result<bool, String>
+	where
+		F: FnMut(&[&str]) -> Result<String, String>,
+		C: FnMut(),
+	{
 		let mut demoting = false;
-		for (address, device) in &devices {
+		for (address, device) in devices {
 			if !sources.contains(address) {
 				if !device.is_profile("a2dp-sink") {
 					self.change_profile(address, device, "a2dp-sink", sources, io)?;
@@ -305,27 +337,43 @@ impl Coordinator {
 				}
 			}
 		}
-		if demoting {
-			return Ok(false);
-		}
-		let mut ready = true;
+		Ok(demoting)
+	}
+
+	fn promote_sources<F, C>(
+		&mut self,
+		sources: &BTreeSet<String>,
+		io: &mut TransitionIo<'_, F, C>,
+		devices: &BTreeMap<String, Device>,
+	) -> Result<(), String>
+	where
+		F: FnMut(&[&str]) -> Result<String, String>,
+		C: FnMut(),
+	{
 		for address in sources {
 			let device = &devices[address];
 			if !device.is_profile(HEADSET_PROFILE) {
 				self.change_profile(address, device, HEADSET_PROFILE, sources, io)?;
-				ready = false;
 			}
 		}
-		Ok(ready
-			&& devices.iter().all(|(address, device)| {
-				headsets.get(address).is_some_and(|headset| {
-					if sources.contains(address) {
-						device.is_profile(HEADSET_PROFILE) && headset.has_duplex_audio()
-					} else {
-						device.is_profile("a2dp-sink") && !headset.sinks.is_empty()
-					}
-				})
-			}))
+		Ok(())
+	}
+
+	fn sources_ready(
+		&self,
+		sources: &BTreeSet<String>,
+		headsets: &BTreeMap<String, crate::pipewire::Headset>,
+		devices: &BTreeMap<String, Device>,
+	) -> bool {
+		devices.iter().all(|(address, device)| {
+			headsets.get(address).is_some_and(|headset| {
+				if sources.contains(address) {
+					device.is_profile(HEADSET_PROFILE) && headset.has_duplex_audio()
+				} else {
+					device.is_profile("a2dp-sink") && !headset.sinks.is_empty()
+				}
+			})
+		})
 	}
 
 	fn change_profile<F, C>(