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