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