diff --git a/src/rust/lqos_bus/src/bus/client.rs b/src/rust/lqos_bus/src/bus/client.rs index 8320035d..a2cb7e4a 100644 --- a/src/rust/lqos_bus/src/bus/client.rs +++ b/src/rust/lqos_bus/src/bus/client.rs @@ -2,9 +2,11 @@ // SPDX-License-Identifier: AGPL-3.0-or-later WITH LicenseRef-LibreQoS-Exception use crate::{BUS_SOCKET_PATH, BusReply, BusRequest, BusResponse, BusSession, bus::BusClientError}; +use std::time::Duration; use tokio::{ io::{AsyncReadExt, AsyncWriteExt}, net::UnixStream, + time::timeout, }; use tracing::error; @@ -50,6 +52,16 @@ impl LibreqosBusClient { }) } + /// Creates a new `LibreqosBusClient`, failing if connection setup exceeds + /// `request_timeout`. + /// + /// Side effects: opens a Unix socket connection to the local `lqosd` bus. + pub async fn new_with_timeout(request_timeout: Duration) -> Result { + timeout(request_timeout, Self::new()) + .await + .map_err(|_| BusClientError::TimedOut)? + } + /// Sends a request to the bus and waits for a response. /// /// ## Arguments @@ -91,6 +103,21 @@ impl LibreqosBusClient { } Ok(response.responses) } + + /// Sends a request to the bus and waits for a bounded response. + /// + /// Side effects: writes the request to the connected Unix socket. If this + /// call times out, discard the client because the stream may later receive + /// the timed-out reply. + pub async fn request_with_timeout( + &mut self, + requests: Vec, + request_timeout: Duration, + ) -> Result, BusClientError> { + timeout(request_timeout, self.request(requests)) + .await + .map_err(|_| BusClientError::TimedOut)? + } } /// Convenient wrapper for accessing the bus, for a single request-response cycle. This @@ -108,3 +135,19 @@ pub async fn bus_request(requests: Vec) -> Result, }; client.request(requests).await } + +/// Convenient wrapper for a single bus request with an end-to-end timeout. +/// +/// Side effects: opens a Unix socket connection, sends the request, and closes +/// the connection when the client is dropped. +pub async fn bus_request_with_timeout( + requests: Vec, + request_timeout: Duration, +) -> Result, BusClientError> { + timeout(request_timeout, async move { + let mut client = LibreqosBusClient::new().await?; + client.request(requests).await + }) + .await + .map_err(|_| BusClientError::TimedOut)? +} diff --git a/src/rust/lqos_bus/src/bus/mod.rs b/src/rust/lqos_bus/src/bus/mod.rs index efd88fe9..91f01315 100644 --- a/src/rust/lqos_bus/src/bus/mod.rs +++ b/src/rust/lqos_bus/src/bus/mod.rs @@ -9,7 +9,7 @@ mod request; pub mod response; mod session; mod unix_socket_server; -pub use client::{LibreqosBusClient, bus_request}; +pub use client::{LibreqosBusClient, bus_request, bus_request_with_timeout}; pub use queue_data::*; pub use reply::BusReply; pub use request::{ @@ -55,6 +55,9 @@ pub enum BusClientError { /// Reading the reply from the socket failed. #[error("Cannot read from socket")] StreamReadError, + /// The socket operation exceeded the caller's deadline. + #[error("Timed out waiting for bus response")] + TimedOut, /// The socket connection is no longer usable. #[error("Stream is no longer connected")] StreamNotConnected, diff --git a/src/rust/lqos_bus/src/bus/unix_socket_server.rs b/src/rust/lqos_bus/src/bus/unix_socket_server.rs index 59b89e3f..ddfae309 100644 --- a/src/rust/lqos_bus/src/bus/unix_socket_server.rs +++ b/src/rust/lqos_bus/src/bus/unix_socket_server.rs @@ -5,21 +5,63 @@ use crate::{ BUS_SOCKET_PATH, BusReply, BusRequest, BusResponse, bus::client::{MAGIC_NUMBER, MAGIC_RESPONSE}, }; -use std::{ffi::CString, fs::remove_file}; +use std::{ffi::CString, fs::remove_file, time::Duration}; use thiserror::Error; use tokio::{ io::{AsyncReadExt, AsyncWriteExt}, net::UnixListener, + task::spawn_blocking, + time::timeout, }; use tracing::{debug, error, info, warn}; use super::BUS_SOCKET_DIRECTORY; use super::protocol::{decode_session_cbor, encode_reply_cbor, read_frame, write_frame}; +const BUS_HANDLER_TIMEOUT: Duration = Duration::from_secs(30); + fn dropped_reply_response_count(reply: &BusReply) -> usize { reply.responses.len() } +fn timeout_reply(request_count: usize) -> BusReply { + BusReply { + responses: (0..request_count) + .map(|_| BusResponse::Fail("Bus request handler timed out".to_string())) + .collect(), + } +} + +async fn handle_requests_with_deadline( + handle_bus_requests: fn(&[BusRequest], &mut Vec), + requests: Vec, +) -> BusReply { + let request_count = requests.len(); + let handler = spawn_blocking(move || { + let mut response = BusReply { + responses: Vec::with_capacity(request_count), + }; + handle_bus_requests(&requests, &mut response.responses); + response + }); + + match timeout(BUS_HANDLER_TIMEOUT, handler).await { + Ok(Ok(response)) => response, + Ok(Err(err)) => { + warn!("Bus request handler task failed: {err}"); + BusReply { + responses: (0..request_count) + .map(|_| BusResponse::Fail("Bus request handler failed".to_string())) + .collect(), + } + } + Err(_) => { + warn!("Bus request handler timed out after {BUS_HANDLER_TIMEOUT:?}"); + timeout_reply(request_count) + } + } +} + /// Implements a Tokio-friendly server using Unix Sockets and the bus protocol. /// Requests are handled and then forwarded to the handler. pub struct UnixSocketServer {} @@ -113,8 +155,7 @@ impl UnixSocketServer { ret = bus_rx.recv() => { // We received a channel-based message if let Some((reply_channel, msg)) = ret { - let mut response = BusReply { responses: Vec::with_capacity(8) }; - handle_bus_requests(&[msg], &mut response.responses); + let response = handle_requests_with_deadline(handle_bus_requests, vec![msg]).await; if let Err(reply) = reply_channel.send(response) { warn!( dropped_response_count = dropped_reply_response_count(&reply), @@ -182,8 +223,9 @@ impl UnixSocketServer { debug!("Received request: {:?}", request); // Handle the request and build the response - let mut response = BusReply { responses: Vec::with_capacity(8) }; - handle_bus_requests(&request.requests, &mut response.responses); + let response = + handle_requests_with_deadline(handle_bus_requests, request.requests) + .await; // Encode the response let Ok(encoded_response) = encode_reply_cbor(&response) else { diff --git a/src/rust/lqos_bus/src/lib.rs b/src/rust/lqos_bus/src/lib.rs index 48f9bd08..a8d0e5f5 100644 --- a/src/rust/lqos_bus/src/lib.rs +++ b/src/rust/lqos_bus/src/lib.rs @@ -37,7 +37,7 @@ pub use bus::{ BUS_SOCKET_PATH, BakeryCapacityReportInterface, BlackboardSystem, BusClientError, BusReply, BusRequest, BusResponse, BusSession, CakeDiffTinTransit, CakeDiffTransit, CakeTransit, LibreqosBusClient, QueueStoreTransit, SchedulerProgressReport, TopFlowType, UnixSocketServer, - UrgentSeverity, UrgentSource, bus_request, + UrgentSeverity, UrgentSource, bus_request, bus_request_with_timeout, }; pub use tc_handle::TcHandle; diff --git a/src/rust/lqos_network_devices/src/actor.rs b/src/rust/lqos_network_devices/src/actor.rs index 6b6f3069..b349e477 100644 --- a/src/rust/lqos_network_devices/src/actor.rs +++ b/src/rust/lqos_network_devices/src/actor.rs @@ -5,18 +5,18 @@ use crate::dynamic_store::{load_dynamic_circuits_from_disk, persist_dynamic_circ use crate::state; use crate::{DaemonHooks, ShapedDevicesCatalog, load_network_json, load_shaped_devices}; use anyhow::{Context, Result, anyhow}; -use crossbeam_channel::{Receiver, Sender}; +use crossbeam_channel::{Receiver, RecvTimeoutError, SendTimeoutError, Sender, bounded}; use ip_network_table::IpNetworkTable; use once_cell::sync::OnceCell; use std::collections::HashSet; use std::net::IpAddr; use std::sync::Arc; use std::time::Duration; -use tokio::sync::oneshot; use tracing::{debug, error, warn}; const RELOAD_RETRY_DELAY_MS: u64 = 500; const RELOAD_ATTEMPTS: usize = 2; +const ACTOR_COMMAND_TIMEOUT: Duration = Duration::from_secs(10); const DYNAMIC_CIRCUITS_MAINTENANCE_TICK_SECONDS: u64 = 60; const MAX_UNKNOWN_IP_PROMOTIONS_PER_OBSERVATION: usize = 32; @@ -44,41 +44,67 @@ fn sender() -> Result> { .ok_or_else(|| anyhow!("lqos_network_devices runtime actor is not running")) } +fn send_actor_command(command: NetworkDevicesCommand, operation: &str) -> Result<()> { + sender()? + .send_timeout(command, ACTOR_COMMAND_TIMEOUT) + .map_err(|err| match err { + SendTimeoutError::Timeout(_) => { + anyhow!("{operation} timed out while queueing lqos_network_devices actor command") + } + SendTimeoutError::Disconnected(_) => { + anyhow!("{operation} failed because lqos_network_devices actor stopped") + } + }) +} + +fn receive_actor_reply(reply_rx: Receiver>, operation: &str) -> Result { + match reply_rx.recv_timeout(ACTOR_COMMAND_TIMEOUT) { + Ok(result) => result, + Err(RecvTimeoutError::Timeout) => { + Err(anyhow!("{operation} timed out waiting for actor reply")) + } + Err(RecvTimeoutError::Disconnected) => Err(anyhow!("{operation} reply channel closed")), + } +} + pub(crate) fn request_reload_shaped_devices(reason: &str) -> Result<()> { - let (reply_tx, reply_rx) = oneshot::channel(); - sender()?.send(NetworkDevicesCommand::ReloadShapedDevices { - reason: reason.to_string(), - reply: reply_tx, - })?; - reply_rx - .blocking_recv() - .map_err(|_| anyhow!("ShapedDevices reload reply channel closed"))? + let (reply_tx, reply_rx) = bounded(1); + send_actor_command( + NetworkDevicesCommand::ReloadShapedDevices { + reason: reason.to_string(), + reply: reply_tx, + }, + "ShapedDevices reload", + )?; + receive_actor_reply(reply_rx, "ShapedDevices reload") } pub(crate) fn request_reload_network_json(reason: &str) -> Result<()> { - let (reply_tx, reply_rx) = oneshot::channel(); - sender()?.send(NetworkDevicesCommand::ReloadNetworkJson { - reason: reason.to_string(), - reply: reply_tx, - })?; - reply_rx - .blocking_recv() - .map_err(|_| anyhow!("NetworkJson reload reply channel closed"))? + let (reply_tx, reply_rx) = bounded(1); + send_actor_command( + NetworkDevicesCommand::ReloadNetworkJson { + reason: reason.to_string(), + reply: reply_tx, + }, + "NetworkJson reload", + )?; + receive_actor_reply(reply_rx, "NetworkJson reload") } pub(crate) fn apply_shaped_devices_snapshot( reason: &str, shaped: lqos_config::ConfigShapedDevices, ) -> Result<()> { - let (reply_tx, reply_rx) = oneshot::channel(); - sender()?.send(NetworkDevicesCommand::ApplyShapedDevicesSnapshot { - reason: reason.to_string(), - shaped: Box::new(shaped), - reply: reply_tx, - })?; - reply_rx - .blocking_recv() - .map_err(|_| anyhow!("Apply shaped devices reply channel closed"))? + let (reply_tx, reply_rx) = bounded(1); + send_actor_command( + NetworkDevicesCommand::ApplyShapedDevicesSnapshot { + reason: reason.to_string(), + shaped: Box::new(shaped), + reply: reply_tx, + }, + "Apply shaped devices", + )?; + receive_actor_reply(reply_rx, "Apply shaped devices") } pub(crate) fn report_observations(observations: &[CircuitObservation]) { @@ -95,51 +121,53 @@ pub(crate) fn report_observations(observations: &[CircuitObservation]) { } pub(crate) fn upsert_dynamic_circuit(shaped_device: lqos_config::ShapedDevice) -> Result<()> { - let (reply_tx, reply_rx) = oneshot::channel(); - sender()?.send(NetworkDevicesCommand::UpsertDynamicCircuit { - shaped_device: Box::new(shaped_device), - reply: reply_tx, - })?; - reply_rx - .blocking_recv() - .map_err(|_| anyhow!("Upsert dynamic circuit reply channel closed"))? + let (reply_tx, reply_rx) = bounded(1); + send_actor_command( + NetworkDevicesCommand::UpsertDynamicCircuit { + shaped_device: Box::new(shaped_device), + reply: reply_tx, + }, + "Upsert dynamic circuit", + )?; + receive_actor_reply(reply_rx, "Upsert dynamic circuit") } pub(crate) fn remove_dynamic_circuit(circuit_id: &str) -> Result { - let (reply_tx, reply_rx) = oneshot::channel(); - sender()?.send(NetworkDevicesCommand::RemoveDynamicCircuit { - circuit_id: circuit_id.to_string(), - reply: reply_tx, - })?; - reply_rx - .blocking_recv() - .map_err(|_| anyhow!("Remove dynamic circuit reply channel closed"))? + let (reply_tx, reply_rx) = bounded(1); + send_actor_command( + NetworkDevicesCommand::RemoveDynamicCircuit { + circuit_id: circuit_id.to_string(), + reply: reply_tx, + }, + "Remove dynamic circuit", + )?; + receive_actor_reply(reply_rx, "Remove dynamic circuit") } enum NetworkDevicesCommand { ReloadShapedDevices { reason: String, - reply: oneshot::Sender>, + reply: Sender>, }, ReloadNetworkJson { reason: String, - reply: oneshot::Sender>, + reply: Sender>, }, ApplyShapedDevicesSnapshot { reason: String, shaped: Box, - reply: oneshot::Sender>, + reply: Sender>, }, ReportObservations { observations: Vec, }, UpsertDynamicCircuit { shaped_device: Box, - reply: oneshot::Sender>, + reply: Sender>, }, RemoveDynamicCircuit { circuit_id: String, - reply: oneshot::Sender>, + reply: Sender>, }, } diff --git a/src/rust/lqos_probe/src/lib.rs b/src/rust/lqos_probe/src/lib.rs index 03c63b92..936cc67c 100644 --- a/src/rust/lqos_probe/src/lib.rs +++ b/src/rust/lqos_probe/src/lib.rs @@ -15,8 +15,14 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH}; use surge_ping::{Client, Config as PingConfig, ICMP, IcmpPacket, PingIdentifier, PingSequence}; use thiserror::Error; use tokio::sync::{Semaphore, mpsc, oneshot}; +use tokio::time::timeout; use tracing::warn; +const PROBE_DNS_RESOLUTION_TIMEOUT: Duration = Duration::from_secs(2); +const PROBE_WORKER_GRACE: Duration = Duration::from_secs(1); +const PROBE_MANAGER_REPLY_GRACE: Duration = Duration::from_secs(5); +const PROBE_BATCH_CONCURRENCY_ESTIMATE: usize = 256; + /// Logical consumer class for a probe request. #[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize, Allocative)] pub enum ProbeClass { @@ -135,18 +141,23 @@ impl ProbeClient { if requests.is_empty() { return Ok(Vec::new()); } + let wait_timeout = probe_batch_wait_timeout(&requests); let (reply_tx, reply_rx) = oneshot::channel(); - self.tx - .send(ProbeCommand::ProbeBatch { - requests, - max_age, - reply: reply_tx, - }) - .await - .map_err(|_| ProbeClientError::ManagerUnavailable)?; - reply_rx - .await - .map_err(|_| ProbeClientError::ManagerUnavailable) + timeout(wait_timeout, async { + self.tx + .send(ProbeCommand::ProbeBatch { + requests, + max_age, + reply: reply_tx, + }) + .await + .map_err(|_| ProbeClientError::ManagerUnavailable)?; + reply_rx + .await + .map_err(|_| ProbeClientError::ManagerUnavailable) + }) + .await + .map_err(|_| ProbeClientError::TimedOut)? } /// Probes a batch of targets for reachability. @@ -192,6 +203,9 @@ pub enum ProbeClientError { /// The manager returned no observations for the request. #[error("shared probe manager returned no observations")] EmptyResponse, + /// The manager did not answer before the caller's deadline. + #[error("shared probe manager timed out")] + TimedOut, } /// Background shared probe manager. @@ -213,6 +227,27 @@ impl ProbeManager { } } +fn probe_worker_timeout(request_timeout: Duration) -> Duration { + PROBE_DNS_RESOLUTION_TIMEOUT + .saturating_add(request_timeout) + .saturating_add(PROBE_WORKER_GRACE) +} + +fn probe_batch_wait_timeout(requests: &[ProbeRequest]) -> Duration { + let worker_timeout = requests + .iter() + .map(|request| probe_worker_timeout(request.timeout)) + .max() + .unwrap_or(PROBE_MANAGER_REPLY_GRACE); + let wave_count = requests + .len() + .max(1) + .div_ceil(PROBE_BATCH_CONCURRENCY_ESTIMATE) as u32; + worker_timeout + .saturating_mul(wave_count) + .saturating_add(PROBE_MANAGER_REPLY_GRACE) +} + #[derive(Clone, Debug)] struct ProbeCacheEntry { normalized_target: String, @@ -313,7 +348,23 @@ impl ProbeManagerState { let semaphore = self.semaphore.clone(); join_set.spawn(async move { let _permit = semaphore.acquire_owned().await.ok(); - let observation = execute_probe(&request, &key.normalized_target).await; + let observation = match timeout( + probe_worker_timeout(request.timeout), + execute_probe(&request, &key.normalized_target), + ) + .await + { + Ok(observation) => observation, + Err(_) => ProbeCacheEntry { + normalized_target: key.normalized_target.clone(), + resolved_ip: None, + kind: request.kind, + observed_at_unix_ms: now_unix_ms(), + reachable: false, + rtt_ms: None, + error: Some("probe worker timed out".to_string()), + }, + }; (key, observation) }); } @@ -529,9 +580,15 @@ async fn resolve_target(normalized_target: &str) -> Result<(IpAddr, String), Str return Ok((ip, ip.to_string())); } - let mut addrs = tokio::net::lookup_host((normalized_target, 80)) - .await - .map_err(|err| format!("unable to resolve target '{normalized_target}': {err}"))?; + let lookup = timeout( + PROBE_DNS_RESOLUTION_TIMEOUT, + tokio::net::lookup_host((normalized_target, 80)), + ) + .await + .map_err(|_| format!("timed out resolving target '{normalized_target}'"))?; + + let mut addrs = + lookup.map_err(|err| format!("unable to resolve target '{normalized_target}': {err}"))?; let Some(addr) = addrs.next() else { return Err(format!("unable to resolve target '{normalized_target}'")); }; @@ -540,7 +597,11 @@ async fn resolve_target(normalized_target: &str) -> Result<(IpAddr, String), Str #[cfg(test)] mod tests { - use super::{ProbeKind, ProbeRequest, duration_to_millis, normalize_target}; + use super::{ + PROBE_DNS_RESOLUTION_TIMEOUT, PROBE_MANAGER_REPLY_GRACE, PROBE_WORKER_GRACE, ProbeKind, + ProbeRequest, duration_to_millis, normalize_target, probe_batch_wait_timeout, + probe_worker_timeout, + }; use std::time::Duration; #[test] @@ -576,4 +637,40 @@ mod tests { assert_eq!(reachability.kind, ProbeKind::Reachability); assert_eq!(rtt.kind, ProbeKind::RoundTripTime); } + + #[test] + fn probe_deadlines_include_dns_probe_and_reply_grace() { + let request = ProbeRequest::reachability( + "example.com", + super::ProbeClass::UiMonitor, + Duration::from_millis(750), + ); + assert_eq!( + probe_worker_timeout(request.timeout), + PROBE_DNS_RESOLUTION_TIMEOUT + Duration::from_millis(750) + PROBE_WORKER_GRACE + ); + assert_eq!( + probe_batch_wait_timeout(&[request]), + PROBE_DNS_RESOLUTION_TIMEOUT + + Duration::from_millis(750) + + PROBE_WORKER_GRACE + + PROBE_MANAGER_REPLY_GRACE + ); + } + + #[test] + fn probe_batch_deadline_scales_for_multiple_probe_waves() { + let requests = vec![ + ProbeRequest::reachability( + "example.com", + super::ProbeClass::UiMonitor, + Duration::from_secs(1), + ); + 257 + ]; + assert_eq!( + probe_batch_wait_timeout(&requests), + probe_worker_timeout(Duration::from_secs(1)) * 2 + PROBE_MANAGER_REPLY_GRACE + ); + } } diff --git a/src/rust/lqos_topology/src/runtime.rs b/src/rust/lqos_topology/src/runtime.rs index 1a8b763d..103f179f 100644 --- a/src/rust/lqos_topology/src/runtime.rs +++ b/src/rust/lqos_topology/src/runtime.rs @@ -1,5 +1,5 @@ use anyhow::{Context, Result}; -use lqos_bus::{BusRequest, BusResponse, bus_request}; +use lqos_bus::{BusRequest, BusResponse, bus_request_with_timeout}; use lqos_config::{ TopologyAttachmentEndpointStatus, TopologyAttachmentHealthEntry, TopologyAttachmentHealthStateFile, TopologyAttachmentHealthStatus, @@ -21,6 +21,7 @@ use crate::{ }; const TOPOLOGY_PROBE_MAX_AGE_MS: u64 = 250; +const TOPOLOGY_PROBE_BUS_TIMEOUT: Duration = Duration::from_secs(8); fn now_unix() -> Option { SystemTime::now() @@ -120,10 +121,13 @@ async fn probe_specs( return Ok(HashMap::new()); } - let responses = bus_request(vec![BusRequest::ProbeBatch { - requests: probe_requests, - max_age_ms: TOPOLOGY_PROBE_MAX_AGE_MS, - }]) + let responses = bus_request_with_timeout( + vec![BusRequest::ProbeBatch { + requests: probe_requests, + max_age_ms: TOPOLOGY_PROBE_MAX_AGE_MS, + }], + TOPOLOGY_PROBE_BUS_TIMEOUT, + ) .await .map_err(|err| anyhow::anyhow!("unable to query shared probe manager: {err}"))?; let Some(response) = responses.into_iter().next() else { diff --git a/src/rust/lqosd/src/node_manager/local_api/config.rs b/src/rust/lqosd/src/node_manager/local_api/config.rs index a024764b..f8d7f9da 100644 --- a/src/rust/lqosd/src/node_manager/local_api/config.rs +++ b/src/rust/lqosd/src/node_manager/local_api/config.rs @@ -7,7 +7,7 @@ use axum::body::Bytes; use axum::http::StatusCode; use axum::http::header; use default_net::get_interfaces; -use lqos_bus::{BusRequest, BusResponse, bus_request}; +use lqos_bus::{BusRequest, BusResponse, bus_request_with_timeout}; use lqos_config::{ Config, ConfigShapedDevices, NetworkJson, ShapedDevice, UserRole, WebUser, WebUsers, }; @@ -17,13 +17,14 @@ use serde_json::Value; use std::collections::{BTreeMap, BTreeSet, HashSet}; use std::io::{Cursor, ErrorKind, Write}; use std::path::{Path, PathBuf}; -use std::time::{SystemTime, UNIX_EPOCH}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; const COBRAND_FILE_NAME: &str = "cobrand.png"; const COBRAND_DISPLAY_HEIGHT_PX: u64 = 48; const COBRAND_MAX_DISPLAY_WIDTH_PX: u64 = 176; const COBRAND_MAX_DECODE_BYTES: usize = 64 * 1024 * 1024; const PNG_SIGNATURE: &[u8; 8] = b"\x89PNG\r\n\x1a\n"; +const CONFIG_BUS_REQUEST_TIMEOUT: Duration = Duration::from_secs(10); #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum CobrandUploadValidationError { @@ -802,9 +803,12 @@ pub async fn update_lqosd_config_data( let existing = lqos_config::load_config().map_err(|_| "Unable to load the current config".to_string())?; apply_secret_updates(existing.as_ref(), &mut config, &clear_secrets); - let mut responses = bus_request(vec![BusRequest::UpdateLqosdConfig(Box::new(config))]) - .await - .map_err(|err| format!("Unable to update config: {err}"))?; + let mut responses = bus_request_with_timeout( + vec![BusRequest::UpdateLqosdConfig(Box::new(config))], + CONFIG_BUS_REQUEST_TIMEOUT, + ) + .await + .map_err(|err| format!("Unable to update config: {err}"))?; match responses.pop() { Some(BusResponse::Ack) => Ok(()), diff --git a/src/rust/lqosd/src/node_manager/local_api/lts.rs b/src/rust/lqosd/src/node_manager/local_api/lts.rs index baca3ceb..dbb02519 100644 --- a/src/rust/lqosd/src/node_manager/local_api/lts.rs +++ b/src/rust/lqosd/src/node_manager/local_api/lts.rs @@ -7,7 +7,7 @@ use crate::node_manager::local_api::circuit_count; use axum::http::StatusCode; pub use last_24_hours::*; use lqos_bus::LtsCapabilitiesSummary; -use lqos_bus::{BusRequest, bus_request}; +use lqos_bus::{BusRequest, bus_request_with_timeout}; use lqos_config::load_config; use serde::{Deserialize, Serialize}; pub use shaper_status::ShaperStatus; @@ -18,6 +18,8 @@ use tracing::{debug, info, warn}; use uuid::Uuid; const SIGNUP_POLL_INTERVAL: Duration = Duration::from_secs(10); +const CONFIG_BUS_REQUEST_TIMEOUT: Duration = Duration::from_secs(10); +const SERVICE_RESTART_TIMEOUT: Duration = Duration::from_secs(10); const INSIGHT_FREE_TRIAL_STATUS_CODE: i32 = 2; const INSIGHT_TRIAL_DAYS_HINT: i32 = 30; @@ -132,9 +134,15 @@ async fn apply_insight_license( .clone(); cfg.long_term_stats.gather_stats = true; cfg.long_term_stats.license_key = Some(license_key.clone()); - bus_request(vec![BusRequest::UpdateLqosdConfig(Box::new(cfg))]) - .await - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + bus_request_with_timeout( + vec![BusRequest::UpdateLqosdConfig(Box::new(cfg))], + CONFIG_BUS_REQUEST_TIMEOUT, + ) + .await + .map_err(|err| { + warn!("Unable to update config with Insight license: {err}"); + StatusCode::GATEWAY_TIMEOUT + })?; crate::lts2_sys::capabilities::set_signup_bootstrap_active(false); crate::lts2_sys::capabilities::clear_bootstrap_suppression(); crate::lts2_sys::capabilities::wake_control_channel(); @@ -147,9 +155,21 @@ async fn apply_insight_license( info!("LQOSD configuration updated with new Insight license key."); if restart_lqos_api { - let _ = Command::new("/bin/systemctl") - .args(["restart", "lqos_api"]) - .output(); + let restart = tokio::time::timeout( + SERVICE_RESTART_TIMEOUT, + tokio::task::spawn_blocking(|| { + Command::new("/bin/systemctl") + .args(["restart", "lqos_api"]) + .output() + }), + ) + .await; + match restart { + Ok(Ok(Ok(_))) => {} + Ok(Ok(Err(err))) => warn!("Unable to restart lqos_api: {err}"), + Ok(Err(err)) => warn!("lqos_api restart task failed: {err}"), + Err(_) => warn!("Timed out restarting lqos_api"), + } } Ok(()) diff --git a/src/rust/lqosd/src/node_manager/local_api/lts/last_24_hours.rs b/src/rust/lqosd/src/node_manager/local_api/lts/last_24_hours.rs index 151f5697..55757df5 100644 --- a/src/rust/lqosd/src/node_manager/local_api/lts/last_24_hours.rs +++ b/src/rust/lqosd/src/node_manager/local_api/lts/last_24_hours.rs @@ -1,8 +1,13 @@ use crate::node_manager::shaper_queries_actor::ShaperQueryCommand; use axum::http::StatusCode; use serde::{Deserialize, Serialize}; +use std::time::Duration; +use tokio::time::timeout; use tracing::warn; +const SHAPER_QUERY_RESPONSE_TIMEOUT: Duration = Duration::from_secs(35); +const SHAPER_QUERY_SEND_TIMEOUT: Duration = Duration::from_secs(5); + #[derive(Serialize, Deserialize, Copy, Clone, Debug)] pub struct ThroughputData { time: i64, // Unix timestamp @@ -136,20 +141,54 @@ pub struct RecentMedians { pub last_week: (i64, i64), } +async fn wait_for_shaper_response( + context: &'static str, + rx: tokio::sync::oneshot::Receiver>, +) -> Result, StatusCode> { + match timeout(SHAPER_QUERY_RESPONSE_TIMEOUT, rx).await { + Ok(Ok(response)) => Ok(response), + Ok(Err(err)) => { + warn!("Error getting {context}: {err:?}"); + Err(StatusCode::INTERNAL_SERVER_ERROR) + } + Err(_) => { + warn!("Timed out waiting for {context}"); + Err(StatusCode::GATEWAY_TIMEOUT) + } + } +} + +async fn send_shaper_query( + context: &'static str, + shaper_query: &tokio::sync::mpsc::Sender, + command: ShaperQueryCommand, +) -> Result<(), StatusCode> { + match timeout(SHAPER_QUERY_SEND_TIMEOUT, shaper_query.send(command)).await { + Ok(Ok(())) => Ok(()), + Ok(Err(_)) => { + warn!("Error sending {context} query"); + Err(StatusCode::INTERNAL_SERVER_ERROR) + } + Err(_) => { + warn!("Timed out queueing {context} query"); + Err(StatusCode::GATEWAY_TIMEOUT) + } + } +} + pub async fn throughput_period_data( shaper_query: tokio::sync::mpsc::Sender, seconds: i32, ) -> Result, StatusCode> { super::insight_gate().await?; let (tx, rx) = tokio::sync::oneshot::channel(); - shaper_query - .send(ShaperQueryCommand::ShaperThroughput { seconds, reply: tx }) - .await - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; - let throughput = rx.await.map_err(|e| { - warn!("Error getting total throughput: {:?}", e); - StatusCode::INTERNAL_SERVER_ERROR - })?; + send_shaper_query( + "throughput", + &shaper_query, + ShaperQueryCommand::ShaperThroughput { seconds, reply: tx }, + ) + .await?; + let throughput = wait_for_shaper_response("total throughput", rx).await?; Ok(throughput) } @@ -159,17 +198,13 @@ pub async fn packets_period_data( ) -> Result, StatusCode> { super::insight_gate().await?; let (tx, rx) = tokio::sync::oneshot::channel(); - shaper_query - .send(ShaperQueryCommand::ShaperPackets { seconds, reply: tx }) - .await - .map_err(|_| { - warn!("Error sending packets period"); - StatusCode::INTERNAL_SERVER_ERROR - })?; - let throughput = rx.await.map_err(|e| { - warn!("Error getting packets period: {:?}", e); - StatusCode::INTERNAL_SERVER_ERROR - })?; + send_shaper_query( + "packets period", + &shaper_query, + ShaperQueryCommand::ShaperPackets { seconds, reply: tx }, + ) + .await?; + let throughput = wait_for_shaper_response("packets period", rx).await?; Ok(throughput) } @@ -179,17 +214,13 @@ pub async fn percent_shaped_period_data( ) -> Result, StatusCode> { super::insight_gate().await?; let (tx, rx) = tokio::sync::oneshot::channel(); - shaper_query - .send(ShaperQueryCommand::ShaperPercent { seconds, reply: tx }) - .await - .map_err(|_| { - warn!("Error sending percent shaped period"); - StatusCode::INTERNAL_SERVER_ERROR - })?; - let throughput = rx.await.map_err(|e| { - warn!("Error getting percent shaped: {:?}", e); - StatusCode::INTERNAL_SERVER_ERROR - })?; + send_shaper_query( + "percent shaped period", + &shaper_query, + ShaperQueryCommand::ShaperPercent { seconds, reply: tx }, + ) + .await?; + let throughput = wait_for_shaper_response("percent shaped", rx).await?; Ok(throughput) } @@ -199,17 +230,13 @@ pub async fn percent_flows_period_data( ) -> Result, StatusCode> { super::insight_gate().await?; let (tx, rx) = tokio::sync::oneshot::channel(); - shaper_query - .send(ShaperQueryCommand::ShaperFlows { seconds, reply: tx }) - .await - .map_err(|_| { - warn!("Error sending flows period"); - StatusCode::INTERNAL_SERVER_ERROR - })?; - let throughput = rx.await.map_err(|e| { - warn!("Error getting flows: {:?}", e); - StatusCode::INTERNAL_SERVER_ERROR - })?; + send_shaper_query( + "flows period", + &shaper_query, + ShaperQueryCommand::ShaperFlows { seconds, reply: tx }, + ) + .await?; + let throughput = wait_for_shaper_response("flows", rx).await?; Ok(throughput) } @@ -218,19 +245,14 @@ pub async fn rtt_histo_period_data( seconds: i32, ) -> Result, StatusCode> { super::insight_gate().await?; - tracing::error!("rtt_histo_period"); let (tx, rx) = tokio::sync::oneshot::channel(); - shaper_query - .send(ShaperQueryCommand::ShaperRttHistogram { seconds, reply: tx }) - .await - .map_err(|_| { - warn!("Error sending flows period"); - StatusCode::INTERNAL_SERVER_ERROR - })?; - let throughput = rx.await.map_err(|e| { - warn!("Error getting flows: {:?}", e); - StatusCode::INTERNAL_SERVER_ERROR - })?; + send_shaper_query( + "RTT histogram period", + &shaper_query, + ShaperQueryCommand::ShaperRttHistogram { seconds, reply: tx }, + ) + .await?; + let throughput = wait_for_shaper_response("RTT histogram", rx).await?; Ok(throughput) } @@ -239,19 +261,14 @@ pub async fn top10_downloaders_period_data( seconds: i32, ) -> Result, StatusCode> { super::insight_gate().await?; - tracing::error!("rtt_histo_period"); let (tx, rx) = tokio::sync::oneshot::channel(); - shaper_query - .send(ShaperQueryCommand::ShaperTopDownloaders { seconds, reply: tx }) - .await - .map_err(|_| { - warn!("Error sending flows period"); - StatusCode::INTERNAL_SERVER_ERROR - })?; - let throughput = rx.await.map_err(|e| { - warn!("Error getting flows: {:?}", e); - StatusCode::INTERNAL_SERVER_ERROR - })?; + send_shaper_query( + "top downloaders period", + &shaper_query, + ShaperQueryCommand::ShaperTopDownloaders { seconds, reply: tx }, + ) + .await?; + let throughput = wait_for_shaper_response("top downloaders", rx).await?; Ok(throughput) } @@ -260,19 +277,14 @@ pub async fn worst10_rtt_period_data( seconds: i32, ) -> Result, StatusCode> { super::insight_gate().await?; - tracing::error!("rtt_histo_period"); let (tx, rx) = tokio::sync::oneshot::channel(); - shaper_query - .send(ShaperQueryCommand::ShaperWorstRtt { seconds, reply: tx }) - .await - .map_err(|_| { - warn!("Error sending flows period"); - StatusCode::INTERNAL_SERVER_ERROR - })?; - let throughput = rx.await.map_err(|e| { - warn!("Error getting flows: {:?}", e); - StatusCode::INTERNAL_SERVER_ERROR - })?; + send_shaper_query( + "worst RTT period", + &shaper_query, + ShaperQueryCommand::ShaperWorstRtt { seconds, reply: tx }, + ) + .await?; + let throughput = wait_for_shaper_response("worst RTT", rx).await?; Ok(throughput) } @@ -281,19 +293,14 @@ pub async fn worst10_rxmit_period_data( seconds: i32, ) -> Result, StatusCode> { super::insight_gate().await?; - tracing::error!("rtt_histo_period"); let (tx, rx) = tokio::sync::oneshot::channel(); - shaper_query - .send(ShaperQueryCommand::ShaperWorstRxmit { seconds, reply: tx }) - .await - .map_err(|_| { - warn!("Error sending flows period"); - StatusCode::INTERNAL_SERVER_ERROR - })?; - let throughput = rx.await.map_err(|e| { - warn!("Error getting flows: {:?}", e); - StatusCode::INTERNAL_SERVER_ERROR - })?; + send_shaper_query( + "worst retransmits period", + &shaper_query, + ShaperQueryCommand::ShaperWorstRxmit { seconds, reply: tx }, + ) + .await?; + let throughput = wait_for_shaper_response("worst retransmits", rx).await?; Ok(throughput) } @@ -302,19 +309,14 @@ pub async fn top10_flows_period_data( seconds: i32, ) -> Result, StatusCode> { super::insight_gate().await?; - tracing::error!("rtt_histo_period"); let (tx, rx) = tokio::sync::oneshot::channel(); - shaper_query - .send(ShaperQueryCommand::ShaperTopFlows { seconds, reply: tx }) - .await - .map_err(|_| { - warn!("Error sending flows period"); - StatusCode::INTERNAL_SERVER_ERROR - })?; - let throughput = rx.await.map_err(|e| { - warn!("Error getting flows: {:?}", e); - StatusCode::INTERNAL_SERVER_ERROR - })?; + send_shaper_query( + "top flows period", + &shaper_query, + ShaperQueryCommand::ShaperTopFlows { seconds, reply: tx }, + ) + .await?; + let throughput = wait_for_shaper_response("top flows", rx).await?; Ok(throughput) } @@ -322,19 +324,14 @@ pub async fn recent_medians_data( shaper_query: tokio::sync::mpsc::Sender, ) -> Result, StatusCode> { super::insight_gate().await?; - tracing::debug!("rtt_histo_period"); let (tx, rx) = tokio::sync::oneshot::channel(); - shaper_query - .send(ShaperQueryCommand::ShaperRecentMedian { reply: tx }) - .await - .map_err(|_| { - warn!("Error sending flows period"); - StatusCode::INTERNAL_SERVER_ERROR - })?; - let throughput = rx.await.map_err(|e| { - warn!("Error getting flows: {:?}", e); - StatusCode::INTERNAL_SERVER_ERROR - })?; + send_shaper_query( + "recent median", + &shaper_query, + ShaperQueryCommand::ShaperRecentMedian { reply: tx }, + ) + .await?; + let throughput = wait_for_shaper_response("recent medians", rx).await?; Ok(throughput) } @@ -344,16 +341,12 @@ pub async fn cake_period_data( ) -> Result, StatusCode> { super::insight_gate().await?; let (tx, rx) = tokio::sync::oneshot::channel(); - shaper_query - .send(ShaperQueryCommand::CakeTotals { seconds, reply: tx }) - .await - .map_err(|_| { - warn!("Error sending cake stats period"); - StatusCode::INTERNAL_SERVER_ERROR - })?; - let response = rx.await.map_err(|e| { - warn!("Error getting cake stats: {:?}", e); - StatusCode::INTERNAL_SERVER_ERROR - })?; + send_shaper_query( + "cake stats period", + &shaper_query, + ShaperQueryCommand::CakeTotals { seconds, reply: tx }, + ) + .await?; + let response = wait_for_shaper_response("cake stats", rx).await?; Ok(response) } diff --git a/src/rust/lqosd/src/node_manager/shaper_queries_actor/queries.rs b/src/rust/lqosd/src/node_manager/shaper_queries_actor/queries.rs index 1b5a1298..b3628ae1 100644 --- a/src/rust/lqosd/src/node_manager/shaper_queries_actor/queries.rs +++ b/src/rust/lqosd/src/node_manager/shaper_queries_actor/queries.rs @@ -9,8 +9,11 @@ use crate::node_manager::shaper_queries_actor::ShaperQueryCommand; use crate::node_manager::shaper_queries_actor::caches::Caches; use std::time::Duration; use tokio::sync::oneshot; +use tokio::time::timeout; use tracing::{info, warn}; +const HISTORY_QUERY_ATTEMPT_TIMEOUT: Duration = Duration::from_secs(30); + #[derive(Clone)] struct HistoryClient { control_tx: tokio::sync::mpsc::Sender, @@ -26,21 +29,30 @@ impl HistoryClient { request: RemoteInsightRequest, ) -> Result { let (tx, rx) = oneshot::channel(); - if self - .control_tx - .send(ControlChannelCommand::FetchHistory { + let send_result = timeout( + HISTORY_QUERY_ATTEMPT_TIMEOUT, + self.control_tx.send(ControlChannelCommand::FetchHistory { request, responder: tx, - }) - .await - .is_err() - { - return Err(()); + }), + ) + .await; + match send_result { + Ok(Ok(())) => {} + Ok(Err(_)) => return Err(()), + Err(_) => { + warn!("Timed out queueing Insight history request"); + return Err(()); + } } - match rx.await { - Ok(result) => result, - Err(_) => Err(()), + match timeout(HISTORY_QUERY_ATTEMPT_TIMEOUT, rx).await { + Ok(Ok(result)) => result, + Ok(Err(_)) => Err(()), + Err(_) => { + warn!("Timed out waiting for Insight history response"); + Err(()) + } } } } @@ -89,6 +101,9 @@ macro_rules! shaper_query { break; } Err(()) => { + if my_reply.is_closed() { + break; + } attempts += 1; if attempts >= 3 { warn!("History request exhausted retries"); diff --git a/src/rust/lqosd/src/node_manager/ws.rs b/src/rust/lqosd/src/node_manager/ws.rs index 4e359bd8..7481e364 100644 --- a/src/rust/lqosd/src/node_manager/ws.rs +++ b/src/rust/lqosd/src/node_manager/ws.rs @@ -7,7 +7,9 @@ use std::collections::{BTreeMap, HashSet}; use std::sync::Arc; +use std::time::Duration; +use crate::lts2_sys::control_channel::ControlChannelCommand; use crate::node_manager::auth::{LoginResult, login_from_token}; use crate::node_manager::local_api::{ circuit, circuit_count, config, cpu_affinity, dashboard_themes, device_counts, directories, @@ -55,6 +57,21 @@ mod ticker; const WS_VERSION: &str = include_str!("../../../../VERSION_STRING"); const HANDSHAKE_TIMEOUT_SECS: u64 = 10; +const CONTROL_CHANNEL_SEND_TIMEOUT: Duration = Duration::from_secs(5); + +async fn send_control_command( + control_tx: &tokio::sync::mpsc::Sender, + command: ControlChannelCommand, +) -> bool { + match tokio::time::timeout(CONTROL_CHANNEL_SEND_TIMEOUT, control_tx.send(command)).await { + Ok(Ok(())) => true, + Ok(Err(_)) => false, + Err(_) => { + warn!("Timed out queueing websocket control-channel command"); + false + } + } +} /// Provides an Axum router for the websocket system. Exposes a single /ws route that supports /// pubsub subscriptions and private commands. @@ -742,14 +759,13 @@ async fn receive_channel_message( _ => { let (reply_tx, reply_rx) = tokio::sync::oneshot::channel(); let control_tx = request_state.private_state.control_tx(); - if control_tx - .send( - crate::lts2_sys::control_channel::ControlChannelCommand::SupportTicketList { - responder: reply_tx, - }, - ) - .await - .is_err() + if !send_control_command( + &control_tx, + ControlChannelCommand::SupportTicketList { + responder: reply_tx, + }, + ) + .await { if send_ws_response( &tx, @@ -766,8 +782,11 @@ async fn receive_channel_message( tokio::time::timeout(std::time::Duration::from_secs(30), reply_rx).await; match result { Ok(Ok(Ok(tickets))) => { - if send_ws_response(&tx, WsResponse::SupportTicketListResult { tickets }) - .await + if send_ws_response( + &tx, + WsResponse::SupportTicketListResult { tickets }, + ) + .await { return true; } @@ -816,15 +835,14 @@ async fn receive_channel_message( _ => { let (reply_tx, reply_rx) = tokio::sync::oneshot::channel(); let control_tx = request_state.private_state.control_tx(); - if control_tx - .send( - crate::lts2_sys::control_channel::ControlChannelCommand::SupportTicketGet { - ticket_id, - responder: reply_tx, - }, - ) - .await - .is_err() + if !send_control_command( + &control_tx, + ControlChannelCommand::SupportTicketGet { + ticket_id, + responder: reply_tx, + }, + ) + .await { if send_ws_response( &tx, @@ -899,18 +917,17 @@ async fn receive_channel_message( } else { let (reply_tx, reply_rx) = tokio::sync::oneshot::channel(); let control_tx = request_state.private_state.control_tx(); - if control_tx - .send( - crate::lts2_sys::control_channel::ControlChannelCommand::SupportTicketCreate { - subject, - priority, - body, - commentor, - responder: reply_tx, - }, - ) - .await - .is_err() + if !send_control_command( + &control_tx, + ControlChannelCommand::SupportTicketCreate { + subject, + priority, + body, + commentor, + responder: reply_tx, + }, + ) + .await { if send_ws_response( &tx, @@ -923,12 +940,15 @@ async fn receive_channel_message( return true; } } else { - let result = tokio::time::timeout(std::time::Duration::from_secs(30), reply_rx) - .await; + let result = + tokio::time::timeout(std::time::Duration::from_secs(30), reply_rx).await; match result { Ok(Ok(Ok(ticket))) => { - if send_ws_response(&tx, WsResponse::SupportTicketCreateResult { ticket }) - .await + if send_ws_response( + &tx, + WsResponse::SupportTicketCreateResult { ticket }, + ) + .await { return true; } @@ -988,18 +1008,17 @@ async fn receive_channel_message( .unwrap_or(0); let (reply_tx, reply_rx) = tokio::sync::oneshot::channel(); let control_tx = request_state.private_state.control_tx(); - if control_tx - .send( - crate::lts2_sys::control_channel::ControlChannelCommand::SupportTicketAddComment { - ticket_id, - commentor, - body, - date, - responder: reply_tx, - }, - ) - .await - .is_err() + if !send_control_command( + &control_tx, + ControlChannelCommand::SupportTicketAddComment { + ticket_id, + commentor, + body, + date, + responder: reply_tx, + }, + ) + .await { if send_ws_response( &tx, @@ -1012,8 +1031,8 @@ async fn receive_channel_message( return true; } } else { - let result = tokio::time::timeout(std::time::Duration::from_secs(30), reply_rx) - .await; + let result = + tokio::time::timeout(std::time::Duration::from_secs(30), reply_rx).await; match result { Ok(Ok(Ok(()))) => { if send_ws_response( diff --git a/src/rust/lqosd/src/node_manager/ws/single_user_channels.rs b/src/rust/lqosd/src/node_manager/ws/single_user_channels.rs index fc4c8eba..43d655a9 100644 --- a/src/rust/lqosd/src/node_manager/ws/single_user_channels.rs +++ b/src/rust/lqosd/src/node_manager/ws/single_user_channels.rs @@ -6,6 +6,7 @@ pub(crate) mod flows_by_circuit; mod ping_monitor; mod tree_attached_circuits; +use crate::lts2_sys::control_channel::ControlChannelCommand; use crate::node_manager::ws::messages::{PrivateRequest, WsResponse, encode_ws_message}; use crate::node_manager::ws::single_user_channels::cake_watcher::cake_watcher; use crate::node_manager::ws::single_user_channels::circuit::circuit_watcher; @@ -13,17 +14,52 @@ use crate::node_manager::ws::single_user_channels::circuit_metrics::watch_circui use crate::node_manager::ws::single_user_channels::ping_monitor::ping_monitor; use crate::node_manager::ws::single_user_channels::tree_attached_circuits::watch_tree_attached_circuits; use lqos_probe::ProbeClient; +use std::{sync::Arc, time::Duration}; use tokio::spawn; -use tokio::sync::mpsc::Sender; -use tracing::info; +use tokio::sync::mpsc::{Sender, error::TrySendError}; +use tokio::time::timeout; +use tracing::{debug, info}; + +const CONTROL_CHANNEL_SEND_TIMEOUT: Duration = Duration::from_secs(5); + +/// Sends a private watcher payload without waiting behind a slow websocket. +pub(super) fn try_send_private_payload( + tx: &Sender>>, + payload: Arc>, + channel_name: &'static str, +) -> bool { + match tx.try_send(payload) { + Ok(()) => true, + Err(TrySendError::Full(_)) => { + debug!("{channel_name} outbound queue full; stopping watcher"); + false + } + Err(TrySendError::Closed(_)) => { + debug!("{channel_name} channel closed"); + false + } + } +} + +async fn send_control_command( + control_tx: &tokio::sync::mpsc::Sender, + command: ControlChannelCommand, + context: &'static str, +) { + match timeout(CONTROL_CHANNEL_SEND_TIMEOUT, control_tx.send(command)).await { + Ok(Ok(())) => {} + Ok(Err(_)) => debug!("{context} control channel is closed"), + Err(_) => debug!("{context} timed out queueing control channel command"), + } +} pub struct PrivateState { - tx: Sender>>, + tx: Sender>>, bus_tx: Sender<( tokio::sync::oneshot::Sender, lqos_bus::BusRequest, )>, - control_tx: tokio::sync::mpsc::Sender, + control_tx: tokio::sync::mpsc::Sender, probe_client: ProbeClient, browser_language: Option, chatbot_request: Option, @@ -41,9 +77,7 @@ impl PrivateState { tokio::sync::oneshot::Sender, lqos_bus::BusRequest, )>, - control_tx: tokio::sync::mpsc::Sender< - crate::lts2_sys::control_channel::ControlChannelCommand, - >, + control_tx: tokio::sync::mpsc::Sender, probe_client: ProbeClient, browser_language: Option, ) -> Self { @@ -62,9 +96,7 @@ impl PrivateState { } } - pub fn control_tx( - &self, - ) -> tokio::sync::mpsc::Sender { + pub fn control_tx(&self) -> tokio::sync::mpsc::Sender { self.control_tx.clone() } @@ -205,7 +237,7 @@ impl PrivateState { message: message.to_string(), }; if let Ok(payload) = encode_ws_message(&response) { - let _ = self.tx.send(payload).await; + let _ = try_send_private_payload(&self.tx, payload, "Chatbot"); } return; } @@ -220,7 +252,7 @@ impl PrivateState { let text = String::from_utf8_lossy(&chunk).to_string(); let response = WsResponse::ChatbotChunk { text }; if let Ok(payload) = encode_ws_message(&response) { - if to_client.send(payload).await.is_err() { + if !try_send_private_payload(&to_client, payload, "Chatbot") { break; } } else { @@ -229,17 +261,17 @@ impl PrivateState { } }); - let _ = self - .control_tx - .send( - crate::lts2_sys::control_channel::ControlChannelCommand::StartChat { - request_id, - browser_ts_ms, - browser_language: self.browser_language.clone(), - stream: stream_tx, - }, - ) - .await; + send_control_command( + &self.control_tx, + ControlChannelCommand::StartChat { + request_id, + browser_ts_ms, + browser_language: self.browser_language.clone(), + stream: stream_tx, + }, + "StartChat", + ) + .await; info!( "[chatbot] starting session request_id={} browser_ts_ms={:?}", request_id, browser_ts_ms @@ -250,15 +282,12 @@ impl PrivateState { let Some(request_id) = self.chatbot_request else { return; }; - let _ = self - .control_tx - .send( - crate::lts2_sys::control_channel::ControlChannelCommand::ChatSend { - request_id, - text, - }, - ) - .await; + send_control_command( + &self.control_tx, + ControlChannelCommand::ChatSend { request_id, text }, + "ChatSend", + ) + .await; } } diff --git a/src/rust/lqosd/src/node_manager/ws/single_user_channels/cake_watcher.rs b/src/rust/lqosd/src/node_manager/ws/single_user_channels/cake_watcher.rs index 1267eac7..d6a63db5 100644 --- a/src/rust/lqosd/src/node_manager/ws/single_user_channels/cake_watcher.rs +++ b/src/rust/lqosd/src/node_manager/ws/single_user_channels/cake_watcher.rs @@ -1,4 +1,5 @@ use crate::node_manager::ws::messages::{WsResponse, encode_ws_message}; +use crate::node_manager::ws::single_user_channels::try_send_private_payload; use lqos_queue_tracker::{add_watched_queue, get_raw_circuit_data, still_watching}; pub(super) async fn cake_watcher( @@ -16,11 +17,10 @@ pub(super) async fn cake_watcher( if let lqos_bus::BusResponse::RawQueueData(Some(msg)) = get_raw_circuit_data(&circuit_id) { let response = WsResponse::CakeWatcher { data: *msg }; - if let Ok(payload) = encode_ws_message(&response) { - let send_result = tx.send(payload).await; - if send_result.is_err() { - break; - } + if let Ok(payload) = encode_ws_message(&response) + && !try_send_private_payload(&tx, payload, "CakeWatcher") + { + break; } } } diff --git a/src/rust/lqosd/src/node_manager/ws/single_user_channels/circuit.rs b/src/rust/lqosd/src/node_manager/ws/single_user_channels/circuit.rs index b5a1eccd..4881b460 100644 --- a/src/rust/lqosd/src/node_manager/ws/single_user_channels/circuit.rs +++ b/src/rust/lqosd/src/node_manager/ws/single_user_channels/circuit.rs @@ -1,5 +1,6 @@ use crate::node_manager::local_api::circuit_activity::{CircuitSummaryData, circuit_flow_counts}; use crate::node_manager::ws::messages::{CircuitDevicesResult, WsResponse, encode_ws_message}; +use crate::node_manager::ws::single_user_channels::try_send_private_payload; use crate::node_manager::ws::ticker::all_circuits; use crate::rtt_exclusions; use crate::throughput_tracker::{circuit_current_qoo, circuit_current_rtt_p50_nanos}; @@ -121,7 +122,7 @@ pub(super) async fn circuit_watcher( }; if let Ok(payload) = encode_ws_message(&response) { - if tx.send(payload).await.is_err() { + if !try_send_private_payload(&tx, payload, "CircuitWatcher") { info!("CircuitWatcher channel is gone"); break; } diff --git a/src/rust/lqosd/src/node_manager/ws/single_user_channels/circuit_metrics.rs b/src/rust/lqosd/src/node_manager/ws/single_user_channels/circuit_metrics.rs index d03dcbd2..c950b88c 100644 --- a/src/rust/lqosd/src/node_manager/ws/single_user_channels/circuit_metrics.rs +++ b/src/rust/lqosd/src/node_manager/ws/single_user_channels/circuit_metrics.rs @@ -2,9 +2,9 @@ use crate::node_manager::local_api::circuit_live::{ CircuitLiveMetrics, CircuitMetricsQuery, circuit_live_metrics, }; use crate::node_manager::ws::messages::{WsResponse, encode_ws_message}; +use crate::node_manager::ws::single_user_channels::try_send_private_payload; use std::time::Duration; use tokio::time::MissedTickBehavior; -use tracing::debug; pub(super) async fn watch_circuit_metrics( query: CircuitMetricsQuery, @@ -33,8 +33,7 @@ pub(super) async fn watch_circuit_metrics( if let Some(response) = response { match encode_ws_message(&response) { Ok(payload) => { - if tx.send(payload).await.is_err() { - debug!("CircuitMetrics watcher channel closed"); + if !try_send_private_payload(&tx, payload, "CircuitMetrics") { break; } } diff --git a/src/rust/lqosd/src/node_manager/ws/single_user_channels/ping_monitor.rs b/src/rust/lqosd/src/node_manager/ws/single_user_channels/ping_monitor.rs index 69133f21..ab0cf73e 100644 --- a/src/rust/lqosd/src/node_manager/ws/single_user_channels/ping_monitor.rs +++ b/src/rust/lqosd/src/node_manager/ws/single_user_channels/ping_monitor.rs @@ -1,8 +1,9 @@ use crate::node_manager::ws::messages::{PingState, WsResponse, encode_ws_message}; +use crate::node_manager::ws::single_user_channels::try_send_private_payload; use lqos_probe::{ProbeClass, ProbeClient}; use std::time::Duration; use tokio::time::MissedTickBehavior; -use tracing::{debug, info}; +use tracing::debug; const UI_MONITOR_PROBE_MAX_AGE: Duration = Duration::from_millis(250); @@ -32,13 +33,15 @@ pub(super) async fn ping_monitor( } }; - for ((ip, label), observation) in ip_addresses.iter().zip(observations) { + 'observations: for ((ip, label), observation) in ip_addresses.iter().zip(observations) { if observation.reachable { let ping_time = observation .rtt_ms .map(|rtt_ms| Duration::from_secs_f64(rtt_ms / 1000.0)) .unwrap_or_else(|| Duration::from_secs(0)); - send_alive(tx.clone(), ip.clone(), ping_time, label.clone()).await; + if !send_alive(&tx, ip.clone(), ping_time, label.clone()) { + break 'observations; + } } else { if let Some(error) = observation.error.as_deref() { debug!( @@ -46,7 +49,9 @@ pub(super) async fn ping_monitor( observation.normalized_target, error ); } - send_timeout(tx.clone(), ip.clone()).await; + if !send_timeout(&tx, ip.clone()) { + break 'observations; + } } } @@ -55,7 +60,7 @@ pub(super) async fn ping_monitor( result: PingState::ChannelTest, }; if let Ok(payload) = encode_ws_message(&channel_test) { - if tx.send(payload).await.is_err() { + if !try_send_private_payload(&tx, payload, "PingMonitor") { debug!("Channel is gone"); break; } @@ -65,24 +70,22 @@ pub(super) async fn ping_monitor( } } -async fn send_timeout(tx: tokio::sync::mpsc::Sender>>, ip: String) { +fn send_timeout(tx: &tokio::sync::mpsc::Sender>>, ip: String) -> bool { let result = WsResponse::PingMonitor { ip, result: PingState::NoResponse, }; - if let Ok(payload) = encode_ws_message(&result) - && tx.send(payload).await.is_err() - { - info!("Channel is gone"); - } + encode_ws_message(&result) + .map(|payload| try_send_private_payload(tx, payload, "PingMonitor")) + .unwrap_or(false) } -async fn send_alive( - tx: tokio::sync::mpsc::Sender>>, +fn send_alive( + tx: &tokio::sync::mpsc::Sender>>, ip: String, ping_time: Duration, label: String, -) { +) -> bool { let result = WsResponse::PingMonitor { ip, result: PingState::Ping { @@ -90,9 +93,7 @@ async fn send_alive( label, }, }; - if let Ok(payload) = encode_ws_message(&result) - && tx.send(payload).await.is_err() - { - info!("Channel is gone"); - } + encode_ws_message(&result) + .map(|payload| try_send_private_payload(tx, payload, "PingMonitor")) + .unwrap_or(false) } diff --git a/src/rust/lqosd/src/node_manager/ws/single_user_channels/tree_attached_circuits.rs b/src/rust/lqosd/src/node_manager/ws/single_user_channels/tree_attached_circuits.rs index ee7e791f..eacdf8f3 100644 --- a/src/rust/lqosd/src/node_manager/ws/single_user_channels/tree_attached_circuits.rs +++ b/src/rust/lqosd/src/node_manager/ws/single_user_channels/tree_attached_circuits.rs @@ -2,9 +2,9 @@ use crate::node_manager::local_api::tree_attached_circuits::{ TreeAttachedCircuitsPage, TreeAttachedCircuitsQuery, tree_attached_circuits, }; use crate::node_manager::ws::messages::{WsResponse, encode_ws_message}; +use crate::node_manager::ws::single_user_channels::try_send_private_payload; use std::time::Duration; use tokio::time::MissedTickBehavior; -use tracing::debug; pub(super) async fn watch_tree_attached_circuits( query: TreeAttachedCircuitsQuery, @@ -29,8 +29,7 @@ pub(super) async fn watch_tree_attached_circuits( if let Some(response) = response { match encode_ws_message(&response) { Ok(payload) => { - if tx.send(payload).await.is_err() { - debug!("TreeAttachedCircuits watcher channel closed"); + if !try_send_private_payload(&tx, payload, "TreeAttachedCircuits") { break; } } diff --git a/src/rust/lqosd/src/node_manager/ws/ticker/network_tree.rs b/src/rust/lqosd/src/node_manager/ws/ticker/network_tree.rs index 4c49a3b3..84906f7e 100644 --- a/src/rust/lqosd/src/node_manager/ws/ticker/network_tree.rs +++ b/src/rust/lqosd/src/node_manager/ws/ticker/network_tree.rs @@ -3,7 +3,42 @@ use crate::node_manager::ws::publish_subscribe::PubSub; use crate::node_manager::ws::published_channels::PublishedChannels; use lqos_bus::{BusReply, BusRequest, BusResponse, Circuit}; use std::sync::Arc; +use std::time::Duration; use tokio::sync::mpsc::Sender; +use tokio::time::timeout; + +const INTERNAL_BUS_TIMEOUT: Duration = Duration::from_secs(5); + +async fn request_internal_bus( + context: &'static str, + bus_tx: Sender<(tokio::sync::oneshot::Sender, BusRequest)>, + request: BusRequest, +) -> Option { + let (tx, rx) = tokio::sync::oneshot::channel::(); + match timeout(INTERNAL_BUS_TIMEOUT, bus_tx.send((tx, request))).await { + Ok(Ok(())) => {} + Ok(Err(err)) => { + tracing::warn!("{context}: failed to send request to bus: {err:?}"); + return None; + } + Err(_) => { + tracing::warn!("{context}: timed out queueing request to bus"); + return None; + } + } + + match timeout(INTERNAL_BUS_TIMEOUT, rx).await { + Ok(Ok(replies)) => Some(replies), + Ok(Err(err)) => { + tracing::warn!("{context}: failed to receive response from bus: {err:?}"); + None + } + Err(_) => { + tracing::warn!("{context}: timed out waiting for bus response"); + None + } + } +} pub async fn network_tree( channels: Arc, @@ -16,21 +51,10 @@ pub async fn network_tree( return; } - let (tx, rx) = tokio::sync::oneshot::channel::(); - let request = BusRequest::GetFullNetworkMap; - if let Err(e) = bus_tx.send((tx, request)).await { - tracing::warn!("NetworkTree: failed to send request to bus: {:?}", e); + let Some(replies) = + request_internal_bus("NetworkTree", bus_tx, BusRequest::GetFullNetworkMap).await + else { return; - } - let replies = match rx.await { - Ok(r) => r, - Err(e) => { - tracing::warn!( - "NetworkTree: failed to receive throughput from bus: {:?}", - e - ); - return; - } }; for reply in replies.responses.into_iter() { if let BusResponse::NetworkMap(nodes) = reply { @@ -43,21 +67,10 @@ pub async fn network_tree( pub async fn all_circuits( bus_tx: Sender<(tokio::sync::oneshot::Sender, BusRequest)>, ) -> Vec { - let (tx, rx) = tokio::sync::oneshot::channel::(); - let request = BusRequest::GetAllCircuits; - if let Err(e) = bus_tx.send((tx, request)).await { - tracing::warn!("AllCircuits: failed to send request to bus: {:?}", e); + let Some(replies) = + request_internal_bus("AllCircuits", bus_tx, BusRequest::GetAllCircuits).await + else { return Vec::new(); - } - let replies = match rx.await { - Ok(r) => r, - Err(e) => { - tracing::warn!( - "AllCircuits: failed to receive throughput from bus: {:?}", - e - ); - return Vec::new(); - } }; for reply in replies.responses.into_iter() { if let BusResponse::CircuitData(circuits) = reply {