fix: session disconnected due to unblock multiple unwatching

`xid` for unwatching requests is not set in unblocking. Response for
later unwatching will not find matching request as it was popped by
earlier response.

Closes #22.

Author: kezhuw

src/session/depot.rs

@@ -107,6 +107,11 @@ impl Depot {
         }
     }
 
+    fn push_request(&mut self, mut operation: SessionOperation) {
+        operation.request.set_xid(self.xid.next());
+        self.push_operation(Operation::Session(operation));
+    }
+
     pub fn pop_ping(&mut self) -> Result<(), Error> {
         self.pop_request(PredefinedXid::Ping.into()).map(|_| ())
     }
@@ -145,13 +150,13 @@ impl Depot {
         if *count == 0 {
             self.watching_paths.remove(&(path, mode));
             if let Some(operation) = self.unwatching_paths.remove(&(path, mode)) {
-                self.push_operation(Operation::Session(operation));
+                self.push_request(operation);
             }
             if self.has_watching_requests(path) {
                 return;
             }
             if let Some(operation) = self.unwatching_paths.remove(&(path, WatchMode::Any)) {
-                self.push_operation(Operation::Session(operation));
+                self.push_request(operation);
             }
         }
     }
@@ -166,7 +171,7 @@ impl Depot {
         self.cancel_unwatch(path, mode);
     }
 
-    pub fn push_session(&mut self, mut operation: SessionOperation) {
+    pub fn push_session(&mut self, operation: SessionOperation) {
         let info = operation.request.get_operation_info();
         log::debug!("ZooKeeper operation request: {:?}", info);
         if let (op_code, OpStat::Watch { path, mode }) = info {
@@ -185,8 +190,7 @@ impl Depot {
                 self.watching_paths.insert((path, mode), count);
             }
         }
-        operation.request.set_xid(self.xid.next());
-        self.push_operation(Operation::Session(operation));
+        self.push_request(operation);
     }
 
     pub fn push_remove_watch(&mut self, path: &str, mode: WatchMode, responser: StateResponser) {

src/session/mod.rs

@@ -203,7 +203,7 @@ impl Session {
     ) {
         if let Err(err) = self.serve_session(&sock, buf, depot, requester, unwatch_requester).await {
             self.resolve_serve_error(&err);
-            log::debug!("ZooKeeper session {} state {} error {}", self.session_id, self.session_state, err);
+            log::info!("ZooKeeper session {} state {} error {}", self.session_id, self.session_state, err);
             depot.error(&err);
         } else {
             self.change_state(SessionState::Disconnected);

tests/zookeeper.rs

@@ -1164,6 +1164,35 @@ async fn test_persistent_watcher_passive_remove() {
     assert_eq!(child_event.path, "/");
 }
 
+#[test_log::test(tokio::test)]
+async fn test_fail_watch_with_multiple_unwatching() {
+    let docker = DockerCli::default();
+    let zookeeper = docker.run(zookeeper_image());
+    let zk_port = zookeeper.get_host_port(2181);
+    let cluster = format!("127.0.0.1:{}", zk_port);
+
+    let client = zk::Client::connect(&cluster).await.unwrap();
+
+    let (_, exist_watcher1) = client.check_and_watch_stat("/a1").await.unwrap();
+    let (_, exist_watcher2) = client.check_and_watch_stat("/a2").await.unwrap();
+
+    let mut state_watcher = client.state_watcher();
+
+    let data_watching1 = client.get_and_watch_data("/a1");
+    let data_watching2 = client.get_and_watch_data("/a2");
+
+    drop(exist_watcher1);
+    drop(exist_watcher2);
+
+    assert_that!(data_watching1.await.unwrap_err()).is_equal_to(zk::Error::NoNode);
+    assert_that!(data_watching2.await.unwrap_err()).is_equal_to(zk::Error::NoNode);
+
+    select! {
+        state = state_watcher.changed() => panic!("expect no state update, but got {state}"),
+        _ = tokio::time::sleep(Duration::from_millis(10)) => {},
+    }
+}
+
 #[test_log::test(tokio::test)]
 async fn test_fail_watch_with_concurrent_passive_remove() {
     let docker = DockerCli::default();