feat(client-pool): expose PoolHandle API with fair scheduling and graceful shutdown

docs/roadmap.md

@@ -502,7 +502,7 @@ document so larger deployments can adopt the library confidently.
 - [x] 11.2.1. Implement a configurable connection pool that preserves preamble
   state, enforces in-flight request limits per socket, and recycles idle
   connections.
-- [ ] 11.2.2. Expose a `PoolHandle` API with fairness policies so callers can
+- [x] 11.2.2. Expose a `PoolHandle` API with fairness policies, so callers can
   multiplex many logical sessions without violating back-pressure.
 
 ### 11.3. Streaming helpers and test utilities

docs/users-guide.md

@@ -1383,13 +1383,14 @@ cargo run --example client_echo_login --features examples
 
 Use `connect_pool` when one warm socket is too little, but opening a new TCP
 connection for every request is too expensive. `WireframeClientPool` keeps a
-bounded set of warm sockets, preserves preamble state on reuse, and recycles
-idle sockets after the configured timeout.
+bounded set of warm sockets, preserves preamble state on reuse, recycles idle
+sockets after the configured timeout, and can schedule blocked logical sessions
+fairly through `PoolHandle`.
 
 ```rust,no_run
 use std::{net::SocketAddr, time::Duration};
 
-use wireframe::client::{ClientPoolConfig, WireframeClient};
+use wireframe::client::{ClientPoolConfig, PoolFairnessPolicy, WireframeClient};
 
 #[derive(bincode::Encode, bincode::Decode)]
 struct Ping(u8);
@@ -1406,7 +1407,8 @@ let pool = WireframeClient::builder()
         ClientPoolConfig::default()
             .pool_size(2)
             .max_in_flight_per_socket(2)
-            .idle_timeout(Duration::from_secs(30)),
+            .idle_timeout(Duration::from_secs(30))
+            .fairness_policy(PoolFairnessPolicy::RoundRobin),
     )
     .await?;
 
@@ -1425,6 +1427,54 @@ serialized per physical connection while still allowing multiple callers to be
 admitted against the same warm slot. Treat `max_in_flight_per_socket` as an
 admission budget, not as a guarantee of parallel writes on one TCP stream.
 
+Create a `PoolHandle` when one logical session needs repeated pooled access and
+that session should participate in the configured fairness policy over time:
+
+```rust,no_run
+# use std::net::SocketAddr;
+use wireframe::client::{ClientPoolConfig, PoolFairnessPolicy, PoolHandle, WireframeClient};
+
+#[derive(bincode::Encode, bincode::Decode)]
+struct Ping(u8);
+
+#[derive(bincode::Encode, bincode::Decode, Debug, PartialEq)]
+struct Pong(u8);
+
+# #[tokio::main]
+# async fn main() -> Result<(), wireframe::client::ClientError> {
+let addr: SocketAddr = "127.0.0.1:7878".parse().expect("valid socket address");
+let pool = WireframeClient::builder()
+    .connect_pool(
+        addr,
+        ClientPoolConfig::default()
+            .pool_size(1)
+            .fairness_policy(PoolFairnessPolicy::RoundRobin),
+    )
+    .await?;
+
+let mut session_a: PoolHandle<_, _, _> = pool.handle();
+let mut session_b = pool.handle();
+
+let pong_a: Pong = session_a.call(&Ping(1)).await?;
+let pong_b: Pong = session_b.call(&Ping(2)).await?;
+
+assert_eq!(pong_a, Pong(1));
+assert_eq!(pong_b, Pong(2));
+# Ok(())
+# }
+```
+
+Use `pool.handle()` instead of repeated `pool.acquire()` when a long-lived
+workflow should receive fair turns relative to other workflows. Fairness
+policies order blocked handles; they do not create a second queue outside the
+pool or bypass its back-pressure. If all permits are busy, the waiting handle
+still blocks until capacity returns.
+
+Continue using `PooledClientLease` for explicit split-phase work such as
+`send()` followed later by `receive()`. `PoolHandle` does not pin a logical
+session to one socket and does not demultiplex arbitrary responses across
+shared handles.
+
 Preamble callbacks and setup hooks run per physical socket creation. Warm reuse
 keeps the existing preamble state; idle recycle closes the old socket and
 creates a fresh one, which reruns preamble and setup.

docs/wireframe-client-design.md

@@ -474,6 +474,40 @@ flowchart TD
   or true per-socket concurrent transport use, revisit this boundary and decide
   whether to extend or replace the slot wrapper.
 
+## Decision record for 11.2.2
+
+- Decision: `PoolHandle` represents the fairness identity of one logical
+  session, not affinity to one physical socket.
+- Rationale: callers needed a durable identity so repeated pooled acquisition
+  by one busy workflow could not crowd out other waiting workflows. Tying the
+  handle to fairness instead of socket ownership preserves warm-socket reuse
+  without implying transport pinning.
+- Scheduling model:
+  - `WireframeClientPool` now owns shared inner state plus a handle scheduler;
+  - `pool.handle()` registers one stable logical-session ID;
+  - blocked `PoolHandle::acquire()` calls queue against that scheduler, which
+    obtains a real slot permit before waking the chosen handle.
+- Fairness distinction:
+  - slot rotation still decides which physical socket to try first across the
+    pool's warm connections;
+  - handle fairness decides which logical session receives the next lease when
+    contention exists.
+- Policy set:
+  - `PoolFairnessPolicy::RoundRobin` is the default because it gives stable
+    logical sessions turns in rotation under repeated contention;
+  - `PoolFairnessPolicy::Fifo` is available when callers want blocked handles
+    served strictly in arrival order.
+- API boundary:
+  - `PoolHandle` exposes fair `acquire()` plus safe whole-operation helpers
+    such as `call()`;
+  - `PooledClientLease` remains the low-level surface for split-phase workflows
+    (`send` followed later by `receive`) because the pool still does not
+    demultiplex arbitrary responses across shared handles.
+- Back-pressure posture:
+  - handle fairness is additive above the existing slot-permit budget;
+  - the scheduler never bypasses `max_in_flight_per_socket`, serialized socket
+    access, or idle recycle behaviour inherited from `11.2.1`.
+
 ## Future work
 
 This initial design focuses on a basic request/response workflow. Future

src/client/mod.rs

@@ -41,7 +41,13 @@ pub use hooks::{
     ClientErrorHandler,
 };
 #[cfg(feature = "pool")]
-pub use pool::{ClientPoolConfig, PooledClientLease, WireframeClientPool};
+pub use pool::{
+    ClientPoolConfig,
+    PoolFairnessPolicy,
+    PoolHandle,
+    PooledClientLease,
+    WireframeClientPool,
+};
 pub use response_stream::ResponseStream;
 pub use runtime::WireframeClient;
 pub use send_streaming::{SendStreamingConfig, SendStreamingOutcome};

src/client/pool/client_pool.rs

@@ -5,18 +5,21 @@ use std::{
     pin::Pin,
     sync::{
         Arc,
-        atomic::{AtomicUsize, Ordering},
+        atomic::{AtomicBool, AtomicUsize, Ordering},
     },
 };
 
 use bb8::Pool;
 use bincode::Encode;
 use futures::{Future, future::select_all};
+use tokio::sync::Notify;
 
 use super::{
     ClientPoolConfig,
+    handle::PoolHandle,
     lease::PooledClientLease,
     manager::WireframeConnectionManager,
+    scheduler::PoolScheduler,
     slot::PoolSlot,
 };
 use crate::{
@@ -29,15 +32,27 @@ type AcquirePermit<S, P, C> = (Arc<PoolSlot<S, P, C>>, tokio::sync::OwnedSemapho
 type AcquirePermitFuture<S, P, C> =
     Pin<Box<dyn Future<Output = Result<AcquirePermit<S, P, C>, ClientError>> + Send>>;
 
+pub(crate) struct ClientPoolInner<S, P = (), C = ()>
+where
+    S: Serializer + Clone + Send + Sync + 'static,
+    P: Encode + Clone + Send + Sync + 'static,
+    C: Send + 'static,
+{
+    pub(crate) slots: Arc<[Arc<PoolSlot<S, P, C>>]>,
+    pub(crate) next_slot: AtomicUsize,
+    pub(crate) scheduler: Arc<PoolScheduler<S, P, C>>,
+    shutdown: AtomicBool,
+    shutdown_notify: Notify,
+}
+
 /// Pool of warm wireframe client sockets.
 pub struct WireframeClientPool<S, P = (), C = ()>
 where
     S: Serializer + Clone + Send + Sync + 'static,
     P: Encode + Clone + Send + Sync + 'static,
     C: Send + 'static,
 {
-    slots: Arc<[Arc<PoolSlot<S, P, C>>]>,
-    next_slot: AtomicUsize,
+    inner: Arc<ClientPoolInner<S, P, C>>,
 }
 
 impl<S, P, C> WireframeClientPool<S, P, C>
@@ -51,6 +66,7 @@ where
         pool_config: ClientPoolConfig,
         parts: ClientBuildParts<S, P, C>,
     ) -> Result<Self, ClientError> {
+        let fairness_policy = pool_config.fairness_policy_value();
         let mut slots = Vec::with_capacity(pool_config.pool_size_value());
         for _ in 0..pool_config.pool_size_value() {
             let manager = WireframeConnectionManager::new(addr, parts.clone());
@@ -68,21 +84,89 @@ where
         }
 
         Ok(Self {
-            slots: Arc::from(slots),
-            next_slot: AtomicUsize::new(0),
+            inner: Arc::new(ClientPoolInner {
+                slots: Arc::from(slots),
+                next_slot: AtomicUsize::new(0),
+                scheduler: Arc::new(PoolScheduler::new(fairness_policy)),
+                shutdown: AtomicBool::new(false),
+                shutdown_notify: Notify::new(),
+            }),
         })
     }
 
+    /// Create a stable logical-session handle governed by the pool scheduler.
+    ///
+    /// Prefer a `PoolHandle` when one logical session repeatedly acquires
+    /// pooled leases and should receive fair turns relative to other blocked
+    /// sessions.
+    ///
+    /// # Examples
+    ///
+    /// ```no_run
+    /// # use std::net::SocketAddr;
+    /// use wireframe::client::{ClientPoolConfig, PoolHandle, WireframeClient};
+    ///
+    /// # async fn demo(addr: SocketAddr) -> Result<(), wireframe::client::ClientError> {
+    /// let pool = WireframeClient::builder()
+    ///     .connect_pool(addr, ClientPoolConfig::default())
+    ///     .await?;
+    /// let mut handle: PoolHandle<_, _, _> = pool.handle();
+    /// let lease = handle.acquire().await?;
+    /// drop(lease);
+    /// # Ok(())
+    /// # }
+    /// ```
+    #[must_use]
+    pub fn handle(&self) -> PoolHandle<S, P, C> {
+        let handle_id = self.inner.scheduler.register_handle();
+        PoolHandle::new(Arc::clone(&self.inner), handle_id)
+    }
+
     /// Acquire one pooled-client lease.
     ///
     /// # Errors
     ///
     /// Returns [`ClientError`] if permit acquisition or pooled checkout fails.
     pub async fn acquire(&self) -> Result<PooledClientLease<S, P, C>, ClientError> {
-        if let Some((slot, permit)) = self.try_acquire_immediately() {
-            return Ok(PooledClientLease::new(slot, permit));
+        let mut handle = self.handle();
+        handle.acquire().await
+    }
+
+    /// Close the pool and drop all warm sockets.
+    pub async fn close(self) {
+        self.inner.shutdown();
+        tokio::task::yield_now().await;
+        drop(self);
+    }
+}
+
+impl<S, P, C> ClientPoolInner<S, P, C>
+where
+    S: Serializer + Clone + Send + Sync + 'static,
+    P: Encode + Clone + Send + Sync + 'static,
+    C: Send + 'static,
+{
+    pub(crate) fn is_shutdown(&self) -> bool { self.shutdown.load(Ordering::Acquire) }
+
+    pub(crate) async fn shutdown_notified(&self) { self.shutdown_notify.notified().await; }
+
+    pub(crate) fn shutdown(&self) {
+        self.shutdown.store(true, Ordering::Release);
+        self.shutdown_notify.notify_waiters();
+        self.scheduler.notify_shutdown();
+    }
+
+    pub(crate) fn try_acquire_immediately(self: &Arc<Self>) -> Option<PooledClientLease<S, P, C>> {
+        if self.is_shutdown() {
+            return None;
         }
+        self.ordered_slots().into_iter().find_map(|slot| {
+            slot.try_acquire_permit()
+                .map(|permit| PooledClientLease::new(slot, permit, Some(Arc::clone(self))))
+        })
+    }
 
+    pub(crate) async fn acquire_slot_permit(&self) -> Result<AcquirePermit<S, P, C>, ClientError> {
         let waiters = self
             .ordered_slots()
             .into_iter()
@@ -94,14 +178,7 @@ where
             })
             .collect::<Vec<_>>();
         let (result, ..) = select_all(waiters).await;
-        let (slot, permit) = result?;
-        Ok(PooledClientLease::new(slot, permit))
-    }
-
-    fn try_acquire_immediately(&self) -> Option<AcquirePermit<S, P, C>> {
-        self.ordered_slots()
-            .into_iter()
-            .find_map(|slot| slot.try_acquire_permit().map(|permit| (slot, permit)))
+        result
     }
 
     fn ordered_slots(&self) -> Vec<Arc<PoolSlot<S, P, C>>> {
@@ -113,10 +190,4 @@ where
         }
         ordered
     }
-
-    /// Close the pool and drop all warm sockets.
-    pub async fn close(self) {
-        tokio::task::yield_now().await;
-        drop(self);
-    }
 }

src/client/pool/config.rs

@@ -2,6 +2,8 @@
 
 use std::time::Duration;
 
+use super::policy::PoolFairnessPolicy;
+
 const DEFAULT_POOL_SIZE: usize = 4;
 const DEFAULT_MAX_IN_FLIGHT_PER_SOCKET: usize = 1;
 const DEFAULT_IDLE_TIMEOUT: Duration = Duration::from_secs(600);
@@ -13,21 +15,24 @@ const DEFAULT_IDLE_TIMEOUT: Duration = Duration::from_secs(600);
 /// ```
 /// use std::time::Duration;
 ///
-/// use wireframe::client::ClientPoolConfig;
+/// use wireframe::client::{ClientPoolConfig, PoolFairnessPolicy};
 ///
 /// let config = ClientPoolConfig::default()
 ///     .pool_size(2)
 ///     .max_in_flight_per_socket(3)
-///     .idle_timeout(Duration::from_secs(30));
+///     .idle_timeout(Duration::from_secs(30))
+///     .fairness_policy(PoolFairnessPolicy::Fifo);
 /// assert_eq!(config.pool_size_value(), 2);
 /// assert_eq!(config.max_in_flight_per_socket_value(), 3);
 /// assert_eq!(config.idle_timeout_value(), Duration::from_secs(30));
+/// assert_eq!(config.fairness_policy_value(), PoolFairnessPolicy::Fifo);
 /// ```
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub struct ClientPoolConfig {
     pool_size: usize,
     max_in_flight_per_socket: usize,
     idle_timeout: Duration,
+    fairness_policy: PoolFairnessPolicy,
 }
 
 impl Default for ClientPoolConfig {
@@ -36,6 +41,7 @@ impl Default for ClientPoolConfig {
             pool_size: DEFAULT_POOL_SIZE,
             max_in_flight_per_socket: DEFAULT_MAX_IN_FLIGHT_PER_SOCKET,
             idle_timeout: DEFAULT_IDLE_TIMEOUT,
+            fairness_policy: PoolFairnessPolicy::RoundRobin,
         }
     }
 }
@@ -68,6 +74,13 @@ impl ClientPoolConfig {
         self
     }
 
+    /// Set how blocked `PoolHandle`s are ordered when capacity returns.
+    #[must_use]
+    pub fn fairness_policy(mut self, fairness_policy: PoolFairnessPolicy) -> Self {
+        self.fairness_policy = fairness_policy;
+        self
+    }
+
     /// Return the configured number of physical sockets.
     #[must_use]
     pub const fn pool_size_value(&self) -> usize { self.pool_size }
@@ -79,4 +92,8 @@ impl ClientPoolConfig {
     /// Return the configured idle recycle timeout.
     #[must_use]
     pub const fn idle_timeout_value(&self) -> Duration { self.idle_timeout }
+
+    /// Return the configured handle fairness policy.
+    #[must_use]
+    pub const fn fairness_policy_value(&self) -> PoolFairnessPolicy { self.fairness_policy }
 }