use serde::{Deserialize, Serialize};
use std::collections::HashMap;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Workload {
#[serde(default)]
pub description: Option<String>,
#[serde(default)]
pub scenarios: HashMap<String, Vec<ScenarioStep>>,
#[serde(default)]
pub ops: Vec<ParsedOp>,
#[serde(default)]
pub bindings: BindingsDef,
#[serde(default)]
pub params: HashMap<String, String>,
#[serde(default)]
pub phases: HashMap<String, WorkloadPhase>,
#[serde(default)]
pub phase_order: Vec<String>,
#[serde(default)]
pub stop_when: Vec<StopConditionSpec>,
#[serde(default)]
pub declared_params: Vec<String>,
#[serde(default)]
pub report: crate::report::Report,
#[serde(default, skip_serializing)]
pub report_warnings: Vec<String>,
#[serde(default, skip_serializing)]
pub resolution_warnings: Vec<String>,
#[serde(default, skip_serializing)]
pub scenario_parse_errors: Vec<String>,
#[serde(default)]
pub status_metrics: Vec<String>,
#[serde(default)]
pub readouts: ReadoutsBindings,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub wrappers: Option<WrappersConfig>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub implements: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub stick_session: Option<bool>,
}
impl Workload {
pub fn synthesize_default_phase(&mut self) {
if !self.phases.is_empty() {
return;
}
if self.ops.is_empty() {
return;
}
const SYNTHETIC: &str = "main";
let cycles = self.params.get("cycles").map(|c| format!("==ops:{c}"));
let concurrency = self.params.get("concurrency").cloned();
let rate = self.params.get("rate").cloned();
let bindings = match &self.bindings {
BindingsDef::PolydatSource(_) => std::mem::take(&mut self.bindings),
BindingsDef::Map(_) => BindingsDef::default(),
};
let phase = WorkloadPhase {
ops: self.ops.clone(),
bindings,
cycles,
concurrency,
rate,
..Default::default()
};
self.phases.insert(SYNTHETIC.to_string(), phase);
self.phase_order.push(SYNTHETIC.to_string());
if self.scenarios.is_empty() {
self.scenarios.insert(
"default".to_string(),
vec![ScenarioStep::Phase(SYNTHETIC.to_string())],
);
}
}
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct WrappersConfig {
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub order: Vec<String>,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct ReadoutsBindings {
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub on_session_start: Vec<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub on_session_end: Vec<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub on_phase_start: Vec<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub on_phase_end: Vec<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub on_each_start: Vec<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub on_each_end: Vec<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub on_scope_start: Vec<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub on_scope_end: Vec<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub on_update: Vec<String>,
}
impl ReadoutsBindings {
pub fn is_empty(&self) -> bool {
self.on_session_start.is_empty()
&& self.on_session_end.is_empty()
&& self.on_phase_start.is_empty()
&& self.on_phase_end.is_empty()
&& self.on_each_start.is_empty()
&& self.on_each_end.is_empty()
&& self.on_scope_start.is_empty()
&& self.on_scope_end.is_empty()
&& self.on_update.is_empty()
}
pub fn get(&self, slot_name: &str) -> &[String] {
match slot_name {
"on_session_start" => &self.on_session_start,
"on_session_end" => &self.on_session_end,
"on_phase_start" => &self.on_phase_start,
"on_phase_end" => &self.on_phase_end,
"on_each_start" => &self.on_each_start,
"on_each_end" => &self.on_each_end,
"on_scope_start" => &self.on_scope_start,
"on_scope_end" => &self.on_scope_end,
"on_update" => &self.on_update,
_ => &[],
}
}
}
#[derive(Debug, Clone)]
pub struct SummaryConfig {
pub columns: Vec<String>,
pub row_filters: Vec<String>,
pub aggregates: Vec<AggregateExpr>,
pub show_details: bool,
pub raw: String,
pub metricsql_columns: Vec<(String, String)>,
pub group_by: Vec<String>,
pub state_query: Option<String>,
pub header_notes: Vec<(String, String)>,
}
#[derive(Debug, Clone)]
pub struct AggregateExpr {
pub function: AggFunction,
pub column_pattern: String,
pub label_key: String,
pub label_pattern: String,
pub group_by: Vec<String>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AggFunction {
Mean,
Min,
Max,
}
impl std::fmt::Display for AggFunction {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
AggFunction::Mean => write!(f, "mean"),
AggFunction::Min => write!(f, "min"),
AggFunction::Max => write!(f, "max"),
}
}
}
impl Serialize for SummaryConfig {
fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
serializer.serialize_str(&self.raw)
}
}
impl<'de> Deserialize<'de> for SummaryConfig {
fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
let raw = String::deserialize(deserializer)?;
Ok(SummaryConfig::parse(&raw))
}
}
impl SummaryConfig {
pub fn parse(raw: &str) -> Self {
let mut columns = Vec::new();
let mut row_filters = Vec::new();
let mut aggregates = Vec::new();
let mut show_details = true;
let mut metricsql_columns: Vec<(String, String)> = Vec::new();
let mut group_by: Vec<String> = Vec::new();
let mut state_query: Option<String> = None;
let mut header_notes: Vec<(String, String)> = Vec::new();
let cleaned = strip_hash_line_comments(raw);
let mut residual_lines: Vec<String> = Vec::new();
for line in cleaned.lines().map(str::trim).filter(|s| !s.is_empty()) {
if let Some(rest) = line
.strip_prefix("group_by:")
.map(str::trim)
.or_else(|| line.strip_prefix("group-by:").map(str::trim))
{
group_by = rest
.split(',')
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
.collect();
continue;
}
if let Some(rest) = line.strip_prefix("header ") {
if let Some((col, note)) = rest.split_once(':') {
let (col, note) = (col.trim(), note.trim());
if !col.is_empty() && !note.is_empty() {
header_notes.push((col.to_string(), note.to_string()));
}
}
continue;
}
if let Some(rest) = line.strip_prefix("state:") {
let expr = rest.trim();
if !expr.is_empty() {
state_query = Some(expr.to_string());
}
continue;
}
if let Some(rest) = line.strip_prefix("query") {
let rest = rest.trim_start();
if let Some(after_colon) = rest.strip_prefix(':') {
let after_colon = after_colon.trim_start();
if let Some((col, expr)) = split_named_query(after_colon) {
metricsql_columns.push((col, expr));
} else {
metricsql_columns
.push(("value".to_string(), after_colon.trim().to_string()));
}
continue;
}
if let Some(colon_idx) = rest.find(':') {
let col = rest[..colon_idx].trim().to_string();
let expr = rest[colon_idx + 1..].trim().to_string();
if !col.is_empty() && !expr.is_empty() {
metricsql_columns.push((col, expr));
continue;
}
}
}
residual_lines.push(line.to_string());
}
let cleaned: String = residual_lines.join(";");
for directive in cleaned.split(';').map(str::trim).filter(|s| !s.is_empty()) {
if directive == "details=hide" {
show_details = false;
} else if let Some(filter) = directive.strip_prefix("filter=") {
row_filters.push(filter.trim().to_string());
} else if let Some(agg) = Self::parse_aggregate(directive) {
aggregates.push(agg);
} else {
for col in directive
.split(',')
.map(str::trim)
.filter(|s| !s.is_empty())
{
if col != "all" && col != "*" {
columns.push(col.to_string());
}
}
}
}
SummaryConfig {
columns,
row_filters,
aggregates,
show_details,
raw: raw.to_string(),
metricsql_columns,
group_by,
state_query,
header_notes,
}
}
fn parse_aggregate(s: &str) -> Option<AggregateExpr> {
let paren_open = s.find('(')?;
let paren_close = s.find(')')?;
if paren_close <= paren_open {
return None;
}
let func_name = s[..paren_open].trim();
let function = match func_name {
"mean" => AggFunction::Mean,
"min" => AggFunction::Min,
"max" => AggFunction::Max,
_ => return None,
};
let column_pattern = s[paren_open + 1..paren_close].trim().to_string();
let after_paren = s[paren_close + 1..].trim();
let over_rest = after_paren.strip_prefix("over")?.trim();
if let Some(tilde) = over_rest.find('~') {
let label_key = over_rest[..tilde].trim().to_string();
let label_pattern = over_rest[tilde + 1..].trim().to_string();
return Some(AggregateExpr {
function,
column_pattern,
label_key,
label_pattern,
group_by: Vec::new(),
});
}
let group_by: Vec<String> = over_rest
.split(',')
.map(str::trim)
.filter(|s| !s.is_empty())
.map(|s| s.to_string())
.collect();
if group_by.is_empty() {
return None;
}
Some(AggregateExpr {
function,
column_pattern,
label_key: String::new(),
label_pattern: String::new(),
group_by,
})
}
}
fn split_named_query(text: &str) -> Option<(String, String)> {
let bytes = text.as_bytes();
let mut i = 0;
while i < bytes.len() {
let b = bytes[i];
let is_first = i == 0;
let ok = if is_first {
b.is_ascii_alphabetic() || b == b'_'
} else {
b.is_ascii_alphanumeric() || b == b'_' || b == b'-'
};
if !ok {
break;
}
i += 1;
}
if i == 0 {
return None;
}
let mut j = i;
while j < bytes.len() && bytes[j].is_ascii_whitespace() {
j += 1;
}
if j >= bytes.len() || bytes[j] != b':' {
return None;
}
let name = text[..i].to_string();
let expr = text[j + 1..].trim().to_string();
if name.is_empty() || expr.is_empty() {
return None;
}
Some((name, expr))
}
fn strip_hash_line_comments(s: &str) -> String {
let mut out = String::with_capacity(s.len());
for line in s.split_inclusive('\n') {
let mut quote: Option<char> = None;
let mut prev_ws = true;
let mut cut: Option<usize> = None;
for (i, ch) in line.char_indices() {
match quote {
Some(q) if ch == q => {
quote = None;
prev_ws = false;
}
Some(_) => {
prev_ws = false;
}
None => match ch {
'"' | '\'' => {
quote = Some(ch);
prev_ws = false;
}
'#' if prev_ws => {
cut = Some(i);
break;
}
c if c.is_whitespace() => {
prev_ws = true;
}
_ => {
prev_ws = false;
}
},
}
}
match cut {
Some(idx) => {
out.push_str(&line[..idx]);
if line.ends_with('\n') {
out.push('\n');
}
}
None => out.push_str(line),
}
}
out
}
#[cfg(test)]
mod summary_config_tests {
use super::*;
#[test]
fn parses_multi_key_grouping() {
let cfg = SummaryConfig::parse("recall; mean(recall) over k,limit,optimize_for");
assert_eq!(cfg.aggregates.len(), 1, "got: {:?}", cfg.aggregates);
let agg = &cfg.aggregates[0];
assert_eq!(agg.group_by, vec!["k", "limit", "optimize_for"]);
assert!(agg.label_key.is_empty());
}
#[test]
fn parses_single_key_filter_form_unchanged() {
let cfg = SummaryConfig::parse("mean(recall) over profile~label");
assert_eq!(cfg.aggregates.len(), 1);
let agg = &cfg.aggregates[0];
assert!(agg.group_by.is_empty());
assert_eq!(agg.label_key, "profile");
assert_eq!(agg.label_pattern, "label");
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ScopeLevel {
#[serde(rename = "self")]
SelfScope,
Op,
Phase,
Scenario,
Workload,
}
fn default_each() -> Vec<ScopeLevel> {
vec![ScopeLevel::SelfScope]
}
fn de_each<'de, D>(d: D) -> Result<Vec<ScopeLevel>, D::Error>
where
D: serde::Deserializer<'de>,
{
#[derive(Deserialize)]
#[serde(untagged)]
enum OneOrMany {
One(ScopeLevel),
Many(Vec<ScopeLevel>),
}
Ok(match OneOrMany::deserialize(d)? {
OneOrMany::One(level) => vec![level],
OneOrMany::Many(levels) => levels,
})
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PulseSpec {
Continuous,
PhaseEnd,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct StopConditionSpec {
#[serde(alias = "condition")]
pub when: String,
#[serde(default = "default_each", deserialize_with = "de_each", alias = "per")]
pub each: Vec<ScopeLevel>,
#[serde(default)]
pub trigger: Option<String>,
#[serde(default)]
pub pulse: Option<PulseSpec>,
#[serde(default, alias = "action")]
pub effect: Option<String>,
#[serde(default)]
pub at: Option<ScopeLevel>,
}
impl StopConditionSpec {
pub const EFFECT_VOCABULARY: [&'static str; 3] = ["stop", "fail", "abort"];
pub fn validate(&self) -> Result<(), String> {
match self.effect.as_deref() {
Some(e) if !Self::EFFECT_VOCABULARY.contains(&e) => Err(format!(
"unknown stop-condition effect '{e}' on `when: {}` — \
expected one of stop|fail|abort",
self.when
)),
_ => Ok(()),
}
}
}
fn default_continue_if_each() -> Vec<ScopeLevel> {
vec![ScopeLevel::Scenario]
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct ContinueIfSpec {
pub when: String,
#[serde(
default = "default_continue_if_each",
deserialize_with = "de_each",
alias = "per"
)]
pub each: Vec<ScopeLevel>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
pub struct BackoffSpec {
#[serde(default)]
pub ratio: Option<f64>,
#[serde(default)]
pub min: Option<String>,
#[serde(default)]
pub max: Option<String>,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq)]
#[serde(untagged)]
pub enum ThrottleField {
Enabled(bool),
Spec(ThrottleSpec),
}
impl ThrottleField {
pub fn to_spec(&self) -> Option<ThrottleSpec> {
match self {
ThrottleField::Enabled(true) => Some(ThrottleSpec::default()),
ThrottleField::Enabled(false) => None,
ThrottleField::Spec(s) => Some(s.clone()),
}
}
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq)]
#[serde(deny_unknown_fields)]
pub struct ThrottleSpec {
#[serde(default = "throttle_default_high")]
pub high: f64,
#[serde(default)]
pub low: Option<f64>,
#[serde(default = "throttle_default_control")]
pub control: String,
#[serde(default)]
pub start: Option<f64>,
#[serde(default = "throttle_default_floor")]
pub floor: f64,
#[serde(default = "throttle_default_window")]
pub window: String,
}
fn throttle_default_high() -> f64 {
0.05
}
fn throttle_default_control() -> String {
"concurrency".to_string()
}
fn throttle_default_floor() -> f64 {
1.0
}
fn throttle_default_window() -> String {
"2s".to_string()
}
impl Default for ThrottleSpec {
fn default() -> Self {
Self {
high: throttle_default_high(),
low: None,
control: throttle_default_control(),
start: None,
floor: throttle_default_floor(),
window: throttle_default_window(),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum KeyAgg {
Min,
Max,
Avg,
Last,
First,
Median,
Stddev,
Sum,
Count,
Rate,
Span,
Delta,
}
impl KeyAgg {
pub const VOCAB: &'static str = "min, max, avg, last, first, median, stddev, sum, count; \
derived: rate(F), span(), delta(F)";
pub fn parse(name: &str) -> Option<Self> {
Some(match name {
"min" => Self::Min,
"max" => Self::Max,
"avg" => Self::Avg,
"last" => Self::Last,
"first" => Self::First,
"median" => Self::Median,
"stddev" => Self::Stddev,
"sum" => Self::Sum,
"count" => Self::Count,
"rate" => Self::Rate,
"span" => Self::Span,
"delta" => Self::Delta,
_ => return None,
})
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct KeyMetric {
pub column: String,
pub agg: KeyAgg,
pub family: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct WorkloadPhase {
#[serde(default)]
pub cycles: Option<String>,
#[serde(default)]
pub concurrency: Option<String>,
#[serde(default)]
pub rate: Option<String>,
#[serde(default)]
pub daemon: bool,
#[serde(default)]
pub adapter: Option<String>,
#[serde(default)]
pub errors: Option<String>,
#[serde(default)]
pub tries: Option<u32>,
#[serde(default)]
pub tries_backoff: Option<BackoffSpec>,
#[serde(default)]
pub interval: Option<String>,
#[serde(default)]
pub repeat: Option<u64>,
#[serde(default)]
pub error_rate_max: Option<f64>,
#[serde(default)]
pub timeout: Option<String>,
#[serde(default)]
pub stop_when: Vec<StopConditionSpec>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub throttle: Option<ThrottleField>,
#[serde(default)]
pub tags: Option<String>,
#[serde(default)]
pub ops: Vec<ParsedOp>,
#[serde(default)]
pub for_each: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub continue_if: Option<ContinueIfSpec>,
#[serde(default)]
pub loop_scope: Option<String>,
#[serde(default)]
pub iter_scope: Option<String>,
#[serde(default)]
pub checkpoint: Option<Checkpoint>,
#[serde(default)]
pub status_metrics: Vec<String>,
#[serde(default, skip_serializing_if = "BindingsDef::is_empty")]
pub bindings: BindingsDef,
#[serde(default, skip_serializing_if = "HashMap::is_empty")]
pub metrics: HashMap<String, MetricSpec>,
#[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
pub dimensions: std::collections::BTreeMap<String, DimensionSpec>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub poll: Option<PhasePollSpec>,
#[serde(
default,
deserialize_with = "de_optimize",
skip_serializing_if = "Option::is_none"
)]
pub optimize: Option<OptimizeBlock>,
#[serde(default)]
pub key_metrics: Vec<KeyMetric>,
}
#[derive(Deserialize)]
#[serde(untagged)]
enum OptimizeSpec {
Inline(String),
Block(OptimizeBlock),
}
impl From<OptimizeSpec> for OptimizeBlock {
fn from(spec: OptimizeSpec) -> Self {
match spec {
OptimizeSpec::Inline(objective) => OptimizeBlock {
method: default_optimize_method(),
objective,
servo: Vec::new(),
max_evals: default_optimize_max_evals(),
seed: default_optimize_seed(),
params: HashMap::new(),
},
OptimizeSpec::Block(b) => b,
}
}
}
impl OptimizeBlock {
pub fn from_yaml_value(v: &serde_json::Value) -> Result<OptimizeBlock, serde_json::Error> {
if let Some(s) = v.as_str() {
Ok(OptimizeSpec::Inline(s.to_string()).into())
} else {
serde_json::from_value::<OptimizeBlock>(v.clone())
}
}
}
fn de_optimize<'de, D>(d: D) -> Result<Option<OptimizeBlock>, D::Error>
where
D: serde::Deserializer<'de>,
{
Ok(Option::<OptimizeSpec>::deserialize(d)?.map(Into::into))
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OptimizeBlock {
#[serde(default = "default_optimize_method")]
pub method: String,
pub objective: String,
#[serde(default, deserialize_with = "de_string_or_seq")]
pub servo: Vec<String>,
#[serde(default = "default_optimize_max_evals")]
pub max_evals: usize,
#[serde(default = "default_optimize_seed")]
pub seed: u64,
#[serde(default)]
pub params: HashMap<String, f64>,
}
fn de_string_or_seq<'de, D>(d: D) -> Result<Vec<String>, D::Error>
where
D: serde::Deserializer<'de>,
{
#[derive(Deserialize)]
#[serde(untagged)]
enum OneOrMany {
One(String),
Many(Vec<String>),
}
Ok(match OneOrMany::deserialize(d)? {
OneOrMany::One(s) => vec![s],
OneOrMany::Many(v) => v,
})
}
fn default_optimize_method() -> String {
"sweep".to_string()
}
fn default_optimize_max_evals() -> usize {
100
}
fn default_optimize_seed() -> u64 {
1
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct PhasePollSpec {
pub until: String,
#[serde(default)]
pub interval_ms: Option<String>,
#[serde(default)]
pub timeout_ms: Option<String>,
#[serde(default)]
pub max_error_retries: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub metric_name: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub on_timeout: Option<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub require: Vec<String>,
}
#[derive(Debug, Clone, Serialize, PartialEq)]
pub struct Checkpoint {
pub idempotent: bool,
pub hashed: bool,
pub verify: Option<serde_json::Value>,
}
impl Default for Checkpoint {
fn default() -> Self {
Self {
idempotent: true,
hashed: true,
verify: None,
}
}
}
impl<'de> serde::Deserialize<'de> for Checkpoint {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
struct CheckpointVisitor;
impl<'de> serde::de::Visitor<'de> for CheckpointVisitor {
type Value = Checkpoint;
fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
f.write_str("checkpoint declaration: short string ('idempotent' / 'none' / etc), bool, or mapping with sub-properties")
}
fn visit_str<E: serde::de::Error>(self, s: &str) -> Result<Checkpoint, E> {
let trimmed = s.trim().to_ascii_lowercase();
match trimmed.as_str() {
"idempotent" => Ok(Checkpoint::default()),
"none" | "no" | "false" | "off" | "" => Ok(Checkpoint {
idempotent: false,
hashed: true,
verify: None,
}),
other => Err(E::custom(format!(
"checkpoint: unknown short form '{other}'; \
expected 'idempotent', 'none', 'no', 'false', or a mapping"
))),
}
}
fn visit_string<E: serde::de::Error>(self, s: String) -> Result<Checkpoint, E> {
self.visit_str(&s)
}
fn visit_bool<E: serde::de::Error>(self, b: bool) -> Result<Checkpoint, E> {
if b {
Ok(Checkpoint::default())
} else {
Ok(Checkpoint {
idempotent: false,
hashed: true,
verify: None,
})
}
}
fn visit_unit<E: serde::de::Error>(self) -> Result<Checkpoint, E> {
Ok(Checkpoint {
idempotent: false,
hashed: true,
verify: None,
})
}
fn visit_map<M>(self, mut map: M) -> Result<Checkpoint, M::Error>
where
M: serde::de::MapAccess<'de>,
{
let mut idempotent = true;
let mut hashed = true;
let mut verify: Option<serde_json::Value> = None;
while let Some(key) = map.next_key::<String>()? {
match key.as_str() {
"idempotent" => idempotent = map.next_value::<bool>()?,
"hashed" => hashed = map.next_value::<bool>()?,
"verify" => verify = Some(map.next_value::<serde_json::Value>()?),
other => {
return Err(serde::de::Error::custom(format!(
"checkpoint: unknown key '{other}'; \
expected 'idempotent', 'hashed', or 'verify'"
)));
}
}
}
Ok(Checkpoint {
idempotent,
hashed,
verify,
})
}
}
deserializer.deserialize_any(CheckpointVisitor)
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ScenarioNode {
Phase(String),
Comprehension {
comprehension: polydat::iteration::comprehension::Comprehension,
children: Vec<ScenarioNode>,
#[serde(default, skip_serializing_if = "Option::is_none")]
continue_if: Option<ContinueIfSpec>,
#[serde(default, skip_serializing_if = "Option::is_none")]
anchor: Option<String>,
},
DoWhile {
condition: String,
counter: Option<String>,
children: Vec<ScenarioNode>,
},
DoUntil {
condition: String,
counter: Option<String>,
children: Vec<ScenarioNode>,
},
IncludedScenario {
name: String,
children: Vec<ScenarioNode>,
},
Bindings {
source: String,
children: Vec<ScenarioNode>,
},
}
pub type ScenarioStep = ScenarioNode;
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(untagged)]
pub enum BindingsDef {
Map(HashMap<String, String>),
PolydatSource(String),
}
impl Default for BindingsDef {
fn default() -> Self {
BindingsDef::Map(HashMap::new())
}
}
impl BindingsDef {
pub fn is_empty(&self) -> bool {
match self {
BindingsDef::Map(m) => m.is_empty(),
BindingsDef::PolydatSource(s) => s.trim().is_empty(),
}
}
pub fn as_map(&self) -> &HashMap<String, String> {
static EMPTY: std::sync::LazyLock<HashMap<String, String>> =
std::sync::LazyLock::new(HashMap::new);
match self {
BindingsDef::Map(m) => m,
BindingsDef::PolydatSource(_) => &EMPTY,
}
}
pub fn insert(&mut self, key: String, value: String) {
match self {
BindingsDef::Map(m) => {
m.insert(key, value);
}
_ => {
let mut m = HashMap::new();
m.insert(key, value);
*self = BindingsDef::Map(m);
}
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq, Eq)]
pub struct OpInterface {
#[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
pub needs: std::collections::BTreeMap<String, String>,
#[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
pub yields: std::collections::BTreeMap<String, String>,
#[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
pub results: std::collections::BTreeMap<String, String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ParsedOp {
pub name: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub description: Option<String>,
pub op: HashMap<String, serde_json::Value>,
#[serde(default, skip_serializing_if = "BindingsDef::is_empty")]
pub bindings: BindingsDef,
#[serde(default, skip_serializing_if = "HashMap::is_empty")]
pub params: HashMap<String, serde_json::Value>,
#[serde(default)]
pub tags: HashMap<String, String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub condition: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub delay: Option<DelaySpec>,
#[serde(default, skip_serializing_if = "HashMap::is_empty")]
pub metrics: HashMap<String, MetricSpec>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub result: Option<ResultSpec>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub traverse: Option<TraverseSpec>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub wrappers: Option<WrappersConfig>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub captures: Vec<crate::bindpoints::CapturePoint>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub abstract_interface: Option<OpInterface>,
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
pub interface_bound: bool,
#[serde(default, skip_serializing_if = "DaemonSpec::is_disabled")]
pub daemon: DaemonSpec,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub daemon_cancel_grace_ms: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub while_cond: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub rate: Option<String>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum DaemonSpec {
#[default]
Disabled,
MaxFibers(u32),
}
impl DaemonSpec {
pub fn is_disabled(&self) -> bool {
matches!(self, DaemonSpec::Disabled)
}
pub fn max_fibers(&self) -> Option<u32> {
match self {
DaemonSpec::Disabled => None,
DaemonSpec::MaxFibers(n) => Some(*n),
}
}
}
impl serde::Serialize for DaemonSpec {
fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
match self {
DaemonSpec::Disabled => s.serialize_bool(false),
DaemonSpec::MaxFibers(1) => s.serialize_bool(true),
DaemonSpec::MaxFibers(n) => s.serialize_u32(*n),
}
}
}
impl<'de> serde::Deserialize<'de> for DaemonSpec {
fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
let v = serde_json::Value::deserialize(d)?;
parse_daemon_spec_value(&v).map_err(serde::de::Error::custom)
}
}
pub fn parse_daemon_spec_value(v: &serde_json::Value) -> Result<DaemonSpec, String> {
match v {
serde_json::Value::Bool(true) => Ok(DaemonSpec::MaxFibers(1)),
serde_json::Value::Bool(false) => Ok(DaemonSpec::Disabled),
serde_json::Value::Number(n) => {
if let Some(u) = n.as_u64() {
if u == 0 {
Ok(DaemonSpec::Disabled)
} else if u <= u32::MAX as u64 {
Ok(DaemonSpec::MaxFibers(u as u32))
} else {
Err(format!(
"daemon: {u} exceeds u32::MAX — caps above {} \
aren't supported (and don't make practical sense)",
u32::MAX,
))
}
} else if n.as_i64().is_some_and(|i| i < 0) {
Err(format!(
"daemon: {n} — negative integers are invalid. \
Use 0 / false / \"off\" to disable, or a positive \
integer for the max-fibers cap.",
))
} else {
Err(format!(
"daemon: {n} — non-integer numbers are invalid. \
Use a boolean or a non-negative integer.",
))
}
}
serde_json::Value::String(s) => match s.trim().to_ascii_lowercase().as_str() {
"true" | "on" => Ok(DaemonSpec::MaxFibers(1)),
"false" | "off" => Ok(DaemonSpec::Disabled),
other => Err(format!(
"daemon: \"{other}\" — unknown string form. \
Accepted: \"on\" / \"off\" / \"true\" / \"false\", \
or use a boolean / non-negative integer directly.",
)),
},
serde_json::Value::Null => {
Err("daemon: null is not a valid value. Use false to disable.".into())
}
other => Err(format!(
"daemon: {other:?} — only boolean, integer, or string forms \
are accepted.",
)),
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DelaySpec {
Before(String),
BeforeAfter {
before: Option<String>,
after: Option<String>,
},
}
impl DelaySpec {
pub fn before(&self) -> Option<&str> {
match self {
DelaySpec::Before(n) => Some(n.as_str()),
DelaySpec::BeforeAfter { before, .. } => before.as_deref(),
}
}
pub fn after(&self) -> Option<&str> {
match self {
DelaySpec::Before(_) => None,
DelaySpec::BeforeAfter { after, .. } => after.as_deref(),
}
}
pub fn names(&self) -> Vec<&str> {
match self {
DelaySpec::Before(n) => vec![n.as_str()],
DelaySpec::BeforeAfter { before, after } => {
let mut out = Vec::with_capacity(2);
if let Some(b) = before.as_deref() {
out.push(b);
}
if let Some(a) = after.as_deref() {
out.push(a);
}
out
}
}
}
}
impl serde::Serialize for DelaySpec {
fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
use serde::ser::SerializeMap;
match self {
DelaySpec::Before(name) => s.serialize_str(name),
DelaySpec::BeforeAfter { before, after } => {
let mut m = s.serialize_map(None)?;
if let Some(b) = before {
m.serialize_entry("before", b)?;
}
if let Some(a) = after {
m.serialize_entry("after", a)?;
}
m.end()
}
}
}
}
impl<'de> serde::Deserialize<'de> for DelaySpec {
fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
let v = serde_json::Value::deserialize(d)?;
parse_delay_spec_value(&v).map_err(serde::de::Error::custom)
}
}
pub fn parse_delay_spec_value(v: &serde_json::Value) -> Result<DelaySpec, String> {
match v {
serde_json::Value::String(s) => {
let trimmed = s.trim();
if trimmed.is_empty() {
Err("delay: empty string is not a valid binding name".into())
} else {
Ok(DelaySpec::Before(trimmed.to_string()))
}
}
serde_json::Value::Object(map) => {
let mut before: Option<String> = None;
let mut after: Option<String> = None;
for (k, v) in map {
match k.as_str() {
"before" => {
before = match v {
serde_json::Value::String(s) => {
let t = s.trim();
if t.is_empty() {
return Err(
"delay.before: empty string is not a valid binding name"
.into(),
);
}
Some(t.to_string())
}
other => {
return Err(format!(
"delay.before: expected string, got {other:?}",
));
}
};
}
"after" => {
after = match v {
serde_json::Value::String(s) => {
let t = s.trim();
if t.is_empty() {
return Err(
"delay.after: empty string is not a valid binding name"
.into(),
);
}
Some(t.to_string())
}
other => {
return Err(
format!("delay.after: expected string, got {other:?}",),
);
}
};
}
other => {
return Err(format!(
"delay: unknown key `{other}` — accepted: `before`, `after`",
));
}
}
}
if before.is_none() && after.is_none() {
Err("delay: map form must set at least one of `before` / `after`".into())
} else {
Ok(DelaySpec::BeforeAfter { before, after })
}
}
serde_json::Value::Null => {
Err("delay: null is not a valid value. Omit the field instead.".into())
}
other => Err(format!(
"delay: {other:?} — accepted forms are a binding-name string or \
a map `{{ before: <name>, after: <name> }}`",
)),
}
}
#[cfg(test)]
mod delay_spec_tests {
use super::*;
use serde_json::json;
#[test]
fn parse_bare_string_is_before() {
let spec = parse_delay_spec_value(&json!("ticks")).unwrap();
assert_eq!(spec, DelaySpec::Before("ticks".into()));
}
#[test]
fn parse_trimmed_string() {
let spec = parse_delay_spec_value(&json!(" ticks ")).unwrap();
assert_eq!(spec, DelaySpec::Before("ticks".into()));
}
#[test]
fn parse_map_with_both() {
let spec = parse_delay_spec_value(&json!({
"before": "pre", "after": "post"
}))
.unwrap();
assert_eq!(
spec,
DelaySpec::BeforeAfter {
before: Some("pre".into()),
after: Some("post".into()),
}
);
}
#[test]
fn parse_map_before_only() {
let spec = parse_delay_spec_value(&json!({ "before": "pre" })).unwrap();
assert_eq!(
spec,
DelaySpec::BeforeAfter {
before: Some("pre".into()),
after: None,
}
);
}
#[test]
fn parse_map_after_only() {
let spec = parse_delay_spec_value(&json!({ "after": "post" })).unwrap();
assert_eq!(
spec,
DelaySpec::BeforeAfter {
before: None,
after: Some("post".into()),
}
);
}
#[test]
fn parse_rejects_empty_string() {
assert!(parse_delay_spec_value(&json!("")).is_err());
assert!(parse_delay_spec_value(&json!(" ")).is_err());
}
#[test]
fn parse_rejects_empty_map() {
assert!(parse_delay_spec_value(&json!({})).is_err());
}
#[test]
fn parse_rejects_unknown_key() {
let e = parse_delay_spec_value(&json!({
"before": "pre", "during": "mid"
}))
.unwrap_err();
assert!(e.contains("during"));
}
#[test]
fn parse_rejects_non_string_value() {
assert!(parse_delay_spec_value(&json!({ "before": 5 })).is_err());
assert!(parse_delay_spec_value(&json!({ "after": true })).is_err());
assert!(parse_delay_spec_value(&json!({ "before": null })).is_err());
}
#[test]
fn parse_rejects_null_top_level() {
assert!(parse_delay_spec_value(&json!(null)).is_err());
}
#[test]
fn parse_rejects_array() {
assert!(parse_delay_spec_value(&json!(["pre"])).is_err());
}
#[test]
fn parse_rejects_number() {
assert!(parse_delay_spec_value(&json!(1.5)).is_err());
assert!(parse_delay_spec_value(&json!(100)).is_err());
}
#[test]
fn parse_rejects_empty_after_value() {
assert!(parse_delay_spec_value(&json!({ "after": "" })).is_err());
}
#[test]
fn round_trip_before_serializes_as_string() {
let spec = DelaySpec::Before("ticks".into());
let v = serde_json::to_value(&spec).unwrap();
assert_eq!(v, json!("ticks"));
let parsed: DelaySpec = serde_json::from_value(v).unwrap();
assert_eq!(parsed, spec);
}
#[test]
fn round_trip_map_serializes_as_object() {
let spec = DelaySpec::BeforeAfter {
before: Some("pre".into()),
after: Some("post".into()),
};
let v = serde_json::to_value(&spec).unwrap();
assert_eq!(v.get("before"), Some(&json!("pre")));
assert_eq!(v.get("after"), Some(&json!("post")));
let parsed: DelaySpec = serde_json::from_value(v).unwrap();
assert_eq!(parsed, spec);
}
#[test]
fn names_returns_all_referenced() {
assert_eq!(DelaySpec::Before("x".into()).names(), vec!["x"]);
let spec = DelaySpec::BeforeAfter {
before: Some("a".into()),
after: Some("b".into()),
};
assert_eq!(spec.names(), vec!["a", "b"]);
let only_before = DelaySpec::BeforeAfter {
before: Some("a".into()),
after: None,
};
assert_eq!(only_before.names(), vec!["a"]);
}
#[test]
fn accessors_return_correct_names() {
let s = DelaySpec::Before("x".into());
assert_eq!(s.before(), Some("x"));
assert_eq!(s.after(), None);
let s = DelaySpec::BeforeAfter {
before: Some("a".into()),
after: Some("b".into()),
};
assert_eq!(s.before(), Some("a"));
assert_eq!(s.after(), Some("b"));
}
}
#[cfg(test)]
mod daemon_spec_tests {
use super::*;
use serde_json::json;
#[test]
fn bool_true_is_max_1() {
assert_eq!(
parse_daemon_spec_value(&json!(true)).unwrap(),
DaemonSpec::MaxFibers(1)
);
}
#[test]
fn bool_false_is_disabled() {
assert_eq!(
parse_daemon_spec_value(&json!(false)).unwrap(),
DaemonSpec::Disabled
);
}
#[test]
fn int_0_is_disabled() {
assert_eq!(
parse_daemon_spec_value(&json!(0)).unwrap(),
DaemonSpec::Disabled
);
}
#[test]
fn int_1_is_max_1() {
assert_eq!(
parse_daemon_spec_value(&json!(1)).unwrap(),
DaemonSpec::MaxFibers(1)
);
}
#[test]
fn int_n_is_max_n() {
assert_eq!(
parse_daemon_spec_value(&json!(10)).unwrap(),
DaemonSpec::MaxFibers(10)
);
}
#[test]
fn str_on_is_max_1() {
assert_eq!(
parse_daemon_spec_value(&json!("on")).unwrap(),
DaemonSpec::MaxFibers(1)
);
assert_eq!(
parse_daemon_spec_value(&json!("true")).unwrap(),
DaemonSpec::MaxFibers(1)
);
assert_eq!(
parse_daemon_spec_value(&json!("ON")).unwrap(),
DaemonSpec::MaxFibers(1)
);
}
#[test]
fn str_off_is_disabled() {
assert_eq!(
parse_daemon_spec_value(&json!("off")).unwrap(),
DaemonSpec::Disabled
);
assert_eq!(
parse_daemon_spec_value(&json!("false")).unwrap(),
DaemonSpec::Disabled
);
assert_eq!(
parse_daemon_spec_value(&json!("OFF")).unwrap(),
DaemonSpec::Disabled
);
}
#[test]
fn negative_int_rejected() {
assert!(parse_daemon_spec_value(&json!(-1)).is_err());
assert!(parse_daemon_spec_value(&json!(-100)).is_err());
}
#[test]
fn float_rejected() {
assert!(parse_daemon_spec_value(&json!(1.5)).is_err());
}
#[test]
fn unknown_string_rejected() {
assert!(parse_daemon_spec_value(&json!("garbage")).is_err());
assert!(parse_daemon_spec_value(&json!("yes")).is_err());
}
#[test]
fn null_rejected() {
assert!(parse_daemon_spec_value(&json!(null)).is_err());
}
#[test]
fn array_object_rejected() {
assert!(parse_daemon_spec_value(&json!([1, 2])).is_err());
assert!(parse_daemon_spec_value(&json!({"max": 5})).is_err());
}
#[test]
fn round_trip_disabled() {
let s = serde_json::to_value(DaemonSpec::Disabled).unwrap();
assert_eq!(parse_daemon_spec_value(&s).unwrap(), DaemonSpec::Disabled);
}
#[test]
fn round_trip_max_1_serialises_as_bool() {
let s = serde_json::to_value(DaemonSpec::MaxFibers(1)).unwrap();
assert_eq!(s, json!(true));
assert_eq!(
parse_daemon_spec_value(&s).unwrap(),
DaemonSpec::MaxFibers(1)
);
}
#[test]
fn round_trip_max_n_serialises_as_int() {
let s = serde_json::to_value(DaemonSpec::MaxFibers(5)).unwrap();
assert_eq!(s, json!(5));
assert_eq!(
parse_daemon_spec_value(&s).unwrap(),
DaemonSpec::MaxFibers(5)
);
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct TraverseSpec {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub path: Option<String>,
#[serde(default)]
pub on_missing: OnMissing,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum OnMissing {
#[default]
Ignore,
Warn,
Error,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MetricSpec {
pub value: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub family: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub kind: Option<MetricKind>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub unit: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub format: Option<String>,
#[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
pub cell: std::collections::BTreeMap<String, String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DimensionSpec {
#[serde(default)]
pub value_type: DimensionType,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum DimensionType {
#[default]
Str,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum MetricKind {
Gauge,
Histogram,
Counter,
}
impl Default for MetricKind {
fn default() -> Self {
MetricKind::Gauge
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(untagged)]
pub enum ResultSpec {
String(String),
List(Vec<ResultSpec>),
Map(std::collections::BTreeMap<String, String>),
}
pub type ResultWireSpec = LegacyResultWireSpec;
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(untagged)]
pub enum LegacyResultWireSpec {
String(String),
Object {
source: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
default: Option<String>,
},
}
impl LegacyResultWireSpec {
pub fn source(&self) -> &str {
match self {
LegacyResultWireSpec::String(s) => s,
LegacyResultWireSpec::Object { source, .. } => source,
}
}
}
impl ResultSpec {
pub fn walk_fragments<F: FnMut(ResultFragment<'_>)>(&self, mut on: F) {
self.walk_fragments_inner(&mut on);
}
fn walk_fragments_inner<F: FnMut(ResultFragment<'_>)>(&self, on: &mut F) {
match self {
ResultSpec::String(s) => on(ResultFragment::Source(s)),
ResultSpec::List(items) => {
for item in items {
item.walk_fragments_inner(on);
}
}
ResultSpec::Map(entries) => {
for (name, source) in entries {
on(ResultFragment::Named { name, source });
}
}
}
}
pub fn is_empty(&self) -> bool {
match self {
ResultSpec::String(s) => s.trim().is_empty(),
ResultSpec::List(items) => items.iter().all(|i| i.is_empty()),
ResultSpec::Map(entries) => entries.is_empty(),
}
}
}
pub enum ResultFragment<'a> {
Source(&'a str),
Named { name: &'a str, source: &'a str },
}
impl ParsedOp {
pub fn simple(name: &str, stmt: &str) -> Self {
let mut op = HashMap::new();
op.insert(
"stmt".to_string(),
serde_json::Value::String(stmt.to_string()),
);
Self {
traverse: None,
name: name.to_string(),
description: None,
op,
bindings: BindingsDef::default(),
params: HashMap::new(),
tags: HashMap::new(),
condition: None,
delay: None,
metrics: HashMap::new(),
result: None,
wrappers: None,
captures: Vec::new(),
abstract_interface: None,
interface_bound: false,
daemon: DaemonSpec::Disabled,
daemon_cancel_grace_ms: None,
while_cond: None,
rate: None,
}
}
}