fix: AuthFailed not reported in Connector::connect

`AuthFailed` is a terminal error. It should be reported eagerly if
possible. Currently, there are two problems:
* `AuthFailed` is not reported in connection establishment.
* Session could immediately fail after successful `connect`.

Author: kezhuw

src/client/mod.rs

@@ -10,7 +10,7 @@ use const_format::formatcp;
 use either::{Either, Left, Right};
 use ignore_result::Ignore;
 use thiserror::Error;
-use tokio::sync::{mpsc, watch};
+use tokio::sync::mpsc;
 use tracing::field::display;
 use tracing::{instrument, Span};
 
@@ -248,9 +248,8 @@ impl Client {
         session: SessionInfo,
         timeout: Duration,
         requester: mpsc::UnboundedSender<SessionOperation>,
-        state_receiver: watch::Receiver<SessionState>,
+        state_watcher: StateWatcher,
     ) -> Client {
-        let state_watcher = StateWatcher::new(state_receiver);
         Client { chroot, version, session, session_timeout: timeout, requester, state_watcher }
     }
 
@@ -1672,11 +1671,14 @@ impl Connector {
         let (sender, receiver) = mpsc::unbounded_channel();
         let session_info = session.session.clone();
         let session_timeout = session.session_timeout;
+        let mut state_watcher = StateWatcher::new(state_receiver);
+        // Consume all state changes so far.
+        state_watcher.state();
         tokio::spawn(async move {
             session.serve(endpoints, conn, buf, connecting_depot, receiver).await;
         });
         let client =
-            Client::new(chroot.to_owned(), self.server_version, session_info, session_timeout, sender, state_receiver);
+            Client::new(chroot.to_owned(), self.server_version, session_info, session_timeout, sender, state_watcher);
         Ok(client)
     }
 

src/error.rs

@@ -100,6 +100,10 @@ pub struct OtherError {
 }
 
 impl Error {
+    pub(crate) fn is_terminated(&self) -> bool {
+        matches!(self, Self::NoHosts | Self::SessionExpired | Self::AuthFailed | Self::ClientClosed)
+    }
+
     pub(crate) fn has_no_data_change(&self) -> bool {
         match self {
             Self::NoNode

src/session/mod.rs

@@ -291,7 +291,9 @@ impl Session {
     fn handle_reply(&mut self, header: ReplyHeader, body: &[u8], depot: &mut Depot) -> Result<(), Error> {
         if header.err == ErrorCode::SessionExpired as i32 {
             return Err(Error::SessionExpired);
-        } else if header.err == ErrorCode::AuthFailed as i32 {
+        } else if header.err == ErrorCode::AuthFailed as i32
+            || header.err == ErrorCode::SessionClosedRequireSaslAuth as i32
+        {
             return Err(Error::AuthFailed);
         }
         if header.xid == PredefinedXid::Notification as i32 {
@@ -423,6 +425,7 @@ impl Session {
         buf: &mut Vec<u8>,
         depot: &mut Depot,
     ) -> Result<(), Error> {
+        let mut pinged = false;
         let mut tick = time::interval(self.tick_timeout);
         tick.set_missed_tick_behavior(time::MissedTickBehavior::Skip);
         while !(self.session_state.is_connected() && depot.is_empty()) {
@@ -441,6 +444,13 @@ impl Session {
                     }
                 },
             }
+            if self.session_state.is_connected() && !pinged {
+                // Send opcode other than `auth` and `sasl` to get possible AuthFailed if
+                // "zookeeper.enforce.auth.enabled".
+                pinged = true;
+                self.send_ping(depot, Instant::now());
+                depot.write_operations(conn)?;
+            }
         }
         Ok(())
     }
@@ -571,6 +581,7 @@ impl Session {
         self.last_send = Instant::now();
         self.last_recv = self.last_send;
         self.last_ping = None;
+        self.change_state(SessionState::Disconnected);
         match self.serve_connecting(&mut conn, buf, depot).await {
             Err(err) => {
                 warn!("fails to establish session to {} due to {}", endpoint, err);
@@ -602,7 +613,7 @@ impl Session {
             Err(err) => err,
             Ok(conn) => return Ok(conn),
         };
-        while last_error != Error::NoHosts && last_error != Error::SessionExpired {
+        while !last_error.is_terminated() {
             if deadline.elapsed() {
                 return Err(Error::Timeout);
             }

tests/zookeeper.rs

@@ -390,7 +390,13 @@ impl Cluster {
                 ..Default::default()
             };
             match tls.as_ref() {
-                None => {},
+                None => {
+                    if standalone && !configs.is_empty() {
+                        configs += "clientPort=2181\n";
+                        configs += "initLimit=5\n";
+                        configs += "syncLimit=2\n";
+                    }
+                },
                 Some(tls) => {
                     configs += r"
 ssl.clientAuth=need
@@ -1181,6 +1187,22 @@ async fn test_no_auth() {
     assert_eq!(no_auth_client.set_data("/acl_test_2", b"set_my_data", None).await.unwrap_err(), zk::Error::NoAuth);
 }
 
+#[test_log::test(tokio::test)]
+#[should_panic(expected = "AuthFailed")]
+async fn test_auth_failed() {
+    let cluster = Cluster::with_options(ClusterOptions {
+        configs: vec![
+            "authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider",
+            "enforce.auth.enabled=true",
+            "enforce.auth.schemes=sasl",
+            "sessionRequireClientSASLAuth=true",
+        ],
+        ..Default::default()
+    })
+    .await;
+    cluster.client(None).await;
+}
+
 #[test_log::test(tokio::test)]
 async fn test_delete() {
     let cluster = Cluster::new().await;