1use crate::headless_project::HeadlessAppState;
2use crate::HeadlessProject;
3use anyhow::{anyhow, Context, Result};
4use client::ProxySettings;
5use fs::{Fs, RealFs};
6use futures::channel::mpsc;
7use futures::{select, select_biased, AsyncRead, AsyncWrite, AsyncWriteExt, FutureExt, SinkExt};
8use git::GitHostingProviderRegistry;
9use gpui::{AppContext, Context as _, ModelContext, UpdateGlobal as _};
10use http_client::{read_proxy_from_env, Uri};
11use language::LanguageRegistry;
12use node_runtime::{NodeBinaryOptions, NodeRuntime};
13use paths::logs_dir;
14use project::project_settings::ProjectSettings;
15
16use remote::proxy::ProxyLaunchError;
17use remote::ssh_session::ChannelClient;
18use remote::{
19 json_log::LogRecord,
20 protocol::{read_message, write_message},
21};
22use reqwest_client::ReqwestClient;
23use rpc::proto::{self, Envelope, SSH_PROJECT_ID};
24use settings::{watch_config_file, Settings, SettingsStore};
25use smol::channel::{Receiver, Sender};
26use smol::io::AsyncReadExt;
27
28use smol::Async;
29use smol::{net::unix::UnixListener, stream::StreamExt as _};
30use std::ops::ControlFlow;
31use std::{
32 io::Write,
33 mem,
34 path::{Path, PathBuf},
35 sync::Arc,
36};
37use util::ResultExt;
38
39fn init_logging_proxy() {
40 env_logger::builder()
41 .format(|buf, record| {
42 let mut log_record = LogRecord::new(record);
43 log_record.message = format!("(remote proxy) {}", log_record.message);
44 serde_json::to_writer(&mut *buf, &log_record)?;
45 buf.write_all(b"\n")?;
46 Ok(())
47 })
48 .init();
49}
50
51fn init_logging_server(log_file_path: PathBuf) -> Result<Receiver<Vec<u8>>> {
52 struct MultiWrite {
53 file: Box<dyn std::io::Write + Send + 'static>,
54 channel: Sender<Vec<u8>>,
55 buffer: Vec<u8>,
56 }
57
58 impl std::io::Write for MultiWrite {
59 fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
60 let written = self.file.write(buf)?;
61 self.buffer.extend_from_slice(&buf[..written]);
62 Ok(written)
63 }
64
65 fn flush(&mut self) -> std::io::Result<()> {
66 self.channel
67 .send_blocking(self.buffer.clone())
68 .map_err(|error| std::io::Error::new(std::io::ErrorKind::Other, error))?;
69 self.buffer.clear();
70 self.file.flush()
71 }
72 }
73
74 let log_file = Box::new(if log_file_path.exists() {
75 std::fs::OpenOptions::new()
76 .append(true)
77 .open(&log_file_path)
78 .context("Failed to open log file in append mode")?
79 } else {
80 std::fs::File::create(&log_file_path).context("Failed to create log file")?
81 });
82
83 let (tx, rx) = smol::channel::unbounded();
84
85 let target = Box::new(MultiWrite {
86 file: log_file,
87 channel: tx,
88 buffer: Vec::new(),
89 });
90
91 env_logger::Builder::from_default_env()
92 .target(env_logger::Target::Pipe(target))
93 .format(|buf, record| {
94 let mut log_record = LogRecord::new(record);
95 log_record.message = format!("(remote server) {}", log_record.message);
96 serde_json::to_writer(&mut *buf, &log_record)?;
97 buf.write_all(b"\n")?;
98 Ok(())
99 })
100 .init();
101
102 Ok(rx)
103}
104
105fn init_panic_hook() {
106 std::panic::set_hook(Box::new(|info| {
107 let payload = info
108 .payload()
109 .downcast_ref::<&str>()
110 .map(|s| s.to_string())
111 .or_else(|| info.payload().downcast_ref::<String>().cloned())
112 .unwrap_or_else(|| "Box<Any>".to_string());
113
114 let backtrace = backtrace::Backtrace::new();
115 let mut backtrace = backtrace
116 .frames()
117 .iter()
118 .flat_map(|frame| {
119 frame
120 .symbols()
121 .iter()
122 .filter_map(|frame| Some(format!("{:#}", frame.name()?)))
123 })
124 .collect::<Vec<_>>();
125
126 // Strip out leading stack frames for rust panic-handling.
127 if let Some(ix) = backtrace
128 .iter()
129 .position(|name| name == "rust_begin_unwind")
130 {
131 backtrace.drain(0..=ix);
132 }
133
134 log::error!(
135 "panic occurred: {}\nBacktrace:\n{}",
136 payload,
137 backtrace.join("\n")
138 );
139
140 std::process::abort();
141 }));
142}
143
144struct ServerListeners {
145 stdin: UnixListener,
146 stdout: UnixListener,
147 stderr: UnixListener,
148}
149
150impl ServerListeners {
151 pub fn new(stdin_path: PathBuf, stdout_path: PathBuf, stderr_path: PathBuf) -> Result<Self> {
152 Ok(Self {
153 stdin: UnixListener::bind(stdin_path).context("failed to bind stdin socket")?,
154 stdout: UnixListener::bind(stdout_path).context("failed to bind stdout socket")?,
155 stderr: UnixListener::bind(stderr_path).context("failed to bind stderr socket")?,
156 })
157 }
158}
159
160fn start_server(
161 listeners: ServerListeners,
162 mut log_rx: Receiver<Vec<u8>>,
163 cx: &mut AppContext,
164) -> Arc<ChannelClient> {
165 // This is the server idle timeout. If no connection comes in in this timeout, the server will shut down.
166 const IDLE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10 * 60);
167
168 let (incoming_tx, incoming_rx) = mpsc::unbounded::<Envelope>();
169 let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded::<Envelope>();
170 let (app_quit_tx, mut app_quit_rx) = mpsc::unbounded::<()>();
171
172 cx.on_app_quit(move |_| {
173 let mut app_quit_tx = app_quit_tx.clone();
174 async move {
175 log::info!("app quitting. sending signal to server main loop");
176 app_quit_tx.send(()).await.ok();
177 }
178 })
179 .detach();
180
181 cx.spawn(|cx| async move {
182 let mut stdin_incoming = listeners.stdin.incoming();
183 let mut stdout_incoming = listeners.stdout.incoming();
184 let mut stderr_incoming = listeners.stderr.incoming();
185
186 loop {
187 let streams = futures::future::join3(stdin_incoming.next(), stdout_incoming.next(), stderr_incoming.next());
188
189 log::info!("accepting new connections");
190 let result = select! {
191 streams = streams.fuse() => {
192 let (Some(Ok(stdin_stream)), Some(Ok(stdout_stream)), Some(Ok(stderr_stream))) = streams else {
193 break;
194 };
195 anyhow::Ok((stdin_stream, stdout_stream, stderr_stream))
196 }
197 _ = futures::FutureExt::fuse(smol::Timer::after(IDLE_TIMEOUT)) => {
198 log::warn!("timed out waiting for new connections after {:?}. exiting.", IDLE_TIMEOUT);
199 cx.update(|cx| {
200 // TODO: This is a hack, because in a headless project, shutdown isn't executed
201 // when calling quit, but it should be.
202 cx.shutdown();
203 cx.quit();
204 })?;
205 break;
206 }
207 _ = app_quit_rx.next().fuse() => {
208 break;
209 }
210 };
211
212 let Ok((mut stdin_stream, mut stdout_stream, mut stderr_stream)) = result else {
213 break;
214 };
215
216 let mut input_buffer = Vec::new();
217 let mut output_buffer = Vec::new();
218
219 let (mut stdin_msg_tx, mut stdin_msg_rx) = mpsc::unbounded::<Envelope>();
220 cx.background_executor().spawn(async move {
221 while let Ok(msg) = read_message(&mut stdin_stream, &mut input_buffer).await {
222 if let Err(_) = stdin_msg_tx.send(msg).await {
223 break;
224 }
225 }
226 }).detach();
227
228 loop {
229
230 select_biased! {
231 _ = app_quit_rx.next().fuse() => {
232 return anyhow::Ok(());
233 }
234
235 stdin_message = stdin_msg_rx.next().fuse() => {
236 let Some(message) = stdin_message else {
237 log::warn!("error reading message on stdin. exiting.");
238 break;
239 };
240 if let Err(error) = incoming_tx.unbounded_send(message) {
241 log::error!("failed to send message to application: {:?}. exiting.", error);
242 return Err(anyhow!(error));
243 }
244 }
245
246 outgoing_message = outgoing_rx.next().fuse() => {
247 let Some(message) = outgoing_message else {
248 log::error!("stdout handler, no message");
249 break;
250 };
251
252 if let Err(error) =
253 write_message(&mut stdout_stream, &mut output_buffer, message).await
254 {
255 log::error!("failed to write stdout message: {:?}", error);
256 break;
257 }
258 if let Err(error) = stdout_stream.flush().await {
259 log::error!("failed to flush stdout message: {:?}", error);
260 break;
261 }
262 }
263
264 log_message = log_rx.next().fuse() => {
265 if let Some(log_message) = log_message {
266 if let Err(error) = stderr_stream.write_all(&log_message).await {
267 log::error!("failed to write log message to stderr: {:?}", error);
268 break;
269 }
270 if let Err(error) = stderr_stream.flush().await {
271 log::error!("failed to flush stderr stream: {:?}", error);
272 break;
273 }
274 }
275 }
276 }
277 }
278 }
279 anyhow::Ok(())
280 })
281 .detach();
282
283 ChannelClient::new(incoming_rx, outgoing_tx, cx, "server")
284}
285
286fn init_paths() -> anyhow::Result<()> {
287 for path in [
288 paths::config_dir(),
289 paths::extensions_dir(),
290 paths::languages_dir(),
291 paths::logs_dir(),
292 paths::temp_dir(),
293 ]
294 .iter()
295 {
296 std::fs::create_dir_all(path)
297 .map_err(|e| anyhow!("Could not create directory {:?}: {}", path, e))?;
298 }
299 Ok(())
300}
301
302pub fn execute_run(
303 log_file: PathBuf,
304 pid_file: PathBuf,
305 stdin_socket: PathBuf,
306 stdout_socket: PathBuf,
307 stderr_socket: PathBuf,
308) -> Result<()> {
309 init_paths()?;
310
311 match daemonize()? {
312 ControlFlow::Break(_) => return Ok(()),
313 ControlFlow::Continue(_) => {}
314 }
315
316 init_panic_hook();
317 let log_rx = init_logging_server(log_file)?;
318 log::info!(
319 "starting up. pid_file: {:?}, stdin_socket: {:?}, stdout_socket: {:?}, stderr_socket: {:?}",
320 pid_file,
321 stdin_socket,
322 stdout_socket,
323 stderr_socket
324 );
325
326 write_pid_file(&pid_file)
327 .with_context(|| format!("failed to write pid file: {:?}", &pid_file))?;
328
329 let listeners = ServerListeners::new(stdin_socket, stdout_socket, stderr_socket)?;
330
331 let git_hosting_provider_registry = Arc::new(GitHostingProviderRegistry::new());
332 gpui::App::headless().run(move |cx| {
333 settings::init(cx);
334 HeadlessProject::init(cx);
335
336 log::info!("gpui app started, initializing server");
337 let session = start_server(listeners, log_rx, cx);
338
339 client::init_settings(cx);
340
341 GitHostingProviderRegistry::set_global(git_hosting_provider_registry, cx);
342 git_hosting_providers::init(cx);
343
344 let project = cx.new_model(|cx| {
345 let fs = Arc::new(RealFs::new(Default::default(), None));
346 let node_settings_rx = initialize_settings(session.clone(), fs.clone(), cx);
347
348 let proxy_url = read_proxy_settings(cx);
349
350 let http_client = Arc::new(
351 ReqwestClient::proxy_and_user_agent(
352 proxy_url,
353 &format!(
354 "Zed-Server/{} ({}; {})",
355 env!("CARGO_PKG_VERSION"),
356 std::env::consts::OS,
357 std::env::consts::ARCH
358 ),
359 )
360 .expect("Could not start HTTP client"),
361 );
362
363 let node_runtime = NodeRuntime::new(http_client.clone(), node_settings_rx);
364
365 let mut languages = LanguageRegistry::new(cx.background_executor().clone());
366 languages.set_language_server_download_dir(paths::languages_dir().clone());
367 let languages = Arc::new(languages);
368
369 HeadlessProject::new(
370 HeadlessAppState {
371 session,
372 fs,
373 http_client,
374 node_runtime,
375 languages,
376 },
377 cx,
378 )
379 });
380
381 mem::forget(project);
382 });
383 log::info!("gpui app is shut down. quitting.");
384 Ok(())
385}
386
387#[derive(Clone)]
388struct ServerPaths {
389 log_file: PathBuf,
390 pid_file: PathBuf,
391 stdin_socket: PathBuf,
392 stdout_socket: PathBuf,
393 stderr_socket: PathBuf,
394}
395
396impl ServerPaths {
397 fn new(identifier: &str) -> Result<Self> {
398 let server_dir = paths::remote_server_state_dir().join(identifier);
399 std::fs::create_dir_all(&server_dir)?;
400 std::fs::create_dir_all(&logs_dir())?;
401
402 let pid_file = server_dir.join("server.pid");
403 let stdin_socket = server_dir.join("stdin.sock");
404 let stdout_socket = server_dir.join("stdout.sock");
405 let stderr_socket = server_dir.join("stderr.sock");
406 let log_file = logs_dir().join(format!("server-{}.log", identifier));
407
408 Ok(Self {
409 pid_file,
410 stdin_socket,
411 stdout_socket,
412 stderr_socket,
413 log_file,
414 })
415 }
416}
417
418pub fn execute_proxy(identifier: String, is_reconnecting: bool) -> Result<()> {
419 init_logging_proxy();
420 init_panic_hook();
421
422 log::info!("starting proxy process. PID: {}", std::process::id());
423
424 let server_paths = ServerPaths::new(&identifier)?;
425
426 let server_pid = check_pid_file(&server_paths.pid_file)?;
427 let server_running = server_pid.is_some();
428 if is_reconnecting {
429 if !server_running {
430 log::error!("attempted to reconnect, but no server running");
431 return Err(anyhow!(ProxyLaunchError::ServerNotRunning));
432 }
433 } else {
434 if let Some(pid) = server_pid {
435 log::info!("proxy found server already running with PID {}. Killing process and cleaning up files...", pid);
436 kill_running_server(pid, &server_paths)?;
437 }
438
439 spawn_server(&server_paths)?;
440 };
441
442 let stdin_task = smol::spawn(async move {
443 let stdin = Async::new(std::io::stdin())?;
444 let stream = smol::net::unix::UnixStream::connect(&server_paths.stdin_socket).await?;
445 handle_io(stdin, stream, "stdin").await
446 });
447
448 let stdout_task: smol::Task<Result<()>> = smol::spawn(async move {
449 let stdout = Async::new(std::io::stdout())?;
450 let stream = smol::net::unix::UnixStream::connect(&server_paths.stdout_socket).await?;
451 handle_io(stream, stdout, "stdout").await
452 });
453
454 let stderr_task: smol::Task<Result<()>> = smol::spawn(async move {
455 let mut stderr = Async::new(std::io::stderr())?;
456 let mut stream = smol::net::unix::UnixStream::connect(&server_paths.stderr_socket).await?;
457 let mut stderr_buffer = vec![0; 2048];
458 loop {
459 match stream.read(&mut stderr_buffer).await {
460 Ok(0) => {
461 let error =
462 std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "stderr closed");
463 Err(anyhow!(error))?;
464 }
465 Ok(n) => {
466 stderr.write_all(&mut stderr_buffer[..n]).await?;
467 stderr.flush().await?;
468 }
469 Err(error) => {
470 Err(anyhow!("error reading stderr: {error:?}"))?;
471 }
472 }
473 }
474 });
475
476 if let Err(forwarding_result) = smol::block_on(async move {
477 futures::select! {
478 result = stdin_task.fuse() => result.context("stdin_task failed"),
479 result = stdout_task.fuse() => result.context("stdout_task failed"),
480 result = stderr_task.fuse() => result.context("stderr_task failed"),
481 }
482 }) {
483 log::error!(
484 "encountered error while forwarding messages: {:?}, terminating...",
485 forwarding_result
486 );
487 return Err(forwarding_result);
488 }
489
490 Ok(())
491}
492
493fn kill_running_server(pid: u32, paths: &ServerPaths) -> Result<()> {
494 log::info!("killing existing server with PID {}", pid);
495 std::process::Command::new("kill")
496 .arg(pid.to_string())
497 .output()
498 .context("failed to kill existing server")?;
499
500 for file in [
501 &paths.pid_file,
502 &paths.stdin_socket,
503 &paths.stdout_socket,
504 &paths.stderr_socket,
505 ] {
506 log::debug!("cleaning up file {:?} before starting new server", file);
507 std::fs::remove_file(file).ok();
508 }
509 Ok(())
510}
511
512fn spawn_server(paths: &ServerPaths) -> Result<()> {
513 if paths.stdin_socket.exists() {
514 std::fs::remove_file(&paths.stdin_socket)?;
515 }
516 if paths.stdout_socket.exists() {
517 std::fs::remove_file(&paths.stdout_socket)?;
518 }
519 if paths.stderr_socket.exists() {
520 std::fs::remove_file(&paths.stderr_socket)?;
521 }
522
523 let binary_name = std::env::current_exe()?;
524 let mut server_process = std::process::Command::new(binary_name);
525 server_process
526 .arg("run")
527 .arg("--log-file")
528 .arg(&paths.log_file)
529 .arg("--pid-file")
530 .arg(&paths.pid_file)
531 .arg("--stdin-socket")
532 .arg(&paths.stdin_socket)
533 .arg("--stdout-socket")
534 .arg(&paths.stdout_socket)
535 .arg("--stderr-socket")
536 .arg(&paths.stderr_socket);
537
538 let status = server_process
539 .status()
540 .context("failed to launch server process")?;
541 anyhow::ensure!(
542 status.success(),
543 "failed to launch and detach server process"
544 );
545
546 let mut total_time_waited = std::time::Duration::from_secs(0);
547 let wait_duration = std::time::Duration::from_millis(20);
548 while !paths.stdout_socket.exists()
549 || !paths.stdin_socket.exists()
550 || !paths.stderr_socket.exists()
551 {
552 log::debug!("waiting for server to be ready to accept connections...");
553 std::thread::sleep(wait_duration);
554 total_time_waited += wait_duration;
555 }
556
557 log::info!(
558 "server ready to accept connections. total time waited: {:?}",
559 total_time_waited
560 );
561
562 Ok(())
563}
564
565fn check_pid_file(path: &Path) -> Result<Option<u32>> {
566 let Some(pid) = std::fs::read_to_string(&path)
567 .ok()
568 .and_then(|contents| contents.parse::<u32>().ok())
569 else {
570 return Ok(None);
571 };
572
573 log::debug!("Checking if process with PID {} exists...", pid);
574 match std::process::Command::new("kill")
575 .arg("-0")
576 .arg(pid.to_string())
577 .output()
578 {
579 Ok(output) if output.status.success() => {
580 log::debug!("Process with PID {} exists. NOT spawning new server, but attaching to existing one.", pid);
581 Ok(Some(pid))
582 }
583 _ => {
584 log::debug!(
585 "Found PID file, but process with that PID does not exist. Removing PID file."
586 );
587 std::fs::remove_file(&path).context("Failed to remove PID file")?;
588 Ok(None)
589 }
590 }
591}
592
593fn write_pid_file(path: &Path) -> Result<()> {
594 if path.exists() {
595 std::fs::remove_file(path)?;
596 }
597 let pid = std::process::id().to_string();
598 log::debug!("writing PID {} to file {:?}", pid, path);
599 std::fs::write(path, pid).context("Failed to write PID file")
600}
601
602async fn handle_io<R, W>(mut reader: R, mut writer: W, socket_name: &str) -> Result<()>
603where
604 R: AsyncRead + Unpin,
605 W: AsyncWrite + Unpin,
606{
607 use remote::protocol::read_message_raw;
608
609 let mut buffer = Vec::new();
610 loop {
611 read_message_raw(&mut reader, &mut buffer)
612 .await
613 .with_context(|| format!("failed to read message from {}", socket_name))?;
614
615 write_size_prefixed_buffer(&mut writer, &mut buffer)
616 .await
617 .with_context(|| format!("failed to write message to {}", socket_name))?;
618
619 writer.flush().await?;
620
621 buffer.clear();
622 }
623}
624
625async fn write_size_prefixed_buffer<S: AsyncWrite + Unpin>(
626 stream: &mut S,
627 buffer: &mut Vec<u8>,
628) -> Result<()> {
629 let len = buffer.len() as u32;
630 stream.write_all(len.to_le_bytes().as_slice()).await?;
631 stream.write_all(buffer).await?;
632 Ok(())
633}
634
635fn initialize_settings(
636 session: Arc<ChannelClient>,
637 fs: Arc<dyn Fs>,
638 cx: &mut AppContext,
639) -> async_watch::Receiver<Option<NodeBinaryOptions>> {
640 let user_settings_file_rx = watch_config_file(
641 &cx.background_executor(),
642 fs,
643 paths::settings_file().clone(),
644 );
645
646 handle_settings_file_changes(user_settings_file_rx, cx, {
647 let session = session.clone();
648 move |err, _cx| {
649 if let Some(e) = err {
650 log::info!("Server settings failed to change: {}", e);
651
652 session
653 .send(proto::Toast {
654 project_id: SSH_PROJECT_ID,
655 notification_id: "server-settings-failed".to_string(),
656 message: format!(
657 "Error in settings on remote host {:?}: {}",
658 paths::settings_file(),
659 e
660 ),
661 })
662 .log_err();
663 } else {
664 session
665 .send(proto::HideToast {
666 project_id: SSH_PROJECT_ID,
667 notification_id: "server-settings-failed".to_string(),
668 })
669 .log_err();
670 }
671 }
672 });
673
674 let (tx, rx) = async_watch::channel(None);
675 cx.observe_global::<SettingsStore>(move |cx| {
676 let settings = &ProjectSettings::get_global(cx).node;
677 log::info!("Got new node settings: {:?}", settings);
678 let options = NodeBinaryOptions {
679 allow_path_lookup: !settings.ignore_system_version.unwrap_or_default(),
680 // TODO: Implement this setting
681 allow_binary_download: true,
682 use_paths: settings.path.as_ref().map(|node_path| {
683 let node_path = PathBuf::from(shellexpand::tilde(node_path).as_ref());
684 let npm_path = settings
685 .npm_path
686 .as_ref()
687 .map(|path| PathBuf::from(shellexpand::tilde(&path).as_ref()));
688 (
689 node_path.clone(),
690 npm_path.unwrap_or_else(|| {
691 let base_path = PathBuf::new();
692 node_path.parent().unwrap_or(&base_path).join("npm")
693 }),
694 )
695 }),
696 };
697 tx.send(Some(options)).log_err();
698 })
699 .detach();
700
701 rx
702}
703
704pub fn handle_settings_file_changes(
705 mut server_settings_file: mpsc::UnboundedReceiver<String>,
706 cx: &mut AppContext,
707 settings_changed: impl Fn(Option<anyhow::Error>, &mut AppContext) + 'static,
708) {
709 let server_settings_content = cx
710 .background_executor()
711 .block(server_settings_file.next())
712 .unwrap();
713 SettingsStore::update_global(cx, |store, cx| {
714 store
715 .set_server_settings(&server_settings_content, cx)
716 .log_err();
717 });
718 cx.spawn(move |cx| async move {
719 while let Some(server_settings_content) = server_settings_file.next().await {
720 let result = cx.update_global(|store: &mut SettingsStore, cx| {
721 let result = store.set_server_settings(&server_settings_content, cx);
722 if let Err(err) = &result {
723 log::error!("Failed to load server settings: {err}");
724 }
725 settings_changed(result.err(), cx);
726 cx.refresh();
727 });
728 if result.is_err() {
729 break; // App dropped
730 }
731 }
732 })
733 .detach();
734}
735
736fn read_proxy_settings(cx: &mut ModelContext<'_, HeadlessProject>) -> Option<Uri> {
737 let proxy_str = ProxySettings::get_global(cx).proxy.to_owned();
738 let proxy_url = proxy_str
739 .as_ref()
740 .and_then(|input: &String| {
741 input
742 .parse::<Uri>()
743 .inspect_err(|e| log::error!("Error parsing proxy settings: {}", e))
744 .ok()
745 })
746 .or_else(read_proxy_from_env);
747 proxy_url
748}
749
750fn daemonize() -> Result<ControlFlow<()>> {
751 match fork::fork().map_err(|e| anyhow::anyhow!("failed to call fork with error code {}", e))? {
752 fork::Fork::Parent(_) => {
753 return Ok(ControlFlow::Break(()));
754 }
755 fork::Fork::Child => {}
756 }
757
758 // Once we've detached from the parent, we want to close stdout/stderr/stdin
759 // so that the outer SSH process is not attached to us in any way anymore.
760 unsafe { redirect_standard_streams() }?;
761
762 Ok(ControlFlow::Continue(()))
763}
764
765unsafe fn redirect_standard_streams() -> Result<()> {
766 let devnull_fd = libc::open(b"/dev/null\0" as *const [u8; 10] as _, libc::O_RDWR);
767 anyhow::ensure!(devnull_fd != -1, "failed to open /dev/null");
768
769 let process_stdio = |name, fd| {
770 let reopened_fd = libc::dup2(devnull_fd, fd);
771 anyhow::ensure!(
772 reopened_fd != -1,
773 format!("failed to redirect {} to /dev/null", name)
774 );
775 Ok(())
776 };
777
778 process_stdio("stdin", libc::STDIN_FILENO)?;
779 process_stdio("stdout", libc::STDOUT_FILENO)?;
780 process_stdio("stderr", libc::STDERR_FILENO)?;
781
782 anyhow::ensure!(
783 libc::close(devnull_fd) != -1,
784 "failed to close /dev/null fd after redirecting"
785 );
786
787 Ok(())
788}