diff --git a/Cargo.lock b/Cargo.lock index ef41e05f..9e2ecd75 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1129,9 +1129,9 @@ dependencies = [ [[package]] name = "kvx" -version = "0.8.0" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "31ba9dfb918a4c8e2f326d72a86891ace7c83d13c550e40c543d4cec57a9bf34" +checksum = "dbd3053d98daca52860c508c21f32107c174396af5772ea54051110807d040cb" dependencies = [ "kvx_macros", "kvx_types", @@ -1148,9 +1148,9 @@ dependencies = [ [[package]] name = "kvx_macros" -version = "0.8.0" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e87fdea2c4bf35291921c3a224b4cc0abf86cadb6818a4c09facefc755f2e3dd" +checksum = "ed06dbcd44cd4f7af95fb5e0c8d020d14e64cbbd2ceacfdd9ed07f2e6d81e87f" dependencies = [ "kvx_types", "proc-macro-error", @@ -1161,9 +1161,9 @@ dependencies = [ [[package]] name = "kvx_types" -version = "0.8.0" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0569372a25dd8a7177baf7fb0575709d1121f4ec829623c424150d653200f8d7" +checksum = "fbff677c127336a6e93df7d54a036824312d4c1e9d165bd284e25b024cc83b06" dependencies = [ "postgres", "postgres-types", diff --git a/Cargo.toml b/Cargo.toml index 5b3947a6..32ad944b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -38,8 +38,8 @@ jmespatch = { version = "^0.3", features = ["sync"], optional = true } kmip = { version = "0.4.2", package = "kmip-protocol", features = [ "tls-with-openssl", ], optional = true } -kvx = { version = "0.8.0", features = ["macros"] } -# kvx = { version = "0.6.0", git = "https://github.com/nlnetlabs/kvx", features = ["macros"] } +kvx = { version = "0.9.0", features = ["macros"] } +# kvx = { version = "0.8.0", git = "https://github.com/nlnetlabs/kvx", branch = "schedule-without-finish", features = [ "macros"] } libflate = "^1" log = "^0.4" once_cell = { version = "^1.7.2", optional = true } diff --git a/src/commons/error.rs b/src/commons/error.rs index 39e1f303..998ba6be 100644 --- a/src/commons/error.rs +++ b/src/commons/error.rs @@ -151,6 +151,23 @@ impl From for ApiAuthError { } } +//------------ FatalError -------------------------------------------------- + +/// Wraps an error so horrible to contemplate that it should result in +/// a server crash, as it would have lost its reason to live. +/// +/// Note that we do not provide any From for this in an attempt +/// to ensure that this is only ever used explicitly and when it is +/// appropriate. +#[derive(Debug)] +pub struct FatalError(pub Error); + +impl fmt::Display for FatalError { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + self.0.fmt(f) + } +} + //------------ Error ------------------------------------------------------- #[derive(Debug)] diff --git a/src/daemon/mq.rs b/src/daemon/mq.rs index 888d8f39..dfd85bbb 100644 --- a/src/daemon/mq.rs +++ b/src/daemon/mq.rs @@ -216,7 +216,7 @@ impl TaskQueue { trace!( "fnd task: {} with priority: {}", pending.name, - Priority(pending.timestamp as i64) + Priority::from_timestamp_ms(pending.timestamp_millis) ); Some(pending) } @@ -227,14 +227,16 @@ impl TaskQueue { /// was already present, then it will get the highest of the two /// priorities. /// - /// Many tasks are planned with a high priority - e.g. if they are - /// triggered through CA events. Other tasks may be planned for the - /// future (e.g. sync with parent tomorrow). The latter can be moved - /// forward when circumstances dictate. - /// - /// Recurring tasks will typically be re-added by the Scheduler when - /// needed (and can then be moved forward if needed). + /// This will NOT finish any possible running task by the same + /// name. pub fn schedule(&self, task: Task, priority: Priority) -> KrillResult<()> { + self.schedule_task(task, ScheduleMode::ReplaceExistingSoonest, priority) + } + + /// Schedules a task for the given priority. If the equivalent task + /// was already present, then it will get the highest of the two + /// priorities. + pub fn schedule_and_finish_existing(&self, task: Task, priority: Priority) -> KrillResult<()> { self.schedule_task(task, ScheduleMode::FinishOrReplaceExistingSoonest, priority) } @@ -244,24 +246,26 @@ impl TaskQueue { fn schedule_task(&self, task: Task, mode: ScheduleMode, priority: Priority) -> KrillResult<()> { let task_name = task.name()?; - trace!("add task: {} with priority: {}", task_name, priority.to_string()); + debug!("add task: {} with priority: {}", task_name, priority.to_string()); let json = serde_json::to_value(&task) .map_err(|e| Error::Custom(format!("could not serialize task {}. error: {}", task_name, e)))?; self.q - .schedule_task(task_name, json, Some(priority.into()), mode) + .schedule_task(task_name, json, Some(priority.to_millis()), mode) .map_err(Error::from) } /// Finish a running task, without rescheduling it. pub fn finish(&self, task: &kvx::Key) -> KrillResult<()> { + debug!("Finish task: {}", task); self.q.finish_running_task(task).map_err(Error::from) } /// Reschedule a running task, without finishing it. pub fn reschedule(&self, task: &kvx::Key, priority: Priority) -> KrillResult<()> { + debug!("Reschedule task: {} to: {}", task, priority); self.q - .reschedule_running_task(task, Some(priority.into())) + .reschedule_running_task(task, Some(priority.to_millis())) .map_err(Error::from) } @@ -442,7 +446,7 @@ impl eventsourcing::PostSaveEventListener for TaskQueue { match event { CertAuthEvent::ChildUpdatedResources { child, .. } | CertAuthEvent::ChildKeyRevoked { child, .. } => { debug!("Schedule a sync from the child to this CA as their parent. This will be a no-op for remote children."); - if let Err(e) = self.schedule( + if let Err(e) = self.schedule_and_finish_existing( Task::SyncParent { ca_handle: child.convert(), ca_version: 0, // no need to wait for updated child @@ -541,12 +545,21 @@ impl eventsourcing::PostSaveEventListener for TaskQueue { #[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] pub struct Priority(i64); -pub fn now() -> Priority { - Time::now().into() +impl Priority { + /// Convenience function, but note that we don't + /// have ms granularity here, so this is rounded + /// down to seconds. + pub fn from_timestamp_ms(millis: u128) -> Self { + Priority((millis / 1000) as i64) + } + + pub fn to_millis(&self) -> u128 { + (self.0 * 1000) as u128 + } } -pub fn in_millis(millis: i64) -> Priority { - (Time::now() + chrono::Duration::milliseconds(millis)).into() +pub fn now() -> Priority { + Time::now().into() } pub fn in_seconds(secs: i64) -> Priority { @@ -604,11 +617,3 @@ impl From