diff --git a/Cargo.lock b/Cargo.lock index bac5b07..e394e9c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -120,6 +120,21 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "bit-set" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08807e080ed7f9d5433fa9b275196cfc35414f66a0c79d864dc51a0d825231a3" +dependencies = [ + "bit-vec", +] + +[[package]] +name = "bit-vec" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e764a1d40d510daf35e07be9eb06e75770908c27d411ee6c92109c9840eaaf7" + [[package]] name = "bitflags" version = "2.10.0" @@ -548,6 +563,12 @@ version = "0.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "52051878f80a721bb68ebfbc930e07b65ba72f2da88968ea5c06fd6ca3d3a127" +[[package]] +name = "fnv" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" + [[package]] name = "foldhash" version = "0.1.5" @@ -727,6 +748,9 @@ dependencies = [ [[package]] name = "invariant-scheduler" version = "0.1.0" +dependencies = [ + "proptest", +] [[package]] name = "invariant-types" @@ -948,6 +972,15 @@ dependencies = [ "serde", ] +[[package]] +name = "ppv-lite86" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" +dependencies = [ + "zerocopy", +] + [[package]] name = "proc-macro2" version = "1.0.103" @@ -957,6 +990,25 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "proptest" +version = "1.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b45fcc2344c680f5025fe57779faef368840d0bd1f42f216291f0dc4ace4744" +dependencies = [ + "bit-set", + "bit-vec", + "bitflags", + "num-traits", + "rand", + "rand_chacha", + "rand_xorshift", + "regex-syntax", + "rusty-fork", + "tempfile", + "unarray", +] + [[package]] name = "pulley-interpreter" version = "38.0.3" @@ -980,6 +1032,12 @@ dependencies = [ "syn", ] +[[package]] +name = "quick-error" +version = "1.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a1d01941d82fa2ab50be1e79e6714289dd7cde78eba4c074bc5a4374f650dfe0" + [[package]] name = "quote" version = "1.0.42" @@ -995,6 +1053,44 @@ version = "5.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" +[[package]] +name = "rand" +version = "0.9.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea" +dependencies = [ + "rand_chacha", + "rand_core", +] + +[[package]] +name = "rand_chacha" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" +dependencies = [ + "ppv-lite86", + "rand_core", +] + +[[package]] +name = "rand_core" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c" +dependencies = [ + "getrandom 0.3.4", +] + +[[package]] +name = "rand_xorshift" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "513962919efc330f829edb2535844d1b912b0fbe2ca165d613e4e8788bb05a5a" +dependencies = [ + "rand_core", +] + [[package]] name = "rayon" version = "1.11.0" @@ -1088,6 +1184,18 @@ version = "1.0.22" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" +[[package]] +name = "rusty-fork" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cc6bf79ff24e648f6da1f8d1f011e9cac26491b619e6b9280f2b47f1774e6ee2" +dependencies = [ + "fnv", + "quick-error", + "tempfile", + "wait-timeout", +] + [[package]] name = "semver" version = "1.0.27" @@ -1417,6 +1525,12 @@ version = "1.19.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "562d481066bde0658276a35467c4af00bdc6ee726305698a55b86e61d7ad82bb" +[[package]] +name = "unarray" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eaea85b334db583fe3274d12b4cd1880032beab409c0d774be044d4480ab9a94" + [[package]] name = "unicode-ident" version = "1.0.22" @@ -1469,6 +1583,15 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "wait-timeout" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ac3b126d3914f9849036f826e054cbabdc8519970b8998ddaf3b5bd3c65f11" +dependencies = [ + "libc", +] + [[package]] name = "wasi" version = "0.11.1+wasi-snapshot-preview1" @@ -2164,6 +2287,26 @@ dependencies = [ "wasmparser 0.239.0", ] +[[package]] +name = "zerocopy" +version = "0.8.50" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b065d4f0e55f82fae73202e189638116a87c55ab6b8e6c2721e13dd9d854ad1" +dependencies = [ + "zerocopy-derive", +] + +[[package]] +name = "zerocopy-derive" +version = "0.8.50" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b631b19d36a892ab55420c92dbc83ccd79274f25be714855d3074aa71cab639" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "zmij" version = "1.0.21" diff --git a/Justfile b/Justfile index b5e7c16..fd5a0c5 100644 --- a/Justfile +++ b/Justfile @@ -79,3 +79,17 @@ _ci: _rust-quality _change-risk ci: nix develop --accept-flake-config -c just _ci + +# Run Kani proof harnesses (bounded model checking). Slow; a separate gate, not part of _ci. +_verify: + cargo kani --tests + +verify: + nix develop --accept-flake-config -c just _verify + +# One-time Kani setup. A no-op inside the Nix dev shell (KANI_HOME is pre-staged); only needed when running cargo kani outside Nix. +_kani-setup: + cargo kani setup + +kani-setup: + nix develop --accept-flake-config -c just _kani-setup diff --git a/crates/invariant-scheduler/Cargo.toml b/crates/invariant-scheduler/Cargo.toml index 8b95538..4d9e289 100644 --- a/crates/invariant-scheduler/Cargo.toml +++ b/crates/invariant-scheduler/Cargo.toml @@ -3,4 +3,14 @@ name = "invariant-scheduler" version = "0.1.0" edition = "2024" +# `kani` is an externally-set cfg (only active when `cargo kani` drives the +# build). Register it so the `unexpected_cfgs` lint recognizes `#[cfg(kani)]` +# on the proof harness instead of flagging it as a typo, while still checking +# every other cfg name. +[lints.rust] +unexpected_cfgs = { level = "warn", check-cfg = ['cfg(kani)'] } + +[dev-dependencies] +proptest = "1.11" + [dependencies] diff --git a/crates/invariant-scheduler/src/domain/error.rs b/crates/invariant-scheduler/src/domain/error.rs index a0f20cc..24f0f8e 100644 --- a/crates/invariant-scheduler/src/domain/error.rs +++ b/crates/invariant-scheduler/src/domain/error.rs @@ -1,12 +1,24 @@ use std::fmt; +/// An error returned by a fallible domain operation. #[derive(Clone, Debug, PartialEq, Eq)] pub enum DomainError { + /// A [`JobId`](crate::domain::JobId) was constructed from an empty string. EmptyJobId, + /// A [`TargetRef`](crate::domain::TargetRef) was constructed from an empty string. EmptyTargetRef, + /// Advancing a [`SchedulerTime`](crate::domain::SchedulerTime) overflowed + /// the representable range. TimeOverflow, + /// Incrementing an [`AttemptNumber`](crate::domain::AttemptNumber) overflowed + /// `u32::MAX`. AttemptOverflow, + /// A transition was attempted on a job already in a terminal status. JobTerminal, + /// A transition was attempted that is not legal from the job's current + /// non-terminal status. + IllegalTransition, + /// The ready queue could not accept another job because it is at capacity. QueueFull, } @@ -18,6 +30,7 @@ impl fmt::Display for DomainError { Self::TimeOverflow => "scheduler time overflow", Self::AttemptOverflow => "attempt number overflow", Self::JobTerminal => "job is terminal", + Self::IllegalTransition => "illegal job state transition", Self::QueueFull => "ready queue is full", }; f.write_str(message) @@ -38,6 +51,10 @@ mod tests { (DomainError::TimeOverflow, "scheduler time overflow"), (DomainError::AttemptOverflow, "attempt number overflow"), (DomainError::JobTerminal, "job is terminal"), + ( + DomainError::IllegalTransition, + "illegal job state transition", + ), (DomainError::QueueFull, "ready queue is full"), ]; diff --git a/crates/invariant-scheduler/src/domain/job.rs b/crates/invariant-scheduler/src/domain/job.rs new file mode 100644 index 0000000..674b36b --- /dev/null +++ b/crates/invariant-scheduler/src/domain/job.rs @@ -0,0 +1,697 @@ +use crate::domain::{Deadline, DomainError, Priority, ReadyAt}; + +/// A non-empty identifier that uniquely names a job. +/// +/// Identifiers order lexicographically by their string value. +/// +/// # Examples +/// +/// ``` +/// use invariant_scheduler::domain::JobId; +/// +/// let id = JobId::new("job-1")?; +/// assert_eq!(id.as_str(), "job-1"); +/// # Ok::<(), invariant_scheduler::domain::DomainError>(()) +/// ``` +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct JobId(String); + +impl JobId { + /// Creates a job identifier from any string-like value. + /// + /// # Errors + /// + /// Returns [`DomainError::EmptyJobId`] if `id` is empty. + pub fn new(id: impl Into) -> Result { + let value = id.into(); + if value.is_empty() { + return Err(DomainError::EmptyJobId); + } + Ok(Self(value)) + } + + /// Returns the identifier as a string slice. + #[must_use] + pub fn as_str(&self) -> &str { + self.0.as_str() + } +} + +/// A non-empty reference to the target a job executes against. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct TargetRef(String); + +impl TargetRef { + /// Creates a target reference from any string-like value. + /// + /// # Errors + /// + /// Returns [`DomainError::EmptyTargetRef`] if `target` is empty. + pub fn new(target: impl Into) -> Result { + let value = target.into(); + if value.is_empty() { + return Err(DomainError::EmptyTargetRef); + } + Ok(Self(value)) + } + + /// Returns the target reference as a string slice. + #[must_use] + pub fn as_str(&self) -> &str { + self.0.as_str() + } +} + +/// The zero-based index of a job's current attempt; the first attempt is zero. +/// +/// A job starts at [`zero`](Self::zero) and the index is incremented by +/// [`Job::retry`]. It is an ordinal, not a count: a job on its first attempt +/// has value `0`. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct AttemptNumber(u32); + +impl AttemptNumber { + /// Returns the index of a job's first attempt (zero), before any retry. + pub const fn zero() -> Self { + Self(0) + } + + /// Returns the raw zero-based attempt index. + #[must_use] + pub const fn value(self) -> u32 { + self.0 + } + + /// Returns the next attempt number. + /// + /// # Errors + /// + /// Returns [`DomainError::AttemptOverflow`] if the current value is + /// `u32::MAX`. + pub fn checked_next(self) -> Result { + self.0 + .checked_add(1) + .map(Self) + .ok_or(DomainError::AttemptOverflow) + } +} + +/// The lifecycle status of a [`Job`]. +/// +/// The live (non-terminal) states are [`Queued`](Self::Queued), +/// [`Running`](Self::Running), and [`Failed`](Self::Failed). The terminal +/// states, from which no transition is legal, are [`Completed`](Self::Completed), +/// [`Cancelled`](Self::Cancelled), and [`Exhausted`](Self::Exhausted). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum JobStatus { + /// Waiting to start. A newly created job begins here. + Queued, + /// Currently executing. + Running, + /// A run failed; the job may be retried, exhausted, or cancelled. + Failed, + /// Finished successfully. Terminal. + Completed, + /// Abandoned before completion. Terminal. + Cancelled, + /// Out of retry attempts after failing. Terminal. + Exhausted, +} + +impl JobStatus { + /// Returns `true` if this is a terminal status, namely + /// [`Completed`](Self::Completed), [`Cancelled`](Self::Cancelled), or + /// [`Exhausted`](Self::Exhausted). + #[must_use] + pub const fn is_terminal(self) -> bool { + matches!( + self, + JobStatus::Completed | JobStatus::Cancelled | JobStatus::Exhausted + ) + } +} + +/// A unit of schedulable work and its lifecycle state. +/// +/// A job is created in [`JobStatus::Queued`] with attempt number zero and moves +/// through its [`JobStatus`] state machine via the transition methods +/// ([`start`](Self::start), [`complete`](Self::complete), [`fail`](Self::fail), +/// [`cancel`](Self::cancel), [`exhaust`](Self::exhaust), [`retry`](Self::retry)). +/// +/// # Examples +/// +/// ``` +/// use invariant_scheduler::domain::{Job, JobId, JobStatus, Priority, TargetRef}; +/// +/// let mut job = Job::new( +/// JobId::new("job-1")?, +/// TargetRef::new("component:greet")?, +/// Priority::DEFAULT, +/// None, +/// None, +/// ); +/// assert_eq!(job.status(), JobStatus::Queued); +/// +/// job.start()?; +/// job.complete()?; +/// assert_eq!(job.status(), JobStatus::Completed); +/// # Ok::<(), invariant_scheduler::domain::DomainError>(()) +/// ``` +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Job { + id: JobId, + target: TargetRef, + priority: Priority, + deadline: Option, + ready_at: Option, + status: JobStatus, + attempt: AttemptNumber, +} + +impl Job { + /// Creates a job in [`JobStatus::Queued`] with attempt number zero. + /// + /// `deadline` and `ready_at` are optional scheduling hints; pass `None` when + /// the job has no completion target or readiness gate. + pub fn new( + id: JobId, + target: TargetRef, + priority: Priority, + deadline: Option, + ready_at: Option, + ) -> Self { + Self { + id, + target, + priority, + deadline, + ready_at, + attempt: AttemptNumber::zero(), + status: JobStatus::Queued, + } + } + + /// Returns the job's identifier. + #[must_use] + pub fn id(&self) -> &JobId { + &self.id + } + + /// Returns the reference to the target this job executes against. + #[must_use] + pub fn target(&self) -> &TargetRef { + &self.target + } + + /// Returns the job's scheduling priority. + #[must_use] + pub fn priority(&self) -> Priority { + self.priority + } + + /// Returns the job's deadline, if one was set. + #[must_use] + pub fn deadline(&self) -> Option { + self.deadline + } + + /// Returns the job's readiness gate, if one was set. + #[must_use] + pub fn ready_at(&self) -> Option { + self.ready_at + } + + /// Returns the job's current lifecycle status. + #[must_use] + pub fn status(&self) -> JobStatus { + self.status + } + + /// Returns how many times the job has been attempted. + #[must_use] + pub fn attempt(&self) -> AttemptNumber { + self.attempt + } + + /// Classifies a rejected transition: terminal source vs. live-but-illegal. + fn reject(&self) -> DomainError { + if self.status.is_terminal() { + DomainError::JobTerminal + } else { + DomainError::IllegalTransition + } + } + + /// Transitions the job from [`Queued`](JobStatus::Queued) to + /// [`Running`](JobStatus::Running). + /// + /// # Errors + /// + /// Returns [`DomainError::IllegalTransition`] if the job is in any other + /// live status, or [`DomainError::JobTerminal`] if it is terminal. The + /// status is left unchanged on error. + pub fn start(&mut self) -> Result<(), DomainError> { + match self.status { + JobStatus::Queued => { + self.status = JobStatus::Running; + Ok(()) + } + _ => Err(self.reject()), + } + } + + /// Transitions the job from [`Running`](JobStatus::Running) to the terminal + /// [`Completed`](JobStatus::Completed). + /// + /// # Errors + /// + /// Returns [`DomainError::IllegalTransition`] if the job is in any other + /// live status, or [`DomainError::JobTerminal`] if it is terminal. The + /// status is left unchanged on error. + pub fn complete(&mut self) -> Result<(), DomainError> { + match self.status { + JobStatus::Running => { + self.status = JobStatus::Completed; + Ok(()) + } + _ => Err(self.reject()), + } + } + + /// Transitions the job from [`Running`](JobStatus::Running) to + /// [`Failed`](JobStatus::Failed), from which it may be retried, exhausted, + /// or cancelled. + /// + /// # Errors + /// + /// Returns [`DomainError::IllegalTransition`] if the job is in any other + /// live status, or [`DomainError::JobTerminal`] if it is terminal. The + /// status is left unchanged on error. + pub fn fail(&mut self) -> Result<(), DomainError> { + match self.status { + JobStatus::Running => { + self.status = JobStatus::Failed; + Ok(()) + } + _ => Err(self.reject()), + } + } + + /// Transitions the job from any live status + /// ([`Queued`](JobStatus::Queued), [`Running`](JobStatus::Running), or + /// [`Failed`](JobStatus::Failed)) to the terminal + /// [`Cancelled`](JobStatus::Cancelled). + /// + /// # Errors + /// + /// Returns [`DomainError::JobTerminal`] if the job is already terminal. The + /// status is left unchanged on error. + pub fn cancel(&mut self) -> Result<(), DomainError> { + match self.status { + JobStatus::Queued | JobStatus::Running | JobStatus::Failed => { + self.status = JobStatus::Cancelled; + Ok(()) + } + _ => Err(self.reject()), + } + } + + /// Transitions the job from [`Failed`](JobStatus::Failed) to the terminal + /// [`Exhausted`](JobStatus::Exhausted), marking it as out of retries. + /// + /// # Errors + /// + /// Returns [`DomainError::IllegalTransition`] if the job is in any other + /// live status, or [`DomainError::JobTerminal`] if it is terminal. The + /// status is left unchanged on error. + pub fn exhaust(&mut self) -> Result<(), DomainError> { + match self.status { + JobStatus::Failed => { + self.status = JobStatus::Exhausted; + Ok(()) + } + _ => Err(self.reject()), + } + } + + /// Transitions the job from [`Failed`](JobStatus::Failed) back to + /// [`Queued`](JobStatus::Queued), incrementing its attempt number. + /// + /// The operation is atomic: if the attempt number cannot be incremented, + /// neither the attempt nor the status is changed. + /// + /// # Errors + /// + /// Returns [`DomainError::AttemptOverflow`] if the attempt number is already + /// `u32::MAX`. Returns [`DomainError::IllegalTransition`] if the job is in + /// any other live status, or [`DomainError::JobTerminal`] if it is terminal. + /// The status is left unchanged on error. + pub fn retry(&mut self) -> Result<(), DomainError> { + match self.status { + JobStatus::Failed => { + // Bump the attempt first: the fallible step must commit + // before any status mutation, so retry stays atomic. + self.attempt = self.attempt.checked_next()?; + self.status = JobStatus::Queued; + Ok(()) + } + _ => Err(self.reject()), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A fresh `Job` forced into `status`. Direct field access is legal + /// because this test module is a descendant of the `job` module. + fn job_in(status: JobStatus) -> Job { + let mut job = Job::new( + JobId::new("job-1").unwrap(), + TargetRef::new("component:greet").unwrap(), + Priority::DEFAULT, + None, + None, + ); + job.status = status; + job + } + + mod job_id { + use super::*; + + #[test] + fn rejects_empty() { + assert_eq!(JobId::new(""), Err(DomainError::EmptyJobId)); + } + + #[test] + fn preserves_value() { + assert_eq!(JobId::new("job-1").unwrap().as_str(), "job-1"); + } + + #[test] + fn orders_lexicographically() { + assert!(JobId::new("a").unwrap() < JobId::new("b").unwrap()); + } + } + + mod target_ref { + use super::*; + + #[test] + fn rejects_empty() { + assert_eq!(TargetRef::new(""), Err(DomainError::EmptyTargetRef)); + } + + #[test] + fn preserves_value() { + assert_eq!( + TargetRef::new("component:greet").unwrap().as_str(), + "component:greet" + ); + } + } + + mod attempt_number { + use super::*; + + #[test] + fn starts_at_zero() { + assert_eq!(AttemptNumber::zero().value(), 0); + } + + #[test] + fn increments() { + assert_eq!(AttemptNumber::zero().checked_next().unwrap().value(), 1); + } + + #[test] + fn rejects_overflow() { + assert_eq!( + AttemptNumber(u32::MAX).checked_next(), + Err(DomainError::AttemptOverflow) + ); + } + } + + mod job_status { + use super::*; + + #[test] + fn terminal_states_are_terminal() { + assert!(JobStatus::Completed.is_terminal()); + assert!(JobStatus::Cancelled.is_terminal()); + assert!(JobStatus::Exhausted.is_terminal()); + } + + #[test] + fn live_states_are_not_terminal() { + assert!(!JobStatus::Queued.is_terminal()); + assert!(!JobStatus::Running.is_terminal()); + assert!(!JobStatus::Failed.is_terminal()); // retryable, the key case + } + } + + mod transitions { + use super::*; + + /// Every transition the `Job` state machine exposes, so the table can + /// drive each one uniformly. + #[derive(Debug, Clone, Copy, PartialEq, Eq)] + enum Event { + Start, + Complete, + Fail, + Cancel, + Exhaust, + Retry, + } + + /// Expected outcome of one (state, event) cell. + #[derive(Debug, Clone, PartialEq, Eq)] + enum Expect { + /// Legal: succeeds and lands in this status. + Lands(JobStatus), + /// Rejected: returns this error and leaves the status untouched. + Rejects(DomainError), + } + + const ALL_STATES: [JobStatus; 6] = [ + JobStatus::Queued, + JobStatus::Running, + JobStatus::Failed, + JobStatus::Completed, + JobStatus::Cancelled, + JobStatus::Exhausted, + ]; + + const ALL_EVENTS: [Event; 6] = [ + Event::Start, + Event::Complete, + Event::Fail, + Event::Cancel, + Event::Exhaust, + Event::Retry, + ]; + + fn apply(job: &mut Job, event: Event) -> Result<(), DomainError> { + match event { + Event::Start => job.start(), + Event::Complete => job.complete(), + Event::Fail => job.fail(), + Event::Cancel => job.cancel(), + Event::Exhaust => job.exhaust(), + Event::Retry => job.retry(), + } + } + + /// The complete transition relation: 6 states x 6 events = 36 cells. + /// Legal edges name their target; every other cell is a rejection, + /// classified `JobTerminal` from a terminal source else `IllegalTransition`. + #[rustfmt::skip] + const TABLE: &[(JobStatus, Event, Expect)] = { + use DomainError::{IllegalTransition, JobTerminal}; + use Event::{Cancel, Complete, Exhaust, Fail, Retry, Start}; + use Expect::{Lands, Rejects}; + use JobStatus::{Cancelled, Completed, Exhausted, Failed, Queued, Running}; + &[ + // from Queued (live) + (Queued, Start, Lands(Running)), + (Queued, Complete, Rejects(IllegalTransition)), + (Queued, Fail, Rejects(IllegalTransition)), + (Queued, Cancel, Lands(Cancelled)), + (Queued, Exhaust, Rejects(IllegalTransition)), + (Queued, Retry, Rejects(IllegalTransition)), + // from Running (live) + (Running, Start, Rejects(IllegalTransition)), + (Running, Complete, Lands(Completed)), + (Running, Fail, Lands(Failed)), + (Running, Cancel, Lands(Cancelled)), + (Running, Exhaust, Rejects(IllegalTransition)), + (Running, Retry, Rejects(IllegalTransition)), + // from Failed (live, retryable) + (Failed, Start, Rejects(IllegalTransition)), + (Failed, Complete, Rejects(IllegalTransition)), + (Failed, Fail, Rejects(IllegalTransition)), + (Failed, Cancel, Lands(Cancelled)), + (Failed, Exhaust, Lands(Exhausted)), + (Failed, Retry, Lands(Queued)), + // from Completed (terminal) + (Completed, Start, Rejects(JobTerminal)), + (Completed, Complete, Rejects(JobTerminal)), + (Completed, Fail, Rejects(JobTerminal)), + (Completed, Cancel, Rejects(JobTerminal)), + (Completed, Exhaust, Rejects(JobTerminal)), + (Completed, Retry, Rejects(JobTerminal)), + // from Cancelled (terminal) + (Cancelled, Start, Rejects(JobTerminal)), + (Cancelled, Complete, Rejects(JobTerminal)), + (Cancelled, Fail, Rejects(JobTerminal)), + (Cancelled, Cancel, Rejects(JobTerminal)), + (Cancelled, Exhaust, Rejects(JobTerminal)), + (Cancelled, Retry, Rejects(JobTerminal)), + // from Exhausted (terminal) + (Exhausted, Start, Rejects(JobTerminal)), + (Exhausted, Complete, Rejects(JobTerminal)), + (Exhausted, Fail, Rejects(JobTerminal)), + (Exhausted, Cancel, Rejects(JobTerminal)), + (Exhausted, Exhaust, Rejects(JobTerminal)), + (Exhausted, Retry, Rejects(JobTerminal)), + ] + }; + + #[test] + fn table_covers_every_state_event_pair_exactly_once() { + for state in ALL_STATES { + for event in ALL_EVENTS { + let hits = TABLE + .iter() + .filter(|(s, e, _)| *s == state && *e == event) + .count(); + assert_eq!(hits, 1, "{state:?} + {event:?} must appear exactly once"); + } + } + assert_eq!(TABLE.len(), 36); + } + + #[test] + fn every_transition_matches_the_table() { + for (from, event, expect) in TABLE { + let mut job = job_in(*from); + let result = apply(&mut job, *event); + match expect { + Expect::Lands(target) => { + assert_eq!(result, Ok(()), "{from:?} + {event:?} should succeed"); + assert_eq!(job.status(), *target, "{from:?} + {event:?} target"); + } + Expect::Rejects(err) => { + assert_eq!(result, Err(err.clone()), "{from:?} + {event:?} error"); + assert_eq!( + job.status(), + *from, + "{from:?} + {event:?} must not mutate status on rejection" + ); + } + } + } + } + + #[test] + fn retry_increments_attempt_and_requeues() { + let mut job = job_in(JobStatus::Failed); + assert_eq!(job.attempt().value(), 0); + assert_eq!(job.retry(), Ok(())); + assert_eq!(job.status(), JobStatus::Queued); + assert_eq!(job.attempt().value(), 1); + } + + #[test] + fn retry_at_max_attempt_is_atomic() { + let mut job = job_in(JobStatus::Failed); + job.attempt = AttemptNumber(u32::MAX); + assert_eq!(job.retry(), Err(DomainError::AttemptOverflow)); + // Atomicity: the failed bump leaves both fields untouched. + assert_eq!(job.status(), JobStatus::Failed); + assert_eq!(job.attempt().value(), u32::MAX); + } + } + + mod property_tests { + use super::*; + use proptest::prelude::*; + + fn terminal_status() -> impl Strategy { + prop_oneof![ + Just(JobStatus::Completed), + Just(JobStatus::Cancelled), + Just(JobStatus::Exhausted) + ] + } + type Transition = fn(&mut Job) -> Result<(), DomainError>; + + const TRANSITIONS: [Transition; 6] = [ + Job::start, + Job::complete, + Job::fail, + Job::cancel, + Job::exhaust, + Job::retry, + ]; + + proptest! { + #[test] + fn terminal_states_are_sinks(status in terminal_status()) { + for transition in TRANSITIONS { + let mut job = job_in(status); + prop_assert_eq!(transition(&mut job), Err(DomainError::JobTerminal)); + prop_assert_eq!(job.status(), status); + } + } + + #[test] + fn retry_increments_or_leaves_untouched(attempt in any::()) { + let mut job = job_in(JobStatus::Failed); + job.attempt = AttemptNumber(attempt); + let pre = job.clone(); + + match job.retry() { + Ok(()) => { + prop_assert_eq!(job.attempt().value(), attempt + 1); + prop_assert_eq!(job.status(), JobStatus::Queued); + } + Err(err) => { + prop_assert_eq!(err, DomainError::AttemptOverflow); + // Atomicity: a failed bump leaves the whole job untouched. + prop_assert_eq!(job, pre); + } + } + } + } + } + + #[cfg(kani)] + mod proofs { + use super::*; + + #[kani::proof] + fn retry_is_atomic_for_all_attempts() { + let attempt: u32 = kani::any(); // symbolic: ALL 2^32 values at once + let mut job = job_in(JobStatus::Failed); + job.attempt = AttemptNumber(attempt); + let before = job.clone(); + + match job.retry() { + Ok(()) => { + assert_eq!(job.attempt().value(), attempt + 1); + assert_eq!(job.status(), JobStatus::Queued); + } + Err(e) => { + assert_eq!(e, DomainError::AttemptOverflow); + assert_eq!(job, before); // atomicity, proven for every value + } + } + } + } +} diff --git a/crates/invariant-scheduler/src/domain/mod.rs b/crates/invariant-scheduler/src/domain/mod.rs index c7cc940..55bb05e 100644 --- a/crates/invariant-scheduler/src/domain/mod.rs +++ b/crates/invariant-scheduler/src/domain/mod.rs @@ -1,5 +1,17 @@ +//! Pure domain model for scheduling. +//! +//! Defines the [`Job`] aggregate and its [`JobStatus`] state machine, the +//! scheduling value objects ([`Priority`], [`Deadline`], [`ReadyAt`]), the +//! [`SchedulerTime`] timeline value, and the [`DomainError`] type shared across +//! fallible operations. This module has no I/O, concurrency, or external +//! dependencies. + pub mod error; -pub mod time; +mod job; +mod priority; +mod time; pub use error::DomainError; +pub use job::{AttemptNumber, Job, JobId, JobStatus, TargetRef}; +pub use priority::{Deadline, Priority, ReadyAt}; pub use time::SchedulerTime; diff --git a/crates/invariant-scheduler/src/domain/priority.rs b/crates/invariant-scheduler/src/domain/priority.rs new file mode 100644 index 0000000..fd3bf10 --- /dev/null +++ b/crates/invariant-scheduler/src/domain/priority.rs @@ -0,0 +1,124 @@ +//! Scheduling value objects for the local execution layer. +//! +//! Immutable, `Copy` newtypes in the pure domain: priority, deadline, and readiness gate. + +use crate::domain::SchedulerTime; + +/// Scheduling priority where a higher value is more urgent. +/// +/// Priority is a `u8`, giving 256 levels ordered from [`MIN`](Self::MIN) +/// (least urgent) to [`MAX`](Self::MAX) (most urgent). +/// +/// # Examples +/// +/// ``` +/// use invariant_scheduler::domain::Priority; +/// +/// assert!(Priority::MAX > Priority::DEFAULT); +/// assert_eq!(Priority::new(200).value(), 200); +/// ``` +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct Priority(u8); + +impl Priority { + /// Least urgent priority. + pub const MIN: Self = Self(u8::MIN); + /// Most urgent priority. + pub const MAX: Self = Self(u8::MAX); + /// Neutral baseline at the middle of the range, leaving headroom in both + /// directions. + pub const DEFAULT: Self = Self(128); + + /// Creates a priority from a raw level, where higher is more urgent. + pub const fn new(value: u8) -> Self { + Self(value) + } + + /// Returns the raw priority level. + #[must_use] + pub const fn value(self) -> u8 { + self.0 + } +} + +impl Default for Priority { + fn default() -> Self { + Self::DEFAULT + } +} + +/// The point by which a job should finish. +/// +/// A deadline is best-effort: non-preemptive multi-worker scheduling cannot +/// guarantee it is met. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct Deadline(SchedulerTime); + +impl Deadline { + /// Creates a deadline at the given scheduler time. + pub const fn new(time: SchedulerTime) -> Self { + Self(time) + } + + /// Returns the scheduler time at which this deadline falls. + #[must_use] + pub const fn time(self) -> SchedulerTime { + self.0 + } +} + +/// The earliest time a job may start. +/// +/// This is a readiness gate that controls eligibility; it is not an ordering or +/// sort key. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct ReadyAt(SchedulerTime); + +impl ReadyAt { + /// Creates a readiness gate at the given scheduler time. + pub const fn new(time: SchedulerTime) -> Self { + Self(time) + } + + /// Returns the scheduler time before which the job is not eligible to start. + #[must_use] + pub const fn time(self) -> SchedulerTime { + self.0 + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::domain::SchedulerTime; + + #[test] + fn priority_default_is_mid_range_baseline() { + assert_eq!(Priority::default(), Priority::DEFAULT); + assert_eq!(Priority::DEFAULT.value(), 128); + } + + #[test] + fn priority_min_max_pin_the_u8_range() { + assert_eq!(Priority::MIN.value(), 0); + assert_eq!(Priority::MAX.value(), 255); + } + + #[test] + fn priority_orders_higher_value_as_greater() { + assert!(Priority::MAX > Priority::DEFAULT); + assert!(Priority::DEFAULT > Priority::MIN); + } + + #[test] + fn deadline_round_trips_its_time() { + let t = SchedulerTime::from_millis_since_epoch(42); + assert_eq!(Deadline::new(t).time(), t); + } + + #[test] + fn ready_at_round_trips_its_time() { + let t = SchedulerTime::from_millis_since_epoch(42); + assert_eq!(ReadyAt::new(t).time(), t); + } +} diff --git a/crates/invariant-scheduler/src/domain/time.rs b/crates/invariant-scheduler/src/domain/time.rs index 50a9985..cf1eb44 100644 --- a/crates/invariant-scheduler/src/domain/time.rs +++ b/crates/invariant-scheduler/src/domain/time.rs @@ -11,22 +11,32 @@ use crate::domain::error::DomainError; pub struct SchedulerTime(Duration); impl SchedulerTime { + /// Creates a time from a [`Duration`] elapsed since the scheduler epoch. pub fn from_duration_since_epoch(value: Duration) -> Self { Self(value) } + /// Creates a time from milliseconds elapsed since the scheduler epoch. pub fn from_millis_since_epoch(value: u64) -> Self { Self(Duration::from_millis(value)) } + /// Creates a time from seconds elapsed since the scheduler epoch. pub fn from_secs_since_epoch(value: u64) -> Self { Self(Duration::from_secs(value)) } + /// Returns the [`Duration`] elapsed since the scheduler epoch. pub fn duration_since_epoch(self) -> Duration { self.0 } + /// Returns this time advanced by `duration`. + /// + /// # Errors + /// + /// Returns [`DomainError::TimeOverflow`] if the result would exceed the + /// representable range. pub fn checked_add_duration(self, duration: Duration) -> Result { self.0 .checked_add(duration) diff --git a/crates/invariant-scheduler/src/lib.rs b/crates/invariant-scheduler/src/lib.rs index d7abca1..bf44f91 100644 --- a/crates/invariant-scheduler/src/lib.rs +++ b/crates/invariant-scheduler/src/lib.rs @@ -1 +1,6 @@ +//! Scheduling primitives for the local execution layer. +//! +//! The [`domain`] module holds the pure, dependency-free core: jobs, their +//! state machine, scheduling value objects, and the errors they can produce. + pub mod domain; diff --git a/flake.nix b/flake.nix index 3a21723..263fca6 100644 --- a/flake.nix +++ b/flake.nix @@ -58,6 +58,134 @@ # change-risk check. doCheck = false; }; + # Kani Rust verifier CLI (`cargo kani` / `kani`). + # + # Kani is not a Cargo dependency of this workspace: the + # `kani::any()` / `#[kani::proof]` API is injected by the Kani + # toolchain at verification time and our usage is gated behind + # `#[cfg(kani)]`, so normal `cargo build`/`cargo test` never see it. + # What the dev shell needs is the `cargo kani` command itself. + # + # nixpkgs has no `kani` package, so we build the thin + # `kani-verifier` installer crate here. That gives us the + # `cargo-kani`/`kani` proxy binaries on PATH. + # + # As with `cargo-crap` above, deps are vendored from the lock file + # shipped inside the published crate (`cargoLock.lockFile`) instead + # of `cargoHash`, to avoid crates.io's HTTP 403 WAF rejection of + # buildRustPackage's python-requests fetch path. + kaniVerifier = rustPlatform.buildRustPackage (finalAttrs: { + pname = "kani-verifier"; + version = "0.67.0"; + + src = pkgs.fetchCrate { + inherit (finalAttrs) pname version; + hash = "sha256-m0khwmHJAiEtICN/f2IE70A2/0JNKwaL3so429YtdOY="; + }; + + cargoLock.lockFile = "${finalAttrs.src}/Cargo.lock"; + + doCheck = false; + }); + + # --------------------------------------------------------------- + # Hermetic Kani bundle (replaces the runtime `cargo kani setup`). + # + # The `kani-verifier` proxy binaries above only know how to *find + # and exec* the real `kani-driver`; on first use they normally run + # `cargo kani setup`, which (a) `curl`s the platform release bundle + # from GitHub into `~/.kani/kani-/`, and (b) `rustup`-installs + # the exact nightly that bundle was built against and symlinks it to + # `~/.kani/kani-/toolchain`. Both steps hit the network and are + # the only non-Nix dependency left in the dev shell. + # + # We reproduce that already-set-up layout entirely through Nix: + # + # * `kaniBundleSrc` — the official prebuilt release tarball, + # fetched by `fetchurl` with a PINNED sha256 (a fixed-output + # derivation; prebuilt-binary toolchains are fetched this way + # throughout nixpkgs). The bundle ships its own `cbmc`, + # `goto-cc`, `goto-instrument`, `goto-analyzer`, `kissat` and + # `kani-compiler` under `bin/`, plus the precompiled kani libs. + # + # * `kaniRustToolchain` — the matching nightly from fenix. The + # bundle does NOT ship a toolchain; it only records the version + # it needs (`rust-toolchain-version` = nightly-2025-11-21, + # `rustc-version` = 1.93.0-nightly (53732d5e0 2025-11-20)). + # fenix builds that exact nightly hermetically, replacing the + # `rustup toolchain install` step. The pinned manifest hash is + # verified by fenix's own fixed-output derivation. + # + # * `kaniHome` — assembles `$out/kani-/` to look exactly like + # a completed `cargo kani setup`: the unpacked bundle plus a + # `toolchain` symlink into the fenix nightly. The driver finds + # everything relative to its own location + # (`base_folder = bin/..`) and, in Release mode, invokes + # `kani-/toolchain/bin/cargo` directly, so a read-only + # store path is sufficient (the driver writes only to the + # project's `target/`). + # + # The dev shell then sets `KANI_HOME` to `$out`; the proxy reads it + # (`${KANI_HOME}/kani-`), sees the layout already present + # (`appears_setup()`), and execs the bundled driver with NO network + # access and NO `~/.kani` download. + # + # On Linux the prebuilt ELF binaries need their interpreter and + # rpath patched for the Nix store (the same fixup `cargo kani setup` + # does via its NixOS `patchelf` hack); on macOS the Mach-O binaries + # run as-is. + kaniVersion = "0.67.0"; + kaniTarget = { + aarch64-darwin = "aarch64-apple-darwin"; + x86_64-darwin = "x86_64-apple-darwin"; + x86_64-linux = "x86_64-unknown-linux-gnu"; + aarch64-linux = "aarch64-unknown-linux-gnu"; + }.${system}; + kaniBundleHash = { + aarch64-darwin = "sha256-f9C3ETCqN70eNG66Z1qtCem2bks4P3B/xlczDOsp7uw="; + x86_64-darwin = throw "kani bundle hash not yet pinned for x86_64-darwin; prefetch the release tarball and add its sha256"; + x86_64-linux = "sha256-O196/TtRYD7nINt7wbxP5GtaT1022q2ZOcS0xli1GsA="; + aarch64-linux = "sha256-l0Eo9E3UNhigbSHl/m2f9nGI3lmG/gvFe1NLDkY577k="; + }.${system}; + kaniBundleSrc = pkgs.fetchurl { + url = "https://github.com/model-checking/kani/releases/download/kani-${kaniVersion}/kani-${kaniVersion}-${kaniTarget}.tar.gz"; + hash = kaniBundleHash; + }; + # The exact nightly the 0.67.0 bundle was built against. Pinned by + # date; fenix verifies the channel manifest with `sha256` below. + kaniRustToolchain = + (fenix.packages.${system}.toolchainOf { + channel = "nightly"; + date = "2025-11-21"; + sha256 = "sha256-P39FCgpfDT04989+ZTNEdM/k/AE869JKSB4qjatYTSs="; + }).toolchain; + kaniHome = pkgs.runCommand "kani-home-${kaniVersion}" + { + nativeBuildInputs = pkgs.lib.optionals pkgs.stdenv.isLinux [ pkgs.patchelf ]; + } + '' + dir="$out/kani-${kaniVersion}" + mkdir -p "$dir" + tar --strip-components=1 -xzf ${kaniBundleSrc} -C "$dir" + + # Recreate the `cargo kani setup` toolchain symlink, but point + # it at the hermetic fenix nightly instead of a rustup install. + ln -s ${kaniRustToolchain} "$dir/toolchain" + + ${pkgs.lib.optionalString pkgs.stdenv.isLinux '' + # Linux release binaries are prebuilt against a standard FHS + # loader and a system libstdc++; patch them for the Nix store. + interp="$(cat ${pkgs.stdenv.cc}/nix-support/dynamic-linker)" + rpath="${pkgs.stdenv.cc.cc.lib}/lib" + for f in kani-compiler kani-driver; do + patchelf --set-interpreter "$interp" "$dir/bin/$f" || true + done + for f in cbmc goto-analyzer goto-cc goto-instrument kissat; do + patchelf --set-interpreter "$interp" "$dir/bin/$f" || true + patchelf --set-rpath "$rpath" "$dir/bin/$f" || true + done + ''} + ''; in { formatter = pkgs.nixfmt-rfc-style; @@ -78,6 +206,16 @@ nodejs # Quint verification tool quint + # Kani Rust verifier (`cargo kani`). The `cargo-kani`/`kani` + # proxy binaries come from this package; the Kani compiler + # bundle + matching nightly toolchain are provided hermetically + # by `kaniHome` (see above) and wired up via `KANI_HOME` below, + # so there is NO runtime `cargo kani setup` download into + # `~/.kani`. The bundle ships its OWN `cbmc`, `goto-cc`, + # `goto-instrument`, `goto-analyzer`, and `kissat` binaries in + # `kani-/bin` (which the driver puts on PATH), so the dev + # shell does NOT need nixpkgs `cbmc`/`kissat`. + kaniVerifier # Testing tools cargo-nextest cargo-llvm-cov @@ -87,6 +225,11 @@ ]; RUST_SRC_PATH = "${toolchain}/lib/rustlib/src/rust/library"; + + # Point Kani's proxy at the hermetic, pre-staged bundle so it + # never runs `cargo kani setup`. The proxy resolves the install + # as `${KANI_HOME}/kani-` and execs the bundled driver. + KANI_HOME = "${kaniHome}"; }; packages.default = package;