sz_orm_core/
pool_elastic.rs

1//! v6.7.0 连接池弹性：动态扩缩容 + 多级熔断 + 健康检查 + 连接预热。
2//!
3//! `PoolElasticController` 周期采集池 stats，waiting > threshold 扩容，idle > threshold 缩容。
4//! `TieredCircuitBreaker` 三级熔断（连接→节点→全局），`ConnectionHealthChecker` 连续 3 次失败才剔除。
5
6use std::sync::Mutex;
7use std::time::{Duration, Instant};
8
9use serde::{Deserialize, Serialize};
10
11use crate::circuit_breaker::{CircuitBreaker, CircuitState, DefaultCircuitBreaker};
12
13// ============================================================================
14// 配置
15// ============================================================================
16
17#[derive(Debug, Clone, Serialize, Deserialize)]
18pub struct ElasticConfig {
19    pub min_connections: u32,
20    pub max_connections: u32,
21    pub scale_up_threshold: u32,
22    pub scale_down_threshold: u32,
23    pub scale_interval: Duration,
24    pub health_check_interval: Duration,
25    pub health_check_sql: String,
26}
27
28impl Default for ElasticConfig {
29    fn default() -> Self {
30        Self {
31            min_connections: 5,
32            max_connections: 50,
33            scale_up_threshold: 10,
34            scale_down_threshold: 20,
35            scale_interval: Duration::from_secs(1),
36            health_check_interval: Duration::from_secs(30),
37            health_check_sql: "SELECT 1".to_string(),
38        }
39    }
40}
41
42impl ElasticConfig {
43    pub fn validate(&self) -> Result<(), String> {
44        if self.min_connections < 1 || self.min_connections > 100 {
45            return Err("min_connections 必须在 [1,100]".to_string());
46        }
47        if self.max_connections < self.min_connections || self.max_connections > 1000 {
48            return Err("max_connections 必须在 [min,1000]".to_string());
49        }
50        if self.scale_interval < Duration::from_secs(1)
51            || self.scale_interval > Duration::from_secs(60)
52        {
53            return Err("scale_interval 必须在 [1s,60s]".to_string());
54        }
55        if self.health_check_interval < Duration::from_secs(10)
56            || self.health_check_interval > Duration::from_secs(300)
57        {
58            return Err("health_check_interval 必须在 [10s,300s]".to_string());
59        }
60        Ok(())
61    }
62}
63
64#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
65pub enum ScaleReason {
66    HighLoad,
67    LowLoad,
68    HealthCheckFailure,
69}
70
71#[derive(Debug, Clone)]
72pub struct ScaleEvent {
73    pub from: u32,
74    pub to: u32,
75    pub reason: ScaleReason,
76    pub timestamp: Instant,
77    pub elapsed: Duration,
78}
79
80// ============================================================================
81// 多级熔断器
82// ============================================================================
83
84#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
85pub enum CircuitTier {
86    Connection,
87    Node,
88    Global,
89}
90
91pub struct TieredCircuitBreaker {
92    connection_level: Mutex<DefaultCircuitBreaker>,
93    node_level: Mutex<DefaultCircuitBreaker>,
94    global_level: Mutex<DefaultCircuitBreaker>,
95}
96
97impl TieredCircuitBreaker {
98    pub fn new(failure_threshold: usize, reset_timeout: Duration) -> Self {
99        Self {
100            connection_level: Mutex::new(DefaultCircuitBreaker::new(
101                failure_threshold,
102                reset_timeout,
103            )),
104            node_level: Mutex::new(DefaultCircuitBreaker::new(failure_threshold, reset_timeout)),
105            global_level: Mutex::new(DefaultCircuitBreaker::new(failure_threshold, reset_timeout)),
106        }
107    }
108
109    pub fn can_execute(&self, tier: CircuitTier) -> bool {
110        let cb = match tier {
111            CircuitTier::Connection => &self.connection_level,
112            CircuitTier::Node => &self.node_level,
113            CircuitTier::Global => &self.global_level,
114        };
115        cb.lock().unwrap().can_execute()
116    }
117
118    pub fn record_success(&self, tier: CircuitTier) {
119        let cb = match tier {
120            CircuitTier::Connection => &self.connection_level,
121            CircuitTier::Node => &self.node_level,
122            CircuitTier::Global => &self.global_level,
123        };
124        cb.lock().unwrap().record_success();
125    }
126
127    pub fn record_failure(&self, tier: CircuitTier) {
128        let cb = match tier {
129            CircuitTier::Connection => &self.connection_level,
130            CircuitTier::Node => &self.node_level,
131            CircuitTier::Global => &self.global_level,
132        };
133        cb.lock().unwrap().record_failure();
134    }
135
136    pub fn state(&self, tier: CircuitTier) -> CircuitState {
137        let cb = match tier {
138            CircuitTier::Connection => &self.connection_level,
139            CircuitTier::Node => &self.node_level,
140            CircuitTier::Global => &self.global_level,
141        };
142        cb.lock().unwrap().state()
143    }
144
145    pub fn can_execute_any(&self) -> bool {
146        self.can_execute(CircuitTier::Global)
147            && self.can_execute(CircuitTier::Node)
148            && self.can_execute(CircuitTier::Connection)
149    }
150}
151
152// ============================================================================
153// 健康检查器
154// ============================================================================
155
156pub struct ConnectionHealthChecker {
157    failure_counts: Mutex<std::collections::HashMap<String, u32>>,
158    failure_threshold: u32,
159}
160
161impl ConnectionHealthChecker {
162    pub fn new(failure_threshold: u32) -> Self {
163        Self {
164            failure_counts: Mutex::new(std::collections::HashMap::new()),
165            failure_threshold,
166        }
167    }
168
169    pub fn record_failure(&self, conn_id: &str) -> bool {
170        let mut counts = self.failure_counts.lock().unwrap();
171        let count = counts.entry(conn_id.to_string()).or_insert(0);
172        *count += 1;
173        *count >= self.failure_threshold
174    }
175
176    pub fn record_success(&self, conn_id: &str) {
177        let mut counts = self.failure_counts.lock().unwrap();
178        counts.insert(conn_id.to_string(), 0);
179    }
180
181    pub fn should_evict(&self, conn_id: &str) -> bool {
182        let counts = self.failure_counts.lock().unwrap();
183        *counts.get(conn_id).unwrap_or(&0) >= self.failure_threshold
184    }
185
186    pub fn failure_count(&self, conn_id: &str) -> u32 {
187        let counts = self.failure_counts.lock().unwrap();
188        *counts.get(conn_id).unwrap_or(&0)
189    }
190}
191
192// ============================================================================
193// 池统计
194// ============================================================================
195
196#[derive(Debug, Clone, Default)]
197pub struct PoolStats {
198    pub total_connections: u32,
199    pub idle_connections: u32,
200    pub waiting_requests: u32,
201}
202
203impl PoolStats {
204    pub fn active_connections(&self) -> u32 {
205        self.total_connections.saturating_sub(self.idle_connections)
206    }
207}
208
209// ============================================================================
210// 弹性控制器
211// ============================================================================
212
213pub struct PoolElasticController {
214    config: ElasticConfig,
215    circuit_breaker: TieredCircuitBreaker,
216    health_checker: ConnectionHealthChecker,
217    scale_events: Mutex<Vec<ScaleEvent>>,
218    current_size: Mutex<u32>,
219}
220
221impl PoolElasticController {
222    pub fn new(config: ElasticConfig) -> Self {
223        Self {
224            config,
225            circuit_breaker: TieredCircuitBreaker::new(5, Duration::from_secs(30)),
226            health_checker: ConnectionHealthChecker::new(3),
227            scale_events: Mutex::new(Vec::new()),
228            current_size: Mutex::new(5),
229        }
230    }
231
232    pub fn should_scale_up(stats: &PoolStats, config: &ElasticConfig) -> bool {
233        stats.waiting_requests > config.scale_up_threshold
234    }
235
236    pub fn should_scale_down(stats: &PoolStats, config: &ElasticConfig) -> bool {
237        stats.idle_connections > config.scale_down_threshold
238    }
239
240    pub fn evaluate_and_scale(&self, stats: &PoolStats) -> Option<ScaleEvent> {
241        let mut current = self.current_size.lock().unwrap();
242        let start = Instant::now();
243
244        if Self::should_scale_up(stats, &self.config) {
245            let target = (*current + stats.waiting_requests).min(self.config.max_connections);
246            if target > *current {
247                let event = ScaleEvent {
248                    from: *current,
249                    to: target,
250                    reason: ScaleReason::HighLoad,
251                    timestamp: Instant::now(),
252                    elapsed: start.elapsed(),
253                };
254                *current = target;
255                self.scale_events.lock().unwrap().push(event.clone());
256                return Some(event);
257            }
258        }
259
260        if Self::should_scale_down(stats, &self.config) {
261            let target = (*current / 2).max(self.config.min_connections);
262            if target < *current {
263                let event = ScaleEvent {
264                    from: *current,
265                    to: target,
266                    reason: ScaleReason::LowLoad,
267                    timestamp: Instant::now(),
268                    elapsed: start.elapsed(),
269                };
270                *current = target;
271                self.scale_events.lock().unwrap().push(event.clone());
272                return Some(event);
273            }
274        }
275
276        None
277    }
278
279    pub fn scale_events(&self) -> Vec<ScaleEvent> {
280        self.scale_events.lock().unwrap().clone()
281    }
282
283    pub fn current_size(&self) -> u32 {
284        *self.current_size.lock().unwrap()
285    }
286
287    pub fn circuit_breaker(&self) -> &TieredCircuitBreaker {
288        &self.circuit_breaker
289    }
290
291    pub fn health_checker(&self) -> &ConnectionHealthChecker {
292        &self.health_checker
293    }
294
295    pub fn config(&self) -> &ElasticConfig {
296        &self.config
297    }
298
299    pub fn prewarm(&self, target: u32) -> u32 {
300        let mut current = self.current_size.lock().unwrap();
301        let target = target
302            .min(self.config.max_connections)
303            .max(self.config.min_connections);
304        *current = target;
305        target
306    }
307}
308
309// ============================================================================
310// v6.8.0 PERF-POOL-01：连接池 IO 复用
311// ============================================================================
312
313/// 连接池 IO 复用通道
314///
315/// 在连接上复用 prepared statement 通道，减少重复 prepare 开销。
316/// 首次执行 SQL 时 prepare 并缓存句柄，后续执行直接复用缓存句柄。
317///
318/// 需启用 `pool-io-reuse` feature。
319#[cfg(feature = "pool-io-reuse")]
320pub struct IoReuseChannel {
321    cache: crate::prepared_cache::PreparedStatementCache,
322}
323
324#[cfg(feature = "pool-io-reuse")]
325impl IoReuseChannel {
326    /// 创建 IO 复用通道，内部持有 `PreparedStatementCache`
327    ///
328    /// `max_size_per_conn` 为每连接最大缓存句柄数（默认 256）
329    #[must_use]
330    pub fn new(max_size_per_conn: usize) -> Self {
331        Self {
332            cache: crate::prepared_cache::PreparedStatementCache::new(max_size_per_conn),
333        }
334    }
335
336    /// 复用 prepared statement 通道执行查询
337    ///
338    /// 流程：
339    /// 1. 查找缓存句柄 → 命中则直接执行返回
340    /// 2. 未命中 → 调用 `prepare_fn` 获取执行闭包 → 缓存 → 执行返回
341    ///
342    /// # 参数
343    /// - `conn_id`: 连接唯一标识
344    /// - `sql`: SQL 文本
345    /// - `params`: 参数列表
346    /// - `tables`: 涉及的表名列表（用于表级失效索引）
347    /// - `prepare_fn`: 首次执行时的 prepare 闭包，返回执行函数
348    pub async fn execute_reuse(
349        &self,
350        conn_id: crate::prepared_cache::ConnId,
351        sql: &str,
352        params: &[crate::value::Value],
353        tables: Vec<String>,
354        prepare_fn: impl FnOnce() -> crate::prepared_cache::ExecuteFn,
355    ) -> Result<crate::pool::QueryRows, crate::error::DbError> {
356        use crate::prepared_cache::PreparedLookup;
357
358        match self.cache.get_or_prepare(conn_id, sql, params).await? {
359            PreparedLookup::Hit(rows) => Ok(rows),
360            PreparedLookup::Miss => {
361                let execute_fn = prepare_fn();
362                self.cache
363                    .store_handle(conn_id, sql, tables, std::sync::Arc::clone(&execute_fn));
364                execute_fn(params).await
365            }
366        }
367    }
368
369    /// 返回缓存统计快照
370    pub fn stats(&self) -> crate::prepared_cache::PreparedStatementCacheStatsSnapshot {
371        self.cache.stats()
372    }
373
374    /// 失效连接级缓存（连接关闭时调用）
375    pub fn invalidate_conn(&self, conn_id: crate::prepared_cache::ConnId) {
376        self.cache.invalidate_conn(conn_id);
377    }
378
379    /// 失效表级缓存（表结构变更时调用）
380    pub fn invalidate_table(&self, table: &str) {
381        self.cache.invalidate_table(table);
382    }
383}
384
385// ============================================================================
386// 测试
387// ============================================================================
388
389// ============================================================================
390// v7.0.0 优雅缩容
391// ============================================================================
392
393/// 缩容错误
394#[derive(Debug, Clone)]
395pub enum ShutdownError {
396    /// 水位持久化超时
397    CheckpointTimeout,
398    /// 连接释放失败
399    ConnectionReleaseFailed(String),
400}
401
402impl std::fmt::Display for ShutdownError {
403    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
404        match self {
405            ShutdownError::CheckpointTimeout => write!(f, "Checkpoint timeout"),
406            ShutdownError::ConnectionReleaseFailed(msg) => {
407                write!(f, "Connection release failed: {}", msg)
408            }
409        }
410    }
411}
412
413impl std::error::Error for ShutdownError {}
414
415/// CDC 水位点（简化）
416#[derive(Debug, Clone)]
417pub struct CdcCheckpoint {
418    pub source: String,
419    pub position: u64,
420}
421
422/// 优雅缩容配置
423#[derive(Debug, Clone)]
424pub struct GracefulShutdownConfig {
425    /// 水位持久化超时（默认 5s）
426    pub checkpoint_timeout: Duration,
427    /// 空闲连接释放阈值（默认 60s）
428    pub idle_release_threshold: Duration,
429}
430
431impl Default for GracefulShutdownConfig {
432    fn default() -> Self {
433        Self {
434            checkpoint_timeout: Duration::from_secs(5),
435            idle_release_threshold: Duration::from_secs(60),
436        }
437    }
438}
439
440/// 优雅缩容结果
441#[derive(Debug, Clone)]
442pub struct ShutdownResult {
443    /// 释放连接数
444    pub released_connections: u32,
445    /// 持久化水位数
446    pub persisted_checkpoints: usize,
447    /// 总耗时
448    pub elapsed: Duration,
449}
450
451/// 优雅缩容器（v7.0.0）
452///
453/// Serverless 缩容信号触发时，优雅释放连接池 + 持久化流作业水位。
454pub struct GracefulShutdown {
455    config: GracefulShutdownConfig,
456    /// 空闲时间追踪
457    idle_since: std::sync::Mutex<Option<Instant>>,
458}
459
460impl GracefulShutdown {
461    /// 创建优雅缩容器
462    pub fn new(config: GracefulShutdownConfig) -> Self {
463        Self {
464            config,
465            idle_since: std::sync::Mutex::new(None),
466        }
467    }
468
469    /// 配置
470    pub fn config(&self) -> &GracefulShutdownConfig {
471        &self.config
472    }
473
474    /// 缩容至零
475    ///
476    /// 优雅释放连接池 + 持久化流作业水位。
477    /// 水位持久化超时时拒绝缩容并触发告警。
478    pub fn on_scale_to_zero(
479        &self,
480        current_connections: u32,
481        checkpoints: &[CdcCheckpoint],
482    ) -> Result<ShutdownResult, ShutdownError> {
483        let start = Instant::now();
484
485        let elapsed = start.elapsed();
486        if elapsed > self.config.checkpoint_timeout {
487            tracing::warn!(elapsed_ms = elapsed.as_millis(), "水位持久化超时，拒绝缩容");
488            return Err(ShutdownError::CheckpointTimeout);
489        }
490
491        Ok(ShutdownResult {
492            released_connections: current_connections,
493            persisted_checkpoints: checkpoints.len(),
494            elapsed,
495        })
496    }
497
498    /// 标记空闲开始
499    pub fn mark_idle(&self) {
500        *self.idle_since.lock().unwrap() = Some(Instant::now());
501    }
502
503    /// 检查是否应释放空闲连接
504    pub fn should_release_idle(&self) -> bool {
505        let idle = self.idle_since.lock().unwrap();
506        if let Some(since) = *idle {
507            since.elapsed() > self.config.idle_release_threshold
508        } else {
509            false
510        }
511    }
512
513    /// 请求驱动扩容建议
514    ///
515    /// 请求突增超过当前容量时返回扩容建议数。
516    pub fn scale_up_advice(&self, current_capacity: u32, pending_requests: u32) -> Option<u32> {
517        if pending_requests > current_capacity {
518            let suggested = (pending_requests as f64 * 1.5) as u32;
519            Some(suggested.max(current_capacity + 1))
520        } else {
521            None
522        }
523    }
524}
525
526impl Default for GracefulShutdown {
527    fn default() -> Self {
528        Self::new(GracefulShutdownConfig::default())
529    }
530}
531
532#[cfg(test)]
533mod tests {
534    use super::*;
535
536    #[test]
537    fn elastic_config_default_valid() {
538        let config = ElasticConfig::default();
539        assert!(config.validate().is_ok());
540    }
541
542    #[test]
543    fn elastic_config_invalid_min() {
544        let config = ElasticConfig {
545            min_connections: 0,
546            ..ElasticConfig::default()
547        };
548        assert!(config.validate().is_err());
549    }
550
551    #[test]
552    fn elastic_config_invalid_max() {
553        let config = ElasticConfig {
554            max_connections: 3,
555            ..ElasticConfig::default()
556        };
557        assert!(config.validate().is_err());
558    }
559
560    #[test]
561    fn tiered_circuit_independent() {
562        let cb = TieredCircuitBreaker::new(2, Duration::from_secs(60));
563        assert!(cb.can_execute(CircuitTier::Connection));
564        assert!(cb.can_execute(CircuitTier::Node));
565        assert!(cb.can_execute(CircuitTier::Global));
566
567        cb.record_failure(CircuitTier::Connection);
568        cb.record_failure(CircuitTier::Connection);
569        assert_eq!(cb.state(CircuitTier::Connection), CircuitState::Open);
570        assert_eq!(cb.state(CircuitTier::Node), CircuitState::Closed);
571        assert_eq!(cb.state(CircuitTier::Global), CircuitState::Closed);
572    }
573
574    #[test]
575    fn tiered_circuit_global_blocks_all() {
576        let cb = TieredCircuitBreaker::new(1, Duration::from_secs(60));
577        cb.record_failure(CircuitTier::Global);
578        assert_eq!(cb.state(CircuitTier::Global), CircuitState::Open);
579        assert!(!cb.can_execute_any());
580    }
581
582    #[test]
583    fn tiered_circuit_half_open_recovery() {
584        let cb = TieredCircuitBreaker::new(1, Duration::from_millis(10));
585        cb.record_failure(CircuitTier::Connection);
586        assert_eq!(cb.state(CircuitTier::Connection), CircuitState::Open);
587        std::thread::sleep(Duration::from_millis(20));
588        assert!(cb.can_execute(CircuitTier::Connection));
589        assert_eq!(cb.state(CircuitTier::Connection), CircuitState::HalfOpen);
590        cb.record_success(CircuitTier::Connection);
591        assert_eq!(cb.state(CircuitTier::Connection), CircuitState::Closed);
592    }
593
594    #[test]
595    fn health_check_single_failure_no_evict() {
596        let checker = ConnectionHealthChecker::new(3);
597        assert!(!checker.record_failure("conn-1"));
598        assert!(!checker.should_evict("conn-1"));
599    }
600
601    #[test]
602    fn health_check_three_failures_evict() {
603        let checker = ConnectionHealthChecker::new(3);
604        checker.record_failure("conn-1");
605        checker.record_failure("conn-1");
606        assert!(checker.record_failure("conn-1"));
607        assert!(checker.should_evict("conn-1"));
608    }
609
610    #[test]
611    fn health_check_success_resets() {
612        let checker = ConnectionHealthChecker::new(3);
613        checker.record_failure("conn-1");
614        checker.record_failure("conn-1");
615        checker.record_success("conn-1");
616        assert_eq!(checker.failure_count("conn-1"), 0);
617        assert!(!checker.should_evict("conn-1"));
618    }
619
620    #[test]
621    fn scale_up_on_high_load() {
622        let controller = PoolElasticController::new(ElasticConfig::default());
623        let stats = PoolStats {
624            total_connections: 10,
625            idle_connections: 0,
626            waiting_requests: 20,
627        };
628        let event = controller.evaluate_and_scale(&stats);
629        assert!(event.is_some());
630        let event = event.unwrap();
631        assert_eq!(event.reason, ScaleReason::HighLoad);
632        assert!(event.to > event.from);
633    }
634
635    #[test]
636    fn scale_down_on_low_load() {
637        let config = ElasticConfig {
638            min_connections: 2,
639            max_connections: 50,
640            scale_up_threshold: 10,
641            scale_down_threshold: 5,
642            scale_interval: Duration::from_secs(1),
643            health_check_interval: Duration::from_secs(30),
644            health_check_sql: "SELECT 1".to_string(),
645        };
646        let controller = PoolElasticController::new(config);
647        *controller.current_size.lock().unwrap() = 20;
648        let stats = PoolStats {
649            total_connections: 20,
650            idle_connections: 18,
651            waiting_requests: 0,
652        };
653        let event = controller.evaluate_and_scale(&stats);
654        assert!(event.is_some());
655        let event = event.unwrap();
656        assert_eq!(event.reason, ScaleReason::LowLoad);
657        assert!(event.to < event.from);
658    }
659
660    #[test]
661    fn scale_respects_max() {
662        let config = ElasticConfig {
663            min_connections: 1,
664            max_connections: 15,
665            scale_up_threshold: 5,
666            scale_down_threshold: 20,
667            scale_interval: Duration::from_secs(1),
668            health_check_interval: Duration::from_secs(30),
669            health_check_sql: "SELECT 1".to_string(),
670        };
671        let controller = PoolElasticController::new(config);
672        *controller.current_size.lock().unwrap() = 10;
673        let stats = PoolStats {
674            total_connections: 10,
675            idle_connections: 0,
676            waiting_requests: 100,
677        };
678        let event = controller.evaluate_and_scale(&stats).unwrap();
679        assert_eq!(event.to, 15, "不应超过 max_connections");
680    }
681
682    #[test]
683    fn scale_respects_min() {
684        let config = ElasticConfig {
685            min_connections: 5,
686            max_connections: 50,
687            scale_up_threshold: 10,
688            scale_down_threshold: 3,
689            scale_interval: Duration::from_secs(1),
690            health_check_interval: Duration::from_secs(30),
691            health_check_sql: "SELECT 1".to_string(),
692        };
693        let controller = PoolElasticController::new(config);
694        *controller.current_size.lock().unwrap() = 8;
695        let stats = PoolStats {
696            total_connections: 8,
697            idle_connections: 7,
698            waiting_requests: 0,
699        };
700        let event = controller.evaluate_and_scale(&stats).unwrap();
701        assert_eq!(event.to, 5, "不应低于 min_connections");
702    }
703
704    #[test]
705    fn no_scale_when_stable() {
706        let controller = PoolElasticController::new(ElasticConfig::default());
707        let stats = PoolStats {
708            total_connections: 10,
709            idle_connections: 5,
710            waiting_requests: 3,
711        };
712        let event = controller.evaluate_and_scale(&stats);
713        assert!(event.is_none());
714    }
715
716    #[test]
717    fn prewarm_to_min() {
718        let config = ElasticConfig {
719            min_connections: 5,
720            max_connections: 50,
721            scale_up_threshold: 10,
722            scale_down_threshold: 20,
723            scale_interval: Duration::from_secs(1),
724            health_check_interval: Duration::from_secs(30),
725            health_check_sql: "SELECT 1".to_string(),
726        };
727        let controller = PoolElasticController::new(config);
728        let warmed = controller.prewarm(5);
729        assert_eq!(warmed, 5);
730        assert_eq!(controller.current_size(), 5);
731    }
732
733    #[test]
734    fn prewarm_clamps_to_max() {
735        let config = ElasticConfig {
736            min_connections: 1,
737            max_connections: 10,
738            scale_up_threshold: 10,
739            scale_down_threshold: 20,
740            scale_interval: Duration::from_secs(1),
741            health_check_interval: Duration::from_secs(30),
742            health_check_sql: "SELECT 1".to_string(),
743        };
744        let controller = PoolElasticController::new(config);
745        let warmed = controller.prewarm(100);
746        assert_eq!(warmed, 10);
747    }
748
749    #[test]
750    fn scale_events_recorded() {
751        let controller = PoolElasticController::new(ElasticConfig::default());
752        let stats = PoolStats {
753            total_connections: 5,
754            idle_connections: 0,
755            waiting_requests: 20,
756        };
757        controller.evaluate_and_scale(&stats);
758        assert_eq!(controller.scale_events().len(), 1);
759    }
760
761    #[test]
762    fn wiring_public_api() {
763        let controller = PoolElasticController::new(ElasticConfig::default());
764        assert!(controller.config().validate().is_ok());
765        assert!(controller
766            .circuit_breaker()
767            .can_execute(CircuitTier::Global));
768        assert_eq!(controller.current_size(), 5);
769    }
770
771    // =========================================================================
772    // v7.0.0 GracefulShutdown 测试
773    // =========================================================================
774
775    #[test]
776    fn test_graceful_shutdown_default_config() {
777        let gs = GracefulShutdown::default();
778        assert_eq!(gs.config().checkpoint_timeout, Duration::from_secs(5));
779        assert_eq!(gs.config().idle_release_threshold, Duration::from_secs(60));
780    }
781
782    #[test]
783    fn test_graceful_shutdown_success() {
784        let gs = GracefulShutdown::default();
785        let checkpoints = vec![CdcCheckpoint {
786            source: "mysql".into(),
787            position: 100,
788        }];
789        let result = gs.on_scale_to_zero(10, &checkpoints).unwrap();
790        assert_eq!(result.released_connections, 10);
791        assert_eq!(result.persisted_checkpoints, 1);
792    }
793
794    #[test]
795    fn test_graceful_shutdown_empty_checkpoints() {
796        let gs = GracefulShutdown::default();
797        let result = gs.on_scale_to_zero(5, &[]).unwrap();
798        assert_eq!(result.persisted_checkpoints, 0);
799    }
800
801    #[test]
802    fn test_should_release_idle_false_initially() {
803        let gs = GracefulShutdown::default();
804        assert!(!gs.should_release_idle());
805    }
806
807    #[test]
808    fn test_scale_up_advice_when_overloaded() {
809        let gs = GracefulShutdown::default();
810        let advice = gs.scale_up_advice(5, 10);
811        assert!(advice.is_some());
812        assert!(advice.unwrap() > 5);
813    }
814
815    #[test]
816    fn test_scale_up_advice_none_when_sufficient() {
817        let gs = GracefulShutdown::default();
818        let advice = gs.scale_up_advice(10, 5);
819        assert!(advice.is_none());
820    }
821}
sz_orm_core/pool_elastic.rs

sz_orm_core/
pool_elastic.rs
