Merge pull request #1070 from LibreQoE/ui_fix_websocket

Harden WebUI websocket recovery, queue diff resets, and Insight weight caching

docs-es/v2.0/libreqos-backend-architecture-es.md

@@ -270,14 +270,15 @@ flowchart TD
 3. Esperar mayor riesgo cuando coinciden muchos circuitos y muchos cambios estructurales.
 4. Diseñar la cadencia operativa para priorizar cambios incrementalmente seguros.
 
-### 7.4 Guardrails de seguridad para recargas completas
+### 7.4 Salvaguardas de seguridad para recargas completas
 
-Las recargas completas actuales de Bakery aplican dos verificaciones conservadoras de seguridad antes y durante reconstrucciones grandes de colas:
+Las recargas completas actuales de Bakery aplican verificaciones conservadoras de seguridad antes y durante reconstrucciones grandes de colas:
 
 1. Un preflight de qdisc estima los qdisc planificados por interfaz y además separa qdisc de infraestructura, hojas `cake` y hojas `fq_codel`.
 2. Ese mismo preflight aplica una proyección conservadora de memoria y bloquea de forma estricta las recargas completas claramente inseguras antes de arrancar `tc -batch`.
-3. Durante la aplicación chunked de una recarga completa, Bakery vuelve a revisar la memoria del host en los límites de chunk y aborta el resto de la aplicación si la memoria disponible cae por debajo de su piso de seguridad.
-4. Estos guardrails están sesgados intencionalmente hacia falsos positivos en recargas grandes para fallar temprano con diagnósticos en lugar de entrar en una espiral OOM.
+3. El piso de memoria de Bakery escala con la RAM del host: mantiene al menos 2 GiB disponibles, o un octavo de la RAM total en sistemas más grandes.
+4. Durante la aplicación por fragmentos de una recarga completa, Bakery vuelve a revisar la memoria del host en los límites de cada fragmento y aborta el resto de la aplicación si la memoria disponible cae por debajo del piso escalado más la memoria qdisc proyectada para el lote.
+5. Estas salvaguardas están sesgadas intencionalmente hacia falsos positivos en recargas grandes para fallar temprano con diagnósticos en lugar de entrar en una espiral OOM.
 
 ## 8) Límites de Diseño para Operadores
 

docs-es/v2.0/troubleshooting-es.md

@@ -253,7 +253,7 @@ WebUI muestra códigos legibles por máquina para triage rápido.
 | `MAPPED_CIRCUIT_LIMIT` | Bakery está forzando límite de circuitos mapeados. | Estado de licencia Insight y `journalctl -u lqosd` con requested/allowed/dropped. | Reducir circuitos mapeados o actualizar licencia/límites. |
 | `TC_U16_OVERFLOW` | IDs minor de clases/colas excedieron rango u16 de tc en una cola CPU. | `journalctl -u lqos_scheduler -u lqosd`, profundidad topológica y distribución por colas. | Aumentar paralelismo de colas y/o simplificar/rebalancear jerarquía. |
 | `TC_QDISC_CAPACITY` | Los qdisc autoasignados planificados exceden el presupuesto seguro por interfaz o el preflight conservador de seguridad de memoria de Bakery antes de aplicar. | Conteos estimados por interfaz, desglose por tipo de qdisc y campos de memoria en el contexto del urgent issue, `journalctl -u lqos_scheduler -u lqosd`, configuración `on_a_stick` y `monitor_only`. | Reducir la carga planificada de qdisc para esta ejecución (por ejemplo menos circuitos/dispositivos en la forma de prueba) antes de reintentar; no confiar en una aplicación parcial. |
-| `BAKERY_MEMORY_GUARD` | Un full reload chunked de Bakery fue detenido a mitad de aplicación porque la memoria disponible del host cayó por debajo del piso de seguridad. | `journalctl -u lqosd`, memoria disponible/total en el contexto del urgent issue y progreso reciente de aplicación de Bakery. | Tratar la ejecución como fallida, reducir presión de memoria o huella de colas y reintentar solo cuando el host esté estable. |
+| `BAKERY_MEMORY_GUARD` | Una recarga completa por fragmentos de Bakery fue detenida a mitad de aplicación porque la memoria disponible del host cayó por debajo del piso de seguridad escalado. | `journalctl -u lqosd`, memoria disponible/total en el contexto del urgent issue y progreso reciente de aplicación de Bakery. | Tratar la ejecución como fallida, reducir presión de memoria o huella de colas y reintentar solo cuando el host esté estable. |
 | `XDP_IP_MAPPING_CAPACITY` | Los mapeos IP requeridos exceden la capacidad actual del mapa XDP en el kernel. | Forma de `ShapedDevices.csv`, mezcla IPv4/IPv6, supuesto de un dispositivo frente a varios dispositivos, `journalctl -u lqos_scheduler -u lqosd`. | Reducir mapeos requeridos de inmediato (por ejemplo menos dispositivos o prueba IPv4-only), o aumentar la capacidad del mapa del kernel en un cambio coordinado. |
 | `XDP_IP_MAPPING_APPLY_FAILED` | Uno o más inserts de mapeo IP fallaron durante la aplicación. | `journalctl -u lqos_scheduler -u lqosd` para ejemplos resumidos y conteos de fallo. | Corregir la causa del fallo y volver a ejecutar; no confiar en shaping parcial. |
 

docs/v2.0/libreqos-backend-architecture.md

@@ -343,12 +343,13 @@ This retained-root strategy reduces avoidable root-level churn, while still pref
 
 ### 8.6 Full-reload safety guards
 
-Current Bakery full reloads apply two conservative safety checks before and during large queue rebuilds:
+Current Bakery full reloads apply conservative safety checks before and during large queue rebuilds:
 
 1. A qdisc preflight estimates planned qdiscs per interface and also separates infrastructure, `cake`, and `fq_codel` leaf qdiscs.
 2. That same preflight applies a conservative memory forecast and hard-blocks clearly unsafe full reloads before `tc -batch` starts.
-3. During chunked full reload apply, Bakery re-checks host memory at chunk boundaries and aborts the remaining apply if available memory drops below its safety floor.
-4. These guards are intentionally biased toward false positives on large reloads so the system fails early with diagnostics instead of spiraling into an OOM event.
+3. Bakery's memory floor scales with host RAM: it keeps at least 2 GiB available, or one eighth of total RAM on larger systems.
+4. During chunked full reload apply, Bakery re-checks host memory at chunk boundaries and aborts the remaining apply if available memory drops below the scaled floor plus the projected qdisc memory for the batch.
+5. These guards are intentionally biased toward false positives on large reloads so the system fails early with diagnostics instead of spiraling into an OOM event.
 
 ### 8.7 Runtime safety model
 

docs/v2.0/troubleshooting.md

@@ -330,7 +330,7 @@ WebUI urgent issues include machine-readable codes. Use them to triage quickly.
 | `MAPPED_CIRCUIT_LIMIT` | Bakery is enforcing a mapped-circuit limit. | Insight license status, `journalctl -u lqosd` for requested/allowed/dropped counts. | Reduce mapped circuits immediately or update license/limits. |
 | `TC_U16_OVERFLOW` | Queue/class minor IDs exceeded the Linux tc u16 range on a CPU queue. | `journalctl -u lqos_scheduler -u lqosd`, topology depth/queue distribution. | Increase queue count and/or simplify/rebalance hierarchy (for example with integration strategy or root promotion changes). |
 | `TC_QDISC_CAPACITY` | Planned auto-allocated qdiscs exceed the per-interface safe budget or Bakery's conservative memory-safety preflight before apply. | Estimated per-interface qdisc counts, qdisc-kind breakdown, and memory fields in the urgent issue context, `journalctl -u lqos_scheduler -u lqosd`, `on_a_stick` and `queue_mode` config. | Reduce the planned qdisc load for this run (for example fewer circuits/devices in the test shape) before retrying; do not trust partial apply. |
-| `BAKERY_MEMORY_GUARD` | A chunked Bakery full reload was stopped mid-apply because available host memory fell below the safety floor. | `journalctl -u lqosd`, available/total memory in the urgent issue context, and recent Bakery apply progress. | Treat the run as failed, reduce memory pressure or queue footprint, and retry only after the host is stable. |
+| `BAKERY_MEMORY_GUARD` | A chunked Bakery full reload was stopped mid-apply because available host memory fell below the scaled safety floor. | `journalctl -u lqosd`, available/total memory in the urgent issue context, and recent Bakery apply progress. | Treat the run as failed, reduce memory pressure or queue footprint, and retry only after the host is stable. |
 | `XDP_IP_MAPPING_CAPACITY` | Required IP mappings exceed the current XDP kernel map capacity. | `ShapedDevices.csv` row shape, IPv4/IPv6 mix, one-device-vs-many-device assumptions, `journalctl -u lqos_scheduler -u lqosd`. | Reduce required mappings immediately (for example fewer devices or IPv4-only test shape), or raise kernel map capacity in a coordinated change. |
 | `XDP_IP_MAPPING_APPLY_FAILED` | One or more IP mapping inserts failed during apply. | `journalctl -u lqos_scheduler -u lqosd` for summarized failure examples and counts. | Fix the underlying mapping failure, then rerun; do not trust partial shaping. |
 

src/rust/lqos_bakery/src/lib.rs

@@ -1513,7 +1513,14 @@ pub const FQ_CODEL_QDISC_ESTIMATED_MEMORY_BYTES: u64 = 64 * 1024;
 /// substantially exceeding the earlier heuristic during busy periods.
 pub const CAKE_QDISC_ESTIMATED_MEMORY_BYTES: u64 = 512 * 1024;
 /// Minimum memory headroom Bakery tries to leave unused after a projected or in-flight apply.
-pub const BAKERY_MEMORY_GUARD_MIN_AVAILABLE_BYTES: u64 = 768 * 1024 * 1024;
+pub const BAKERY_MEMORY_GUARD_MIN_AVAILABLE_BYTES: u64 = 2 * 1024 * 1024 * 1024;
+/// Divisor used to scale Bakery's memory guard floor with installed RAM.
+pub const BAKERY_MEMORY_GUARD_TOTAL_RAM_DIVISOR: u64 = 8;
+
+/// Returns Bakery's effective memory guard floor for a host with `total_bytes` RAM.
+pub fn bakery_memory_guard_min_available_bytes(total_bytes: u64) -> u64 {
+    BAKERY_MEMORY_GUARD_MIN_AVAILABLE_BYTES.max(total_bytes / BAKERY_MEMORY_GUARD_TOTAL_RAM_DIVISOR)
+}
 /// Maximum number of mapped circuits allowed without Insight.
 const DEFAULT_MAPPED_CIRCUITS_LIMIT: usize = 1000;
 /// Minimum interval between repeated mapped-circuit-limit urgent issues.
@@ -2055,22 +2062,69 @@ fn telemetry_state() -> &'static RwLock<BakeryTelemetryState> {
     BAKERY_TELEMETRY.get_or_init(|| RwLock::new(BakeryTelemetryState::default()))
 }
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum TcCommandType {
+    Class,
+    Qdisc,
+}
+
+fn tc_command_type(argv: &[String]) -> Option<TcCommandType> {
+    match argv.first()?.as_str() {
+        "class" => Some(TcCommandType::Class),
+        "qdisc" => Some(TcCommandType::Qdisc),
+        _ => None,
+    }
+}
+
 fn count_tc_command_types(commands: &[Vec<String>]) -> (usize, usize, usize) {
     let total = commands.len();
     let mut class_count = 0usize;
     let mut qdisc_count = 0usize;
     for argv in commands {
-        if let Some(kind) = argv.first() {
-            match kind.as_str() {
-                "class" => class_count += 1,
-                "qdisc" => qdisc_count += 1,
-                _ => {}
-            }
+        match tc_command_type(argv) {
+            Some(TcCommandType::Class) => class_count += 1,
+            Some(TcCommandType::Qdisc) => qdisc_count += 1,
+            None => {}
         }
     }
     (total, class_count, qdisc_count)
 }
 
+fn count_tc_command_types_and_estimate_qdisc_memory(
+    commands: &[Vec<String>],
+) -> (usize, usize, usize, u64) {
+    let total = commands.len();
+    let mut class_count = 0usize;
+    let mut qdisc_count = 0usize;
+    let mut estimated_qdisc_memory_bytes = 0u64;
+    let mut seen_qdiscs = HashSet::new();
+
+    for argv in commands {
+        match tc_command_type(argv) {
+            Some(TcCommandType::Class) => class_count += 1,
+            Some(TcCommandType::Qdisc) => qdisc_count += 1,
+            None => {}
+        }
+
+        let Some(qdisc_identity) = planned_qdisc_identity(argv) else {
+            continue;
+        };
+        if !seen_qdiscs.insert(qdisc_identity) {
+            continue;
+        }
+        let kind = planned_qdisc_kind(argv).unwrap_or(PlannedQdiscKind::Infra);
+        estimated_qdisc_memory_bytes =
+            estimated_qdisc_memory_bytes.saturating_add(qdisc_kind_estimated_memory_bytes(kind));
+    }
+
+    (
+        total,
+        class_count,
+        qdisc_count,
+        estimated_qdisc_memory_bytes,
+    )
+}
+
 fn push_bakery_event(event: &str, status: &str, summary: String) {
     push_bakery_event_with_site(event, status, None, summary);
 }
@@ -2505,6 +2559,8 @@ pub struct QdiscBudgetEstimate {
     pub estimated_total_memory_bytes: u64,
     /// Current host memory snapshot used for preflight, if available.
     pub memory_snapshot: Option<MemorySnapshot>,
+    /// Effective memory floor used for the memory preflight.
+    pub memory_guard_min_available_bytes: u64,
     /// Whether the memory preflight passed.
     pub memory_ok: bool,
 }
@@ -2649,10 +2705,14 @@ pub fn estimate_full_reload_auto_qdisc_budget(
         acc.saturating_add(detail.estimated_memory_bytes)
     });
     let memory_snapshot = read_memory_snapshot().ok();
+    let memory_guard_min_available_bytes = memory_snapshot
+        .as_ref()
+        .map(|snapshot| bakery_memory_guard_min_available_bytes(snapshot.total_bytes))
+        .unwrap_or(BAKERY_MEMORY_GUARD_MIN_AVAILABLE_BYTES);
     let memory_ok = memory_snapshot.as_ref().is_none_or(|snapshot| {
         snapshot
             .available_bytes
-            .saturating_sub(BAKERY_MEMORY_GUARD_MIN_AVAILABLE_BYTES)
+            .saturating_sub(memory_guard_min_available_bytes)
             >= estimated_total_memory_bytes
     });
 
@@ -2663,6 +2723,7 @@ pub fn estimate_full_reload_auto_qdisc_budget(
         hard_limit: HARD_QDISC_HANDLE_LIMIT_PER_INTERFACE,
         estimated_total_memory_bytes,
         memory_snapshot,
+        memory_guard_min_available_bytes,
         memory_ok,
     }
 }
@@ -3889,16 +3950,26 @@ fn maybe_emit_mapped_circuit_limit_urgent(stats: &MappedLimitStats) {
     });
 }
 
-fn maybe_emit_memory_guard_urgent(summary: &str) {
+fn maybe_emit_memory_guard_urgent(
+    summary: &str,
+    memory_guard_required_available_bytes: u64,
+    estimated_qdisc_memory_bytes: u64,
+) {
     if !summary.contains("Bakery memory guard stopped") {
         return;
     }
 
     let message = summary.to_string();
     let context = read_memory_snapshot().ok().map(|snapshot| {
+        let memory_guard_floor_bytes =
+            bakery_memory_guard_min_available_bytes(snapshot.total_bytes);
         format!(
-            "{{\"available_bytes\":{},\"total_bytes\":{},\"memory_guard_floor_bytes\":{}}}",
-            snapshot.available_bytes, snapshot.total_bytes, BAKERY_MEMORY_GUARD_MIN_AVAILABLE_BYTES
+            "{{\"available_bytes\":{},\"total_bytes\":{},\"memory_guard_floor_bytes\":{},\"estimated_qdisc_memory_bytes\":{},\"memory_guard_required_available_bytes\":{}}}",
+            snapshot.available_bytes,
+            snapshot.total_bytes,
+            memory_guard_floor_bytes,
+            estimated_qdisc_memory_bytes,
+            memory_guard_required_available_bytes
         )
     });
 
@@ -9059,6 +9130,34 @@ fn site_commands_observed_live(
     })
 }
 
+fn read_optional_restore_class_snapshots(
+    config: &Arc<Config>,
+    site_label: &str,
+) -> Option<(
+    HashMap<TcHandle, LiveTcClassEntry>,
+    HashMap<TcHandle, LiveTcClassEntry>,
+)> {
+    let down_snapshot = match read_live_class_snapshot(&config.isp_interface()) {
+        Ok(snapshot) => snapshot,
+        Err(summary) => {
+            debug!(
+                "Bakery: skipping live restore shortcut for {site_label}; unable to snapshot downlink classes: {summary}"
+            );
+            return None;
+        }
+    };
+    let up_snapshot = match read_live_class_snapshot(&config.internet_interface()) {
+        Ok(snapshot) => snapshot,
+        Err(summary) => {
+            debug!(
+                "Bakery: skipping live restore shortcut for {site_label}; unable to snapshot uplink classes: {summary}"
+            );
+            return None;
+        }
+    };
+    Some((down_snapshot, up_snapshot))
+}
+
 fn live_parent_matches(observed: Option<TcHandle>, expected: TcHandle) -> bool {
     if observed == Some(expected) {
         return true;
@@ -10593,28 +10692,27 @@ fn handle_treeguard_set_node_virtual_live(
         }
 
         let reversible_standby = saved_state.active_branch_hides_original_site();
-        let live_restore_snapshots = if reversible_standby {
-            Some((
-                read_live_class_snapshot(&config.isp_interface())?,
-                read_live_class_snapshot(&config.internet_interface())?,
-            ))
-        } else {
-            None
-        };
+        let mut live_restore_snapshots = None;
 
         if !saved_state.pending_prune
             && let Some(cmds) = saved_state
                 .site
                 .to_commands(&config, ExecutionMode::Builder)
         {
-            let root_already_live =
+            let root_already_live = if reversible_standby {
                 live_restore_snapshots
+                    .get_or_insert_with(|| {
+                        read_optional_restore_class_snapshots(&config, &site_label)
+                    })
                     .as_ref()
                     .is_some_and(|(down_snapshot, up_snapshot)| {
                         let mut only_root = HashMap::new();
                         only_root.insert(site_hash, saved_state.site.clone());
                         site_commands_observed_live(&only_root, down_snapshot, up_snapshot)
-                    });
+                    })
+            } else {
+                false
+            };
             if !root_already_live {
                 let result =
                     execute_and_record_live_change(&cmds, "TreeGuard runtime hidden site restore");
@@ -10643,8 +10741,9 @@ fn handle_treeguard_set_node_virtual_live(
                 )
             })
             .collect();
-        let originals_already_live =
+        let originals_already_live = if reversible_standby && !saved_state.saved_sites.is_empty() {
             live_restore_snapshots
+                .get_or_insert_with(|| read_optional_restore_class_snapshots(&config, &site_label))
                 .as_ref()
                 .is_some_and(|(down_snapshot, up_snapshot)| {
                     site_commands_observed_live(
@@ -10652,7 +10751,10 @@ fn handle_treeguard_set_node_virtual_live(
                         down_snapshot,
                         up_snapshot,
                     )
-                });
+                })
+        } else {
+            false
+        };
         if reversible_standby && originals_already_live {
             for (hash, command) in &saved_state.saved_sites {
                 sites.insert(*hash, Arc::clone(command));
@@ -11037,7 +11139,12 @@ fn process_batch(
     let path = config.debug_state_file_path("linux_tc_rust.txt");
     write_command_file(&path, &commands);
     let build_duration_ms = build_started.elapsed().as_millis() as u64;
-    let (total_tc_commands, class_commands, qdisc_commands) = count_tc_command_types(&commands);
+    let (total_tc_commands, class_commands, qdisc_commands, estimated_qdisc_memory_bytes) =
+        count_tc_command_types_and_estimate_qdisc_memory(&commands);
+    let memory_guard_required_available_bytes = read_memory_snapshot()
+        .map(|snapshot| bakery_memory_guard_min_available_bytes(snapshot.total_bytes))
+        .unwrap_or(BAKERY_MEMORY_GUARD_MIN_AVAILABLE_BYTES)
+        .saturating_add(estimated_qdisc_memory_bytes);
     let total_chunks = if total_tc_commands == 0 {
         0
     } else {
@@ -11054,7 +11161,7 @@ fn process_batch(
         &commands,
         "processing batch",
         FULL_RELOAD_TC_CHUNK_SIZE,
-        Some(BAKERY_MEMORY_GUARD_MIN_AVAILABLE_BYTES),
+        Some(memory_guard_required_available_bytes),
         |completed_tc_commands, total_tc_commands, completed_chunks, total_chunks| {
             update_bakery_apply_progress(
                 Some("Applying tc command chunks"),
@@ -11066,7 +11173,11 @@ fn process_batch(
         },
     );
     let summary = summarize_apply_result("processing batch", &result);
-    maybe_emit_memory_guard_urgent(&summary);
+    maybe_emit_memory_guard_urgent(
+        &summary,
+        memory_guard_required_available_bytes,
+        estimated_qdisc_memory_bytes,
+    );
     BakeryFullReloadBatchResult {
         result,
         summary,
@@ -13363,6 +13474,65 @@ mod tests {
         assert!(estimate.ok());
     }
 
+    #[test]
+    fn bakery_memory_guard_floor_scales_with_host_ram() {
+        assert_eq!(
+            bakery_memory_guard_min_available_bytes(8 * 1024 * 1024 * 1024),
+            BAKERY_MEMORY_GUARD_MIN_AVAILABLE_BYTES
+        );
+        assert_eq!(
+            bakery_memory_guard_min_available_bytes(64 * 1024 * 1024 * 1024),
+            8 * 1024 * 1024 * 1024
+        );
+    }
+
+    #[test]
+    fn command_qdisc_memory_estimate_deduplicates_qdisc_identity() {
+        let commands = vec![
+            vec![
+                "qdisc".to_string(),
+                "add".to_string(),
+                "dev".to_string(),
+                "eth0".to_string(),
+                "parent".to_string(),
+                "1:10".to_string(),
+                "cake".to_string(),
+            ],
+            vec![
+                "qdisc".to_string(),
+                "replace".to_string(),
+                "dev".to_string(),
+                "eth0".to_string(),
+                "parent".to_string(),
+                "1:10".to_string(),
+                "cake".to_string(),
+            ],
+            vec![
+                "qdisc".to_string(),
+                "add".to_string(),
+                "dev".to_string(),
+                "eth1".to_string(),
+                "parent".to_string(),
+                "1:10".to_string(),
+                "cake".to_string(),
+            ],
+            vec![
+                "qdisc".to_string(),
+                "add".to_string(),
+                "dev".to_string(),
+                "eth0".to_string(),
+                "parent".to_string(),
+                "1:11".to_string(),
+                "fq_codel".to_string(),
+            ],
+        ];
+
+        assert_eq!(
+            count_tc_command_types_and_estimate_qdisc_memory(&commands).3,
+            (CAKE_QDISC_ESTIMATED_MEMORY_BYTES * 2) + FQ_CODEL_QDISC_ESTIMATED_MEMORY_BYTES
+        );
+    }
+
     #[test]
     fn qdisc_budget_estimate_counts_only_root_mq_in_observe_mode() {
         let _guard = bakery_test_lock().lock().expect("lock");

src/rust/lqos_bakery/src/utils.rs

@@ -269,7 +269,7 @@ fn memory_guard_failure_summary(
     phase: &str,
 ) -> String {
     format!(
-        "Bakery memory guard stopped {purpose} {phase} chunk {}/{}: available memory {} bytes is below safety floor {} bytes (total RAM {} bytes).",
+        "Bakery memory guard stopped {purpose} {phase} chunk {}/{}: available memory {} bytes is below required available memory {} bytes (total RAM {} bytes).",
         chunk_number,
         total_chunks,
         snapshot.available_bytes,

src/rust/lqos_python/src/device_weights.rs

@@ -16,7 +16,13 @@ use lqos_bus::{BusRequest, BusResponse};
 use lqos_config::{ShapedDevice, load_config};
 use pyo3::pyclass;
 use serde::{Deserialize, Serialize};
-use std::time::Duration;
+use std::{
+    sync::{Mutex, OnceLock},
+    time::{Duration, Instant},
+};
+
+const REMOTE_WEIGHT_SUCCESS_TTL: Duration = Duration::from_secs(300);
+const REMOTE_WEIGHT_FAILURE_BACKOFF: Duration = Duration::from_secs(300);
 
 /// This struct is used to send a request to the Long Term Stats API
 #[derive(Serialize, Deserialize)]
@@ -30,7 +36,7 @@ pub struct DeviceWeightRequest {
 
 /// This struct is used to receive a response from the Long Term Stats API
 /// It contains the circuit_id and the weight of the device
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Serialize, Deserialize, Debug, Clone)]
 #[pyclass]
 pub struct DeviceWeightResponse {
     #[pyo3(get)]
@@ -39,6 +45,57 @@ pub struct DeviceWeightResponse {
     pub weight: i64,
 }
 
+#[derive(Clone)]
+struct CachedRemoteWeights {
+    fetched_at: Instant,
+    weights: Vec<DeviceWeightResponse>,
+}
+
+#[derive(Default)]
+struct RemoteWeightCache {
+    last_success: Option<CachedRemoteWeights>,
+    last_failure: Option<Instant>,
+}
+
+fn remote_weight_cache() -> &'static Mutex<RemoteWeightCache> {
+    static CACHE: OnceLock<Mutex<RemoteWeightCache>> = OnceLock::new();
+    CACHE.get_or_init(|| Mutex::new(RemoteWeightCache::default()))
+}
+
+fn cached_remote_weights() -> Option<Vec<DeviceWeightResponse>> {
+    let cache = remote_weight_cache().lock().ok()?;
+    let cached = cache.last_success.as_ref()?;
+    if cached.fetched_at.elapsed() <= REMOTE_WEIGHT_SUCCESS_TTL {
+        return Some(cached.weights.clone());
+    }
+    None
+}
+
+fn recent_remote_weight_failure() -> bool {
+    let Ok(cache) = remote_weight_cache().lock() else {
+        return false;
+    };
+    cache
+        .last_failure
+        .is_some_and(|failed_at| failed_at.elapsed() <= REMOTE_WEIGHT_FAILURE_BACKOFF)
+}
+
+fn remember_remote_weight_success(weights: &[DeviceWeightResponse]) {
+    if let Ok(mut cache) = remote_weight_cache().lock() {
+        cache.last_success = Some(CachedRemoteWeights {
+            fetched_at: Instant::now(),
+            weights: weights.to_vec(),
+        });
+        cache.last_failure = None;
+    }
+}
+
+fn remember_remote_weight_failure() {
+    if let Ok(mut cache) = remote_weight_cache().lock() {
+        cache.last_failure = Some(Instant::now());
+    }
+}
+
 /// This function is used to get the device weights from the Long Term Stats API
 fn get_weights_from_lts(
     org_key: &str,
@@ -147,10 +204,28 @@ pub(crate) fn get_weights_rust() -> Result<Vec<DeviceWeightResponse>> {
         let duration_seconds = 60 * 60 * 24; // 1 day
         let percentile = 0.95;
 
-        eprintln!("Getting weights from Insight");
-        let weights = get_weights_from_lts(&org_key, &node_id, start, duration_seconds, percentile);
-        if let Ok(weights) = weights {
-            eprintln!("Retrieved {} weights from Insight", weights.len());
+        let remote_weights = if let Some(weights) = cached_remote_weights() {
+            eprintln!("Using cached Insight weights");
+            Some(weights)
+        } else if recent_remote_weight_failure() {
+            eprintln!("Skipping Insight weight fetch; recent request failed");
+            None
+        } else {
+            eprintln!("Getting weights from Insight");
+            match get_weights_from_lts(&org_key, &node_id, start, duration_seconds, percentile) {
+                Ok(weights) => {
+                    eprintln!("Retrieved {} weights from Insight", weights.len());
+                    remember_remote_weight_success(&weights);
+                    Some(weights)
+                }
+                Err(err) => {
+                    remember_remote_weight_failure();
+                    eprintln!("Failed to get weights from Insight: {:?}", err);
+                    None
+                }
+            }
+        };
+        if let Some(weights) = remote_weights {
             // Merge them
             for weight in weights.iter() {
                 if let Some(existing) = shaped_devices_weights
@@ -160,8 +235,6 @@ pub(crate) fn get_weights_rust() -> Result<Vec<DeviceWeightResponse>> {
                     existing.weight = weight.weight;
                 }
             }
-        } else {
-            eprintln!("Failed to get weights from Insight: {:?}", weights);
         }
     }
 

src/rust/lqos_python/src/lib.rs

@@ -3571,15 +3571,15 @@ impl Bakery {
             (false, false) => "exceeds qdisc-count preflight and failed memory preflight",
         };
         let memory_summary = if let Some(snapshot) = estimate.memory_snapshot.as_ref() {
-            let memory_floor = lqos_bakery::BAKERY_MEMORY_GUARD_MIN_AVAILABLE_BYTES;
-            let required_available_bytes =
-                memory_floor.saturating_add(estimate.estimated_total_memory_bytes);
+            let required_available_bytes = estimate
+                .memory_guard_min_available_bytes
+                .saturating_add(estimate.estimated_total_memory_bytes);
             format!(
                 "estimated qdisc memory {} bytes; available memory {} bytes; required minimum {} bytes (safety floor {} bytes plus estimate)",
                 estimate.estimated_total_memory_bytes,
                 snapshot.available_bytes,
                 required_available_bytes,
-                memory_floor
+                estimate.memory_guard_min_available_bytes
             )
         } else {
             format!(
@@ -3625,7 +3625,7 @@ impl Bakery {
                 .memory_snapshot
                 .as_ref()
                 .map(|snapshot| snapshot.available_bytes),
-            memory_guard_min_available_bytes: lqos_bakery::BAKERY_MEMORY_GUARD_MIN_AVAILABLE_BYTES,
+            memory_guard_min_available_bytes: estimate.memory_guard_min_available_bytes,
             memory_ok: estimate.memory_ok,
             interfaces: interface_reports,
         }]);
@@ -3658,7 +3658,7 @@ impl Bakery {
         result.set_item("memory_ok", estimate.memory_ok)?;
         result.set_item(
             "memory_guard_min_available_bytes",
-            lqos_bakery::BAKERY_MEMORY_GUARD_MIN_AVAILABLE_BYTES,
+            estimate.memory_guard_min_available_bytes,
         )?;
         if let Some(snapshot) = estimate.memory_snapshot {
             result.set_item("memory_total_bytes", snapshot.total_bytes)?;

src/rust/lqos_queue_tracker/src/queue_diff.rs

@@ -1,7 +1,7 @@
 use crate::queue_types::QueueType;
 use serde::Serialize;
 use thiserror::Error;
-use tracing::error;
+use tracing::warn;
 
 #[derive(Debug, Clone, Serialize)]
 pub enum QueueDiff {
@@ -21,27 +21,46 @@ pub(crate) fn make_queue_diff(
         QueueType::FqCodel(..) => match current {
             QueueType::FqCodel(..) => Ok(fq_codel_diff(previous, current)?),
             _ => {
-                error!(
-                    "Queue diffs are not implemented for FqCodel to {:?}",
-                    current
+                warn!(
+                    previous = queue_kind(previous),
+                    current = queue_kind(current),
+                    "Queue type changed; resetting queue diff sample"
                 );
-                Err(QueueDiffError::NotImplemented)
+                Ok(QueueDiff::None)
             }
         },
         QueueType::Cake(..) => match current {
             QueueType::Cake(..) => Ok(cake_diff(previous, current)?),
             _ => {
-                error!("Queue diffs are not implemented for Cake to {:?}", current);
-                Err(QueueDiffError::NotImplemented)
+                warn!(
+                    previous = queue_kind(previous),
+                    current = queue_kind(current),
+                    "Queue type changed; resetting queue diff sample"
+                );
+                Ok(QueueDiff::None)
             }
         },
         _ => {
-            error!("Queue diffs are not implemented for {:?}", current);
-            Err(QueueDiffError::NotImplemented)
+            warn!(
+                previous = queue_kind(previous),
+                current = queue_kind(current),
+                "Queue diff unavailable for queue type; resetting queue diff sample"
+            );
+            Ok(QueueDiff::None)
         }
     }
 }
 
+fn queue_kind(queue: &QueueType) -> &'static str {
+    match queue {
+        QueueType::FqCodel(_) => "fq_codel",
+        QueueType::Cake(_) => "cake",
+        QueueType::Mq(_) => "mq",
+        QueueType::Htb(_) => "htb",
+        QueueType::ClsAct => "clsact",
+    }
+}
+
 #[derive(Serialize, Clone, Debug)]
 pub struct CakeDiff {
     pub bytes: u64,
@@ -116,3 +135,42 @@ fn fq_codel_diff(previous: &QueueType, current: &QueueType) -> Result<QueueDiff,
     }
     Err(QueueDiffError::NotImplemented)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::deserialize_tc_tree;
+
+    #[test]
+    fn queue_type_change_resets_diff_sample() {
+        let fq_codel = deserialize_tc_tree(
+            r#"[{
+                "kind":"fq_codel",
+                "handle":"9000:",
+                "parent":"1:2",
+                "options":{},
+                "bytes":1000,
+                "packets":10
+            }]"#,
+        )
+        .expect("fq_codel qdisc should parse")
+        .remove(0);
+        let cake = deserialize_tc_tree(
+            r#"[{
+                "kind":"cake",
+                "handle":"9001:",
+                "parent":"1:2",
+                "options":{},
+                "bytes":2000,
+                "packets":20,
+                "tins":[]
+            }]"#,
+        )
+        .expect("cake qdisc should parse")
+        .remove(0);
+
+        let diff = make_queue_diff(&fq_codel, &cake).expect("type changes should not error");
+
+        assert!(matches!(diff, QueueDiff::None));
+    }
+}

src/rust/lqosd/src/node_manager/js_build/src/pubsub/ws.js

@@ -123,7 +123,10 @@ export class WsClient {
 
     connect() {
         if (this.ws) {
-            return;
+            if (this.ws.readyState === WebSocket.CONNECTING || this.ws.readyState === WebSocket.OPEN) {
+                return;
+            }
+            this._dropSocket("stale-socket");
         }
         if (this.reconnectTimer) {
             clearTimeout(this.reconnectTimer);
@@ -197,6 +200,11 @@ export class WsClient {
             });
             this.ws = null;
             this.handshake_done = false;
+            try {
+                socket.close();
+            } catch (_) {
+                // The browser may already have closed the socket.
+            }
             if (!this.versionReloading) {
                 this._scheduleReconnect();
             }
@@ -312,7 +320,7 @@ export class WsClient {
                 dashboardWsInteresting("subscribe-wire-interesting", [channel], {
                     desiredChannels: Array.from(this.desiredChannels.keys()),
                 });
-                this.ws.send(encoder.encode({ Subscribe: { channel } }));
+                this._sendControl({ Subscribe: { channel } });
             }
         }
         if (channels.length > 0) {
@@ -339,7 +347,7 @@ export class WsClient {
                     dashboardWsInteresting("unsubscribe-wire-interesting", [channel], {
                         desiredChannels: Array.from(this.desiredChannels.keys()),
                     });
-                    this.ws.send(encoder.encode({ Unsubscribe: { channel } }));
+                    this._sendControl({ Unsubscribe: { channel } });
                 }
             } else {
                 this.desiredChannels.set(channel, current - 1);
@@ -355,11 +363,16 @@ export class WsClient {
         if (!this.ws) {
             this.connect();
         }
-        if (!this.handshake_done) {
+        if (!this.handshake_done || !this._socketIsOpen()) {
             this.pending.push(normalized);
+            if (this.handshake_done && this.ws && !this._socketIsOpen()) {
+                this._retireSocket();
+            }
             return;
         }
-        this.ws.send(encoder.encode(normalized));
+        if (!this._sendControl(normalized)) {
+            this.pending.push(normalized);
+        }
     }
 
     on(event_name, handler) {
@@ -406,28 +419,58 @@ export class WsClient {
         dashboardWsDebug("handshake-complete", {
             desiredChannels: Array.from(this.desiredChannels.keys()),
         });
-        this.ws.send(
-            encoder.encode({
-                HelloReply: {
-                    ack: ACK_TEXT,
-                    token: get_user_token(),
-                },
-            }),
-        );
-        for (let i = 0; i < this.pending.length; i++) {
-            this.ws.send(encoder.encode(this.pending[i]));
+        if (!this._sendControl({
+            HelloReply: {
+                ack: ACK_TEXT,
+                token: get_user_token(),
+            },
+        })) {
+            return;
         }
+        const pending = this.pending;
         this.pending = [];
+        for (let i = 0; i < pending.length; i++) {
+            if (!this._sendControl(pending[i])) {
+                this.pending = pending.slice(i).concat(this.pending);
+                return;
+            }
+        }
         for (const channel of this.desiredChannels.keys()) {
             dashboardWsInteresting("subscribe-wire-interesting", [channel], {
                 desiredChannels: Array.from(this.desiredChannels.keys()),
                 fromHandshake: true,
             });
-            this.ws.send(encoder.encode({ Subscribe: { channel } }));
+            if (!this._sendControl({ Subscribe: { channel } })) {
+                return;
+            }
         }
         this.reconnectDelayMs = 1000;
     }
 
+    _socketIsOpen() {
+        return this.ws && this.ws.readyState === WebSocket.OPEN;
+    }
+
+    _retireSocket() {
+        this._dropSocket("retire-socket");
+        this._scheduleReconnect();
+    }
+
+    _sendControl(request_obj) {
+        if (!this._socketIsOpen()) {
+            this._retireSocket();
+            return false;
+        }
+        try {
+            this.ws.send(encoder.encode(request_obj));
+            return true;
+        } catch (err) {
+            console.warn("Websocket send failed; reconnecting", err);
+            this._retireSocket();
+            return false;
+        }
+    }
+
     _dispatch(msg) {
         if (!msg || !msg.event) {
             return;

src/rust/lqosd/src/node_manager/ws.rs

@@ -55,6 +55,7 @@ mod ticker;
 
 const WS_VERSION: &str = include_str!("../../../../VERSION_STRING");
 const HANDSHAKE_TIMEOUT_SECS: u64 = 10;
+const REQUEST_TIMEOUT_SECS: u64 = 15;
 
 /// Provides an Axum router for the websocket system. Exposes a single /ws route that supports
 /// pubsub subscriptions and private commands.
@@ -182,19 +183,29 @@ async fn handle_socket(
                 // Received a websocket message
                 match inbound {
                     Some(Ok(msg)) => {
-                        let should_close = receive_channel_message(
-                            msg,
-                            channels.clone(),
-                            tx.clone(),
-                            &mut subscribed_channels,
-                            &mut handshake_complete,
-                            &mut WsRequestState {
-                                private_state: &mut private_state,
-                                login: &mut login,
-                                shaper_query: shaper_query.clone(),
-                            },
+                        let should_close = match tokio::time::timeout(
+                            std::time::Duration::from_secs(REQUEST_TIMEOUT_SECS),
+                            receive_channel_message(
+                                msg,
+                                channels.clone(),
+                                tx.clone(),
+                                &mut subscribed_channels,
+                                &mut handshake_complete,
+                                &mut WsRequestState {
+                                    private_state: &mut private_state,
+                                    login: &mut login,
+                                    shaper_query: shaper_query.clone(),
+                                },
+                            ),
                         )
-                        .await;
+                        .await
+                        {
+                            Ok(should_close) => should_close,
+                            Err(_) => {
+                                warn!("Websocket request timed out; closing connection");
+                                true
+                            }
+                        };
                         if should_close {
                             break;
                         }
@@ -212,6 +223,11 @@ async fn handle_socket(
             }
         }
     }
+    for channel in subscribed_channels.drain() {
+        channels.unsubscribe(channel, tx.clone()).await;
+    }
+    drop(private_state);
+    drop(tx);
     outbound_handle.abort();
     let _ = outbound_handle.await;
     info!("Websocket disconnected");