use std::collections::BTreeMap;
use std::fmt;
use std::str::FromStr;
use std::time::Duration;
use jiff::Timestamp;
use serde::{Deserialize, Serialize};
use crate::agent::AgentName;
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Deserialize, Serialize)]
#[serde(transparent)]
pub struct PipelineName(String);
impl PipelineName {
#[must_use]
pub fn new(name: impl Into<String>) -> Self {
Self(name.into())
}
#[must_use]
pub fn as_str(&self) -> &str {
&self.0
}
}
impl fmt::Display for PipelineName {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(&self.0)
}
}
impl From<&str> for PipelineName {
fn from(value: &str) -> Self {
Self(value.to_owned())
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Schedule {
Every(Duration),
Cron(String),
}
impl Schedule {
pub const MIN_INTERVAL_SECS: u64 = 60;
#[must_use]
pub fn interval(&self) -> Option<Duration> {
match self {
Self::Every(duration) => Some(*duration),
Self::Cron(_) => None,
}
}
#[must_use]
pub fn next_after(&self, now: Timestamp) -> Option<Timestamp> {
match self {
Self::Every(interval) => {
let step = i64::try_from(interval.as_secs()).ok()?.max(1);
now.checked_add(jiff::SignedDuration::from_secs(step)).ok()
}
Self::Cron(expression) => {
let cron = croner::Cron::from_str(expression).ok()?;
let zoned = now.to_zoned(jiff::tz::TimeZone::system());
cron.find_next_occurrence(&zoned, false)
.ok()
.map(|at| at.timestamp())
}
}
}
#[must_use]
pub fn min_gap_secs(&self) -> Option<u64> {
match self {
Self::Every(duration) => Some(duration.as_secs()),
Self::Cron(expression) => {
let minutes = expression.split_whitespace().next()?;
if minutes == "*" {
return Some(60);
}
let step: u64 = minutes.strip_prefix("*/")?.parse().ok()?;
step.checked_mul(60)
}
}
}
}
impl fmt::Display for Schedule {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Every(duration) => write!(f, "every {}s", duration.as_secs()),
Self::Cron(expression) => write!(f, "cron `{expression}`"),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Trigger {
Manual,
Scheduled(Schedule),
}
impl fmt::Display for Trigger {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Manual => f.write_str("manual"),
Self::Scheduled(schedule) => write!(f, "{schedule}"),
}
}
}
impl Trigger {
#[must_use]
pub fn schedule(&self) -> Option<&Schedule> {
match self {
Self::Manual => None,
Self::Scheduled(schedule) => Some(schedule),
}
}
#[must_use]
pub fn is_manual(&self) -> bool {
matches!(self, Self::Manual)
}
}
#[derive(Debug, Clone, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct FlagSpec {
#[serde(default)]
pub default: bool,
#[serde(default)]
pub description: Option<String>,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Deserialize, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum Workspace {
#[default]
Shared,
PerItinerary,
}
impl Workspace {
#[must_use]
pub fn is_isolated(&self) -> bool {
matches!(self, Self::PerItinerary)
}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Deserialize, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum Overlap {
#[default]
Skip,
Allow,
}
impl Overlap {
#[must_use]
pub fn allows_second_instance(&self) -> bool {
matches!(self, Self::Allow)
}
}
#[derive(Debug, Clone, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Pipeline {
#[serde(default)]
pub description: Option<String>,
pub entry: AgentName,
#[serde(default = "manual_trigger")]
pub trigger: Trigger,
#[serde(default)]
pub workspace: Workspace,
#[serde(default)]
pub overlap: Overlap,
#[serde(default)]
pub resumes: bool,
#[serde(default)]
pub flags: BTreeMap<String, FlagSpec>,
}
impl Pipeline {
#[must_use]
pub fn allows_overlap(&self) -> bool {
self.overlap.allows_second_instance()
}
pub fn flags_for_run(&self, overrides: &BTreeMap<String, bool>) -> Result<Flags, FlagError> {
if let Some(unknown) = overrides.keys().find(|key| !self.flags.contains_key(*key)) {
return Err(FlagError::Undeclared {
flag: unknown.clone(),
});
}
let values = self
.flags
.iter()
.map(|(name, spec)| {
let value = overrides.get(name).copied().unwrap_or(spec.default);
(name.clone(), value)
})
.collect();
Ok(Flags(values))
}
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Flags(BTreeMap<String, bool>);
impl Flags {
#[must_use]
pub fn new(values: BTreeMap<String, bool>) -> Self {
Self(values)
}
#[must_use]
pub fn get(&self, name: &str) -> Option<bool> {
self.0.get(name).copied()
}
pub fn iter(&self) -> impl Iterator<Item = (&str, bool)> {
self.0.iter().map(|(name, value)| (name.as_str(), *value))
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.0.is_empty()
}
}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum FlagError {
#[error("flag `{flag}` is not declared by this pipeline")]
Undeclared {
flag: String,
},
}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum TriggerError {
#[error(
"unknown trigger `{found}`; expected `manual`, `{{ every = \"1h\" }}` or `{{ cron = \"0 * * * *\" }}`"
)]
UnknownKeyword {
found: String,
},
#[error("could not read `{found}` as a duration; expected a number followed by s, m, h or d")]
BadDuration {
found: String,
},
#[error(
"`every = \"{found}\"` fires more often than once every {minimum}s, which is not allowed \
for unattended work"
)]
TooFrequent {
found: String,
minimum: u64,
},
#[error("could not read `{found}` as a cron expression: {reason}")]
BadCron {
found: String,
reason: String,
},
#[error(
"cron expression `{found}` has {fields} fields; Layover accepts five-field expressions \
only, because a seconds field can schedule work faster than a run can finish"
)]
SubMinuteCron {
found: String,
fields: usize,
},
#[error("a trigger sets both `every` and `cron`; give exactly one")]
AmbiguousSchedule,
#[error("a trigger table sets neither `every` nor `cron`")]
EmptySchedule,
}
const fn manual_trigger() -> Trigger {
Trigger::Manual
}
fn parse_duration(text: &str) -> Result<Duration, TriggerError> {
let trimmed = text.trim();
let bad = || TriggerError::BadDuration {
found: text.to_owned(),
};
let (digits, unit) = match trimmed.char_indices().next_back() {
Some((index, unit)) => (&trimmed[..index], unit),
None => return Err(bad()),
};
let multiplier = match unit {
's' => 1_u64,
'm' => 60,
'h' => 60 * 60,
'd' => 24 * 60 * 60,
_ => return Err(bad()),
};
let amount: u64 = digits.parse().map_err(|_| bad())?;
amount
.checked_mul(multiplier)
.map(Duration::from_secs)
.ok_or_else(bad)
}
fn parse_cron(expression: &str) -> Result<String, TriggerError> {
let fields = expression.split_whitespace().count();
if fields > 5 {
return Err(TriggerError::SubMinuteCron {
found: expression.to_owned(),
fields,
});
}
croner::Cron::from_str(expression).map_err(|error| TriggerError::BadCron {
found: expression.to_owned(),
reason: error.to_string(),
})?;
Ok(expression.to_owned())
}
fn trigger_from_parts(
every: Option<String>,
cron: Option<String>,
) -> Result<Trigger, TriggerError> {
match (every, cron) {
(Some(every), None) => {
let duration = parse_duration(&every)?;
if duration.as_secs() < Schedule::MIN_INTERVAL_SECS {
return Err(TriggerError::TooFrequent {
found: every,
minimum: Schedule::MIN_INTERVAL_SECS,
});
}
Ok(Trigger::Scheduled(Schedule::Every(duration)))
}
(None, Some(cron)) => Ok(Trigger::Scheduled(Schedule::Cron(parse_cron(&cron)?))),
(Some(_), Some(_)) => Err(TriggerError::AmbiguousSchedule),
(None, None) => Err(TriggerError::EmptySchedule),
}
}
impl<'de> Deserialize<'de> for Trigger {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
struct TriggerVisitor;
impl<'de> serde::de::Visitor<'de> for TriggerVisitor {
type Value = Trigger;
fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(r#""manual", { every = "1h" } or { cron = "0 * * * *" }"#)
}
fn visit_str<E>(self, value: &str) -> Result<Trigger, E>
where
E: serde::de::Error,
{
if value == "manual" {
return Ok(Trigger::Manual);
}
Err(E::custom(TriggerError::UnknownKeyword {
found: value.to_owned(),
}))
}
fn visit_map<M>(self, mut map: M) -> Result<Trigger, M::Error>
where
M: serde::de::MapAccess<'de>,
{
const FIELDS: &[&str] = &["every", "cron"];
let mut every: Option<String> = None;
let mut cron: Option<String> = None;
while let Some(key) = map.next_key::<String>()? {
match key.as_str() {
"every" if every.is_some() => {
return Err(serde::de::Error::duplicate_field("every"));
}
"every" => every = Some(map.next_value()?),
"cron" if cron.is_some() => {
return Err(serde::de::Error::duplicate_field("cron"));
}
"cron" => cron = Some(map.next_value()?),
other => return Err(serde::de::Error::unknown_field(other, FIELDS)),
}
}
trigger_from_parts(every, cron).map_err(serde::de::Error::custom)
}
}
deserializer.deserialize_any(TriggerVisitor)
}
}
#[cfg(test)]
mod tests {
use super::*;
fn pipeline(body: &str) -> Pipeline {
toml::from_str(body).expect("pipeline parses")
}
fn trigger_error(body: &str) -> String {
toml::from_str::<Pipeline>(body)
.expect_err("trigger must be rejected")
.to_string()
}
#[test]
fn a_pipeline_defaults_to_manual() {
let pipeline = pipeline(r#"entry = "analyst""#);
assert_eq!(pipeline.trigger, Trigger::Manual);
assert!(pipeline.trigger.is_manual());
assert!(pipeline.trigger.schedule().is_none());
}
#[test]
fn manual_may_be_stated_explicitly() {
let pipeline = pipeline(
r#"
entry = "analyst"
trigger = "manual"
"#,
);
assert_eq!(pipeline.trigger, Trigger::Manual);
}
#[test]
fn an_interval_schedule_is_parsed() {
let pipeline = pipeline(
r#"
entry = "pr_scanner"
trigger = { every = "1h" }
"#,
);
assert_eq!(
pipeline.trigger,
Trigger::Scheduled(Schedule::Every(Duration::from_secs(3_600)))
);
assert_eq!(
pipeline.trigger.schedule().and_then(Schedule::interval),
Some(Duration::from_secs(3_600))
);
}
#[test]
fn every_unit_is_understood() {
assert_eq!(parse_duration("90s"), Ok(Duration::from_secs(90)));
assert_eq!(parse_duration("15m"), Ok(Duration::from_mins(15)));
assert_eq!(parse_duration("2h"), Ok(Duration::from_secs(7_200)));
assert_eq!(parse_duration("1d"), Ok(Duration::from_hours(24)));
}
#[test]
fn a_malformed_duration_is_rejected() {
assert!(parse_duration("soon").is_err());
assert!(parse_duration("10").is_err());
assert!(parse_duration("").is_err());
assert!(parse_duration("1w").is_err());
assert!(parse_duration("-5m").is_err());
assert!(
parse_duration("1é").is_err(),
"must not panic on a multi-byte tail"
);
assert!(parse_duration("99999999999999999999d").is_err());
}
#[test]
fn a_trigger_setting_both_forms_is_refused() {
let message = trigger_error(
r#"
entry = "pr_scanner"
trigger = { every = "1h", cron = "0 * * * *" }
"#,
);
assert!(
message.contains("both"),
"a trigger must not silently pick one of two schedules, got: {message}"
);
}
#[test]
fn an_empty_trigger_table_is_refused() {
let message = trigger_error(
r#"
entry = "pr_scanner"
trigger = {}
"#,
);
assert!(message.contains("neither"), "got: {message}");
}
#[test]
fn an_unknown_trigger_field_is_refused_by_name() {
let message = trigger_error(
r#"
entry = "pr_scanner"
trigger = { evry = "1h" }
"#,
);
assert!(
message.contains("evry"),
"the error must name the offending field, got: {message}"
);
assert!(
message.contains("every") && message.contains("cron"),
"the error must list what was expected, got: {message}"
);
}
#[test]
fn a_duplicate_trigger_field_is_refused() {
let message = trigger_error(
r#"
entry = "pr_scanner"
trigger = { every = "1h", every = "2h" }
"#,
);
assert!(!message.is_empty(), "got: {message}");
}
#[test]
fn a_schedule_faster_than_the_floor_is_refused() {
let message = trigger_error(
r#"
entry = "pr_scanner"
trigger = { every = "30s" }
"#,
);
assert!(
message.contains("fires more often"),
"expected a frequency refusal, got: {message}"
);
}
#[test]
fn a_cron_schedule_is_validated_at_load_time() {
let pipeline = pipeline(
r#"
entry = "pr_scanner"
trigger = { cron = "0 * * * *" }
"#,
);
assert_eq!(
pipeline.trigger,
Trigger::Scheduled(Schedule::Cron("0 * * * *".to_owned()))
);
}
#[test]
fn a_nonsense_cron_expression_is_refused() {
let message = trigger_error(
r#"
entry = "pr_scanner"
trigger = { cron = "every hour please" }
"#,
);
assert!(
message.contains("cron expression"),
"expected a cron refusal, got: {message}"
);
}
#[test]
fn a_six_field_cron_expression_is_refused() {
let message = trigger_error(
r#"
entry = "pr_scanner"
trigger = { cron = "*/5 * * * * *" }
"#,
);
assert!(
message.contains("five-field"),
"expected a granularity refusal, got: {message}"
);
}
#[test]
fn an_unknown_trigger_keyword_is_refused() {
let message = trigger_error(
r#"
entry = "pr_scanner"
trigger = "whenever"
"#,
);
assert!(message.contains("unknown trigger"));
}
#[test]
fn flags_fall_back_to_their_declared_defaults() {
let pipeline = pipeline(
r#"
entry = "analyst"
[flags]
run_e2e = { default = false, description = "Run the remote suite" }
verbose = { default = true }
"#,
);
let flags = pipeline
.flags_for_run(&BTreeMap::new())
.expect("no overrides is always valid");
assert_eq!(flags.get("run_e2e"), Some(false));
assert_eq!(flags.get("verbose"), Some(true));
assert_eq!(flags.get("undeclared"), None);
assert!(!flags.is_empty());
}
#[test]
fn a_trigger_override_wins_over_the_default() {
let pipeline = pipeline(
r#"
entry = "analyst"
[flags]
run_e2e = { default = false }
"#,
);
let overrides = BTreeMap::from([("run_e2e".to_owned(), true)]);
let flags = pipeline.flags_for_run(&overrides).expect("declared flag");
assert_eq!(flags.get("run_e2e"), Some(true));
}
#[test]
fn setting_an_undeclared_flag_is_an_error() {
let pipeline = pipeline(r#"entry = "analyst""#);
let overrides = BTreeMap::from([("run_e2ee".to_owned(), true)]);
assert_eq!(
pipeline.flags_for_run(&overrides),
Err(FlagError::Undeclared {
flag: "run_e2ee".to_owned()
})
);
}
#[test]
fn flags_iterate_in_a_stable_order() {
let flags = Flags::new(BTreeMap::from([
("b".to_owned(), true),
("a".to_owned(), false),
]));
assert_eq!(
flags.iter().collect::<Vec<_>>(),
[("a", false), ("b", true)]
);
}
}