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