Fix Keeper ephemeral count

Author: antonio2368

src/Coordination/KeeperSnapshotManager.cpp

@@ -443,7 +443,10 @@ void KeeperStorageSnapshot<Storage>::deserialize(SnapshotDeserializationResult<S
                 node.getChildren().reserve(node.stats.numChildren());
 
         if (ephemeral_owner != 0)
+        {
             storage.committed_ephemerals[node.stats.ephemeralOwner()].insert(std::string{path});
+            ++storage.committed_ephemeral_nodes;
+        }
 
         if (recalculate_digest)
             storage.nodes_digest += node.getDigest(path);

src/Coordination/KeeperStorage.cpp

@@ -122,12 +122,14 @@ void unregisterEphemeralPath(KeeperStorageBase::Ephemerals & ephemerals, int64_t
     if (ephemerals_it == ephemerals.end())
     {
         if (throw_if_missing)
-            throw Exception(ErrorCodes::LOGICAL_ERROR, "Session {} is missing ephemeral path {}", session_id, path);
+            throw Exception(ErrorCodes::LOGICAL_ERROR, "Session {} is expected to have ephemeral paths but no path is registered", session_id);
 
         return;
     }
 
-    ephemerals_it->second.erase(path);
+    if (auto erased = ephemerals_it->second.erase(path); !erased && throw_if_missing)
+        throw Exception(ErrorCodes::LOGICAL_ERROR, "Session {} is missing ephemeral path {}", session_id, path);
+
     if (ephemerals_it->second.empty())
         ephemerals.erase(ephemerals_it);
 }
@@ -255,14 +257,9 @@ void NodeStats::copyStats(const Coordination::Stat & stat)
     aversion = stat.aversion;
 
     if (stat.ephemeralOwner == 0)
-    {
-        is_ephemeral_and_ctime.is_ephemeral = false;
         setNumChildren(stat.numChildren);
-    }
     else
-    {
         setEphemeralOwner(stat.ephemeralOwner);
-    }
 }
 
 void KeeperRocksNodeInfo::copyStats(const Coordination::Stat & stat)
@@ -1356,6 +1353,7 @@ bool KeeperStorage<Container>::removeNode(const std::string & path, int32_t vers
 
     if (prev_node.stats.ephemeralOwner() != 0)
     {
+        chassert(committed_ephemeral_nodes != 0);
         --committed_ephemeral_nodes;
         std::lock_guard lock(ephemeral_mutex);
         unregisterEphemeralPath(committed_ephemerals, prev_node.stats.ephemeralOwner(), path, /*throw_if_missing=*/true);

src/Coordination/KeeperStorage.h

@@ -60,7 +60,7 @@ struct NodeStats
 
     void setEphemeralOwner(int64_t ephemeral_owner)
     {
-        is_ephemeral_and_ctime.is_ephemeral = ephemeral_owner != 0;
+        is_ephemeral_and_ctime.is_ephemeral = true;
         ephemeral_or_children_data.ephemeral_owner = ephemeral_owner;
     }