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                    let (Some(Ok(stdin_stream)), Some(Ok(stdout_stream)), Some(Ok(stderr_stream))) = streams else {
190                        break;
191                    };
192                    anyhow::Ok((stdin_stream, stdout_stream, stderr_stream))
193                }
194                _ = futures::FutureExt::fuse(smol::Timer::after(IDLE_TIMEOUT)) => {
195                    log::warn!("timed out waiting for new connections after {:?}. exiting.", IDLE_TIMEOUT);
196                    cx.update(|cx| {
197                        // TODO: This is a hack, because in a headless project, shutdown isn't executed
198                        // when calling quit, but it should be.
199                        cx.shutdown();
200                        cx.quit();
201                    })?;
202                    break;
203                }
204                _ = app_quit_rx.next().fuse() => {
205                    break;
206                }
207            };
208
209            let Ok((mut stdin_stream, mut stdout_stream, mut stderr_stream)) = result else {
210                break;
211            };
212
213            let mut input_buffer = Vec::new();
214            let mut output_buffer = Vec::new();
215            loop {
216                select_biased! {
217                    _ = app_quit_rx.next().fuse() => {
218                        return anyhow::Ok(());
219                    }
220
221                    stdin_message = read_message(&mut stdin_stream, &mut input_buffer).fuse() => {
222                        let message = match stdin_message {
223                            Ok(message) => message,
224                            Err(error) => {
225                                log::warn!("error reading message on stdin: {}. exiting.", error);
226                                break;
227                            }
228                        };
229                        if let Err(error) = incoming_tx.unbounded_send(message) {
230                            log::error!("failed to send message to application: {:?}. exiting.", error);
231                            return Err(anyhow!(error));
232                        }
233                    }
234
235                    outgoing_message  = outgoing_rx.next().fuse() => {
236                        let Some(message) = outgoing_message else {
237                            log::error!("stdout handler, no message");
238                            break;
239                        };
240
241                        if let Err(error) =
242                            write_message(&mut stdout_stream, &mut output_buffer, message).await
243                        {
244                            log::error!("failed to write stdout message: {:?}", error);
245                            break;
246                        }
247                        if let Err(error) = stdout_stream.flush().await {
248                            log::error!("failed to flush stdout message: {:?}", error);
249                            break;
250                        }
251                    }
252
253                    log_message = log_rx.next().fuse() => {
254                        if let Some(log_message) = log_message {
255                            if let Err(error) = stderr_stream.write_all(&log_message).await {
256                                log::error!("failed to write log message to stderr: {:?}", error);
257                                break;
258                            }
259                            if let Err(error) = stderr_stream.flush().await {
260                                log::error!("failed to flush stderr stream: {:?}", error);
261                                break;
262                            }
263                        }
264                    }
265                }
266            }
267        }
268        anyhow::Ok(())
269    })
270    .detach();
271
272    ChannelClient::new(incoming_rx, outgoing_tx, cx)
273}
274
275fn init_paths() -> anyhow::Result<()> {
276    for path in [
277        paths::config_dir(),
278        paths::extensions_dir(),
279        paths::languages_dir(),
280        paths::logs_dir(),
281        paths::temp_dir(),
282    ]
283    .iter()
284    {
285        std::fs::create_dir_all(path)
286            .map_err(|e| anyhow!("Could not create directory {:?}: {}", path, e))?;
287    }
288    Ok(())
289}
290
291pub fn execute_run(
292    log_file: PathBuf,
293    pid_file: PathBuf,
294    stdin_socket: PathBuf,
295    stdout_socket: PathBuf,
296    stderr_socket: PathBuf,
297) -> Result<()> {
298    let log_rx = init_logging_server(log_file)?;
299    init_panic_hook();
300    init_paths()?;
301
302    log::info!(
303        "starting up. pid_file: {:?}, stdin_socket: {:?}, stdout_socket: {:?}, stderr_socket: {:?}",
304        pid_file,
305        stdin_socket,
306        stdout_socket,
307        stderr_socket
308    );
309
310    write_pid_file(&pid_file)
311        .with_context(|| format!("failed to write pid file: {:?}", &pid_file))?;
312
313    let listeners = ServerListeners::new(stdin_socket, stdout_socket, stderr_socket)?;
314
315    log::info!("starting headless gpui app");
316    gpui::App::headless().run(move |cx| {
317        settings::init(cx);
318        HeadlessProject::init(cx);
319
320        log::info!("gpui app started, initializing server");
321        let session = start_server(listeners, log_rx, cx);
322
323        client::init_settings(cx);
324
325        let project = cx.new_model(|cx| {
326            let fs = Arc::new(RealFs::new(Default::default(), None));
327            let node_settings_rx = initialize_settings(session.clone(), fs.clone(), cx);
328
329            let proxy_url = read_proxy_settings(cx);
330
331            let http_client = Arc::new(
332                ReqwestClient::proxy_and_user_agent(
333                    proxy_url,
334                    &format!(
335                        "Zed-Server/{} ({}; {})",
336                        env!("CARGO_PKG_VERSION"),
337                        std::env::consts::OS,
338                        std::env::consts::ARCH
339                    ),
340                )
341                .expect("Could not start HTTP client"),
342            );
343
344            let node_runtime = NodeRuntime::new(http_client.clone(), node_settings_rx);
345
346            let mut languages = LanguageRegistry::new(cx.background_executor().clone());
347            languages.set_language_server_download_dir(paths::languages_dir().clone());
348            let languages = Arc::new(languages);
349
350            HeadlessProject::new(
351                HeadlessAppState {
352                    session,
353                    fs,
354                    http_client,
355                    node_runtime,
356                    languages,
357                },
358                cx,
359            )
360        });
361
362        mem::forget(project);
363    });
364    log::info!("gpui app is shut down. quitting.");
365    Ok(())
366}
367
368#[derive(Clone)]
369struct ServerPaths {
370    log_file: PathBuf,
371    pid_file: PathBuf,
372    stdin_socket: PathBuf,
373    stdout_socket: PathBuf,
374    stderr_socket: PathBuf,
375}
376
377impl ServerPaths {
378    fn new(identifier: &str) -> Result<Self> {
379        let server_dir = paths::remote_server_state_dir().join(identifier);
380        std::fs::create_dir_all(&server_dir)?;
381        std::fs::create_dir_all(&logs_dir())?;
382
383        let pid_file = server_dir.join("server.pid");
384        let stdin_socket = server_dir.join("stdin.sock");
385        let stdout_socket = server_dir.join("stdout.sock");
386        let stderr_socket = server_dir.join("stderr.sock");
387        let log_file = logs_dir().join(format!("server-{}.log", identifier));
388
389        Ok(Self {
390            pid_file,
391            stdin_socket,
392            stdout_socket,
393            stderr_socket,
394            log_file,
395        })
396    }
397}
398
399pub fn execute_proxy(identifier: String, is_reconnecting: bool) -> Result<()> {
400    init_logging_proxy();
401    init_panic_hook();
402
403    log::info!("starting proxy process. PID: {}", std::process::id());
404
405    let server_paths = ServerPaths::new(&identifier)?;
406
407    let server_pid = check_pid_file(&server_paths.pid_file)?;
408    let server_running = server_pid.is_some();
409    if is_reconnecting {
410        if !server_running {
411            log::error!("attempted to reconnect, but no server running");
412            return Err(anyhow!(ProxyLaunchError::ServerNotRunning));
413        }
414    } else {
415        if let Some(pid) = server_pid {
416            log::info!("proxy found server already running with PID {}. Killing process and cleaning up files...", pid);
417            kill_running_server(pid, &server_paths)?;
418        }
419
420        spawn_server(&server_paths)?;
421    };
422
423    let stdin_task = smol::spawn(async move {
424        let stdin = Async::new(std::io::stdin())?;
425        let stream = smol::net::unix::UnixStream::connect(&server_paths.stdin_socket).await?;
426        handle_io(stdin, stream, "stdin").await
427    });
428
429    let stdout_task: smol::Task<Result<()>> = smol::spawn(async move {
430        let stdout = Async::new(std::io::stdout())?;
431        let stream = smol::net::unix::UnixStream::connect(&server_paths.stdout_socket).await?;
432        handle_io(stream, stdout, "stdout").await
433    });
434
435    let stderr_task: smol::Task<Result<()>> = smol::spawn(async move {
436        let mut stderr = Async::new(std::io::stderr())?;
437        let mut stream = smol::net::unix::UnixStream::connect(&server_paths.stderr_socket).await?;
438        let mut stderr_buffer = vec![0; 2048];
439        loop {
440            match stream.read(&mut stderr_buffer).await {
441                Ok(0) => {
442                    let error =
443                        std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "stderr closed");
444                    Err(anyhow!(error))?;
445                }
446                Ok(n) => {
447                    stderr.write_all(&mut stderr_buffer[..n]).await?;
448                    stderr.flush().await?;
449                }
450                Err(error) => {
451                    Err(anyhow!("error reading stderr: {error:?}"))?;
452                }
453            }
454        }
455    });
456
457    if let Err(forwarding_result) = smol::block_on(async move {
458        futures::select! {
459            result = stdin_task.fuse() => result,
460            result = stdout_task.fuse() => result,
461            result = stderr_task.fuse() => result,
462        }
463    }) {
464        if let Some(error) = forwarding_result.downcast_ref::<std::io::Error>() {
465            if error.kind() == std::io::ErrorKind::UnexpectedEof {
466                log::error!("connection to server closed due to unexpected EOF");
467                return Err(anyhow!("connection to server closed"));
468            }
469        }
470        log::error!(
471            "failed to forward messages: {:?}, terminating...",
472            forwarding_result
473        );
474        return Err(forwarding_result);
475    }
476
477    Ok(())
478}
479
480fn kill_running_server(pid: u32, paths: &ServerPaths) -> Result<()> {
481    log::info!("killing existing server with PID {}", pid);
482    std::process::Command::new("kill")
483        .arg(pid.to_string())
484        .output()
485        .context("failed to kill existing server")?;
486
487    for file in [
488        &paths.pid_file,
489        &paths.stdin_socket,
490        &paths.stdout_socket,
491        &paths.stderr_socket,
492    ] {
493        log::debug!("cleaning up file {:?} before starting new server", file);
494        std::fs::remove_file(file).ok();
495    }
496    Ok(())
497}
498
499fn spawn_server(paths: &ServerPaths) -> Result<()> {
500    if paths.stdin_socket.exists() {
501        std::fs::remove_file(&paths.stdin_socket)?;
502    }
503    if paths.stdout_socket.exists() {
504        std::fs::remove_file(&paths.stdout_socket)?;
505    }
506    if paths.stderr_socket.exists() {
507        std::fs::remove_file(&paths.stderr_socket)?;
508    }
509
510    let binary_name = std::env::current_exe()?;
511    let server_process = smol::process::Command::new(binary_name)
512        .arg("run")
513        .arg("--log-file")
514        .arg(&paths.log_file)
515        .arg("--pid-file")
516        .arg(&paths.pid_file)
517        .arg("--stdin-socket")
518        .arg(&paths.stdin_socket)
519        .arg("--stdout-socket")
520        .arg(&paths.stdout_socket)
521        .arg("--stderr-socket")
522        .arg(&paths.stderr_socket)
523        .spawn()?;
524
525    log::info!(
526        "proxy spawned server process. PID: {:?}",
527        server_process.id()
528    );
529
530    let mut total_time_waited = std::time::Duration::from_secs(0);
531    let wait_duration = std::time::Duration::from_millis(20);
532    while !paths.stdout_socket.exists()
533        || !paths.stdin_socket.exists()
534        || !paths.stderr_socket.exists()
535    {
536        log::debug!("waiting for server to be ready to accept connections...");
537        std::thread::sleep(wait_duration);
538        total_time_waited += wait_duration;
539    }
540
541    log::info!(
542        "server ready to accept connections. total time waited: {:?}",
543        total_time_waited
544    );
545
546    Ok(())
547}
548
549fn check_pid_file(path: &Path) -> Result<Option<u32>> {
550    let Some(pid) = std::fs::read_to_string(&path)
551        .ok()
552        .and_then(|contents| contents.parse::<u32>().ok())
553    else {
554        return Ok(None);
555    };
556
557    log::debug!("Checking if process with PID {} exists...", pid);
558    match std::process::Command::new("kill")
559        .arg("-0")
560        .arg(pid.to_string())
561        .output()
562    {
563        Ok(output) if output.status.success() => {
564            log::debug!("Process with PID {} exists. NOT spawning new server, but attaching to existing one.", pid);
565            Ok(Some(pid))
566        }
567        _ => {
568            log::debug!(
569                "Found PID file, but process with that PID does not exist. Removing PID file."
570            );
571            std::fs::remove_file(&path).context("Failed to remove PID file")?;
572            Ok(None)
573        }
574    }
575}
576
577fn write_pid_file(path: &Path) -> Result<()> {
578    if path.exists() {
579        std::fs::remove_file(path)?;
580    }
581    let pid = std::process::id().to_string();
582    log::debug!("writing PID {} to file {:?}", pid, path);
583    std::fs::write(path, pid).context("Failed to write PID file")
584}
585
586async fn handle_io<R, W>(mut reader: R, mut writer: W, socket_name: &str) -> Result<()>
587where
588    R: AsyncRead + Unpin,
589    W: AsyncWrite + Unpin,
590{
591    use remote::protocol::read_message_raw;
592
593    let mut buffer = Vec::new();
594    loop {
595        read_message_raw(&mut reader, &mut buffer)
596            .await
597            .with_context(|| format!("failed to read message from {}", socket_name))?;
598
599        write_size_prefixed_buffer(&mut writer, &mut buffer)
600            .await
601            .with_context(|| format!("failed to write message to {}", socket_name))?;
602
603        writer.flush().await?;
604
605        buffer.clear();
606    }
607}
608
609async fn write_size_prefixed_buffer<S: AsyncWrite + Unpin>(
610    stream: &mut S,
611    buffer: &mut Vec<u8>,
612) -> Result<()> {
613    let len = buffer.len() as u32;
614    stream.write_all(len.to_le_bytes().as_slice()).await?;
615    stream.write_all(buffer).await?;
616    Ok(())
617}
618
619fn initialize_settings(
620    session: Arc<ChannelClient>,
621    fs: Arc<dyn Fs>,
622    cx: &mut AppContext,
623) -> async_watch::Receiver<Option<NodeBinaryOptions>> {
624    let user_settings_file_rx = watch_config_file(
625        &cx.background_executor(),
626        fs,
627        paths::settings_file().clone(),
628    );
629
630    handle_settings_file_changes(user_settings_file_rx, cx, {
631        let session = session.clone();
632        move |err, _cx| {
633            if let Some(e) = err {
634                log::info!("Server settings failed to change: {}", e);
635
636                session
637                    .send(proto::Toast {
638                        project_id: SSH_PROJECT_ID,
639                        notification_id: "server-settings-failed".to_string(),
640                        message: format!(
641                            "Error in settings on remote host {:?}: {}",
642                            paths::settings_file(),
643                            e
644                        ),
645                    })
646                    .log_err();
647            } else {
648                session
649                    .send(proto::HideToast {
650                        project_id: SSH_PROJECT_ID,
651                        notification_id: "server-settings-failed".to_string(),
652                    })
653                    .log_err();
654            }
655        }
656    });
657
658    let (tx, rx) = async_watch::channel(None);
659    cx.observe_global::<SettingsStore>(move |cx| {
660        let settings = &ProjectSettings::get_global(cx).node;
661        log::info!("Got new node settings: {:?}", settings);
662        let options = NodeBinaryOptions {
663            allow_path_lookup: !settings.ignore_system_version.unwrap_or_default(),
664            // TODO: Implement this setting
665            allow_binary_download: true,
666            use_paths: settings.path.as_ref().map(|node_path| {
667                let node_path = PathBuf::from(shellexpand::tilde(node_path).as_ref());
668                let npm_path = settings
669                    .npm_path
670                    .as_ref()
671                    .map(|path| PathBuf::from(shellexpand::tilde(&path).as_ref()));
672                (
673                    node_path.clone(),
674                    npm_path.unwrap_or_else(|| {
675                        let base_path = PathBuf::new();
676                        node_path.parent().unwrap_or(&base_path).join("npm")
677                    }),
678                )
679            }),
680        };
681        tx.send(Some(options)).log_err();
682    })
683    .detach();
684
685    rx
686}
687
688pub fn handle_settings_file_changes(
689    mut server_settings_file: mpsc::UnboundedReceiver<String>,
690    cx: &mut AppContext,
691    settings_changed: impl Fn(Option<anyhow::Error>, &mut AppContext) + 'static,
692) {
693    let server_settings_content = cx
694        .background_executor()
695        .block(server_settings_file.next())
696        .unwrap();
697    SettingsStore::update_global(cx, |store, cx| {
698        store
699            .set_server_settings(&server_settings_content, cx)
700            .log_err();
701    });
702    cx.spawn(move |cx| async move {
703        while let Some(server_settings_content) = server_settings_file.next().await {
704            let result = cx.update_global(|store: &mut SettingsStore, cx| {
705                let result = store.set_server_settings(&server_settings_content, cx);
706                if let Err(err) = &result {
707                    log::error!("Failed to load server settings: {err}");
708                }
709                settings_changed(result.err(), cx);
710                cx.refresh();
711            });
712            if result.is_err() {
713                break; // App dropped
714            }
715        }
716    })
717    .detach();
718}
719
720fn read_proxy_settings(cx: &mut ModelContext<'_, HeadlessProject>) -> Option<Uri> {
721    let proxy_str = ProxySettings::get_global(cx).proxy.to_owned();
722    let proxy_url = proxy_str
723        .as_ref()
724        .and_then(|input: &String| {
725            input
726                .parse::<Uri>()
727                .inspect_err(|e| log::error!("Error parsing proxy settings: {}", e))
728                .ok()
729        })
730        .or_else(read_proxy_from_env);
731    proxy_url
732}