| 204 | // Run the recon protocol conversation to completion. |
| 205 | #[instrument(skip_all)] |
| 206 | async fn protocol<R, S, E>( |
| 207 | sync_id: Option<String>, |
| 208 | mut role: R, |
| 209 | stream: S, |
| 210 | metrics: Metrics, |
| 211 | ) -> Result<usize> |
| 212 | where |
| 213 | R: Role, |
| 214 | R::Out: std::fmt::Debug + Send + 'static, |
| 215 | R::In: std::fmt::Debug, |
| 216 | MessageLabels: for<'a> From<&'a R::Out>, |
| 217 | MessageLabels: for<'a> From<&'a R::In>, |
| 218 | S: Stream<Item = Result<ReconMessage<R::In>, E>> |
| 219 | + Sink<ReconMessage<R::Out>, Error = E> |
| 220 | + Send |
| 221 | + 'static, |
| 222 | E: std::error::Error + Send + Sync + 'static, |
| 223 | { |
| 224 | let start = Instant::now(); |
| 225 | let (sink, stream) = stream.split(); |
| 226 | let (to_writer_tx, to_writer_rx) = mpsc::channel(1000); |
| 227 | |
| 228 | let init = role.init().await?; |
| 229 | let write = write( |
| 230 | sync_id.clone(), |
| 231 | sink.sink_map_err(anyhow::Error::from), |
| 232 | to_writer_rx, |
| 233 | init, |
| 234 | role.finish(), |
| 235 | PENDING_RANGES_LIMIT, |
| 236 | metrics.clone(), |
| 237 | ); |
| 238 | |
| 239 | let read = read(sync_id, stream, &mut role, to_writer_tx, metrics.clone()); |
| 240 | |
| 241 | // In a recon conversation there are 4 futures being polled: |
| 242 | // |
| 243 | // * Initiator Read |
| 244 | // * Initiator Write |
| 245 | // * Responder Read |
| 246 | // * Responder Write |
| 247 | // |
| 248 | // The following sequence occurs to end the conversation: |
| 249 | // |
| 250 | // 1. Initator Read determines there is no more work to do when there are no interests in |
| 251 | // common, or it reads the final [`ResponderMessage::RangeResponse`] from the Responder. |
| 252 | // 2. Initator Read sends [`ToWriter::Finish`] to the Initator Writer. |
| 253 | // 3. Initiator Writer sends the [`InitiatorMessage::Finished`] to the Responder and |
| 254 | // completes. |
| 255 | // 4. Responder Read receives the Finished message and completes, dropping the |
| 256 | // to_writer_tx sender. |
| 257 | // 5. Responder Write completes because the to_writer_rx has completed. This drops the substream |
| 258 | // to the remote which closes it. |
| 259 | // 6. Initiator Read sees the substream has closed and completes. |
| 260 | // |
| 261 | // This is analogous to the FIN -> FIN ACK sequence in TCP which ensures that boths ends of |
| 262 | // the conversation agree it has completed. This prevents a class of bugs where the Initiator |
| 263 | // may try and start a new conversation before Responder is aware the previous one has completed. |