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