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