working impl

Author: abr-egn

.evergreen/run-tests.sh

@@ -33,10 +33,13 @@ if [ "Windows_NT" == "$OS" ]; then
   export SSL_CERT_DIR=$(cygpath /etc/ssl/certs --windows)
 fi
 
-cargo_test ""
+CARGO_OPTIONS+=("--nocapture")
+export TEST_FILE=aggregate-write-readPreference.json
+cargo_test "test::spec::crud::run_unified"
+#cargo_test ""
 
 # cargo-nextest doesn't support doc tests
-RUST_BACKTRACE=1 cargo test --doc $(cargo_test_options)
-((CARGO_RESULT = ${CARGO_RESULT} || $?))
+#RUST_BACKTRACE=1 cargo test --doc $(cargo_test_options)
+#((CARGO_RESULT = ${CARGO_RESULT} || $?))
 
 exit $CARGO_RESULT

src/client.rs

@@ -447,7 +447,7 @@ impl Client {
         criteria: Option<&SelectionCriteria>,
     ) -> Result<ServerAddress> {
         let (server, _) = self
-            .select_server(criteria, "Test select server", None)
+            .select_server(criteria, "Test select server", None, false)
             .await?;
         Ok(server.address.clone())
     }
@@ -460,6 +460,7 @@ impl Client {
         #[allow(unused_variables)] // we only use the operation_name for tracing.
         operation_name: &str,
         deprioritized: Option<&ServerAddress>,
+        is_out_or_merge: bool,
     ) -> Result<(SelectedServer, SelectionCriteria)> {
         let criteria =
             criteria.unwrap_or(&SelectionCriteria::ReadPreference(ReadPreference::Primary));
@@ -488,10 +489,15 @@ impl Client {
         let mut watcher = self.inner.topology.watch();
         loop {
             let state = watcher.observe_latest();
-            for server in state.description.servers.values() {
-                eprintln!("at selection: {:?}", server.hello_response());
-            }
-            let effective_criteria = criteria; // TODO
+            let override_criteria;
+            let effective_criteria = if let Some(oc) =
+                Self::override_criteria(criteria, &state.description, is_out_or_merge)
+            {
+                override_criteria = oc;
+                &override_criteria
+            } else {
+                criteria
+            };
             let result = server_selection::attempt_to_select_server(
                 effective_criteria,
                 &state.description,
@@ -543,6 +549,38 @@ impl Client {
         }
     }
 
+    /// Check to see if selection criteria need to be overridden.  Currently only required for
+    /// aggregate operations with $merge/$out stages.
+    fn override_criteria(
+        criteria: &SelectionCriteria,
+        desc: &crate::sdam::TopologyDescription,
+        is_out_or_merge: bool,
+    ) -> Option<SelectionCriteria> {
+        if is_out_or_merge {
+            eprintln!("aggregate: checking override");
+        }
+        if !is_out_or_merge
+            || criteria == &SelectionCriteria::ReadPreference(ReadPreference::Primary)
+            || desc.topology_type() == crate::TopologyType::LoadBalanced
+        {
+            if is_out_or_merge {
+                eprintln!("aggregate: skipping override");
+            }
+            return None;
+        }
+        for server in desc.servers.values() {
+            let _ = dbg!(server.hello_response());
+            if let Ok(Some(v)) = server.max_wire_version() {
+                static SERVER_5_0_0_WIRE_VERSION: i32 = 13;
+                if v < SERVER_5_0_0_WIRE_VERSION {
+                    eprintln!("aggregate: overriding criteria");
+                    return Some(SelectionCriteria::ReadPreference(ReadPreference::Primary));
+                }
+            }
+        }
+        return None;
+    }
+
     #[cfg(all(test, feature = "dns-resolver"))]
     pub(crate) fn get_hosts(&self) -> Vec<String> {
         let watcher = self.inner.topology.watch();

src/client/executor.rs

@@ -326,6 +326,7 @@ impl Client {
                     selection_criteria,
                     op.name(),
                     retry.as_ref().map(|r| &r.first_server),
+                    op.is_out_or_merge(),
                 )
                 .await
             {
@@ -489,15 +490,15 @@ impl Client {
         connection: &mut PooledConnection,
         session: &mut Option<&mut ClientSession>,
         txn_number: Option<i64>,
-        _effective_critera: SelectionCriteria,
+        effective_criteria: SelectionCriteria,
     ) -> Result<Command> {
         let stream_description = connection.stream_description()?;
         let is_sharded = stream_description.initial_server_type == ServerType::Mongos;
         let mut cmd = op.build(stream_description)?;
         self.inner.topology.update_command_with_read_pref(
             connection.address(),
             &mut cmd,
-            op.selection_criteria(),
+            Some(&effective_criteria),
         );
 
         match session {
@@ -854,7 +855,7 @@ impl Client {
                 || server_type.is_data_bearing()
         }));
         let _ = self
-            .select_server(Some(&criteria), operation_name, None)
+            .select_server(Some(&criteria), operation_name, None, false)
             .await?;
         Ok(())
     }

src/operation.rs

@@ -76,7 +76,6 @@ pub(crate) use update::{Update, UpdateOrReplace};
 
 const SERVER_4_2_0_WIRE_VERSION: i32 = 8;
 const SERVER_4_4_0_WIRE_VERSION: i32 = 9;
-const _SERVER_5_0_0_WIRE_VERSION: i32 = 13;
 const SERVER_8_0_0_WIRE_VERSION: i32 = 25;
 // The maximum number of bytes that may be included in a write payload when auto-encryption is
 // enabled.
@@ -149,6 +148,9 @@ pub(crate) trait Operation {
     /// Updates this operation as needed for a retry.
     fn update_for_retry(&mut self);
 
+    /// Returns whether this is a $out or $merge aggregation operation.
+    fn is_out_or_merge(&self) -> bool;
+
     fn pinned_connection(&self) -> Option<&PinnedConnectionHandle>;
 
     fn name(&self) -> &str;
@@ -236,6 +238,11 @@ pub(crate) trait OperationWithDefaults: Send + Sync {
     /// Updates this operation as needed for a retry.
     fn update_for_retry(&mut self) {}
 
+    /// Returns whether this is a $out or $merge aggregation operation.
+    fn is_out_or_merge(&self) -> bool {
+        false
+    }
+
     fn pinned_connection(&self) -> Option<&PinnedConnectionHandle> {
         None
     }
@@ -288,6 +295,9 @@ where
     fn update_for_retry(&mut self) {
         self.update_for_retry()
     }
+    fn is_out_or_merge(&self) -> bool {
+        self.is_out_or_merge()
+    }
     fn pinned_connection(&self) -> Option<&PinnedConnectionHandle> {
         self.pinned_connection()
     }

src/operation/aggregate.rs

@@ -109,33 +109,6 @@ impl OperationWithDefaults for Aggregate {
         ))
     }
 
-    /*
-    fn update_for_topology(&mut self, topology: &crate::sdam::TopologyDescription) {
-        eprintln!("aggregate: update_for_topology");
-        if !self.is_out_or_merge()
-            || matches!(
-                self.selection_criteria(),
-                None | Some(SelectionCriteria::ReadPreference(ReadPreference::Primary))
-            )
-            || topology.topology_type() == TopologyType::LoadBalanced
-        {
-            eprintln!("aggregate: skipping topology update");
-            return;
-        }
-        for server in topology.servers.values() {
-            let _ = dbg!(server.hello_response());
-            if let Ok(Some(v)) = server.max_wire_version() {
-                if v < SERVER_5_0_0_WIRE_VERSION {
-                    eprintln!("aggregate: updating topology");
-                    self.options.get_or_insert_default().selection_criteria =
-                        Some(SelectionCriteria::ReadPreference(ReadPreference::Primary));
-                    break;
-                }
-            }
-        }
-    }
-    */
-
     fn selection_criteria(&self) -> Option<&SelectionCriteria> {
         self.options
             .as_ref()
@@ -161,10 +134,7 @@ impl OperationWithDefaults for Aggregate {
             Retryability::Read
         }
     }
-}
 
-impl Aggregate {
-    /// Returns whether this is a $out or $merge aggregation operation.
     fn is_out_or_merge(&self) -> bool {
         self.pipeline
             .last()

src/operation/raw_output.rs

@@ -68,6 +68,10 @@ impl<Op: Operation> Operation for RawOutput<Op> {
         self.0.update_for_retry()
     }
 
+    fn is_out_or_merge(&self) -> bool {
+        self.0.is_out_or_merge()
+    }
+
     fn pinned_connection(&self) -> Option<&crate::cmap::conn::PinnedConnectionHandle> {
         self.0.pinned_connection()
     }

src/operation/run_cursor_command.rs

@@ -79,6 +79,10 @@ impl Operation for RunCursorCommand<'_> {
         self.run_command.update_for_retry()
     }
 
+    fn is_out_or_merge(&self) -> bool {
+        self.run_command.is_out_or_merge()
+    }
+
     fn pinned_connection(&self) -> Option<&PinnedConnectionHandle> {
         self.run_command.pinned_connection()
     }

src/test/spec/crud.rs

@@ -43,13 +43,6 @@ async fn run_unified() {
          pre-5.0 server",
         "Requesting unacknowledged write with verboseResults is a client-side error",
         "Requesting unacknowledged write with ordered is a client-side error",
-        // TODO RUST-663: Unskip these tests.
-        "Aggregate with $out includes read preference for 5.0+ server",
-        "Aggregate with $out omits read preference for pre-5.0 server",
-        "Aggregate with $merge includes read preference for 5.0+ server",
-        "Aggregate with $merge omits read preference for pre-5.0 server",
-        "Database-level aggregate with $out omits read preference for pre-5.0 server",
-        "Database-level aggregate with $merge omits read preference for pre-5.0 server",
     ];
     // TODO: remove this manual skip when this test is fixed to skip on serverless
     if *SERVERLESS {