@@ -1,25 +1,40 @@
11//! Run only configured executable basenames beside the application executable.
22//! Application arguments and policy belong to the caller, not this plugin.
33
4+use parking_lot::Mutex;
45use serde::Serialize;
6+use shared_child::SharedChild;
57use std::{
8+ collections::HashMap,
69 env,
710 io::{self, Read, Write},
811 path::PathBuf,
9− process::{Child, Command, ExitStatus, Stdio},
10− sync::mpsc,
11− thread,
12+ process::{Child, ChildStdin, Command, ExitStatus, Stdio},
13+ sync::{
14+ atomic::{AtomicBool, Ordering},
15+ mpsc, Arc,
16+ },
17+ thread::{self, JoinHandle},
1218};
1319use tauri::{
20+ ipc::{InvokeBody, Request},
1421 plugin::{Builder, TauriPlugin},
15− Manager, Runtime, State,
22+ webview::PageLoadEvent,
23+ AppHandle, Manager, RunEvent, Runtime, State, Webview, WindowEvent,
1624};
1725
26+const STREAM_HEADER: &str = "x-tauri-bundled-process-stream";
27+
1828struct BundledProcess {
1929 directory: PathBuf,
2030 executables: Vec<String>,
2131}
2232
33+#[derive(Default)]
34+struct Streams {
35+ active: Mutex<HashMap<String, Arc<Stream>>>,
36+}
37+
2338fn validate_basename(name: &str) -> Result<(), String> {
2439 // Reject Windows separators/drive prefixes even on Unix, and vice versa.
2540 if name.is_empty() || name == "." || name == ".." || name.contains(['/', '\\', ':', '\0']) {
@@ -60,23 +75,36 @@ struct ProcessResult {
6075
6176// Every exit path, including pipe-worker spawn errors and panics, reaps the child.
6277struct ChildGuard {
63− child: Child,
78+ child: Option<Child>,
6479 reaped: bool,
6580}
6681
6782impl ChildGuard {
6883 fn wait(&mut self) -> io::Result<ExitStatus> {
69− let status = self.child.wait()?;
84+ let status = self
85+ .child
86+ .as_mut()
87+ .expect("child guard always owns a child before reaping")
88+ .wait()?;
7089 self.reaped = true;
7190 Ok(status)
7291 }
92+
93+ fn kill_and_reap(&mut self) {
94+ let Some(child) = self.child.as_mut() else {
95+ return;
96+ };
97+ let _ = child.kill();
98+ if child.wait().is_ok() {
99+ self.reaped = true;
100+ }
101+ }
73102}
74103
75104impl Drop for ChildGuard {
76105 fn drop(&mut self) {
77106 if !self.reaped {
78− let _ = self.child.kill();
79− let _ = self.child.wait();
107+ self.kill_and_reap();
80108 }
81109 }
82110}
@@ -120,26 +148,20 @@ fn execute(mut command: Command, input: String) -> Result<ProcessResult, String>
120148 // Keep the guard inside the scope: on error, kill before scope joins
121149 // workers that may otherwise remain blocked on the child's pipes.
122150 let mut child = ChildGuard {
123− child: command
124− .spawn()
125− .map_err(|error| format!("Cannot spawn bundled executable: {error}"))?,
151+ child: Some(
152+ command
153+ .spawn()
154+ .map_err(|error| format!("Cannot spawn bundled executable: {error}"))?,
155+ ),
126156 reaped: false,
127157 };
128− let mut stdin = child
158+ let process = child
129159 .child
130− .stdin
131− .take()
132− .ok_or("Child stdin is unavailable")?;
133− let mut stdout = child
134− .child
135− .stdout
136− .take()
137− .ok_or("Child stdout is unavailable")?;
138− let mut stderr = child
139− .child
140− .stderr
141− .take()
142− .ok_or("Child stderr is unavailable")?;
160+ .as_mut()
161+ .expect("child guard always owns a child before reaping");
162+ let mut stdin = process.stdin.take().ok_or("Child stdin is unavailable")?;
163+ let mut stdout = process.stdout.take().ok_or("Child stdout is unavailable")?;
164+ let mut stderr = process.stderr.take().ok_or("Child stderr is unavailable")?;
143165 let (sender, receiver) = mpsc::channel();
144166 pipe_worker(scope, &sender, "stdin", move || {
145167 // Closing stdin is required even for empty input. A child is allowed
@@ -187,6 +209,307 @@ fn execute(mut command: Command, input: String) -> Result<ProcessResult, String>
187209 })
188210}
189211
212+type OutputReader = JoinHandle<io::Result<Vec<u8>>>;
213+
214+struct Stream {
215+ owner: String,
216+ // stdin has its own lock so cancellation can kill the child while a write is
217+ // blocked in the kernel. The stream registry lock is never held for I/O.
218+ stdin: Mutex<Option<ChildStdin>>,
219+ // SharedChild keeps waiting, killing, and reaping independently callable.
220+ // In particular, cancel can kill while finish is blocked in wait().
221+ child: Arc<SharedChild>,
222+ stdout: Mutex<Option<OutputReader>>,
223+ stderr: Mutex<Option<OutputReader>>,
224+ closing: AtomicBool,
225+ writing: AtomicBool,
226+ waiting: AtomicBool,
227+}
228+
229+struct Writing<'a>(&'a AtomicBool);
230+
231+impl Drop for Writing<'_> {
232+ fn drop(&mut self) {
233+ self.0.store(false, Ordering::Release);
234+ }
235+}
236+
237+struct Waiting<'a>(&'a AtomicBool);
238+
239+impl Drop for Waiting<'_> {
240+ fn drop(&mut self) {
241+ self.0.store(false, Ordering::Release);
242+ }
243+}
244+
245+impl Stream {
246+ fn write(&self, bytes: &[u8]) -> Result<(), String> {
247+ if self.closing.load(Ordering::Acquire) {
248+ return Err("Process stream is closing".into());
249+ }
250+ self.writing
251+ .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
252+ .map_err(|_| "A process stream write is already in flight".to_owned())?;
253+ let _writing = Writing(&self.writing);
254+ let mut stdin = self.stdin.lock();
255+ if self.closing.load(Ordering::Acquire) {
256+ return Err("Process stream is closing".into());
257+ }
258+ let stdin = stdin.as_mut().ok_or("Process stream stdin is closed")?;
259+ stdin
260+ .write_all(bytes)
261+ .map_err(|error| format!("Cannot write child stdin: {error}"))
262+ }
263+
264+ fn finish(&self) -> Result<ProcessResult, String> {
265+ if self.closing.swap(true, Ordering::AcqRel) {
266+ return Err("Process stream is already closing".into());
267+ }
268+ self.stdin.lock().take();
269+ let status = self.reap(false)?;
270+ let (stdout, stderr) = self.collect_output()?;
271+ Ok(ProcessResult {
272+ success: status.success(),
273+ code: status.code().unwrap_or(-1),
274+ stdout: decode_output(stdout),
275+ stderr: decode_output(stderr),
276+ })
277+ }
278+
279+ fn cancel(&self) -> Result<(), String> {
280+ self.closing.store(true, Ordering::Release);
281+ self.reap(true)?;
282+ self.discard_output();
283+ Ok(())
284+ }
285+
286+ fn abandon(&self) {
287+ self.closing.store(true, Ordering::Release);
288+ let _ = self.reap(true);
289+ self.discard_output();
290+ }
291+
292+ fn reap(&self, kill: bool) -> Result<ExitStatus, String> {
293+ // SharedChild allows kill and wait from independent threads. This means
294+ // cancellation remains live while either a stdin write or finish wait
295+ // is blocked.
296+ if kill {
297+ let _ = self.child.kill();
298+ }
299+ self.waiting.store(true, Ordering::Release);
300+ let _waiting = Waiting(&self.waiting);
301+ self.child
302+ .wait()
303+ .map_err(|error| format!("Cannot wait for bundled executable: {error}"))
304+ }
305+
306+ fn collect_output(&self) -> Result<(Vec<u8>, Vec<u8>), String> {
307+ let stdout = self
308+ .stdout
309+ .lock()
310+ .take()
311+ .ok_or("Process stream stdout is unavailable")?;
312+ let stderr = self
313+ .stderr
314+ .lock()
315+ .take()
316+ .ok_or("Process stream stderr is unavailable")?;
317+ let stdout = join_output(stdout, "stdout");
318+ let stderr = join_output(stderr, "stderr");
319+ Ok((stdout?, stderr?))
320+ }
321+
322+ fn discard_output(&self) {
323+ if let Some(stdout) = self.stdout.lock().take() {
324+ let _ = stdout.join();
325+ }
326+ if let Some(stderr) = self.stderr.lock().take() {
327+ let _ = stderr.join();
328+ }
329+ }
330+}
331+
332+impl Drop for Stream {
333+ fn drop(&mut self) {
334+ self.abandon();
335+ }
336+}
337+
338+fn join_output(reader: OutputReader, name: &str) -> Result<Vec<u8>, String> {
339+ reader
340+ .join()
341+ .map_err(|_| format!("Child {name} reader panicked"))?
342+ .map_err(|error| format!("Cannot read child {name}: {error}"))
343+}
344+
345+fn output_reader<R: Read + Send + 'static>(
346+ name: &str,
347+ mut pipe: R,
348+) -> Result<OutputReader, String> {
349+ thread::Builder::new()
350+ .name(format!("bundled-process-{name}"))
351+ .spawn(move || {
352+ let mut output = Vec::new();
353+ pipe.read_to_end(&mut output)?;
354+ Ok(output)
355+ })
356+ .map_err(|error| format!("Cannot start child {name} reader: {error}"))
357+}
358+
359+fn spawn_stream(mut command: Command, owner: String) -> Result<Arc<Stream>, String> {
360+ command
361+ .stdin(Stdio::piped())
362+ .stdout(Stdio::piped())
363+ .stderr(Stdio::piped());
364+ let child = Arc::new(
365+ SharedChild::spawn(&mut command)
366+ .map_err(|error| format!("Cannot spawn bundled executable: {error}"))?,
367+ );
368+ let stdin = child.take_stdin().ok_or_else(|| {
369+ let _ = child.kill();
370+ let _ = child.wait();
371+ "Child stdin is unavailable"
372+ })?;
373+ let stdout_pipe = child.take_stdout().ok_or_else(|| {
374+ let _ = child.kill();
375+ let _ = child.wait();
376+ "Child stdout is unavailable"
377+ })?;
378+ let stderr_pipe = child.take_stderr().ok_or_else(|| {
379+ let _ = child.kill();
380+ let _ = child.wait();
381+ "Child stderr is unavailable"
382+ })?;
383+ let stdout = match output_reader("stdout", stdout_pipe) {
384+ Ok(reader) => reader,
385+ Err(error) => {
386+ let _ = child.kill();
387+ let _ = child.wait();
388+ return Err(error);
389+ }
390+ };
391+ let stderr = match output_reader("stderr", stderr_pipe) {
392+ Ok(reader) => reader,
393+ Err(error) => {
394+ let _ = child.kill();
395+ let _ = child.wait();
396+ let _ = stdout.join();
397+ return Err(error);
398+ }
399+ };
400+ Ok(Arc::new(Stream {
401+ owner,
402+ stdin: Mutex::new(Some(stdin)),
403+ child,
404+ stdout: Mutex::new(Some(stdout)),
405+ stderr: Mutex::new(Some(stderr)),
406+ closing: AtomicBool::new(false),
407+ writing: AtomicBool::new(false),
408+ waiting: AtomicBool::new(false),
409+ }))
410+}
411+
412+fn stream_id() -> Result<String, String> {
413+ let mut bytes = [0u8; 32];
414+ getrandom::fill(&mut bytes).map_err(|_| "Unable to obtain stream identifier randomness")?;
415+ const HEX: &[u8; 16] = b"0123456789abcdef";
416+ let mut id = String::with_capacity(bytes.len() * 2);
417+ for byte in bytes {
418+ id.push(HEX[(byte >> 4) as usize] as char);
419+ id.push(HEX[(byte & 15) as usize] as char);
420+ }
421+ Ok(id)
422+}
423+
424+fn insert_stream(streams: &Streams, stream: Arc<Stream>) -> Result<String, String> {
425+ let mut active = streams.active.lock();
426+ loop {
427+ let id = stream_id()?;
428+ if !active.contains_key(&id) {
429+ active.insert(id.clone(), stream);
430+ return Ok(id);
431+ }
432+ }
433+}
434+
435+fn owned_stream<R: Runtime>(
436+ streams: &Streams,
437+ webview: &Webview<R>,
438+ id: &str,
439+) -> Result<Arc<Stream>, String> {
440+ let stream = streams
441+ .active
442+ .lock()
443+ .get(id)
444+ .cloned()
445+ .ok_or("Unknown process stream")?;
446+ if stream.owner != webview.label() {
447+ return Err("Process stream belongs to a different webview".into());
448+ }
449+ Ok(stream)
450+}
451+
452+fn remove_stream(streams: &Streams, id: &str, stream: &Arc<Stream>) {
453+ let mut active = streams.active.lock();
454+ if active
455+ .get(id)
456+ .is_some_and(|current| Arc::ptr_eq(current, stream))
457+ {
458+ active.remove(id);
459+ }
460+}
461+
462+fn requested_stream<'a>(request: &'a Request<'_>) -> Result<&'a str, String> {
463+ request
464+ .headers()
465+ .get(STREAM_HEADER)
466+ .ok_or("Missing process stream header")?
467+ .to_str()
468+ .map_err(|_| "Process stream header is not valid text".to_owned())
469+}
470+
471+fn raw_stream_bytes<'a>(request: &'a Request<'_>) -> Result<&'a [u8], String> {
472+ if request
473+ .headers()
474+ .get("content-type")
475+ .and_then(|value| value.to_str().ok())
476+ != Some("application/octet-stream")
477+ {
478+ return Err("Process stream writes require application/octet-stream".into());
479+ }
480+ let InvokeBody::Raw(bytes) = request.body() else {
481+ return Err("Process stream writes require a raw ArrayBuffer body".into());
482+ };
483+ Ok(bytes)
484+}
485+
486+fn take_streams(streams: &Streams, owner: Option<&str>) -> Vec<Arc<Stream>> {
487+ let mut active = streams.active.lock();
488+ let ids: Vec<String> = active
489+ .iter()
490+ .filter_map(|(id, stream)| {
491+ if owner.map_or(true, |owner| stream.owner == owner) {
492+ Some(id.clone())
493+ } else {
494+ None
495+ }
496+ })
497+ .collect();
498+ ids.into_iter()
499+ .filter_map(|id| active.remove(&id))
500+ .collect()
501+}
502+
503+fn cleanup_streams(streams: Vec<Arc<Stream>>) {
504+ for stream in streams {
505+ stream.abandon();
506+ }
507+}
508+
509+fn cleanup_owner<R: Runtime>(app: &AppHandle<R>, owner: Option<&str>) {
510+ cleanup_streams(take_streams(app.state::<Streams>().inner(), owner));
511+}
512+
190513#[tauri::command]
191514async fn run(
192515 process: State<'_, BundledProcess>,
@@ -204,10 +527,82 @@ async fn run(
204527 .map_err(|error| format!("Bundled process worker failed: {error}"))?
205528}
206529
530+#[tauri::command]
531+fn spawn<R: Runtime>(
532+ process: State<'_, BundledProcess>,
533+ streams: State<'_, Streams>,
534+ webview: Webview<R>,
535+ executable: String,
536+ arguments: Vec<String>,
537+) -> Result<String, String> {
538+ let executable = process.resolve(&executable)?;
539+ let mut command = Command::new(executable);
540+ command.args(arguments);
541+ let stream = spawn_stream(command, webview.label().to_owned())?;
542+ insert_stream(streams.inner(), stream)
543+}
544+
545+#[tauri::command]
546+async fn write<R: Runtime>(
547+ streams: State<'_, Streams>,
548+ webview: Webview<R>,
549+ request: Request<'_>,
550+) -> Result<(), String> {
551+ let id = requested_stream(&request)?;
552+ let stream = owned_stream(streams.inner(), &webview, id)?;
553+ // Request borrows Tauri's raw body, so moving the blocking pipe write off
554+ // the IPC task requires one owned frame copy. No JSON/base64 conversion or
555+ // additional stream buffering is introduced.
556+ let bytes = raw_stream_bytes(&request)?.to_vec();
557+ tauri::async_runtime::spawn_blocking(move || stream.write(&bytes))
558+ .await
559+ .map_err(|error| format!("Bundled process worker failed: {error}"))?
560+}
561+
562+#[tauri::command]
563+async fn finish<R: Runtime>(
564+ streams: State<'_, Streams>,
565+ webview: Webview<R>,
566+ stream: String,
567+) -> Result<ProcessResult, String> {
568+ let stream_handle = owned_stream(streams.inner(), &webview, &stream)?;
569+ let result = match tauri::async_runtime::spawn_blocking({
570+ let stream_handle = Arc::clone(&stream_handle);
571+ move || stream_handle.finish()
572+ })
573+ .await
574+ {
575+ Ok(result) => result,
576+ Err(error) => Err(format!("Bundled process worker failed: {error}")),
577+ };
578+ remove_stream(streams.inner(), &stream, &stream_handle);
579+ result
580+}
581+
582+#[tauri::command]
583+async fn cancel<R: Runtime>(
584+ streams: State<'_, Streams>,
585+ webview: Webview<R>,
586+ stream: String,
587+) -> Result<(), String> {
588+ let stream_handle = owned_stream(streams.inner(), &webview, &stream)?;
589+ let result = match tauri::async_runtime::spawn_blocking({
590+ let stream_handle = Arc::clone(&stream_handle);
591+ move || stream_handle.cancel()
592+ })
593+ .await
594+ {
595+ Ok(result) => result,
596+ Err(error) => Err(format!("Bundled process worker failed: {error}")),
597+ };
598+ remove_stream(streams.inner(), &stream, &stream_handle);
599+ result
600+}
601+
207602pub fn init<R: Runtime>(executables: &[&str]) -> TauriPlugin<R> {
208603 let executables: Vec<String> = executables.iter().map(|name| (*name).to_owned()).collect();
209604 Builder::new("bundled-process")
210− .invoke_handler(tauri::generate_handler![run])
605+ .invoke_handler(tauri::generate_handler![run, spawn, write, finish, cancel])
211606 .setup(move |app, _| {
212607 for name in &executables {
213608 validate_basename(name)
@@ -227,14 +622,39 @@ pub fn init<R: Runtime>(executables: &[&str]) -> TauriPlugin<R> {
227622 directory,
228623 executables,
229624 });
625+ app.manage(Streams::default());
230626 Ok(())
231627 })
628+ .on_page_load(|webview, payload| {
629+ if matches!(payload.event(), PageLoadEvent::Started) {
630+ cleanup_owner(&webview.app_handle(), Some(webview.label()));
631+ }
632+ })
633+ .on_event(|app, event| match event {
634+ RunEvent::Exit => cleanup_owner(app, None),
635+ RunEvent::WindowEvent {
636+ label,
637+ event: WindowEvent::Destroyed,
638+ ..
639+ } => cleanup_owner(app, Some(label)),
640+ _ => {}
641+ })
642+ .on_drop(|app| cleanup_owner(&app, None))
232643 .build()
233644}
234645
235646#[cfg(test)]
236647mod tests {
237648 use super::*;
649+ use std::time::{Duration, Instant};
650+
651+ fn wait_until(label: &str, predicate: impl Fn() -> bool) {
652+ let deadline = Instant::now() + Duration::from_secs(5);
653+ while !predicate() {
654+ assert!(Instant::now() < deadline, "timed out waiting for {label}");
655+ thread::sleep(Duration::from_millis(1));
656+ }
657+ }
238658
239659 const FIXTURE_MODE: &str = "TAURI_BUNDLED_PROCESS_TEST_CHILD";
240660 const PIPE_BYTES: usize = 512 * 1024;
@@ -300,6 +720,41 @@ mod tests {
300720 assert_eq!(result.stderr, "input rejected");
301721 }
302722
723+ #[test]
724+ fn cancel_unblocks_a_blocked_stream_write() {
725+ let stream = spawn_stream(fixture("blocked-write"), "test".into()).unwrap();
726+ let writer_stream = Arc::clone(&stream);
727+ let writer = thread::spawn(move || writer_stream.write(&vec![b'i'; PIPE_BYTES * 16]));
728+
729+ wait_until("the stream write to block", || {
730+ stream.writing.load(Ordering::Acquire)
731+ });
732+ let cancelled_at = Instant::now();
733+ stream.cancel().unwrap();
734+
735+ assert!(writer.join().unwrap().is_err());
736+ assert!(cancelled_at.elapsed() < Duration::from_secs(5));
737+ }
738+
739+ #[test]
740+ fn cancel_kills_a_child_while_finish_waits() {
741+ let stream = spawn_stream(fixture("finish-wait"), "test".into()).unwrap();
742+ let finish_stream = Arc::clone(&stream);
743+ let finish = thread::spawn(move || finish_stream.finish());
744+
745+ wait_until("finish to begin waiting for the child", || {
746+ stream.waiting.load(Ordering::Acquire)
747+ });
748+ let cancelled_at = Instant::now();
749+ stream.cancel().unwrap();
750+ assert!(cancelled_at.elapsed() < Duration::from_secs(5));
751+
752+ match finish.join().unwrap() {
753+ Ok(result) => assert!(!result.success),
754+ Err(_) => {}
755+ }
756+ }
757+
303758 #[test]
304759 fn pipe_child_fixture() {
305760 let Ok(mode) = env::var(FIXTURE_MODE) else {
@@ -309,6 +764,16 @@ mod tests {
309764 io::stderr().write_all(b"input rejected").unwrap();
310765 std::process::exit(17);
311766 }
767+ if mode == "blocked-write" {
768+ thread::sleep(Duration::from_secs(30));
769+ return;
770+ }
771+ if mode == "finish-wait" {
772+ let mut input = Vec::new();
773+ io::stdin().read_to_end(&mut input).unwrap();
774+ thread::sleep(Duration::from_secs(30));
775+ return;
776+ }
312777 if mode == "large" {
313778 // Fill both outputs before reading stdin: sequential I/O deadlocks.
314779 io::stdout().write_all(&vec![b'o'; PIPE_BYTES]).unwrap();