Fix SessionPool blocking indefinitely after a server outage(#62)

Author: CritasWang

docs/API.md

@@ -44,7 +44,7 @@ var tablet =
 | -------------- | ------------------------- | ------------------------ | ----------------------------- |
 | Open           | bool                      | open session             | session_pool.Open(false)      |
 | Close          | null                      | close session            | session_pool.Close()          |
-| IsOpen         | null                      | check if session is open | session_pool.IsOpen()         |
+| IsOpen         | null                      | check if the pool was opened and not yet closed by the caller. It is a lifecycle flag, **not** a connectivity probe: it stays `true` after the server goes down, because the client keeps no heartbeat and reconnects on demand instead. | session_pool.IsOpen()         |
 | OpenDebugMode  | LoggingConfiguration=null | open debug mode          | session_pool.OpenDebugMode()  |
 | CloseDebugMode | null                      | close debug mode         | session_pool.CloseDebugMode() |
 | SetTimeZone    | string                    | set time zone            | session_pool.GetTimeZone()    |

docs/SessionPool_Exception_Handling.md

@@ -34,9 +34,9 @@ using System;
 try
 {
     var sessionPool = new SessionPool.Builder()
-        .Host("127.0.0.1")
-        .Port(6667)
-        .PoolSize(4)
+        .SetHost("127.0.0.1")
+        .SetPort(6667)
+        .SetPoolSize(4)
         .Build();
 
     await sessionPool.Open();
@@ -54,6 +54,52 @@ catch (SessionPoolDepletedException ex)
 }
 ```
 
+## `IsOpen()` is a lifecycle flag, not a health check
+
+`SessionPool.IsOpen()` reports whether **you** have opened the pool and not yet closed it. It is not a
+connectivity probe:
+
+- It becomes `true` after a successful `Open()` and only returns to `false` when you call `Close()`.
+- The client runs no heartbeat, so a server that goes down does **not** flip it back to `false`.
+  Reconnection happens lazily, on the next operation.
+
+This means the following common guard never re-opens the pool, because the flag stays `true` forever:
+
+```csharp
+// Anti-pattern: this short-circuits even while every connection is dead
+if (_pool != null && _pool.IsOpen()) return;
+```
+
+To reason about actual availability, use the health metrics below, or simply let an operation throw
+`SessionPoolDepletedException` and handle it.
+
+## Pool Wait Timeout
+
+Two independent timeouts govern a pool operation:
+
+| Setting                                   | Unit | Default | Controls                                                                  |
+| ----------------------------------------- | ---- | ------- | ------------------------------------------------------------------------- |
+| `SetConnectionTimeoutInMs(int)`           | ms   | 500     | Socket-level send/receive timeout of an individual connection             |
+| `SetPoolWaitTimeoutInMs(int)`             | ms   | 10000   | How long an operation waits for a free client before the pool gives up    |
+
+```csharp
+var sessionPool = new SessionPool.Builder()
+    .SetHost("127.0.0.1")
+    .SetPort(6667)
+    .SetPoolSize(8)
+    .SetConnectionTimeoutInMs(500)   // socket timeout
+    .SetPoolWaitTimeoutInMs(10_000)  // give up after 10s of waiting for a free client
+    .Build();
+```
+
+When the wait budget is exhausted, the operation throws `SessionPoolDepletedException` with the reason
+`Connection pool is empty and wait time out(...ms)`. Raise `SetPoolWaitTimeoutInMs` if your workload
+legitimately queues behind long operations; lower it if you would rather fail fast and retry.
+
+> **Note:** before this setting existed, the wait budget was derived from the connection timeout and then
+> misinterpreted as seconds, which turned the 500 ms default into a ~41 minute block. If you are upgrading
+> from an older version and relied on that (unintended) long wait, set `SetPoolWaitTimeoutInMs` explicitly.
+
 ## Pool Health Metrics
 
 ### Monitoring Pool Status
@@ -62,16 +108,17 @@ The `SessionPool` class exposes real-time health metrics that can be used for mo
 
 ```csharp
 var sessionPool = new SessionPool.Builder()
-    .Host("127.0.0.1")
-    .Port(6667)
-    .PoolSize(8)
+    .SetHost("127.0.0.1")
+    .SetPort(6667)
+    .SetPoolSize(8)
     .Build();
 
 await sessionPool.Open();
 
 // Check pool health
 Console.WriteLine($"Available Clients: {sessionPool.AvailableClients}");
 Console.WriteLine($"Total Pool Size: {sessionPool.TotalPoolSize}");
+Console.WriteLine($"Unrealized Capacity: {sessionPool.UnrealizedCapacity}");
 Console.WriteLine($"Failed Reconnections: {sessionPool.FailedReconnections}");
 ```
 
@@ -81,8 +128,27 @@ Console.WriteLine($"Failed Reconnections: {sessionPool.FailedReconnections}");
 | -------------------- | --------------------- | ------------------------------------------------ | --------------------------- |
 | Available Clients    | `AvailableClients`    | Number of idle clients ready for use             | Alert if < 25% of pool size |
 | Total Pool Size      | `TotalPoolSize`       | Configured maximum pool size                     | N/A (constant)              |
+| Unrealized Capacity  | `UnrealizedCapacity`  | Configured capacity currently holding no connection, refilled on demand | Not an alert signal on its own - see below |
 | Failed Reconnections | `FailedReconnections` | Cumulative count of failed reconnection attempts | Alert if > 0 and increasing |
 
+### Capacity is demand-driven
+
+When an operation fails and reconnection also fails, the dead connection is discarded but its **capacity is
+retained** rather than lost. `UnrealizedCapacity` counts the capacity left without a connection, and an
+acquisition that finds no idle client materializes one connection before falling back to waiting.
+Consequences:
+
+- The pool no longer shrinks by one on every failure, so it cannot reach the state where every caller blocks
+  on a queue nobody will feed.
+- Once the server is reachable again, the pool repopulates itself as load demands it - no `Close()` +
+  `Open()` cycle is required.
+- **Capacity is refilled on demand, not eagerly.** A connection is only created when an acquisition finds the
+  idle queue empty. Under sequential or light workloads one connection is enough to serve every request, so
+  `UnrealizedCapacity` legitimately stays above zero long after the server has fully recovered. It measures
+  how much of the configured pool has not been materialized, not server availability.
+- Therefore **do not alert on `UnrealizedCapacity` alone.** Use `FailedReconnections` to reason about server
+  reachability: it only increases when a reconnection actually fails.
+
 ## Failure Scenarios and Recovery Strategies
 
 ### Scenario 1: Pool Exhaustion (High Load)
@@ -101,9 +167,9 @@ Console.WriteLine($"Failed Reconnections: {sessionPool.FailedReconnections}");
 
 ```csharp
 var sessionPool = new SessionPool.Builder()
-    .Host("127.0.0.1")
-    .Port(6667)
-    .PoolSize(16)  // Increased from 8
+    .SetHost("127.0.0.1")
+    .SetPort(6667)
+    .SetPoolSize(16)  // Increased from 8
     .Build();
 ```
 
@@ -137,13 +203,25 @@ for (int i = 0; i < maxRetries; i++)
 **Symptoms:**
 
 - `SessionPoolDepletedException` with reason "Reconnection failed"
-- `AvailableClients` decreases over time
-- `FailedReconnections` > 0 and increasing
+- `AvailableClients` drops toward 0 while `UnrealizedCapacity` rises
+- `FailedReconnections` > 0 and increasing (this, not `UnrealizedCapacity`, is the outage signal)
 
 **Root Cause:** IoTDB server unreachable or network issues
 
 **Recovery Strategies:**
 
+0. **Do nothing but retry.** Capacity is retained and refilled on demand, so once the server comes back a
+   plain retry succeeds. Reinitialising is only needed if you want to change configuration or drop
+   accumulated state:
+
+```csharp
+catch (SessionPoolDepletedException ex)
+{
+    // Capacity is retained and refilled on demand - just back off and try again
+    await Task.Delay(2000);
+}
+```
+
 1. **Reinitialize SessionPool:**
 
 ```csharp
@@ -159,9 +237,9 @@ catch (SessionPoolDepletedException ex) when (ex.FailedReconnections > 5)
 
     // Create new pool
     sessionPool = new SessionPool.Builder()
-        .Host("127.0.0.1")
-        .Port(6667)
-        .PoolSize(8)
+        .SetHost("127.0.0.1")
+        .SetPort(6667)
+        .SetPoolSize(8)
         .Build();
 
     await sessionPool.Open();
@@ -245,9 +323,9 @@ public async Task RateLimitedInsert(string deviceId, RowRecord record)
 
 ```csharp
 var sessionPool = new SessionPool.Builder()
-    .Host("127.0.0.1")
-    .Port(6667)
-    .Timeout(120)  // Increased timeout for slow server
+    .SetHost("127.0.0.1")
+    .SetPort(6667)
+    .SetConnectionTimeoutInMs(5000)  // Increased socket timeout for a slow server
     .Build();
 ```
 
@@ -381,10 +459,10 @@ public class ProductionSessionPoolManager
     public async Task Initialize()
     {
         _pool = new SessionPool.Builder()
-            .Host("127.0.0.1")
-            .Port(6667)
-            .PoolSize(8)
-            .Timeout(60)
+            .SetHost("127.0.0.1")
+            .SetPort(6667)
+            .SetPoolSize(8)
+            .SetConnectionTimeoutInMs(5000)
             .Build();
 
         await _pool.Open();
@@ -481,7 +559,11 @@ public class ProductionSessionPoolManager
 The SessionPool exception handling and health monitoring features provide comprehensive tools for building robust IoTDB applications:
 
 - Use `SessionPoolDepletedException` to understand and react to pool issues
-- Monitor `AvailableClients`, `TotalPoolSize`, and `FailedReconnections` metrics
+- Treat `IsOpen()` as a lifecycle flag, never as a connectivity check
+- Tune `SetPoolWaitTimeoutInMs` separately from `SetConnectionTimeoutInMs`
+- Monitor `AvailableClients`, `TotalPoolSize`, and `FailedReconnections`; read `UnrealizedCapacity` as
+  capacity not yet materialized rather than as an outage signal
+- Rely on demand-driven capacity refill for recovery; reinitialise only when you need to change configuration
 - Implement appropriate recovery strategies based on failure scenarios
 - Set up proactive monitoring and alerting to prevent issues
 - Follow best practices for pool sizing and resource management

src/Apache.IoTDB/ConcurrentClientQueue.cs

@@ -58,26 +58,52 @@ public void Return(Client client)
         public void AddRef() => Interlocked.Increment(ref _ref);
         public int GetRef() => Volatile.Read(ref _ref);
         public void RemoveRef() => Interlocked.Decrement(ref _ref);
-        public int Timeout { get; set; } = 10;
+
+        /// <summary>
+        /// The maximum time, in milliseconds, that <see cref="Take"/> waits for a client to be
+        /// returned to the pool before throwing. Defaults to 10000 (10 seconds).
+        /// </summary>
+        public int TimeoutInMs { get; set; } = DefaultTimeoutInMs;
+
+        internal const int DefaultTimeoutInMs = 10_000;
+
+        /// <summary>
+        /// The wait timeout expressed in seconds. Kept for backward compatibility only; it is a thin
+        /// wrapper over <see cref="TimeoutInMs"/>. Prefer <see cref="TimeoutInMs"/>, which avoids the
+        /// unit ambiguity that previously caused millisecond values to be interpreted as seconds.
+        /// </summary>
+        [Obsolete("Use TimeoutInMs instead. This property interprets its value as seconds.")]
+        public int Timeout
+        {
+            get => TimeoutInMs / 1000;
+            set => TimeoutInMs = value * 1000;
+        }
+
         public Client Take()
         {
             Client client = null;
+            // One overall deadline for the whole call. Return() uses PulseAll, so every waiter wakes up
+            // while only one of them can dequeue the returned client; re-arming the full timeout on each
+            // wake-up would let an unlucky waiter exceed the configured bound indefinitely under churn.
+            var budgetMs = TimeoutInMs;
+            var elapsed = Stopwatch.StartNew();
             Monitor.Enter(ClientQueue);
             try
             {
                 while (true)
                 {
-                    bool timeout = false;
-                    if (ClientQueue.IsEmpty)
+                    if (ClientQueue.TryDequeue(out client))
                     {
-                        timeout = !Monitor.Wait(ClientQueue, TimeSpan.FromSeconds(Timeout));
+                        break;
                     }
-                    ClientQueue.TryDequeue(out client);
 
-                    if (client != null || timeout)
+                    var remainingMs = budgetMs - (int)elapsed.ElapsedMilliseconds;
+                    if (remainingMs <= 0)
                     {
                         break;
                     }
+
+                    Monitor.Wait(ClientQueue, TimeSpan.FromMilliseconds(remainingMs));
                 }
             }
             finally
@@ -86,7 +112,7 @@ public Client Take()
             }
             if (client == null)
             {
-                var reasonPhrase = $"Connection pool is empty and wait time out({Timeout}s)";
+                var reasonPhrase = $"Connection pool is empty and wait time out({budgetMs}ms)";
                 if (DiagnosticReporter != null)
                 {
                     throw DiagnosticReporter.BuildDepletionException(reasonPhrase);

src/Apache.IoTDB/SessionPool.Builder.cs

@@ -34,6 +34,7 @@ public class Builder
         private int _poolSize = 8;
         private bool _enableRpcCompression = false;
         private int _connectionTimeoutInMs = 500;
+        private int _poolWaitTimeoutInMs = DefaultPoolWaitTimeoutInMs;
         private bool _useSsl = false;
         private string _certificatePath = null;
         private string _sqlDialect = IoTDBConstant.TREE_SQL_DIALECT;
@@ -94,6 +95,18 @@ public Builder SetConnectionTimeoutInMs(int timeout)
             return this;
         }
 
+        /// <summary>
+        /// Sets how long, in milliseconds, an operation waits for a client to become available in the pool
+        /// before a <see cref="SessionPoolDepletedException"/> is thrown. Defaults to
+        /// <see cref="DefaultPoolWaitTimeoutInMs"/> (10 seconds). This is independent of
+        /// <see cref="SetConnectionTimeoutInMs"/>, which controls the socket-level timeout.
+        /// </summary>
+        public Builder SetPoolWaitTimeoutInMs(int poolWaitTimeoutInMs)
+        {
+            _poolWaitTimeoutInMs = poolWaitTimeoutInMs;
+            return this;
+        }
+
         public Builder SetUseSsl(bool useSsl)
         {
             _useSsl = useSsl;
@@ -135,6 +148,7 @@ public Builder()
             _poolSize = 8;
             _enableRpcCompression = false;
             _connectionTimeoutInMs = 500;
+            _poolWaitTimeoutInMs = DefaultPoolWaitTimeoutInMs;
             _useSsl = false;
             _certificatePath = null;
             _sqlDialect = IoTDBConstant.TREE_SQL_DIALECT;
@@ -146,9 +160,9 @@ public SessionPool Build()
             // if nodeUrls is not empty, use nodeUrls to create session pool
             if (_nodeUrls.Count > 0)
             {
-                return new SessionPool(_nodeUrls, _username, _password, _fetchSize, _zoneId, _poolSize, _enableRpcCompression, _connectionTimeoutInMs, _useSsl, _certificatePath, _sqlDialect, _database);
+                return new SessionPool(_nodeUrls, _username, _password, _fetchSize, _zoneId, _poolSize, _enableRpcCompression, _connectionTimeoutInMs, _useSsl, _certificatePath, _sqlDialect, _database, _poolWaitTimeoutInMs);
             }
-            return new SessionPool(_host, _port, _username, _password, _fetchSize, _zoneId, _poolSize, _enableRpcCompression, _connectionTimeoutInMs, _useSsl, _certificatePath, _sqlDialect, _database);
+            return new SessionPool(_host, _port, _username, _password, _fetchSize, _zoneId, _poolSize, _enableRpcCompression, _connectionTimeoutInMs, _useSsl, _certificatePath, _sqlDialect, _database, _poolWaitTimeoutInMs);
         }
     }
 }