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.
This commit is contained in:
Herbert Wolverson committed 2026-08-06 10:13:45 -05:00
1 parent 5d0f94fd2d
commit 94b3c82679
2 files changed
+124 -10

No files matched your search

+75 -2
View File
@@ -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();
+49 -8
View File
@@ -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";