Merge pull request #1139 from LibreQoE/fix/bakery-license-cache-startup

Gate Bakery startup on local license-cache initialization

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<crossbeam_channel::Sender<BakeryCommands>> {
+/// Starts the Bakery system after receiving the one-shot startup readiness token.
+pub fn start_bakery(
+    startup_ready: Receiver<()>,
+) -> anyhow::Result<crossbeam_channel::Sender<BakeryCommands>> {
     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<crossbeam_channel::Sender<BakeryCommands
     std::thread::Builder::new()
         .name("lqos_bakery".to_string())
         .spawn(move || {
+            if !wait_for_startup_ready(startup_ready) {
+                return;
+            }
             bakery_main(rx, inner_sender);
         })
         .map_err(|e| anyhow::anyhow!("Failed to start Bakery thread: {}", e))?;
     Ok(tx)
 }
 
+fn wait_for_startup_ready(startup_ready: Receiver<()>) -> 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<BakeryCommands>, tx: Sender<BakeryCommands>) {
     // Current operation batch
     let mut batch: Option<Vec<Arc<BakeryCommands>>> = None;
@@ -11702,6 +11719,62 @@ mod tests {
 
     static TC_BYPASS_SETTER_CALLS: Mutex<Vec<bool>> = 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();

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";