From 94b3c82679b8468198c4e0ac9142baab58ca093b Mon Sep 17 00:00:00 2001 From: Herbert Wolverson Date: Thu, 6 Aug 2026 10:13:45 -0500 Subject: [PATCH] Gate Bakery startup on local license-cache initialization Keep queued Bakery commands intact while lqosd checks local .keys state, releasing the gate for both successful and non-fatal cache outcomes without tying startup to Insight network availability. --- src/rust/lqos_bakery/src/lib.rs | 77 ++++++++++++++++++++++++++++++++- src/rust/lqosd/src/main.rs | 57 ++++++++++++++++++++---- 2 files changed, 124 insertions(+), 10 deletions(-) diff --git a/src/rust/lqos_bakery/src/lib.rs b/src/rust/lqos_bakery/src/lib.rs index d55a3b21..a0cc5efe 100644 --- a/src/rust/lqos_bakery/src/lib.rs +++ b/src/rust/lqos_bakery/src/lib.rs @@ -4073,8 +4073,10 @@ fn log_mapped_limit_decision( ); } -/// Starts the Bakery system, returning a channel sender for sending commands to the Bakery. -pub fn start_bakery() -> anyhow::Result> { +/// Starts the Bakery system after receiving the one-shot startup readiness token. +pub fn start_bakery( + startup_ready: Receiver<()>, +) -> anyhow::Result> { let (tx, rx) = crossbeam_channel::bounded(CHANNEL_CAPACITY); let inner_sender = tx.clone(); if BAKERY_SENDER.set(tx.clone()).is_err() { @@ -4083,12 +4085,27 @@ pub fn start_bakery() -> anyhow::Result) -> bool { + match startup_ready.recv() { + Ok(()) => true, + Err(error) => { + error!( + "Bakery startup readiness channel closed before release: {error}; stopping worker" + ); + false + } + } +} + fn bakery_main(rx: Receiver, tx: Sender) { // Current operation batch let mut batch: Option>> = None; @@ -11702,6 +11719,62 @@ mod tests { static TC_BYPASS_SETTER_CALLS: Mutex> = Mutex::new(Vec::new()); + #[test] + fn startup_gate_blocks_worker_and_preserves_queued_command() { + let (startup_tx, startup_rx) = crossbeam_channel::bounded(1); + let (command_tx, command_rx) = crossbeam_channel::bounded(1); + let (processed_tx, processed_rx) = crossbeam_channel::bounded(1); + let (worker_started_tx, worker_started_rx) = crossbeam_channel::bounded(1); + + let worker = std::thread::spawn(move || { + worker_started_tx + .send(()) + .expect("worker-started receiver should remain connected"); + assert!(wait_for_startup_ready(startup_rx)); + let command = command_rx + .recv() + .expect("queued Bakery command should remain available"); + processed_tx + .send(command) + .expect("processed Bakery command receiver should remain connected"); + }); + + command_tx + .send(BakeryCommands::Tick) + .expect("queue the Bakery command before startup release"); + worker_started_rx + .recv() + .expect("Bakery worker should reach the startup gate"); + assert!(processed_rx.try_recv().is_err()); + + startup_tx + .send(()) + .expect("release the Bakery startup gate"); + assert!(matches!( + processed_rx.recv_timeout(Duration::from_secs(1)), + Ok(BakeryCommands::Tick) + )); + worker.join().expect("Bakery worker test should finish"); + } + + #[test] + fn startup_gate_stops_worker_when_readiness_channel_closes() { + let (startup_tx, startup_rx) = crossbeam_channel::bounded(1); + drop(startup_tx); + let (progress_tx, progress_rx) = crossbeam_channel::bounded(1); + + let worker = std::thread::spawn(move || { + if wait_for_startup_ready(startup_rx) { + progress_tx + .send(()) + .expect("progress receiver should remain connected"); + } + }); + + worker.join().expect("closed-gate worker test should finish"); + assert!(progress_rx.try_recv().is_err()); + } + #[test] fn stormguard_live_requires_enabled_non_dry_run_config() { let mut config = Config::default(); diff --git a/src/rust/lqosd/src/main.rs b/src/rust/lqosd/src/main.rs index 47a41bbb..1676fab6 100644 --- a/src/rust/lqosd/src/main.rs +++ b/src/rust/lqosd/src/main.rs @@ -119,6 +119,22 @@ pub fn set_console_logging() -> anyhow::Result<()> { Ok(()) } +fn initialize_license_storage_and_release( + initialize_storage: impl FnOnce() -> Result<()>, + license_cache_ready_tx: crossbeam_channel::Sender<()>, +) { + info!("Starting local Insight license-cache initialization"); + match initialize_storage() { + Ok(()) => info!("Completed local Insight license-cache initialization successfully"), + Err(err) => { + warn!("Completed local Insight license-cache initialization with error: {err:?}") + } + } + if let Err(err) = license_cache_ready_tx.send(()) { + warn!("Unable to release the Bakery startup gate: {err}"); + } +} + fn normalize_mapping_request( tc_handle: lqos_bus::TcHandle, cpu: u32, @@ -206,15 +222,11 @@ fn main() -> Result<()> { ensure_rustls_crypto_provider()?; - if let Err(e) = lts2_sys::license_grant::init_license_storage(&config) { - warn!("Failed to initialize Insight license storage: {e:?}"); - } + let (license_cache_ready_tx, license_cache_ready_rx) = crossbeam_channel::bounded(1); - // Apply Tunings - tuning::tune_lqosd_from_config_file()?; - - // Start the Bakery for TC command execution - let bakery_sender = match lqos_bakery::start_bakery() { + // Start the Bakery before checking the local license cache so its command + // queue is available, but keep its worker behind the readiness token. + let bakery_sender = match lqos_bakery::start_bakery(license_cache_ready_rx) { Ok(sender) => Some(sender), Err(err) => { record_shaping_failure( @@ -226,6 +238,14 @@ fn main() -> Result<()> { } }; + initialize_license_storage_and_release( + || lts2_sys::license_grant::init_license_storage(&config), + license_cache_ready_tx, + ); + + // Apply Tunings + tuning::tune_lqosd_from_config_file()?; + // Spawn tracking sub-systems let Ok(control_channel) = lts2_sys::control_channel::init_control_channel() else { error!("Failed to initialize Insight control channel, exiting."); @@ -1464,6 +1484,27 @@ mod tests { use super::*; use lqos_bus::{UrgentSeverity, UrgentSource}; + #[test] + fn license_cache_error_releases_bakery_startup_gate() { + let (ready_tx, ready_rx) = crossbeam_channel::bounded(1); + + initialize_license_storage_and_release( + || Err(anyhow::anyhow!("synthetic license-cache failure")), + ready_tx, + ); + + assert!(matches!(ready_rx.try_recv(), Ok(()))); + } + + #[test] + fn license_cache_success_releases_bakery_startup_gate() { + let (ready_tx, ready_rx) = crossbeam_channel::bounded(1); + + initialize_license_storage_and_release(|| Ok(()), ready_tx); + + assert!(matches!(ready_rx.try_recv(), Ok(()))); + } + #[test] fn bus_clear_urgent_issue_by_identity_reaches_urgent_store() { let code = "TEST_BUS_XDP_IP_MAPPING_APPLY_FAILED";