Fail fast when there are something wrong with snapshot

src/Common/ErrorCodes.cpp

@@ -105,6 +105,10 @@
     M(114, RAFT_FWD_NO_CONN) \
     M(115, FORWARD_NOT_CONNECTED) \
     M(116, ILLEGAL_SETTING_VALUE) \
+    M(117, CORRUPTED_LOG) \
+    M(118, CORRUPTED_SNAPSHOT) \
+    M(119, SNAPSHOT_OBJECT_NOT_EXISTS) \
+    M(120, SNAPSHOT_NOT_EXISTS) \
     /* See END */
 
 namespace RK

src/Service/ACLMap.cpp

@@ -100,9 +100,9 @@ void ACLMap::addMapping(uint64_t acls_id, const Coordination::ACLs & acls)
 
 void ACLMap::addUsage(uint64_t acl_id, uint64_t count)
 {
-    std::lock_guard lock(acl_mutex);
     if (acl_id == 0)
         return;
+    std::lock_guard lock(acl_mutex);
     usage_counter[acl_id] += count;
 }
 

src/Service/ConnectionHandler.h

@@ -37,7 +37,7 @@ using Poco::Thread;
  * User connection handler with TCP protocol. It is a core class who process
  * Zookeeper network protocol and send it to dispatcher.
  *
- * We utilize a getWorkerReactor network programming model. We allocate a handler for
+ * We utilize reactor network programming model. We allocate a handler for
  * every connection and ensure that every handler run in the same network thread.
  *
  * So there is no multi-thread issues.

src/Service/KeeperStore.h

@@ -73,6 +73,12 @@ struct KeeperNode
     bool operator!=(const KeeperNode & rhs) const { return !(rhs == *this); }
 };
 
+struct KeeperNodeWithPath
+{
+    String path;
+    std::shared_ptr<KeeperNode> node;
+};
+
 /// Map for data tree, it is thread safe with read write lock.
 /// ConcurrentMap is a two-level unordered_map which is designed
 /// to reduce latency for unordered_map scales.

src/Service/NuRaftLogSnapshot.h

@@ -95,7 +95,7 @@ class KeeperSnapshotStore
     SnapshotVersion version;
 
 private:
-    /// For snapshot version v1
+    /// For snapshot version v1 /// TODO delete
     size_t createObjectsV1(KeeperStore & store, int64_t next_zxid = 0, int64_t next_session_id = 0);
     /// For snapshot version v3
     size_t createObjectsV2(KeeperStore & store, int64_t next_zxid = 0, int64_t next_session_id = 0);
@@ -104,24 +104,24 @@ class KeeperSnapshotStore
     void getObjectPath(ulong object_id, String & path);
 
     /// Parse an snapshot object. We should take the version from snapshot in general.
-    bool parseObject(KeeperStore & store, String obj_path);
+    void parseObject(KeeperStore & store, String obj_path);
 
     /// Parse batch header in an object
     /// TODO use internal buffer
-    bool parseBatchHeader(ptr<std::fstream> fs, SnapshotBatchHeader & head);
+    void parseBatchHeader(ptr<std::fstream> fs, SnapshotBatchHeader & head);
 
+    /// Parse a batch /// TODO delete
+    void parseBatchBody(KeeperStore & store, char * batch_buf, size_t length, SnapshotVersion version_);
     /// Parse a batch
-    bool parseBatchBody(KeeperStore & store, char * batch_buf, size_t length, SnapshotVersion version_);
-    /// Parse a batch
-    bool parseBatchBodyV2(KeeperStore & store, char * buf, size_t length, SnapshotVersion version_);
+    void parseBatchBodyV2(KeeperStore & store, char * buf, size_t length, SnapshotVersion version_);
 
-    /// Serialize whole data tree
+    /// Serialize whole data tree /// TODO delete
     size_t serializeDataTree(KeeperStore & storage);
 
-    /// For snapshot version v3
+    /// For snapshot version v2
     size_t serializeDataTreeV2(KeeperStore & storage);
 
-    /// Serialize data tree by deep traversal.
+    /// Serialize data tree by deep traversal. /// TODO delete
     void serializeNode(
         ptr<WriteBufferFromFile> & out,
         ptr<SnapshotBatchPB> & batch,
@@ -139,10 +139,10 @@ class KeeperSnapshotStore
         uint64_t & processed,
         uint32_t & checksum);
 
-    /// Append node to batch
+    /// Append node to batch /// TODO delete
     inline static void
     appendNodeToBatch(ptr<SnapshotBatchPB> batch, const String & path, std::shared_ptr<KeeperNode> node, SnapshotVersion version);
-    /// For snapshot version v3
+    /// For snapshot version v2
     inline static void
     appendNodeToBatchV2(ptr<SnapshotBatchBody> batch, const String & path, std::shared_ptr<KeeperNode> node, SnapshotVersion version);