Acquire a connection from the pool. Returns an idle connection if available, otherwise creates a new one. Blocks if `max_size` connections are already in use.
(&self)
| 112 | /// Returns an idle connection if available, otherwise creates a new one. |
| 113 | /// Blocks if `max_size` connections are already in use. |
| 114 | pub async fn acquire(&self) -> NodeDbResult<PooledConnection<'_>> { |
| 115 | let permit = tokio::time::timeout(self.config.connect_timeout, self.semaphore.acquire()) |
| 116 | .await |
| 117 | .map_err(|_| NodeDbError::sync_connection_failed("pool acquire timeout"))? |
| 118 | .map_err(|_| NodeDbError::sync_connection_failed("pool closed"))?; |
| 119 | |
| 120 | // std::sync::Mutex is intentional here (not tokio::sync::Mutex): |
| 121 | // 1. Critical section is trivial (pop_front / push_back only) |
| 122 | // 2. No async operations while holding the lock |
| 123 | // 3. Enables synchronous return in Drop (no spawned tasks) |
| 124 | // 4. Poison is handled gracefully via unwrap_or_else |
| 125 | let idle_conn = { |
| 126 | let mut idle = self.inner.idle.lock().unwrap_or_else(|e| e.into_inner()); |
| 127 | idle.pop_front() |
| 128 | }; |
| 129 | |
| 130 | if let Some(mut conn) = idle_conn { |
| 131 | // Health check: ping to verify the connection is still alive. |
| 132 | if conn.ping().await.is_ok() { |
| 133 | return Ok(PooledConnection { |
| 134 | conn: Some(conn), |
| 135 | inner: Arc::clone(&self.inner), |
| 136 | _permit: permit, |
| 137 | }); |
| 138 | } |
| 139 | // Connection dead — create a new one below. |
| 140 | } |
| 141 | |
| 142 | // Create a new connection (plain TCP or TLS). |
| 143 | let addr = self.config.addr.clone(); |
| 144 | let tls_cfg = self.config.tls.clone(); |
| 145 | let timeout = self.config.connect_timeout; |
| 146 | let mut conn = tokio::time::timeout(timeout, async move { |
| 147 | if tls_cfg.enabled { |
| 148 | NativeConnection::connect_tls(&addr, &tls_cfg).await |
| 149 | } else { |
| 150 | NativeConnection::connect(&addr).await |
| 151 | } |
| 152 | }) |
| 153 | .await |
| 154 | .map_err(|_| NodeDbError::sync_connection_failed("connect timeout"))??; |
| 155 | |
| 156 | // `NativeConnection::connect` already performed the handshake. |
| 157 | // Calling it a second time here would write a stale HelloFrame |
| 158 | // into the post-handshake stream and confuse the framed read |
| 159 | // path (the bytes get mis-parsed as a regular response frame). |
| 160 | |
| 161 | // Authenticate — pass the optional database name for handshake binding. |
| 162 | conn.authenticate(self.config.auth.clone(), self.config.database.as_deref()) |
| 163 | .await?; |
| 164 | |
| 165 | // Capture negotiated metadata from the first handshake. |
| 166 | { |
| 167 | let mut meta = self.inner.meta.lock().unwrap_or_else(|e| e.into_inner()); |
| 168 | if meta.is_none() { |
| 169 | *meta = Some(NegotiatedMeta { |
| 170 | proto_version: conn.proto_version, |
| 171 | capabilities: conn.capabilities, |
no test coverage detected