fix: clean errors instead of crashes/hangs on unsupported requests; fix SetWatches child-watch restore

src/Service/ConnectionHandler.cpp

@@ -577,19 +577,35 @@ std::pair<Coordination::OpNum, Coordination::XID> ConnectionHandler::receiveRequ
 {
     ReadBufferFromMemory body(req_body_buf->begin(), req_body_buf->used());
     int32_t xid;
+    int32_t raw_opnum;
     Coordination::read(xid, body);
+    /// Read the raw int32 instead of the validated OpNum overload: getOpNum throws
+    /// for unknown opnums, and we want that to happen inside the try below so the
+    /// client gets an error reply rather than a silently dropped request.
+    Coordination::read(raw_opnum, body);
+    Coordination::OpNum opnum = static_cast<Coordination::OpNum>(raw_opnum);
 
-    Coordination::OpNum opnum;
-    Coordination::read(opnum, body);
-
-    LOG_DEBUG(log, "Receive request #{}#{}#{}", toHexString(session_id.load()), xid, Coordination::toString(opnum));
+    try
+    {
+        LOG_DEBUG(log, "Receive request #{}#{}#{}", toHexString(session_id.load()), xid, Coordination::toString(opnum));
 
-    Coordination::ZooKeeperRequestPtr request = Coordination::ZooKeeperRequestFactory::instance().get(opnum);
-    request->xid = xid;
-    request->readImpl(body);
+        Coordination::ZooKeeperRequestPtr request = Coordination::ZooKeeperRequestFactory::instance().get(opnum);
+        request->xid = xid;
+        request->readImpl(body);
 
-    if (!keeper_dispatcher->pushRequest(request, session_id))
-        throw Exception(ErrorCodes::TIMEOUT_EXCEEDED, "Session {} already disconnected", toHexString(session_id.load()));
+        if (!keeper_dispatcher->pushRequest(request, session_id))
+            throw Exception(ErrorCodes::TIMEOUT_EXCEEDED, "Session {} already disconnected", toHexString(session_id.load()));
+    }
+    catch (const Coordination::Exception & e)
+    {
+        /// Unknown opnum, or a request we can't parse. The body is already fully read,
+        /// so the stream stays aligned; answer with an error instead of dropping the
+        /// request silently, which would leave the client hanging until its timeout.
+        auto response = std::make_shared<Coordination::ZooKeeperErrorResponse>();
+        response->xid = xid;
+        response->error = e.code;
+        pushUserResponseToSendingQueue(response);
+    }
     return std::make_pair(opnum, xid);
 }
 

src/Service/KeeperStore.cpp

@@ -10,7 +10,6 @@ namespace RK
 namespace ErrorCodes
 {
     extern const int LOGICAL_ERROR;
-    extern const int BAD_ARGUMENTS;
 }
 
 static inline void set_response(
@@ -1274,8 +1273,16 @@ struct StoreRequestMultiTxn final : public StoreRequest
     using OperationType = Coordination::ZooKeeperMultiRequest::OperationType;
     OperationType operation_type = OperationType::Unspecified;
 
+    /// Set when a subrequest is not supported (unknown opnum, or a mix of read and
+    /// write subrequests). The constructor must not throw: it runs at Raft apply time,
+    /// where RequestProcessor::applyRequest aborts the whole server on any exception.
+    Coordination::Error construction_error = Coordination::Error::ZOK;
+
     bool checkAuth(KeeperStore & store, int64_t session_id) const override
     {
+        /// Let process() answer with construction_error instead of a misleading ZNOAUTH.
+        if (construction_error != Coordination::Error::ZOK)
+            return true;
         for (const auto & concrete_request : concrete_requests)
             if (!concrete_request->checkAuth(store, session_id))
                 return false;
@@ -1291,7 +1298,7 @@ struct StoreRequestMultiTxn final : public StoreRequest
         const auto check_operation_type = [&](OperationType type)
         {
             if (operation_type != OperationType::Unspecified && operation_type != type)
-                throw RK::Exception(ErrorCodes::BAD_ARGUMENTS, "Illegal mixing of read and write operations in multi request");
+                construction_error = Coordination::Error::ZBADARGUMENTS;
             operation_type = type;
         };
 
@@ -1346,8 +1353,7 @@ struct StoreRequestMultiTxn final : public StoreRequest
                 concrete_requests.push_back(std::make_shared<StoreRequestList>(sub_zk_request));
             }
             else
-                throw RK::Exception(
-                    ErrorCodes::BAD_ARGUMENTS, "Illegal command as part of multi ZooKeeper request {}", toString(sub_zk_request->getOpNum()));
+                construction_error = Coordination::Error::ZBADARGUMENTS;
         }
     }
 
@@ -1356,6 +1362,24 @@ struct StoreRequestMultiTxn final : public StoreRequest
     {
         Coordination::ZooKeeperResponsePtr response = zk_request->makeResponse();
         Coordination::ZooKeeperMultiResponse & response_typed = dynamic_cast<Coordination::ZooKeeperMultiResponse &>(*response);
+
+        /// Reject unsupported multis with a clean error instead of throwing at apply
+        /// time, where the exception would abort the server process. Leave the
+        /// top-level error at ZOK (its default): ZooKeeperResponse::writeNoCopy only
+        /// serializes the per-subrequest body when the top-level error is ZOK, so a
+        /// non-ZOK top-level error here would silently suppress the body — the client
+        /// would see an empty multi response instead of the per-op errors. Matches the
+        /// shape the write-rollback path below already produces.
+        if (construction_error != Coordination::Error::ZOK)
+        {
+            for (auto & sub_response : response_typed.responses)
+            {
+                sub_response = std::make_shared<Coordination::ZooKeeperErrorResponse>();
+                sub_response->error = construction_error;
+            }
+            return {response, {}};
+        }
+
         std::vector<Undo> undo_actions;
 
         try
@@ -1642,12 +1666,22 @@ void KeeperStore::processRequest(
         auto * request = dynamic_cast<Coordination::ZooKeeperSetWatchesRequest *>(zk_request.get());
 
         /// path -> mzixd pzxid
+        /// Populated from every watch array: child and exist watches need the same node
+        /// info as data watches, otherwise every re-established child watch is treated
+        /// as "node missing" and spuriously fires DELETED (and is dropped), and exist
+        /// watches never fire CREATED for nodes created during the disconnect.
         std::unordered_map<String, std::pair<int64_t, int64_t>> watch_nodes_info;
-        for (String & path : request->data_watches)
+        const auto add_watch_nodes_info = [&](const std::vector<String> & paths)
         {
-            if (auto node = data_tree.get(path))
-                watch_nodes_info.emplace(path, std::make_pair(node->stat.mzxid, node->stat.pzxid));
-        }
+            for (const String & path : paths)
+            {
+                if (auto node = data_tree.get(path))
+                    watch_nodes_info.emplace(path, std::make_pair(node->stat.mzxid, node->stat.pzxid));
+            }
+        };
+        add_watch_nodes_info(request->data_watches);
+        add_watch_nodes_info(request->exist_watches);
+        add_watch_nodes_info(request->list_watches);
 
         auto watch_responses = watch_manager.processRequestSetWatch(request_for_session, watch_nodes_info);
         set_response(responses_queue, watch_responses, ignore_response);
@@ -1667,17 +1701,21 @@ void KeeperStore::processRequest(
             auto sub_opnum = sub_zk_request->getOpNum();
 
             /// MultiRead only allows read operations. Writes routed here would bypass Raft consensus.
+            /// Reject with a clean per-subrequest error instead of throwing: read-path
+            /// exceptions produce no response and the client would hang until timeout.
             if (sub_opnum != Coordination::OpNum::Get
                 && sub_opnum != Coordination::OpNum::Exists
                 && sub_opnum != Coordination::OpNum::List
                 && sub_opnum != Coordination::OpNum::SimpleList
                 && sub_opnum != Coordination::OpNum::FilteredList
                 && sub_opnum != Coordination::OpNum::GetACL)
             {
-                throw Exception(
-                    ErrorCodes::BAD_ARGUMENTS,
-                    "Illegal command {} as part of MultiRead request",
-                    Coordination::toString(sub_opnum));
+                auto sub_response = std::make_shared<Coordination::ZooKeeperErrorResponse>();
+                sub_response->error = Coordination::Error::ZBADARGUMENTS;
+                sub_response->xid = sub_zk_request->xid;
+                sub_response->zxid = zxid.load();
+                multi_response.responses[i] = sub_response;
+                continue;
             }
 
             auto sub_store_request = StoreRequestFactory::instance().get(sub_zk_request);