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