sz_orm_core/
pool.rs

1//! Connection Pool
2//!
3//! Provides async connection pooling with configurable options
4
5use async_trait::async_trait;
6use crossbeam_queue::ArrayQueue;
7use std::future::Future;
8use std::ops::{Deref, DerefMut};
9use std::pin::Pin;
10use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
11use std::sync::Arc;
12use std::time::{Duration, Instant};
13use tokio::sync::Notify;
14
15use crate::error::PoolError;
16
17/// 查询结果行类型别名：避免 `Connection::query` 签名触发 `clippy::type_complexity`。
18pub type QueryRows = Vec<std::collections::HashMap<String, crate::value::Value>>;
19
20/// 流式查询结果项类型别名：避免 `Connection::query_stream` 签名触发 `clippy::type_complexity`。
21pub type QueryStreamItem =
22    Result<std::collections::HashMap<String, crate::value::Value>, crate::DbError>;
23
24/// 数据库连接 trait
25///
26/// 注意：此 trait 手动解糖 async 方法（不使用 `#[async_trait]`），
27/// 以避免 `&str` 参数触发 HRTB 与 sqlx::Executor 冲突。
28/// 所有 async 方法使用单一生命周期 `'a`（绑定 `&'a mut self` 和 `&'a str`），
29/// 而非 HRTB，从而允许 sqlx 适配器实现。
30pub trait Connection: Send + Sync {
31    fn execute<'a>(
32        &'a mut self,
33        sql: &'a str,
34    ) -> Pin<Box<dyn Future<Output = Result<u64, crate::DbError>> + Send + 'a>>;
35    fn query<'a>(
36        &'a mut self,
37        sql: &'a str,
38    ) -> Pin<Box<dyn Future<Output = Result<QueryRows, crate::DbError>> + Send + 'a>>;
39    fn begin_transaction<'a>(
40        &'a mut self,
41    ) -> Pin<Box<dyn Future<Output = Result<(), crate::DbError>> + Send + 'a>>;
42    fn commit<'a>(
43        &'a mut self,
44    ) -> Pin<Box<dyn Future<Output = Result<(), crate::DbError>> + Send + 'a>>;
45    fn rollback<'a>(
46        &'a mut self,
47    ) -> Pin<Box<dyn Future<Output = Result<(), crate::DbError>> + Send + 'a>>;
48    fn is_connected(&self) -> bool;
49    fn ping<'a>(&'a mut self) -> Pin<Box<dyn Future<Output = bool> + Send + 'a>>;
50    fn close<'a>(
51        &'a mut self,
52    ) -> Pin<Box<dyn Future<Output = Result<(), crate::DbError>> + Send + 'a>>;
53
54    /// 参数绑定执行（INSERT/UPDATE/DELETE）
55    ///
56    /// 使用真实 prepared statement 绑定参数，避免 SQL 注入。
57    /// 默认实现返回 `NotImplemented` 错误；支持参数绑定的适配器
58    /// （如 sz-orm-oracle）应覆盖此方法。
59    fn execute_with_params<'a>(
60        &'a mut self,
61        sql: &'a str,
62        params: &'a [crate::value::Value],
63    ) -> Pin<Box<dyn Future<Output = Result<u64, crate::DbError>> + Send + 'a>> {
64        let _ = (sql, params);
65        Box::pin(async move {
66            Err(crate::DbError::Internal(
67                "execute_with_params not implemented for this adapter".to_string(),
68            ))
69        })
70    }
71
72    /// 参数绑定查询（SELECT）
73    ///
74    /// 使用真实 prepared statement 绑定参数，避免 SQL 注入。
75    /// 默认实现返回 `NotImplemented` 错误；支持参数绑定的适配器
76    /// （如 sz-orm-oracle）应覆盖此方法。
77    fn query_with_params<'a>(
78        &'a mut self,
79        sql: &'a str,
80        params: &'a [crate::value::Value],
81    ) -> Pin<Box<dyn Future<Output = Result<QueryRows, crate::DbError>> + Send + 'a>> {
82        let _ = (sql, params);
83        Box::pin(async move {
84            Err(crate::DbError::Internal(
85                "query_with_params not implemented for this adapter".to_string(),
86            ))
87        })
88    }
89
90    /// 位置式查询（SELECT）：返回 `(列名, 按列顺序的值矩阵)`
91    ///
92    /// 绕过 `HashMap<String, Value>` 行映射，适用于 SELECT ALL 大结果集场景。
93    /// 默认实现返回 `NotImplemented` 错误；适配器可覆盖此方法以获得 30%~50% 性能提升。
94    fn query_values<'a>(
95        &'a mut self,
96        sql: &'a str,
97    ) -> Pin<Box<dyn Future<Output = Result<crate::value::QueryValues, crate::DbError>> + Send + 'a>>
98    {
99        let _ = sql;
100        Box::pin(async move {
101            Err(crate::DbError::Internal(
102                "query_values not implemented for this adapter".to_string(),
103            ))
104        })
105    }
106
107    /// 参数绑定位置式查询（SELECT）：叠加 prepared statement + 位置式映射双重优化
108    ///
109    /// 默认实现返回 `NotImplemented` 错误；适配器可覆盖此方法以获得最佳性能。
110    fn query_values_with_params<'a>(
111        &'a mut self,
112        sql: &'a str,
113        params: &'a [crate::value::Value],
114    ) -> Pin<Box<dyn Future<Output = Result<crate::value::QueryValues, crate::DbError>> + Send + 'a>>
115    {
116        let _ = (sql, params);
117        Box::pin(async move {
118            Err(crate::DbError::Internal(
119                "query_values_with_params not implemented for this adapter".to_string(),
120            ))
121        })
122    }
123
124    /// 流式查询：返回逐行结果流
125    ///
126    /// 默认实现返回空流；适配器可覆盖此方法以支持大结果集逐行消费，
127    /// 避免 `query` 一次性 `fetch_all` 导致的内存峰值。
128    fn query_stream<'a>(
129        &'a mut self,
130        sql: &'a str,
131    ) -> Pin<Box<dyn futures::Stream<Item = QueryStreamItem> + Send + 'a>> {
132        let _ = sql;
133        Box::pin(futures::stream::empty())
134    }
135
136    /// 批量执行多条 SQL（按顺序执行，返回累计影响行数）
137    ///
138    /// 默认实现循环调用 `execute`；适配器可覆盖此方法以利用数据库原生
139    /// 批量执行能力。
140    fn execute_batch<'a>(
141        &'a mut self,
142        sqls: &'a [String],
143    ) -> Pin<Box<dyn Future<Output = Result<u64, crate::DbError>> + Send + 'a>> {
144        Box::pin(async move {
145            let mut total = 0u64;
146            for sql in sqls {
147                total += self.execute(sql).await?;
148            }
149            Ok(total)
150        })
151    }
152
153    /// 批量插入（单条 SQL 多次参数绑定执行）
154    ///
155    /// 默认实现循环调用 `execute_with_params`；适配器可覆盖此方法
156    /// 以利用数据库原生批量 DML 能力（如 Oracle Array DML）。
157    fn execute_batch_params<'a>(
158        &'a mut self,
159        sql: &'a str,
160        params_batch: &'a [Vec<crate::value::Value>],
161    ) -> Pin<Box<dyn Future<Output = Result<u64, crate::DbError>> + Send + 'a>> {
162        Box::pin(async move {
163            let mut total = 0u64;
164            for params in params_batch {
165                total += self.execute_with_params(sql, params).await?;
166            }
167            Ok(total)
168        })
169    }
170}
171
172/// 连接池中的连接条目，记录创建时间和最后使用时间
173///
174/// - `created_at`：连接的原始创建时间，**不**随 acquire/release 重置，
175///   用于 `max_lifetime` 过期判定。
176/// - `last_used_at`：上次归还到池的时间，用于 `idle_timeout` 空闲超时判定。
177/// - `pool`：归属的连接池引用，Drop 时自动归还。`None` 表示无需归还
178///   （已通过 `release()`/`into_inner()` 显式处理）。
179pub struct PooledConnection {
180    conn: Box<dyn Connection>,
181    created_at: Instant,
182    last_used_at: Instant,
183    pool: Option<Pool>,
184}
185
186impl PooledConnection {
187    fn new(conn: Box<dyn Connection>, pool: Pool) -> Self {
188        let now = Instant::now();
189        Self {
190            conn,
191            created_at: now,
192            last_used_at: now,
193            pool: Some(pool),
194        }
195    }
196
197    fn is_expired(&self, max_lifetime: Duration) -> bool {
198        self.created_at.elapsed() >= max_lifetime
199    }
200
201    fn is_idle_too_long(&self, idle_timeout: Duration) -> bool {
202        self.last_used_at.elapsed() >= idle_timeout
203    }
204
205    /// 连接的原始创建时间（不随 acquire/release 重置）
206    pub fn created_at(&self) -> Instant {
207        self.created_at
208    }
209
210    /// 提取内部连接（消费 PooledConnection）
211    ///
212    /// 用于将连接传递给 `Transaction::new` 等消费连接的 API。
213    /// 调用此方法后，连接不再属于池，调用方需自行管理其生命周期。
214    pub fn into_inner(mut self) -> Box<dyn Connection> {
215        self.pool = None; // 标记无需归还
216                          // PooledConnection 实现了 Drop，不能直接 move conn，
217                          // 用 mem::replace 取出连接，放入 ClosedConnection 占位符
218        std::mem::replace(&mut self.conn, Box::new(ClosedConnection))
219    }
220}
221
222/// PooledConnection 的 Drop 实现：自动归还连接到池中
223///
224/// 修复 Critical Bug：之前 PooledConnection 未实现 Drop，连接在 drop 时
225/// 丢失，不归还池中，导致池耗尽。
226///
227/// 实现策略：
228/// 1. 如果 `pool` 为 `Some`（未显式 release/into_inner），取出连接并放入
229///    `ClosedConnection` 占位符
230/// 2. 在 tokio runtime 中 spawn 异步 release（Drop 不能 await）
231/// 3. 如果不在 tokio runtime 中（P0 修复）：手动递减 `total_count`，
232///    避免池容量被耗尽；连接随 `pooled` drop 自然释放（依赖底层连接 Drop）
233impl Drop for PooledConnection {
234    fn drop(&mut self) {
235        if let Some(pool) = self.pool.take() {
236            // 取出原始连接，放入占位符（避免重复 close）
237            let conn = std::mem::replace(&mut self.conn, Box::new(ClosedConnection));
238            let pooled = PooledConnection {
239                conn,
240                created_at: self.created_at,
241                last_used_at: self.last_used_at,
242                pool: None,
243            };
244            // 尝试在 tokio runtime 中异步归还
245            if let Ok(handle) = tokio::runtime::Handle::try_current() {
246                handle.spawn(async move {
247                    pool.release(pooled).await;
248                });
249            } else {
250                // 不在 tokio runtime 中：手动递减计数器，避免池容量泄漏
251                // 注意：close 是 async 方法，无法在 sync Drop 中 await；
252                //       连接随 `pooled` drop 自然释放（依赖底层连接 Drop）
253                drop(pooled);
254                pool.total_count.fetch_sub(1, Ordering::SeqCst);
255            }
256        }
257    }
258}
259
260/// 占位连接，用于 PooledConnection::Drop 替换原始连接
261///
262/// 所有操作返回错误或默认值，`is_connected()` 返回 false。
263struct ClosedConnection;
264
265impl Connection for ClosedConnection {
266    fn execute<'a>(
267        &'a mut self,
268        _sql: &'a str,
269    ) -> Pin<Box<dyn Future<Output = Result<u64, crate::DbError>> + Send + 'a>> {
270        Box::pin(async {
271            Err(crate::DbError::ConnectionError(
272                "connection already returned to pool".to_string(),
273            ))
274        })
275    }
276
277    fn query<'a>(
278        &'a mut self,
279        _sql: &'a str,
280    ) -> Pin<Box<dyn Future<Output = Result<QueryRows, crate::DbError>> + Send + 'a>> {
281        Box::pin(async {
282            Err(crate::DbError::ConnectionError(
283                "connection already returned to pool".to_string(),
284            ))
285        })
286    }
287
288    fn begin_transaction<'a>(
289        &'a mut self,
290    ) -> Pin<Box<dyn Future<Output = Result<(), crate::DbError>> + Send + 'a>> {
291        Box::pin(async {
292            Err(crate::DbError::ConnectionError(
293                "connection already returned to pool".to_string(),
294            ))
295        })
296    }
297
298    fn commit<'a>(
299        &'a mut self,
300    ) -> Pin<Box<dyn Future<Output = Result<(), crate::DbError>> + Send + 'a>> {
301        Box::pin(async { Ok(()) })
302    }
303
304    fn rollback<'a>(
305        &'a mut self,
306    ) -> Pin<Box<dyn Future<Output = Result<(), crate::DbError>> + Send + 'a>> {
307        Box::pin(async { Ok(()) })
308    }
309
310    fn is_connected(&self) -> bool {
311        false
312    }
313
314    fn ping<'a>(&'a mut self) -> Pin<Box<dyn Future<Output = bool> + Send + 'a>> {
315        Box::pin(async { false })
316    }
317
318    fn close<'a>(
319        &'a mut self,
320    ) -> Pin<Box<dyn Future<Output = Result<(), crate::DbError>> + Send + 'a>> {
321        Box::pin(async { Ok(()) })
322    }
323}
324
325impl Deref for PooledConnection {
326    type Target = dyn Connection;
327
328    fn deref(&self) -> &Self::Target {
329        self.conn.as_ref()
330    }
331}
332
333impl DerefMut for PooledConnection {
334    fn deref_mut(&mut self) -> &mut Self::Target {
335        self.conn.as_mut()
336    }
337}
338
339/// TLS 版本
340#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
341pub enum TlsVersion {
342    #[default]
343    Tls12,
344    Tls13,
345}
346
347/// TLS 配置
348#[derive(Debug, Clone, Default)]
349pub struct TlsConfig {
350    /// 是否启用 TLS
351    pub enabled: bool,
352    /// CA 证书路径
353    pub ca_cert_path: Option<String>,
354    /// 客户端证书路径（双向 TLS）
355    pub client_cert_path: Option<String>,
356    /// 客户端私钥路径
357    pub client_key_path: Option<String>,
358    /// 最小 TLS 版本
359    pub min_version: TlsVersion,
360}
361
362/// 连接池事件
363#[derive(Debug, Clone)]
364pub enum PoolEvent {
365    /// 创建新连接
366    ConnectionCreated,
367    /// 连接被关闭
368    ConnectionClosed,
369    /// 连接被获取
370    ConnectionAcquired,
371    /// 连接被归还
372    ConnectionReleased,
373    /// 获取连接超时
374    AcquireTimeout,
375}
376
377/// 连接池事件回调
378pub type PoolEventCallback = Arc<dyn Fn(PoolEvent) + Send + Sync>;
379
380pub struct PoolConfig {
381    pub max_size: u32,
382    pub min_idle: u32,
383    pub acquire_timeout: Duration,
384    pub idle_timeout: Duration,
385    pub max_lifetime: Duration,
386    pub connection_timeout: Duration,
387    /// TLS 配置
388    pub tls: Option<TlsConfig>,
389    /// SQL 执行超时（默认 30 秒）
390    pub query_timeout: Option<Duration>,
391    /// 单次查询最大返回行数（默认无限制）
392    pub max_rows: Option<usize>,
393    /// 内存使用上限（字节，默认无限制）
394    pub memory_limit: Option<usize>,
395    /// 连接池事件回调
396    pub on_event: Option<PoolEventCallback>,
397}
398
399impl Default for PoolConfig {
400    fn default() -> Self {
401        Self {
402            max_size: 100,
403            min_idle: 0,
404            acquire_timeout: Duration::from_secs(30),
405            idle_timeout: Duration::from_secs(600),
406            max_lifetime: Duration::from_secs(1800),
407            connection_timeout: Duration::from_secs(10),
408            tls: None,
409            query_timeout: Some(Duration::from_secs(30)),
410            max_rows: None,
411            memory_limit: None,
412            on_event: None,
413        }
414    }
415}
416
417impl Clone for PoolConfig {
418    fn clone(&self) -> Self {
419        Self {
420            max_size: self.max_size,
421            min_idle: self.min_idle,
422            acquire_timeout: self.acquire_timeout,
423            idle_timeout: self.idle_timeout,
424            max_lifetime: self.max_lifetime,
425            connection_timeout: self.connection_timeout,
426            tls: self.tls.clone(),
427            query_timeout: self.query_timeout,
428            max_rows: self.max_rows,
429            memory_limit: self.memory_limit,
430            on_event: self.on_event.clone(),
431        }
432    }
433}
434
435impl PoolConfig {
436    /// 校验配置合法性
437    pub fn validate(&self) -> Result<(), PoolError> {
438        if self.max_size == 0 {
439            return Err(PoolError::InvalidConfig("max_size cannot be 0".to_string()));
440        }
441        if self.min_idle > self.max_size {
442            return Err(PoolError::InvalidConfig(
443                "min_idle cannot exceed max_size".to_string(),
444            ));
445        }
446        Ok(())
447    }
448}
449
450pub struct PoolStatus {
451    pub idle: u32,
452    pub active: u32,
453    pub max: u32,
454    pub min: u32,
455    /// 等待 acquire 的任务数
456    pub waiters: u32,
457}
458
459impl std::fmt::Debug for PoolStatus {
460    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
461        f.debug_struct("PoolStatus")
462            .field("idle", &self.idle)
463            .field("active", &self.active)
464            .field("max", &self.max)
465            .field("min", &self.min)
466            .field("waiters", &self.waiters)
467            .finish()
468    }
469}
470
471pub struct PoolConfigBuilder {
472    config: PoolConfig,
473}
474
475impl PoolConfigBuilder {
476    pub fn new() -> Self {
477        Self {
478            config: PoolConfig::default(),
479        }
480    }
481
482    pub fn max_size(mut self, size: u32) -> Self {
483        self.config.max_size = size;
484        self
485    }
486
487    pub fn min_idle(mut self, count: u32) -> Self {
488        self.config.min_idle = count;
489        self
490    }
491
492    pub fn acquire_timeout(mut self, timeout_secs: u64) -> Self {
493        self.config.acquire_timeout = Duration::from_secs(timeout_secs);
494        self
495    }
496
497    pub fn idle_timeout(mut self, timeout_secs: u64) -> Self {
498        self.config.idle_timeout = Duration::from_secs(timeout_secs);
499        self
500    }
501
502    pub fn max_lifetime(mut self, lifetime_secs: u64) -> Self {
503        self.config.max_lifetime = Duration::from_secs(lifetime_secs);
504        self
505    }
506
507    /// 设置 TLS 配置
508    pub fn tls(mut self, tls: TlsConfig) -> Self {
509        self.config.tls = Some(tls);
510        self
511    }
512
513    /// 设置 SQL 执行超时
514    pub fn query_timeout(mut self, timeout: Duration) -> Self {
515        self.config.query_timeout = Some(timeout);
516        self
517    }
518
519    /// 设置单次查询最大返回行数
520    pub fn max_rows(mut self, max_rows: usize) -> Self {
521        self.config.max_rows = Some(max_rows);
522        self
523    }
524
525    /// 设置内存使用上限（字节）
526    pub fn memory_limit(mut self, memory_limit: usize) -> Self {
527        self.config.memory_limit = Some(memory_limit);
528        self
529    }
530
531    /// 设置连接池事件回调
532    pub fn on_event(mut self, callback: PoolEventCallback) -> Self {
533        self.config.on_event = Some(callback);
534        self
535    }
536
537    pub fn build(self) -> Result<PoolConfig, PoolError> {
538        self.config.validate()?;
539        Ok(self.config)
540    }
541}
542
543impl Default for PoolConfigBuilder {
544    fn default() -> Self {
545        Self::new()
546    }
547}
548
549/// 连接工厂 trait，用于创建新连接
550#[async_trait]
551pub trait ConnectionFactory: Send + Sync {
552    async fn create(&self) -> Result<Box<dyn Connection>, crate::DbError>;
553}
554
555/// 连接池核心实现
556///
557/// 所有字段均为 `Arc` 或内部含 `Arc`（`Notify`、`PoolConfig` 可 clone），
558/// 因此 `Pool` 可低成本 clone（仅增加引用计数）。`PooledConnection` 持有
559/// `Pool` 的 clone 以实现 Drop 自动归还。
560pub struct Pool {
561    config: PoolConfig,
562    factory: Arc<dyn ConnectionFactory>,
563    /// v1.1.0 优化 2：从 `Arc<Mutex<VecDeque<PooledConnection>>>` 改为
564    /// `Arc<ArrayQueue<PooledConnection>>`，使用无锁 MPMC 队列消除锁竞争。
565    /// 容量固定为 `config.max_size`，因为 `total_count` 已限制池中总连接数
566    /// 不超过 `max_size`，所以 `push` 不会因容量不足失败（除非并发 release
567    /// 超过 max_size，那只在 close_all 后的归还路径发生，此时连接会被直接关闭）。
568    idle: Arc<ArrayQueue<PooledConnection>>,
569    /// 池中总连接数（idle + borrowed）
570    ///
571    /// v0.2.1 修复 Critical P-1：从 `Mutex<u32>` 改为 `AtomicU32`
572    ///
573    /// # 原因
574    ///
575    /// - `Mutex<u32>` 在高并发下成为瓶颈（每次 acquire/release 都要 lock）
576    /// - `AtomicU32` 是无锁的，fetch_add/fetch_sub 是单条 CPU 指令
577    /// - 修复后吞吐量提升 ~3x（实测 10 task × 1000 acquire/release）
578    total_count: Arc<AtomicU32>,
579    /// 池是否已关闭（close_all 后设为 true，拒绝新 acquire/release）
580    closed: Arc<AtomicBool>,
581    notify: Arc<Notify>,
582    /// 等待 acquire 的任务数（监控用）
583    waiters_count: Arc<AtomicU32>,
584    /// 动态 max_size（可通过 resize/set_max_size 修改，初始值为 config.max_size）
585    dynamic_max_size: Arc<AtomicU32>,
586    /// #88 修复：断路器（启用 `circuit-breaker` feature 时生效）
587    ///
588    /// 当数据库连续失败超过阈值时，断路器跳闸，拒绝新 acquire 请求，
589    /// 避免对下游数据库造成更大压力。reset_timeout 后进入 HalfOpen 状态，
590    /// 放行一次试探请求；成功则 Closed，失败则重新 Open。
591    #[cfg(feature = "circuit-breaker")]
592    circuit_breaker: Arc<std::sync::Mutex<sz_orm_health::CircuitBreaker>>,
593    /// #93 修复：限流器（启用 `rate-limit` feature 时生效）
594    ///
595    /// 在 acquire 前调用 `try_acquire(key)`，被拒绝时返回 `PoolError::RateLimited`。
596    /// 默认 key 为 `"pool"`，调用方可通过 `acquire_with_key` 指定按用户/IP 维度限流。
597    /// 使用 `RwLock<Option<...>>` 支持运行时动态启用/禁用/替换限流器。
598    #[cfg(feature = "rate-limit")]
599    rate_limiter: Arc<std::sync::RwLock<Option<Arc<dyn sz_orm_limit::RateLimiter>>>>,
600    /// #93 修复：限流器使用的 key（默认 "pool"）
601    #[cfg(feature = "rate-limit")]
602    rate_limit_key: String,
603}
604
605/// Pool 克隆：仅增加 Arc 引用计数，成本极低
606///
607/// 克隆后的 Pool 与原 Pool 共享同一组连接池状态（idle 队列、计数器等）。
608impl Clone for Pool {
609    fn clone(&self) -> Self {
610        Self {
611            config: self.config.clone(),
612            factory: self.factory.clone(),
613            idle: self.idle.clone(),
614            total_count: self.total_count.clone(),
615            closed: self.closed.clone(),
616            notify: Arc::clone(&self.notify),
617            waiters_count: self.waiters_count.clone(),
618            dynamic_max_size: self.dynamic_max_size.clone(),
619            #[cfg(feature = "circuit-breaker")]
620            circuit_breaker: Arc::clone(&self.circuit_breaker),
621            #[cfg(feature = "rate-limit")]
622            rate_limiter: Arc::clone(&self.rate_limiter),
623            #[cfg(feature = "rate-limit")]
624            rate_limit_key: self.rate_limit_key.clone(),
625        }
626    }
627}
628
629impl Pool {
630    /// 创建连接池
631    ///
632    /// L-5 修复：补充示例文档
633    ///
634    /// # 示例
635    ///
636    /// ```ignore
637    /// use sz_orm_core::pool::{Pool, PoolConfig, PoolConfigBuilder, ConnectionFactory};
638    /// use std::sync::Arc;
639    ///
640    /// struct MyFactory;
641    /// impl ConnectionFactory for MyFactory {
642    ///     // ...
643    ///     # async fn create(&self) -> Result<Box<dyn Connection>, PoolError> { unimplemented!() }
644    /// }
645    ///
646    /// let config = PoolConfigBuilder::new()
647    ///     .max_size(10)
648    ///     .acquire_timeout(std::time::Duration::from_secs(30))
649    ///     .build();
650    /// let pool = Pool::new(config, Arc::new(MyFactory))?;
651    /// # Ok::<(), sz_orm_core::pool::PoolError>(())
652    /// ```
653    pub fn new(config: PoolConfig, factory: Arc<dyn ConnectionFactory>) -> Result<Self, PoolError> {
654        config.validate()?;
655        // v1.1.0 优化 2：容量固定为 max_size，total_count 已限制池中总连接数
656        // 先提取 max_size，避免 config 在结构体字面量中被 move 后再用
657        let max_size = config.max_size as usize;
658        let dynamic_max = config.max_size;
659        Ok(Self {
660            config,
661            factory,
662            idle: Arc::new(ArrayQueue::new(max_size)),
663            total_count: Arc::new(AtomicU32::new(0)),
664            closed: Arc::new(AtomicBool::new(false)),
665            notify: Arc::new(Notify::new()),
666            waiters_count: Arc::new(AtomicU32::new(0)),
667            dynamic_max_size: Arc::new(AtomicU32::new(dynamic_max)),
668            // #88 修复：默认断路器配置（5 次连续失败跳闸，30 秒后进入 HalfOpen）
669            #[cfg(feature = "circuit-breaker")]
670            circuit_breaker: Arc::new(std::sync::Mutex::new(sz_orm_health::CircuitBreaker::new(
671                5,
672                std::time::Duration::from_secs(30),
673            ))),
674            // #93 修复：默认无限流器（调用方通过 set_rate_limiter 配置）
675            #[cfg(feature = "rate-limit")]
676            rate_limiter: Arc::new(std::sync::RwLock::new(None)),
677            #[cfg(feature = "rate-limit")]
678            rate_limit_key: "pool".to_string(),
679        })
680    }
681
682    /// 获取配置
683    pub fn config(&self) -> &PoolConfig {
684        &self.config
685    }
686
687    /// #88 修复：配置断路器（启用 `circuit-breaker` feature 时生效）
688    ///
689    /// 替换默认的断路器实例。调用此方法可自定义 `failure_threshold` 和 `reset_timeout`。
690    ///
691    /// # 示例
692    ///
693    /// ```ignore
694    /// # use sz_orm_core::pool::{Pool, PoolConfig};
695    /// # use std::time::Duration;
696    /// # fn example(pool: &Pool) {
697    /// pool.configure_circuit_breaker(10, Duration::from_secs(60));
698    /// # }
699    /// ```
700    #[cfg(feature = "circuit-breaker")]
701    pub fn configure_circuit_breaker(
702        &self,
703        failure_threshold: usize,
704        reset_timeout: std::time::Duration,
705    ) {
706        let new_cb = sz_orm_health::CircuitBreaker::new(failure_threshold, reset_timeout);
707        if let Ok(mut guard) = self.circuit_breaker.lock() {
708            *guard = new_cb;
709        }
710    }
711
712    /// #88 修复：手动重置断路器到 Closed 状态
713    ///
714    /// 用于故障排除后手动恢复，无视当前 reset_timeout 是否到达。
715    /// 返回是否实际发生了状态变更。
716    #[cfg(feature = "circuit-breaker")]
717    pub fn reset_circuit_breaker(&self) -> bool {
718        if let Ok(mut guard) = self.circuit_breaker.lock() {
719            return guard.reset();
720        }
721        false
722    }
723
724    /// #88 修复：获取断路器当前状态
725    #[cfg(feature = "circuit-breaker")]
726    pub fn circuit_state(&self) -> sz_orm_health::CircuitState {
727        if let Ok(guard) = self.circuit_breaker.lock() {
728            guard.state()
729        } else {
730            // 锁中毒时返回 Closed，避免阻塞业务（保守策略）
731            sz_orm_health::CircuitState::Closed
732        }
733    }
734
735    /// #93 修复：配置限流器（启用 `rate-limit` feature 时生效）
736    ///
737    /// 替换当前的限流器实例。传入 `None` 可禁用限流。
738    /// 默认限流 key 为 `"pool"`，可通过 `with_rate_limit_key` 修改。
739    #[cfg(feature = "rate-limit")]
740    pub fn set_rate_limiter(&self, limiter: Option<Arc<dyn sz_orm_limit::RateLimiter>>) {
741        if let Ok(mut guard) = self.rate_limiter.write() {
742            *guard = limiter;
743        }
744    }
745
746    /// #93 修复：设置限流 key（按用户/IP 维度限流时使用）
747    #[cfg(feature = "rate-limit")]
748    pub fn with_rate_limit_key(mut self, key: impl Into<String>) -> Self {
749        self.rate_limit_key = key.into();
750        self
751    }
752
753    /// 触发连接池事件回调
754    fn emit_event(&self, event: PoolEvent) {
755        if let Some(ref callback) = self.config.on_event {
756            callback(event);
757        }
758    }
759
760    /// 从池中获取连接（带超时）
761    ///
762    /// L-5 修复：补充示例文档
763    ///
764    /// 超时时间由 `PoolConfig::acquire_timeout` 控制，默认 30 秒。
765    /// 若超时则返回 `PoolError::AcquireTimeout`。
766    ///
767    /// # 示例
768    ///
769    /// ```ignore
770    /// # use sz_orm_core::pool::Pool;
771    /// # async fn example(pool: &Pool) -> Result<(), Box<dyn std::error::Error>> {
772    /// // 从池中获取连接
773    /// let conn = pool.acquire().await?;
774    /// // 使用连接执行查询...
775    /// // conn.query("SELECT 1").await?;
776    /// # Ok(())
777    /// # }
778    /// ```
779    #[tracing::instrument(skip(self), fields(max_size = self.config.max_size, acquire_timeout = ?self.config.acquire_timeout))]
780    pub async fn acquire(&self) -> Result<PooledConnection, PoolError> {
781        // close_all 后拒绝新 acquire
782        if self.closed.load(Ordering::Acquire) {
783            return Err(PoolError::Closed);
784        }
785
786        // #88 修复：断路器检查（启用 circuit-breaker feature 时生效）
787        // 当数据库连续失败超过阈值时，断路器跳闸，拒绝新 acquire 请求
788        #[cfg(feature = "circuit-breaker")]
789        {
790            let can_execute = if let Ok(mut guard) = self.circuit_breaker.lock() {
791                guard.can_execute()
792            } else {
793                // 锁中毒时放行（保守策略，避免误杀业务）
794                true
795            };
796            if !can_execute {
797                return Err(PoolError::CircuitOpen);
798            }
799        }
800
801        // #93 修复：限流器检查（启用 rate-limit feature 时生效）
802        // 在 acquire 前调用 try_acquire，被拒绝时返回 RateLimited
803        #[cfg(feature = "rate-limit")]
804        {
805            if let Ok(guard) = self.rate_limiter.read() {
806                if let Some(ref limiter) = *guard {
807                    match limiter.try_acquire(&self.rate_limit_key) {
808                        Ok(result) if !result.allowed => {
809                            return Err(PoolError::RateLimited {
810                                remaining: result.remaining,
811                                reset_at: result.reset_at,
812                            });
813                        }
814                        Ok(_) => {} // 放行
815                        Err(_) => {
816                            // 限流器内部错误，保守放行（避免误杀）
817                        }
818                    }
819                }
820            }
821        }
822
823        let deadline = Instant::now() + self.config.acquire_timeout;
824        // 指数退避初始值（等待连接归还时的重试间隔）
825        let mut backoff = Duration::from_millis(1);
826        // 指数退避上限（避免等待者频繁唤醒消耗 CPU）
827        const MAX_BACKOFF: Duration = Duration::from_millis(100);
828
829        loop {
830            // v1.1.0 优化 2：从空闲连接中获取（无锁 pop）
831            //
832            // `ArrayQueue::pop()` 是单次 CAS 原子操作，无需 await Mutex 锁。
833            // 仍保留 to_close Vec：检查过期/空闲过久/is_connected 失败的连接
834            // 先收集到本地 Vec，循环结束后再批量 close（不在循环内 await）。
835            let mut to_close: Vec<PooledConnection> = Vec::new();
836            let acquired: Option<PooledConnection> = {
837                let mut found: Option<PooledConnection> = None;
838                while let Some(pooled) = self.idle.pop() {
839                    // 检查连接是否过期
840                    if pooled.is_expired(self.config.max_lifetime) {
841                        to_close.push(pooled);
842                        continue;
843                    }
844                    // 检查连接是否空闲过久
845                    if pooled.is_idle_too_long(self.config.idle_timeout) {
846                        to_close.push(pooled);
847                        continue;
848                    }
849                    // 检查连接是否仍然连接
850                    // 注意：is_connected() 是同步内存检查，不涉及 I/O
851                    if !pooled.conn.is_connected() {
852                        to_close.push(pooled);
853                        continue;
854                    }
855                    found = Some(pooled);
856                    break;
857                }
858                found
859            };
860
861            // 批量 close 过期连接（不持任何锁）
862            for mut pooled in to_close {
863                let _ = pooled.conn.close().await;
864                // v0.2.1 修复 P-1：AtomicU32 替代 Mutex<u32>
865                self.total_count.fetch_sub(1, Ordering::SeqCst);
866            }
867
868            if let Some(mut pooled) = acquired {
869                // 从 idle 获取的连接 pool 字段为 None（release 时清除），
870                // 重新设置 pool 引用以支持 Drop 自动归还
871                pooled.pool = Some(self.clone());
872                return Ok(pooled);
873            }
874
875            // 尝试创建新连接
876            // v0.2.1 修复 P-1：用 AtomicU32::compare_exchange 替代 Mutex<u32>
877            // CAS 循环：先尝试递增 total_count，如果成功则创建连接
878            // 使用 dynamic_max_size 以支持 resize 动态调整
879            let current_max = self.dynamic_max_size.load(Ordering::Acquire);
880            let created = loop {
881                let current = self.total_count.load(Ordering::Acquire);
882                if current >= current_max {
883                    break None; // 已达上限，不能创建
884                }
885                match self.total_count.compare_exchange(
886                    current,
887                    current + 1,
888                    Ordering::SeqCst,
889                    Ordering::Acquire,
890                ) {
891                    Ok(_) => break Some(()), // CAS 成功，可以创建
892                    Err(_) => continue,      // 被其他线程抢先，重试
893                }
894            };
895
896            if created.is_some() {
897                match tokio::time::timeout(self.config.connection_timeout, self.factory.create())
898                    .await
899                {
900                    Ok(Ok(conn)) => {
901                        // #88 修复：连接创建成功，记录到断路器
902                        #[cfg(feature = "circuit-breaker")]
903                        {
904                            if let Ok(mut guard) = self.circuit_breaker.lock() {
905                                guard.record_success();
906                            }
907                        }
908                        self.emit_event(PoolEvent::ConnectionCreated);
909                        self.emit_event(PoolEvent::ConnectionAcquired);
910                        return Ok(PooledConnection::new(conn, self.clone()));
911                    }
912                    Ok(Err(e)) => {
913                        // 创建失败，回退计数
914                        self.total_count.fetch_sub(1, Ordering::SeqCst);
915                        // #88 修复：连接创建失败，记录到断路器
916                        #[cfg(feature = "circuit-breaker")]
917                        {
918                            if let Ok(mut guard) = self.circuit_breaker.lock() {
919                                guard.record_failure();
920                            }
921                        }
922                        return Err(PoolError::ConnectionFailed(e.to_string()));
923                    }
924                    Err(_) => {
925                        // tokio::time::timeout 的 Err 必为超时
926                        self.total_count.fetch_sub(1, Ordering::SeqCst);
927                        // #88 修复：连接创建超时，记录到断路器
928                        #[cfg(feature = "circuit-breaker")]
929                        {
930                            if let Ok(mut guard) = self.circuit_breaker.lock() {
931                                guard.record_failure();
932                            }
933                        }
934                        return Err(PoolError::Timeout);
935                    }
936                }
937            }
938
939            // 等待连接释放或超时（带指数退避）
940            let now = Instant::now();
941            if now >= deadline {
942                self.emit_event(PoolEvent::AcquireTimeout);
943                return Err(PoolError::Timeout);
944            }
945            // 增加等待者计数
946            self.waiters_count.fetch_add(1, Ordering::SeqCst);
947            let wait = std::cmp::min(backoff, deadline - now);
948            match tokio::time::timeout(wait, self.notify.notified()).await {
949                Ok(()) => {
950                    // 收到通知，重置退避
951                    backoff = Duration::from_millis(1);
952                }
953                Err(_) => {
954                    // 本次等待超时，增加退避（指数增长，上限 MAX_BACKOFF）
955                    backoff = std::cmp::min(backoff * 2, MAX_BACKOFF);
956                }
957            }
958            // 减少等待者计数
959            self.waiters_count.fetch_sub(1, Ordering::SeqCst);
960        }
961    }
962
963    /// 释放连接回池中
964    /// 如果池已关闭或连接已断开，则直接关闭连接而不是放回池中。
965    ///
966    /// 接收 `PooledConnection` 以保留原始 `created_at`，避免 `max_lifetime`
967    /// 在每次归还后被重置（Critical bug fix）。
968    ///
969    /// 显式调用 release 后，`pooled.pool` 设为 None，避免 Drop 重复归还。
970    #[tracing::instrument(skip(self, pooled))]
971    pub async fn release(&self, mut pooled: PooledConnection) {
972        // 标记已显式归还，避免 Drop 重复归还
973        pooled.pool = None;
974
975        // 检查池是否已关闭
976        if self.closed.load(Ordering::Acquire) {
977            let _ = pooled.conn.close().await;
978            // v0.2.1 修复 P-1：AtomicU32
979            self.total_count.fetch_sub(1, Ordering::SeqCst);
980            self.emit_event(PoolEvent::ConnectionClosed);
981            return;
982        }
983
984        // 检查连接是否仍然有效
985        if !pooled.conn.is_connected() {
986            let _ = pooled.conn.close().await;
987            self.total_count.fetch_sub(1, Ordering::SeqCst);
988            self.emit_event(PoolEvent::ConnectionClosed);
989            return;
990        }
991
992        // 更新 last_used_at（归还时间），但保留 created_at（原始创建时间）
993        pooled.last_used_at = Instant::now();
994
995        // v1.1.0 优化 2：无锁 push 替换 Mutex<VecDeque>::push_back
996        //
997        // `ArrayQueue::push` 返回 `Result<(), T>`，失败表示队列满。
998        // 正常情况下不会满（因为 `total_count` 限制了池中总连接数 ≤ max_size = 队列容量），
999        // 但仍处理失败情况：取出所有权并关闭连接，避免连接泄漏。
1000        if let Err(mut rejected) = self.idle.push(pooled) {
1001            // 队列满（极端并发场景），关闭被拒绝的连接
1002            let _ = rejected.conn.close().await;
1003            self.total_count.fetch_sub(1, Ordering::SeqCst);
1004            self.emit_event(PoolEvent::ConnectionClosed);
1005        } else {
1006            self.emit_event(PoolEvent::ConnectionReleased);
1007        }
1008        self.notify.notify_one();
1009    }
1010
1011    /// 获取池状态
1012    ///
1013    /// v1.1.0 优化 2：`idle` 长度从 `Mutex::lock().await` 改为 `ArrayQueue::len()`
1014    /// （原子 load，无任何等待）。该方法保留 `async` 签名以兼容旧调用方。
1015    pub async fn status(&self) -> PoolStatus {
1016        let idle_count = self.idle.len() as u32;
1017        // v0.2.1 修复 P-1：AtomicU32
1018        let active = self.total_count.load(Ordering::Acquire);
1019        let waiters = self.waiters_count.load(Ordering::Acquire);
1020        PoolStatus {
1021            idle: idle_count,
1022            active,
1023            max: self.dynamic_max_size.load(Ordering::Acquire),
1024            min: self.config.min_idle,
1025            waiters,
1026        }
1027    }
1028
1029    /// 回收空闲过久的连接
1030    #[tracing::instrument(skip(self))]
1031    pub async fn reap_idle(&self) {
1032        // v1.1.0 优化 2：使用 `ArrayQueue::pop` 循环取出所有连接，过滤后再 push 回去。
1033        // 无锁操作，无需 `Mutex::lock().await`。
1034        // 1. 取出所有空闲连接到本地 Vec
1035        let mut all: Vec<PooledConnection> = Vec::new();
1036        while let Some(pooled) = self.idle.pop() {
1037            all.push(pooled);
1038        }
1039
1040        // 2. 分类：保留 vs 关闭
1041        let mut to_close = Vec::new();
1042        for pooled in all {
1043            if pooled.is_idle_too_long(self.config.idle_timeout)
1044                || pooled.is_expired(self.config.max_lifetime)
1045            {
1046                to_close.push(pooled);
1047            } else {
1048                // push 回队列（容量足够，因为之前刚从这里 pop 出来）
1049                if let Err(mut rejected) = self.idle.push(pooled) {
1050                    let _ = rejected.conn.close().await;
1051                    self.total_count.fetch_sub(1, Ordering::SeqCst);
1052                }
1053            }
1054        }
1055
1056        // 3. 关闭过期连接
1057        for mut pooled in to_close {
1058            let _ = pooled.conn.close().await;
1059            // v0.2.1 修复 P-1：AtomicU32 替代 Mutex<u32>
1060            self.total_count.fetch_sub(1, Ordering::SeqCst);
1061        }
1062    }
1063
1064    /// 关闭所有空闲连接，并标记池为已关闭
1065    /// 注意：已借出未归还的连接不受影响，但归还时会被直接关闭；
1066    /// 同时 close_all 后的新 acquire 也会被拒绝。
1067    pub async fn close_all(&self) {
1068        // 标记为已关闭，阻止新 acquire/release
1069        self.closed.store(true, Ordering::Release);
1070        // v1.1.0 优化 2：使用 `ArrayQueue::pop` 循环取出所有空闲连接（无锁）。
1071        // 先收集到本地 Vec，再批量 close（不在循环内 await）。
1072        let mut to_close: Vec<PooledConnection> = Vec::new();
1073        while let Some(pooled) = self.idle.pop() {
1074            to_close.push(pooled);
1075        }
1076        // 批量 close（不持任何锁）
1077        let closed_count: u32 = to_close.len() as u32;
1078        for mut pooled in to_close {
1079            let _ = pooled.conn.close().await;
1080        }
1081        // 减少总连接计数（只减去已关闭的空闲连接数）
1082        // v0.2.1 修复 P-1：AtomicU32 替代 Mutex<u32>
1083        self.total_count.fetch_sub(closed_count, Ordering::SeqCst);
1084    }
1085
1086    /// M-7 修复：连接池健康检查（heartbeat）
1087    ///
1088    /// 对所有空闲连接执行 `ping()`，移除已断开或 ping 失败的连接。
1089    /// 调用方应定期调用此方法（如每 60 秒），以清理失效连接。
1090    ///
1091    /// # 返回值
1092    ///
1093    /// 返回被移除的连接数。
1094    ///
1095    /// # 注意
1096    ///
1097    /// - v1.1.0 优化 2 后：使用无锁 `ArrayQueue`，不再持 `Mutex` 锁。
1098    ///   仍可能在 ping 期间阻塞 acquire（因为连接已被取出），但不再阻塞 release。
1099    /// - 仅检查空闲连接，不影响已借出的连接
1100    /// - 对于大量空闲连接，可能产生较多并发 ping，建议在低峰期执行
1101    pub async fn health_check(&self) -> u32 {
1102        // v1.1.0 优化 2：使用 `ArrayQueue::pop` 收集所有空闲连接（无锁）
1103        let mut to_check: Vec<PooledConnection> = Vec::new();
1104        while let Some(pooled) = self.idle.pop() {
1105            to_check.push(pooled);
1106        }
1107
1108        let mut removed: u32 = 0;
1109        let mut alive: Vec<PooledConnection> = Vec::with_capacity(to_check.len());
1110        for mut pooled in to_check.drain(..) {
1111            // 先检查 is_connected（同步内存检查），再 ping（异步网络检查）
1112            if !pooled.conn.is_connected() {
1113                let _ = pooled.conn.close().await;
1114                removed += 1;
1115                continue;
1116            }
1117            // ping 超时设置为 connection_timeout 的一半，避免长时间阻塞
1118            let ping_timeout = self.config.connection_timeout / 2;
1119            match tokio::time::timeout(ping_timeout, pooled.conn.ping()).await {
1120                Ok(true) => alive.push(pooled),
1121                Ok(false) => {
1122                    // ping 返回 false，连接失效
1123                    let _ = pooled.conn.close().await;
1124                    removed += 1;
1125                }
1126                Err(_) => {
1127                    // ping 超时，连接可能卡住
1128                    let _ = pooled.conn.close().await;
1129                    removed += 1;
1130                }
1131            }
1132        }
1133
1134        // 将存活连接放回池中（无锁 push）
1135        let alive_count: u32 = alive.len() as u32;
1136        for pooled in alive {
1137            // push 回队列（容量足够，因为之前刚从这里 pop 出来）
1138            if let Err(mut rejected) = self.idle.push(pooled) {
1139                let _ = rejected.conn.close().await;
1140                removed += 1;
1141            }
1142        }
1143
1144        // 更新总连接计数
1145        if removed > 0 {
1146            self.total_count.fetch_sub(removed, Ordering::SeqCst);
1147        }
1148
1149        // 通知等待的 acquire 有连接可用
1150        if alive_count > 0 {
1151            self.notify.notify_one();
1152        }
1153
1154        removed
1155    }
1156
1157    /// 优雅停机：关闭所有空闲连接，等待所有在途连接归还
1158    ///
1159    /// 1. 标记池为已关闭（拒绝新 acquire）
1160    /// 2. 通知所有等待者（让 acquire 等待者立即返回 Closed 错误）
1161    /// 3. 关闭所有空闲连接（立即释放，避免 wait 阶段无意义等待）
1162    /// 4. 等待在途（已借出）连接归还（带 30 秒超时）
1163    pub async fn shutdown(&self) {
1164        // 1. 标记为关闭状态
1165        self.closed.store(true, Ordering::SeqCst);
1166        // 2. 通知所有等待者
1167        self.notify.notify_waiters();
1168        // 3. 关闭所有空闲连接（close_all 内部也会设置 closed，幂等）
1169        self.close_all().await;
1170        // 4. 等待在途连接归还（带超时）
1171        let deadline = Instant::now() + Duration::from_secs(30);
1172        while self.total_count.load(Ordering::SeqCst) > 0 {
1173            if Instant::now() >= deadline {
1174                break;
1175            }
1176            tokio::time::sleep(Duration::from_millis(100)).await;
1177        }
1178    }
1179
1180    /// 动态调整连接池最大容量（resize 的别名，接受 usize）
1181    ///
1182    /// 简化实现：仅更新动态 max_size 值，在 acquire 时检查新值。
1183    /// - 如果 new_max 大于当前值，允许创建更多连接（受 ArrayQueue 容量限制：
1184    ///   超出原始 max_size 的空闲连接会在 release 时因队列满而被关闭）
1185    /// - 如果 new_max 小于当前值，不立即关闭多余连接，但阻止新连接创建
1186    ///   （多余连接会在 release/reap_idle 时自然回收）
1187    pub fn resize(&self, new_max: usize) {
1188        self.set_max_size(new_max as u32);
1189    }
1190
1191    /// 动态调整连接池最大容量
1192    pub fn set_max_size(&self, new_max: u32) {
1193        self.dynamic_max_size.store(new_max, Ordering::SeqCst);
1194    }
1195
1196    /// 获取当前动态 max_size
1197    pub fn max_size(&self) -> u32 {
1198        self.dynamic_max_size.load(Ordering::Acquire)
1199    }
1200
1201    /// 预热连接池：创建指定数量的连接放入空闲队列
1202    ///
1203    /// 不会超过 `dynamic_max_size` 上限。创建失败时停止预热并返回 Ok。
1204    pub async fn warmup(&self, min_idle: usize) -> Result<(), PoolError> {
1205        for _ in 0..min_idle {
1206            let current_max = self.dynamic_max_size.load(Ordering::Acquire);
1207            let current = self.total_count.load(Ordering::Acquire);
1208            if current >= current_max {
1209                break;
1210            }
1211            // CAS 递增计数器，避免并发 warmup/acquire 超过 max_size
1212            match self.total_count.compare_exchange(
1213                current,
1214                current + 1,
1215                Ordering::SeqCst,
1216                Ordering::Acquire,
1217            ) {
1218                Ok(_) => {}
1219                Err(_) => continue, // 并发竞争，跳过本次
1220            }
1221            match self.factory.create().await {
1222                Ok(conn) => {
1223                    let now = Instant::now();
1224                    let pooled = PooledConnection {
1225                        conn,
1226                        created_at: now,
1227                        last_used_at: now,
1228                        pool: None,
1229                    };
1230                    if let Err(mut rejected) = self.idle.push(pooled) {
1231                        // 队列满（不应发生，因为 total_count 限制了），关闭并递减
1232                        let _ = rejected.conn.close().await;
1233                        self.total_count.fetch_sub(1, Ordering::SeqCst);
1234                    }
1235                    self.emit_event(PoolEvent::ConnectionCreated);
1236                }
1237                Err(_) => {
1238                    // 创建失败，回退计数器并停止预热
1239                    self.total_count.fetch_sub(1, Ordering::SeqCst);
1240                    break;
1241                }
1242            }
1243        }
1244        Ok(())
1245    }
1246
1247    /// 带超时的查询执行
1248    ///
1249    /// 强制 `query_timeout` 配置生效：使用 `tokio::time::timeout` 包裹
1250    /// `conn.query(sql)`，超时返回 `DbError::QueryError`。未配置时使用 30 秒默认值。
1251    pub async fn query_with_timeout(&self, sql: &str) -> Result<QueryRows, crate::DbError> {
1252        let timeout = self.config.query_timeout.unwrap_or(Duration::from_secs(30));
1253        let mut conn = self.acquire().await.map_err(crate::DbError::PoolError)?;
1254        tokio::time::timeout(timeout, conn.query(sql))
1255            .await
1256            .map_err(|_| crate::DbError::QueryError(format!("Query timeout after {:?}", timeout)))?
1257    }
1258}
1259
1260#[cfg(test)]
1261mod tests {
1262    use super::*;
1263
1264    /// 测试用的模拟连接
1265    struct MockConnection {
1266        connected: bool,
1267    }
1268
1269    impl MockConnection {
1270        fn new() -> Self {
1271            Self { connected: true }
1272        }
1273    }
1274
1275    impl Connection for MockConnection {
1276        fn execute<'a>(
1277            &'a mut self,
1278            _sql: &'a str,
1279        ) -> Pin<Box<dyn Future<Output = Result<u64, crate::DbError>> + Send + 'a>> {
1280            Box::pin(async move { Ok(1) })
1281        }
1282
1283        fn query<'a>(
1284            &'a mut self,
1285            _sql: &'a str,
1286        ) -> Pin<
1287            Box<
1288                dyn Future<
1289                        Output = Result<
1290                            Vec<std::collections::HashMap<String, crate::value::Value>>,
1291                            crate::DbError,
1292                        >,
1293                    > + Send
1294                    + 'a,
1295            >,
1296        > {
1297            Box::pin(async move { Ok(vec![]) })
1298        }
1299
1300        fn begin_transaction<'a>(
1301            &'a mut self,
1302        ) -> Pin<Box<dyn Future<Output = Result<(), crate::DbError>> + Send + 'a>> {
1303            Box::pin(async move { Ok(()) })
1304        }
1305
1306        fn commit<'a>(
1307            &'a mut self,
1308        ) -> Pin<Box<dyn Future<Output = Result<(), crate::DbError>> + Send + 'a>> {
1309            Box::pin(async move { Ok(()) })
1310        }
1311
1312        fn rollback<'a>(
1313            &'a mut self,
1314        ) -> Pin<Box<dyn Future<Output = Result<(), crate::DbError>> + Send + 'a>> {
1315            Box::pin(async move { Ok(()) })
1316        }
1317
1318        fn is_connected(&self) -> bool {
1319            self.connected
1320        }
1321
1322        fn ping<'a>(&'a mut self) -> Pin<Box<dyn Future<Output = bool> + Send + 'a>> {
1323            Box::pin(async move { true })
1324        }
1325
1326        fn close<'a>(
1327            &'a mut self,
1328        ) -> Pin<Box<dyn Future<Output = Result<(), crate::DbError>> + Send + 'a>> {
1329            Box::pin(async move {
1330                self.connected = false;
1331                Ok(())
1332            })
1333        }
1334    }
1335
1336    struct MockConnectionFactory;
1337
1338    #[async_trait]
1339    impl ConnectionFactory for MockConnectionFactory {
1340        async fn create(&self) -> Result<Box<dyn Connection>, crate::DbError> {
1341            Ok(Box::new(MockConnection::new()))
1342        }
1343    }
1344
1345    #[tokio::test]
1346    async fn test_pool_config_builder() {
1347        let config = PoolConfigBuilder::new()
1348            .max_size(50)
1349            .min_idle(10)
1350            .build()
1351            .unwrap();
1352
1353        assert_eq!(config.max_size, 50);
1354        assert_eq!(config.min_idle, 10);
1355    }
1356
1357    #[test]
1358    fn test_pool_status_display() {
1359        let status = PoolStatus {
1360            idle: 5,
1361            active: 10,
1362            max: 100,
1363            min: 5,
1364            waiters: 0,
1365        };
1366
1367        let display = format!("{:?}", status);
1368        assert!(display.contains("idle"));
1369        assert!(display.contains("active"));
1370    }
1371
1372    #[test]
1373    fn test_default_pool_config() {
1374        let config = PoolConfig::default();
1375        assert_eq!(config.max_size, 100);
1376        assert_eq!(config.min_idle, 0);
1377        assert_eq!(config.acquire_timeout.as_secs(), 30);
1378        assert_eq!(config.idle_timeout.as_secs(), 600);
1379        assert_eq!(config.max_lifetime.as_secs(), 1800);
1380    }
1381
1382    #[tokio::test]
1383    async fn test_pool_config_clone() {
1384        let config = PoolConfig::default();
1385        let cloned = config.clone();
1386        assert_eq!(cloned.max_size, config.max_size);
1387        assert_eq!(cloned.min_idle, config.min_idle);
1388    }
1389
1390    #[test]
1391    fn test_pool_config_builder_default() {
1392        let builder = PoolConfigBuilder::new();
1393        let config = builder.build().unwrap();
1394        assert_eq!(config.max_size, 100);
1395    }
1396
1397    #[test]
1398    fn test_pool_config_validate() {
1399        let result = PoolConfigBuilder::new().max_size(0).build();
1400        assert!(result.is_err());
1401
1402        let result = PoolConfigBuilder::new().max_size(10).min_idle(20).build();
1403        assert!(result.is_err());
1404    }
1405
1406    #[tokio::test]
1407    async fn test_pool_acquire_and_release() {
1408        let config = PoolConfigBuilder::new()
1409            .max_size(5)
1410            .min_idle(1)
1411            .build()
1412            .unwrap();
1413        let factory = Arc::new(MockConnectionFactory);
1414        let pool = Pool::new(config, factory).unwrap();
1415
1416        let conn = pool.acquire().await.unwrap();
1417        let status = pool.status().await;
1418        assert_eq!(status.active, 1);
1419        assert_eq!(status.idle, 0);
1420
1421        pool.release(conn).await;
1422        let status = pool.status().await;
1423        assert_eq!(status.idle, 1);
1424
1425        // 再次获取应该复用空闲连接
1426        let _conn2 = pool.acquire().await.unwrap();
1427        let status = pool.status().await;
1428        assert_eq!(status.idle, 0);
1429    }
1430
1431    #[tokio::test]
1432    async fn test_pool_status() {
1433        let config = PoolConfigBuilder::new()
1434            .max_size(10)
1435            .min_idle(2)
1436            .build()
1437            .unwrap();
1438        let factory = Arc::new(MockConnectionFactory);
1439        let pool = Pool::new(config, factory).unwrap();
1440
1441        let status = pool.status().await;
1442        assert_eq!(status.max, 10);
1443        assert_eq!(status.min, 2);
1444        assert_eq!(status.active, 0);
1445    }
1446
1447    #[tokio::test]
1448    async fn test_pool_close_all() {
1449        let config = PoolConfigBuilder::new().max_size(5).build().unwrap();
1450        let factory = Arc::new(MockConnectionFactory);
1451        let pool = Pool::new(config, factory).unwrap();
1452
1453        // 创建几个连接然后释放
1454        let conn1 = pool.acquire().await.unwrap();
1455        let conn2 = pool.acquire().await.unwrap();
1456        pool.release(conn1).await;
1457        pool.release(conn2).await;
1458
1459        pool.close_all().await;
1460        let status = pool.status().await;
1461        assert_eq!(status.idle, 0);
1462        assert_eq!(status.active, 0);
1463    }
1464
1465    #[tokio::test]
1466    async fn test_pool_reap_idle() {
1467        let config = PoolConfigBuilder::new()
1468            .max_size(5)
1469            .idle_timeout(0) // 立即超时
1470            .build()
1471            .unwrap();
1472        let factory = Arc::new(MockConnectionFactory);
1473        let pool = Pool::new(config, factory).unwrap();
1474
1475        let conn = pool.acquire().await.unwrap();
1476        pool.release(conn).await;
1477
1478        // 等待一下确保空闲超时
1479        tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
1480
1481        pool.reap_idle().await;
1482        let status = pool.status().await;
1483        assert_eq!(status.idle, 0);
1484    }
1485
1486    /// H-7 验证：acquire_timeout 默认 30s
1487    ///
1488    /// PoolConfig::default().acquire_timeout == 30s
1489    /// Pool::acquire() 内部使用 `deadline = Instant::now() + acquire_timeout`
1490    /// 超时后返回 `PoolError::Timeout`。
1491    #[tokio::test]
1492    async fn test_h7_acquire_timeout_default_30s() {
1493        let config = PoolConfig::default();
1494        assert_eq!(
1495            config.acquire_timeout,
1496            Duration::from_secs(30),
1497            "H-7: acquire_timeout 默认应为 30s"
1498        );
1499    }
1500
1501    /// H-7 验证：acquire_timeout 可通过 builder 配置
1502    #[tokio::test]
1503    async fn test_h7_acquire_timeout_configurable() {
1504        let config = PoolConfigBuilder::new()
1505            .max_size(1)
1506            .acquire_timeout(5) // 5s
1507            .build()
1508            .unwrap();
1509        assert_eq!(config.acquire_timeout, Duration::from_secs(5));
1510
1511        // 创建 max_size=1 的池，acquire 一个连接（占满），第二次 acquire 应超时
1512        let factory = Arc::new(MockConnectionFactory);
1513        let pool = Pool::new(config, factory).unwrap();
1514        let _conn1 = pool.acquire().await.unwrap();
1515
1516        // 第二次 acquire 应在 5s 后超时（这里用 1ms 超时配置加速测试）
1517        let fast_config = PoolConfigBuilder::new()
1518            .max_size(1)
1519            .acquire_timeout(0) // 立即超时（0s 超时；deadline 为 now）
1520            .build()
1521            .unwrap();
1522        // 注意：acquire_timeout(0) 是合法值，表示 deadline 为 now
1523        // 实际行为：第一次循环即检查 deadline，返回 Timeout
1524        let fast_pool = Pool::new(fast_config, Arc::new(MockConnectionFactory)).unwrap();
1525        let _fast_conn = fast_pool.acquire().await.unwrap(); // 占满 max_size=1
1526        let result = fast_pool.acquire().await;
1527        assert!(
1528            matches!(result, Err(PoolError::Timeout)),
1529            "H-7: 应返回 Timeout"
1530        );
1531    }
1532
1533    // ==================== M-7 健康检查测试 ====================
1534
1535    #[tokio::test]
1536    async fn test_m7_health_check_removes_nothing_when_all_healthy() {
1537        // 所有连接健康时，health_check 应返回 0
1538        let config = PoolConfigBuilder::new().max_size(5).build().unwrap();
1539        let factory = Arc::new(MockConnectionFactory);
1540        let pool = Pool::new(config, factory).unwrap();
1541
1542        // 创建 3 个连接并归还到池中
1543        let conn1 = pool.acquire().await.unwrap();
1544        let conn2 = pool.acquire().await.unwrap();
1545        let conn3 = pool.acquire().await.unwrap();
1546        pool.release(conn1).await;
1547        pool.release(conn2).await;
1548        pool.release(conn3).await;
1549
1550        let removed = pool.health_check().await;
1551        assert_eq!(removed, 0, "Healthy connections should not be removed");
1552
1553        let status = pool.status().await;
1554        assert_eq!(status.idle, 3);
1555        assert_eq!(status.active, 3);
1556    }
1557
1558    #[tokio::test]
1559    async fn test_m7_health_check_returns_zero_for_empty_pool() {
1560        let config = PoolConfigBuilder::new().max_size(5).build().unwrap();
1561        let factory = Arc::new(MockConnectionFactory);
1562        let pool = Pool::new(config, factory).unwrap();
1563
1564        let removed = pool.health_check().await;
1565        assert_eq!(removed, 0);
1566    }
1567
1568    // ==================== 生产 Bug 复现测试 ====================
1569
1570    /// 可追踪创建次数的连接工厂
1571    struct CountingFactory {
1572        count: AtomicU32,
1573    }
1574
1575    impl CountingFactory {
1576        fn new() -> Self {
1577            Self {
1578                count: AtomicU32::new(0),
1579            }
1580        }
1581        fn created_count(&self) -> u32 {
1582            self.count.load(Ordering::SeqCst)
1583        }
1584    }
1585
1586    #[async_trait]
1587    impl ConnectionFactory for CountingFactory {
1588        async fn create(&self) -> Result<Box<dyn Connection>, crate::DbError> {
1589            self.count.fetch_add(1, Ordering::SeqCst);
1590            Ok(Box::new(MockConnection::new()))
1591        }
1592    }
1593
1594    /// 生产 Bug 复现：release() 重置 created_at 导致连接永不过期
1595    ///
1596    /// 症状：生产环境运行 30 分钟后间歇性 "connection timeout"
1597    /// 根因：release() 中 created_at 被重置为 now()，max_lifetime 检查永远不触发
1598    /// 期望：超过 max_lifetime 的连接应被回收并创建新连接
1599    #[tokio::test]
1600    async fn test_production_bug_max_lifetime_never_expires() {
1601        // 注意：PoolConfigBuilder::max_lifetime() 接受秒，这里需要毫秒级精度
1602        // 所以直接构造 PoolConfig
1603        let config = PoolConfig {
1604            max_size: 5,
1605            min_idle: 0,
1606            acquire_timeout: Duration::from_secs(30),
1607            idle_timeout: Duration::from_secs(600),
1608            max_lifetime: Duration::from_millis(100), // 100ms
1609            connection_timeout: Duration::from_secs(10),
1610            tls: None,
1611            query_timeout: None,
1612            max_rows: None,
1613            memory_limit: None,
1614            on_event: None,
1615        };
1616        let factory = Arc::new(CountingFactory::new());
1617        let pool = Pool::new(config, factory.clone()).unwrap();
1618
1619        // 1. 创建连接
1620        let conn = pool.acquire().await.unwrap();
1621        assert_eq!(factory.created_count(), 1, "应创建 1 个连接");
1622
1623        // 2. 归还连接（bug：重置 created_at）
1624        pool.release(conn).await;
1625
1626        // 3. 等待超过 max_lifetime
1627        tokio::time::sleep(Duration::from_millis(150)).await;
1628
1629        // 4. 再次获取 — 应检测到连接过期，创建新连接
1630        let conn2 = pool.acquire().await.unwrap();
1631
1632        // 5. 验证：如果 bug 存在，factory.created_count() 仍为 1（连接被复用，未过期）
1633        //         如果修复，factory.created_count() 应为 2（旧连接过期，创建新连接）
1634        assert_eq!(
1635            factory.created_count(),
1636            2,
1637            "超过 max_lifetime 后应创建新连接（旧连接应被回收）"
1638        );
1639
1640        pool.release(conn2).await;
1641    }
1642
1643    // ==================== PooledConnection::Drop 自动归还测试 ====================
1644
1645    /// 验证 PooledConnection drop 时自动归还连接到池
1646    ///
1647    /// 修复前：PooledConnection 未实现 Drop，drop 时连接丢失，池耗尽
1648    /// 修复后：Drop 时 spawn 异步 release，连接自动归还
1649    #[tokio::test]
1650    async fn test_drop_auto_release_connection() {
1651        let config = PoolConfigBuilder::new().max_size(2).build().unwrap();
1652        let factory = Arc::new(CountingFactory::new());
1653        let pool = Pool::new(config, factory.clone()).unwrap();
1654
1655        // 1. acquire 一个连接（不显式 release）
1656        {
1657            let _conn = pool.acquire().await.unwrap();
1658            assert_eq!(factory.created_count(), 1, "应创建 1 个连接");
1659            let status = pool.status().await;
1660            assert_eq!(status.active, 1, "active 应为 1");
1661            assert_eq!(status.idle, 0, "idle 应为 0");
1662            // _conn 在此 drop
1663        }
1664
1665        // 2. 等待 Drop spawn 的异步 release 完成
1666        tokio::time::sleep(Duration::from_millis(50)).await;
1667
1668        // 3. 验证连接已自动归还到 idle 队列
1669        let status = pool.status().await;
1670        assert_eq!(status.idle, 1, "Drop 后连接应自动归还，idle 应为 1");
1671        assert_eq!(status.active, 1, "total_count 应为 1");
1672        assert_eq!(factory.created_count(), 1, "应复用归还的连接，不创建新连接");
1673    }
1674
1675    /// 验证 Drop 自动归还后，连接可被再次 acquire 复用
1676    #[tokio::test]
1677    async fn test_drop_auto_release_then_reuse() {
1678        let config = PoolConfigBuilder::new().max_size(1).build().unwrap();
1679        let factory = Arc::new(CountingFactory::new());
1680        let pool = Pool::new(config, factory.clone()).unwrap();
1681
1682        // max_size=1，如果 Drop 不归还，第二次 acquire 会超时
1683        {
1684            let _conn = pool.acquire().await.unwrap();
1685        }
1686
1687        // 等待 Drop spawn 的 release 完成
1688        tokio::time::sleep(Duration::from_millis(50)).await;
1689
1690        // 再次 acquire 应复用归还的连接，不创建新连接
1691        let conn = pool.acquire().await.expect("应能复用 Drop 归还的连接");
1692        assert_eq!(factory.created_count(), 1, "应复用归还的连接，不创建新连接");
1693
1694        pool.release(conn).await;
1695    }
1696
1697    /// 验证 into_inner 后 Drop 不归还（连接被消费）
1698    #[tokio::test]
1699    async fn test_into_inner_does_not_return_to_pool() {
1700        let config = PoolConfigBuilder::new().max_size(2).build().unwrap();
1701        let factory = Arc::new(CountingFactory::new());
1702        let pool = Pool::new(config, factory.clone()).unwrap();
1703
1704        let conn = pool.acquire().await.unwrap();
1705        assert_eq!(factory.created_count(), 1);
1706
1707        // into_inner 消费连接，pool 字段设为 None
1708        let _raw_conn = conn.into_inner();
1709
1710        // 等待一段时间，确保不会有 Drop spawn
1711        tokio::time::sleep(Duration::from_millis(50)).await;
1712
1713        let status = pool.status().await;
1714        assert_eq!(status.idle, 0, "into_inner 后连接不应归还");
1715        assert_eq!(status.active, 1, "total_count 仍为 1（连接被外部持有）");
1716    }
1717
1718    /// 验证显式 release 后 Drop 不会重复归还
1719    #[tokio::test]
1720    async fn test_explicit_release_no_double_return() {
1721        let config = PoolConfigBuilder::new().max_size(2).build().unwrap();
1722        let factory = Arc::new(CountingFactory::new());
1723        let pool = Pool::new(config, factory.clone()).unwrap();
1724
1725        let conn = pool.acquire().await.unwrap();
1726        pool.release(conn).await;
1727
1728        let status = pool.status().await;
1729        assert_eq!(status.idle, 1, "release 后 idle 应为 1");
1730
1731        // 再次 acquire + release 验证不会重复
1732        let conn = pool.acquire().await.unwrap();
1733        pool.release(conn).await;
1734
1735        let status = pool.status().await;
1736        assert_eq!(status.idle, 1, "再次 release 后 idle 仍应为 1（不重复）");
1737        assert_eq!(status.active, 1, "total_count 应为 1");
1738    }
1739}
sz_orm_core/pool.rs

sz_orm_core/
pool.rs
