Handle all bidi streams on a single connection. Exits cleanly (Ok) on shutdown, on normal connection close, or on unrecoverable transport error.
(
conn: quinn::Connection,
handler: Arc<H>,
auth: Arc<AuthContext>,
identity_store: Arc<S>,
mut shutdown: watch::Receiver<bool>,
)
| 128 | /// Exits cleanly (Ok) on shutdown, on normal connection close, |
| 129 | /// or on unrecoverable transport error. |
| 130 | pub(crate) async fn handle_connection<H: RaftRpcHandler, S: PeerIdentityStore>( |
| 131 | conn: quinn::Connection, |
| 132 | handler: Arc<H>, |
| 133 | auth: Arc<AuthContext>, |
| 134 | identity_store: Arc<S>, |
| 135 | mut shutdown: watch::Receiver<bool>, |
| 136 | ) -> Result<()> { |
| 137 | // Extract the peer cert once per connection; it does not change. |
| 138 | let peer_cert_der: Option<Vec<u8>> = peer_leaf_cert_der(&conn); |
| 139 | let peer_addr = conn.remote_address(); |
| 140 | |
| 141 | // Perform the wire-version handshake on the first bidi stream before |
| 142 | // dispatching any RPCs. The client opens a dedicated stream for this |
| 143 | // exchange; subsequent streams on the same connection are RPC streams. |
| 144 | let agreed_version = { |
| 145 | let accepted = tokio::select! { |
| 146 | biased; |
| 147 | _ = shutdown.changed() => { |
| 148 | if *shutdown.borrow() { |
| 149 | return Ok(()); |
| 150 | } |
| 151 | // Spurious change — retry the accept. |
| 152 | conn.accept_bi().await |
| 153 | } |
| 154 | result = conn.accept_bi() => result, |
| 155 | }; |
| 156 | |
| 157 | let (mut hs_send, mut hs_recv) = match accepted { |
| 158 | Ok(streams) => streams, |
| 159 | Err(quinn::ConnectionError::ApplicationClosed(_)) => return Ok(()), |
| 160 | Err(quinn::ConnectionError::LocallyClosed) => return Ok(()), |
| 161 | Err(e) => { |
| 162 | return Err(ClusterError::Transport { |
| 163 | detail: format!("accept handshake stream from {peer_addr}: {e}"), |
| 164 | }); |
| 165 | } |
| 166 | }; |
| 167 | |
| 168 | let local = local_version_range(); |
| 169 | match perform_version_handshake_server(&conn, &mut hs_send, &mut hs_recv).await { |
| 170 | Ok(v) => v, |
| 171 | Err(e) => { |
| 172 | warn!( |
| 173 | peer_addr = %peer_addr, |
| 174 | local_min = %local.min, |
| 175 | local_max = %local.max, |
| 176 | error = %e, |
| 177 | "wire version handshake failed; closing connection" |
| 178 | ); |
| 179 | // perform_version_handshake_server already closed the QUIC |
| 180 | // connection on range mismatch; propagate the error so the |
| 181 | // caller logs it and the connection task exits. |
| 182 | return Err(e); |
| 183 | } |
| 184 | } |
| 185 | }; |
| 186 | |
| 187 | debug!( |
nothing calls this directly
no test coverage detected