poolsim_core/
advanced.rs

1//! Optional advanced sizing helpers for acquisition waits, transaction mixes, and leaks.
2//!
3//! These helpers are additive and do not change the default simulation model.
4
5use crate::{erlang, error::PoolsimError, types::WorkloadConfig};
6
7/// Two-stage acquisition wait estimate for a fixed pool size.
8#[derive(Debug, Clone, PartialEq)]
9pub struct AcquisitionEstimate {
10    pool_size: u32,
11    utilisation_rho: f64,
12    mean_acquisition_wait_ms: f64,
13    p99_acquisition_wait_ms: f64,
14    acquisition_timeout_ms: f64,
15    timeout_risk: bool,
16}
17
18impl AcquisitionEstimate {
19    /// Fixed pool size used by the estimate.
20    pub fn pool_size(&self) -> u32 {
21        self.pool_size
22    }
23
24    /// Estimated utilisation ratio.
25    pub fn utilisation_rho(&self) -> f64 {
26        self.utilisation_rho
27    }
28
29    /// Estimated mean wait to acquire a pool slot in milliseconds.
30    pub fn mean_acquisition_wait_ms(&self) -> f64 {
31        self.mean_acquisition_wait_ms
32    }
33
34    /// Estimated p99 wait to acquire a pool slot in milliseconds.
35    pub fn p99_acquisition_wait_ms(&self) -> f64 {
36        self.p99_acquisition_wait_ms
37    }
38
39    /// Configured acquisition timeout in milliseconds.
40    pub fn acquisition_timeout_ms(&self) -> f64 {
41        self.acquisition_timeout_ms
42    }
43
44    /// Whether the p99 acquisition wait is at or above the timeout.
45    pub fn timeout_risk(&self) -> bool {
46        self.timeout_risk
47    }
48}
49
50/// Estimates pool-slot acquisition wait before database service time.
51///
52/// # Errors
53///
54/// Returns [`PoolsimError::InvalidInput`] when pool size or timeout is invalid,
55/// and propagates workload validation or Erlang-C errors.
56pub fn estimate_acquisition_wait(
57    workload: &WorkloadConfig,
58    pool_size: u32,
59    acquisition_timeout_ms: f64,
60) -> Result<AcquisitionEstimate, PoolsimError> {
61    workload.validate()?;
62    if pool_size == 0 {
63        return Err(PoolsimError::invalid_input(
64            "INVALID_POOL_SIZE",
65            "pool_size must be greater than 0",
66            None,
67        ));
68    }
69    if acquisition_timeout_ms <= 0.0 || !acquisition_timeout_ms.is_finite() {
70        return Err(PoolsimError::invalid_input(
71            "INVALID_ACQUISITION_TIMEOUT",
72            "acquisition_timeout_ms must be finite and greater than 0",
73            None,
74        ));
75    }
76
77    let lambda = workload.requests_per_second;
78    let service_mean_ms =
79        (workload.latency_p50_ms + workload.latency_p95_ms + workload.latency_p99_ms) / 3.0;
80    let mu = 1_000.0 / service_mean_ms;
81    let rho = erlang::utilisation(lambda, mu, pool_size);
82    let mean = erlang::mean_queue_wait_ms(lambda, mu, pool_size)?;
83    let p99 = erlang::queue_wait_percentile_ms(lambda, mu, pool_size, 0.99)?;
84
85    Ok(AcquisitionEstimate {
86        pool_size,
87        utilisation_rho: rho,
88        mean_acquisition_wait_ms: mean,
89        p99_acquisition_wait_ms: p99,
90        acquisition_timeout_ms,
91        timeout_risk: p99 >= acquisition_timeout_ms,
92    })
93}
94
95/// One traffic class in a transaction-level workload mix.
96#[derive(Debug, Clone, PartialEq)]
97pub struct TransactionClass {
98    name: String,
99    requests_per_second: f64,
100    latency_p50_ms: f64,
101    latency_p95_ms: f64,
102    latency_p99_ms: f64,
103}
104
105impl TransactionClass {
106    /// Creates a transaction class with its own request rate and latency percentiles.
107    pub fn new(
108        name: impl Into<String>,
109        requests_per_second: f64,
110        latency_p50_ms: f64,
111        latency_p95_ms: f64,
112        latency_p99_ms: f64,
113    ) -> Self {
114        Self {
115            name: name.into(),
116            requests_per_second,
117            latency_p50_ms,
118            latency_p95_ms,
119            latency_p99_ms,
120        }
121    }
122
123    /// Transaction class name.
124    pub fn name(&self) -> &str {
125        &self.name
126    }
127
128    /// Request rate for this class.
129    pub fn requests_per_second(&self) -> f64 {
130        self.requests_per_second
131    }
132
133    fn validate(&self) -> Result<(), PoolsimError> {
134        if self.name.trim().is_empty() {
135            return Err(PoolsimError::invalid_input(
136                "INVALID_TRANSACTION_NAME",
137                "transaction class name must not be empty",
138                None,
139            ));
140        }
141        WorkloadConfig {
142            requests_per_second: self.requests_per_second,
143            latency_p50_ms: self.latency_p50_ms,
144            latency_p95_ms: self.latency_p95_ms,
145            latency_p99_ms: self.latency_p99_ms,
146            raw_samples_ms: None,
147            step_load_profile: None,
148        }
149        .validate()
150    }
151}
152
153/// A transaction-level workload mix.
154#[derive(Debug, Clone, PartialEq)]
155pub struct TransactionMix {
156    classes: Vec<TransactionClass>,
157}
158
159impl TransactionMix {
160    /// Creates a transaction mix after validating each class.
161    ///
162    /// # Errors
163    ///
164    /// Returns [`PoolsimError::InvalidInput`] if the mix is empty or a class is invalid.
165    pub fn new(classes: Vec<TransactionClass>) -> Result<Self, PoolsimError> {
166        if classes.is_empty() {
167            return Err(PoolsimError::invalid_input(
168                "INVALID_TRANSACTION_MIX",
169                "transaction mix must contain at least one class",
170                None,
171            ));
172        }
173        for class in &classes {
174            class.validate()?;
175        }
176        Ok(Self { classes })
177    }
178
179    /// Classes in this mix.
180    pub fn classes(&self) -> &[TransactionClass] {
181        &self.classes
182    }
183
184    /// Aggregates the transaction mix into a workload usable by the existing simulator.
185    pub fn aggregate_workload(&self) -> WorkloadConfig {
186        let total_rps: f64 = self
187            .classes
188            .iter()
189            .map(|class| class.requests_per_second)
190            .sum();
191        let weighted = |latency: fn(&TransactionClass) -> f64| {
192            self.classes
193                .iter()
194                .map(|class| latency(class) * class.requests_per_second / total_rps)
195                .sum()
196        };
197
198        WorkloadConfig {
199            requests_per_second: total_rps,
200            latency_p50_ms: weighted(|class| class.latency_p50_ms),
201            latency_p95_ms: weighted(|class| class.latency_p95_ms),
202            latency_p99_ms: weighted(|class| class.latency_p99_ms),
203            raw_samples_ms: None,
204            step_load_profile: None,
205        }
206    }
207}
208
209/// Result of connection leak modeling.
210#[derive(Debug, Clone, PartialEq)]
211pub struct LeakSimulation {
212    initial_pool_size: u32,
213    final_available_connections: u32,
214    leaked_connections: u32,
215    minutes_to_exhaustion: Option<u32>,
216}
217
218impl LeakSimulation {
219    /// Initial pool size before leakage.
220    pub fn initial_pool_size(&self) -> u32 {
221        self.initial_pool_size
222    }
223
224    /// Available connections at the modeled duration.
225    pub fn final_available_connections(&self) -> u32 {
226        self.final_available_connections
227    }
228
229    /// Total leaked connections by the modeled duration.
230    pub fn leaked_connections(&self) -> u32 {
231        self.leaked_connections
232    }
233
234    /// First minute where all pool slots are leaked, if exhaustion occurs.
235    pub fn minutes_to_exhaustion(&self) -> Option<u32> {
236        self.minutes_to_exhaustion
237    }
238}
239
240/// Simulates gradual connection leakage over time.
241///
242/// # Errors
243///
244/// Returns [`PoolsimError::InvalidInput`] when pool size, leak rate, or duration is invalid.
245pub fn simulate_connection_leak(
246    initial_pool_size: u32,
247    leak_rate_per_minute: f64,
248    duration_minutes: u32,
249) -> Result<LeakSimulation, PoolsimError> {
250    if initial_pool_size == 0 {
251        return Err(PoolsimError::invalid_input(
252            "INVALID_POOL_SIZE",
253            "initial_pool_size must be greater than 0",
254            None,
255        ));
256    }
257    if leak_rate_per_minute < 0.0 || !leak_rate_per_minute.is_finite() {
258        return Err(PoolsimError::invalid_input(
259            "INVALID_LEAK_RATE",
260            "leak_rate_per_minute must be finite and non-negative",
261            None,
262        ));
263    }
264
265    let leaked = (leak_rate_per_minute * f64::from(duration_minutes)).floor() as u32;
266    let capped_leaked = leaked.min(initial_pool_size);
267    let minutes_to_exhaustion = if leak_rate_per_minute > 0.0 {
268        Some((f64::from(initial_pool_size) / leak_rate_per_minute).ceil() as u32)
269            .filter(|minutes| *minutes <= duration_minutes)
270    } else {
271        None
272    };
273
274    Ok(LeakSimulation {
275        initial_pool_size,
276        final_available_connections: initial_pool_size.saturating_sub(capped_leaked),
277        leaked_connections: capped_leaked,
278        minutes_to_exhaustion,
279    })
280}
281
282#[cfg(test)]
283mod tests {
284    use super::*;
285
286    fn workload() -> WorkloadConfig {
287        WorkloadConfig {
288            requests_per_second: 100.0,
289            latency_p50_ms: 5.0,
290            latency_p95_ms: 20.0,
291            latency_p99_ms: 40.0,
292            raw_samples_ms: None,
293            step_load_profile: None,
294        }
295    }
296
297    #[test]
298    fn acquisition_wait_estimates_timeout_risk() {
299        let estimate = estimate_acquisition_wait(&workload(), 4, 100.0).expect("estimate");
300        assert_eq!(estimate.pool_size(), 4);
301        assert!(estimate.utilisation_rho().is_finite());
302        assert!(estimate.p99_acquisition_wait_ms() >= 0.0);
303        assert_eq!(
304            estimate.timeout_risk(),
305            estimate.p99_acquisition_wait_ms() >= 100.0
306        );
307    }
308
309    #[test]
310    fn transaction_mix_aggregates_weighted_workload() {
311        let mix = TransactionMix::new(vec![
312            TransactionClass::new("fast-read", 80.0, 4.0, 12.0, 30.0),
313            TransactionClass::new("slow-write", 20.0, 20.0, 80.0, 160.0),
314        ])
315        .expect("valid mix");
316        assert_eq!(mix.classes()[0].name(), "fast-read");
317        let workload = mix.aggregate_workload();
318        assert_eq!(workload.requests_per_second, 100.0);
319        assert!(workload.latency_p99_ms > 30.0);
320    }
321
322    #[test]
323    fn connection_leak_reports_exhaustion() {
324        let leak = simulate_connection_leak(10, 2.0, 6).expect("leak simulation");
325        assert_eq!(leak.initial_pool_size(), 10);
326        assert_eq!(leak.leaked_connections(), 10);
327        assert_eq!(leak.final_available_connections(), 0);
328        assert_eq!(leak.minutes_to_exhaustion(), Some(5));
329    }
330
331    #[test]
332    fn invalid_advanced_inputs_are_rejected() {
333        assert!(estimate_acquisition_wait(&workload(), 0, 100.0).is_err());
334        assert!(TransactionMix::new(Vec::new()).is_err());
335        assert!(TransactionMix::new(vec![TransactionClass::new("", 1.0, 1.0, 2.0, 3.0)]).is_err());
336        assert!(simulate_connection_leak(1, f64::NAN, 10).is_err());
337    }
338}
poolsim_core/advanced.rs

poolsim_core/
advanced.rs
