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