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