..

core/src/tasks/scheduler.rs

@@ -70,26 +70,26 @@ impl Scheduler {
 					Ok(fut) = rx.recv() => {
 						fut.await;
 					}
-					Ok((id, mut task)) = file.recv() => {
+					Ok((id, mut op)) = file.recv() => {
 						if !running.read().exists(id) {
-							trace!("Skipping task {:?} as it was removed", task);
+							trace!("Skipping task {:?} as it was removed", op);
 							continue;
 						}
-						if let Err(e) = file.work(&mut task).await {
+						if let Err(e) = file.work(&mut op).await {
-							info!("Failed to work on task {:?}: {}", task, e);
+							info!("Failed to work on task {:?}: {}", op, e);
 						} else {
-							trace!("Finished task {:?}", task);
+							trace!("Finished task {:?}", op);
 						}
 					}
-					Ok((id, mut task)) = precache.recv() => {
+					Ok((id, mut op)) = precache.recv() => {
 						if !running.read().exists(id) {
-							trace!("Skipping task {:?} as it was removed", task);
+							trace!("Skipping task {:?} as it was removed", op);
 							continue;
 						}
-						if let Err(e) = precache.work(&mut task).await {
+						if let Err(e) = precache.work(&mut op).await {
-							info!("Failed to work on task {:?}: {}", task, e);
+							info!("Failed to work on task {:?}: {}", op, e);
 						} else {
-							trace!("Finished task {:?}", task);
+							trace!("Finished task {:?}", op);
 						}
 					}
 				}
@@ -102,14 +102,20 @@ impl Scheduler {
 		let running = self.running.clone();
 
 		tokio::spawn(async move {
-			while let Some(task) = rx.recv().await {
+			while let Some(op) = rx.recv().await {
-				match task {
+				match op {
 					TaskOp::New(id, size) => {
 						if let Some(task) = running.write().get(id) {
 							task.found += 1;
 							task.todo += size;
 						}
 					}
+					TaskOp::Log(id, line) => {
+						if let Some(task) = running.write().get(id) {
+							task.logs.push_str(&line);
+							task.logs.push('\n');
+						}
+					}
 					TaskOp::Adv(id, processed, size) => {
 						let mut running = running.write();
 						if let Some(task) = running.get(id) {
@@ -303,7 +309,7 @@ impl Scheduler {
 			})
 		});
 
-		let _ = self.todo.send_blocking({
+		tokio::spawn({
 			let process = self.process.clone();
 			let opener = opener.clone();
 			async move {
@@ -312,7 +318,6 @@ impl Scheduler {
 					.await
 					.ok();
 			}
-			.boxed()
 		});
 	}
 

core/src/tasks/tasks.rs

@@ -12,6 +12,7 @@ pub struct Task {
 	pub id:    usize,
 	pub name:  String,
 	pub stage: TaskStage,
+	pub logs:  String,
 
 	pub found:     u32,
 	pub processed: u32,
@@ -22,7 +23,16 @@ pub struct Task {
 
 impl Task {
 	pub fn new(id: usize, name: String) -> Self {
-		Self { id, name, stage: Default::default(), found: 0, processed: 0, todo: 0, done: 0 }
+		Self {
+			id,
+			name,
+			stage: Default::default(),
+			logs: Default::default(),
+			found: 0,
+			processed: 0,
+			todo: 0,
+			done: 0,
+		}
 	}
 }
 
@@ -30,6 +40,8 @@ impl Task {
 pub enum TaskOp {
 	// task_id, size
 	New(usize, u64),
+	// task_id, line
+	Log(usize, String),
 	// task_id, processed, size
 	Adv(usize, u32, u64),
 	// task_id

core/src/tasks/workers/file.rs

@@ -4,7 +4,7 @@ use anyhow::Result;
 use futures::{future::BoxFuture, FutureExt};
 use shared::{calculate_size, copy_with_progress};
 use tokio::{fs, io::{self, ErrorKind::{AlreadyExists, NotFound}}, sync::mpsc};
-use tracing::{info, trace};
+use tracing::trace;
 
 use crate::tasks::TaskOp;
 
@@ -72,8 +72,8 @@ impl File {
 		})
 	}
 
-	pub(crate) async fn work(&self, task: &mut FileOp) -> Result<()> {
+	pub(crate) async fn work(&self, op: &mut FileOp) -> Result<()> {
-		match task {
+		match op {
 			FileOp::Paste(task) => {
 				match fs::remove_file(&task.to).await {
 					Err(e) if e.kind() != NotFound => Err(e)?,
@@ -90,7 +90,7 @@ impl File {
 							break;
 						}
 						Ok(n) => {
-							trace!("Paste task advanced {}: {:?}", n, task);
+							self.log(task.id, format!("Paste task advanced {}: {:?}", n, task))?;
 							self.sch.send(TaskOp::Adv(task.id, 0, n))?
 						}
 						Err(e) if e.kind() == NotFound => {
@@ -100,7 +100,7 @@ impl File {
 						// Operation not permitted (os error 1)
 						// Attribute not found (os error 93)
 						Err(e) if task.retry < 3 && matches!(e.raw_os_error(), Some(1) | Some(93)) => {
-							trace!("Paste task retry: {:?}", task);
+							self.log(task.id, format!("Paste task retry: {:?}", task))?;
 							task.retry += 1;
 							return Ok(self.tx.send(FileOp::Paste(task.clone())).await?);
 						}
@@ -113,7 +113,7 @@ impl File {
 				let src = match fs::read_link(&task.from).await {
 					Ok(src) => src,
 					Err(e) if e.kind() == NotFound => {
-						trace!("Link task partially done: {:?}", task);
+						self.log(task.id, format!("Link task partially done: {:?}", task))?;
 						return Ok(self.sch.send(TaskOp::Adv(task.id, 1, task.length))?);
 					}
 					Err(e) => Err(e)?,
@@ -132,7 +132,7 @@ impl File {
 			FileOp::Delete(task) => {
 				if let Err(e) = fs::remove_file(&task.target).await {
 					if e.kind() != NotFound && fs::symlink_metadata(&task.target).await.is_ok() {
-						info!("Delete task failed: {:?}, {}", task, e);
+						self.log(task.id, format!("Delete task failed: {:?}, {}", task, e))?;
 						Err(e)?
 					}
 				}
@@ -156,6 +156,9 @@ impl File {
 		Ok(())
 	}
 
+	#[inline]
+	fn log(&self, id: usize, line: String) -> Result<()> { Ok(self.sch.send(TaskOp::Log(id, line))?) }
+
 	#[inline]
 	fn done(&self, id: usize) -> Result<()> { Ok(self.sch.send(TaskOp::Done(id))?) }
 
@@ -189,7 +192,7 @@ impl File {
 			let dest = root.join(src.components().skip(skip).collect::<PathBuf>());
 			match fs::create_dir(&dest).await {
 				Err(e) if e.kind() != AlreadyExists => {
-					info!("Create dir failed: {:?}, {}", dest, e);
+					self.log(task.id, format!("Create dir failed: {:?}, {}", dest, e))?;
 					continue;
 				}
 				_ => {}
@@ -198,7 +201,7 @@ impl File {
 			let mut it = match fs::read_dir(&src).await {
 				Ok(it) => it,
 				Err(e) => {
-					info!("Read dir failed: {:?}, {}", src, e);
+					self.log(task.id, format!("Read dir failed: {:?}, {}", src, e))?;
 					continue;
 				}
 			};

core/src/tasks/workers/process.rs

@@ -1,7 +1,7 @@
 use std::{ffi::OsString, process::Stdio};
 
 use anyhow::Result;
-use tokio::{process::Command, select, sync::{mpsc, oneshot}};
+use tokio::{io::{AsyncBufReadExt, BufReader}, process::Command, select, sync::{mpsc, oneshot}};
 use tracing::trace;
 
 use crate::{emit, tasks::TaskOp, BLOCKER};
@@ -22,6 +22,9 @@ pub(crate) struct ProcessOpOpen {
 impl Process {
 	pub(crate) fn new(sch: mpsc::UnboundedSender<TaskOp>) -> Self { Self { sch } }
 
+	#[inline]
+	fn log(&self, id: usize, line: String) -> Result<()> { Ok(self.sch.send(TaskOp::Log(id, line))?) }
+
 	#[inline]
 	fn done(&self, id: usize) -> Result<()> { Ok(self.sch.send(TaskOp::Done(id))?) }
 
@@ -43,18 +46,29 @@ impl Process {
 			return Ok(());
 		}
 
-		let status = Command::new(&task.cmd)
-			.args(&task.args)
-			.stdout(Stdio::null())
-			.stderr(Stdio::null())
-			.kill_on_drop(true)
-			.status();
-
 		self.sch.send(TaskOp::New(task.id, 0))?;
-		select! {
+		let mut child = Command::new(&task.cmd)
-			_ = task.cancel.closed() => {},
+			.args(&task.args)
-			Ok(status) = status => {
+			.stdout(Stdio::piped())
-				trace!("{} exited with {:?}", task.cmd, status);
+			.stderr(Stdio::piped())
+			.kill_on_drop(true)
+			.spawn()?;
+
+		let mut stdout = BufReader::new(child.stdout.take().unwrap()).lines();
+		let mut stderr = BufReader::new(child.stderr.take().unwrap()).lines();
+		loop {
+			select! {
+				_ = task.cancel.closed() => break,
+				Ok(Some(line)) = stdout.next_line() => {
+					self.log(task.id, line)?;
+				}
+				Ok(Some(line)) = stderr.next_line() => {
+					self.log(task.id, line)?;
+				}
+				Ok(status) = child.wait() => {
+					self.log(task.id, format!("Exited with {:?}", status))?;
+					break;
+				}
 			}
 		}
 		self.done(task.id)

cspell.json

@@ -1 +1 @@
-{"language":"en","flagWords":[],"words":["Punct","KEYMAP","splitn","crossterm","YAZI","unar","peekable","ratatui","syntect","pbpaste","pbcopy","ffmpegthumbnailer","oneshot","Posix","Lsar","XADDOS","zoxide","cands","Deque","precache","imageops","IFBLK","IFCHR","IFDIR","IFIFO","IFLNK","IFMT","IFSOCK","IRGRP","IROTH","IRUSR","ISGID","ISUID","ISVTX","IWGRP","IWOTH","IWUSR","IXGRP","IXOTH","IXUSR","libc","winsize","TIOCGWINSZ","xpixel","ypixel","ioerr","appender","Catppuccin","macchiato","gitmodules","Dotfiles","bashprofile","vimrc","flac","webp","exiftool","mediainfo","ripgrep","nvim","indexmap","indexmap","unwatch","canonicalize","serde","fsevent","Ueberzug","iterm","wezterm","sixel","chafa","ueberzugpp","️ Überzug","️ Überzug","Konsole","Alacritty","Überzug"],"version":"0.2"}
+{"words":["Punct","KEYMAP","splitn","crossterm","YAZI","unar","peekable","ratatui","syntect","pbpaste","pbcopy","ffmpegthumbnailer","oneshot","Posix","Lsar","XADDOS","zoxide","cands","Deque","precache","imageops","IFBLK","IFCHR","IFDIR","IFIFO","IFLNK","IFMT","IFSOCK","IRGRP","IROTH","IRUSR","ISGID","ISUID","ISVTX","IWGRP","IWOTH","IWUSR","IXGRP","IXOTH","IXUSR","libc","winsize","TIOCGWINSZ","xpixel","ypixel","ioerr","appender","Catppuccin","macchiato","gitmodules","Dotfiles","bashprofile","vimrc","flac","webp","exiftool","mediainfo","ripgrep","nvim","indexmap","indexmap","unwatch","canonicalize","serde","fsevent","Ueberzug","iterm","wezterm","sixel","chafa","ueberzugpp","️ Überzug","️ Überzug","Konsole","Alacritty","Überzug","pkgs","paru","unarchiver"],"version":"0.2","language":"en","flagWords":[]}