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