diff --git a/Cargo.lock b/Cargo.lock index e6d0ba7..effc024 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1176,6 +1176,7 @@ version = "0.1.5" dependencies = [ "axum", "cursor-server", + "libc", "rfd", "serde", "serde_json", diff --git a/apps/desktop/src-tauri/Cargo.toml b/apps/desktop/src-tauri/Cargo.toml index 8ec2faa..7f6328e 100644 --- a/apps/desktop/src-tauri/Cargo.toml +++ b/apps/desktop/src-tauri/Cargo.toml @@ -14,6 +14,7 @@ tauri-build = { version = "2", features = [] } [dependencies] axum = "0.8" cursor-server = { path = "../../../server" } +libc = "0.2" rfd = "0.15" serde = { version = "1", features = ["derive"] } serde_json = "1" diff --git a/apps/desktop/src-tauri/src/desktop.rs b/apps/desktop/src-tauri/src/desktop.rs index 31f2d85..6afa44c 100644 --- a/apps/desktop/src-tauri/src/desktop.rs +++ b/apps/desktop/src-tauri/src/desktop.rs @@ -149,6 +149,23 @@ pub fn run() -> ExitCode { return ExitCode::FAILURE; } }; + #[cfg(unix)] + { + let open_file_limit = match crate::resource_limits::raise_open_file_limit() { + Ok(limit) => limit, + Err(error) => { + diagnostics.report_fatal(&error); + return ExitCode::FAILURE; + } + }; + tracing::info!( + requested = crate::resource_limits::REQUESTED_OPEN_FILE_LIMIT, + previous = open_file_limit.previous, + effective = open_file_limit.effective, + hard = open_file_limit.hard, + "open file limit configured" + ); + } tracing::info!( version = env!("CARGO_PKG_VERSION"), os = std::env::consts::OS, diff --git a/apps/desktop/src-tauri/src/lib.rs b/apps/desktop/src-tauri/src/lib.rs index ec2df31..14e5164 100644 --- a/apps/desktop/src-tauri/src/lib.rs +++ b/apps/desktop/src-tauri/src/lib.rs @@ -1,6 +1,7 @@ mod desktop; #[cfg(not(dev))] mod frontend; +mod resource_limits; mod startup; mod tray; diff --git a/apps/desktop/src-tauri/src/resource_limits.rs b/apps/desktop/src-tauri/src/resource_limits.rs new file mode 100644 index 0000000..b553f66 --- /dev/null +++ b/apps/desktop/src-tauri/src/resource_limits.rs @@ -0,0 +1,56 @@ +//! Configures process resource limits before the desktop runtime starts. + +#[cfg(unix)] +use std::io; + +#[cfg(unix)] +pub(crate) const REQUESTED_OPEN_FILE_LIMIT: u64 = 65_536; + +#[cfg(unix)] +pub(crate) struct OpenFileLimit { + pub(crate) previous: u64, + pub(crate) effective: u64, + pub(crate) hard: u64, +} + +#[cfg(unix)] +pub(crate) fn raise_open_file_limit() -> io::Result { + let mut limits = libc::rlimit { + rlim_cur: 0, + rlim_max: 0, + }; + // SAFETY: `limits` points to writable memory for one `rlimit` value. + if unsafe { libc::getrlimit(libc::RLIMIT_NOFILE, &mut limits) } != 0 { + return Err(io::Error::last_os_error()); + } + + let previous = limits.rlim_cur; + let target = limits + .rlim_max + .min(REQUESTED_OPEN_FILE_LIMIT as libc::rlim_t); + if previous < target { + let requested = libc::rlimit { + rlim_cur: target, + rlim_max: limits.rlim_max, + }; + // SAFETY: `requested` is a valid `rlimit` value and does not raise the hard limit. + if unsafe { libc::setrlimit(libc::RLIMIT_NOFILE, &requested) } != 0 { + return Err(io::Error::last_os_error()); + } + } + + let mut effective = libc::rlimit { + rlim_cur: 0, + rlim_max: 0, + }; + // SAFETY: `effective` points to writable memory for one `rlimit` value. + if unsafe { libc::getrlimit(libc::RLIMIT_NOFILE, &mut effective) } != 0 { + return Err(io::Error::last_os_error()); + } + + Ok(OpenFileLimit { + previous: previous as u64, + effective: effective.rlim_cur as u64, + hard: effective.rlim_max as u64, + }) +} diff --git a/server/src/api/cursor/handlers.rs b/server/src/api/cursor/handlers.rs index 6a608d2..aa0c0d3 100644 --- a/server/src/api/cursor/handlers.rs +++ b/server/src/api/cursor/handlers.rs @@ -27,8 +27,11 @@ use crate::{ Result, }; -pub fn router(registry: TransportRegistry) -> Result { - let proxy = CursorProxy::cursor(registry.store().clone())?; +pub fn router( + registry: TransportRegistry, + clients: crate::network::NetworkClients, +) -> Result { + let proxy = CursorProxy::cursor(clients); let knowledge = knowledge::KnowledgeService::managed()?; Ok(router_with_proxy(registry, proxy, knowledge)) } diff --git a/server/src/api/cursor/proxy.rs b/server/src/api/cursor/proxy.rs index df1dff6..6463fe3 100644 --- a/server/src/api/cursor/proxy.rs +++ b/server/src/api/cursor/proxy.rs @@ -14,8 +14,7 @@ pub const UPSTREAM_URL_HEADER: &str = "x-server-upstream-url"; #[derive(Clone)] pub struct CursorProxy { - client: Option, - store: Option, + clients: crate::network::NetworkClients, upstream: String, } @@ -47,23 +46,15 @@ impl BufferedResponse { } impl CursorProxy { - pub fn cursor(store: crate::store::Store) -> Result { - Ok(Self { - client: None, - store: Some(store), + pub fn cursor(clients: crate::network::NetworkClients) -> Self { + Self { + clients, upstream: CURSOR_UPSTREAM.into(), - }) + } } async fn client(&self) -> Result { - match (&self.client, &self.store) { - (Some(client), _) => Ok(client.clone()), - (_, Some(store)) => Ok(crate::network::client_builder(store) - .await? - .redirect(reqwest::redirect::Policy::none()) - .build()?), - _ => unreachable!("Cursor proxy always has a client or store"), - } + self.clients.cursor_client().await } } diff --git a/server/src/api/router.rs b/server/src/api/router.rs index 4a7e8e9..377831c 100644 --- a/server/src/api/router.rs +++ b/server/src/api/router.rs @@ -1,7 +1,7 @@ //! Builds the top-level server router. -use crate::{cursor::transport::TransportRegistry, Result}; +use crate::{cursor::transport::TransportRegistry, network::NetworkClients, Result}; -pub fn router(registry: TransportRegistry) -> Result { - super::cursor::router(registry) +pub fn router(registry: TransportRegistry, clients: NetworkClients) -> Result { + super::cursor::router(registry, clients) } diff --git a/server/src/app.rs b/server/src/app.rs index d4ec4b1..62cb015 100644 --- a/server/src/app.rs +++ b/server/src/app.rs @@ -44,9 +44,11 @@ impl App { plugin_runtime.clone(), config.app_version.clone(), )?; + let clients = crate::network::NetworkClients::new(store.clone()); let provider = std::sync::Arc::new(ProviderRouter::new( store.clone(), plugins.clone(), + clients.clone(), config.provider_request_timeout, config.provider_stream_idle_timeout, )); @@ -58,10 +60,15 @@ impl App { plugins.clone(), crate::config::managed_data_dir()?.join("rules"), ); - let control = - control::ControlService::new(store.clone(), provider, plugin_runtime, plugins)?; + let control = control::ControlService::new( + store.clone(), + provider, + plugin_runtime, + plugins, + clients.clone(), + )?; let harness = control.cursor_harness().clone(); - let mut router = api::router(registry.clone())?; + let mut router = api::router(registry.clone(), clients)?; router = match &config.console { Some(ConsoleSource::Directory(directory)) => { router.merge(control::web_router(control.clone(), directory)) diff --git a/server/src/control/service.rs b/server/src/control/service.rs index 5c3fa02..28e9e09 100644 --- a/server/src/control/service.rs +++ b/server/src/control/service.rs @@ -41,6 +41,7 @@ pub struct ControlService { provider: Arc, plugin_runtime: PluginRuntime, plugins: PluginRegistry, + clients: crate::network::NetworkClients, model_tests: Arc>>, } @@ -151,6 +152,7 @@ impl ControlService { provider: Arc, plugin_runtime: PluginRuntime, plugins: PluginRegistry, + clients: crate::network::NetworkClients, ) -> Result { Ok(Self { cursor_harness: CursorHarness::new(store.clone())?, @@ -158,6 +160,7 @@ impl ControlService { provider, plugin_runtime, plugins, + clients, model_tests: Arc::new(Mutex::new(BTreeMap::new())), }) } @@ -256,7 +259,7 @@ impl ControlService { disabled_ad_ids: Option<&str>, language: &str, ) -> Result { - let client = crate::network::client(&self.store).await?; + let client = self.clients.default_client().await?; let installation_id = self.store.installation_id().await?; let mut request = client .get(ADS_ENDPOINT) @@ -281,7 +284,7 @@ impl ControlService { } pub(super) async fn dismiss_ad(&self, ad_id: &str, input: &AdDismissalInput) -> Result<()> { - let client = crate::network::client(&self.store).await?; + let client = self.clients.default_client().await?; let installation_id = self.store.installation_id().await?; let mut endpoint = Url::parse(ADS_ENDPOINT).map_err(|error| { Error::Config(format!("advertisement endpoint is invalid: {error}")) @@ -510,7 +513,7 @@ impl ControlService { } pub async fn discover_models(&self, input: &ModelDiscoveryInput) -> Result { - let client = crate::network::client(&self.store).await?; + let client = self.clients.default_client().await?; let base_url = crate::model::normalize_request_url(&input.base_url)?; discover_models_from_endpoint( &client, @@ -677,7 +680,9 @@ impl ControlService { } pub async fn set_proxy_settings(&self, settings: ProxySettingsInput) -> Result { - self.store.set_proxy_settings(settings).await + let settings = self.store.set_proxy_settings(settings).await?; + self.clients.invalidate().await; + Ok(settings) } pub async fn tab_settings(&self) -> Result { diff --git a/server/src/network.rs b/server/src/network.rs index d31f02c..b790dae 100644 --- a/server/src/network.rs +++ b/server/src/network.rs @@ -1,8 +1,93 @@ -//! Provides shared network client and transport configuration. -//! Outbound HTTP clients configured from persisted application proxy settings. +//! Owns reusable outbound HTTP clients configured from persisted proxy settings. + +use std::{sync::Arc, time::Duration}; + +use tokio::sync::RwLock; use crate::{store::Store, Result}; +#[derive(Clone)] +pub struct NetworkClients { + store: Store, + cache: Arc>, +} + +#[derive(Default)] +struct ClientCache { + default: Option, + cursor: Option, + provider: Option<(Duration, reqwest::Client)>, +} + +impl NetworkClients { + pub fn new(store: Store) -> Self { + Self { + store, + cache: Arc::new(RwLock::new(ClientCache::default())), + } + } + + pub async fn default_client(&self) -> Result { + if let Some(client) = self.cache.read().await.default.clone() { + return Ok(client); + } + let mut cache = self.cache.write().await; + if let Some(client) = cache.default.clone() { + return Ok(client); + } + let client = client_builder(&self.store).await?.build()?; + cache.default = Some(client.clone()); + Ok(client) + } + + pub async fn cursor_client(&self) -> Result { + if let Some(client) = self.cache.read().await.cursor.clone() { + return Ok(client); + } + let mut cache = self.cache.write().await; + if let Some(client) = cache.cursor.clone() { + return Ok(client); + } + let client = client_builder(&self.store) + .await? + .redirect(reqwest::redirect::Policy::none()) + .build()?; + cache.cursor = Some(client.clone()); + Ok(client) + } + + pub async fn provider_client(&self, timeout: Duration) -> Result { + if let Some((_, client)) = self + .cache + .read() + .await + .provider + .as_ref() + .filter(|(cached_timeout, _)| *cached_timeout == timeout) + { + return Ok(client.clone()); + } + let mut cache = self.cache.write().await; + if let Some((_, client)) = cache + .provider + .as_ref() + .filter(|(cached_timeout, _)| *cached_timeout == timeout) + { + return Ok(client.clone()); + } + let client = client_builder(&self.store) + .await? + .timeout(timeout) + .build()?; + cache.provider = Some((timeout, client.clone())); + Ok(client) + } + + pub async fn invalidate(&self) { + *self.cache.write().await = ClientCache::default(); + } +} + pub async fn client_builder(store: &Store) -> Result { let settings = store.proxy_settings_secret().await?; // Use the platform TLS stack for compatibility with provider gateways that diff --git a/server/src/provider/router.rs b/server/src/provider/router.rs index 7eabe64..577c0eb 100644 --- a/server/src/provider/router.rs +++ b/server/src/provider/router.rs @@ -21,6 +21,7 @@ use super::{ pub struct ProviderRouter { store: Store, plugins: PluginRegistry, + clients: crate::network::NetworkClients, request_timeout: Duration, stream_idle_timeout: Duration, } @@ -29,12 +30,14 @@ impl ProviderRouter { pub fn new( store: Store, plugins: PluginRegistry, + clients: crate::network::NetworkClients, request_timeout: Duration, stream_idle_timeout: Duration, ) -> Self { Self { store, plugins, + clients, request_timeout, stream_idle_timeout, } @@ -49,6 +52,7 @@ impl Provider for ProviderRouter { ) -> ProviderStream { let store = self.store.clone(); let plugins = self.plugins.clone(); + let clients = self.clients.clone(); let request_timeout = self.request_timeout; let stream_idle_timeout = self.stream_idle_timeout; Box::pin(try_stream! { @@ -91,7 +95,7 @@ impl Provider for ProviderRouter { request_timeout, allowed_body_fields: None, }; - let client = crate::network::client_builder(&store).await?.timeout(request_timeout).build()?; + let client = clients.provider_client(request_timeout).await?; let provider = build_observed(&config, recorder.clone(), client)?; (recorder, guard, provider.stream(routed, cancellation.clone())) }; diff --git a/server/tests/knowledge_rules.rs b/server/tests/knowledge_rules.rs index 1dc4bc5..05687b0 100644 --- a/server/tests/knowledge_rules.rs +++ b/server/tests/knowledge_rules.rs @@ -106,7 +106,7 @@ async fn decode(response: Response) -> M { #[tokio::test] async fn offline_crud_round_trip_persists_markdown() { let (_store_dir, store) = fixtures::temp_store().await; - let upstream = CursorProxy::cursor(store).unwrap(); + let upstream = CursorProxy::cursor(cursor_server::network::NetworkClients::new(store)); let rules_dir = tempfile::tempdir().unwrap(); let rules_root = rules_dir.path().join("rules"); let service = KnowledgeService::with_root(rules_root.clone()).unwrap(); @@ -193,7 +193,7 @@ async fn offline_crud_round_trip_persists_markdown() { #[tokio::test] async fn updating_missing_rule_reports_failure() { let (_store_dir, store) = fixtures::temp_store().await; - let upstream = CursorProxy::cursor(store).unwrap(); + let upstream = CursorProxy::cursor(cursor_server::network::NetworkClients::new(store)); let rules_dir = tempfile::tempdir().unwrap(); let service = KnowledgeService::with_root(rules_dir.path().join("rules")).unwrap();