tenant.openagents/omega

1mod headless_project;
2
3#[cfg(test)]
4mod remote_editing_tests;
5
6#[cfg(windows)]
7pub mod windows;
8
9pub use headless_project::{HeadlessAppState, HeadlessProject};
10
11use anyhow::{Context as _, Result, anyhow};
12use clap::Subcommand;
13use client::ProxySettings;
14use collections::HashMap;
15use extension::ExtensionHostProxy;
16use fs::{Fs, RealFs};
17use futures::{
18    AsyncRead, AsyncWrite, AsyncWriteExt, FutureExt, SinkExt,
19    channel::{mpsc, oneshot},
20    select, select_biased,
21};
22use git::GitHostingProviderRegistry;
23use gpui::{App, AppContext as _, Context, Entity, UpdateGlobal as _};
24use gpui_tokio::Tokio;
25use http_client::{Url, read_proxy_from_env};
26use language::LanguageRegistry;
27use net::async_net::{UnixListener, UnixStream};
28use node_runtime::{NodeBinaryOptions, NodeRuntime};
29use paths::logs_dir;
30use project::{project_settings::ProjectSettings, trusted_worktrees};
31use proto::CrashReport;
32use release_channel::{AppCommitSha, AppVersion, RELEASE_CHANNEL, ReleaseChannel};
33use remote::{
34    RemoteClient,
35    json_log::LogRecord,
36    protocol::{read_message, write_message},
37    proxy::ProxyLaunchError,
38};
39use reqwest_client::ReqwestClient;
40use rpc::proto::{self, Envelope, REMOTE_SERVER_PROJECT_ID};
41use rpc::{AnyProtoClient, TypedEnvelope};
42use settings::{Settings, SettingsStore, watch_config_file};
43use smol::{
44    Timer,
45    channel::{Receiver, Sender},
46    io::AsyncReadExt,
47    stream::StreamExt as _,
48};
49use std::{
50    ffi::OsStr,
51    fs::File,
52    io::Write,
53    mem,
54    path::{Path, PathBuf},
55    str::FromStr,
56    sync::{Arc, LazyLock},
57    time::Instant,
58};
59use thiserror::Error;
60use util::{ResultExt, command::new_command};
61
62#[derive(Subcommand)]
63pub enum Commands {
64    Run {
65        #[arg(long)]
66        log_file: PathBuf,
67        #[arg(long)]
68        pid_file: PathBuf,
69        #[arg(long)]
70        stdin_socket: PathBuf,
71        #[arg(long)]
72        stdout_socket: PathBuf,
73        #[arg(long)]
74        stderr_socket: PathBuf,
75    },
76    Proxy {
77        #[arg(long)]
78        reconnect: bool,
79        #[arg(long)]
80        identifier: String,
81    },
82    Version,
83}
84
85pub fn run(command: Commands) -> anyhow::Result<()> {
86    use anyhow::Context;
87    use release_channel::{RELEASE_CHANNEL, ReleaseChannel};
88
89    match command {
90        Commands::Run {
91            log_file,
92            pid_file,
93            stdin_socket,
94            stdout_socket,
95            stderr_socket,
96        } => execute_run(
97            log_file,
98            pid_file,
99            stdin_socket,
100            stdout_socket,
101            stderr_socket,
102        ),
103        Commands::Proxy {
104            identifier,
105            reconnect,
106        } => execute_proxy(identifier, reconnect).context("running proxy on the remote server"),
107        Commands::Version => {
108            let release_channel = *RELEASE_CHANNEL;
109            match release_channel {
110                ReleaseChannel::Stable | ReleaseChannel::Preview => {
111                    println!("{}", env!("ZED_PKG_VERSION"))
112                }
113                ReleaseChannel::Nightly | ReleaseChannel::Dev => {
114                    let commit_sha =
115                        option_env!("ZED_COMMIT_SHA").unwrap_or(release_channel.dev_name());
116                    let build_id = option_env!("ZED_BUILD_ID");
117                    if let Some(build_id) = build_id {
118                        println!("{}+{}", build_id, commit_sha)
119                    } else {
120                        println!("{commit_sha}");
121                    }
122                }
123            };
124            Ok(())
125        }
126    }
127}
128
129pub static VERSION: LazyLock<String> = LazyLock::new(|| match *RELEASE_CHANNEL {
130    ReleaseChannel::Stable | ReleaseChannel::Preview => env!("ZED_PKG_VERSION").to_owned(),
131    ReleaseChannel::Nightly | ReleaseChannel::Dev => {
132        let commit_sha = option_env!("ZED_COMMIT_SHA").unwrap_or("missing-zed-commit-sha");
133        let build_identifier = option_env!("ZED_BUILD_ID");
134        if let Some(build_id) = build_identifier {
135            format!("{build_id}+{commit_sha}")
136        } else {
137            commit_sha.to_owned()
138        }
139    }
140});
141
142fn init_logging_proxy() {
143    env_logger::builder()
144        .format(|buf, record| {
145            let mut log_record = LogRecord::new(record);
146            log_record.message =
147                std::borrow::Cow::Owned(format!("(remote proxy) {}", log_record.message));
148            serde_json::to_writer(&mut *buf, &log_record)?;
149            buf.write_all(b"\n")?;
150            Ok(())
151        })
152        .init();
153}
154
155const REMOTE_SERVER_LOG_MAX_BYTES: u64 = 1024 * 1024;
156
157struct RotatingLogFile {
158    path: PathBuf,
159    file: File,
160    size_bytes: u64,
161}
162
163impl RotatingLogFile {
164    fn open(path: &Path) -> Result<Self> {
165        if std::fs::metadata(path)
166            .map(|metadata| metadata.len() >= REMOTE_SERVER_LOG_MAX_BYTES)
167            .unwrap_or(false)
168        {
169            rotate_log_file(path, &rotated_log_path(path))
170                .context("failed to rotate existing remote server log")?;
171        }
172
173        let file = open_log_file(path).context("failed to open remote server log")?;
174        let size_bytes = file
175            .metadata()
176            .context("failed to read remote server log metadata")?
177            .len();
178
179        Ok(Self {
180            path: path.to_path_buf(),
181            file,
182            size_bytes,
183        })
184    }
185
186    fn rotate(&mut self) -> std::io::Result<()> {
187        self.file.flush()?;
188        rotate_log_file(&self.path, &rotated_log_path(&self.path))?;
189        self.file = open_log_file(&self.path)?;
190        self.size_bytes = 0;
191        Ok(())
192    }
193}
194
195impl Write for RotatingLogFile {
196    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
197        if self.size_bytes.saturating_add(buf.len() as u64) > REMOTE_SERVER_LOG_MAX_BYTES {
198            self.rotate()?;
199        }
200
201        self.file.write_all(buf)?;
202        self.size_bytes += buf.len() as u64;
203
204        Ok(buf.len())
205    }
206
207    fn flush(&mut self) -> std::io::Result<()> {
208        self.file.flush()
209    }
210}
211
212fn open_log_file(path: &Path) -> std::io::Result<File> {
213    std::fs::OpenOptions::new()
214        .create(true)
215        .append(true)
216        .open(path)
217}
218
219fn rotated_log_path(path: &Path) -> PathBuf {
220    path.with_extension("1.log")
221}
222
223fn rotate_log_file(path: &Path, rotated_path: &Path) -> std::io::Result<()> {
224    match std::fs::remove_file(rotated_path) {
225        Ok(()) => {}
226        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
227        Err(error) => return Err(error),
228    }
229
230    match std::fs::rename(path, rotated_path) {
231        Ok(()) => {}
232        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
233        Err(error) => return Err(error),
234    }
235
236    Ok(())
237}
238
239fn init_logging_server(log_file_path: &Path) -> Result<Receiver<Vec<u8>>> {
240    struct MultiWrite {
241        file: RotatingLogFile,
242        channel: Sender<Vec<u8>>,
243        buffer: Vec<u8>,
244    }
245
246    impl Write for MultiWrite {
247        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
248            self.file.write_all(buf)?;
249            self.buffer.extend_from_slice(buf);
250            Ok(buf.len())
251        }
252
253        fn flush(&mut self) -> std::io::Result<()> {
254            self.channel
255                .send_blocking(self.buffer.clone())
256                .map_err(std::io::Error::other)?;
257            self.buffer.clear();
258            self.file.flush()
259        }
260    }
261
262    let log_file = RotatingLogFile::open(log_file_path)
263        .context("Failed to open rotating remote server log file")?;
264
265    let (tx, rx) = async_channel::unbounded();
266
267    let target = Box::new(MultiWrite {
268        file: log_file,
269        channel: tx,
270        buffer: Vec::new(),
271    });
272
273    let old_hook = std::panic::take_hook();
274    std::panic::set_hook(Box::new(move |info| {
275        let backtrace = std::backtrace::Backtrace::force_capture();
276        let message = info.payload_as_str().unwrap_or("Box<Any>").to_owned();
277        let location = info
278            .location()
279            .map_or_else(|| "<unknown>".to_owned(), |location| location.to_string());
280        let current_thread = std::thread::current();
281        let thread_name = current_thread.name().unwrap_or("<unnamed>");
282
283        let msg = format!("thread '{thread_name}' panicked at {location}:\n{message}\n{backtrace}");
284        // NOTE: This log never reaches the client, as the communication is handled on a main thread task
285        // which will never run once we panic.
286        log::error!("{msg}");
287        old_hook(info);
288    }));
289    env_logger::Builder::new()
290        .filter_level(log::LevelFilter::Info)
291        .parse_default_env()
292        .target(env_logger::Target::Pipe(target))
293        .format(|buf, record| {
294            let mut log_record = LogRecord::new(record);
295            log_record.message =
296                std::borrow::Cow::Owned(format!("(remote server) {}", log_record.message));
297            serde_json::to_writer(&mut *buf, &log_record)?;
298            buf.write_all(b"\n")?;
299            Ok(())
300        })
301        .init();
302
303    Ok(rx)
304}
305
306/// Initializes the telemetry queue on the remote server, forwarding every
307/// emitted event to the connected client over the proto channel.
308///
309/// The remote server cannot upload telemetry itself (it has no logged-in user,
310/// no checksum seed, and no Telemetry instance), so without this its
311/// `telemetry::event!` calls are silently dropped. The client attributes these
312/// events to the remote host using the platform it already detected during
313/// connection setup, so no OS metadata needs to be sent here.
314fn init_telemetry_forwarding(session: AnyProtoClient, cx: &mut App) {
315    let (tx, mut rx) = mpsc::unbounded::<telemetry::Event>();
316    telemetry::init(tx);
317
318    cx.background_spawn(async move {
319        while let Some(event) = rx.next().await {
320            let Some(event_json) = serde_json::to_string(&event).log_err() else {
321                continue;
322            };
323            session
324                .send(proto::TelemetryEvent {
325                    project_id: REMOTE_SERVER_PROJECT_ID,
326                    event_json,
327                })
328                .log_err();
329        }
330    })
331    .detach();
332}
333
334fn handle_crash_files_requests(project: &Entity<HeadlessProject>, client: &AnyProtoClient) {
335    client.add_request_handler(
336        project.downgrade(),
337        |_, _: TypedEnvelope<proto::GetCrashFiles>, _cx| async move {
338            let mut legacy_panics = Vec::new();
339            let mut crashes = Vec::new();
340            let mut children = smol::fs::read_dir(paths::logs_dir()).await?;
341            while let Some(child) = children.next().await {
342                let child = child?;
343                let child_path = child.path();
344
345                let extension = child_path.extension();
346                if extension == Some(OsStr::new("panic")) {
347                    let filename = if let Some(filename) = child_path.file_name() {
348                        filename.to_string_lossy()
349                    } else {
350                        continue;
351                    };
352
353                    if !filename.starts_with("zed") {
354                        continue;
355                    }
356
357                    let file_contents = smol::fs::read_to_string(&child_path)
358                        .await
359                        .context("error reading panic file")?;
360
361                    legacy_panics.push(file_contents);
362                    smol::fs::remove_file(&child_path)
363                        .await
364                        .context("error removing panic")
365                        .log_err();
366                } else if extension == Some(OsStr::new("dmp")) {
367                    let mut json_path = child_path.clone();
368                    json_path.set_extension("json");
369                    if let Ok(json_content) = smol::fs::read_to_string(&json_path).await {
370                        crashes.push(CrashReport {
371                            metadata: json_content,
372                            minidump_contents: smol::fs::read(&child_path).await?,
373                        });
374                        smol::fs::remove_file(&child_path).await.log_err();
375                        smol::fs::remove_file(&json_path).await.log_err();
376                    } else {
377                        log::error!("Couldn't find json metadata for crash: {child_path:?}");
378                    }
379                }
380            }
381
382            anyhow::Ok(proto::GetCrashFilesResponse { crashes })
383        },
384    );
385}
386
387struct ServerListeners {
388    stdin: UnixListener,
389    stdout: UnixListener,
390    stderr: UnixListener,
391}
392
393impl ServerListeners {
394    pub fn new(stdin_path: PathBuf, stdout_path: PathBuf, stderr_path: PathBuf) -> Result<Self> {
395        Ok(Self {
396            stdin: UnixListener::bind(stdin_path).context("failed to bind stdin socket")?,
397            stdout: UnixListener::bind(stdout_path).context("failed to bind stdout socket")?,
398            stderr: UnixListener::bind(stderr_path).context("failed to bind stderr socket")?,
399        })
400    }
401}
402
403fn start_server(
404    listeners: ServerListeners,
405    log_rx: Receiver<Vec<u8>>,
406    cx: &mut App,
407    is_wsl_interop: bool,
408) -> AnyProtoClient {
409    // This is the server idle timeout. If no connection comes in this timeout, the server will shut down.
410    const IDLE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10 * 60);
411
412    let (incoming_tx, incoming_rx) = mpsc::unbounded::<Envelope>();
413    let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded::<Envelope>();
414    let (app_quit_tx, mut app_quit_rx) = mpsc::unbounded::<()>();
415
416    cx.on_app_quit(move |_| {
417        let mut app_quit_tx = app_quit_tx.clone();
418        async move {
419            log::info!("app quitting. sending signal to server main loop");
420            app_quit_tx.send(()).await.ok();
421        }
422    })
423    .detach();
424
425    cx.spawn(async move |cx| {
426        loop {
427            let streams = futures::future::join3(
428                listeners.stdin.accept(),
429                listeners.stdout.accept(),
430                listeners.stderr.accept(),
431            );
432
433            log::info!("accepting new connections");
434            let result = select! {
435                streams = streams.fuse() => {
436                    let (Ok((stdin_stream, _)), Ok((stdout_stream, _)), Ok((stderr_stream, _))) = streams else {
437                        log::error!("failed to accept new connections");
438                        break;
439                    };
440                    log::info!("accepted new connections");
441                    anyhow::Ok((stdin_stream, stdout_stream, stderr_stream))
442                }
443                _ = futures::FutureExt::fuse(cx.background_executor().timer(IDLE_TIMEOUT)) => {
444                    log::warn!("timed out waiting for new connections after {:?}. exiting.", IDLE_TIMEOUT);
445                    cx.update(|cx| {
446                        // TODO: This is a hack, because in a headless project, shutdown isn't executed
447                        // when calling quit, but it should be.
448                        cx.shutdown();
449                        cx.quit();
450                    });
451                    break;
452                }
453                _ = app_quit_rx.next().fuse() => {
454                    log::info!("app quit requested");
455                    break;
456                }
457            };
458
459            let Ok((mut stdin_stream, mut stdout_stream, mut stderr_stream)) = result else {
460                break;
461            };
462
463            let mut input_buffer = Vec::new();
464            let mut output_buffer = Vec::new();
465
466            let (mut stdin_msg_tx, mut stdin_msg_rx) = mpsc::unbounded::<Envelope>();
467            cx.background_spawn(async move {
468                loop {
469                    match read_message(&mut stdin_stream, &mut input_buffer).await {
470                        Ok(msg) => {
471                            if (stdin_msg_tx.send(msg).await).is_err() {
472                                log::info!("stdin message channel closed, stopping stdin reader");
473                                break;
474                            }
475                        }
476                        Err(error) => {
477                            log::warn!("stdin read failed: {error:?}");
478                            break;
479                        }
480                    }
481                }
482            }).detach();
483
484            loop {
485
486                select_biased! {
487                    _ = app_quit_rx.next().fuse() => {
488                        return anyhow::Ok(());
489                    }
490
491                    stdin_message = stdin_msg_rx.next().fuse() => {
492                        let Some(message) = stdin_message else {
493                            log::warn!("error reading message on stdin, dropping connection.");
494                            break;
495                        };
496                        if let Err(error) = incoming_tx.unbounded_send(message) {
497                            log::error!("failed to send message to application: {error:?}. exiting.");
498                            return Err(anyhow!(error));
499                        }
500                    }
501
502                    outgoing_message  = outgoing_rx.next().fuse() => {
503                        let Some(message) = outgoing_message else {
504                            log::error!("stdout handler, no message");
505                            break;
506                        };
507
508                        if let Err(error) =
509                            write_message(&mut stdout_stream, &mut output_buffer, message).await
510                        {
511                            log::error!("failed to write stdout message: {:?}", error);
512                            break;
513                        }
514                        if let Err(error) = stdout_stream.flush().await {
515                            log::error!("failed to flush stdout message: {:?}", error);
516                            break;
517                        }
518                    }
519
520                    log_message = log_rx.recv().fuse() => {
521                        if let Ok(log_message) = log_message {
522                            if let Err(error) = stderr_stream.write_all(&log_message).await {
523                                log::error!("failed to write log message to stderr: {:?}", error);
524                                break;
525                            }
526                            if let Err(error) = stderr_stream.flush().await {
527                                log::error!("failed to flush stderr stream: {:?}", error);
528                                break;
529                            }
530                        }
531                    }
532                }
533            }
534        }
535        anyhow::Ok(())
536    })
537    .detach();
538
539    RemoteClient::proto_client_from_channels(incoming_rx, outgoing_tx, cx, "server", is_wsl_interop)
540}
541
542fn init_paths() -> anyhow::Result<()> {
543    for path in [
544        paths::config_dir(),
545        paths::extensions_dir(),
546        paths::languages_dir(),
547        paths::logs_dir(),
548        paths::temp_dir(),
549        paths::hang_traces_dir(),
550        paths::remote_extensions_dir(),
551        paths::remote_extensions_uploads_dir(),
552    ]
553    .iter()
554    {
555        std::fs::create_dir_all(path).with_context(|| format!("creating directory {path:?}"))?;
556    }
557    Ok(())
558}
559
560pub fn execute_run(
561    log_file: PathBuf,
562    pid_file: PathBuf,
563    stdin_socket: PathBuf,
564    stdout_socket: PathBuf,
565    stderr_socket: PathBuf,
566) -> Result<()> {
567    init_paths()?;
568
569    let startup_time = Instant::now();
570    let app = gpui_platform::headless();
571    let pid = std::process::id();
572    let id = pid.to_string();
573    let should_install_crash_handler =
574        client::telemetry::should_install_crash_handler(*RELEASE_CHANNEL);
575
576    let crash_handler = if should_install_crash_handler {
577        Some(app.background_executor().spawn(crashes::init(
578            crashes::InitCrashHandler {
579                session_id: id,
580                zed_version: VERSION.to_owned(),
581                binary: "zed-remote-server".to_string(),
582                release_channel: release_channel::RELEASE_CHANNEL_NAME.clone(),
583                commit_sha: option_env!("ZED_COMMIT_SHA").unwrap_or("no_sha").to_owned(),
584            },
585            {
586                let background_executor = app.background_executor();
587                move |task| {
588                    background_executor.spawn(task).detach();
589                }
590            },
591            |pid| paths::temp_dir().join(format!("zed-remote-server-crash-handler-{pid}")),
592            // we are running outside gpui
593            #[allow(clippy::disallowed_methods)]
594            |duration| FutureExt::map(Timer::after(duration), |_| ()),
595        )))
596    } else {
597        crashes::force_backtrace();
598        None
599    };
600    let log_rx = init_logging_server(&log_file)?;
601    log::info!(
602        "starting up with PID {}:\npid_file: {:?}, log_file: {:?}, stdin_socket: {:?}, stdout_socket: {:?}, stderr_socket: {:?}",
603        pid,
604        pid_file,
605        log_file,
606        stdin_socket,
607        stdout_socket,
608        stderr_socket
609    );
610
611    write_pid_file(&pid_file, pid)
612        .with_context(|| format!("failed to write pid file: {:?}", &pid_file))?;
613
614    let listeners = ServerListeners::new(stdin_socket, stdout_socket, stderr_socket)?;
615
616    rayon::ThreadPoolBuilder::new()
617        .num_threads(std::thread::available_parallelism().map_or(1, |n| n.get().div_ceil(2)))
618        .stack_size(10 * 1024 * 1024)
619        .thread_name(|ix| format!("RayonWorker{}", ix))
620        .build_global()
621        .unwrap();
622
623    #[cfg(unix)]
624    let shell_env_loaded_rx = {
625        let (shell_env_loaded_tx, shell_env_loaded_rx) = oneshot::channel();
626        app.background_executor()
627            .spawn(async {
628                util::load_login_shell_environment().await.log_err();
629                shell_env_loaded_tx.send(()).ok();
630            })
631            .detach();
632        Some(shell_env_loaded_rx)
633    };
634    #[cfg(windows)]
635    let shell_env_loaded_rx: Option<oneshot::Receiver<()>> = None;
636
637    let git_hosting_provider_registry = Arc::new(GitHostingProviderRegistry::new());
638    let run = move |cx: &mut App| {
639        if let Some(crash_handler) = crash_handler {
640            cx.spawn(async move |_cx| {
641                let _crash_handler = crash_handler.await;
642                // cx.update(|cx| cx.set_global(CrashHandler(crash_handler)))
643            })
644            .detach();
645        }
646        settings::init(cx);
647        let app_commit_sha = option_env!("ZED_COMMIT_SHA").map(|s| AppCommitSha::new(s.to_owned()));
648        let app_version = AppVersion::load(
649            env!("ZED_PKG_VERSION"),
650            option_env!("ZED_BUILD_ID"),
651            app_commit_sha,
652        );
653        release_channel::init(app_version, cx);
654        gpui_tokio::init(cx);
655
656        HeadlessProject::init(cx);
657
658        let is_wsl_interop = if cfg!(target_os = "linux") {
659            // See: https://learn.microsoft.com/en-us/windows/wsl/filesystems#disable-interoperability
660            matches!(std::fs::read_to_string("/proc/sys/fs/binfmt_misc/WSLInterop").or_else(|_| std::fs::read_to_string("/proc/sys/fs/binfmt_misc/WSLInterop-late")), Ok(s) if s.contains("enabled"))
661        } else {
662            false
663        };
664
665        log::info!("gpui app started, initializing server");
666        let session = start_server(listeners, log_rx, cx, is_wsl_interop);
667        init_telemetry_forwarding(session.clone(), cx);
668        trusted_worktrees::init(HashMap::default(), cx);
669
670        GitHostingProviderRegistry::set_global(git_hosting_provider_registry, cx);
671        git_hosting_providers::init(cx);
672        dap_adapters::init(cx);
673
674        extension::init(cx);
675        let extension_host_proxy = ExtensionHostProxy::global(cx);
676
677        json_schema_store::init(cx);
678
679        let project = cx.new(|cx| {
680            let fs = Arc::new(RealFs::new(None, cx.background_executor().clone()));
681            let node_settings_rx = initialize_settings(session.clone(), fs.clone(), cx);
682
683            let proxy_url = read_proxy_settings(cx);
684
685            let http_client = {
686                let _guard = Tokio::handle(cx).enter();
687                Arc::new(
688                    ReqwestClient::proxy_and_user_agent(
689                        proxy_url,
690                        &format!(
691                            "Zed-Server/{} ({}; {})",
692                            env!("CARGO_PKG_VERSION"),
693                            std::env::consts::OS,
694                            std::env::consts::ARCH
695                        ),
696                    )
697                    .expect("Could not start HTTP client"),
698                )
699            };
700
701            let node_runtime =
702                NodeRuntime::new(http_client.clone(), shell_env_loaded_rx, node_settings_rx);
703
704            let mut languages = LanguageRegistry::new(cx.background_executor().clone());
705            languages.set_language_server_download_dir(paths::languages_dir().clone());
706            let languages = Arc::new(languages);
707
708            HeadlessProject::new(
709                HeadlessAppState {
710                    session: session.clone(),
711                    fs,
712                    http_client,
713                    node_runtime,
714                    languages,
715                    extension_host_proxy,
716                    startup_time,
717                },
718                true,
719                cx,
720            )
721        });
722
723        handle_crash_files_requests(&project, &session);
724
725        cx.background_spawn(async move {
726            cleanup_old_binaries_wsl();
727            cleanup_old_binaries()
728        })
729        .detach();
730
731        mem::forget(project);
732    };
733    // We do not reuse any of the state after unwinding, so we don't run risk of observing broken invariants.
734    let app = std::panic::AssertUnwindSafe(app);
735    let run = std::panic::AssertUnwindSafe(run);
736    let res = std::panic::catch_unwind(move || { app }.0.run({ run }.0));
737    if let Err(_) = res {
738        log::error!("app panicked. quitting.");
739        Err(anyhow::anyhow!("panicked"))
740    } else {
741        log::info!("gpui app is shut down. quitting.");
742        Ok(())
743    }
744}
745
746#[derive(Debug, Error)]
747pub enum ServerPathError {
748    #[error("Failed to create server_dir `{path}`")]
749    CreateServerDir {
750        #[source]
751        source: std::io::Error,
752        path: PathBuf,
753    },
754    #[error("Failed to create logs_dir `{path}`")]
755    CreateLogsDir {
756        #[source]
757        source: std::io::Error,
758        path: PathBuf,
759    },
760}
761
762#[derive(Clone, Debug)]
763struct ServerPaths {
764    log_file: PathBuf,
765    pid_file: PathBuf,
766    stdin_socket: PathBuf,
767    stdout_socket: PathBuf,
768    stderr_socket: PathBuf,
769}
770
771impl ServerPaths {
772    fn new(identifier: &str) -> Result<Self, ServerPathError> {
773        let server_dir = paths::remote_server_state_dir().join(identifier);
774        std::fs::create_dir_all(&server_dir).map_err(|source| {
775            ServerPathError::CreateServerDir {
776                source,
777                path: server_dir.clone(),
778            }
779        })?;
780        let log_dir = logs_dir();
781        std::fs::create_dir_all(log_dir).map_err(|source| ServerPathError::CreateLogsDir {
782            source,
783            path: log_dir.clone(),
784        })?;
785
786        let pid_file = server_dir.join("server.pid");
787        let stdin_socket = server_dir.join("stdin.sock");
788        let stdout_socket = server_dir.join("stdout.sock");
789        let stderr_socket = server_dir.join("stderr.sock");
790        let log_file = logs_dir().join(format!("server-{}.log", identifier));
791
792        Ok(Self {
793            pid_file,
794            stdin_socket,
795            stdout_socket,
796            stderr_socket,
797            log_file,
798        })
799    }
800}
801
802#[derive(Debug, Error)]
803pub enum ExecuteProxyError {
804    #[error("Failed to init server paths: {0:#}")]
805    ServerPath(#[from] ServerPathError),
806
807    #[error(transparent)]
808    ServerNotRunning(#[from] ProxyLaunchError),
809
810    #[error("Failed to check PidFile '{path}': {source:#}")]
811    CheckPidFile {
812        #[source]
813        source: CheckPidError,
814        path: PathBuf,
815    },
816
817    #[error("Failed to kill existing server with pid '{pid}'")]
818    KillRunningServer { pid: u32 },
819
820    #[error("failed to spawn server")]
821    SpawnServer(#[source] SpawnServerError),
822
823    #[error("stdin_task failed: {0:#}")]
824    StdinTask(#[source] anyhow::Error),
825    #[error("stdout_task failed: {0:#}")]
826    StdoutTask(#[source] anyhow::Error),
827    #[error("stderr_task failed: {0:#}")]
828    StderrTask(#[source] anyhow::Error),
829}
830
831impl ExecuteProxyError {
832    pub fn to_exit_code(&self) -> i32 {
833        match self {
834            ExecuteProxyError::ServerNotRunning(proxy_launch_error) => {
835                proxy_launch_error.to_exit_code()
836            }
837            _ => 1,
838        }
839    }
840}
841
842pub(crate) fn execute_proxy(
843    identifier: String,
844    is_reconnecting: bool,
845) -> Result<(), ExecuteProxyError> {
846    init_logging_proxy();
847
848    let server_paths = ServerPaths::new(&identifier)?;
849
850    let id = std::process::id().to_string();
851    let should_install_crash_handler =
852        client::telemetry::should_install_crash_handler(*RELEASE_CHANNEL);
853
854    if should_install_crash_handler {
855        smol::spawn(crashes::init(
856            crashes::InitCrashHandler {
857                session_id: id,
858                zed_version: VERSION.to_owned(),
859                binary: "zed-remote-proxy".to_string(),
860                release_channel: release_channel::RELEASE_CHANNEL_NAME.clone(),
861                commit_sha: option_env!("ZED_COMMIT_SHA").unwrap_or("no_sha").to_owned(),
862            },
863            |task| {
864                smol::spawn(task).detach();
865            },
866            |pid| paths::temp_dir().join(format!("zed-remote-server-proxy-crash-handler-{pid}")),
867            // we are running outside gpui
868            #[allow(clippy::disallowed_methods)]
869            |duration| FutureExt::map(Timer::after(duration), |_| ()),
870        ))
871        .detach();
872    };
873    log::info!("starting proxy process. PID: {}", std::process::id());
874    let server_pid = {
875        let server_pid = check_pid_file(&server_paths.pid_file).map_err(|source| {
876            ExecuteProxyError::CheckPidFile {
877                source,
878                path: server_paths.pid_file.clone(),
879            }
880        })?;
881        if is_reconnecting {
882            match server_pid {
883                None => {
884                    log::error!("attempted to reconnect, but no server running");
885                    return Err(ExecuteProxyError::ServerNotRunning(
886                        ProxyLaunchError::ServerNotRunning,
887                    ));
888                }
889                Some(server_pid) => server_pid,
890            }
891        } else {
892            if let Some(pid) = server_pid {
893                log::info!(
894                    "proxy found server already running with PID {}. Killing process and cleaning up files...",
895                    pid
896                );
897                kill_running_server(pid, &server_paths)?;
898            }
899            gpui::block_on(spawn_server(&server_paths)).map_err(ExecuteProxyError::SpawnServer)?;
900            std::fs::read_to_string(&server_paths.pid_file)
901                .and_then(|contents| {
902                    contents.parse::<u32>().map_err(|_| {
903                        std::io::Error::new(
904                            std::io::ErrorKind::InvalidData,
905                            "Invalid PID file contents",
906                        )
907                    })
908                })
909                .map_err(SpawnServerError::ProcessStatus)
910                .map_err(ExecuteProxyError::SpawnServer)?
911        }
912    };
913
914    let stdin_task = smol::spawn(async move {
915        let stdin = smol::Unblock::new(std::io::stdin());
916        let stream = UnixStream::connect(&server_paths.stdin_socket)
917            .await
918            .with_context(|| {
919                format!(
920                    "Failed to connect to stdin socket {}",
921                    server_paths.stdin_socket.display()
922                )
923            })?;
924        handle_io(stdin, stream, "stdin").await
925    });
926
927    let stdout_task: smol::Task<Result<()>> = smol::spawn(async move {
928        let stdout = smol::Unblock::new(std::io::stdout());
929        let stream = UnixStream::connect(&server_paths.stdout_socket)
930            .await
931            .with_context(|| {
932                format!(
933                    "Failed to connect to stdout socket {}",
934                    server_paths.stdout_socket.display()
935                )
936            })?;
937        handle_io(stream, stdout, "stdout").await
938    });
939
940    let stderr_task: smol::Task<Result<()>> = smol::spawn(async move {
941        let mut stderr = smol::Unblock::new(std::io::stderr());
942        let mut stream = UnixStream::connect(&server_paths.stderr_socket)
943            .await
944            .with_context(|| {
945                format!(
946                    "Failed to connect to stderr socket {}",
947                    server_paths.stderr_socket.display()
948                )
949            })?;
950        let mut stderr_buffer = vec![0; 2048];
951        loop {
952            match stream
953                .read(&mut stderr_buffer)
954                .await
955                .context("reading stderr")?
956            {
957                0 => {
958                    let error =
959                        std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "stderr closed");
960                    Err(anyhow!(error))?;
961                }
962                n => {
963                    stderr.write_all(&stderr_buffer[..n]).await?;
964                    stderr.flush().await?;
965                }
966            }
967        }
968    });
969
970    if let Err(forwarding_result) = gpui::block_on(async move {
971        futures::select! {
972            result = stdin_task.fuse() => result.map_err(ExecuteProxyError::StdinTask),
973            result = stdout_task.fuse() => result.map_err(ExecuteProxyError::StdoutTask),
974            result = stderr_task.fuse() => result.map_err(ExecuteProxyError::StderrTask),
975        }
976    }) {
977        log::error!("encountered error while forwarding messages: {forwarding_result:#}",);
978        if !matches!(gpui::block_on(check_server_running(server_pid)), Ok(true)) {
979            log::error!("server exited unexpectedly");
980            return Err(ExecuteProxyError::ServerNotRunning(
981                ProxyLaunchError::ServerNotRunning,
982            ));
983        }
984        return Err(forwarding_result);
985    }
986
987    Ok(())
988}
989
990fn kill_running_server(pid: u32, paths: &ServerPaths) -> Result<(), ExecuteProxyError> {
991    log::info!("killing existing server with PID {}", pid);
992    let system = sysinfo::System::new_with_specifics(
993        sysinfo::RefreshKind::nothing().with_processes(sysinfo::ProcessRefreshKind::nothing()),
994    );
995
996    if let Some(process) = system.process(sysinfo::Pid::from_u32(pid)) {
997        let killed = process.kill();
998        if !killed {
999            return Err(ExecuteProxyError::KillRunningServer { pid });
1000        }
1001    }
1002
1003    for file in [
1004        &paths.pid_file,
1005        &paths.stdin_socket,
1006        &paths.stdout_socket,
1007        &paths.stderr_socket,
1008    ] {
1009        log::debug!("cleaning up file {:?} before starting new server", file);
1010        std::fs::remove_file(file).ok();
1011    }
1012
1013    Ok(())
1014}
1015
1016#[derive(Debug, Error)]
1017pub enum SpawnServerError {
1018    #[error("failed to remove stdin socket")]
1019    RemoveStdinSocket(#[source] std::io::Error),
1020
1021    #[error("failed to remove stdout socket")]
1022    RemoveStdoutSocket(#[source] std::io::Error),
1023
1024    #[error("failed to remove stderr socket")]
1025    RemoveStderrSocket(#[source] std::io::Error),
1026
1027    #[error("failed to get current_exe")]
1028    CurrentExe(#[source] std::io::Error),
1029
1030    #[error("failed to launch server process")]
1031    ProcessStatus(#[source] std::io::Error),
1032
1033    #[error("failed to wait for server to be ready to accept connections")]
1034    Timeout,
1035}
1036
1037async fn spawn_server(paths: &ServerPaths) -> Result<(), SpawnServerError> {
1038    log::info!("spawning server process",);
1039    if paths.stdin_socket.exists() {
1040        std::fs::remove_file(&paths.stdin_socket).map_err(SpawnServerError::RemoveStdinSocket)?;
1041    }
1042    if paths.stdout_socket.exists() {
1043        std::fs::remove_file(&paths.stdout_socket).map_err(SpawnServerError::RemoveStdoutSocket)?;
1044    }
1045    if paths.stderr_socket.exists() {
1046        std::fs::remove_file(&paths.stderr_socket).map_err(SpawnServerError::RemoveStderrSocket)?;
1047    }
1048
1049    let binary_name = std::env::current_exe().map_err(SpawnServerError::CurrentExe)?;
1050
1051    #[cfg(windows)]
1052    {
1053        spawn_server_windows(&binary_name, paths)?;
1054    }
1055
1056    #[cfg(not(windows))]
1057    {
1058        spawn_server_normal(&binary_name, paths)?;
1059    }
1060
1061    let mut total_time_waited = std::time::Duration::from_secs(0);
1062    let wait_duration = std::time::Duration::from_millis(20);
1063    while !paths.stdout_socket.exists()
1064        || !paths.stdin_socket.exists()
1065        || !paths.stderr_socket.exists()
1066    {
1067        log::debug!("waiting for server to be ready to accept connections...");
1068        std::thread::sleep(wait_duration);
1069        total_time_waited += wait_duration;
1070        if total_time_waited > std::time::Duration::from_secs(10) {
1071            return Err(SpawnServerError::Timeout);
1072        }
1073    }
1074
1075    log::info!(
1076        "server ready to accept connections. total time waited: {:?}",
1077        total_time_waited
1078    );
1079
1080    Ok(())
1081}
1082
1083#[cfg(windows)]
1084fn spawn_server_windows(binary_name: &Path, paths: &ServerPaths) -> Result<(), SpawnServerError> {
1085    let binary_path = binary_name.to_string_lossy().to_string();
1086    let parameters = format!(
1087        "run --log-file \"{}\" --pid-file \"{}\" --stdin-socket \"{}\" --stdout-socket \"{}\" --stderr-socket \"{}\"",
1088        paths.log_file.to_string_lossy(),
1089        paths.pid_file.to_string_lossy(),
1090        paths.stdin_socket.to_string_lossy(),
1091        paths.stdout_socket.to_string_lossy(),
1092        paths.stderr_socket.to_string_lossy()
1093    );
1094
1095    let directory = binary_name
1096        .parent()
1097        .map(|p| p.to_string_lossy().to_string())
1098        .unwrap_or_default();
1099
1100    crate::windows::shell_execute_from_explorer(&binary_path, &parameters, &directory)
1101        .map_err(|e| SpawnServerError::ProcessStatus(std::io::Error::other(e)))?;
1102
1103    Ok(())
1104}
1105
1106#[cfg(not(windows))]
1107fn spawn_server_normal(binary_name: &Path, paths: &ServerPaths) -> Result<(), SpawnServerError> {
1108    let mut server_process = new_command(binary_name);
1109    server_process
1110        .stdin(util::command::Stdio::null())
1111        .stdout(util::command::Stdio::null())
1112        .stderr(util::command::Stdio::null())
1113        .arg("run")
1114        .arg("--log-file")
1115        .arg(&paths.log_file)
1116        .arg("--pid-file")
1117        .arg(&paths.pid_file)
1118        .arg("--stdin-socket")
1119        .arg(&paths.stdin_socket)
1120        .arg("--stdout-socket")
1121        .arg(&paths.stdout_socket)
1122        .arg("--stderr-socket")
1123        .arg(&paths.stderr_socket);
1124
1125    server_process
1126        .spawn()
1127        .map_err(SpawnServerError::ProcessStatus)?;
1128
1129    Ok(())
1130}
1131
1132#[derive(Debug, Error)]
1133#[error("Failed to remove PID file for missing process (pid `{pid}`")]
1134pub struct CheckPidError {
1135    #[source]
1136    source: std::io::Error,
1137    pid: u32,
1138}
1139async fn check_server_running(pid: u32) -> std::io::Result<bool> {
1140    new_command("kill")
1141        .arg("-0")
1142        .arg(pid.to_string())
1143        .output()
1144        .await
1145        .map(|output| output.status.success())
1146}
1147
1148fn check_pid_file(path: &Path) -> Result<Option<u32>, CheckPidError> {
1149    let Some(pid) = std::fs::read_to_string(path)
1150        .ok()
1151        .and_then(|contents| contents.parse::<u32>().ok())
1152    else {
1153        return Ok(None);
1154    };
1155
1156    log::debug!("Checking if process with PID {} exists...", pid);
1157
1158    let system = sysinfo::System::new_with_specifics(
1159        sysinfo::RefreshKind::nothing().with_processes(sysinfo::ProcessRefreshKind::nothing()),
1160    );
1161
1162    if system.process(sysinfo::Pid::from_u32(pid)).is_some() {
1163        log::debug!(
1164            "Process with PID {} exists. NOT spawning new server, but attaching to existing one.",
1165            pid
1166        );
1167        Ok(Some(pid))
1168    } else {
1169        log::debug!("Found PID file, but process with that PID does not exist. Removing PID file.");
1170        std::fs::remove_file(path).map_err(|source| CheckPidError { source, pid })?;
1171        Ok(None)
1172    }
1173}
1174
1175fn write_pid_file(path: &Path, pid: u32) -> Result<()> {
1176    if path.exists() {
1177        std::fs::remove_file(path)?;
1178    }
1179    log::debug!("writing PID {} to file {:?}", pid, path);
1180    std::fs::write(path, pid.to_string()).context("Failed to write PID file")
1181}
1182
1183async fn handle_io<R, W>(mut reader: R, mut writer: W, socket_name: &str) -> Result<()>
1184where
1185    R: AsyncRead + Unpin,
1186    W: AsyncWrite + Unpin,
1187{
1188    use remote::protocol::{read_message_raw, write_size_prefixed_buffer};
1189
1190    let mut buffer = Vec::new();
1191    loop {
1192        read_message_raw(&mut reader, &mut buffer)
1193            .await
1194            .with_context(|| format!("failed to read message from {}", socket_name))?;
1195        write_size_prefixed_buffer(&mut writer, &mut buffer)
1196            .await
1197            .with_context(|| format!("failed to write message to {}", socket_name))?;
1198        writer.flush().await?;
1199        buffer.clear();
1200    }
1201}
1202
1203fn initialize_settings(
1204    session: AnyProtoClient,
1205    fs: Arc<dyn Fs>,
1206    cx: &mut App,
1207) -> watch::Receiver<Option<NodeBinaryOptions>> {
1208    let (user_settings_file_rx, watcher_task) =
1209        watch_config_file(cx.background_executor(), fs, paths::settings_file().clone());
1210
1211    handle_settings_file_changes(user_settings_file_rx, watcher_task, cx, {
1212        move |err, _cx| {
1213            if let Some(e) = err {
1214                log::info!("Server settings failed to change: {}", e);
1215
1216                session
1217                    .send(proto::Toast {
1218                        project_id: REMOTE_SERVER_PROJECT_ID,
1219                        notification_id: "server-settings-failed".to_string(),
1220                        message: format!(
1221                            "Error in settings on remote host {:?}: {}",
1222                            paths::settings_file(),
1223                            e
1224                        ),
1225                    })
1226                    .log_err();
1227            } else {
1228                session
1229                    .send(proto::HideToast {
1230                        project_id: REMOTE_SERVER_PROJECT_ID,
1231                        notification_id: "server-settings-failed".to_string(),
1232                    })
1233                    .log_err();
1234            }
1235        }
1236    });
1237
1238    let (mut tx, rx) = watch::channel(None);
1239    let mut node_settings = None;
1240    cx.observe_global::<SettingsStore>(move |cx| {
1241        let new_node_settings = &ProjectSettings::get_global(cx).node;
1242        if Some(new_node_settings) != node_settings.as_ref() {
1243            log::info!("Got new node settings: {new_node_settings:?}");
1244            let options = NodeBinaryOptions {
1245                allow_path_lookup: !new_node_settings.ignore_system_version,
1246                // TODO: Implement this setting
1247                allow_binary_download: true,
1248                use_paths: new_node_settings.path.as_ref().map(|node_path| {
1249                    let node_path = PathBuf::from(shellexpand::tilde(node_path).as_ref());
1250                    let npm_path = new_node_settings
1251                        .npm_path
1252                        .as_ref()
1253                        .map(|path| PathBuf::from(shellexpand::tilde(&path).as_ref()));
1254                    (
1255                        node_path.clone(),
1256                        npm_path.unwrap_or_else(|| {
1257                            let base_path = PathBuf::new();
1258                            node_path.parent().unwrap_or(&base_path).join("npm")
1259                        }),
1260                    )
1261                }),
1262            };
1263            node_settings = Some(new_node_settings.clone());
1264            tx.send(Some(options)).ok();
1265        }
1266    })
1267    .detach();
1268
1269    rx
1270}
1271
1272pub fn handle_settings_file_changes(
1273    mut server_settings_file: mpsc::UnboundedReceiver<String>,
1274    watcher_task: gpui::Task<()>,
1275    cx: &mut App,
1276    settings_changed: impl Fn(Option<anyhow::Error>, &mut App) + 'static,
1277) {
1278    let server_settings_content = cx
1279        .foreground_executor()
1280        .block_on(server_settings_file.next())
1281        .unwrap();
1282    SettingsStore::update_global(cx, |store, cx| {
1283        store
1284            .set_server_settings(&server_settings_content, cx)
1285            .log_err();
1286    });
1287    cx.spawn(async move |cx| {
1288        let _watcher_task = watcher_task;
1289        while let Some(server_settings_content) = server_settings_file.next().await {
1290            cx.update_global(|store: &mut SettingsStore, cx| {
1291                let result = store.set_server_settings(&server_settings_content, cx);
1292                if let Err(err) = &result {
1293                    log::error!("Failed to load server settings: {err}");
1294                }
1295                settings_changed(result.err(), cx);
1296                cx.refresh_windows();
1297            });
1298        }
1299    })
1300    .detach();
1301}
1302
1303fn read_proxy_settings(cx: &mut Context<HeadlessProject>) -> Option<Url> {
1304    let proxy_str = ProxySettings::get_global(cx).proxy.to_owned();
1305
1306    proxy_str
1307        .as_deref()
1308        .map(str::trim)
1309        .filter(|input| !input.is_empty())
1310        .and_then(|input| {
1311            input
1312                .parse::<Url>()
1313                .inspect_err(|e| log::error!("Error parsing proxy settings: {}", e))
1314                .ok()
1315        })
1316        .or_else(read_proxy_from_env)
1317}
1318
1319fn cleanup_old_binaries() -> Result<()> {
1320    let server_dir = paths::remote_server_dir_relative();
1321    let release_channel = release_channel::RELEASE_CHANNEL.dev_name();
1322    let prefix = format!("zed-remote-server-{}-", release_channel);
1323
1324    for entry in std::fs::read_dir(server_dir.as_std_path())? {
1325        let path = entry?.path();
1326
1327        if let Some(file_name) = path.file_name()
1328            && let Some(version) = file_name.to_string_lossy().strip_prefix(&prefix)
1329            && !is_new_version(version)
1330            && !is_file_in_use(file_name)
1331        {
1332            log::info!("removing old remote server binary: {:?}", path);
1333            std::fs::remove_file(&path)?;
1334        }
1335    }
1336
1337    Ok(())
1338}
1339
1340// Remove this once 223 goes stable, we only have this to clean up old binaries on WSL
1341// we no longer download them into this folder, we use the same folder as other remote servers
1342fn cleanup_old_binaries_wsl() {
1343    let server_dir = paths::remote_wsl_server_dir_relative();
1344    if let Ok(()) = std::fs::remove_dir_all(server_dir.as_std_path()) {
1345        log::info!("removing old wsl remote server folder: {:?}", server_dir);
1346    }
1347}
1348
1349fn is_new_version(version: &str) -> bool {
1350    semver::Version::from_str(version)
1351        .ok()
1352        .zip(semver::Version::from_str(env!("ZED_PKG_VERSION")).ok())
1353        .is_some_and(|(version, current_version)| version >= current_version)
1354}
1355
1356fn is_file_in_use(file_name: &OsStr) -> bool {
1357    let info = sysinfo::System::new_with_specifics(sysinfo::RefreshKind::nothing().with_processes(
1358        sysinfo::ProcessRefreshKind::nothing().with_exe(sysinfo::UpdateKind::Always),
1359    ));
1360
1361    for process in info.processes().values() {
1362        if process
1363            .exe()
1364            .is_some_and(|exe| exe.file_name().is_some_and(|name| name == file_name))
1365        {
1366            return true;
1367        }
1368    }
1369
1370    false
1371}
1372
1373#[cfg(test)]
1374mod tests {
1375    use super::*;
1376
1377    #[test]
1378    fn rotated_remote_log_path_uses_numbered_log_suffix() {
1379        assert_eq!(
1380            rotated_log_path(Path::new("server-workspace-12.log")),
1381            PathBuf::from("server-workspace-12.1.log")
1382        );
1383    }
1384
1385    #[test]
1386    fn opening_remote_log_rotates_existing_oversized_log() {
1387        let temp_dir = tempfile::tempdir().expect("create temp dir");
1388        let log_path = temp_dir.path().join("server-workspace-12.log");
1389        let rotated_path = temp_dir.path().join("server-workspace-12.1.log");
1390        let existing_contents = vec![b'x'; REMOTE_SERVER_LOG_MAX_BYTES as usize];
1391        std::fs::write(&log_path, &existing_contents).expect("write oversized log");
1392
1393        let _log_file = RotatingLogFile::open(&log_path).expect("open rotating log file");
1394
1395        assert_eq!(
1396            std::fs::read(&rotated_path).expect("read rotated log"),
1397            existing_contents
1398        );
1399        assert_eq!(
1400            std::fs::metadata(&log_path)
1401                .expect("active log metadata")
1402                .len(),
1403            0
1404        );
1405    }
1406
1407    #[test]
1408    fn writing_remote_log_rotates_before_exceeding_size_limit() {
1409        let temp_dir = tempfile::tempdir().expect("create temp dir");
1410        let log_path = temp_dir.path().join("server-workspace-12.log");
1411        let rotated_path = temp_dir.path().join("server-workspace-12.1.log");
1412        let existing_contents = vec![b'x'; REMOTE_SERVER_LOG_MAX_BYTES as usize - 1];
1413        let new_contents = b"yz";
1414        std::fs::write(&log_path, &existing_contents).expect("write existing log contents");
1415        let mut log_file = RotatingLogFile::open(&log_path).expect("open rotating log file");
1416
1417        log_file
1418            .write_all(new_contents)
1419            .expect("write log contents");
1420        log_file.flush().expect("flush log file");
1421
1422        assert_eq!(
1423            std::fs::read(&rotated_path).expect("read rotated log"),
1424            existing_contents
1425        );
1426        assert_eq!(
1427            std::fs::read(&log_path).expect("read active log"),
1428            new_contents
1429        );
1430    }
1431}
1432