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