1use crate::headless_project::HeadlessAppState;
2use crate::HeadlessProject;
3use anyhow::{anyhow, Context, Result};
4use client::ProxySettings;
5use fs::{Fs, RealFs};
6use futures::channel::mpsc;
7use futures::{select, select_biased, AsyncRead, AsyncWrite, AsyncWriteExt, FutureExt, SinkExt};
8use git::GitHostingProviderRegistry;
9use gpui::{AppContext, Context as _, ModelContext, UpdateGlobal as _};
10use http_client::{read_proxy_from_env, Uri};
11use language::LanguageRegistry;
12use node_runtime::{NodeBinaryOptions, NodeRuntime};
13use paths::logs_dir;
14use project::project_settings::ProjectSettings;
15use remote::proxy::ProxyLaunchError;
16use remote::ssh_session::ChannelClient;
17use remote::{
18 json_log::LogRecord,
19 protocol::{read_message, write_message},
20};
21use reqwest_client::ReqwestClient;
22use rpc::proto::{self, Envelope, SSH_PROJECT_ID};
23use settings::{watch_config_file, Settings, SettingsStore};
24use smol::channel::{Receiver, Sender};
25use smol::io::AsyncReadExt;
26
27use smol::Async;
28use smol::{net::unix::UnixListener, stream::StreamExt as _};
29use std::{
30 io::Write,
31 mem,
32 path::{Path, PathBuf},
33 sync::Arc,
34};
35use util::ResultExt;
36
37fn init_logging_proxy() {
38 env_logger::builder()
39 .format(|buf, record| {
40 let mut log_record = LogRecord::new(record);
41 log_record.message = format!("(remote proxy) {}", log_record.message);
42 serde_json::to_writer(&mut *buf, &log_record)?;
43 buf.write_all(b"\n")?;
44 Ok(())
45 })
46 .init();
47}
48
49fn init_logging_server(log_file_path: PathBuf) -> Result<Receiver<Vec<u8>>> {
50 struct MultiWrite {
51 file: Box<dyn std::io::Write + Send + 'static>,
52 channel: Sender<Vec<u8>>,
53 buffer: Vec<u8>,
54 }
55
56 impl std::io::Write for MultiWrite {
57 fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
58 let written = self.file.write(buf)?;
59 self.buffer.extend_from_slice(&buf[..written]);
60 Ok(written)
61 }
62
63 fn flush(&mut self) -> std::io::Result<()> {
64 self.channel
65 .send_blocking(self.buffer.clone())
66 .map_err(|error| std::io::Error::new(std::io::ErrorKind::Other, error))?;
67 self.buffer.clear();
68 self.file.flush()
69 }
70 }
71
72 let log_file = Box::new(if log_file_path.exists() {
73 std::fs::OpenOptions::new()
74 .append(true)
75 .open(&log_file_path)
76 .context("Failed to open log file in append mode")?
77 } else {
78 std::fs::File::create(&log_file_path).context("Failed to create log file")?
79 });
80
81 let (tx, rx) = smol::channel::unbounded();
82
83 let target = Box::new(MultiWrite {
84 file: log_file,
85 channel: tx,
86 buffer: Vec::new(),
87 });
88
89 env_logger::Builder::from_default_env()
90 .target(env_logger::Target::Pipe(target))
91 .format(|buf, record| {
92 let mut log_record = LogRecord::new(record);
93 log_record.message = format!("(remote server) {}", log_record.message);
94 serde_json::to_writer(&mut *buf, &log_record)?;
95 buf.write_all(b"\n")?;
96 Ok(())
97 })
98 .init();
99
100 Ok(rx)
101}
102
103fn init_panic_hook() {
104 std::panic::set_hook(Box::new(|info| {
105 let payload = info
106 .payload()
107 .downcast_ref::<&str>()
108 .map(|s| s.to_string())
109 .or_else(|| info.payload().downcast_ref::<String>().cloned())
110 .unwrap_or_else(|| "Box<Any>".to_string());
111
112 let backtrace = backtrace::Backtrace::new();
113 let mut backtrace = backtrace
114 .frames()
115 .iter()
116 .flat_map(|frame| {
117 frame
118 .symbols()
119 .iter()
120 .filter_map(|frame| Some(format!("{:#}", frame.name()?)))
121 })
122 .collect::<Vec<_>>();
123
124 // Strip out leading stack frames for rust panic-handling.
125 if let Some(ix) = backtrace
126 .iter()
127 .position(|name| name == "rust_begin_unwind")
128 {
129 backtrace.drain(0..=ix);
130 }
131
132 log::error!(
133 "panic occurred: {}\nBacktrace:\n{}",
134 payload,
135 backtrace.join("\n")
136 );
137
138 std::process::abort();
139 }));
140}
141
142struct ServerListeners {
143 stdin: UnixListener,
144 stdout: UnixListener,
145 stderr: UnixListener,
146}
147
148impl ServerListeners {
149 pub fn new(stdin_path: PathBuf, stdout_path: PathBuf, stderr_path: PathBuf) -> Result<Self> {
150 Ok(Self {
151 stdin: UnixListener::bind(stdin_path).context("failed to bind stdin socket")?,
152 stdout: UnixListener::bind(stdout_path).context("failed to bind stdout socket")?,
153 stderr: UnixListener::bind(stderr_path).context("failed to bind stderr socket")?,
154 })
155 }
156}
157
158fn start_server(
159 listeners: ServerListeners,
160 mut log_rx: Receiver<Vec<u8>>,
161 cx: &mut AppContext,
162) -> Arc<ChannelClient> {
163 // This is the server idle timeout. If no connection comes in in this timeout, the server will shut down.
164 const IDLE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10 * 60);
165
166 let (incoming_tx, incoming_rx) = mpsc::unbounded::<Envelope>();
167 let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded::<Envelope>();
168 let (app_quit_tx, mut app_quit_rx) = mpsc::unbounded::<()>();
169
170 cx.on_app_quit(move |_| {
171 let mut app_quit_tx = app_quit_tx.clone();
172 async move {
173 log::info!("app quitting. sending signal to server main loop");
174 app_quit_tx.send(()).await.ok();
175 }
176 })
177 .detach();
178
179 cx.spawn(|cx| async move {
180 let mut stdin_incoming = listeners.stdin.incoming();
181 let mut stdout_incoming = listeners.stdout.incoming();
182 let mut stderr_incoming = listeners.stderr.incoming();
183
184 loop {
185 let streams = futures::future::join3(stdin_incoming.next(), stdout_incoming.next(), stderr_incoming.next());
186
187 log::info!("accepting new connections");
188 let result = select! {
189 streams = streams.fuse() => {
190 let (Some(Ok(stdin_stream)), Some(Ok(stdout_stream)), Some(Ok(stderr_stream))) = streams else {
191 break;
192 };
193 anyhow::Ok((stdin_stream, stdout_stream, stderr_stream))
194 }
195 _ = futures::FutureExt::fuse(smol::Timer::after(IDLE_TIMEOUT)) => {
196 log::warn!("timed out waiting for new connections after {:?}. exiting.", IDLE_TIMEOUT);
197 cx.update(|cx| {
198 // TODO: This is a hack, because in a headless project, shutdown isn't executed
199 // when calling quit, but it should be.
200 cx.shutdown();
201 cx.quit();
202 })?;
203 break;
204 }
205 _ = app_quit_rx.next().fuse() => {
206 break;
207 }
208 };
209
210 let Ok((mut stdin_stream, mut stdout_stream, mut stderr_stream)) = result else {
211 break;
212 };
213
214 let mut input_buffer = Vec::new();
215 let mut output_buffer = Vec::new();
216 loop {
217 select_biased! {
218 _ = app_quit_rx.next().fuse() => {
219 return anyhow::Ok(());
220 }
221
222 stdin_message = read_message(&mut stdin_stream, &mut input_buffer).fuse() => {
223 let message = match stdin_message {
224 Ok(message) => message,
225 Err(error) => {
226 log::warn!("error reading message on stdin: {}. exiting.", error);
227 break;
228 }
229 };
230 if let Err(error) = incoming_tx.unbounded_send(message) {
231 log::error!("failed to send message to application: {:?}. exiting.", error);
232 return Err(anyhow!(error));
233 }
234 }
235
236 outgoing_message = outgoing_rx.next().fuse() => {
237 let Some(message) = outgoing_message else {
238 log::error!("stdout handler, no message");
239 break;
240 };
241
242 if let Err(error) =
243 write_message(&mut stdout_stream, &mut output_buffer, message).await
244 {
245 log::error!("failed to write stdout message: {:?}", error);
246 break;
247 }
248 if let Err(error) = stdout_stream.flush().await {
249 log::error!("failed to flush stdout message: {:?}", error);
250 break;
251 }
252 }
253
254 log_message = log_rx.next().fuse() => {
255 if let Some(log_message) = log_message {
256 if let Err(error) = stderr_stream.write_all(&log_message).await {
257 log::error!("failed to write log message to stderr: {:?}", error);
258 break;
259 }
260 if let Err(error) = stderr_stream.flush().await {
261 log::error!("failed to flush stderr stream: {:?}", error);
262 break;
263 }
264 }
265 }
266 }
267 }
268 }
269 anyhow::Ok(())
270 })
271 .detach();
272
273 ChannelClient::new(incoming_rx, outgoing_tx, cx)
274}
275
276fn init_paths() -> anyhow::Result<()> {
277 for path in [
278 paths::config_dir(),
279 paths::extensions_dir(),
280 paths::languages_dir(),
281 paths::logs_dir(),
282 paths::temp_dir(),
283 ]
284 .iter()
285 {
286 std::fs::create_dir_all(path)
287 .map_err(|e| anyhow!("Could not create directory {:?}: {}", path, e))?;
288 }
289 Ok(())
290}
291
292pub fn execute_run(
293 log_file: PathBuf,
294 pid_file: PathBuf,
295 stdin_socket: PathBuf,
296 stdout_socket: PathBuf,
297 stderr_socket: PathBuf,
298) -> Result<()> {
299 let log_rx = init_logging_server(log_file)?;
300 init_panic_hook();
301 init_paths()?;
302
303 log::info!(
304 "starting up. pid_file: {:?}, stdin_socket: {:?}, stdout_socket: {:?}, stderr_socket: {:?}",
305 pid_file,
306 stdin_socket,
307 stdout_socket,
308 stderr_socket
309 );
310
311 write_pid_file(&pid_file)
312 .with_context(|| format!("failed to write pid file: {:?}", &pid_file))?;
313
314 let listeners = ServerListeners::new(stdin_socket, stdout_socket, stderr_socket)?;
315
316 log::info!("starting headless gpui app");
317
318 let git_hosting_provider_registry = Arc::new(GitHostingProviderRegistry::new());
319 gpui::App::headless().run(move |cx| {
320 settings::init(cx);
321 HeadlessProject::init(cx);
322
323 log::info!("gpui app started, initializing server");
324 let session = start_server(listeners, log_rx, cx);
325
326 client::init_settings(cx);
327
328 GitHostingProviderRegistry::set_global(git_hosting_provider_registry, cx);
329 git_hosting_providers::init(cx);
330
331 let project = cx.new_model(|cx| {
332 let fs = Arc::new(RealFs::new(Default::default(), None));
333 let node_settings_rx = initialize_settings(session.clone(), fs.clone(), cx);
334
335 let proxy_url = read_proxy_settings(cx);
336
337 let http_client = Arc::new(
338 ReqwestClient::proxy_and_user_agent(
339 proxy_url,
340 &format!(
341 "Zed-Server/{} ({}; {})",
342 env!("CARGO_PKG_VERSION"),
343 std::env::consts::OS,
344 std::env::consts::ARCH
345 ),
346 )
347 .expect("Could not start HTTP client"),
348 );
349
350 let node_runtime = NodeRuntime::new(http_client.clone(), node_settings_rx);
351
352 let mut languages = LanguageRegistry::new(cx.background_executor().clone());
353 languages.set_language_server_download_dir(paths::languages_dir().clone());
354 let languages = Arc::new(languages);
355
356 HeadlessProject::new(
357 HeadlessAppState {
358 session,
359 fs,
360 http_client,
361 node_runtime,
362 languages,
363 },
364 cx,
365 )
366 });
367
368 mem::forget(project);
369 });
370 log::info!("gpui app is shut down. quitting.");
371 Ok(())
372}
373
374#[derive(Clone)]
375struct ServerPaths {
376 log_file: PathBuf,
377 pid_file: PathBuf,
378 stdin_socket: PathBuf,
379 stdout_socket: PathBuf,
380 stderr_socket: PathBuf,
381}
382
383impl ServerPaths {
384 fn new(identifier: &str) -> Result<Self> {
385 let server_dir = paths::remote_server_state_dir().join(identifier);
386 std::fs::create_dir_all(&server_dir)?;
387 std::fs::create_dir_all(&logs_dir())?;
388
389 let pid_file = server_dir.join("server.pid");
390 let stdin_socket = server_dir.join("stdin.sock");
391 let stdout_socket = server_dir.join("stdout.sock");
392 let stderr_socket = server_dir.join("stderr.sock");
393 let log_file = logs_dir().join(format!("server-{}.log", identifier));
394
395 Ok(Self {
396 pid_file,
397 stdin_socket,
398 stdout_socket,
399 stderr_socket,
400 log_file,
401 })
402 }
403}
404
405pub fn execute_proxy(identifier: String, is_reconnecting: bool) -> Result<()> {
406 init_logging_proxy();
407 init_panic_hook();
408
409 log::info!("starting proxy process. PID: {}", std::process::id());
410
411 let server_paths = ServerPaths::new(&identifier)?;
412
413 let server_pid = check_pid_file(&server_paths.pid_file)?;
414 let server_running = server_pid.is_some();
415 if is_reconnecting {
416 if !server_running {
417 log::error!("attempted to reconnect, but no server running");
418 return Err(anyhow!(ProxyLaunchError::ServerNotRunning));
419 }
420 } else {
421 if let Some(pid) = server_pid {
422 log::info!("proxy found server already running with PID {}. Killing process and cleaning up files...", pid);
423 kill_running_server(pid, &server_paths)?;
424 }
425
426 spawn_server(&server_paths)?;
427 };
428
429 let stdin_task = smol::spawn(async move {
430 let stdin = Async::new(std::io::stdin())?;
431 let stream = smol::net::unix::UnixStream::connect(&server_paths.stdin_socket).await?;
432 handle_io(stdin, stream, "stdin").await
433 });
434
435 let stdout_task: smol::Task<Result<()>> = smol::spawn(async move {
436 let stdout = Async::new(std::io::stdout())?;
437 let stream = smol::net::unix::UnixStream::connect(&server_paths.stdout_socket).await?;
438 handle_io(stream, stdout, "stdout").await
439 });
440
441 let stderr_task: smol::Task<Result<()>> = smol::spawn(async move {
442 let mut stderr = Async::new(std::io::stderr())?;
443 let mut stream = smol::net::unix::UnixStream::connect(&server_paths.stderr_socket).await?;
444 let mut stderr_buffer = vec![0; 2048];
445 loop {
446 match stream.read(&mut stderr_buffer).await {
447 Ok(0) => {
448 let error =
449 std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "stderr closed");
450 Err(anyhow!(error))?;
451 }
452 Ok(n) => {
453 stderr.write_all(&mut stderr_buffer[..n]).await?;
454 stderr.flush().await?;
455 }
456 Err(error) => {
457 Err(anyhow!("error reading stderr: {error:?}"))?;
458 }
459 }
460 }
461 });
462
463 if let Err(forwarding_result) = smol::block_on(async move {
464 futures::select! {
465 result = stdin_task.fuse() => result,
466 result = stdout_task.fuse() => result,
467 result = stderr_task.fuse() => result,
468 }
469 }) {
470 if let Some(error) = forwarding_result.downcast_ref::<std::io::Error>() {
471 if error.kind() == std::io::ErrorKind::UnexpectedEof {
472 log::error!("connection to server closed due to unexpected EOF");
473 return Err(anyhow!("connection to server closed"));
474 }
475 }
476 log::error!(
477 "failed to forward messages: {:?}, terminating...",
478 forwarding_result
479 );
480 return Err(forwarding_result);
481 }
482
483 Ok(())
484}
485
486fn kill_running_server(pid: u32, paths: &ServerPaths) -> Result<()> {
487 log::info!("killing existing server with PID {}", pid);
488 std::process::Command::new("kill")
489 .arg(pid.to_string())
490 .output()
491 .context("failed to kill existing server")?;
492
493 for file in [
494 &paths.pid_file,
495 &paths.stdin_socket,
496 &paths.stdout_socket,
497 &paths.stderr_socket,
498 ] {
499 log::debug!("cleaning up file {:?} before starting new server", file);
500 std::fs::remove_file(file).ok();
501 }
502 Ok(())
503}
504
505fn spawn_server(paths: &ServerPaths) -> Result<()> {
506 if paths.stdin_socket.exists() {
507 std::fs::remove_file(&paths.stdin_socket)?;
508 }
509 if paths.stdout_socket.exists() {
510 std::fs::remove_file(&paths.stdout_socket)?;
511 }
512 if paths.stderr_socket.exists() {
513 std::fs::remove_file(&paths.stderr_socket)?;
514 }
515
516 let binary_name = std::env::current_exe()?;
517 let server_process = smol::process::Command::new(binary_name)
518 .arg("run")
519 .arg("--log-file")
520 .arg(&paths.log_file)
521 .arg("--pid-file")
522 .arg(&paths.pid_file)
523 .arg("--stdin-socket")
524 .arg(&paths.stdin_socket)
525 .arg("--stdout-socket")
526 .arg(&paths.stdout_socket)
527 .arg("--stderr-socket")
528 .arg(&paths.stderr_socket)
529 .spawn()?;
530
531 log::info!(
532 "proxy spawned server process. PID: {:?}",
533 server_process.id()
534 );
535
536 let mut total_time_waited = std::time::Duration::from_secs(0);
537 let wait_duration = std::time::Duration::from_millis(20);
538 while !paths.stdout_socket.exists()
539 || !paths.stdin_socket.exists()
540 || !paths.stderr_socket.exists()
541 {
542 log::debug!("waiting for server to be ready to accept connections...");
543 std::thread::sleep(wait_duration);
544 total_time_waited += wait_duration;
545 }
546
547 log::info!(
548 "server ready to accept connections. total time waited: {:?}",
549 total_time_waited
550 );
551
552 Ok(())
553}
554
555fn check_pid_file(path: &Path) -> Result<Option<u32>> {
556 let Some(pid) = std::fs::read_to_string(&path)
557 .ok()
558 .and_then(|contents| contents.parse::<u32>().ok())
559 else {
560 return Ok(None);
561 };
562
563 log::debug!("Checking if process with PID {} exists...", pid);
564 match std::process::Command::new("kill")
565 .arg("-0")
566 .arg(pid.to_string())
567 .output()
568 {
569 Ok(output) if output.status.success() => {
570 log::debug!("Process with PID {} exists. NOT spawning new server, but attaching to existing one.", pid);
571 Ok(Some(pid))
572 }
573 _ => {
574 log::debug!(
575 "Found PID file, but process with that PID does not exist. Removing PID file."
576 );
577 std::fs::remove_file(&path).context("Failed to remove PID file")?;
578 Ok(None)
579 }
580 }
581}
582
583fn write_pid_file(path: &Path) -> Result<()> {
584 if path.exists() {
585 std::fs::remove_file(path)?;
586 }
587 let pid = std::process::id().to_string();
588 log::debug!("writing PID {} to file {:?}", pid, path);
589 std::fs::write(path, pid).context("Failed to write PID file")
590}
591
592async fn handle_io<R, W>(mut reader: R, mut writer: W, socket_name: &str) -> Result<()>
593where
594 R: AsyncRead + Unpin,
595 W: AsyncWrite + Unpin,
596{
597 use remote::protocol::read_message_raw;
598
599 let mut buffer = Vec::new();
600 loop {
601 read_message_raw(&mut reader, &mut buffer)
602 .await
603 .with_context(|| format!("failed to read message from {}", socket_name))?;
604
605 write_size_prefixed_buffer(&mut writer, &mut buffer)
606 .await
607 .with_context(|| format!("failed to write message to {}", socket_name))?;
608
609 writer.flush().await?;
610
611 buffer.clear();
612 }
613}
614
615async fn write_size_prefixed_buffer<S: AsyncWrite + Unpin>(
616 stream: &mut S,
617 buffer: &mut Vec<u8>,
618) -> Result<()> {
619 let len = buffer.len() as u32;
620 stream.write_all(len.to_le_bytes().as_slice()).await?;
621 stream.write_all(buffer).await?;
622 Ok(())
623}
624
625fn initialize_settings(
626 session: Arc<ChannelClient>,
627 fs: Arc<dyn Fs>,
628 cx: &mut AppContext,
629) -> async_watch::Receiver<Option<NodeBinaryOptions>> {
630 let user_settings_file_rx = watch_config_file(
631 &cx.background_executor(),
632 fs,
633 paths::settings_file().clone(),
634 );
635
636 handle_settings_file_changes(user_settings_file_rx, cx, {
637 let session = session.clone();
638 move |err, _cx| {
639 if let Some(e) = err {
640 log::info!("Server settings failed to change: {}", e);
641
642 session
643 .send(proto::Toast {
644 project_id: SSH_PROJECT_ID,
645 notification_id: "server-settings-failed".to_string(),
646 message: format!(
647 "Error in settings on remote host {:?}: {}",
648 paths::settings_file(),
649 e
650 ),
651 })
652 .log_err();
653 } else {
654 session
655 .send(proto::HideToast {
656 project_id: SSH_PROJECT_ID,
657 notification_id: "server-settings-failed".to_string(),
658 })
659 .log_err();
660 }
661 }
662 });
663
664 let (tx, rx) = async_watch::channel(None);
665 cx.observe_global::<SettingsStore>(move |cx| {
666 let settings = &ProjectSettings::get_global(cx).node;
667 log::info!("Got new node settings: {:?}", settings);
668 let options = NodeBinaryOptions {
669 allow_path_lookup: !settings.ignore_system_version.unwrap_or_default(),
670 // TODO: Implement this setting
671 allow_binary_download: true,
672 use_paths: settings.path.as_ref().map(|node_path| {
673 let node_path = PathBuf::from(shellexpand::tilde(node_path).as_ref());
674 let npm_path = settings
675 .npm_path
676 .as_ref()
677 .map(|path| PathBuf::from(shellexpand::tilde(&path).as_ref()));
678 (
679 node_path.clone(),
680 npm_path.unwrap_or_else(|| {
681 let base_path = PathBuf::new();
682 node_path.parent().unwrap_or(&base_path).join("npm")
683 }),
684 )
685 }),
686 };
687 tx.send(Some(options)).log_err();
688 })
689 .detach();
690
691 rx
692}
693
694pub fn handle_settings_file_changes(
695 mut server_settings_file: mpsc::UnboundedReceiver<String>,
696 cx: &mut AppContext,
697 settings_changed: impl Fn(Option<anyhow::Error>, &mut AppContext) + 'static,
698) {
699 let server_settings_content = cx
700 .background_executor()
701 .block(server_settings_file.next())
702 .unwrap();
703 SettingsStore::update_global(cx, |store, cx| {
704 store
705 .set_server_settings(&server_settings_content, cx)
706 .log_err();
707 });
708 cx.spawn(move |cx| async move {
709 while let Some(server_settings_content) = server_settings_file.next().await {
710 let result = cx.update_global(|store: &mut SettingsStore, cx| {
711 let result = store.set_server_settings(&server_settings_content, cx);
712 if let Err(err) = &result {
713 log::error!("Failed to load server settings: {err}");
714 }
715 settings_changed(result.err(), cx);
716 cx.refresh();
717 });
718 if result.is_err() {
719 break; // App dropped
720 }
721 }
722 })
723 .detach();
724}
725
726fn read_proxy_settings(cx: &mut ModelContext<'_, HeadlessProject>) -> Option<Uri> {
727 let proxy_str = ProxySettings::get_global(cx).proxy.to_owned();
728 let proxy_url = proxy_str
729 .as_ref()
730 .and_then(|input: &String| {
731 input
732 .parse::<Uri>()
733 .inspect_err(|e| log::error!("Error parsing proxy settings: {}", e))
734 .ok()
735 })
736 .or_else(read_proxy_from_env);
737 proxy_url
738}