fix: last_zxid_seen in ConnectRequest is not set (#39)

Closes #32.

Author: kezhuw

src/client/mod.rs

@@ -2257,4 +2257,36 @@ mod tests {
             .unwrap_err())
         .is_equal_to(Error::BadArguments(&"directory node must not be sequential"));
     }
+
+    #[test_log::test(tokio::test)]
+    async fn session_last_zxid_seen() {
+        use testcontainers::clients::Cli as DockerCli;
+        use testcontainers::core::{Healthcheck, WaitFor};
+        use testcontainers::images::generic::GenericImage;
+
+        let healthcheck = Healthcheck::default()
+            .with_cmd(["./bin/zkServer.sh", "status"].iter())
+            .with_interval(Duration::from_secs(2))
+            .with_retries(60);
+        let image =
+            GenericImage::new("zookeeper", "3.9.0").with_healthcheck(healthcheck).with_wait_for(WaitFor::Healthcheck);
+        let docker = DockerCli::default();
+        let container = docker.run(image);
+        let endpoint = format!("127.0.0.1:{}", container.get_host_port(2181));
+
+        let client1 = Client::connector().detached().connect(&endpoint).await.unwrap();
+        client1.create("/n1", b"", &CreateMode::Persistent.with_acls(Acls::anyone_all())).await.unwrap();
+
+        let mut session = client1.into_session();
+
+        // Fail to connect with large zxid.
+        session.last_zxid = i64::MAX;
+        assert_that!(Client::connector().fail_eagerly().session(session.clone()).connect(&endpoint).await.unwrap_err())
+            .is_equal_to(Error::NoHosts);
+
+        // Succeed to connect with small zxid.
+        session.last_zxid = 0;
+        let client2 = Client::connector().fail_eagerly().session(session.clone()).connect(&endpoint).await.unwrap();
+        client2.create("/n2", b"", &CreateMode::Persistent.with_acls(Acls::anyone_all())).await.unwrap();
+    }
 }

src/session/mod.rs

@@ -102,7 +102,7 @@ impl Session {
 
             configured_connection_timeout: connection_timeout,
 
-            last_zxid: 0,
+            last_zxid: session.last_zxid,
             last_recv: now,
             last_send: now,
             last_ping: None,
@@ -286,15 +286,15 @@ impl Session {
     }
 
     fn handle_reply(&mut self, header: ReplyHeader, body: &[u8], depot: &mut Depot) -> Result<(), Error> {
-        if header.err == ErrorCode::SessionExpired.into() {
+        if header.err == ErrorCode::SessionExpired as i32 {
             return Err(Error::SessionExpired);
-        } else if header.err == ErrorCode::AuthFailed.into() {
+        } else if header.err == ErrorCode::AuthFailed as i32 {
             return Err(Error::AuthFailed);
         }
-        if header.xid == PredefinedXid::Notification.into() {
+        if header.xid == PredefinedXid::Notification as i32 {
             self.handle_notification(header.zxid, body, depot)?;
             return Ok(());
-        } else if header.xid == PredefinedXid::Ping.into() {
+        } else if header.xid == PredefinedXid::Ping as i32 {
             depot.pop_ping()?;
             if let Some(last_ping) = self.last_ping.take() {
                 let elapsed = Instant::now() - last_ping;
@@ -514,7 +514,7 @@ impl Session {
     fn send_connect(&self, depot: &mut Depot) {
         let request = ConnectRequest {
             protocol_version: 0,
-            last_zxid_seen: 0,
+            last_zxid_seen: self.last_zxid,
             timeout: self.session_timeout.as_millis() as i32,
             session_id: if self.session.readonly { 0 } else { self.session.id.0 },
             password: self.session.password.as_slice(),

src/session/types.rs

@@ -30,11 +30,18 @@ pub struct SessionInfo {
     #[derive_where(skip(Debug))]
     pub(crate) password: Vec<u8>,
     pub(crate) readonly: bool,
+    /// Only set through test otherwise 0.
+    ///
+    /// I thought to carry [Session::last_zxid] from [Client::session] to [Connector::session].
+    /// This way session reestablishment API has not major difference with internal reconnection.
+    /// But I think it is a ZooKeeper tradition to [Client::sync] after session reestablishment.
+    /// We probably should not challenge this.
+    pub(crate) last_zxid: i64,
 }
 
 impl SessionInfo {
     pub(crate) fn new(id: SessionId, password: Vec<u8>) -> Self {
-        Self { id, password, readonly: id.0 == 0 }
+        Self { id, password, readonly: id.0 == 0, last_zxid: 0 }
     }
 
     /// Session id.