diff --git a/nebula_app/src/atomic_file.rs b/nebula_app/src/atomic_file.rs index 8e8e35c..6c1dc11 100644 --- a/nebula_app/src/atomic_file.rs +++ b/nebula_app/src/atomic_file.rs @@ -4,7 +4,7 @@ //! so a crash cannot leave a half-written JSON document. A best-effort lock //! prevents two Nebula processes from compacting the same store at once. -use std::fs::OpenOptions; +use std::fs::{File, OpenOptions}; use std::io::{self, Write}; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; @@ -64,6 +64,50 @@ pub(crate) fn try_lock(path: &Path) -> io::Result> { } } +/// Try to own a lock for the full lifetime of a resident process. +/// +/// 与短状态写锁不同，这里依赖 OS 句柄锁：进程崩溃时内核立即释放，健康进程 +/// 无论运行多久都不会被“超时回收”。锁文件本身可以保留，所有权只看句柄。 +pub(crate) fn try_lifetime_lock(path: &Path) -> io::Result> { + let lock_path = lock_path(path); + if let Some(parent) = lock_path.parent() { + std::fs::create_dir_all(parent)?; + } + + #[cfg(windows)] + let mut file = { + use std::os::windows::fs::OpenOptionsExt as _; + + match OpenOptions::new().read(true).write(true).create(true).share_mode(0).open(&lock_path) + { + Ok(file) => file, + Err(error) if matches!(error.raw_os_error(), Some(32) | Some(33)) => return Ok(None), + Err(error) => return Err(error), + } + }; + + #[cfg(unix)] + let mut file = { + use std::os::fd::AsRawFd as _; + + let file = OpenOptions::new().read(true).write(true).create(true).open(&lock_path)?; + let result = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) }; + if result != 0 { + let error = io::Error::last_os_error(); + if error.kind() == io::ErrorKind::WouldBlock { + return Ok(None); + } + return Err(error); + } + file + }; + + file.set_len(0)?; + write!(file, "{}", std::process::id())?; + file.sync_data()?; + Ok(Some(LifetimeFileLock { file })) +} + fn create_lock(path: &Path) -> io::Result { if let Some(parent) = path.parent() { std::fs::create_dir_all(parent)?; @@ -99,6 +143,20 @@ impl Drop for FileLock { } } +pub(crate) struct LifetimeFileLock { + file: File, +} + +impl Drop for LifetimeFileLock { + fn drop(&mut self) { + #[cfg(unix)] + { + use std::os::fd::AsRawFd as _; + let _ = unsafe { libc::flock(self.file.as_raw_fd(), libc::LOCK_UN) }; + } + } +} + #[cfg(windows)] fn replace(source: &Path, destination: &Path) -> io::Result<()> { use std::os::windows::ffi::OsStrExt; @@ -131,7 +189,7 @@ fn replace(source: &Path, destination: &Path) -> io::Result<()> { #[cfg(test)] mod tests { - use super::{try_lock, write}; + use super::{try_lifetime_lock, try_lock, write}; #[test] fn atomic_write_replaces_and_lock_is_exclusive() { @@ -145,5 +203,10 @@ mod tests { assert!(try_lock(&path).unwrap().is_none()); drop(first); assert!(try_lock(&path).unwrap().is_some()); + + let owner = try_lifetime_lock(&path).unwrap().expect("first lifetime lock"); + assert!(try_lifetime_lock(&path).unwrap().is_none()); + drop(owner); + assert!(try_lifetime_lock(&path).unwrap().is_some()); } } diff --git a/nebula_app/src/gpui_shell/mod.rs b/nebula_app/src/gpui_shell/mod.rs index 2aafd79..bcaffbd 100644 --- a/nebula_app/src/gpui_shell/mod.rs +++ b/nebula_app/src/gpui_shell/mod.rs @@ -15,9 +15,9 @@ pub mod config; pub mod doc_tabs; pub mod http; pub mod math_view; +pub mod network_settings; pub mod prelude; pub mod session_restore; -pub mod network_settings; pub mod settings_pane; pub mod ssh_hosts; pub mod ssh_settings; @@ -64,7 +64,7 @@ pub fn run_shell() { }); // 驻留进程写 runtime.port：第二份 `nebula --gpui` 经 ATTACH + tab.new // 并入当前窗口，而不是再拉一套进程。Drop 会删 port 文件。 - let _runtime_server = crate::runtime_api::RuntimeServer::spawn_callback( + let runtime_server = crate::runtime_api::RuntimeServer::spawn_callback( { let tx = shell_tx; move |callback| { @@ -79,6 +79,18 @@ pub fn run_shell() { }, runtime_hub.clone(), ); + if runtime_server.is_none() { + // 与另一份进程同时启动时，对方可能已经拿到 owner lock、但尚未来得及 + // 发布 endpoint。短暂等待并交接，避免继续打开一个无控制面的空窗口。 + for _ in 0..40 { + if crate::runtime_api::try_open_default_tab_existing() { + crate::tray::shutdown(); + return; + } + std::thread::sleep(std::time::Duration::from_millis(25)); + } + } + let _runtime_server = runtime_server; gpui::Application::new().with_assets(Assets).run(move |cx| { // GPUI is the only event loop in `--gpui` mode, so it owns the same // per-process hook pipe before the first TerminalView spawns. @@ -173,9 +185,8 @@ fn open_main_window( }; cx.open_window(options, move |window, cx| { - let workspace = cx.new(|cx| { - NebulaWorkspace::new(window, ai_events, shell_events, 1, runtime_hub, cx) - }); + let workspace = + cx.new(|cx| NebulaWorkspace::new(window, ai_events, shell_events, 1, runtime_hub, cx)); // 调试/验收后门：启动即打开指定文档（见 open_document_at_startup）。 if let Ok(path) = std::env::var("NEBULA_GPUI_OPEN_DOC") && !path.is_empty() diff --git a/nebula_app/src/input/terminal_input.rs b/nebula_app/src/input/terminal_input.rs index a7cabcd..2645f26 100644 --- a/nebula_app/src/input/terminal_input.rs +++ b/nebula_app/src/input/terminal_input.rs @@ -282,33 +282,10 @@ pub(crate) fn build_runtime_sequence( result } -/// Encode text injected by the Runtime API for the pane's active input protocol. -/// -/// ConPTY's Win32 input mode deserializes `CSI Vk;Sc;Uc;Kd;Cs;Rc_` into -/// `KEY_EVENT_RECORD`s. Mixing raw UTF-8 text and a serialized Enter record in -/// one pipe batch can leave the control record in an interactive TUI's text -/// input batch. Encode the text as `VK_PACKET` records as well, so a submitted -/// prompt is one homogeneous protocol stream and needs no timing boundary. -pub(crate) fn build_runtime_text_sequence(text: &str, mode: TermMode) -> Vec { - if use_win32_input_mode(mode) { - #[cfg(target_os = "windows")] - { - use windows_sys::Win32::UI::Input::KeyboardAndMouse::VK_PACKET; - - let mut sequence = String::with_capacity(text.len().saturating_mul(40)); - for unicode_char in text.encode_utf16() { - for key_down in [1, 0] { - // VK_PACKET is the documented virtual key for Unicode - // supplied by a non-keyboard input source. A zero scan - // code and the UTF-16 unit preserve arbitrary prompt text. - let _ = - write!(sequence, "\x1b[{};0;{};{key_down};0;1_", VK_PACKET, unicode_char); - } - } - return sequence.into_bytes(); - } - } - +/// Runtime 文本始终走终端的原生文本路径；ConPTY 会在 `WriteString` 中按当前 +/// Console/键盘布局生成事件。强制伪造成 `VK_PACKET` 会被旧 PSReadLine 丢弃。 +/// Enter 等控制键由 submit barrier 在下一批按当前协议单独编码。 +pub(crate) fn build_runtime_text_sequence(text: &str, _mode: TermMode) -> Vec { text.as_bytes().to_vec() } @@ -1122,28 +1099,14 @@ mod vt_tests { #[cfg(target_os = "windows")] #[test] - fn runtime_win32_text_and_enter_form_one_record_stream() { - use windows_sys::Win32::UI::Input::KeyboardAndMouse::VK_PACKET; - + fn runtime_win32_text_uses_conpty_native_text_path() { let mode = TermMode::WIN32_INPUT_MODE; - let mut bytes = build_runtime_text_sequence("A中", mode); - bytes.extend(build_runtime_sequence( - RuntimeKey::Enter, - RuntimeKeyModifiers::default(), - 1, - mode, - )); + assert_eq!(build_runtime_text_sequence("A中", mode), "A中".as_bytes()); - let text = String::from_utf8(bytes).unwrap(); - let records: Vec<_> = text.split_inclusive('_').collect(); - assert_eq!(records.len(), 6); - assert_eq!(records[0], format!("\x1b[{VK_PACKET};0;65;1;0;1_")); - assert_eq!(records[1], format!("\x1b[{VK_PACKET};0;65;0;0;1_")); - assert_eq!(records[2], format!("\x1b[{VK_PACKET};0;20013;1;0;1_")); - assert_eq!(records[3], format!("\x1b[{VK_PACKET};0;20013;0;0;1_")); - assert!(records[4].starts_with("\x1b[13;")); - assert!(records[4].contains(";13;1;0;1_")); - assert!(records[5].contains(";13;0;0;1_")); + let enter = + build_runtime_sequence(RuntimeKey::Enter, RuntimeKeyModifiers::default(), 1, mode); + assert_ne!(enter, b"\r"); + assert!(String::from_utf8(enter).unwrap().starts_with("\x1b[13;")); } #[test] diff --git a/nebula_app/src/runtime_api.rs b/nebula_app/src/runtime_api.rs index 1313e60..c21e13e 100644 --- a/nebula_app/src/runtime_api.rs +++ b/nebula_app/src/runtime_api.rs @@ -8,12 +8,22 @@ mod agent_api; mod cli; mod command; +mod server; #[cfg(test)] mod tests; pub use cli::run_cli; -pub(crate) use command::{capture_process_tree, capture_terminal_tail, validate_command_line, validate_prompt}; -use command::{RuntimeWaitState, SubscribeParams, WaitParams, default_read_lines, default_true, parse_params, validate_agent_name, validate_agent_selector, wait_matches}; +use command::{ + RuntimeWaitState, SubscribeParams, WaitParams, default_read_lines, default_true, parse_params, + validate_agent_name, validate_agent_selector, wait_matches, +}; +pub(crate) use command::{ + capture_process_tree, capture_terminal_tail, validate_command_line, validate_prompt, +}; +use server::{Endpoint, endpoint_addr, read_endpoint}; +pub use server::{ + RuntimeServer, dispatch_prompt, try_open_default_tab_existing, try_open_directory_existing, +}; use std::error::Error; use std::fmt; @@ -1379,164 +1389,6 @@ fn run_result_or_capability_error(result: RuntimeRunResult) -> Result PathBuf { - crate::display::nebula_data_dir().join("runtime.port") -} - -fn legacy_port_file() -> PathBuf { - crate::display::nebula_data_dir().join("mux.port") -} - -fn read_endpoint() -> Option { - read_endpoint_from(port_file()) -} - -fn read_endpoint_from(path: PathBuf) -> Option { - let data = std::fs::read_to_string(path).ok()?; - let mut parts = data.split_whitespace(); - Some(Endpoint { port: parts.next()?.parse().ok()?, token: parts.next()?.to_owned() }) -} - -fn endpoint_addr(endpoint: &Endpoint) -> SocketAddr { - SocketAddr::from((Ipv4Addr::LOCALHOST, endpoint.port)) -} - -fn fresh_token() -> String { - use std::hash::{BuildHasher, Hasher, RandomState}; - let mut a = RandomState::new().build_hasher(); - let mut b = RandomState::new().build_hasher(); - a.write_u32(std::process::id()); - b.write_u128( - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|duration| duration.as_nanos()) - .unwrap_or(0), - ); - format!("{:016x}{:016x}", a.finish(), b.finish()) -} - -/// 普通二次启动并入驻留实例：先恢复/聚焦窗口，再新建一个默认 shell 标签页。 -pub fn try_open_default_tab_existing() -> bool { - try_open_tab_existing(None) -} - -/// 后台任务把一行文本作为输入敲进某个 pane（不回车）。复用 runtime Prompt -/// 的窗口定位与写入路径（`window_id=None` 按 pane 自动定位）；响应通道即弃 -/// ——调用方不关心结果，失败在事件线程侧留日志即可。SFTP 图片粘贴用它回粘 -/// 远端路径。 -pub fn dispatch_prompt(proxy: &EventLoopProxy, pane_id: u64, text: String) { - let (dispatch, _receiver) = RuntimeDispatch::new(RuntimeCommand::Prompt { - window_id: None, - pane_id, - text, - submit: false, - }); - if proxy.send_event(Event::new(EventType::RuntimeControl(dispatch), None)).is_err() { - warn!("prompt dispatch failed: event loop is gone"); - } -} - -/// Explorer 右键「在 Nebula 中打开」/ CLI 带 `--working-directory` 的启动并入 -/// 驻留实例：先经 ATTACH 恢复或聚焦既有窗口（与平启动同一套交接，detached -/// 的标签因此先回来），再请求在该窗口打开定目录标签。任一步失败都返回 -/// false，调用方回落为独立启动——绝不吞掉用户的手势。 -pub fn try_open_directory_existing(dir: &std::path::Path) -> bool { - try_open_tab_existing(Some(dir)) -} - -fn try_open_tab_existing(dir: Option<&std::path::Path>) -> bool { - if legacy_request("ATTACH").is_none() { - return false; - } - // ATTACH 与 tab.new 都落到同一条 winit 事件队列上，先后有序：窗口先 - // 恢复，新标签随后落在恢复出来的窗口里。 - let params = dir.map_or_else(|| json!({}), |dir| json!({ "cwd": dir })); - cli::request_once("tab.new", params, IO_TIMEOUT).map(|response| response.ok).unwrap_or(false) -} - -fn legacy_request(verb: &str) -> Option<()> { - read_endpoint() - .and_then(|endpoint| legacy_request_to(verb, &endpoint)) - // Upgrade compatibility: an already-running pre-v1 Nebula still - // publishes mux.port. A plain launch can attach to it and exit; a new - // resident process writes runtime.port after the old process ends. - .or_else(|| { - read_endpoint_from(legacy_port_file()) - .and_then(|endpoint| legacy_request_to(verb, &endpoint)) - }) -} - -fn legacy_request_to(verb: &str, endpoint: &Endpoint) -> Option<()> { - let mut stream = TcpStream::connect_timeout(&endpoint_addr(&endpoint), CONNECT_TIMEOUT).ok()?; - stream.set_read_timeout(Some(IO_TIMEOUT)).ok()?; - stream.set_write_timeout(Some(IO_TIMEOUT)).ok()?; - stream.write_all(format!("{verb} {}\n", endpoint.token).as_bytes()).ok()?; - let mut line = String::new(); - BufReader::new(stream).read_line(&mut line).ok()?; - (line.trim() == "OK").then_some(()) -} - -/// Resident versioned runtime API server. -pub struct RuntimeServer { - endpoint: Endpoint, - port_file: PathBuf, -} - -impl RuntimeServer { - pub fn spawn(proxy: EventLoopProxy, hub: RuntimeHub) -> Option { - Self::spawn_with_sink(EventSink::Winit(proxy), hub) - } - - /// Same discovery file and ATTACH/PING/JSON protocol as [`Self::spawn`], - /// but attach/control are delivered through a callback (GPUI has no winit - /// `EventLoopProxy`). - pub fn spawn_callback( - on_event: impl Fn(RuntimeCallback) + Send + Sync + 'static, - hub: RuntimeHub, - ) -> Option { - Self::spawn_with_sink(EventSink::Callback(Arc::new(on_event)), hub) - } - - fn spawn_with_sink(sink: EventSink, hub: RuntimeHub) -> Option { - if read_endpoint().and_then(|endpoint| legacy_request_to("PING", &endpoint)).is_some() { - info!("Runtime API server already running; this instance stays client-only"); - return None; - } - - let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).ok()?; - let endpoint = Endpoint { port: listener.local_addr().ok()?.port(), token: fresh_token() }; - let path = port_file(); - let contents = format!("{} {} {}\n", endpoint.port, endpoint.token, PROTOCOL_VERSION); - if crate::atomic_file::write(&path, contents.as_bytes()).is_err() { - warn!("Runtime API: cannot write {path:?}; control plane disabled"); - return None; - } - - let server_token = endpoint.token.clone(); - let spawned = std::thread::Builder::new() - .name("nebula-runtime-api".into()) - .spawn(move || serve(listener, server_token, sink, hub)) - .is_ok(); - spawned.then(|| Self { endpoint, port_file: path }) - } -} - -impl Drop for RuntimeServer { - fn drop(&mut self) { - // Do not delete another process's newer discovery record. - if read_endpoint().as_ref() == Some(&self.endpoint) { - let _ = std::fs::remove_file(&self.port_file); - } - } -} - fn serve(listener: TcpListener, token: String, sink: EventSink, hub: RuntimeHub) { let active = Arc::new(AtomicUsize::new(0)); for stream in listener.incoming() { diff --git a/nebula_app/src/runtime_api/cli.rs b/nebula_app/src/runtime_api/cli.rs index 55086fc..3b90bb8 100644 --- a/nebula_app/src/runtime_api/cli.rs +++ b/nebula_app/src/runtime_api/cli.rs @@ -22,14 +22,18 @@ pub(super) fn request_once( params: Value, timeout: Duration, ) -> Result> { - let endpoint = - read_endpoint().ok_or_else(|| CliError("no resident Nebula runtime found".into()))?; + let endpoint = read_endpoint() + .ok_or_else(|| CliError::new("runtime_unavailable", "no resident Nebula runtime found"))?; let request = ApiRequest::new(endpoint.token.clone(), method, params); let stream = client_stream(&endpoint, &request, Some(timeout))?; let mut line = String::new(); BufReader::new(stream).read_line(&mut line)?; if line.is_empty() { - return Err(CliError("runtime closed the connection without a response".into()).into()); + return Err(CliError::new( + "runtime_no_response", + "runtime closed the connection without a response", + ) + .into()); } Ok(serde_json::from_str(&line)?) } @@ -45,15 +49,31 @@ fn print_response(response: &ApiResponse, pretty: bool) -> Result<(), Box Result<(), Box> { + let pretty = options.pretty; + match run_cli_inner(options) { + result @ Ok(_) => result, + Err(error) if error.downcast_ref::().is_some() => Err(error), + Err(error) => { + let code = + error.downcast_ref::().map_or("cli_transport_error", |error| error.code); + let response = ApiResponse::failure("cli", ApiError::new(code, error.to_string())); + print_response(&response, pretty) + }, + } +} + +fn run_cli_inner(options: ControlOptions) -> Result<(), Box> { if options.timeout_ms == 0 || Duration::from_millis(options.timeout_ms) > MAX_WAIT { - return Err(CliError("--timeout-ms must be between 1 and 86400000".into()).into()); + return Err( + CliError::new("invalid_params", "--timeout-ms must be between 1 and 86400000").into() + ); } let timeout = Duration::from_millis(options.timeout_ms); match options.command { @@ -310,8 +330,8 @@ fn wait_state_name(state: ControlWaitState) -> &'static str { } fn subscribe_cli(since: Option, timeout: Duration) -> Result<(), Box> { - let endpoint = - read_endpoint().ok_or_else(|| CliError("no resident Nebula runtime found".into()))?; + let endpoint = read_endpoint() + .ok_or_else(|| CliError::new("runtime_unavailable", "no resident Nebula runtime found"))?; let request = ApiRequest::new( endpoint.token.clone(), "events.subscribe", @@ -321,9 +341,11 @@ fn subscribe_cli(since: Option, timeout: Duration) -> Result<(), Box, timeout: Duration) -> Result<(), Box) -> Self { + Self { code, message: message.into() } + } +} impl fmt::Display for CliError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(&self.message) + } +} + +impl Error for CliError {} + +#[derive(Debug)] +struct PrintedCliError(String); + +impl fmt::Display for PrintedCliError { fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { formatter.write_str(&self.0) } } -impl Error for CliError {} +impl Error for PrintedCliError {} diff --git a/nebula_app/src/runtime_api/server.rs b/nebula_app/src/runtime_api/server.rs new file mode 100644 index 0000000..17c2a3c --- /dev/null +++ b/nebula_app/src/runtime_api/server.rs @@ -0,0 +1,161 @@ +//! Runtime endpoint discovery and resident server ownership. + +use super::*; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(super) struct Endpoint { + pub(super) port: u16, + pub(super) token: String, +} + +fn port_file() -> PathBuf { + crate::display::nebula_data_dir().join("runtime.port") +} + +fn legacy_port_file() -> PathBuf { + crate::display::nebula_data_dir().join("mux.port") +} + +pub(super) fn read_endpoint() -> Option { + read_endpoint_from(port_file()) +} + +fn read_endpoint_from(path: PathBuf) -> Option { + let data = std::fs::read_to_string(path).ok()?; + let mut parts = data.split_whitespace(); + Some(Endpoint { port: parts.next()?.parse().ok()?, token: parts.next()?.to_owned() }) +} + +pub(super) fn endpoint_addr(endpoint: &Endpoint) -> SocketAddr { + SocketAddr::from((Ipv4Addr::LOCALHOST, endpoint.port)) +} + +fn fresh_token() -> String { + use std::hash::{BuildHasher, Hasher, RandomState}; + let mut a = RandomState::new().build_hasher(); + let mut b = RandomState::new().build_hasher(); + a.write_u32(std::process::id()); + b.write_u128( + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|duration| duration.as_nanos()) + .unwrap_or(0), + ); + format!("{:016x}{:016x}", a.finish(), b.finish()) +} + +/// 普通二次启动并入驻留实例：先恢复/聚焦窗口，再新建一个默认 shell 标签页。 +pub fn try_open_default_tab_existing() -> bool { + try_open_tab_existing(None) +} + +/// 后台任务把一行文本作为输入敲进某个 pane（不回车）。 +pub fn dispatch_prompt(proxy: &EventLoopProxy, pane_id: u64, text: String) { + let (dispatch, _receiver) = RuntimeDispatch::new(RuntimeCommand::Prompt { + window_id: None, + pane_id, + text, + submit: false, + }); + if proxy.send_event(Event::new(EventType::RuntimeControl(dispatch), None)).is_err() { + warn!("prompt dispatch failed: event loop is gone"); + } +} + +/// Explorer 右键或带 `--working-directory` 的启动并入驻留实例。 +pub fn try_open_directory_existing(dir: &std::path::Path) -> bool { + try_open_tab_existing(Some(dir)) +} + +fn try_open_tab_existing(dir: Option<&std::path::Path>) -> bool { + if legacy_request("ATTACH").is_none() { + return false; + } + // ATTACH 与 tab.new 落到同一事件队列，窗口先恢复，新标签随后创建。 + let params = dir.map_or_else(|| json!({}), |dir| json!({ "cwd": dir })); + cli::request_once("tab.new", params, IO_TIMEOUT).map(|response| response.ok).unwrap_or(false) +} + +fn legacy_request(verb: &str) -> Option<()> { + read_endpoint() + .and_then(|endpoint| legacy_request_to(verb, &endpoint)) + // 已运行的 pre-v1 版本只发布 mux.port；升级期间仍允许普通启动交接。 + .or_else(|| { + read_endpoint_from(legacy_port_file()) + .and_then(|endpoint| legacy_request_to(verb, &endpoint)) + }) +} + +fn legacy_request_to(verb: &str, endpoint: &Endpoint) -> Option<()> { + let mut stream = TcpStream::connect_timeout(&endpoint_addr(endpoint), CONNECT_TIMEOUT).ok()?; + stream.set_read_timeout(Some(IO_TIMEOUT)).ok()?; + stream.set_write_timeout(Some(IO_TIMEOUT)).ok()?; + stream.write_all(format!("{verb} {}\n", endpoint.token).as_bytes()).ok()?; + let mut line = String::new(); + BufReader::new(stream).read_line(&mut line).ok()?; + (line.trim() == "OK").then_some(()) +} + +/// Resident versioned runtime API server. +pub struct RuntimeServer { + endpoint: Endpoint, + port_file: PathBuf, + _owner_lock: crate::atomic_file::LifetimeFileLock, +} + +impl RuntimeServer { + pub fn spawn(proxy: EventLoopProxy, hub: RuntimeHub) -> Option { + Self::spawn_with_sink(EventSink::Winit(proxy), hub) + } + + /// GPUI variant of [`Self::spawn`], delivered through a callback. + pub fn spawn_callback( + on_event: impl Fn(RuntimeCallback) + Send + Sync + 'static, + hub: RuntimeHub, + ) -> Option { + Self::spawn_with_sink(EventSink::Callback(Arc::new(on_event)), hub) + } + + fn spawn_with_sink(sink: EventSink, hub: RuntimeHub) -> Option { + let path = port_file(); + let owner_lock = match crate::atomic_file::try_lifetime_lock(&path) { + Ok(Some(lock)) => lock, + Ok(None) => { + info!("Runtime API owner is starting or already running; staying client-only"); + return None; + }, + Err(error) => { + warn!("Runtime API: cannot lock {path:?}: {error}; control plane disabled"); + return None; + }, + }; + if read_endpoint().and_then(|endpoint| legacy_request_to("PING", &endpoint)).is_some() { + info!("Runtime API server already running; this instance stays client-only"); + return None; + } + + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).ok()?; + let endpoint = Endpoint { port: listener.local_addr().ok()?.port(), token: fresh_token() }; + let contents = format!("{} {} {}\n", endpoint.port, endpoint.token, PROTOCOL_VERSION); + if crate::atomic_file::write(&path, contents.as_bytes()).is_err() { + warn!("Runtime API: cannot write {path:?}; control plane disabled"); + return None; + } + + let server_token = endpoint.token.clone(); + let spawned = std::thread::Builder::new() + .name("nebula-runtime-api".into()) + .spawn(move || serve(listener, server_token, sink, hub)) + .is_ok(); + spawned.then(|| Self { endpoint, port_file: path, _owner_lock: owner_lock }) + } +} + +impl Drop for RuntimeServer { + fn drop(&mut self) { + // Do not delete another process's newer discovery record. + if read_endpoint().as_ref() == Some(&self.endpoint) { + let _ = std::fs::remove_file(&self.port_file); + } + } +}