use serde::{Deserialize, Serialize};
use std::num::NonZeroUsize;
pub const MIN_DEFAULT_FILES_IN_FLIGHT: usize = 4;
#[derive(Clone, Copy, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
pub enum ConcurrencyLimit {
Unlimited,
Limited(std::num::NonZeroUsize),
}
impl ConcurrencyLimit {
#[must_use]
pub fn meet(self, other: Self) -> Self {
match (self, other) {
(Self::Unlimited, limit) | (limit, Self::Unlimited) => limit,
(Self::Limited(left), Self::Limited(right)) => {
Self::Limited(std::cmp::min(left, right))
}
}
}
}
impl std::fmt::Display for ConcurrencyLimit {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Unlimited => formatter.write_str("unlimited"),
Self::Limited(value) => value.fmt(formatter),
}
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum FilesInFlightSource {
Automatic,
Explicit,
DeprecatedMaxOpenFiles,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct ResolvedFilesInFlight {
limit: ConcurrencyLimit,
source: FilesInFlightSource,
warn_deprecated_max_open_files: bool,
}
fn default_files_in_flight_from(available: Option<NonZeroUsize>) -> NonZeroUsize {
let available = available.map_or(1, NonZeroUsize::get);
NonZeroUsize::new(available.max(MIN_DEFAULT_FILES_IN_FLIGHT))
.expect("the minimum files-in-flight value is nonzero")
}
fn default_files_in_flight() -> NonZeroUsize {
default_files_in_flight_from(std::thread::available_parallelism().ok())
}
impl ResolvedFilesInFlight {
pub fn automatic() -> Self {
Self::automatic_with(default_files_in_flight())
}
pub const fn automatic_with(value: NonZeroUsize) -> Self {
Self {
limit: ConcurrencyLimit::Limited(value),
source: FilesInFlightSource::Automatic,
warn_deprecated_max_open_files: false,
}
}
pub const fn explicit(value: NonZeroUsize) -> Self {
Self::explicit_limit(ConcurrencyLimit::Limited(value))
}
pub const fn explicit_limit(limit: ConcurrencyLimit) -> Self {
Self {
limit,
source: FilesInFlightSource::Explicit,
warn_deprecated_max_open_files: false,
}
}
pub const fn unlimited() -> Self {
Self::explicit_limit(ConcurrencyLimit::Unlimited)
}
pub fn legacy(value: usize) -> Self {
Self::legacy_with_warning(value, true)
}
pub fn forwarded_legacy(value: usize) -> Self {
Self::legacy_with_warning(value, false)
}
fn legacy_with_warning(value: usize, warn_deprecated_max_open_files: bool) -> Self {
Self {
limit: NonZeroUsize::new(value)
.map_or(ConcurrencyLimit::Unlimited, ConcurrencyLimit::Limited),
source: FilesInFlightSource::DeprecatedMaxOpenFiles,
warn_deprecated_max_open_files,
}
}
pub const fn limit(self) -> ConcurrencyLimit {
self.limit
}
pub const fn source(self) -> FilesInFlightSource {
self.source
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, clap::ValueEnum, Serialize, Deserialize)]
pub enum DryRunMode {
#[value(name = "brief")]
Brief,
#[value(name = "all")]
All,
#[value(name = "explain")]
Explain,
}
#[derive(Debug, Clone, Copy, Default)]
pub struct RuntimeConfig {
pub max_workers: usize,
pub max_blocking_threads: usize,
}
#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)]
pub struct AutoMetaThrottleConfig {
pub initial_cwnd: u32,
pub min_cwnd: u32,
pub max_cwnd: u32,
pub alpha: f64,
pub beta: f64,
pub increase_step: u32,
pub decrease_step: u32,
pub baseline_percentile: f64,
pub current_percentile: f64,
pub long_window: std::time::Duration,
pub short_window: std::time::Duration,
pub tick_interval: std::time::Duration,
}
#[derive(Debug, Clone)]
pub struct ThrottleConfig {
pub files_in_flight: ResolvedFilesInFlight,
pub apply_files_in_flight: bool,
pub ops_throttle: usize,
pub iops_throttle: usize,
pub chunk_size: u64,
pub auto_meta: Option<AutoMetaThrottleConfig>,
pub histogram_enabled: bool,
pub histogram_log_path: Option<std::path::PathBuf>,
pub histogram_interval: std::time::Duration,
}
impl Default for ThrottleConfig {
fn default() -> Self {
Self {
files_in_flight: ResolvedFilesInFlight::automatic(),
apply_files_in_flight: true,
ops_throttle: 0,
iops_throttle: 0,
chunk_size: 0,
auto_meta: None,
histogram_enabled: false,
histogram_log_path: None,
histogram_interval: std::time::Duration::from_secs(1),
}
}
}
pub const AUTO_META_MIN_OPS_THROTTLE: usize = 10;
impl ThrottleConfig {
#[must_use]
pub fn deprecated_max_open_files_warning(&self) -> Option<&'static str> {
if !self.files_in_flight.warn_deprecated_max_open_files {
return None;
}
match self.files_in_flight.limit() {
ConcurrencyLimit::Unlimited => Some(
"--max-open-files=0 is deprecated; use --max-files-in-flight=unlimited instead. It removes the user ceiling but no longer disables descriptor safety.",
),
ConcurrencyLimit::Limited(_) => {
Some("--max-open-files is deprecated; use --max-files-in-flight instead.")
}
}
}
pub fn validate(&self) -> Result<(), String> {
if self.iops_throttle > 0 && self.chunk_size == 0 {
return Err("chunk_size must be specified when using iops_throttle".to_string());
}
if let Some(auto) = &self.auto_meta {
if auto.max_cwnd == 0 {
return Err("auto-meta-max-cwnd must be > 0".to_string());
}
if auto.min_cwnd == 0 {
return Err("auto-meta-min-cwnd must be >= 1".to_string());
}
if auto.min_cwnd > auto.max_cwnd {
return Err("auto-meta-min-cwnd must be <= auto-meta-max-cwnd".to_string());
}
if !(0.0..1.0).contains(&auto.baseline_percentile) {
return Err("auto-meta-baseline-percentile must be in [0.0, 1.0)".to_string());
}
if !(0.0..1.0).contains(&auto.current_percentile) {
return Err("auto-meta-current-percentile must be in [0.0, 1.0)".to_string());
}
if auto.baseline_percentile > auto.current_percentile {
return Err(
"auto-meta-baseline-percentile must be <= auto-meta-current-percentile"
.to_string(),
);
}
if !auto.alpha.is_finite() || auto.alpha <= 0.0 {
return Err("auto-meta-alpha must be a finite value > 0".to_string());
}
if !auto.beta.is_finite() || auto.beta <= 0.0 {
return Err("auto-meta-beta must be a finite value > 0".to_string());
}
if auto.alpha >= auto.beta {
return Err("auto-meta-alpha must be < auto-meta-beta".to_string());
}
if auto.tick_interval.is_zero() {
return Err("auto-meta-tick-interval must be > 0".to_string());
}
if auto.long_window.is_zero() {
return Err("auto-meta-long-window must be > 0".to_string());
}
if auto.short_window.is_zero() {
return Err("auto-meta-short-window must be > 0".to_string());
}
if auto.short_window >= auto.long_window {
return Err("auto-meta-short-window must be < auto-meta-long-window".to_string());
}
if self.ops_throttle > 0 && self.ops_throttle < AUTO_META_MIN_OPS_THROTTLE {
return Err(format!(
"--auto-meta-throttle is incompatible with --ops-throttle={} \
(auto-meta uses a fixed 100ms replenish interval; rates below \
{} ops/sec round to zero tokens per interval and would pause \
the throttle after the initial token). Either raise ops-throttle \
to >= {} or drop --auto-meta-throttle to get the legacy adaptive \
interval.",
self.ops_throttle, AUTO_META_MIN_OPS_THROTTLE, AUTO_META_MIN_OPS_THROTTLE,
));
}
}
let histogram_active = self.histogram_enabled || self.histogram_log_path.is_some();
if histogram_active && self.auto_meta.is_none() {
return Err(
"--auto-meta-histogram and --auto-meta-histogram-log require \
--auto-meta-throttle to be enabled"
.into(),
);
}
if histogram_active {
let min = std::time::Duration::from_millis(100);
let max = std::time::Duration::from_secs(60);
if self.histogram_interval < min || self.histogram_interval > max {
return Err(format!(
"--auto-meta-histogram-interval must be in [{}ms, {}s]",
min.as_millis(),
max.as_secs(),
));
}
if let Some(path) = &self.histogram_log_path {
let parent = match path.parent() {
Some(p) if p.as_os_str().is_empty() => std::path::Path::new("."),
Some(p) => p,
None => std::path::Path::new("."),
};
if !parent.exists() {
return Err(format!(
"--auto-meta-histogram-log parent directory does not exist: {parent:?}",
));
}
if !parent.is_dir() {
return Err(format!(
"--auto-meta-histogram-log parent is not a directory: {parent:?}",
));
}
let suffix: u64 = rand::random();
let probe = parent.join(format!(
".rcp-auto-meta-probe-{}-{:016x}",
std::process::id(),
suffix,
));
match std::fs::OpenOptions::new()
.create_new(true)
.write(true)
.open(&probe)
{
Ok(_) => {
let _ = std::fs::remove_file(&probe);
}
Err(err) => {
return Err(format!(
"--auto-meta-histogram-log parent {parent:?} is not writable: {err:#}",
));
}
}
}
}
Ok(())
}
}
#[derive(Debug, Clone, Copy, Default)]
pub struct OutputConfig {
pub quiet: bool,
pub verbose: u8,
pub print_summary: bool,
pub suppress_runtime_stats: bool,
pub startup_error_prefix: Option<&'static str>,
}
pub struct DryRunWarnings {
warnings: Vec<String>,
}
impl DryRunWarnings {
#[must_use]
pub fn new(
has_progress: bool,
has_summary: bool,
verbose: u8,
has_overwrite: bool,
has_filters: bool,
has_destination: bool,
has_ignore_existing: bool,
) -> Self {
let mut warnings = Vec::new();
if has_progress {
warnings.push("dry-run: --progress was ignored".to_string());
}
if has_summary && verbose == 0 {
warnings.push("dry-run: --summary was ignored".to_string());
}
if has_overwrite {
warnings.push(
"dry-run: --overwrite was ignored; dry-run does not check destination state"
.to_string(),
);
}
if !has_filters && !has_ignore_existing {
if has_destination {
warnings.push(
"dry-run: no filtering specified. dry-run is primarily useful to preview \
--include/--exclude/--filter-file filtering; it does not check whether \
files already exist at the destination."
.to_string(),
);
} else {
warnings.push(
"dry-run: no filtering specified. dry-run is primarily useful to preview \
--include/--exclude/--filter-file filtering."
.to_string(),
);
}
}
Self { warnings }
}
pub fn print(&self) {
for warning in &self.warnings {
eprintln!("{warning}");
}
}
}
#[derive(Debug)]
pub struct TracingConfig {
pub remote_layer: Option<crate::remote_tracing::RemoteTracingLayer>,
pub debug_log_file: Option<String>,
pub chrome_trace_prefix: Option<String>,
pub flamegraph_prefix: Option<String>,
pub trace_identifier: String,
pub profile_level: Option<String>,
pub tokio_console: bool,
pub tokio_console_port: Option<u16>,
}
impl TracingConfig {
#[must_use]
pub fn local(trace_identifier: &str) -> Self {
Self {
remote_layer: None,
debug_log_file: None,
chrome_trace_prefix: None,
flamegraph_prefix: None,
trace_identifier: trace_identifier.to_string(),
profile_level: None,
tokio_console: false,
tokio_console_port: None,
}
}
}
impl Default for TracingConfig {
fn default() -> Self {
Self {
remote_layer: None,
debug_log_file: None,
chrome_trace_prefix: None,
flamegraph_prefix: None,
trace_identifier: "unknown".to_string(),
profile_level: None,
tokio_console: false,
tokio_console_port: None,
}
}
}
#[cfg(test)]
mod files_in_flight_policy_tests {
use super::*;
use std::num::NonZeroUsize;
#[test]
fn limit_meet_uses_the_lower_finite_value() {
let four = ConcurrencyLimit::Limited(NonZeroUsize::new(4).unwrap());
let eight = ConcurrencyLimit::Limited(NonZeroUsize::new(8).unwrap());
assert_eq!(four.meet(eight), four);
assert_eq!(ConcurrencyLimit::Unlimited.meet(eight), eight);
assert_eq!(eight.meet(ConcurrencyLimit::Unlimited), eight);
assert_eq!(
ConcurrencyLimit::Unlimited.meet(ConcurrencyLimit::Unlimited),
ConcurrencyLimit::Unlimited,
);
}
#[test]
fn concurrency_limit_display_is_operator_friendly() {
let eight = ConcurrencyLimit::Limited(NonZeroUsize::new(8).unwrap());
assert_eq!(eight.to_string(), "8");
assert_eq!(ConcurrencyLimit::Unlimited.to_string(), "unlimited");
}
#[test]
fn explicit_unlimited_file_ceiling_removes_only_the_user_cap() {
let limit = ResolvedFilesInFlight::unlimited();
assert_eq!(limit.limit(), ConcurrencyLimit::Unlimited);
assert_eq!(limit.source(), FilesInFlightSource::Explicit);
}
#[test]
fn automatic_files_in_flight_has_a_floor_of_four() {
assert_eq!(default_files_in_flight_from(None).get(), 4);
assert_eq!(default_files_in_flight_from(NonZeroUsize::new(2)).get(), 4);
assert_eq!(default_files_in_flight_from(NonZeroUsize::new(4)).get(), 4);
assert_eq!(
default_files_in_flight_from(NonZeroUsize::new(12)).get(),
12
);
}
#[test]
fn injected_automatic_file_limit_retains_automatic_provenance() {
let limit = ResolvedFilesInFlight::automatic_with(NonZeroUsize::new(12).unwrap());
assert_eq!(
limit.limit(),
ConcurrencyLimit::Limited(NonZeroUsize::new(12).unwrap())
);
assert_eq!(limit.source(), FilesInFlightSource::Automatic);
}
#[test]
fn forwarded_legacy_finite_retains_origin_without_a_duplicate_warning() {
let files_in_flight = ResolvedFilesInFlight::forwarded_legacy(7);
let throttle = ThrottleConfig {
files_in_flight,
..ThrottleConfig::default()
};
assert_eq!(
files_in_flight.limit(),
ConcurrencyLimit::Limited(NonZeroUsize::new(7).unwrap())
);
assert_eq!(
files_in_flight.source(),
FilesInFlightSource::DeprecatedMaxOpenFiles
);
assert_eq!(throttle.deprecated_max_open_files_warning(), None);
}
#[test]
fn forwarded_legacy_unlimited_retains_origin_without_a_duplicate_warning() {
let files_in_flight = ResolvedFilesInFlight::forwarded_legacy(0);
let throttle = ThrottleConfig {
files_in_flight,
..ThrottleConfig::default()
};
assert_eq!(files_in_flight.limit(), ConcurrencyLimit::Unlimited);
assert_eq!(
files_in_flight.source(),
FilesInFlightSource::DeprecatedMaxOpenFiles
);
assert_eq!(throttle.deprecated_max_open_files_warning(), None);
}
}
#[cfg(test)]
mod auto_meta_validation_tests {
use super::*;
fn valid_auto_meta() -> AutoMetaThrottleConfig {
AutoMetaThrottleConfig {
initial_cwnd: 1,
min_cwnd: 1,
max_cwnd: 4096,
alpha: 1.3,
beta: 1.8,
increase_step: 1,
decrease_step: 1,
baseline_percentile: 0.1,
current_percentile: 0.5,
long_window: std::time::Duration::from_secs(10),
short_window: std::time::Duration::from_secs(1),
tick_interval: std::time::Duration::from_millis(50),
}
}
fn config_with(auto: AutoMetaThrottleConfig) -> ThrottleConfig {
ThrottleConfig {
files_in_flight: ResolvedFilesInFlight::automatic(),
apply_files_in_flight: true,
ops_throttle: 0,
iops_throttle: 0,
chunk_size: 0,
auto_meta: Some(auto),
histogram_enabled: false,
histogram_log_path: None,
histogram_interval: std::time::Duration::from_secs(1),
}
}
#[test]
fn defaults_validate() {
assert!(config_with(valid_auto_meta()).validate().is_ok());
}
#[test]
fn min_cwnd_zero_is_rejected() {
let mut auto = valid_auto_meta();
auto.min_cwnd = 0;
let err = config_with(auto).validate().unwrap_err();
assert!(err.contains("min-cwnd"), "got: {err}");
}
#[test]
fn alpha_at_or_below_zero_is_rejected() {
let mut auto = valid_auto_meta();
auto.alpha = 0.0;
assert!(config_with(auto).validate().is_err());
let mut auto = valid_auto_meta();
auto.alpha = -0.5;
assert!(config_with(auto).validate().is_err());
}
#[test]
fn alpha_below_one_is_accepted() {
let mut auto = valid_auto_meta();
auto.alpha = 0.9;
auto.beta = 1.1;
assert!(config_with(auto).validate().is_ok());
}
#[test]
fn beta_at_or_below_zero_is_rejected() {
let mut auto = valid_auto_meta();
auto.alpha = 0.5;
auto.beta = 0.0;
let err = config_with(auto).validate().unwrap_err();
assert!(err.contains("beta"), "got: {err}");
}
#[test]
fn cross_percentile_config_validates() {
let mut auto = valid_auto_meta();
auto.baseline_percentile = 0.4;
auto.current_percentile = 0.6;
assert!(config_with(auto).validate().is_ok());
}
#[test]
fn baseline_percentile_above_current_is_rejected() {
let mut auto = valid_auto_meta();
auto.baseline_percentile = 0.6;
auto.current_percentile = 0.4;
let err = config_with(auto).validate().unwrap_err();
assert!(
err.contains("baseline-percentile") && err.contains("current-percentile"),
"got: {err}",
);
}
#[test]
fn baseline_percentile_out_of_range_is_rejected() {
let mut auto = valid_auto_meta();
auto.baseline_percentile = 1.0;
let err = config_with(auto).validate().unwrap_err();
assert!(err.contains("baseline-percentile"), "got: {err}");
}
#[test]
fn non_finite_alpha_or_beta_is_rejected() {
for bad in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY] {
let mut auto = valid_auto_meta();
auto.alpha = bad;
assert!(
config_with(auto).validate().is_err(),
"alpha={bad} must be rejected",
);
let mut auto = valid_auto_meta();
auto.beta = bad;
assert!(
config_with(auto).validate().is_err(),
"beta={bad} must be rejected",
);
}
}
#[test]
fn current_percentile_out_of_range_is_rejected() {
let mut auto = valid_auto_meta();
auto.current_percentile = 1.0;
let err = config_with(auto).validate().unwrap_err();
assert!(err.contains("current-percentile"), "got: {err}");
}
#[test]
fn ops_throttle_below_floor_is_rejected_under_auto_meta() {
let mut config = config_with(valid_auto_meta());
config.ops_throttle = 5;
let err = config.validate().unwrap_err();
assert!(
err.contains("ops-throttle") && err.contains("auto-meta-throttle"),
"got: {err}",
);
}
#[test]
fn ops_throttle_at_or_above_floor_is_accepted_under_auto_meta() {
let mut config = config_with(valid_auto_meta());
config.ops_throttle = AUTO_META_MIN_OPS_THROTTLE;
assert!(config.validate().is_ok());
config.ops_throttle = AUTO_META_MIN_OPS_THROTTLE + 100;
assert!(config.validate().is_ok());
}
#[test]
fn ops_throttle_below_floor_is_fine_without_auto_meta() {
let config = ThrottleConfig {
files_in_flight: ResolvedFilesInFlight::automatic(),
apply_files_in_flight: true,
ops_throttle: 5,
iops_throttle: 0,
chunk_size: 0,
auto_meta: None,
histogram_enabled: false,
histogram_log_path: None,
histogram_interval: std::time::Duration::from_secs(1),
};
assert!(config.validate().is_ok());
}
#[test]
fn alpha_greater_than_beta_is_rejected() {
let mut auto = valid_auto_meta();
auto.alpha = 1.6;
auto.beta = 1.5;
let err = config_with(auto).validate().unwrap_err();
assert!(err.contains("alpha") && err.contains("beta"), "got: {err}");
}
#[test]
fn histogram_log_without_throttle_is_rejected() {
let config = ThrottleConfig {
files_in_flight: ResolvedFilesInFlight::automatic(),
apply_files_in_flight: true,
ops_throttle: 0,
iops_throttle: 0,
chunk_size: 0,
auto_meta: None,
histogram_log_path: Some("/tmp/x.hdr".into()),
histogram_enabled: false,
histogram_interval: std::time::Duration::from_secs(1),
};
let err = config.validate().unwrap_err();
assert!(
err.contains("histogram") && err.contains("auto-meta-throttle"),
"got: {err}"
);
}
#[test]
fn histogram_enabled_without_throttle_is_rejected() {
let config = ThrottleConfig {
files_in_flight: ResolvedFilesInFlight::automatic(),
apply_files_in_flight: true,
ops_throttle: 0,
iops_throttle: 0,
chunk_size: 0,
auto_meta: None,
histogram_enabled: true,
histogram_log_path: None,
histogram_interval: std::time::Duration::from_secs(1),
};
let err = config.validate().unwrap_err();
assert!(
err.contains("histogram") && err.contains("auto-meta-throttle"),
"got: {err}"
);
}
#[test]
fn histogram_interval_below_floor_is_rejected() {
let mut config = config_with(valid_auto_meta());
config.histogram_enabled = true;
config.histogram_interval = std::time::Duration::from_millis(50);
let err = config.validate().unwrap_err();
assert!(err.contains("histogram-interval"), "got: {err}");
}
#[test]
fn histogram_interval_above_ceiling_is_rejected() {
let mut config = config_with(valid_auto_meta());
config.histogram_enabled = true;
config.histogram_interval = std::time::Duration::from_secs(120);
let err = config.validate().unwrap_err();
assert!(err.contains("histogram-interval"), "got: {err}");
}
#[test]
fn histogram_defaults_pass_validation() {
let mut config = config_with(valid_auto_meta());
config.histogram_enabled = true;
config.histogram_interval = std::time::Duration::from_secs(1);
assert!(config.validate().is_ok());
}
#[test]
fn histogram_log_with_missing_parent_is_rejected() {
let mut config = config_with(valid_auto_meta());
config.histogram_log_path = Some("/nonexistent-dir-12345/foo.hdr".into());
let err = config.validate().unwrap_err();
assert!(
err.contains("histogram-log") && err.contains("parent"),
"got: {err}",
);
}
#[test]
fn histogram_log_with_writable_parent_is_accepted() {
let dir = tempfile::tempdir().unwrap();
let mut config = config_with(valid_auto_meta());
config.histogram_log_path = Some(dir.path().join("foo.hdr"));
assert!(config.validate().is_ok());
}
#[test]
fn histogram_log_with_bare_filename_is_accepted() {
let mut config = config_with(valid_auto_meta());
config.histogram_log_path = Some("bare-filename.hdr".into());
assert!(
config.validate().is_ok(),
"validate err: {:?}",
config.validate(),
);
}
#[test]
fn histogram_log_validation_uses_unique_probe_per_call() {
let dir = tempfile::tempdir().unwrap();
let mut config = config_with(valid_auto_meta());
config.histogram_log_path = Some(dir.path().join("log.hdr"));
assert!(config.validate().is_ok());
assert!(config.validate().is_ok());
}
}