mod headless_project; #[cfg(test)]mod remote_editing_tests; #[cfg(windows)]pub mod windows; pub use headless_project::{HeadlessAppState, HeadlessProject}; use anyhow::{Context as _, Result, anyhow};use clap::Subcommand;use client::ProxySettings;use collections::HashMap;use extension::ExtensionHostProxy;use fs::{Fs, RealFs};use futures::{    AsyncRead, AsyncWrite, AsyncWriteExt, FutureExt, SinkExt,    channel::{mpsc, oneshot},    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 net::async_net::{UnixListener, UnixStream};use node_runtime::{NodeBinaryOptions, NodeRuntime};use paths::logs_dir;use project::{project_settings::ProjectSettings, trusted_worktrees};use proto::CrashReport;use release_channel::{AppCommitSha, AppVersion, RELEASE_CHANNEL, ReleaseChannel};use remote::{    RemoteClient,    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},    io::AsyncReadExt,    stream::StreamExt as _,};use std::{    env,    ffi::OsStr,    fs::File,    io::Write,    mem,    path::{Path, PathBuf},    str::FromStr,    sync::{Arc, LazyLock},    time::Instant,};use thiserror::Error;use util::{ResultExt, command::new_command}; #[derive(Subcommand)]pub enum Commands {    Run {        #[arg(long)]        log_file: PathBuf,        #[arg(long)]        pid_file: PathBuf,        #[arg(long)]        stdin_socket: PathBuf,        #[arg(long)]        stdout_socket: PathBuf,        #[arg(long)]        stderr_socket: PathBuf,    },    Proxy {        #[arg(long)]        reconnect: bool,        #[arg(long)]        identifier: String,    },    Version,} pub fn run(command: Commands) -> anyhow::Result<()> {    use anyhow::Context;    use release_channel::{RELEASE_CHANNEL, ReleaseChannel};     match command {        Commands::Run {            log_file,            pid_file,            stdin_socket,            stdout_socket,            stderr_socket,        } => execute_run(            log_file,            pid_file,            stdin_socket,            stdout_socket,            stderr_socket,        ),        Commands::Proxy {            identifier,            reconnect,        } => execute_proxy(identifier, reconnect).context("running proxy on the remote server"),        Commands::Version => {            let release_channel = *RELEASE_CHANNEL;            match release_channel {                ReleaseChannel::Stable | ReleaseChannel::Preview => {                    println!("{}", env!("ZED_PKG_VERSION"))                }                ReleaseChannel::Nightly | ReleaseChannel::Dev => {                    let commit_sha =                        option_env!("ZED_COMMIT_SHA").unwrap_or(release_channel.dev_name());                    let build_id = option_env!("ZED_BUILD_ID");                    if let Some(build_id) = build_id {                        println!("{}+{}", build_id, commit_sha)                    } else {                        println!("{commit_sha}");                    }                }            };            Ok(())        }    }} 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 =                std::borrow::Cow::Owned(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: &Path) -> 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(),    });     let old_hook = std::panic::take_hook();    std::panic::set_hook(Box::new(move |info| {        let backtrace = std::backtrace::Backtrace::force_capture();        let message = info.payload_as_str().unwrap_or("Box<Any>").to_owned();        let location = info            .location()            .map_or_else(|| "<unknown>".to_owned(), |location| location.to_string());        let current_thread = std::thread::current();        let thread_name = current_thread.name().unwrap_or("<unnamed>");         let msg = format!("thread '{thread_name}' panicked at {location}:\n{message}\n{backtrace}");        // NOTE: This log never reaches the client, as the communication is handled on a main thread task        // which will never run once we panic.        log::error!("{msg}");        old_hook(info);    }));    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 =                std::borrow::Cow::Owned(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| {        loop {            let streams = futures::future::join3(                listeners.stdin.accept(),                listeners.stdout.accept(),                listeners.stderr.accept(),            );             log::info!("accepting new connections");            let result = select! {                streams = streams.fuse() => {                    let (Ok((stdin_stream, _)), Ok((stdout_stream, _)), Ok((stderr_stream, _))) = streams else {                        log::error!("failed to accept new connections");                        break;                    };                    log::info!("accepted new connections");                    anyhow::Ok((stdin_stream, stdout_stream, stderr_stream))                }                _ = futures::FutureExt::fuse(cx.background_executor().timer(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() => {                    log::info!("app quit requested");                    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 {                loop {                    match read_message(&mut stdin_stream, &mut input_buffer).await {                        Ok(msg) => {                            if (stdin_msg_tx.send(msg).await).is_err() {                                log::info!("stdin message channel closed, stopping stdin reader");                                break;                            }                        }                        Err(error) => {                            log::warn!("stdin read failed: {error:?}");                            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, dropping connection.");                            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()?;     let startup_time = Instant::now();    let app = gpui_platform::headless();    let pid = std::process::id();    let id = pid.to_string();    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(),        },        |task| {            app.background_executor().spawn(task).detach();        },    );    let log_rx = init_logging_server(&log_file)?;    log::info!(        "starting up with PID {}:\npid_file: {:?}, log_file: {:?}, stdin_socket: {:?}, stdout_socket: {:?}, stderr_socket: {:?}",        pid,        pid_file,        log_file,        stdin_socket,        stdout_socket,        stderr_socket    );     write_pid_file(&pid_file, pid)        .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();     #[cfg(unix)]    let shell_env_loaded_rx = {

Lines 1–500 of 1230. Load more