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