1use crate::client::{LLMClient, Provider};
38use ares_types::types::{AppError, Result};
39use parking_lot::{Mutex, RwLock};
40use serde::{Deserialize, Serialize};
41use std::collections::HashMap;
42use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
43use std::sync::Arc;
44use std::time::{Duration, Instant};
45use tokio::sync::{OwnedSemaphorePermit, Semaphore};
46
47#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
49pub enum PoolError {
50 #[serde(rename = "pool_exhausted")]
51 PoolExhausted { max: usize },
52 #[serde(rename = "timeout")]
53 Timeout { timeout_ms: u64 },
54 #[serde(rename = "invalid_client")]
55 InvalidClient { reason: String },
56}
57
58impl std::fmt::Display for PoolError {
59 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
60 match self {
61 Self::PoolExhausted { max } => write!(f, "pool exhausted (max={max})"),
62 Self::Timeout { timeout_ms } => write!(f, "pool acquire timeout after {timeout_ms}ms"),
63 Self::InvalidClient { reason } => write!(f, "invalid pooled client: {reason}"),
64 }
65 }
66}
67
68impl std::error::Error for PoolError {}
69
70impl From<PoolError> for AppError {
71 fn from(err: PoolError) -> Self {
72 AppError::LLM(err.to_string())
73 }
74}
75
76mod duration_secs {
77 use serde::{Deserialize, Deserializer, Serializer};
78 use std::time::Duration;
79
80 pub fn serialize<S>(value: &Duration, serializer: S) -> std::result::Result<S::Ok, S::Error>
81 where
82 S: Serializer,
83 {
84 serializer.serialize_u64(value.as_secs())
85 }
86
87 pub fn deserialize<'de, D>(deserializer: D) -> std::result::Result<Duration, D::Error>
88 where
89 D: Deserializer<'de>,
90 {
91 Ok(Duration::from_secs(u64::deserialize(deserializer)?))
92 }
93}
94
95#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
97pub struct PoolConfig {
98 pub max_connections_per_provider: usize,
100
101 pub min_idle_connections: usize,
103
104 #[serde(with = "duration_secs")]
106 pub idle_timeout: Duration,
107
108 #[serde(with = "duration_secs")]
110 pub max_lifetime: Duration,
111
112 #[serde(with = "duration_secs")]
114 pub health_check_interval: Duration,
115
116 #[serde(with = "duration_secs")]
118 pub acquire_timeout: Duration,
119
120 pub enable_health_check: bool,
122}
123
124impl Default for PoolConfig {
125 fn default() -> Self {
126 Self {
127 max_connections_per_provider: 10,
128 min_idle_connections: 2,
129 idle_timeout: Duration::from_secs(300), max_lifetime: Duration::from_secs(1800), health_check_interval: Duration::from_secs(60),
132 acquire_timeout: Duration::from_secs(30),
133 enable_health_check: true,
134 }
135 }
136}
137
138impl PoolConfig {
139 pub fn with_max_connections(mut self, max: usize) -> Self {
141 self.max_connections_per_provider = max;
142 self
143 }
144
145 pub fn with_idle_timeout(mut self, timeout: Duration) -> Self {
147 self.idle_timeout = timeout;
148 self
149 }
150
151 pub fn with_max_lifetime(mut self, lifetime: Duration) -> Self {
153 self.max_lifetime = lifetime;
154 self
155 }
156
157 pub fn without_health_check(mut self) -> Self {
159 self.enable_health_check = false;
160 self
161 }
162}
163
164
165impl std::fmt::Display for PoolConfig {
166 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
167 write!(
168 f,
169 "PoolConfig(max={}, min_idle={}, health_check={})",
170 self.max_connections_per_provider,
171 self.min_idle_connections,
172 self.enable_health_check
173 )
174 }
175}
176
177#[derive(Debug)]
179struct PooledClientMeta {
180 created_at: Instant,
182 last_used: Instant,
184 #[allow(dead_code)] use_count: AtomicU64,
187}
188
189impl PooledClientMeta {
190 fn new() -> Self {
191 let now = Instant::now();
192 Self {
193 created_at: now,
194 last_used: now,
195 use_count: AtomicU64::new(0),
196 }
197 }
198
199 fn mark_used(&mut self) {
200 self.last_used = Instant::now();
201 self.use_count.fetch_add(1, Ordering::Relaxed);
202 }
203
204 fn is_stale(&self, config: &PoolConfig) -> bool {
205 let now = Instant::now();
206 let idle_duration = now.duration_since(self.last_used);
207 let lifetime = now.duration_since(self.created_at);
208
209 idle_duration > config.idle_timeout || lifetime > config.max_lifetime
210 }
211}
212
213
214#[derive(Debug)]
215enum BorrowFromIdle {
216 Found(PooledClient),
217 Exhausted,
218}
219
220#[derive(Debug, PartialEq, Eq, Clone, Copy)]
221pub enum ReturnDisposition {
222 Returned,
223 Dropped,
224}
225
226fn borrow_client(idle: &mut Vec<PooledClient>, config: &PoolConfig) -> BorrowFromIdle {
227 let mut found_idx = None;
228 for (idx, pooled) in idle.iter().enumerate() {
229 if health_check(&pooled.meta, config) {
230 found_idx = Some(idx);
231 break;
232 }
233 }
234 if let Some(idx) = found_idx {
235 return BorrowFromIdle::Found(idle.swap_remove(idx));
236 }
237 idle.retain(|c| health_check(&c.meta, config));
238 BorrowFromIdle::Exhausted
239}
240
241fn return_client(
242 idle: &mut Vec<PooledClient>,
243 client: Box<dyn LLMClient>,
244 config: &PoolConfig,
245) -> ReturnDisposition {
246 if idle.len() < config.max_connections_per_provider {
247 idle.push(PooledClient {
248 client,
249 meta: PooledClientMeta::new(),
250 });
251 ReturnDisposition::Returned
252 } else {
253 ReturnDisposition::Dropped
254 }
255}
256
257fn health_check(meta: &PooledClientMeta, config: &PoolConfig) -> bool {
258 if !config.enable_health_check {
259 return true;
260 }
261 !meta.is_stale(config)
262}
263
264fn pool_stats(
265 available: usize,
266 in_use: usize,
267 total_created: u64,
268 max_size: usize,
269 borrow_count: u64,
270 error_count: u64,
271) -> ProviderPoolStats {
272 ProviderPoolStats {
273 available,
274 in_use,
275 total: available.saturating_add(in_use),
276 total_created,
277 max_size,
278 borrow_count,
279 error_count,
280 }
281}
282
283fn validate_pooled_client(client: &dyn LLMClient) -> std::result::Result<(), PoolError> {
284 if client.model_name().trim().is_empty() {
285 return Err(PoolError::InvalidClient {
286 reason: "empty model name".to_string(),
287 });
288 }
289 Ok(())
290}
291
292struct PooledClient {
294 client: Box<dyn LLMClient>,
295 meta: PooledClientMeta,
296}
297
298impl std::fmt::Debug for PooledClient {
299 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
300 f.debug_struct("PooledClient")
301 .field("meta", &self.meta)
302 .finish()
303 }
304}
305
306#[derive(Debug)]
308struct ProviderPool {
309 provider: Provider,
311 clients: Mutex<Vec<PooledClient>>,
313 semaphore: Arc<Semaphore>,
315 in_use_count: AtomicUsize,
317 total_created: AtomicU64,
319 borrow_count: AtomicU64,
320 error_count: AtomicU64,
321 config: PoolConfig,
323}
324
325impl ProviderPool {
326 fn new(provider: Provider, config: PoolConfig) -> Self {
327 let semaphore = Arc::new(Semaphore::new(config.max_connections_per_provider));
328 Self {
329 provider,
330 clients: Mutex::new(Vec::with_capacity(config.max_connections_per_provider)),
331 semaphore,
332 in_use_count: AtomicUsize::new(0),
333 total_created: AtomicU64::new(0),
334 borrow_count: AtomicU64::new(0),
335 error_count: AtomicU64::new(0),
336 config,
337 }
338 }
339
340 async fn acquire(&self) -> std::result::Result<(Box<dyn LLMClient>, OwnedSemaphorePermit), PoolError> {
342 let permit = match tokio::time::timeout(
343 self.config.acquire_timeout,
344 self.semaphore.clone().acquire_owned(),
345 )
346 .await
347 {
348 Ok(Ok(permit)) => permit,
349 Ok(Err(_)) => {
350 self.error_count.fetch_add(1, Ordering::Relaxed);
351 return Err(PoolError::PoolExhausted {
352 max: self.config.max_connections_per_provider,
353 });
354 }
355 Err(_) => {
356 self.error_count.fetch_add(1, Ordering::Relaxed);
357 return Err(PoolError::Timeout {
358 timeout_ms: self.config.acquire_timeout.as_millis() as u64,
359 });
360 }
361 };
362
363 self.borrow_count.fetch_add(1, Ordering::Relaxed);
364
365 let client = {
366 let borrowed = {
367 let mut clients = self.clients.lock();
368 match borrow_client(&mut clients, &self.config) {
369 BorrowFromIdle::Found(mut pooled) => {
370 validate_pooled_client(pooled.client.as_ref())?;
371 pooled.meta.mark_used();
372 Ok(pooled.client)
373 }
374 BorrowFromIdle::Exhausted => Err(()),
375 }
376 };
377 match borrowed {
378 Ok(client) => client,
379 Err(()) => {
380 self.total_created.fetch_add(1, Ordering::Relaxed);
381 let created = self.provider.create_client().await.map_err(|e| {
382 self.error_count.fetch_add(1, Ordering::Relaxed);
383 PoolError::InvalidClient {
384 reason: e.to_string(),
385 }
386 })?;
387 validate_pooled_client(created.as_ref())?;
388 created
389 }
390 }
391 };
392
393 self.in_use_count.fetch_add(1, Ordering::Relaxed);
394 Ok((client, permit))
395 }
396
397 async fn try_acquire(&self) -> std::result::Result<(Box<dyn LLMClient>, OwnedSemaphorePermit), PoolError> {
398 let permit = self.semaphore.clone().try_acquire_owned().map_err(|_| {
399 self.error_count.fetch_add(1, Ordering::Relaxed);
400 PoolError::PoolExhausted {
401 max: self.config.max_connections_per_provider,
402 }
403 })?;
404
405 self.borrow_count.fetch_add(1, Ordering::Relaxed);
406
407 let client = {
408 let borrowed = {
409 let mut clients = self.clients.lock();
410 match borrow_client(&mut clients, &self.config) {
411 BorrowFromIdle::Found(mut pooled) => {
412 validate_pooled_client(pooled.client.as_ref())?;
413 pooled.meta.mark_used();
414 Ok(pooled.client)
415 }
416 BorrowFromIdle::Exhausted => Err(()),
417 }
418 };
419 match borrowed {
420 Ok(client) => client,
421 Err(()) => {
422 self.total_created.fetch_add(1, Ordering::Relaxed);
423 let created = self.provider.create_client().await.map_err(|e| {
424 self.error_count.fetch_add(1, Ordering::Relaxed);
425 PoolError::InvalidClient {
426 reason: e.to_string(),
427 }
428 })?;
429 validate_pooled_client(created.as_ref())?;
430 created
431 }
432 }
433 };
434
435 self.in_use_count.fetch_add(1, Ordering::Relaxed);
436 Ok((client, permit))
437 }
438
439 fn release(&self, client: Box<dyn LLMClient>) {
441 self.in_use_count.fetch_sub(1, Ordering::Relaxed);
442 let mut clients = self.clients.lock();
443 let _ = return_client(&mut clients, client, &self.config);
444 }
445
446 fn cleanup_stale(&self) -> usize {
448 let mut clients = self.clients.lock();
449 let before = clients.len();
450 clients.retain(|c| !c.meta.is_stale(&self.config));
451 before - clients.len()
452 }
453
454 fn stats(&self) -> ProviderPoolStats {
456 let clients = self.clients.lock();
457 pool_stats(
458 clients.len(),
459 self.in_use_count.load(Ordering::Relaxed),
460 self.total_created.load(Ordering::Relaxed),
461 self.config.max_connections_per_provider,
462 self.borrow_count.load(Ordering::Relaxed),
463 self.error_count.load(Ordering::Relaxed),
464 )
465 }
466
467 fn drain(&self) {
469 let mut clients = self.clients.lock();
470 clients.clear();
471 }
472}
473
474#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
476pub struct ProviderPoolStats {
477 pub available: usize,
478 pub in_use: usize,
479 pub total: usize,
480 pub total_created: u64,
481 pub max_size: usize,
482 pub borrow_count: u64,
483 pub error_count: u64,
484}
485
486impl std::fmt::Display for ProviderPoolStats {
487 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
488 write!(
489 f,
490 "idle={} active={} total={} created={} max={} borrows={} errors={}",
491 self.available,
492 self.in_use,
493 self.total,
494 self.total_created,
495 self.max_size,
496 self.borrow_count,
497 self.error_count
498 )
499 }
500}
501
502#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
504pub struct PoolStats {
505 pub providers: HashMap<String, ProviderPoolStats>,
506 pub total_available: usize,
507 pub total_in_use: usize,
508 pub total_connections: usize,
509 pub borrow_count: u64,
510 pub error_count: u64,
511}
512
513impl std::fmt::Display for PoolStats {
514 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
515 write!(
516 f,
517 "providers={} idle={} active={} total={} borrows={} errors={}",
518 self.providers.len(),
519 self.total_available,
520 self.total_in_use,
521 self.total_connections,
522 self.borrow_count,
523 self.error_count
524 )
525 }
526}
527
528pub struct PooledClientGuard {
530 client: Option<Box<dyn LLMClient>>,
531 pool: Arc<ProviderPool>,
532 _permit: OwnedSemaphorePermit,
533}
534
535impl std::fmt::Debug for PooledClientGuard {
536 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
537 f.debug_struct("PooledClientGuard")
538 .field("has_client", &self.client.is_some())
539 .field("pool", &self.pool)
540 .finish()
541 }
542}
543
544impl PooledClientGuard {
545 pub fn client(&self) -> &dyn LLMClient {
547 self.client.as_ref().expect("Client already taken").as_ref()
548 }
549
550 pub fn client_mut(&mut self) -> &mut dyn LLMClient {
552 self.client.as_mut().expect("Client already taken").as_mut()
553 }
554
555 pub fn take(mut self) -> Box<dyn LLMClient> {
560 self.client.take().expect("Client already taken")
561 }
562}
563
564impl Drop for PooledClientGuard {
565 fn drop(&mut self) {
566 if let Some(client) = self.client.take() {
567 self.pool.release(client);
568 }
569 }
570}
571
572impl std::ops::Deref for PooledClientGuard {
573 type Target = Box<dyn LLMClient>;
574
575 fn deref(&self) -> &Self::Target {
576 self.client.as_ref().expect("Client already taken")
577 }
578}
579
580
581#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
582pub struct LLMPoolSnapshot {
583 pub config: PoolConfig,
584 pub stats: PoolStats,
585 pub providers: Vec<String>,
586 pub shutdown: bool,
587}
588
589pub type LLMPool = ClientPool;
590
591pub struct ClientPool {
596 config: PoolConfig,
597 providers: RwLock<HashMap<String, Arc<ProviderPool>>>,
598 shutdown: std::sync::atomic::AtomicBool,
599}
600
601impl ClientPool {
602 pub fn new(config: PoolConfig) -> Self {
604 Self {
605 config,
606 providers: RwLock::new(HashMap::new()),
607 shutdown: std::sync::atomic::AtomicBool::new(false),
608 }
609 }
610
611 pub fn with_defaults() -> Self {
613 Self::new(PoolConfig::default())
614 }
615
616 #[allow(unreachable_code, unused_variables)]
621 pub fn register_provider(&self, name: &str, provider: Provider) {
622 let pool = Arc::new(ProviderPool::new(provider, self.config.clone()));
623 let mut providers = self.providers.write();
624 providers.insert(name.to_string(), pool);
625 }
626
627 pub fn has_provider(&self, name: &str) -> bool {
629 self.providers.read().contains_key(name)
630 }
631
632 pub fn provider_names(&self) -> Vec<String> {
634 self.providers.read().keys().cloned().collect()
635 }
636
637 pub async fn get(&self, provider_name: &str) -> Result<PooledClientGuard> {
642 self.get_with_error(provider_name).await.map_err(Into::into)
643 }
644
645 pub async fn get_with_error(
646 &self,
647 provider_name: &str,
648 ) -> std::result::Result<PooledClientGuard, PoolError> {
649 if self.shutdown.load(Ordering::Relaxed) {
650 return Err(PoolError::InvalidClient {
651 reason: "pool is shutting down".to_string(),
652 });
653 }
654
655 let pool = {
656 let providers = self.providers.read();
657 providers.get(provider_name).cloned().ok_or_else(|| PoolError::InvalidClient {
658 reason: format!("provider '{provider_name}' not registered in pool"),
659 })?
660 };
661
662 let (client, permit) = pool.acquire().await?;
663
664 Ok(PooledClientGuard {
665 client: Some(client),
666 pool,
667 _permit: permit,
668 })
669 }
670
671 pub async fn try_get(
672 &self,
673 provider_name: &str,
674 ) -> std::result::Result<PooledClientGuard, PoolError> {
675 if self.shutdown.load(Ordering::Relaxed) {
676 return Err(PoolError::InvalidClient {
677 reason: "pool is shutting down".to_string(),
678 });
679 }
680
681 let pool = {
682 let providers = self.providers.read();
683 providers.get(provider_name).cloned().ok_or_else(|| PoolError::InvalidClient {
684 reason: format!("provider '{provider_name}' not registered in pool"),
685 })?
686 };
687
688 let (client, permit) = pool.try_acquire().await?;
689
690 Ok(PooledClientGuard {
691 client: Some(client),
692 pool,
693 _permit: permit,
694 })
695 }
696
697 pub fn stats(&self) -> PoolStats {
699 let providers = self.providers.read();
700 let mut stats = PoolStats {
701 providers: HashMap::new(),
702 total_available: 0,
703 total_in_use: 0,
704 total_connections: 0,
705 borrow_count: 0,
706 error_count: 0,
707 };
708
709 for (name, pool) in providers.iter() {
710 let provider_stats = pool.stats();
711 stats.total_available += provider_stats.available;
712 stats.total_in_use += provider_stats.in_use;
713 stats.total_connections += provider_stats.total;
714 stats.borrow_count += provider_stats.borrow_count;
715 stats.error_count += provider_stats.error_count;
716 stats.providers.insert(name.clone(), provider_stats);
717 }
718
719 stats
720 }
721
722 pub fn cleanup_stale(&self) -> usize {
726 let providers = self.providers.read();
727 providers.values().map(|p| p.cleanup_stale()).sum()
728 }
729
730 pub fn start_cleanup_task(self: &Arc<Self>) -> tokio::task::JoinHandle<()> {
734 let pool = Arc::clone(self);
735 let interval = pool.config.health_check_interval;
736
737 tokio::spawn(async move {
738 let mut interval_timer = tokio::time::interval(interval);
739 loop {
740 interval_timer.tick().await;
741
742 if pool.shutdown.load(Ordering::Relaxed) {
743 break;
744 }
745
746 let removed = pool.cleanup_stale();
747 if removed > 0 {
748 tracing::debug!("Pool cleanup: removed {} stale connections", removed);
749 }
750 }
751 })
752 }
753
754 pub fn shutdown(&self) {
759 self.shutdown.store(true, Ordering::Relaxed);
760
761 let providers = self.providers.read();
762 for pool in providers.values() {
763 pool.drain();
764 }
765 }
766
767 pub fn is_shutdown(&self) -> bool {
769 self.shutdown.load(Ordering::Relaxed)
770 }
771
772 pub fn snapshot(&self) -> LLMPoolSnapshot {
773 LLMPoolSnapshot {
774 config: self.config.clone(),
775 stats: self.stats(),
776 providers: self.provider_names(),
777 shutdown: self.is_shutdown(),
778 }
779 }
780}
781
782
783impl std::fmt::Display for ClientPool {
784 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
785 let snap = self.snapshot();
786 write!(f, "LLMPool(shutdown={}, {})", snap.shutdown, snap.stats)
787 }
788}
789
790impl Default for ClientPool {
791 fn default() -> Self {
792 Self::with_defaults()
793 }
794}
795
796pub struct ClientPoolBuilder {
798 config: PoolConfig,
799 providers: Vec<(String, Provider)>,
800}
801
802impl ClientPoolBuilder {
803 pub fn new() -> Self {
805 Self {
806 config: PoolConfig::default(),
807 providers: Vec::new(),
808 }
809 }
810
811 pub fn config(mut self, config: PoolConfig) -> Self {
813 self.config = config;
814 self
815 }
816
817 pub fn provider(mut self, name: impl Into<String>, provider: Provider) -> Self {
819 self.providers.push((name.into(), provider));
820 self
821 }
822
823 pub fn build(self) -> ClientPool {
825 let pool = ClientPool::new(self.config);
826 for (name, provider) in self.providers {
827 pool.register_provider(&name, provider);
828 }
829 pool
830 }
831
832 pub fn build_arc(self) -> Arc<ClientPool> {
834 Arc::new(self.build())
835 }
836}
837
838impl Default for ClientPoolBuilder {
839 fn default() -> Self {
840 Self::new()
841 }
842}
843
844#[cfg(test)]
845mod tests {
846 use super::*;
847 use crate::client::test_support::MockLLMClient;
848 use std::sync::atomic::Ordering as AtomicOrdering;
849 fn mock_client(model: &str) -> Box<dyn LLMClient> {
850 Box::new(MockLLMClient::new(model))
851 }
852
853 fn test_stub_provider() -> Provider {
854 Provider::TestStub {
855 model: "mock".to_string(),
856 }
857 }
858
859 fn provider_pool(config: PoolConfig) -> Arc<ProviderPool> {
860 Arc::new(ProviderPool::new(test_stub_provider(), config))
861 }
862
863 fn seed_pool(pool: &ProviderPool, clients: Vec<Box<dyn LLMClient>>) {
864 let mut guard = pool.clients.lock();
865 for client in clients {
866 guard.push(PooledClient {
867 client,
868 meta: PooledClientMeta::new(),
869 });
870 }
871 }
872
873 fn register_seeded_pool(
874 client_pool: &ClientPool,
875 name: &str,
876 config: PoolConfig,
877 clients: Vec<Box<dyn LLMClient>>,
878 ) {
879 let sub = provider_pool(config);
880 seed_pool(&sub, clients);
881 client_pool.providers.write().insert(name.to_string(), sub);
882 }
883
884 #[test]
885 fn test_pool_config_defaults() {
886 let config = PoolConfig::default();
887 assert_eq!(config.max_connections_per_provider, 10);
888 assert_eq!(config.min_idle_connections, 2);
889 assert_eq!(config.idle_timeout, Duration::from_secs(300));
890 assert_eq!(config.max_lifetime, Duration::from_secs(1800));
891 assert_eq!(config.health_check_interval, Duration::from_secs(60));
892 assert_eq!(config.acquire_timeout, Duration::from_secs(30));
893 assert!(config.enable_health_check);
894 }
895
896 #[test]
897 fn test_pool_config_builder() {
898 let config = PoolConfig::default()
899 .with_max_connections(20)
900 .with_idle_timeout(Duration::from_secs(60))
901 .without_health_check();
902
903 assert_eq!(config.max_connections_per_provider, 20);
904 assert_eq!(config.idle_timeout, Duration::from_secs(60));
905 assert!(!config.enable_health_check);
906 }
907
908 #[test]
909 fn test_pool_config_with_max_lifetime() {
910 let config = PoolConfig::default().with_max_lifetime(Duration::from_secs(42));
911 assert_eq!(config.max_lifetime, Duration::from_secs(42));
912 }
913
914 #[test]
915 fn test_pool_config_clone_preserves_fields() {
916 let config = PoolConfig::default()
917 .with_max_connections(7)
918 .with_idle_timeout(Duration::from_secs(11))
919 .with_max_lifetime(Duration::from_secs(22))
920 .without_health_check();
921 let cloned = config.clone();
922 assert_eq!(cloned.max_connections_per_provider, 7);
923 assert_eq!(cloned.idle_timeout, Duration::from_secs(11));
924 assert_eq!(cloned.max_lifetime, Duration::from_secs(22));
925 assert!(!cloned.enable_health_check);
926 }
927
928 #[test]
929 fn test_pool_config_debug_format() {
930 let debug = format!("{:?}", PoolConfig::default().with_max_connections(3));
931 assert!(debug.contains("max_connections_per_provider"));
932 assert!(debug.contains('3'));
933 }
934
935 #[test]
936 fn test_pool_config_max_lifetime_stale() {
937 let config = PoolConfig::default()
938 .with_idle_timeout(Duration::from_millis(1))
939 .with_max_lifetime(Duration::from_millis(5));
940 let meta = PooledClientMeta::new();
941 std::thread::sleep(Duration::from_millis(6));
942 assert!(meta.is_stale(&config));
943 }
944
945 #[test]
946 fn test_pooled_client_meta_stale_detection() {
947 let config = PoolConfig::default()
948 .with_idle_timeout(Duration::from_millis(10))
949 .with_max_lifetime(Duration::from_millis(50));
950
951 let meta = PooledClientMeta::new();
952 assert!(!meta.is_stale(&config));
953
954 std::thread::sleep(Duration::from_millis(15));
955 assert!(meta.is_stale(&config));
956 }
957
958 #[test]
959 fn test_pooled_client_meta_mark_used_increments() {
960 let mut meta = PooledClientMeta::new();
961 meta.mark_used();
962 meta.mark_used();
963 assert_eq!(meta.use_count.load(AtomicOrdering::Relaxed), 2);
964 }
965
966 #[test]
967 fn test_pooled_client_meta_mark_used_refreshes_idle_timer() {
968 let config = PoolConfig::default().with_idle_timeout(Duration::from_millis(30));
969 let mut meta = PooledClientMeta::new();
970 std::thread::sleep(Duration::from_millis(20));
971 meta.mark_used();
972 assert!(!meta.is_stale(&config));
973 }
974
975 #[test]
976 fn test_pooled_client_debug_output() {
977 let pooled = PooledClient {
978 client: mock_client("debug-model"),
979 meta: PooledClientMeta::new(),
980 };
981 let debug = format!("{pooled:?}");
982 assert!(debug.contains("PooledClient"));
983 assert!(debug.contains("meta"));
984 }
985
986 #[test]
987 fn test_provider_pool_stats_clone_and_debug() {
988 let stats = pool_stats(2, 1, 5, 10, 3, 1);
989 let cloned = stats.clone();
990 assert_eq!(cloned.available, 2);
991 assert_eq!(cloned.in_use, 1);
992 let debug = format!("{stats:?}");
993 assert!(debug.contains("available"));
994 }
995
996 #[test]
997 fn test_pool_stats_clone_and_debug() {
998 let mut providers = HashMap::new();
999 providers.insert(
1000 "p1".to_string(),
1001 pool_stats(1, 0, 1, 3, 0, 0),
1002 );
1003 let stats = PoolStats {
1004 providers,
1005 total_available: 1,
1006 total_in_use: 0,
1007 total_connections: 1,
1008 borrow_count: 0,
1009 error_count: 0,
1010 };
1011 let cloned = stats.clone();
1012 assert_eq!(cloned.total_available, 1);
1013 assert_eq!(cloned.providers.len(), 1);
1014 assert!(format!("{stats:?}").contains("total_in_use"));
1015 }
1016
1017 #[test]
1018 fn test_client_pool_default_impl() {
1019 let pool = ClientPool::default();
1020 assert!(!pool.is_shutdown());
1021 assert_eq!(pool.stats().total_available, 0);
1022 }
1023
1024 #[test]
1025 fn test_client_pool_builder_default_impl() {
1026 let builder = ClientPoolBuilder::default();
1027 let pool = builder.build();
1028 assert!(!pool.has_provider("anything"));
1029 }
1030
1031 #[test]
1032 fn test_builder_empty_build() {
1033 let pool = ClientPoolBuilder::new().build();
1034 assert!(pool.provider_names().is_empty());
1035 assert!(!pool.has_provider("missing"));
1036 }
1037
1038 #[test]
1039 fn test_builder_build_arc() {
1040 let pool = ClientPoolBuilder::new()
1041 .config(PoolConfig::default().with_max_connections(4))
1042 .build_arc();
1043 assert_eq!(pool.stats().total_available, 0);
1044 assert!(!pool.is_shutdown());
1045 }
1046
1047 #[test]
1048 fn test_pool_stats() {
1049 let pool = ClientPool::with_defaults();
1050 let stats = pool.stats();
1051 assert_eq!(stats.total_available, 0);
1052 assert_eq!(stats.total_in_use, 0);
1053 assert!(stats.providers.is_empty());
1054 }
1055
1056 #[test]
1057 fn test_cleanup_stale_on_empty_pool() {
1058 let pool = ClientPool::with_defaults();
1059 assert_eq!(pool.cleanup_stale(), 0);
1060 }
1061
1062 #[test]
1063 fn test_pool_shutdown() {
1064 let pool = ClientPool::with_defaults();
1065 assert!(!pool.is_shutdown());
1066 pool.shutdown();
1067 assert!(pool.is_shutdown());
1068 }
1069
1070 #[test]
1071 fn test_pool_double_shutdown_is_safe() {
1072 let pool = ClientPool::with_defaults();
1073 pool.shutdown();
1074 pool.shutdown();
1075 assert!(pool.is_shutdown());
1076 }
1077
1078 #[test]
1079 fn test_provider_registration() {
1080 let pool = ClientPool::with_defaults();
1081 pool.register_provider("ollama", test_stub_provider());
1082 assert!(pool.has_provider("ollama"));
1083 assert!(!pool.has_provider("openai"));
1084 assert_eq!(pool.provider_names(), vec!["ollama"]);
1085 }
1086
1087 #[test]
1088 fn test_builder_pattern() {
1089 let pool = ClientPoolBuilder::new()
1090 .config(PoolConfig::default().with_max_connections(5))
1091 .provider("ollama", test_stub_provider())
1092 .build();
1093 assert!(pool.has_provider("ollama"));
1094 }
1095
1096 #[test]
1097 fn test_builder_multiple_providers() {
1098 let pool = ClientPoolBuilder::new()
1099 .provider("a", test_stub_provider())
1100 .provider("b", test_stub_provider())
1101 .build();
1102 let mut names = pool.provider_names();
1103 names.sort();
1104 assert_eq!(names, vec!["a", "b"]);
1105 }
1106
1107 #[tokio::test]
1108 async fn test_get_unregistered_provider_error() {
1109 let pool = ClientPool::with_defaults();
1110 let result = pool.get("nonexistent").await;
1111 assert!(result.is_err());
1112 assert!(matches!(result.unwrap_err(), AppError::LLM(_)));
1113 }
1114
1115 #[tokio::test]
1116 async fn test_acquire_reuses_seeded_mock_without_network() {
1117 let config = PoolConfig::default()
1118 .with_max_connections(2)
1119 .without_health_check();
1120 let pool = ClientPool::new(config.clone());
1121 register_seeded_pool(&pool, "mock", config, vec![mock_client("seeded")]);
1122
1123 let guard = pool.get("mock").await.expect("seeded client should be available");
1124 assert_eq!(guard.client().model_name(), "seeded");
1125 assert_eq!(pool.stats().providers["mock"].total_created, 0);
1126 }
1127
1128 #[tokio::test]
1129 async fn test_acquire_skips_stale_prefers_first_fresh() {
1130 let config = PoolConfig::default()
1131 .with_idle_timeout(Duration::from_millis(5))
1132 .with_max_lifetime(Duration::from_secs(60));
1133 let sub = provider_pool(config);
1134
1135 {
1136 let mut guard = sub.clients.lock();
1137 guard.push(PooledClient {
1138 client: mock_client("stale"),
1139 meta: PooledClientMeta::new(),
1140 });
1141 std::thread::sleep(Duration::from_millis(8));
1142 guard.push(PooledClient {
1143 client: mock_client("fresh"),
1144 meta: PooledClientMeta::new(),
1145 });
1146 }
1147
1148 let (client, permit) = sub.acquire().await.expect("acquire fresh client");
1149 assert_eq!(client.model_name(), "fresh");
1150 drop(client);
1151 drop(permit);
1152 }
1153
1154 #[tokio::test]
1155 async fn test_release_and_reacquire_reuses_pooled_client() {
1156 let config = PoolConfig::default()
1157 .with_max_connections(1)
1158 .without_health_check();
1159 let pool = ClientPool::new(config.clone());
1160 register_seeded_pool(&pool, "mock", config, vec![mock_client("reusable")]);
1161
1162 {
1163 let guard = pool.get("mock").await.expect("first acquire");
1164 assert_eq!(guard.client().model_name(), "reusable");
1165 }
1166
1167 let guard = pool.get("mock").await.expect("second acquire");
1168 assert_eq!(guard.client().model_name(), "reusable");
1169 let stats = pool.stats().providers["mock"].clone();
1170 assert_eq!(stats.total_created, 0);
1171 assert_eq!(stats.in_use, 1);
1172 }
1173
1174 #[tokio::test]
1175 async fn test_release_drops_client_when_idle_pool_full() {
1176 let config = PoolConfig::default()
1177 .with_max_connections(1)
1178 .without_health_check();
1179 let sub = provider_pool(config);
1180 seed_pool(&sub, vec![mock_client("idle-0")]);
1181
1182 let (client, permit) = sub.acquire().await.expect("acquire seeded client");
1183 sub.release(client);
1184 drop(permit);
1185 assert_eq!(sub.stats().available, 1);
1186 assert_eq!(sub.stats().in_use, 0);
1187
1188 let (overflow, permit2) = sub.acquire().await.expect("acquire again");
1189 sub.release(overflow);
1190 drop(permit2);
1191
1192 let stats = sub.stats();
1193 assert_eq!(stats.available, 1);
1194 assert_eq!(stats.in_use, 0);
1195 assert_eq!(stats.total_created, 0);
1196 }
1197
1198 #[tokio::test]
1199 async fn test_provider_pool_in_use_accounting() {
1200 let config = PoolConfig::default()
1201 .with_max_connections(2)
1202 .without_health_check();
1203 let sub = provider_pool(config);
1204 seed_pool(&sub, vec![mock_client("a")]);
1205
1206 let (_client, permit) = sub.acquire().await.expect("acquire");
1207 let stats = sub.stats();
1208 assert_eq!(stats.in_use, 1);
1209 assert_eq!(stats.available, 0);
1210 drop(permit);
1211 }
1212
1213 #[tokio::test]
1214 async fn test_cleanup_stale_removes_idle_clients() {
1215 let config = PoolConfig::default()
1216 .with_idle_timeout(Duration::from_millis(5))
1217 .without_health_check();
1218 let sub = provider_pool(config);
1219 {
1220 let mut guard = sub.clients.lock();
1221 guard.push(PooledClient {
1222 client: mock_client("old"),
1223 meta: PooledClientMeta::new(),
1224 });
1225 std::thread::sleep(Duration::from_millis(8));
1226 }
1227 let removed = sub.cleanup_stale();
1228 assert_eq!(removed, 1);
1229 assert_eq!(sub.stats().available, 0);
1230 }
1231
1232 #[tokio::test]
1233 async fn test_pooled_client_guard_debug_and_deref() {
1234 let config = PoolConfig::default()
1235 .with_max_connections(1)
1236 .without_health_check();
1237 let pool = ClientPool::new(config.clone());
1238 register_seeded_pool(&pool, "mock", config, vec![mock_client("guard")]);
1239
1240 let guard = pool.get("mock").await.expect("guard acquire");
1241 let debug = format!("{guard:?}");
1242 assert!(debug.contains("PooledClientGuard"));
1243 assert!(debug.contains("has_client"));
1244 assert_eq!(guard.model_name(), "guard");
1245 }
1246
1247 #[test]
1248 fn test_register_provider_overwrites_existing_name() {
1249 let pool = ClientPool::with_defaults();
1250 pool.register_provider("ollama", test_stub_provider());
1251 pool.register_provider("ollama", test_stub_provider());
1252 assert_eq!(pool.provider_names(), vec!["ollama"]);
1253 }
1254
1255 #[tokio::test]
1256 async fn test_stats_aggregate_multiple_providers() {
1257 let config = PoolConfig::default()
1258 .with_max_connections(2)
1259 .without_health_check();
1260 let pool = ClientPool::new(config.clone());
1261 register_seeded_pool(&pool, "a", config.clone(), vec![mock_client("a1")]);
1262 register_seeded_pool(&pool, "b", config, vec![mock_client("b1")]);
1263
1264 let _ga = pool.get("a").await.expect("provider a");
1265 let stats = pool.stats();
1266 assert_eq!(stats.providers.len(), 2);
1267 assert_eq!(stats.total_in_use, 1);
1268 assert_eq!(stats.total_available, 1);
1269 }
1270
1271 #[tokio::test]
1272 async fn test_shutdown_drains_seeded_clients() {
1273 let config = PoolConfig::default()
1274 .with_max_connections(2)
1275 .without_health_check();
1276 let pool = ClientPool::new(config.clone());
1277 register_seeded_pool(&pool, "mock", config, vec![mock_client("seeded")]);
1278 assert_eq!(pool.stats().total_available, 1);
1279
1280 pool.shutdown();
1281 assert!(pool.is_shutdown());
1282 assert_eq!(pool.stats().total_available, 0);
1283 }
1284
1285
1286 #[tokio::test]
1287 async fn test_acquire_creates_client_when_pool_empty() {
1288 let config = PoolConfig::default()
1289 .with_max_connections(1)
1290 .without_health_check();
1291 let sub = provider_pool(config);
1292
1293 let (client, permit) = sub.acquire().await.expect("create via TestStub");
1294 assert_eq!(client.model_name(), "mock");
1295 drop(client);
1296 drop(permit);
1297 assert_eq!(sub.stats().total_created, 1);
1298 }
1299
1300 #[tokio::test]
1301 async fn test_pooled_client_guard_client_mut_and_take() {
1302 let config = PoolConfig::default()
1303 .with_max_connections(1)
1304 .without_health_check();
1305 let pool = ClientPool::new(config.clone());
1306 register_seeded_pool(&pool, "mock", config, vec![mock_client("guard-mut")]);
1307
1308 let mut guard = pool.get("mock").await.expect("guard acquire");
1309 assert_eq!(guard.client_mut().model_name(), "guard-mut");
1310 let taken = guard.take();
1311 assert_eq!(taken.model_name(), "guard-mut");
1312 }
1313
1314 fn serde_roundtrip<T>(value: &T) -> T
1315 where
1316 T: serde::Serialize + for<'de> serde::Deserialize<'de> + PartialEq + std::fmt::Debug,
1317 {
1318 let json = serde_json::to_string(value).unwrap();
1319 let parsed: T = serde_json::from_str(&json).unwrap();
1320 assert_eq!(*value, parsed);
1321 parsed
1322 }
1323
1324 #[test]
1325 fn test_pool_config_serde_roundtrip() {
1326 let config = PoolConfig::default()
1327 .with_max_connections(4)
1328 .with_idle_timeout(Duration::from_secs(90))
1329 .without_health_check();
1330 serde_roundtrip(&config);
1331 }
1332
1333 #[test]
1334 fn test_pool_stats_serde_roundtrip() {
1335 let mut providers = HashMap::new();
1336 providers.insert("mock".into(), pool_stats(1, 2, 3, 4, 5, 6));
1337 let stats = PoolStats {
1338 providers,
1339 total_available: 1,
1340 total_in_use: 2,
1341 total_connections: 3,
1342 borrow_count: 5,
1343 error_count: 6,
1344 };
1345 serde_roundtrip(&stats);
1346 }
1347
1348 #[test]
1349 fn test_provider_pool_stats_serde_roundtrip() {
1350 serde_roundtrip(&pool_stats(2, 1, 9, 10, 7, 2));
1351 }
1352
1353 #[test]
1354 fn test_llm_pool_snapshot_serde_roundtrip() {
1355 let pool = ClientPoolBuilder::new()
1356 .provider("mock", test_stub_provider())
1357 .build();
1358 serde_roundtrip(&pool.snapshot());
1359 }
1360
1361 #[test]
1362 fn test_llm_pool_type_alias() {
1363 let pool: LLMPool = ClientPool::with_defaults();
1364 assert_eq!(pool.stats().total_available, 0);
1365 }
1366
1367 #[test]
1368 fn test_pool_error_serde_roundtrip_exhausted() {
1369 serde_roundtrip(&PoolError::PoolExhausted { max: 3 });
1370 }
1371
1372 #[test]
1373 fn test_pool_error_serde_roundtrip_timeout() {
1374 serde_roundtrip(&PoolError::Timeout { timeout_ms: 250 });
1375 }
1376
1377 #[test]
1378 fn test_pool_error_serde_roundtrip_invalid_client() {
1379 serde_roundtrip(&PoolError::InvalidClient {
1380 reason: "bad".into(),
1381 });
1382 }
1383
1384 #[test]
1385 fn test_pool_error_display_variants() {
1386 assert!(PoolError::PoolExhausted { max: 2 }
1387 .to_string()
1388 .contains("pool exhausted"));
1389 assert!(PoolError::Timeout { timeout_ms: 10 }
1390 .to_string()
1391 .contains("timeout"));
1392 assert!(PoolError::InvalidClient {
1393 reason: "x".into()
1394 }
1395 .to_string()
1396 .contains("invalid"));
1397 }
1398
1399 #[test]
1400 fn test_pool_error_clone_debug() {
1401 let err = PoolError::PoolExhausted { max: 1 };
1402 assert_eq!(err, err.clone());
1403 assert!(format!("{err:?}").contains("PoolExhausted"));
1404 }
1405
1406 #[test]
1407 fn test_pool_config_display() {
1408 let s = PoolConfig::default().with_max_connections(6).to_string();
1409 assert!(s.contains("max=6"));
1410 }
1411
1412 #[test]
1413 fn test_pool_stats_display() {
1414 let stats = PoolStats {
1415 providers: HashMap::new(),
1416 total_available: 1,
1417 total_in_use: 2,
1418 total_connections: 3,
1419 borrow_count: 4,
1420 error_count: 5,
1421 };
1422 let s = stats.to_string();
1423 assert!(s.contains("idle=1"));
1424 assert!(s.contains("errors=5"));
1425 }
1426
1427 #[test]
1428 fn test_provider_pool_stats_display() {
1429 let s = pool_stats(1, 2, 0, 4, 9, 1).to_string();
1430 assert!(s.contains("borrows=9"));
1431 assert!(s.contains("total=3"));
1432 }
1433
1434 #[test]
1435 fn test_client_pool_display() {
1436 let pool = ClientPool::with_defaults();
1437 let s = pool.to_string();
1438 assert!(s.contains("LLMPool"));
1439 assert!(s.contains("shutdown=false"));
1440 }
1441
1442 #[test]
1443 fn test_borrow_client_returns_first_healthy() {
1444 let config = PoolConfig::default().without_health_check();
1445 let mut idle = vec![PooledClient {
1446 client: mock_client("a"),
1447 meta: PooledClientMeta::new(),
1448 }];
1449 match borrow_client(&mut idle, &config) {
1450 BorrowFromIdle::Found(p) => assert_eq!(p.client.model_name(), "a"),
1451 _ => panic!("expected found"),
1452 }
1453 assert!(idle.is_empty());
1454 }
1455
1456 #[test]
1457 fn test_borrow_client_purges_stale_entries() {
1458 let config = PoolConfig::default()
1459 .with_idle_timeout(Duration::from_millis(1));
1460 let mut idle = vec![PooledClient {
1461 client: mock_client("stale"),
1462 meta: PooledClientMeta::new(),
1463 }];
1464 std::thread::sleep(Duration::from_millis(3));
1465 assert!(matches!(borrow_client(&mut idle, &config), BorrowFromIdle::Exhausted));
1466 assert!(idle.is_empty());
1467 }
1468
1469 #[test]
1470 fn test_return_client_respects_capacity() {
1471 let config = PoolConfig::default()
1472 .with_max_connections(1)
1473 .without_health_check();
1474 let mut idle = vec![];
1475 assert_eq!(
1476 return_client(&mut idle, mock_client("a"), &config),
1477 ReturnDisposition::Returned
1478 );
1479 assert_eq!(
1480 return_client(&mut idle, mock_client("b"), &config),
1481 ReturnDisposition::Dropped
1482 );
1483 assert_eq!(idle.len(), 1);
1484 }
1485
1486 #[test]
1487 fn test_health_check_disabled_ignores_stale_meta() {
1488 let config = PoolConfig::default()
1489 .with_idle_timeout(Duration::from_millis(1))
1490 .without_health_check();
1491 let meta = PooledClientMeta::new();
1492 std::thread::sleep(Duration::from_millis(3));
1493 assert!(health_check(&meta, &config));
1494 }
1495
1496 #[test]
1497 fn test_health_check_enabled_rejects_stale_meta() {
1498 let config = PoolConfig::default().with_idle_timeout(Duration::from_millis(1));
1499 let meta = PooledClientMeta::new();
1500 std::thread::sleep(Duration::from_millis(3));
1501 assert!(!health_check(&meta, &config));
1502 }
1503
1504 #[test]
1505 fn test_pool_stats_helper_totals() {
1506 let stats = pool_stats(2, 3, 10, 8, 4, 1);
1507 assert_eq!(stats.total, 5);
1508 assert_eq!(stats.borrow_count, 4);
1509 assert_eq!(stats.error_count, 1);
1510 }
1511
1512 #[test]
1513 fn test_validate_pooled_client_rejects_empty_model() {
1514 let client = mock_client("");
1515 let err = validate_pooled_client(client.as_ref()).unwrap_err();
1516 assert!(matches!(err, PoolError::InvalidClient { .. }));
1517 }
1518
1519 #[test]
1520 fn test_validate_pooled_client_accepts_named_model() {
1521 validate_pooled_client(mock_client("ok").as_ref()).unwrap();
1522 }
1523
1524 #[tokio::test]
1525 async fn test_try_get_pool_exhausted_when_at_capacity() {
1526 let config = PoolConfig::default()
1527 .with_max_connections(1)
1528 .without_health_check();
1529 let pool = ClientPool::new(config.clone());
1530 register_seeded_pool(&pool, "mock", config, vec![mock_client("only")]);
1531
1532 let _guard = pool.try_get("mock").await.expect("first borrow");
1533 let err = pool.try_get("mock").await.unwrap_err();
1534 assert!(matches!(err, PoolError::PoolExhausted { max: 1 }));
1535 assert_eq!(pool.stats().providers["mock"].error_count, 1);
1536 }
1537
1538 #[tokio::test]
1539 async fn test_acquire_timeout_increments_error_count() {
1540 let mut config = PoolConfig::default()
1541 .with_max_connections(1)
1542 .with_idle_timeout(Duration::from_secs(60))
1543 .without_health_check();
1544 config.acquire_timeout = Duration::from_millis(50);
1545 let sub = provider_pool(config);
1546 seed_pool(&sub, vec![mock_client("held")]);
1547
1548 let (_c, permit) = sub.acquire().await.expect("hold permit");
1549 let err = match sub.acquire().await {
1550 Err(e) => e,
1551 Ok(_) => panic!("expected pool acquire timeout"),
1552 };
1553 assert!(matches!(err, PoolError::Timeout { .. }));
1554 assert_eq!(sub.stats().error_count, 1);
1555 drop(permit);
1556 }
1557
1558 #[tokio::test]
1559 async fn test_borrow_count_increments_on_success() {
1560 let config = PoolConfig::default()
1561 .with_max_connections(2)
1562 .without_health_check();
1563 let sub = provider_pool(config);
1564 seed_pool(&sub, vec![mock_client("x")]);
1565 let (_c, permit) = sub.acquire().await.unwrap();
1566 drop(permit);
1567 assert_eq!(sub.stats().borrow_count, 1);
1568 }
1569
1570 #[tokio::test]
1571 async fn test_concurrent_borrows_respect_max_connections() {
1572 let config = PoolConfig::default()
1573 .with_max_connections(2)
1574 .without_health_check();
1575 let pool = Arc::new(ClientPool::new(config.clone()));
1576 register_seeded_pool(&pool, "mock", config, vec![mock_client("c1"), mock_client("c2")]);
1577
1578 let (tx, mut rx) = tokio::sync::oneshot::channel();
1579 let pool_bg = Arc::clone(&pool);
1580 tokio::spawn(async move {
1581 let result = pool_bg.get("mock").await;
1582 let _ = tx.send(result);
1583 });
1584
1585 let g1 = pool.get("mock").await.expect("first");
1586 let g2 = pool.get("mock").await.expect("second");
1587 assert!(
1588 tokio::time::timeout(Duration::from_millis(100), &mut rx)
1589 .await
1590 .is_err(),
1591 "third borrow should still be waiting"
1592 );
1593 drop(g1);
1594 drop(g2);
1595 let third_result = rx.await.expect("channel").expect("third completes after release");
1596 drop(third_result);
1597 }
1598
1599 #[tokio::test]
1600 async fn test_get_with_error_unregistered_invalid_client() {
1601 let pool = ClientPool::with_defaults();
1602 let err = pool.get_with_error("missing").await.unwrap_err();
1603 assert!(matches!(err, PoolError::InvalidClient { .. }));
1604 }
1605
1606 #[tokio::test]
1607 async fn test_get_with_error_shutdown_invalid_client() {
1608 let pool = ClientPool::with_defaults();
1609 pool.shutdown();
1610 let err = pool.get_with_error("anything").await.unwrap_err();
1611 assert!(matches!(err, PoolError::InvalidClient { .. }));
1612 }
1613
1614 #[tokio::test]
1615 async fn test_cleanup_stale_client_pool_aggregates() {
1616 let config = PoolConfig::default()
1617 .with_idle_timeout(Duration::from_millis(5))
1618 .without_health_check();
1619 let pool = ClientPool::new(config.clone());
1620 register_seeded_pool(&pool, "mock", config, vec![mock_client("old")]);
1621 std::thread::sleep(Duration::from_millis(8));
1622 assert_eq!(pool.cleanup_stale(), 1);
1623 assert_eq!(pool.stats().total_available, 0);
1624 }
1625
1626 #[tokio::test]
1627 async fn test_stats_track_aggregate_borrow_and_error_counts() {
1628 let config = PoolConfig::default()
1629 .with_max_connections(1)
1630 .without_health_check();
1631 let pool = ClientPool::new(config.clone());
1632 register_seeded_pool(&pool, "mock", config, vec![mock_client("one")]);
1633 let _g = pool.try_get("mock").await.unwrap();
1634 let _ = pool.try_get("mock").await;
1635 let stats = pool.stats();
1636 assert_eq!(stats.borrow_count, 1);
1637 assert_eq!(stats.error_count, 1);
1638 }
1639
1640 #[tokio::test]
1641 async fn test_acquire_removes_stale_before_creating_client() {
1642 let config = PoolConfig::default()
1643 .with_idle_timeout(Duration::from_millis(5));
1644 let sub = provider_pool(config);
1645 {
1646 let mut guard = sub.clients.lock();
1647 guard.push(PooledClient {
1648 client: mock_client("stale-only"),
1649 meta: PooledClientMeta::new(),
1650 });
1651 std::thread::sleep(Duration::from_millis(8));
1652 }
1653 let (client, permit) = sub.acquire().await.expect("creates fresh client");
1654 assert_eq!(client.model_name(), "mock");
1655 assert_eq!(sub.stats().total_created, 1);
1656 drop(client);
1657 drop(permit);
1658 }
1659
1660 #[tokio::test]
1661 async fn test_race_release_then_reacquire() {
1662 let config = PoolConfig::default()
1663 .with_max_connections(1)
1664 .without_health_check();
1665 let pool = Arc::new(ClientPool::new(config.clone()));
1666 register_seeded_pool(&pool, "mock", config, vec![mock_client("race")]);
1667
1668 let g1 = pool.get("mock").await.expect("first");
1669 let pool2 = Arc::clone(&pool);
1670 let j = tokio::spawn(async move {
1671 tokio::time::sleep(Duration::from_millis(5)).await;
1672 pool2.get("mock").await
1673 });
1674 drop(g1);
1675 let g2 = j.await.expect("join").expect("second acquire after release");
1676 assert_eq!(g2.model_name(), "race");
1677 }
1678
1679 #[test]
1680 fn test_return_disposition_debug_clone() {
1681 let d = ReturnDisposition::Returned;
1682 assert_eq!(d, d);
1683 assert!(format!("{d:?}").contains("Returned"));
1684 }
1685
1686 #[test]
1687 fn test_pool_stats_clone_preserves_aggregate_fields() {
1688 let stats = PoolStats {
1689 providers: HashMap::new(),
1690 total_available: 0,
1691 total_in_use: 0,
1692 total_connections: 0,
1693 borrow_count: 2,
1694 error_count: 3,
1695 };
1696 let cloned = stats.clone();
1697 assert_eq!(cloned.borrow_count, 2);
1698 assert_eq!(cloned.error_count, 3);
1699 }
1700
1701 #[tokio::test]
1702 async fn test_provider_pool_stats_after_release_shows_idle() {
1703 let config = PoolConfig::default()
1704 .with_max_connections(1)
1705 .without_health_check();
1706 let sub = provider_pool(config);
1707 let (client, permit) = sub.acquire().await.unwrap();
1708 sub.release(client);
1709 drop(permit);
1710 let stats = sub.stats();
1711 assert_eq!(stats.available, 1);
1712 assert_eq!(stats.in_use, 0);
1713 assert_eq!(stats.total, 1);
1714 }
1715
1716
1717
1718
1719 #[test]
1720 fn test_pool_stats_default_aggregate_fields() {
1721 let stats = ClientPool::with_defaults().stats();
1722 assert_eq!(stats.total_connections, 0);
1723 assert_eq!(stats.borrow_count, 0);
1724 assert_eq!(stats.error_count, 0);
1725 }
1726
1727 #[test]
1728 fn test_return_disposition_variants() {
1729 assert_ne!(ReturnDisposition::Returned, ReturnDisposition::Dropped);
1730 }
1731
1732 #[tokio::test]
1733 async fn test_try_get_success_returns_guard() {
1734 let config = PoolConfig::default()
1735 .with_max_connections(1)
1736 .without_health_check();
1737 let pool = ClientPool::new(config.clone());
1738 register_seeded_pool(&pool, "mock", config, vec![mock_client("ok")]);
1739 let guard = pool.try_get("mock").await.expect("try_get ok");
1740 assert_eq!(guard.model_name(), "ok");
1741 }
1742
1743 #[tokio::test]
1744 async fn test_get_with_error_success_path() {
1745 let config = PoolConfig::default()
1746 .with_max_connections(1)
1747 .without_health_check();
1748 let pool = ClientPool::new(config.clone());
1749 register_seeded_pool(&pool, "mock", config, vec![mock_client("ok2")]);
1750 let guard = pool.get_with_error("mock").await.expect("get ok");
1751 assert_eq!(guard.model_name(), "ok2");
1752 }
1753
1754 #[test]
1755 fn test_llm_pool_snapshot_lists_providers() {
1756 let pool = ClientPoolBuilder::new()
1757 .provider("a", test_stub_provider())
1758 .provider("b", test_stub_provider())
1759 .build();
1760 let snap = pool.snapshot();
1761 let mut names = snap.providers;
1762 names.sort();
1763 assert_eq!(names, vec!["a", "b"]);
1764 }
1765
1766 #[test]
1767 fn test_health_check_fresh_client_meta() {
1768 let config = PoolConfig::default().with_idle_timeout(Duration::from_secs(60));
1769 let meta = PooledClientMeta::new();
1770 assert!(health_check(&meta, &config));
1771 }
1772
1773 #[tokio::test]
1774 async fn test_cleanup_stale_on_provider_pool() {
1775 let config = PoolConfig::default().with_idle_timeout(Duration::from_millis(5));
1776 let sub = provider_pool(config);
1777 {
1778 let mut guard = sub.clients.lock();
1779 guard.push(PooledClient {
1780 client: mock_client("gone"),
1781 meta: PooledClientMeta::new(),
1782 });
1783 std::thread::sleep(Duration::from_millis(8));
1784 }
1785 assert_eq!(sub.cleanup_stale(), 1);
1786 }
1787
1788 #[tokio::test]
1789 async fn test_get_after_shutdown() {
1790 let pool = ClientPool::with_defaults();
1791 pool.shutdown();
1792
1793 let result = pool.get("anything").await;
1794 assert!(result.is_err());
1795 assert!(matches!(result.unwrap_err(), AppError::LLM(_)));
1796 }
1797}
1798