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