git.delta.rocks / remowt / refs/commits / 6d9cf16dada2

difftreelog

feat drain stdout/stderr after signal

vmkslqpuYaroslav Bolyukin4 days agoparent: #eadb0bb.patch.diff
in: trunk

6 files changed

modifiedCargo.lockdiffbeforeafterboth
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -308,9 +308,9 @@
 
 [[package]]
 name = "bifrostlink"
-version = "0.2.4"
+version = "0.2.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ad2d0e30a2aa432b78f41f9676572f88201d4dc73bc2b7bc90704d2e02b7d062"
+checksum = "910f9286588d13e3dbdbbc1ad4d292656e704bc93e1f41b8a13b48e3a8e95f39"
 dependencies = [
  "async-trait",
  "async_fn_traits",
@@ -327,9 +327,9 @@
 
 [[package]]
 name = "bifrostlink-macros"
-version = "0.2.4"
+version = "0.2.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e2121559c45cbe89c4f8d1d741360d5b028577254f6beca053dc02332da85b43"
+checksum = "a0ea5c423c3831c523c8ef78debdf6a64e72b21ec92148a44163a4c25c05dfd0"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -338,9 +338,9 @@
 
 [[package]]
 name = "bifrostlink-ports"
-version = "0.2.4"
+version = "0.2.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e9395c4ccca497b0c50583e6de57aca921c046ae0c10f56030cd2c5a20db05f8"
+checksum = "9e3a9a01ec1b8bd7d44b47cd0183a1465880e241027d9f5afcb076e11704ec70"
 dependencies = [
  "bifrostlink",
  "bytes",
@@ -1835,7 +1835,7 @@
 
 [[package]]
 name = "polkit-backend"
-version = "0.1.4"
+version = "0.1.6"
 dependencies = [
  "anyhow",
  "clap",
@@ -2055,7 +2055,7 @@
 
 [[package]]
 name = "remowt-agent"
-version = "0.1.4"
+version = "0.1.6"
 dependencies = [
  "anyhow",
  "bifrostlink",
@@ -2083,7 +2083,7 @@
 
 [[package]]
 name = "remowt-client"
-version = "0.1.4"
+version = "0.1.6"
 dependencies = [
  "anyhow",
  "bifrostlink",
@@ -2106,7 +2106,7 @@
 
 [[package]]
 name = "remowt-endpoints"
-version = "0.1.4"
+version = "0.1.6"
 dependencies = [
  "anyhow",
  "bifrostlink",
@@ -2124,7 +2124,7 @@
 
 [[package]]
 name = "remowt-link-shared"
-version = "0.1.4"
+version = "0.1.6"
 dependencies = [
  "bifrostlink",
  "bytes",
@@ -2138,7 +2138,7 @@
 
 [[package]]
 name = "remowt-plugin"
-version = "0.1.4"
+version = "0.1.6"
 dependencies = [
  "anyhow",
  "bifrostlink",
@@ -2152,7 +2152,7 @@
 
 [[package]]
 name = "remowt-polkit-shared"
-version = "0.1.4"
+version = "0.1.6"
 dependencies = [
  "nix",
  "serde",
@@ -2161,7 +2161,7 @@
 
 [[package]]
 name = "remowt-ssh"
-version = "0.1.4"
+version = "0.1.6"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -2189,7 +2189,7 @@
 
 [[package]]
 name = "remowt-ui-prompt"
-version = "0.1.4"
+version = "0.1.6"
 dependencies = [
  "anyhow",
  "bifrostlink",
modifiedCargo.tomldiffbeforeafterboth
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -3,18 +3,18 @@
 resolver = "2"
 
 [workspace.package]
-version = "0.1.4"
+version = "0.1.6"
 license = "MIT"
 edition = "2021"
 repository = "https://git.delta.rocks/r/remowt"
 
 [workspace.dependencies]
-remowt-client = { version = "0.1.3", path = "crates/remowt-client" }
-remowt-polkit-shared = { version = "0.1.3", path = "crates/polkit-shared" }
-remowt-link-shared = { version = "0.1.3", path = "crates/remowt-link-shared" }
-remowt-plugin = { version = "0.1.3", path = "crates/remowt-plugin" }
-remowt-ui-prompt = { version = "0.1.3", path = "crates/remowt-ui-prompt" }
-remowt-endpoints = { version = "0.1.3", path = "crates/remowt-endpoints" }
+remowt-client = { version = "0.1.6", path = "crates/remowt-client" }
+remowt-polkit-shared = { version = "0.1.6", path = "crates/polkit-shared" }
+remowt-link-shared = { version = "0.1.6", path = "crates/remowt-link-shared" }
+remowt-plugin = { version = "0.1.6", path = "crates/remowt-plugin" }
+remowt-ui-prompt = { version = "0.1.6", path = "crates/remowt-ui-prompt" }
+remowt-endpoints = { version = "0.1.6", path = "crates/remowt-endpoints" }
 
 bifrostlink = "0.2.0"
 bifrostlink-macros = "0.2.0"
@@ -52,7 +52,9 @@
 thiserror = "2.0.18"
 
 [profile.release]
-panic = "abort"
+panic = "unwind"
 opt-level = "z"
 lto = true
 codegen-units = 1
+debug = "full"
+split-debuginfo = "off"
modifiedcmds/remowt-agent/src/main.rsdiffbeforeafterboth
--- a/cmds/remowt-agent/src/main.rs
+++ b/cmds/remowt-agent/src/main.rs
@@ -2,6 +2,7 @@
 use std::collections::{BTreeMap, HashMap};
 use std::fs::Permissions;
 use std::future::pending;
+use std::io;
 use std::os::unix::fs::PermissionsExt as _;
 use std::path::PathBuf;
 use std::sync::{Arc, Mutex, OnceLock};
@@ -230,7 +231,7 @@
 
 fn main() -> anyhow::Result<()> {
 	tracing_subscriber::fmt()
-		.with_writer(std::io::stderr)
+		.with_writer(io::stderr)
 		.without_time()
 		.init();
 	let opts = Opts::parse();
modifiedcrates/remowt-client/src/lib.rsdiffbeforeafterboth
--- a/crates/remowt-client/src/lib.rs
+++ b/crates/remowt-client/src/lib.rs
@@ -17,13 +17,15 @@
 use russh::keys::ssh_key::PublicKey;
 use russh::Channel;
 use tempfile::TempDir;
+use tokio::io::AsyncRead;
 use tokio::net::UnixListener;
 use tokio::sync::oneshot;
+use tokio::task::JoinHandle;
 use tokio::{
 	fs,
-	io::{AsyncBufReadExt as _, AsyncReadExt as _, AsyncWriteExt as _, BufReader, DuplexStream},
+	io::{AsyncBufReadExt as _, AsyncReadExt as _, AsyncWriteExt as _, BufReader},
 };
-use tracing::{debug, warn};
+use tracing::{debug, info, warn};
 use uuid::Uuid;
 
 pub mod editor;
@@ -97,7 +99,7 @@
 
 	let mut child = SshExecChild::from_exec(ch);
 	drop(child.stdin);
-	drain_stderr(child.stderr, cmd.to_owned());
+	drain_to_tracing(child.stderr, cmd.to_owned(), true);
 
 	let mut out = Vec::new();
 	child.stdout.read_to_end(&mut out).await?;
@@ -317,7 +319,7 @@
 			.await?;
 
 		let child = SshExecChild::from_exec(cmd_chan);
-		drain_stderr(child.stderr, "agent".to_owned());
+		drain_to_tracing(child.stderr, "agent".to_owned(), true);
 		rpc.add_direct(
 			Address::Agent,
 			child_port(child.stdout, child.stdin),
@@ -518,20 +520,33 @@
 	}
 }
 
-fn drain_stderr(stream: DuplexStream, context: String) {
+pub(crate) fn drain_to_tracing(
+	stream: impl AsyncRead + Unpin + 'static + Send,
+	context: String,
+	stderr: bool,
+) -> JoinHandle<()> {
 	tokio::spawn(async move {
-		let mut reader = BufReader::new(stream).lines();
+		let mut reader = BufReader::new(stream);
+		let mut buf = Vec::with_capacity(4096);
 		loop {
-			match reader.next_line().await {
-				Ok(Some(line)) => warn!(context = %context, "{line}"),
-				Ok(None) => break,
+			buf.clear();
+			match reader.read_until(b'\n', &mut buf).await {
+				Ok(0) => break,
+				Ok(_) => {
+					let line = String::from_utf8_lossy(buf.strip_suffix(b"\n").unwrap_or(&buf));
+					if stderr {
+						warn!(context = %context, "{line}");
+					} else {
+						info!(context = %context, "{line}");
+					}
+				}
 				Err(e) => {
-					warn!(context = %context, "stderr read failed: {e}");
+					warn!(context = %context, "child stdio read failed: {e}");
 					break;
 				}
 			}
 		}
-	});
+	})
 }
 
 fn local_runtime_dir() -> Result<(Utf8PathBuf, Option<TempDir>)> {
modifiedcrates/remowt-client/src/subprocess.rsdiffbeforeafterboth
6use remowt_endpoints::subprocess::{ProcId, SpawnSpec, StderrSpec, StdioSpec, SubprocessClient};6use remowt_endpoints::subprocess::{ProcId, SpawnSpec, StderrSpec, StdioSpec, SubprocessClient};
7use remowt_link_shared::BifConfig;7use remowt_link_shared::BifConfig;
8use serde::de::DeserializeOwned;8use serde::de::DeserializeOwned;
9use tokio::io::{AsyncBufReadExt as _, AsyncWriteExt as _, BufReader};9use tokio::io::AsyncWriteExt as _;
10use tokio::select;10use tokio::select;
11use tokio_util::codec::{BytesCodec, FramedRead, LinesCodec};11use tokio_util::codec::{BytesCodec, FramedRead, LinesCodec};
12use tracing::{debug, info, warn};12use tracing::{debug, info, warn};
1313
14use crate::forwarded::{RemowtListener, RemowtStream};14use crate::forwarded::{RemowtListener, RemowtStream};
15use crate::Remowt;15use crate::{drain_to_tracing, Remowt};
1616
17#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]17#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
18pub enum StdioMode {18pub enum StdioMode {
62			client,62			client,
63		} = self;63		} = self;
64		drop(stdin);64		drop(stdin);
65		let drain_out = async move {
65		drop(stdout);66			if let Some(s) = stdout {
67				drain_to_tracing(s, "<child stdout>".to_owned(), false).await;
68			}
69		};
70		let drain_err = async move {
66		drop(stderr);71			if let Some(s) = stderr {
72				drain_to_tracing(s, "<child stderr>".to_owned(), true).await;
73			}
74		};
75		let wait = async move {
67		client76			client
68			.wait(id)77				.wait(id)
69			.await?78				.await?
70			.map_err(|e| anyhow!("agent wait failed: {e}"))79				.map_err(|e| anyhow!("agent wait failed: {e}"))
80		};
81		let (code, _, _) = tokio::join!(wait, drain_out, drain_err);
82		code
71	}83	}
7284
73	pub async fn kill(&self, signal: i32) -> Result<()> {85	pub async fn kill(&self, signal: i32) -> Result<()> {
163		);175		);
164176
165		let stdin_stream = handle_stdin(stdin, stdin_res?, &program);177		let stdin_stream = handle_stdin(stdin, stdin_res?, &program);
166		let stdout_stream = handle_output(stdout, stdout_res?, &program, false);178		let stdout_stream = handle_output(stdout, stdout_res?, &program);
167		let stderr_stream = handle_output_err(stderr, stderr_res?, &program);179		let stderr_stream = handle_output_err(stderr, stderr_res?, &program);
168180
169		Ok(RemowtChild {181		Ok(RemowtChild {
218fn handle_output(230fn handle_output(mode: StdioMode, s: Option<RemowtStream>, program: &str) -> Option<RemowtStream> {
219	mode: StdioMode,
220	s: Option<RemowtStream>,
221	program: &str,
222	is_stderr: bool,
223) -> Option<RemowtStream> {
224	match mode {231	match mode {
225		StdioMode::Pipe => s,232		StdioMode::Pipe => s,
226		StdioMode::Inherit => {233		StdioMode::Inherit => {
227			if let Some(s) = s {234			if let Some(s) = s {
228				let program = program.to_owned();235				let program = program.to_owned();
229				tokio::spawn(pump_to_tracing(s, program, is_stderr));236				tokio::spawn(drain_to_tracing(s, program, false));
230			}237			}
231			None238			None
232		}239		}
244		StderrMode::Inherit => {251		StderrMode::Inherit => {
245			if let Some(s) = s {252			if let Some(s) = s {
246				let program = program.to_owned();253				let program = program.to_owned();
247				tokio::spawn(pump_to_tracing(s, program, true));254				tokio::spawn(drain_to_tracing(s, program, true));
248			}255			}
249			None256			None
250		}257		}
251		StderrMode::MergeWithStdout | StderrMode::Null => None,258		StderrMode::MergeWithStdout | StderrMode::Null => None,
252	}259	}
253}260}
254
255async fn pump_to_tracing(stream: RemowtStream, program: String, is_stderr: bool) {
256	let mut reader = BufReader::new(stream).lines();
257	loop {
258		match reader.next_line().await {
259			Ok(Some(line)) => {
260				if is_stderr {
261					warn!(program, "{line}");
262				} else {
263					info!(program, "{line}");
264				}
265			}
266			Ok(None) => break,
267			Err(e) => {
268				warn!(program, "child log stream error: {e}");
269				break;
270			}
271		}
272	}
273}
274261
275fn escape_bash(input: &str, out: &mut String) {262fn escape_bash(input: &str, out: &mut String) {
276	const TO_ESCAPE: &str = "$ !\"#&'()*,;<>?[\\]^`{|}";263	const TO_ESCAPE: &str = "$ !\"#&'()*,;<>?[\\]^`{|}";
422		}409		}
423	};410	};
411
412	while let Some(e) = err.next().await {
413		if let Ok(line) = e {
414			warn!(program = %program, "{line}");
415		}
416	}
417	if want_stdout {
418		if let Some(out_bytes) = out_bytes.as_mut() {
419			while let Some(o) = out_bytes.next().await {
420				if let Ok(chunk) = o {
421					buf.as_mut().expect("want_stdout").extend_from_slice(&chunk);
422				}
423			}
424		}
425	} else if let Some(out_lines) = out_lines.as_mut() {
426		while let Some(o) = out_lines.next().await {
427			if let Ok(line) = o {
428				info!(program = %program, "{line}");
429			}
430		}
431	}
424432
425	match exit {433	match exit {
426		Some(0) => Ok(buf),434		Some(0) => Ok(buf),
modifiedcrates/remowt-link-shared/src/port.rsdiffbeforeafterboth
--- a/crates/remowt-link-shared/src/port.rs
+++ b/crates/remowt-link-shared/src/port.rs
@@ -3,10 +3,8 @@
 use bifrostlink::Port;
 use bytes::{Bytes, BytesMut};
 use tokio::io::{AsyncRead, AsyncReadExt as _, AsyncWrite, AsyncWriteExt as _};
+use tokio::select;
 
-/// Wire a length-prefixed duplex byte stream (e.g. a child process's
-/// stdout/stdin) into a bifrost [`Port`]. Each frame is a big-endian `u32`
-/// length followed by that many payload bytes.
 pub fn child_port<R, W>(mut reader: R, mut writer: W) -> Port
 where
 	R: AsyncRead + Unpin + Send + 'static,
@@ -18,13 +16,13 @@
 				let len = match reader.read_u32().await {
 					Ok(len) => len,
 					Err(e) => {
-						tracing::error!("child read failed: {e}");
+						log_read_end(&e);
 						break;
 					}
 				};
 				let mut buf = BytesMut::zeroed(len as usize);
 				if let Err(e) = reader.read_exact(&mut buf).await {
-					tracing::error!("child read failed: {e}");
+					log_read_end(&e);
 					break;
 				}
 				if tx.send(buf.freeze()).is_err() {
@@ -35,15 +33,45 @@
 		let write_task = async move {
 			while let Some(msg) = rx.recv().await {
 				if let Err(e) = write_frame(&mut writer, msg).await {
-					tracing::error!("child write failed: {e}");
+					log_write_end(&e);
 					break;
 				}
 			}
 		};
-		tokio::join!(read_task, write_task);
+		select! {
+			_ = read_task => {},
+			_ = write_task => {},
+		}
 	})
 }
 
+fn log_read_end(e: &io::Error) {
+	if matches!(
+		e.kind(),
+		io::ErrorKind::UnexpectedEof
+			| io::ErrorKind::BrokenPipe
+			| io::ErrorKind::ConnectionReset
+			| io::ErrorKind::ConnectionAborted
+	) {
+		tracing::debug!("child read ended: {e}");
+	} else {
+		tracing::error!("child read failed: {e}");
+	}
+}
+
+fn log_write_end(e: &io::Error) {
+	if matches!(
+		e.kind(),
+		io::ErrorKind::BrokenPipe
+			| io::ErrorKind::ConnectionReset
+			| io::ErrorKind::ConnectionAborted
+	) {
+		tracing::debug!("child write ended: {e}");
+	} else {
+		tracing::error!("child write failed: {e}");
+	}
+}
+
 async fn write_frame<W: AsyncWrite + Unpin>(writer: &mut W, msg: Bytes) -> io::Result<()> {
 	let len = u32::try_from(msg.len())
 		.map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "message larger than 4GB"))?;