use crate::HeadlessProject;use crate::headless_project::HeadlessAppState;use anyhow::{Context as _, Result, anyhow};use chrono::Utc;use client::{ProxySettings, telemetry}; use extension::ExtensionHostProxy;use fs::{Fs, RealFs};use futures::channel::mpsc;use futures::{AsyncRead, AsyncWrite, AsyncWriteExt, FutureExt, SinkExt, select, select_biased};use git::GitHostingProviderRegistry;use gpui::{App, AppContext as _, Context, Entity, SemanticVersion, 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 release_channel::{AppVersion, RELEASE_CHANNEL, ReleaseChannel};use remote::proxy::ProxyLaunchError;use remote::ssh_session::ChannelClient;use remote::{    json_log::LogRecord,    protocol::{read_message, write_message},};use reqwest_client::ReqwestClient;use rpc::proto::{self, Envelope, SSH_PROJECT_ID};use rpc::{AnyProtoClient, TypedEnvelope};use settings::{Settings, SettingsStore, watch_config_file};use smol::channel::{Receiver, Sender};use smol::io::AsyncReadExt; use smol::Async;use smol::{net::unix::UnixListener, stream::StreamExt as _};use std::ffi::OsStr;use std::ops::ControlFlow;use std::str::FromStr;use std::{env, thread};use std::{    io::Write,    mem,    path::{Path, PathBuf},    sync::Arc,};use telemetry_events::LocationData;use util::ResultExt; 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: std::fs::File,        channel: Sender<Vec<u8>>,        buffer: Vec<u8>,    }     impl std::io::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(|error| std::io::Error::new(std::io::ErrorKind::Other, error))?;            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::from_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 init_panic_hook() {    std::panic::set_hook(Box::new(|info| {        let payload = info            .payload()            .downcast_ref::<&str>()            .map(|s| s.to_string())            .or_else(|| info.payload().downcast_ref::<String>().cloned())            .unwrap_or_else(|| "Box<Any>".to_string());         let backtrace = backtrace::Backtrace::new();        let mut backtrace = backtrace            .frames()            .iter()            .flat_map(|frame| {                frame                    .symbols()                    .iter()                    .filter_map(|frame| Some(format!("{:#}", frame.name()?)))            })            .collect::<Vec<_>>();         // Strip out leading stack frames for rust panic-handling.        if let Some(ix) = backtrace            .iter()            .position(|name| name == "rust_begin_unwind")        {            backtrace.drain(0..=ix);        }         let thread = thread::current();        let thread_name = thread.name().unwrap_or("<unnamed>");         log::error!(            "panic occurred: {}\nBacktrace:\n{}",            &payload,            (&backtrace).join("\n")        );         let release_channel = *RELEASE_CHANNEL;        let version = match release_channel {            ReleaseChannel::Stable | ReleaseChannel::Preview => env!("ZED_PKG_VERSION"),            ReleaseChannel::Nightly | ReleaseChannel::Dev => {                option_env!("ZED_COMMIT_SHA").unwrap_or("missing-zed-commit-sha")            }        };         let panic_data = telemetry_events::Panic {            thread: thread_name.into(),            payload: payload.clone(),            location_data: info.location().map(|location| LocationData {                file: location.file().into(),                line: location.line(),            }),            app_version: format!("remote-server-{version}"),            app_commit_sha: option_env!("ZED_COMMIT_SHA").map(|sha| sha.into()),            release_channel: release_channel.display_name().into(),            target: env!("TARGET").to_owned().into(),            os_name: telemetry::os_name(),            os_version: Some(telemetry::os_version()),            architecture: env::consts::ARCH.into(),            panicked_on: Utc::now().timestamp_millis(),            backtrace,            system_id: None,            // Set on SSH client            installation_id: None,      // Set on SSH client            session_id: "".to_string(), // Set on SSH client        };         if let Some(panic_data_json) = serde_json::to_string(&panic_data).log_err() {            let timestamp = chrono::Utc::now().format("%Y_%m_%d %H_%M_%S").to_string();            let panic_file_path = paths::logs_dir().join(format!("zed-{timestamp}.panic"));            let panic_file = std::fs::OpenOptions::new()                .append(true)                .create(true)                .open(&panic_file_path)                .log_err();            if let Some(mut panic_file) = panic_file {                writeln!(&mut panic_file, "{panic_data_json}").log_err();                panic_file.flush().log_err();            }        }         std::process::abort();    }));} fn handle_panic_requests(project: &Entity<HeadlessProject>, client: &Arc<ChannelClient>) {    let client: AnyProtoClient = client.clone().into();    client.add_request_handler(        project.downgrade(),        |_, _: TypedEnvelope<proto::GetPanicFiles>, _cx| async move {            let mut children = smol::fs::read_dir(paths::logs_dir()).await?;            let mut panic_files = Vec::new();            while let Some(child) = children.next().await {                let child = child?;                let child_path = child.path();                 if child_path.extension() != Some(OsStr::new("panic")) {                    continue;                }                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")?;                 panic_files.push(file_contents);                 // We've done what we can, delete the file                std::fs::remove_file(child_path)                    .context("error removing panic")                    .log_err();            }            anyhow::Ok(proto::GetPanicFilesResponse {                file_contents: panic_files,            })        },    );} 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,) -> Arc<ChannelClient> {    // 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 let Err(_) = stdin_msg_tx.send(msg).await {                        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();     ChannelClient::new(incoming_rx, outgoing_tx, cx, "server")} fn init_paths() -> anyhow::Result<()> {    for path in [        paths::config_dir(),        paths::extensions_dir(),        paths::languages_dir(),        paths::logs_dir(),        paths::temp_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(_) => {}    }     init_panic_hook();    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)?;     let git_hosting_provider_registry = Arc::new(GitHostingProviderRegistry::new());    gpui::Application::headless().run(move |cx| {        settings::init(cx);        let app_version = AppVersion::load(env!("ZED_PKG_VERSION"));        release_channel::init(app_version, cx);        gpui_tokio::init(cx);         HeadlessProject::init(cx);         log::info!("gpui app started, initializing server");        let session = start_server(listeners, log_rx, cx);         client::init_settings(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);         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(), None, 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,                },                cx,            )        });         handle_panic_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(Clone)]

Lines 1–500 of 954. Load more