Single-attempt RPC send (no retry, no circuit breaker).
(
&self,
target: u64,
envelope: &[u8],
)
| 125 | |
| 126 | /// Single-attempt RPC send (no retry, no circuit breaker). |
| 127 | async fn try_send_once( |
| 128 | &self, |
| 129 | target: u64, |
| 130 | envelope: &[u8], |
| 131 | ) -> std::result::Result<Result<RaftRpc>, ClusterError> { |
| 132 | let conn = self.get_or_connect(target).await?; |
| 133 | |
| 134 | let (mut send, mut recv) = conn.open_bi().await.map_err(|e| ClusterError::Transport { |
| 135 | detail: format!("open_bi to node {target}: {e}"), |
| 136 | })?; |
| 137 | |
| 138 | send.write_all(envelope) |
| 139 | .await |
| 140 | .map_err(|e| ClusterError::Transport { |
| 141 | detail: format!("write to node {target}: {e}"), |
| 142 | })?; |
| 143 | send.finish().map_err(|e| ClusterError::Transport { |
| 144 | detail: format!("finish send to node {target}: {e}"), |
| 145 | })?; |
| 146 | |
| 147 | let response_envelope = |
| 148 | tokio::time::timeout(self.rpc_timeout, server::read_envelope(&mut recv)) |
| 149 | .await |
| 150 | .map_err(|_| ClusterError::Transport { |
| 151 | detail: format!( |
| 152 | "RPC timeout ({}ms) to node {target}", |
| 153 | self.rpc_timeout.as_millis() |
| 154 | ), |
| 155 | })??; |
| 156 | |
| 157 | // Envelope / MAC / replay-window / codec errors are not transport |
| 158 | // errors — return them wrapped in Ok so retry logic doesn't retry |
| 159 | // a failed MAC as if it were a flaky network. |
| 160 | Ok(self.parse_inbound(&response_envelope)) |
| 161 | } |
| 162 | |
| 163 | /// Encode and wrap an RPC in an authenticated envelope. |
| 164 | fn wrap_outbound(&self, rpc: &RaftRpc) -> Result<Vec<u8>> { |
no test coverage detected