chore: add debug log for diagnosis

Author: kezhuw

.github/workflows/ci.yml

@@ -5,7 +5,7 @@ on:
 
 env:
   RUSTFLAGS: -Dwarnings
-  RUST_BACKTRACE: 1
+  RUST_BACKTRACE: full
 
 jobs:
   fmt:
@@ -36,6 +36,8 @@ jobs:
   test:
     needs: [build]
     runs-on: ubuntu-latest
+    env:
+      RUST_LOG: debug
     steps:
     - uses: actions/checkout@v2
     - uses: Swatinem/rust-cache@v1
@@ -45,7 +47,7 @@ jobs:
           toolchain: stable
           override: true
     - name: Test code
-      run: cargo test
+      run: cargo test -- --nocapture
   coverage:
     if: github.event_name == 'pull_request' || (github.event_name == 'push' && github.ref_type == 'branch' && github.ref_name == 'master')
     needs: [test]

src/session/depot.rs

@@ -5,7 +5,7 @@ use hashbrown::HashMap;
 use strum::IntoEnumIterator;
 use tokio::net::TcpStream;
 
-use super::request::{MarshalledRequest, Operation, SessionOperation, StateResponser};
+use super::request::{MarshalledRequest, OpStat, Operation, SessionOperation, StateResponser};
 use super::types::WatchMode;
 use super::xid::Xid;
 use super::SessionId;
@@ -167,7 +167,9 @@ impl Depot {
     }
 
     pub fn push_session(&mut self, mut operation: SessionOperation) {
-        if let (op_code, Some((path, mode))) = operation.request.get_operation_info() {
+        let info = operation.request.get_operation_info();
+        log::debug!("ZooKeeper operation request: {:?}", info);
+        if let (op_code, OpStat::Watch { path, mode }) = info {
             let path = unsafe { std::mem::transmute::<&str, &'_ str>(path) };
             if op_code == OpCode::RemoveWatches {
                 if self.watching_paths.contains_key(&(path, mode))

src/session/mod.rs

@@ -19,6 +19,7 @@ use self::event::WatcherEvent;
 pub use self::request::{
     ConnectOperation,
     MarshalledRequest,
+    OpStat,
     Operation,
     SessionOperation,
     StateReceiver,
@@ -218,11 +219,11 @@ impl Session {
 
     fn handle_session_failure(&mut self, operation: SessionOperation, err: Error, depot: &mut Depot) {
         let SessionOperation { responser, request, .. } = operation;
-        let (op_code, watch_info) = request.get_operation_info();
-        if let Some((path, mode)) = watch_info {
-            if op_code != OpCode::RemoveWatches {
-                depot.fail_watch(path, mode);
-            }
+        let info = request.get_operation_info();
+        log::debug!("ZooKeeper operation unknown failure: {:?}, {:?}", info, err);
+        match info {
+            (op_code, OpStat::Watch { path, mode }) if op_code != OpCode::RemoveWatches => depot.fail_watch(path, mode),
+            _ => {},
         }
         responser.send(Err(err));
     }
@@ -233,11 +234,12 @@ impl Session {
         error_code: ErrorCode,
         depot: &mut Depot,
     ) -> (OpCode, WatchReceiver) {
-        let (op_code, watch_info) = request.get_operation_info();
-        if op_code == OpCode::RemoveWatches || watch_info.is_none() {
-            return (op_code, WatchReceiver::None);
-        }
-        let (path, mode) = watch_info.unwrap();
+        let info = request.get_operation_info();
+        log::debug!("ZooKeeper operation get reply: {:?}, {:?}", info, error_code);
+        let (op_code, path, mode) = match info {
+            (op_code, OpStat::Watch { path, mode }) if op_code != OpCode::RemoveWatches => (op_code, path, mode),
+            (op_code, _) => return (op_code, WatchReceiver::None),
+        };
         let watcher = self.watch_manager.create_watcher(path, mode, request.get_code(), error_code);
         if watcher.is_none() {
             depot.fail_watch(path, mode);

src/session/request.rs

@@ -15,6 +15,13 @@ use crate::record::{self, Record, StaticRecord};
 #[derive(Clone, Debug, Default)]
 pub struct MarshalledRequest(pub Vec<u8>);
 
+#[derive(Clone, Copy, Debug)]
+pub enum OpStat<'a> {
+    None,
+    Path(&'a str),
+    Watch { path: &'a str, mode: WatchMode },
+}
+
 impl MarshalledRequest {
     pub fn new(code: OpCode, body: &dyn Record) -> MarshalledRequest {
         let header = RequestHeader::with_code(code);
@@ -45,42 +52,64 @@ impl MarshalledRequest {
         xid_buf.put_i32(xid);
     }
 
-    pub fn get_operation_info(&self) -> (OpCode, Option<(&str, WatchMode)>) {
+    fn get_body(&self) -> &[u8] {
+        let offset = 4 + RequestHeader::record_len();
+        &self.0[offset..]
+    }
+
+    pub fn get_operation_info(&self) -> (OpCode, OpStat<'_>) {
         let op_code = self.get_code();
-        let watcher_info = match op_code {
+        let stat = match op_code {
+            OpCode::Create
+            | OpCode::Create2
+            | OpCode::CreateTtl
+            | OpCode::CreateContainer
+            | OpCode::SetData
+            | OpCode::Delete
+            | OpCode::DeleteContainer
+            | OpCode::GetACL
+            | OpCode::SetACL
+            | OpCode::Sync
+            | OpCode::Check
+            | OpCode::CheckWatches
+            | OpCode::GetEphemerals
+            | OpCode::GetAllChildrenNumber => {
+                let mut body = self.get_body();
+                let server_path = record::deserialize::<&str>(&mut body).unwrap();
+                OpStat::Path(server_path)
+            },
             OpCode::GetData
             | OpCode::Exists
             | OpCode::GetChildren
             | OpCode::GetChildren2
             | OpCode::AddWatch
             | OpCode::RemoveWatches => {
-                let offset = 4 + RequestHeader::record_len();
-                let mut body = &self.0[offset..];
+                let mut body = self.get_body();
                 let server_path = record::deserialize::<&str>(&mut body).unwrap();
                 if op_code == OpCode::AddWatch || op_code == OpCode::RemoveWatches {
                     body.advance(3);
                 }
                 let watch = body.get_u8();
                 if op_code == OpCode::AddWatch {
                     let add_mode = AddWatchMode::try_from(watch as i32).unwrap();
-                    Some((server_path, WatchMode::from(add_mode)))
+                    OpStat::Watch { path: server_path, mode: WatchMode::from(add_mode) }
                 } else if op_code == OpCode::RemoveWatches {
-                    Some((server_path, WatchMode::try_from(watch as i32).unwrap()))
+                    OpStat::Watch { path: server_path, mode: WatchMode::try_from(watch as i32).unwrap() }
                 } else if watch == 1 {
                     let mode = if op_code == OpCode::GetData || op_code == OpCode::Exists {
                         WatchMode::Data
                     } else {
                         WatchMode::Child
                     };
-                    Some((server_path, mode))
+                    OpStat::Watch { path: server_path, mode }
                 } else {
                     assert!(watch == 0);
-                    None
+                    OpStat::Path(server_path)
                 }
             },
-            _ => None,
+            _ => OpStat::None,
         };
-        (op_code, watcher_info)
+        (op_code, stat)
     }
 
     pub fn is_empty(&self) -> bool {