use crate::HeadlessProject;use crate::headless_project::HeadlessAppState;use anyhow::{Context as _, Result, anyhow};use client::{ProjectId, ProxySettings};use collections::HashMap;use project::trusted_worktrees;use util::ResultExt; use extension::ExtensionHostProxy;use fs::{Fs, RealFs};use futures::channel::{mpsc, oneshot};use futures::{AsyncRead, AsyncWrite, AsyncWriteExt, FutureExt, SinkExt, select, select_biased};use git::GitHostingProviderRegistry;use gpui::{App, AppContext as _, Context, Entity, UpdateGlobal as _};use gpui_tokio::Tokio;use http_client::{Url, read_proxy_from_env};use language::LanguageRegistry;use node_runtime::{NodeBinaryOptions, NodeRuntime};use paths::logs_dir;use project::project_settings::ProjectSettings;use util::command::new_smol_command; use proto::CrashReport;use release_channel::{AppCommitSha, AppVersion, RELEASE_CHANNEL, ReleaseChannel};use remote::RemoteClient;use remote::{    json_log::LogRecord,    protocol::{read_message, write_message},    proxy::ProxyLaunchError,};use reqwest_client::ReqwestClient;use rpc::proto::{self, Envelope, REMOTE_SERVER_PROJECT_ID};use rpc::{AnyProtoClient, TypedEnvelope};use settings::{Settings, SettingsStore, watch_config_file}; use smol::channel::{Receiver, Sender};use smol::io::AsyncReadExt;use smol::{net::unix::UnixListener, stream::StreamExt as _};use std::{    env,    ffi::OsStr,    fs::File,    io::Write,    mem,    ops::ControlFlow,    path::{Path, PathBuf},    process::ExitStatus,    str::FromStr,    sync::{Arc, LazyLock},};use thiserror::Error; pub static VERSION: LazyLock<String> = LazyLock::new(|| match *RELEASE_CHANNEL {    ReleaseChannel::Stable | ReleaseChannel::Preview => env!("ZED_PKG_VERSION").to_owned(),    ReleaseChannel::Nightly | ReleaseChannel::Dev => {        let commit_sha = option_env!("ZED_COMMIT_SHA").unwrap_or("missing-zed-commit-sha");        let build_identifier = option_env!("ZED_BUILD_ID");        if let Some(build_id) = build_identifier {            format!("{build_id}+{commit_sha}")        } else {            commit_sha.to_owned()        }    }}); fn init_logging_proxy() {    env_logger::builder()        .format(|buf, record| {            let mut log_record = LogRecord::new(record);            log_record.message = format!("(remote proxy) {}", log_record.message);            serde_json::to_writer(&mut *buf, &log_record)?;            buf.write_all(b"\n")?;            Ok(())        })        .init();} fn init_logging_server(log_file_path: PathBuf) -> Result<Receiver<Vec<u8>>> {    struct MultiWrite {        file: File,        channel: Sender<Vec<u8>>,        buffer: Vec<u8>,    }     impl Write for MultiWrite {        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {            let written = self.file.write(buf)?;            self.buffer.extend_from_slice(&buf[..written]);            Ok(written)        }         fn flush(&mut self) -> std::io::Result<()> {            self.channel                .send_blocking(self.buffer.clone())                .map_err(std::io::Error::other)?;            self.buffer.clear();            self.file.flush()        }    }     let log_file = std::fs::OpenOptions::new()        .create(true)        .append(true)        .open(&log_file_path)        .context("Failed to open log file in append mode")?;     let (tx, rx) = smol::channel::unbounded();     let target = Box::new(MultiWrite {        file: log_file,        channel: tx,        buffer: Vec::new(),    });     env_logger::Builder::new()        .filter_level(log::LevelFilter::Info)        .parse_default_env()        .target(env_logger::Target::Pipe(target))        .format(|buf, record| {            let mut log_record = LogRecord::new(record);            log_record.message = format!("(remote server) {}", log_record.message);            serde_json::to_writer(&mut *buf, &log_record)?;            buf.write_all(b"\n")?;            Ok(())        })        .init();     Ok(rx)} fn handle_crash_files_requests(project: &Entity<HeadlessProject>, client: &AnyProtoClient) {    client.add_request_handler(        project.downgrade(),        |_, _: TypedEnvelope<proto::GetCrashFiles>, _cx| async move {            let mut legacy_panics = Vec::new();            let mut crashes = Vec::new();            let mut children = smol::fs::read_dir(paths::logs_dir()).await?;            while let Some(child) = children.next().await {                let child = child?;                let child_path = child.path();                 let extension = child_path.extension();                if extension == Some(OsStr::new("panic")) {                    let filename = if let Some(filename) = child_path.file_name() {                        filename.to_string_lossy()                    } else {                        continue;                    };                     if !filename.starts_with("zed") {                        continue;                    }                     let file_contents = smol::fs::read_to_string(&child_path)                        .await                        .context("error reading panic file")?;                     legacy_panics.push(file_contents);                    smol::fs::remove_file(&child_path)                        .await                        .context("error removing panic")                        .log_err();                } else if extension == Some(OsStr::new("dmp")) {                    let mut json_path = child_path.clone();                    json_path.set_extension("json");                    if let Ok(json_content) = smol::fs::read_to_string(&json_path).await {                        crashes.push(CrashReport {                            metadata: json_content,                            minidump_contents: smol::fs::read(&child_path).await?,                        });                        smol::fs::remove_file(&child_path).await.log_err();                        smol::fs::remove_file(&json_path).await.log_err();                    } else {                        log::error!("Couldn't find json metadata for crash: {child_path:?}");                    }                }            }             anyhow::Ok(proto::GetCrashFilesResponse { crashes })        },    );} struct ServerListeners {    stdin: UnixListener,    stdout: UnixListener,    stderr: UnixListener,} impl ServerListeners {    pub fn new(stdin_path: PathBuf, stdout_path: PathBuf, stderr_path: PathBuf) -> Result<Self> {        Ok(Self {            stdin: UnixListener::bind(stdin_path).context("failed to bind stdin socket")?,            stdout: UnixListener::bind(stdout_path).context("failed to bind stdout socket")?,            stderr: UnixListener::bind(stderr_path).context("failed to bind stderr socket")?,        })    }} fn start_server(    listeners: ServerListeners,    log_rx: Receiver<Vec<u8>>,    cx: &mut App,    is_wsl_interop: bool,) -> AnyProtoClient {    // This is the server idle timeout. If no connection comes in this timeout, the server will shut down.    const IDLE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10 * 60);     let (incoming_tx, incoming_rx) = mpsc::unbounded::<Envelope>();    let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded::<Envelope>();    let (app_quit_tx, mut app_quit_rx) = mpsc::unbounded::<()>();     cx.on_app_quit(move |_| {        let mut app_quit_tx = app_quit_tx.clone();        async move {            log::info!("app quitting. sending signal to server main loop");            app_quit_tx.send(()).await.ok();        }    })    .detach();     cx.spawn(async move |cx| {        let mut stdin_incoming = listeners.stdin.incoming();        let mut stdout_incoming = listeners.stdout.incoming();        let mut stderr_incoming = listeners.stderr.incoming();         loop {            let streams = futures::future::join3(stdin_incoming.next(), stdout_incoming.next(), stderr_incoming.next());             log::info!("accepting new connections");            let result = select! {                streams = streams.fuse() => {                    let (Some(Ok(stdin_stream)), Some(Ok(stdout_stream)), Some(Ok(stderr_stream))) = streams else {                        break;                    };                    anyhow::Ok((stdin_stream, stdout_stream, stderr_stream))                }                _ = futures::FutureExt::fuse(smol::Timer::after(IDLE_TIMEOUT)) => {                    log::warn!("timed out waiting for new connections after {:?}. exiting.", IDLE_TIMEOUT);                    cx.update(|cx| {                        // TODO: This is a hack, because in a headless project, shutdown isn't executed                        // when calling quit, but it should be.                        cx.shutdown();                        cx.quit();                    })?;                    break;                }                _ = app_quit_rx.next().fuse() => {                    break;                }            };             let Ok((mut stdin_stream, mut stdout_stream, mut stderr_stream)) = result else {                break;            };             let mut input_buffer = Vec::new();            let mut output_buffer = Vec::new();             let (mut stdin_msg_tx, mut stdin_msg_rx) = mpsc::unbounded::<Envelope>();            cx.background_spawn(async move {                while let Ok(msg) = read_message(&mut stdin_stream, &mut input_buffer).await {                    if (stdin_msg_tx.send(msg).await).is_err() {                        break;                    }                }            }).detach();             loop {                 select_biased! {                    _ = app_quit_rx.next().fuse() => {                        return anyhow::Ok(());                    }                     stdin_message = stdin_msg_rx.next().fuse() => {                        let Some(message) = stdin_message else {                            log::warn!("error reading message on stdin. exiting.");                            break;                        };                        if let Err(error) = incoming_tx.unbounded_send(message) {                            log::error!("failed to send message to application: {error:?}. exiting.");                            return Err(anyhow!(error));                        }                    }                     outgoing_message  = outgoing_rx.next().fuse() => {                        let Some(message) = outgoing_message else {                            log::error!("stdout handler, no message");                            break;                        };                         if let Err(error) =                            write_message(&mut stdout_stream, &mut output_buffer, message).await                        {                            log::error!("failed to write stdout message: {:?}", error);                            break;                        }                        if let Err(error) = stdout_stream.flush().await {                            log::error!("failed to flush stdout message: {:?}", error);                            break;                        }                    }                     log_message = log_rx.recv().fuse() => {                        if let Ok(log_message) = log_message {                            if let Err(error) = stderr_stream.write_all(&log_message).await {                                log::error!("failed to write log message to stderr: {:?}", error);                                break;                            }                            if let Err(error) = stderr_stream.flush().await {                                log::error!("failed to flush stderr stream: {:?}", error);                                break;                            }                        }                    }                }            }        }        anyhow::Ok(())    })    .detach();     RemoteClient::proto_client_from_channels(incoming_rx, outgoing_tx, cx, "server", is_wsl_interop)} fn init_paths() -> anyhow::Result<()> {    for path in [        paths::config_dir(),        paths::extensions_dir(),        paths::languages_dir(),        paths::logs_dir(),        paths::temp_dir(),        paths::hang_traces_dir(),        paths::remote_extensions_dir(),        paths::remote_extensions_uploads_dir(),    ]    .iter()    {        std::fs::create_dir_all(path).with_context(|| format!("creating directory {path:?}"))?;    }    Ok(())} pub fn execute_run(    log_file: PathBuf,    pid_file: PathBuf,    stdin_socket: PathBuf,    stdout_socket: PathBuf,    stderr_socket: PathBuf,) -> Result<()> {    init_paths()?;     match daemonize()? {        ControlFlow::Break(_) => return Ok(()),        ControlFlow::Continue(_) => {}    }     let app = gpui::Application::headless();    let id = std::process::id().to_string();    app.background_executor()        .spawn(crashes::init(crashes::InitCrashHandler {            session_id: id,            zed_version: VERSION.to_owned(),            binary: "zed-remote-server".to_string(),            release_channel: release_channel::RELEASE_CHANNEL_NAME.clone(),            commit_sha: option_env!("ZED_COMMIT_SHA").unwrap_or("no_sha").to_owned(),        }))        .detach();    let log_rx = init_logging_server(log_file)?;    log::info!(        "starting up. pid_file: {:?}, stdin_socket: {:?}, stdout_socket: {:?}, stderr_socket: {:?}",        pid_file,        stdin_socket,        stdout_socket,        stderr_socket    );     write_pid_file(&pid_file)        .with_context(|| format!("failed to write pid file: {:?}", &pid_file))?;     let listeners = ServerListeners::new(stdin_socket, stdout_socket, stderr_socket)?;     rayon::ThreadPoolBuilder::new()        .num_threads(std::thread::available_parallelism().map_or(1, |n| n.get().div_ceil(2)))        .stack_size(10 * 1024 * 1024)        .thread_name(|ix| format!("RayonWorker{}", ix))        .build_global()        .unwrap();     let (shell_env_loaded_tx, shell_env_loaded_rx) = oneshot::channel();    app.background_executor()        .spawn(async {            util::load_login_shell_environment().await.log_err();            shell_env_loaded_tx.send(()).ok();        })        .detach();     let git_hosting_provider_registry = Arc::new(GitHostingProviderRegistry::new());    app.run(move |cx| {        settings::init(cx);        let app_commit_sha = option_env!("ZED_COMMIT_SHA").map(|s| AppCommitSha::new(s.to_owned()));        let app_version = AppVersion::load(            env!("ZED_PKG_VERSION"),            option_env!("ZED_BUILD_ID"),            app_commit_sha,        );        release_channel::init(app_version, cx);        gpui_tokio::init(cx);         HeadlessProject::init(cx);         let is_wsl_interop = if cfg!(target_os = "linux") {            // See: https://learn.microsoft.com/en-us/windows/wsl/filesystems#disable-interoperability            matches!(std::fs::read_to_string("/proc/sys/fs/binfmt_misc/WSLInterop"), Ok(s) if s.contains("enabled"))        } else {            false        };         log::info!("gpui app started, initializing server");        let session = start_server(listeners, log_rx, cx, is_wsl_interop);        trusted_worktrees::init(HashMap::default(), Some((session.clone(), ProjectId(REMOTE_SERVER_PROJECT_ID))), None, cx);         GitHostingProviderRegistry::set_global(git_hosting_provider_registry, cx);        git_hosting_providers::init(cx);        dap_adapters::init(cx);         extension::init(cx);        let extension_host_proxy = ExtensionHostProxy::global(cx);         json_schema_store::init(cx);         let project = cx.new(|cx| {            let fs = Arc::new(RealFs::new(None, cx.background_executor().clone()));            let node_settings_rx = initialize_settings(session.clone(), fs.clone(), cx);             let proxy_url = read_proxy_settings(cx);             let http_client = {                let _guard = Tokio::handle(cx).enter();                Arc::new(                    ReqwestClient::proxy_and_user_agent(                        proxy_url,                        &format!(                            "Zed-Server/{} ({}; {})",                            env!("CARGO_PKG_VERSION"),                            std::env::consts::OS,                            std::env::consts::ARCH                        ),                    )                    .expect("Could not start HTTP client"),                )            };             let node_runtime = NodeRuntime::new(                http_client.clone(),                Some(shell_env_loaded_rx),                node_settings_rx,            );             let mut languages = LanguageRegistry::new(cx.background_executor().clone());            languages.set_language_server_download_dir(paths::languages_dir().clone());            let languages = Arc::new(languages);             HeadlessProject::new(                HeadlessAppState {                    session: session.clone(),                    fs,                    http_client,                    node_runtime,                    languages,                    extension_host_proxy,                },                true,                cx,            )        });         handle_crash_files_requests(&project, &session);         cx.background_spawn(async move { cleanup_old_binaries() })            .detach();         mem::forget(project);    });    log::info!("gpui app is shut down. quitting.");    Ok(())} #[derive(Debug, Error)]pub(crate) enum ServerPathError {    #[error("Failed to create server_dir `{path}`")]    CreateServerDir {        #[source]        source: std::io::Error,        path: PathBuf,    },    #[error("Failed to create logs_dir `{path}`")]    CreateLogsDir {        #[source]

Lines 1–500 of 1057. Load more