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::{    Timer,    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();} const REMOTE_SERVER_LOG_MAX_BYTES: u64 = 1024 * 1024; struct RotatingLogFile {    path: PathBuf,    file: File,    size_bytes: u64,} impl RotatingLogFile {    fn open(path: &Path) -> Result<Self> {        if std::fs::metadata(path)            .map(|metadata| metadata.len() >= REMOTE_SERVER_LOG_MAX_BYTES)            .unwrap_or(false)        {            rotate_log_file(path, &rotated_log_path(path))                .context("failed to rotate existing remote server log")?;        }         let file = open_log_file(path).context("failed to open remote server log")?;        let size_bytes = file            .metadata()            .context("failed to read remote server log metadata")?            .len();         Ok(Self {            path: path.to_path_buf(),            file,            size_bytes,        })    }     fn rotate(&mut self) -> std::io::Result<()> {        self.file.flush()?;        rotate_log_file(&self.path, &rotated_log_path(&self.path))?;        self.file = open_log_file(&self.path)?;        self.size_bytes = 0;        Ok(())    }} impl Write for RotatingLogFile {    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {        if self.size_bytes.saturating_add(buf.len() as u64) > REMOTE_SERVER_LOG_MAX_BYTES {            self.rotate()?;        }         self.file.write_all(buf)?;        self.size_bytes += buf.len() as u64;         Ok(buf.len())    }     fn flush(&mut self) -> std::io::Result<()> {        self.file.flush()    }} fn open_log_file(path: &Path) -> std::io::Result<File> {    std::fs::OpenOptions::new()        .create(true)        .append(true)        .open(path)} fn rotated_log_path(path: &Path) -> PathBuf {    path.with_extension("1.log")} fn rotate_log_file(path: &Path, rotated_path: &Path) -> std::io::Result<()> {    match std::fs::remove_file(rotated_path) {        Ok(()) => {}        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}        Err(error) => return Err(error),    }     match std::fs::rename(path, rotated_path) {        Ok(()) => {}        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}        Err(error) => return Err(error),    }     Ok(())} fn init_logging_server(log_file_path: &Path) -> Result<Receiver<Vec<u8>>> {    struct MultiWrite {        file: RotatingLogFile,        channel: Sender<Vec<u8>>,        buffer: Vec<u8>,    }     impl Write for MultiWrite {        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {            self.file.write_all(buf)?;            self.buffer.extend_from_slice(buf);            Ok(buf.len())        }         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 = RotatingLogFile::open(log_file_path)        .context("Failed to open rotating remote server log file")?;     let (tx, rx) = async_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);

Lines 1–500 of 1407. Load more