RUST-2219 Cherrypick domain name parsing fix into 3.2.x

.evergreen/check-clippy.sh

@@ -5,7 +5,7 @@ set -o errexit
 source ./.evergreen/env.sh
 
 # Pin clippy to the latest version. This should be updated when new versions of Rust are released.
-CLIPPY_VERSION=1.83.0
+CLIPPY_VERSION=1.85.0
 
 rustup install $CLIPPY_VERSION
 

.evergreen/config.yml

@@ -271,16 +271,16 @@ buildvariants:
     tasks:
       - test-plain-auth
 
-  - name: serverless
-    display_name: "Serverless"
-    run_on:
-      - rhel80-small
-    expansions:
-      LIBMONGOCRYPT_OS: rhel-80-64-bit
-      AUTH: auth
-      SSL: ssl
-    tasks:
-      - serverless-task-group
+  # - name: serverless
+  #   display_name: "Serverless"
+  #   run_on:
+  #     - rhel80-small
+  #   expansions:
+  #     LIBMONGOCRYPT_OS: rhel-80-64-bit
+  #     AUTH: auth
+  #     SSL: ssl
+  #   tasks:
+  #     - serverless-task-group
 
   - name: oidc-linux
     display_name: "OIDC Linux"
@@ -675,13 +675,21 @@ task_groups:
       - func: init test-results
       - func: make files executable
       - func: install rust
+      - command: ec2.assume_role
+        params:
+          role_arn: ${aws_test_secrets_role}
+          duration_seconds: 3600
       - command: subprocess.exec
         params:
           binary: bash
           env:
             AZUREOIDC_VMNAME_PREFIX: "RUST_DRIVER"
           args:
             - ${DRIVERS_TOOLS}/.evergreen/auth_oidc/azure/create-and-setup-vm.sh
+          include_expansions_in_env:
+            - AWS_ACCESS_KEY_ID
+            - AWS_SECRET_ACCESS_KEY
+            - AWS_SESSION_TOKEN
     teardown_task:
       - command: subprocess.exec
         params:
@@ -702,13 +710,21 @@ task_groups:
       - func: init test-results
       - func: make files executable
       - func: install rust
+      - command: ec2.assume_role
+        params:
+          role_arn: ${aws_test_secrets_role}
+          duration_seconds: 3600
       - command: subprocess.exec
         params:
           binary: bash
           env:
             GCPOIDC_VMNAME_PREFIX: "RUST_DRIVER"
           args:
             - ${DRIVERS_TOOLS}/.evergreen/auth_oidc/gcp/setup.sh
+          include_expansions_in_env:
+            - AWS_ACCESS_KEY_ID
+            - AWS_SECRET_ACCESS_KEY
+            - AWS_SESSION_TOKEN
     teardown_task:
       - command: subprocess.exec
         params:
@@ -729,11 +745,19 @@ task_groups:
       - func: init test-results
       - func: make files executable
       - func: install rust
+      - command: ec2.assume_role
+        params:
+          role_arn: ${aws_test_secrets_role}
+          duration_seconds: 3600
       - command: subprocess.exec
         params:
           binary: bash
           args:
             - ${DRIVERS_TOOLS}/.evergreen/auth_oidc/k8s/setup.sh
+          include_expansions_in_env:
+            - AWS_ACCESS_KEY_ID
+            - AWS_SECRET_ACCESS_KEY
+            - AWS_SESSION_TOKEN
     teardown_task:
       - command: subprocess.exec
         params:

.evergreen/install-dependencies.sh

@@ -21,6 +21,15 @@ for arg; do
   if [ $arg == "rust" ]; then
     curl https://sh.rustup.rs -sSf | sh -s -- -y --no-modify-path $DEFAULT_HOST_OPTIONS
 
+    # Cygwin has a bug with reporting symlink paths that breaks rustup; see 
+    # https://github.com/rust-lang/rustup/issues/4239.  This works around it by replacing the
+    # symlinks with copies.
+    if [ "Windows_NT" == "$OS" ]; then
+      pushd ${CARGO_HOME}/bin
+      python3 ../../.evergreen/unsymlink.py
+      popd
+    fi
+
     # This file is not created by default on Windows
     echo 'export PATH="$PATH:${CARGO_HOME}/bin"' >>${CARGO_HOME}/env
     echo "export CARGO_NET_GIT_FETCH_WITH_CLI=true" >>${CARGO_HOME}/env

.evergreen/unsymlink.py

@@ -0,0 +1,19 @@
+import os
+import shutil
+
+found = []
+for entry in os.scandir():
+    if not entry.is_symlink():
+        print(f"Skipping {entry.name}: not a symlink")
+        continue
+    target = os.readlink(entry.name)
+    if target != "rustup.exe":
+        print(f"Skipping {entry.name}: not rustup.exe")
+        continue
+    print(f"Found {entry.name}")
+    found.append(entry.name)
+
+for name in found:
+    print(f"Replacing {name} symlink with copy")
+    os.remove(name)
+    shutil.copy2("rustup.exe", name)

Cargo.toml

@@ -72,7 +72,7 @@ tracing-unstable = ["dep:tracing", "dep:log"]
 async-trait = "0.1.42"
 base64 = "0.13.0"
 bitflags = "1.1.0"
-bson = { git = "https://github.com/mongodb/bson-rust", branch = "main", version = "2.14.0" }
+bson = "2.14.0"
 chrono = { version = "0.4.7", default-features = false, features = [
     "clock",
     "std",
@@ -91,7 +91,7 @@ hmac = "0.12.1"
 once_cell = "1.19.0"
 log = { version = "0.4.17", optional = true }
 md-5 = "0.10.1"
-mongocrypt = { git = "https://github.com/mongodb/libmongocrypt-rust.git", branch = "main", optional = true, version = "0.2.1" }
+mongocrypt = { optional = true, version = "0.2.1" }
 mongodb-internal-macros = { path = "macros", version = "3.2.3" }
 num_cpus = { version = "1.13.1", optional = true }
 openssl = { version = "0.10.38", optional = true }

deny.toml

@@ -35,6 +35,7 @@ allow = [
     "Apache-2.0",
     "Apache-2.0 WITH LLVM-exception",
     "CC0-1.0",
+    "CDLA-Permissive-2.0",
     "ISC",
     "OpenSSL",
     "BSD-2-Clause",

src/action/find_and_modify.rs

@@ -27,12 +27,12 @@ use crate::{
 use super::{action_impl, deeplink, export_doc, option_setters, options_doc};
 
 impl<T: DeserializeOwned + Send + Sync> Collection<T> {
-    async fn find_and_modify<'a>(
+    async fn find_and_modify(
         &self,
         filter: Document,
         modification: Modification,
         mut options: Option<FindAndModifyOptions>,
-        session: Option<&'a mut ClientSession>,
+        session: Option<&mut ClientSession>,
     ) -> Result<Option<T>> {
         resolve_write_concern_with_session!(self, options, session.as_ref())?;
 

src/client.rs

@@ -261,7 +261,7 @@ impl Client {
             .read()
             .await
             .as_ref()
-            .map_or(false, |cs| cs.exec().mongocryptd_spawned())
+            .is_some_and(|cs| cs.exec().mongocryptd_spawned())
     }
 
     #[cfg(all(test, feature = "in-use-encryption"))]
@@ -271,7 +271,7 @@ impl Client {
             .read()
             .await
             .as_ref()
-            .map_or(false, |cs| cs.exec().has_mongocryptd_client())
+            .is_some_and(|cs| cs.exec().has_mongocryptd_client())
     }
 
     fn test_command_event_channel(&self) -> Option<&options::TestEventSender> {

src/client/action/perf.rs

@@ -10,7 +10,7 @@ impl<'a> Action for crate::action::WarmConnectionPool<'a> {
             .inner
             .options
             .min_pool_size
-            .map_or(false, |s| s > 0)
+            .is_some_and(|size| size > 0)
         {
             // No-op when min_pool_size is zero.
             return;

src/client/auth/aws.rs

@@ -542,7 +542,7 @@ impl ServerFirst {
                 MECH_NAME,
                 "sts host must be non-empty",
             ))
-        } else if self.sts_host.as_bytes().len() > 255 {
+        } else if self.sts_host.len() > 255 {
             Err(Error::authentication_error(
                 MECH_NAME,
                 "sts host cannot be more than 255 bytes",

src/client/auth/oidc.rs

@@ -968,7 +968,7 @@ pub(super) fn validate_credential(credential: &Credential) -> Result<()> {
     if credential
         .source
         .as_ref()
-        .map_or(false, |s| s != "$external")
+        .is_some_and(|source| source != "$external")
     {
         return Err(Error::invalid_argument(format!(
             "source must be $external for {} authentication, found: {:?}",

src/client/csfle/client_encryption.rs

@@ -183,7 +183,7 @@ impl ClientEncryption {
 
     /// Decrypts an encrypted value (BSON binary of subtype 6).
     /// Returns the original BSON value.
-    pub async fn decrypt<'a>(&self, value: RawBinaryRef<'a>) -> Result<bson::RawBson> {
+    pub async fn decrypt(&self, value: RawBinaryRef<'_>) -> Result<bson::RawBson> {
         if value.subtype != BinarySubtype::Encrypted {
             return Err(Error::invalid_argument(format!(
                 "Invalid binary subtype for decrypt: expected {:?}, got {:?}",

src/client/options/test.rs

@@ -154,9 +154,7 @@ async fn run_tests(path: &[&str], skipped_files: &[&str]) {
 
                         let (_, actual_value) = actual_options
                             .iter()
-                            .find(|(actual_key, _)| {
-                                actual_key.to_ascii_lowercase() == expected_key.to_ascii_lowercase()
-                            })
+                            .find(|(actual_key, _)| actual_key.eq_ignore_ascii_case(expected_key))
                             .unwrap_or_else(|| {
                                 panic!(
                                     "{}: parsed options missing {} key",

src/client/session/action.rs

@@ -99,7 +99,7 @@ impl<'a> Action for StartTransaction<&'a mut ClientSession> {
     }
 }
 
-impl<'a> StartTransaction<&'a mut ClientSession> {
+impl StartTransaction<&mut ClientSession> {
     /// Starts a transaction, runs the given callback, and commits or aborts the transaction.
     /// Transient transaction errors will cause the callback or the commit to be retried;
     /// other errors will cause the transaction to be aborted and the error returned to the

src/cmap/conn/stream_description.rs

@@ -76,6 +76,6 @@ impl StreamDescription {
     pub(crate) fn supports_retryable_writes(&self) -> bool {
         self.initial_server_type != ServerType::Standalone
             && self.logical_session_timeout.is_some()
-            && self.max_wire_version.map_or(false, |version| version >= 6)
+            && self.max_wire_version.is_some_and(|version| version >= 6)
     }
 }

src/cmap/establish/handshake.rs

@@ -217,15 +217,15 @@ impl RuntimeEnvironment {
 }
 
 fn var_set(name: &str) -> bool {
-    env::var_os(name).map_or(false, |v| !v.is_empty())
+    env::var_os(name).is_some_and(|v| !v.is_empty())
 }
 
 impl FaasEnvironmentName {
     pub(crate) fn new() -> Option<Self> {
         use FaasEnvironmentName::*;
         let mut found: Option<Self> = None;
         let lambda_env = env::var_os("AWS_EXECUTION_ENV")
-            .map_or(false, |v| v.to_string_lossy().starts_with("AWS_Lambda_"));
+            .is_some_and(|v| v.to_string_lossy().starts_with("AWS_Lambda_"));
         if lambda_env || var_set("AWS_LAMBDA_RUNTIME_API") {
             found = Some(AwsLambda);
         }

src/error.rs

@@ -290,6 +290,14 @@ impl Error {
         .into()
     }
 
+    #[cfg(feature = "dns-resolver")]
+    pub(crate) fn from_resolve_proto_error(error: hickory_proto::error::ProtoError) -> Self {
+        ErrorKind::DnsResolve {
+            message: error.to_string(),
+        }
+        .into()
+    }
+
     pub(crate) fn is_non_timeout_network_error(&self) -> bool {
         matches!(self.kind.as_ref(), ErrorKind::Io(ref io_err) if io_err.kind() != std::io::ErrorKind::TimedOut)
     }

src/operation/aggregate/change_stream.rs

@@ -61,7 +61,7 @@ impl OperationWithDefaults for ChangeStreamAggregate {
                     .start_at_operation_time
                     .as_ref()
                     .or(data.initial_operation_time.as_ref());
-                if saved_time.is_some() && description.max_wire_version.map_or(false, |v| v >= 7) {
+                if saved_time.is_some() && description.max_wire_version.is_some_and(|v| v >= 7) {
                     new_opts.start_at_operation_time = saved_time.cloned();
                 }
             }
@@ -109,7 +109,7 @@ impl OperationWithDefaults for ChangeStreamAggregate {
 
         let description = context.connection.stream_description()?;
         if self.args.options.as_ref().map_or(true, has_no_time)
-            && description.max_wire_version.map_or(false, |v| v >= 7)
+            && description.max_wire_version.is_some_and(|v| v >= 7)
             && spec.initial_buffer.is_empty()
             && spec.post_batch_resume_token.is_none()
         {

src/operation/create_indexes.rs

@@ -48,7 +48,7 @@ impl OperationWithDefaults for CreateIndexes {
             && self
                 .options
                 .as_ref()
-                .map_or(false, |options| options.commit_quorum.is_some())
+                .is_some_and(|options| options.commit_quorum.is_some())
         {
             return Err(ErrorKind::InvalidArgument {
                 message: "Specifying a commit quorum to create_index(es) is not supported on \

src/runtime/resolver.rs

@@ -2,7 +2,7 @@ use hickory_resolver::{
     config::ResolverConfig,
     error::ResolveErrorKind,
     lookup::{SrvLookup, TxtLookup},
-    IntoName,
+    Name,
 };
 
 use crate::error::{Error, Result};
@@ -25,17 +25,19 @@ impl AsyncResolver {
 }
 
 impl AsyncResolver {
-    pub async fn srv_lookup<N: IntoName>(&self, query: N) -> Result<SrvLookup> {
+    pub async fn srv_lookup(&self, query: &str) -> Result<SrvLookup> {
+        let name = Name::from_str_relaxed(query).map_err(Error::from_resolve_proto_error)?;
         let lookup = self
             .resolver
-            .srv_lookup(query)
+            .srv_lookup(name)
             .await
             .map_err(Error::from_resolve_error)?;
         Ok(lookup)
     }
 
-    pub async fn txt_lookup<N: IntoName>(&self, query: N) -> Result<Option<TxtLookup>> {
-        let lookup_result = self.resolver.txt_lookup(query).await;
+    pub async fn txt_lookup(&self, query: &str) -> Result<Option<TxtLookup>> {
+        let name = Name::from_str_relaxed(query).map_err(Error::from_resolve_proto_error)?;
+        let lookup_result = self.resolver.txt_lookup(name).await;
         match lookup_result {
             Ok(lookup) => Ok(Some(lookup)),
             Err(e) => match e.kind() {

src/sdam/description/topology/server_selection/test.rs

@@ -75,10 +75,7 @@ struct TestServerDescription {
 
 impl TestServerDescription {
     fn into_server_description(self) -> Option<ServerDescription> {
-        let server_type = match self.server_type.into_server_type() {
-            Some(server_type) => server_type,
-            None => return None,
-        };
+        let server_type = self.server_type.into_server_type()?;
 
         let server_address = ServerAddress::parse(self.address).ok()?;
         let tags = self.tags;

src/sdam/srv_polling.rs

@@ -33,10 +33,7 @@ impl SrvPollingMonitor {
         topology_watcher: TopologyWatcher,
         mut client_options: ClientOptions,
     ) -> Option<Self> {
-        let initial_info = match client_options.original_srv_info.take() {
-            Some(info) => info,
-            None => return None,
-        };
+        let initial_info = client_options.original_srv_info.take()?;
 
         Some(Self {
             initial_hostname: initial_info.hostname,

src/test/client.rs

@@ -931,7 +931,7 @@ async fn warm_connection_pool() {
     client.list_database_names().await.unwrap();
 }
 
-async fn get_end_session_event_count<'a>(event_stream: &mut EventStream<'a, Event>) -> usize {
+async fn get_end_session_event_count(event_stream: &mut EventStream<'_, Event>) -> usize {
     // Use collect_successful_command_execution to assert that the call to endSessions succeeded.
     event_stream
         .collect_successful_command_execution(Duration::from_millis(500), "endSessions")

src/test/csfle.rs

@@ -217,7 +217,7 @@ static EXTRA_OPTIONS: Lazy<Document> =
 static KV_NAMESPACE: Lazy<Namespace> =
     Lazy::new(|| Namespace::from_str("keyvault.datakeys").unwrap());
 static DISABLE_CRYPT_SHARED: Lazy<bool> =
-    Lazy::new(|| env::var("DISABLE_CRYPT_SHARED").map_or(false, |s| s == "true"));
+    Lazy::new(|| env::var("DISABLE_CRYPT_SHARED").is_ok_and(|s| s == "true"));
 
 fn check_env(name: &str, kmip: bool) -> bool {
     if env::var("CSFLE_LOCAL_KEY").is_err() {