git.delta.rocks / fleet / refs/commits / b66112ef1437

--- a/cmds/fleet/src/main.rs
+++ b/cmds/fleet/src/main.rs
@@ -180,7 +180,7 @@
 		)
 	};
 
-	let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
+	let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("noq_udp=warn,info"));
 
 	let tree = {
 		#[cfg(feature = "indicatif")]

18use remowt_ui_prompt::{PrependSourcePrompter, Source};18use remowt_ui_prompt::{PrependSourcePrompter, Source};
19use tokio::io::unix::AsyncFd;19use tokio::io::unix::AsyncFd;
20use tokio::io::{AsyncRead, ReadBuf};20use tokio::io::{AsyncRead, ReadBuf};
21use tokio::signal::unix::{signal, SignalKind};21use tokio::signal::unix::{SignalKind, signal};
22use tracing::debug;22use tracing::debug;
23use tracing_subscriber::prelude::*;23use tracing_subscriber::EnvFilter;
24use tracing_subscriber::EnvFilter;24use tracing_subscriber::prelude::*;
2525
26#[derive(Parser)]26#[derive(Parser)]
27enum Opts {27enum Opts {

--- a/remowt/crates/remowt-client/src/lib.rs
+++ b/remowt/crates/remowt-client/src/lib.rs
@@ -27,7 +27,7 @@
 	fs,
 	io::{AsyncBufReadExt as _, AsyncReadExt as _, AsyncWriteExt as _, BufReader},
 };
-use tracing::{debug, info, warn};
+use tracing::{Instrument as _, debug, info, warn};
 use uuid::Uuid;
 
 pub mod editor;
@@ -655,28 +655,31 @@
 	context: String,
 	stderr: bool,
 ) -> JoinHandle<()> {
-	tokio::spawn(async move {
-		let mut reader = BufReader::new(stream);
-		let mut buf = Vec::with_capacity(4096);
-		loop {
-			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}");
+	tokio::spawn(
+		async move {
+			let mut reader = BufReader::new(stream);
+			let mut buf = Vec::with_capacity(4096);
+			loop {
+				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, "child stdio read failed: {e}");
+						break;
 					}
-				}
-				Err(e) => {
-					warn!(context = %context, "child stdio read failed: {e}");
-					break;
 				}
 			}
 		}
-	})
+		.in_current_span(),
+	)
 }
 
 fn local_runtime_dir() -> Result<(Utf8PathBuf, Option<TempDir>)> {

--- a/remowt/crates/remowt-endpoints/src/nix_daemon.rs
+++ b/remowt/crates/remowt-endpoints/src/nix_daemon.rs
@@ -1,11 +1,12 @@
 use std::process::Stdio;
 use std::sync::Arc;
 
+use bifrostlink::Config;
 use bifrostlink::declarative::endpoints;
-use bifrostlink::Config;
 use remowt_link_shared::iroh_tunnel::{TunnelAddr, TunnelDialer};
 use serde::{Deserialize, Serialize};
 use std::result::Result;
+use tokio::io::{self, AsyncWriteExt as _};
 use tokio::process::Command;
 
 #[derive(Clone)]
@@ -47,11 +48,16 @@
 		let mut stdin = child.stdin.take().expect("piped");
 		let mut stdout = child.stdout.take().expect("piped");
 		tokio::spawn(async move {
-			let (mut tr, mut tw) = tokio::io::split(tunnel);
-			let _ = tokio::join!(
-				tokio::io::copy(&mut tr, &mut stdin),
-				tokio::io::copy(&mut stdout, &mut tw),
-			);
+			let (mut tr, mut tw) = io::split(tunnel);
+			let to_child = async {
+				let _ = io::copy(&mut tr, &mut stdin).await;
+				let _ = stdin.shutdown().await;
+			};
+			let from_child = async {
+				let _ = io::copy(&mut stdout, &mut tw).await;
+				let _ = tw.shutdown().await;
+			};
+			tokio::join!(to_child, from_child);
 			let _ = child.wait().await;
 		});
 		Ok(())