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