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