feat: allow disabling updates to job COBs in config

This commit is contained in:
Lars Wirzenius
2026-05-11 15:03:21 +03:00
parent ecce2a73a5
commit bf8be65677
3 changed files with 28 additions and 3 deletions
+2
View File
@@ -219,6 +219,7 @@ impl QueuedCmd {
.triggers(&config.to_triggers()) .triggers(&config.to_triggers())
.adapters(&adapters) .adapters(&adapters)
.concurrent_adapters(config.concurrent_adapters()) .concurrent_adapters(config.concurrent_adapters())
.update_job_cobs(config.update_job_cobs())
.build() .build()
.map_err(CibError::process_queue)?; .map_err(CibError::process_queue)?;
let thread = start_thread(processor); let thread = start_thread(processor);
@@ -303,6 +304,7 @@ impl ProcessEventsCmd {
.triggers(&config.to_triggers()) .triggers(&config.to_triggers())
.adapters(&adapters) .adapters(&adapters)
.concurrent_adapters(config.concurrent_adapters()) .concurrent_adapters(config.concurrent_adapters())
.update_job_cobs(config.update_job_cobs())
.build() .build()
.map_err(CibError::process_queue)?; .map_err(CibError::process_queue)?;
+11
View File
@@ -54,6 +54,9 @@ pub struct Config {
concurrent_adapters: Option<usize>, concurrent_adapters: Option<usize>,
description: Option<String>, description: Option<String>,
#[serde(default = "default_update_job_cobs")]
update_job_cobs: bool,
} }
fn default_max_run_time() -> Duration { fn default_max_run_time() -> Duration {
@@ -64,6 +67,10 @@ fn default_queue_len_interval() -> Duration {
DEFAULT_QUEUE_LEN_INTERVAL DEFAULT_QUEUE_LEN_INTERVAL
} }
fn default_update_job_cobs() -> bool {
true
}
impl Config { impl Config {
pub fn load(filename: &Path) -> Result<Self, ConfigError> { pub fn load(filename: &Path) -> Result<Self, ConfigError> {
let config = std::fs::read(filename).map_err(|e| ConfigError::read_config(filename, e))?; let config = std::fs::read(filename).map_err(|e| ConfigError::read_config(filename, e))?;
@@ -140,6 +147,10 @@ impl Config {
&self.db &self.db
} }
pub fn update_job_cobs(&self) -> bool {
self.update_job_cobs
}
pub fn to_json(&self) -> Result<String, ConfigError> { pub fn to_json(&self) -> Result<String, ConfigError> {
serde_json::to_string_pretty(self).map_err(ConfigError::to_json) serde_json::to_string_pretty(self).map_err(ConfigError::to_json)
} }
+15 -3
View File
@@ -41,6 +41,7 @@ pub struct QueueProcessorBuilder {
run_tx: Option<NotificationSender>, run_tx: Option<NotificationSender>,
queue_len_interval: Option<Duration>, queue_len_interval: Option<Duration>,
concurrent_adapters: Option<usize>, concurrent_adapters: Option<usize>,
update_known_job_cobs: bool,
} }
const DEFAULT_QUEUE_LEN_DURATION: Duration = Duration::from_secs(10); const DEFAULT_QUEUE_LEN_DURATION: Duration = Duration::from_secs(10);
@@ -58,6 +59,14 @@ impl QueueProcessorBuilder {
.ok_or(QueueError::Missing("concurrent_adapters"))?; .ok_or(QueueError::Missing("concurrent_adapters"))?;
let (child_pid_tx, child_pid_rx) = channel(); let (child_pid_tx, child_pid_rx) = channel();
let known_job_cobs = if self.update_known_job_cobs {
eprintln!("XXX update job COBs");
KnownJobCobs::updater().map_err(QueueError::KnownJobCobs)?
} else {
eprintln!("XXX do NOT update job COBs");
KnownJobCobs::NoUpdates
};
Ok(QueueProcessor { Ok(QueueProcessor {
profile, profile,
broker, broker,
@@ -75,9 +84,7 @@ impl QueueProcessorBuilder {
current: CurrentlyPicked::default(), current: CurrentlyPicked::default(),
child_pid_tx, child_pid_tx,
child_pid_rx, child_pid_rx,
known_job_cobs: Arc::new(Mutex::new( known_job_cobs: Arc::new(Mutex::new(known_job_cobs)),
KnownJobCobs::updater().map_err(QueueError::KnownJobCobs)?,
)),
}) })
} }
@@ -125,6 +132,11 @@ impl QueueProcessorBuilder {
self.adapters = Some(adapters.clone()); self.adapters = Some(adapters.clone());
self self
} }
pub fn update_job_cobs(mut self, value: bool) -> Self {
self.update_known_job_cobs = value;
self
}
} }
// The queue processor gets events from the event queue in // The queue processor gets events from the event queue in