refactor: merge connecting and serving depots into one

Those two are there from initial day to keep `pending_authes` cross
reconnection. But turns out, it is verbose and confusing. Treating
`pending_authes` specially is enough for us to go.

By merging them into one, we could share partial state among connection
states, e.g. `xid`.

Author: kezhuw

src/client/mod.rs

@@ -1671,16 +1671,16 @@ impl Connector {
             endpoints.cycle();
         }
         let mut buf = Vec::with_capacity(4096);
-        let mut connecting_depot = Depot::for_connecting();
-        let conn = session.start(&mut endpoints, &mut buf, &mut connecting_depot).await?;
+        let mut depot = Depot::new();
+        let conn = session.start(&mut endpoints, &mut buf, &mut depot).await?;
         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;
+            session.serve(endpoints, conn, buf, depot, receiver).await;
         });
         let client =
             Client::new(chroot.to_owned(), self.server_version, session_info, session_timeout, sender, state_watcher);

src/session/depot.rs

@@ -16,6 +16,9 @@ use crate::proto::{OpCode, PredefinedXid, RemoveWatchesRequest};
 pub struct Depot {
     xid: Xid,
 
+    // If ongoing authes are interruptted due to disconnection, they will be retried after
+    // session reestablishment as auth failure will terminate session and should not be dropped
+    // silently in case of disconnection.
     pending_authes: Vec<SessionOperation>,
 
     writing_slices: Vec<IoSlice<'static>>,
@@ -30,7 +33,7 @@ pub struct Depot {
 }
 
 impl Depot {
-    pub fn for_serving() -> Depot {
+    pub fn new() -> Self {
         let writing_capacity = 128usize;
         Depot {
             xid: Default::default(),
@@ -39,34 +42,18 @@ impl Depot {
             writing_slices: Vec::with_capacity(writing_capacity),
             writing_operations: VecDeque::with_capacity(writing_capacity),
             written_operations: HashMap::with_capacity(128),
-            pending_operations: Default::default(),
+            pending_operations: VecDeque::with_capacity(32),
             watching_paths: HashMap::with_capacity(32),
             unwatching_paths: HashMap::with_capacity(32),
         }
     }
 
-    pub fn for_connecting() -> Depot {
-        Depot {
-            xid: Default::default(),
-            sasl: false,
-            pending_authes: Default::default(),
-            writing_slices: Vec::with_capacity(10),
-            writing_operations: VecDeque::with_capacity(10),
-            written_operations: HashMap::with_capacity(10),
-            pending_operations: VecDeque::with_capacity(10),
-            watching_paths: HashMap::new(),
-            unwatching_paths: HashMap::new(),
-        }
-    }
-
-    /// Clear all buffered operations from previous run.
-    pub fn clear(&mut self) {
-        self.xid = Default::default();
+    /// Reset state and clear buffered operations from previous run except `pending_authes`.
+    pub fn reset(&mut self) {
+        // We don't reset xid, so we get continuous xid (ignoring the overflow) even in case of
+        // reconnection. This is helpful in diagnosis.
         self.sasl = false;
-        self.pending_authes.clear();
         self.writing_slices.clear();
-        self.watching_paths.clear();
-        self.unwatching_paths.clear();
         self.writing_operations.clear();
         self.written_operations.clear();
         self.pending_operations.clear();

src/session/mod.rs

@@ -225,16 +225,15 @@ impl Session {
         mut endpoints: IterableEndpoints,
         conn: Connection,
         mut buf: Vec<u8>,
-        mut connecting_trans: Depot,
+        mut depot: Depot,
         mut requester: mpsc::UnboundedReceiver<SessionOperation>,
     ) {
-        let mut depot = Depot::for_serving();
         let mut unwatch_requester = self.unwatch_receiver.take().unwrap();
         endpoints.cycle();
         endpoints.reset();
         self.serve_once(conn, &mut endpoints, &mut buf, &mut depot, &mut requester, &mut unwatch_requester).await;
         while !self.session_state.is_terminated() {
-            let conn = match self.start(&mut endpoints, &mut buf, &mut connecting_trans).await {
+            let conn = match self.start(&mut endpoints, &mut buf, &mut depot).await {
                 Err(err) => {
                     warn!("fail to connect to cluster {:?} due to {}", endpoints.endpoints(), err);
                     self.resolve_start_error(&err);
@@ -683,7 +682,7 @@ impl Session {
                 conn
             },
         };
-        depot.clear();
+        depot.reset();
         buf.clear();
         self.send_connect(depot);
         #[cfg(any(feature = "sasl-digest-md5", feature = "sasl-gssapi"))]