RUST-360 Streaming monitoring protocol (#721)

This also fixes RUST-1464

Author: patrickfreed

Cargo.toml

@@ -88,7 +88,6 @@ md-5 = "0.10.1"
 mongocrypt = { git = "https://github.com/mongodb/libmongocrypt-rust.git", branch = "main", optional = true }
 openssl = { version = "0.10.38", optional = true }
 openssl-probe = { version = "0.1.5", optional = true }
-os_info = { version = "3.0.1", default-features = false }
 percent-encoding = "2.0.0"
 rand = { version = "0.8.3", features = ["small_rng"] }
 rayon = { version = "1.5.3", optional = true }
@@ -170,6 +169,7 @@ serde = { version = "*", features = ["rc"] }
 serde_json = "1.0.64"
 semver = "1.0.0"
 time = "0.3.9"
+regex = "1.6.0"
 
 [package.metadata.docs.rs]
 rustdoc-args = ["--cfg", "docsrs"]

src/client/csfle/state_machine.rs

@@ -15,10 +15,9 @@ use tokio::{
 
 use crate::{
     client::{options::ServerAddress, WeakClient},
-    cmap::options::StreamOptions,
     error::{Error, Result},
     operation::{RawOutput, RunCommand},
-    runtime::{AsyncStream, Process},
+    runtime::{AsyncStream, Process, TlsConfig},
     Client,
     Namespace,
 };
@@ -165,13 +164,9 @@ impl CryptExecutor {
                                 .and_then(|tls| tls.get(&provider))
                                 .cloned()
                                 .unwrap_or_default();
-                            let mut stream = AsyncStream::connect(
-                                StreamOptions::builder()
-                                    .address(addr)
-                                    .tls_options(tls_options)
-                                    .build(),
-                            )
-                            .await?;
+                            let mut stream =
+                                AsyncStream::connect(addr, Some(&TlsConfig::new(tls_options)?))
+                                    .await?;
                             stream.write_all(kms_ctx.message()?).await?;
                             let mut buf = vec![0];
                             while kms_ctx.bytes_needed() > 0 {

src/client/executor.rs

@@ -620,6 +620,7 @@ impl Client {
         let raw_cmd = RawCommand {
             name: cmd_name.clone(),
             target_db,
+            exhaust_allowed: false,
             bytes: serialized,
         };
 

src/client/mod.rs

@@ -409,22 +409,21 @@ impl Client {
             if let Some(server) = server_selection::attempt_to_select_server(
                 criteria,
                 &state.description,
-                &state.servers,
+                &state.servers(),
             )? {
                 return Ok(server);
             }
 
-            self.inner.topology.request_update();
+            watcher.request_immediate_check();
 
             let change_occurred = start_time.elapsed() < timeout
                 && watcher
                     .wait_for_update(timeout - start_time.elapsed())
                     .await;
             if !change_occurred {
                 return Err(ErrorKind::ServerSelection {
-                    message: self
-                        .inner
-                        .topology
+                    message: state
+                        .description
                         .server_selection_timeout_error_message(criteria),
                 }
                 .into());
@@ -433,12 +432,13 @@ impl Client {
     }
 
     #[cfg(all(test, not(feature = "sync"), not(feature = "tokio-sync")))]
-    pub(crate) async fn get_hosts(&self) -> Vec<String> {
+    pub(crate) fn get_hosts(&self) -> Vec<String> {
         let watcher = self.inner.topology.watch();
         let state = watcher.peek_latest();
 
-        let servers = state.servers.keys();
-        servers
+        state
+            .servers()
+            .keys()
             .map(|stream_address| format!("{}", stream_address))
             .collect()
     }
@@ -458,6 +458,11 @@ impl Client {
             .clone()
     }
 
+    #[cfg(test)]
+    pub(crate) fn topology(&self) -> &crate::sdam::Topology {
+        &self.inner.topology
+    }
+
     #[cfg(feature = "csfle")]
     pub(crate) fn weak(&self) -> WeakClient {
         WeakClient {
@@ -475,11 +480,6 @@ impl Client {
         })
         .ok()
     }
-
-    #[cfg(test)]
-    pub(crate) fn topology(&self) -> &Topology {
-        &self.inner.topology
-    }
 }
 
 #[cfg(feature = "csfle")]

src/client/options/mod.rs

@@ -33,7 +33,7 @@ use crate::{
     error::{Error, ErrorKind, Result},
     event::{cmap::CmapEventHandler, command::CommandEventHandler, sdam::SdamEventHandler},
     options::ReadConcernLevel,
-    sdam::{DEFAULT_HEARTBEAT_FREQUENCY, IDLE_WRITE_PERIOD, MIN_HEARTBEAT_FREQUENCY},
+    sdam::{verify_max_staleness, DEFAULT_HEARTBEAT_FREQUENCY, MIN_HEARTBEAT_FREQUENCY},
     selection_criteria::{ReadPreference, SelectionCriteria, TagSet},
     srv::{OriginalSrvInfo, SrvResolver},
 };
@@ -568,7 +568,7 @@ pub struct ClientOptions {
 #[derive(Debug, Clone, Default)]
 pub(crate) struct TestOptions {
     /// Override MIN_HEARTBEAT_FREQUENCY.
-    pub(crate) heartbeat_freq: Option<Duration>,
+    pub(crate) min_heartbeat_freq: Option<Duration>,
 
     /// Disable server and SRV-polling monitor threads.
     pub(crate) disable_monitoring_threads: bool,
@@ -1318,17 +1318,10 @@ impl ClientOptions {
 
         if let Some(SelectionCriteria::ReadPreference(ref rp)) = self.selection_criteria {
             if let Some(max_staleness) = rp.max_staleness() {
-                let smallest_max_staleness = std::cmp::max(
-                    Duration::from_secs(90),
-                    self.heartbeat_freq.unwrap_or(DEFAULT_HEARTBEAT_FREQUENCY) + IDLE_WRITE_PERIOD,
-                );
-
-                if max_staleness < smallest_max_staleness {
-                    return Err(Error::invalid_argument(format!(
-                        "invalid maxStaleness value: must be at least {} seconds",
-                        smallest_max_staleness.as_secs()
-                    )));
-                }
+                verify_max_staleness(
+                    max_staleness,
+                    self.heartbeat_freq.unwrap_or(DEFAULT_HEARTBEAT_FREQUENCY),
+                )?;
             }
         }
 

src/cmap/conn/command.rs

@@ -1,5 +1,3 @@
-use std::time::Duration;
-
 use bson::{RawDocument, RawDocumentBuf};
 use serde::{de::DeserializeOwned, Deserialize, Serialize};
 
@@ -20,6 +18,8 @@ use crate::{
 pub(crate) struct RawCommand {
     pub(crate) name: String,
     pub(crate) target_db: String,
+    /// Whether or not the server may respond to this command multiple times via the moreToComeBit.
+    pub(crate) exhaust_allowed: bool,
     pub(crate) bytes: Vec<u8>,
 }
 
@@ -38,6 +38,9 @@ pub(crate) struct Command<T = Document> {
     #[serde(skip)]
     pub(crate) name: String,
 
+    #[serde(skip)]
+    pub(crate) exhaust_allowed: bool,
+
     #[serde(flatten)]
     pub(crate) body: T,
 
@@ -71,6 +74,7 @@ impl<T> Command<T> {
         Self {
             name,
             target_db,
+            exhaust_allowed: false,
             body,
             lsid: None,
             cluster_time: None,
@@ -93,6 +97,7 @@ impl<T> Command<T> {
         Self {
             name,
             target_db,
+            exhaust_allowed: false,
             body,
             lsid: None,
             cluster_time: None,
@@ -245,15 +250,14 @@ impl RawCommandResponse {
             .map_err(|_| Error::invalid_authentication_response(mechanism_name))
     }
 
-    pub(crate) fn into_hello_reply(self, round_trip_time: Duration) -> Result<HelloReply> {
+    pub(crate) fn into_hello_reply(self) -> Result<HelloReply> {
         match self.body::<CommandResponse<HelloCommandResponse>>() {
             Ok(response) if response.is_success() => {
                 let server_address = self.source_address().clone();
                 let cluster_time = response.cluster_time().cloned();
                 Ok(HelloReply {
                     server_address,
                     command_response: response.body,
-                    round_trip_time,
                     cluster_time,
                     raw_command_response: self.into_raw_document_buf(),
                 })