diff --git a/src/Service/KeeperCommon.h b/src/Service/KeeperCommon.h index f9a41d667c..8a42c47664 100644 --- a/src/Service/KeeperCommon.h +++ b/src/Service/KeeperCommon.h @@ -34,6 +34,11 @@ using ThreadPoolPtr = std::shared_ptr; MULTI_READ, CHECK_NOT_EXISTS, CREATE_IF_NOT_EXISTS, + REMOVE_RECURSIVE, + CHECK_STAT, + TRY_REMOVE, + LIST_WITH_STAT_AND_DATA, + GET_CHILDREN_RECURSIVE, MAX, }; @@ -60,9 +65,10 @@ using ThreadPoolPtr = std::shared_ptr; }; inline constexpr auto CURRENT_KEEPER_API_VERSION = KeeperApiVersion::WITH_MULTI_READ; - // 0b11110000 enable FILTERED_LIST & MULTI_READ & CHECK_NOT_EXISTS & CREATE_IF_NOT_EXISTS, - // For example, set 0b10110000 to disable MULTI_READ - inline const std::string CURRENT_KEEPER_FEATURE_FLAGS = "\xF0"; + // 0b11111111 0b10000000 enable all 9 feature flags: + // FILTERED_LIST, MULTI_READ, CHECK_NOT_EXISTS, CREATE_IF_NOT_EXISTS, + // REMOVE_RECURSIVE, CHECK_STAT, TRY_REMOVE, LIST_WITH_STAT_AND_DATA, GET_CHILDREN_RECURSIVE + inline const std::string CURRENT_KEEPER_FEATURE_FLAGS = "\xFF\x80"; #endif struct RequestId; diff --git a/src/Service/KeeperStore.cpp b/src/Service/KeeperStore.cpp index 94d2b79938..ecd14d433a 100644 --- a/src/Service/KeeperStore.cpp +++ b/src/Service/KeeperStore.cpp @@ -120,7 +120,8 @@ static bool shouldIncreaseZxid(const Coordination::ZooKeeperRequestPtr & zk_requ || dynamic_cast(zk_request.get()) || dynamic_cast(zk_request.get()) || dynamic_cast(zk_request.get()) - || zk_request->getOpNum() == Coordination::OpNum::MultiRead); + || zk_request->getOpNum() == Coordination::OpNum::MultiRead + || zk_request->getOpNum() == Coordination::OpNum::ListRecursive); } KeeperNodePtr KeeperNode::clone() const @@ -455,6 +456,11 @@ struct StoreRequestRemove final : public StoreRequest { using StoreRequest::StoreRequest; + /// Filled by process(): true only when a node was actually removed. + /// TryRemove returns ZOK even for a missing node, so watch firing must + /// check this (mirrors RemoveRecursive::removed_paths). + mutable bool removed = false; + bool checkAuth(KeeperStore & store, int64_t session_id) const override { auto parent = store.getNode(getParentPath(zk_request->getPath())); @@ -493,7 +499,10 @@ struct StoreRequestRemove final : public StoreRequest auto node = store.getNode(request.path); if (node == nullptr) { - response.error = Coordination::Error::ZNONODE; + if (request.try_remove) + response.error = Coordination::Error::ZOK; + else + response.error = Coordination::Error::ZNONODE; } else if (request.version != -1 && request.version != node->stat.version) { @@ -544,12 +553,235 @@ struct StoreRequestRemove final : public StoreRequest undo_parent->children.insert(child_basename); } }; + removed = true; } return {response_ptr, undo}; } }; +struct StoreRequestRemoveRecursive final : public StoreRequest +{ + using StoreRequest::StoreRequest; + + /// Paths actually removed (root + descendants), filled on success so the caller + /// can fire a DELETED watch for each. Recursive delete touches many nodes, but + /// watch firing is centralized in processRequest which only sees the root path. + mutable std::vector removed_paths; + + bool checkAuth(KeeperStore & store, int64_t session_id) const override + { + auto parent = store.getNode(getParentPath(zk_request->getPath())); + if (parent == nullptr) + return true; + + const auto & node_acls = store.acl_map.convertNumber(parent->acl_id); + if (node_acls.empty()) + return true; + + std::shared_lock r_lock(store.auth_mutex); + auto it = store.session_and_auth.find(session_id); + const auto & session_auths = (it != store.session_and_auth.end()) ? it->second : std::vector{}; + return checkACL(Coordination::ACL::Delete, node_acls, session_auths); + } + + std::pair + process(KeeperStore & store, int64_t zxid, int64_t /*session_id*/, int64_t /* time */) const override + { + auto response = zk_request->makeResponse(); + auto & request_typed = dynamic_cast(*zk_request); + auto & response_typed = dynamic_cast(*response); + + auto root_node = store.getNode(request_typed.path); + if (root_node == nullptr) + { + response_typed.error = Coordination::Error::ZNONODE; + return {response, {}}; + } + + /// Collect all descendant paths via DFS + std::vector paths_to_remove; + paths_to_remove.push_back(request_typed.path); + std::function collect_descendants + = [&](const KeeperNodePtr & node, const String & node_path) + { + for (const auto & child : node->children) + { + String child_path = node_path + "/" + child; + auto child_node = store.getNode(child_path); + if (child_node) + { + paths_to_remove.push_back(child_path); + if (!child_node->children.empty()) + collect_descendants(child_node, child_path); + } + } + }; + collect_descendants(root_node, request_typed.path); + + /// Check limit + if (request_typed.remove_nodes_limit > 0 + && paths_to_remove.size() > static_cast(request_typed.remove_nodes_limit)) + { + response_typed.error = Coordination::Error::ZNOTEMPTY; + return {response, {}}; + } + + /// Snapshot every node about to be removed so the operation can be rolled back + /// (required when RemoveRecursive is a subrequest of a multi transaction). + /// clone() preserves each node's own children set, so internal parent-child + /// links inside the subtree are restored automatically on re-add; only the + /// root's link into its surviving parent must be restored manually. + std::vector> removed_nodes; + removed_nodes.reserve(paths_to_remove.size()); + for (const auto & path : paths_to_remove) + { + if (auto n = store.getNode(path)) + removed_nodes.emplace_back(path, n->clone()); + } + + String root_base = getBaseName(request_typed.path); + String root_parent_path = getParentPath(request_typed.path); + int64_t root_parent_pzxid = 0; + bool has_root_parent = false; + + /// Update the surviving parent of the recursive-delete root: its child count + /// drops by one and pzxid advances, matching StoreRequestRemove's behaviour. + /// (Descendants' parents are themselves removed, so only this parent survives.) + if (auto root_parent = store.getNode(root_parent_path)) + { + has_root_parent = true; + root_parent_pzxid = root_parent->stat.pzxid; + --root_parent->stat.numChildren; + root_parent->stat.pzxid = zxid; + } + + /// Remove from leaves up. Each node is removed individually via KeeperStore public API. + for (auto it = paths_to_remove.rbegin(); it != paths_to_remove.rend(); ++it) + { + const auto & path = *it; + auto node = store.getNode(path); + if (!node) + continue; + + /// Clear from parent's children set + auto parent = store.getNode(getParentPath(path)); + if (parent) + parent->children.erase(getBaseName(path)); + + /// Clean ephemeral and ACL references + if (node->is_ephemeral) + store.removeEphemeralNode(node->stat.ephemeralOwner, path); + store.acl_map.removeUsage(node->acl_id); + store.removeNode(path); + } + + response_typed.error = Coordination::Error::ZOK; + removed_paths = paths_to_remove; + + Undo undo = [&store, removed_nodes, root_base, root_parent_path, root_parent_pzxid, has_root_parent] + { + /// Re-add nodes root-first so parents exist before children are relinked. + /// Each clone already carries its own children set, so subtree links restore + /// automatically; we only relink the root into its surviving parent below. + for (const auto & [path, node] : removed_nodes) + { + store.addNode(path, node); + store.acl_map.addUsage(node->acl_id); + if (node->is_ephemeral) + store.addEphemeralNode(node->stat.ephemeralOwner, path); + } + + if (has_root_parent) + { + if (auto root_parent = store.getNode(root_parent_path)) + { + root_parent->children.insert(root_base); + ++root_parent->stat.numChildren; + root_parent->stat.pzxid = root_parent_pzxid; + } + } + }; + + return {response, undo}; + } +}; + +struct StoreRequestListRecursive final : public StoreRequest +{ + using StoreRequest::StoreRequest; + + bool checkAuth(KeeperStore & store, int64_t session_id) const override + { + return checkACLForNode(store, session_id, zk_request->getPath(), Coordination::ACL::Read); + } + + std::pair + process(KeeperStore & store, int64_t /*zxid*/, int64_t /*session_id*/, int64_t /* time */) const override + { + auto response = zk_request->makeResponse(); + auto & response_typed = dynamic_cast(*response); + + /// ponytail: access path directly from the request, same as other StoreRequest implementations + auto & request_typed = dynamic_cast(*zk_request); + + auto node = store.getNode(request_typed.path); + if (!node) + { + response_typed.error = Coordination::Error::ZNONODE; + return {response, {}}; + } + + std::vector all_paths; + bool stopped = false; + + std::function collect_recursive + = [&](const KeeperNodePtr & current, const String & current_path) + { + for (const auto & child : current->children) + { + if (stopped) + return; + + String child_path = current_path + "/" + child; + all_paths.push_back(child_path); + + if (request_typed.max_entries > 0 + && all_paths.size() >= static_cast(request_typed.max_entries)) + { + stopped = true; + return; + } + + auto child_node = store.getNode(child_path); + if (child_node && !child_node->children.empty()) + collect_recursive(child_node, child_path); + } + }; + + collect_recursive(node, request_typed.path); + + response_typed.names.push_back(all_paths.begin(), all_paths.end()); + response_typed.error = Coordination::Error::ZOK; + return {response, {}}; + } + +private: + static bool checkACLForNode(KeeperStore & store, int64_t session_id, const String & path, int32_t permission) + { + auto node = store.getNode(path); + if (!node) + return true; + const auto & node_acls = store.acl_map.convertNumber(node->acl_id); + if (node_acls.empty()) + return true; + std::shared_lock r_lock(store.auth_mutex); + auto it = store.session_and_auth.find(session_id); + const auto & session_auths = (it != store.session_and_auth.end()) ? it->second : std::vector{}; + return checkACL(permission, node_acls, session_auths); + } +}; + struct StoreRequestExists final : public StoreRequest { using StoreRequest::StoreRequest; @@ -690,30 +922,33 @@ struct StoreRequestList final : public StoreRequest if (path_prefix.empty()) throw RK::Exception(ErrorCodes::LOGICAL_ERROR, "Logical error: path cannot be empty"); - if (response->getOpNum() == Coordination::OpNum::List || response->getOpNum() == Coordination::OpNum::FilteredList) + if (response->getOpNum() == Coordination::OpNum::List + || response->getOpNum() == Coordination::OpNum::FilteredList + || response->getOpNum() == Coordination::OpNum::FilteredListWithStatsAndData) { using enum Coordination::ZooKeeperFilteredListRequest::ListRequestType; auto list_request_type = ALL; + bool with_stat = false; + bool with_data = false; if (auto * filtered_list_request = dynamic_cast(&request_typed)) { list_request_type = filtered_list_request->list_request_type; + with_stat = filtered_list_request->with_stat; + with_data = filtered_list_request->with_data; } - auto & response_typed = dynamic_cast(*response); + /// Cast to the ListResponse base: for OpNum 506 the concrete type is + /// ZooKeeperFilteredListWithStatsAndDataResponse, which is NOT a ZooKeeperListResponse. + auto & response_typed = dynamic_cast(*response); response_typed.stat = node->statForResponse(); - if (list_request_type == ALL) - { - response_typed.names.reserve(node->children.size()); - response_typed.names.push_back(node->children.begin(), node->children.end()); - return {response, {}}; - } - - auto add_child = [&](const auto & child) + auto matches_filter = [&](const auto & child) -> bool { + if (list_request_type == ALL) + return true; auto child_node = store.getNode(request_typed.path + "/" + child); - if (node == nullptr) + if (child_node == nullptr) { LOG_ERROR( &Poco::Logger::get("StoreRequestList"), @@ -722,15 +957,26 @@ struct StoreRequestList final : public StoreRequest request_typed.path); std::terminate(); } - const auto is_ephemeral = child_node->stat.ephemeralOwner != 0; return (is_ephemeral && list_request_type == EPHEMERAL_ONLY) || (!is_ephemeral && list_request_type == PERSISTENT_ONLY); }; - for (const auto & child: node->children) + response_typed.names.reserve(node->children.size()); + for (const auto & child : node->children) { - if (add_child(child)) - response_typed.names.push_back(child); + if (!matches_filter(child)) + continue; + + response_typed.names.push_back(child); + + if (with_stat || with_data) + { + auto child_node = store.getNode(request_typed.path + "/" + child); + if (with_stat) + response_typed.stats.push_back(child_node ? child_node->statForResponse() : Coordination::Stat{}); + if (with_data) + response_typed.data.push_back(child_node ? child_node->data : String{}); + } } } else @@ -815,6 +1061,48 @@ struct StoreRequestCheck final : public StoreRequest bool check_not_exists; }; +struct StoreRequestCheckStat final : public StoreRequest +{ + using StoreRequest::StoreRequest; + + bool checkAuth(KeeperStore & store, int64_t session_id) const override + { + auto node = store.getNode(zk_request->getPath()); + if (node == nullptr) + return true; + + const auto & node_acls = store.acl_map.convertNumber(node->acl_id); + if (node_acls.empty()) + return true; + + std::shared_lock r_lock(store.auth_mutex); + auto it = store.session_and_auth.find(session_id); + const auto & session_auths = (it != store.session_and_auth.end()) ? it->second : std::vector{}; + return checkACL(Coordination::ACL::Read, node_acls, session_auths); + } + + std::pair + process(KeeperStore & store, int64_t /*zxid*/, int64_t /*session_id*/, int64_t /* time */) const override + { + auto response = zk_request->makeResponse(); + auto & request_typed = dynamic_cast(*zk_request); + + auto node = store.getNode(request_typed.path); + if (node == nullptr) + response->error = Coordination::Error::ZNONODE; + else if (request_typed.version != -1 && request_typed.version != node->stat.version) + response->error = Coordination::Error::ZBADVERSION; + else if (request_typed.cversion != -1 && request_typed.cversion != node->stat.cversion) + response->error = Coordination::Error::ZBADVERSION; + else if (request_typed.aversion != -1 && request_typed.aversion != node->stat.aversion) + response->error = Coordination::Error::ZBADVERSION; + else + response->error = Coordination::Error::ZOK; + + return {response, {}}; + } +}; + struct StoreRequestSetACL final : public StoreRequest { using StoreRequest::StoreRequest; @@ -1015,7 +1303,8 @@ struct StoreRequestMultiTxn final : public StoreRequest check_operation_type(OperationType::Write); concrete_requests.push_back(std::make_shared(sub_zk_request)); } - else if (sub_zk_request->getOpNum() == Coordination::OpNum::Remove) + else if (sub_zk_request->getOpNum() == Coordination::OpNum::Remove + || sub_zk_request->getOpNum() == Coordination::OpNum::TryRemove) { check_operation_type(OperationType::Write); concrete_requests.push_back(std::make_shared(sub_zk_request)); @@ -1030,6 +1319,16 @@ struct StoreRequestMultiTxn final : public StoreRequest check_operation_type(OperationType::Write); concrete_requests.push_back(std::make_shared(sub_zk_request)); } + else if (sub_zk_request->getOpNum() == Coordination::OpNum::CheckStat) + { + check_operation_type(OperationType::Write); + concrete_requests.push_back(std::make_shared(sub_zk_request)); + } + else if (sub_zk_request->getOpNum() == Coordination::OpNum::RemoveRecursive) + { + check_operation_type(OperationType::Write); + concrete_requests.push_back(std::make_shared(sub_zk_request)); + } else if (sub_zk_request->getOpNum() == Coordination::OpNum::Get) { check_operation_type(OperationType::Read); @@ -1221,6 +1520,11 @@ StoreRequestFactory::StoreRequestFactory() registerNuKeeperRequestWrapper(*this); registerNuKeeperRequestWrapper(*this); registerNuKeeperRequestWrapper(*this); + registerNuKeeperRequestWrapper(*this); + registerNuKeeperRequestWrapper(*this); + registerNuKeeperRequestWrapper(*this); + registerNuKeeperRequestWrapper(*this); + registerNuKeeperRequestWrapper(*this); } @@ -1461,9 +1765,16 @@ void KeeperStore::processRequest( if (!multi_response->responses.empty() && multi_response->responses.back()->error == Coordination::Error::ZOK) { auto * multi_request = dynamic_cast(zk_request.get()); - for (auto & concrete_request : multi_request->requests) + const auto * multi_txn = dynamic_cast(store_request.get()); + for (size_t i = 0; i < multi_request->requests.size(); ++i) { - const auto * sub_zk_request = dynamic_cast(concrete_request.get()); + const auto * sub_zk_request = dynamic_cast(multi_request->requests[i].get()); + /// TryRemove sub-op returns ZOK without deleting a missing + /// node — must not fire a spurious DELETED watch for it. + const auto * sub_remove + = multi_txn ? dynamic_cast(multi_txn->concrete_requests[i].get()) : nullptr; + if (sub_remove && !sub_remove->removed) + continue; auto watch_responses = watch_manager.processWatches(sub_zk_request->getPath(), sub_zk_request->getOpNum()); if (!watch_responses.empty()) { @@ -1475,11 +1786,34 @@ void KeeperStore::processRequest( } else { - auto watch_responses = watch_manager.processWatches(zk_request->getPath(), zk_request->getOpNum()); - if (!watch_responses.empty()) + /// RemoveRecursive deletes a whole subtree; fire a DELETED watch for + /// every removed node, not just the root (processWatches by opnum only + /// covers the single request path). + if (auto * rr = dynamic_cast(store_request.get())) + { + for (const auto & removed_path : rr->removed_paths) + { + auto watch_responses = watch_manager.processWatches(removed_path, Coordination::Event::DELETED); + if (!watch_responses.empty()) + set_response(responses_queue, watch_responses, ignore_response); + } + } + else { - LOG_TRACE(log, "{} triggered {} watches", request_for_session.toSimpleString(), watch_responses.size()); - set_response(responses_queue, watch_responses, ignore_response); + /// TryRemove succeeds (ZOK) without deleting anything when the + /// node is missing — must not fire a spurious DELETED watch. + const auto * remove_request = dynamic_cast(store_request.get()); + const bool actually_removed = !remove_request || remove_request->removed; + + if (actually_removed) + { + auto watch_responses = watch_manager.processWatches(zk_request->getPath(), zk_request->getOpNum()); + if (!watch_responses.empty()) + { + LOG_TRACE(log, "{} triggered {} watches", request_for_session.toSimpleString(), watch_responses.size()); + set_response(responses_queue, watch_responses, ignore_response); + } + } } } } diff --git a/src/Service/WatchManager.cpp b/src/Service/WatchManager.cpp index 7bb42c6633..d02b5ae160 100644 --- a/src/Service/WatchManager.cpp +++ b/src/Service/WatchManager.cpp @@ -13,7 +13,8 @@ void WatchManager::registerWatches(const String & path, int64_t session_id, Coor std::lock_guard lock(watch_mutex); auto watches_type = opnum == Coordination::OpNum::List || opnum == Coordination::OpNum::SimpleList - || opnum == Coordination::OpNum::FilteredList ? WatchType::List : WatchType::Data; + || opnum == Coordination::OpNum::FilteredList + || opnum == Coordination::OpNum::FilteredListWithStatsAndData ? WatchType::List : WatchType::Data; switch (watches_type) { @@ -37,6 +38,7 @@ ResponsesForSessions WatchManager::processWatches(const String & path, Coordinat case Coordination::OpNum::Create: return processWatches(path, Coordination::Event::CREATED); case Coordination::OpNum::Remove: + case Coordination::OpNum::TryRemove: return processWatches(path, Coordination::Event::DELETED); case Coordination::OpNum::Set: return processWatches(path, Coordination::Event::CHANGED); diff --git a/src/Service/tests/gtest_raft_state_machine.cpp b/src/Service/tests/gtest_raft_state_machine.cpp index 60d0873eaf..25f9bef17d 100644 --- a/src/Service/tests/gtest_raft_state_machine.cpp +++ b/src/Service/tests/gtest_raft_state_machine.cpp @@ -662,3 +662,432 @@ TEST(RaftStateMachine, MultiReadAuthCheckPerSubrequest) cleanDirectory(snap_dir); cleanDirectory(log_dir); } + +TEST(RaftStateMachine, RemoveRecursive) +{ + String snap_dir(SNAP_DIR + "/rem_rec"); + String log_dir(LOG_DIR + "/rem_rec"); + cleanDirectory(snap_dir); + cleanDirectory(log_dir); + + KeeperResponsesQueue queue; + RaftSettingsPtr setting_ptr = RaftSettings::getDefault(); + std::mutex new_session_id_callback_mutex; + std::unordered_map> new_session_id_callback; + + NuRaftStateMachine machine(queue, setting_ptr, snap_dir, log_dir, 10, 3, new_session_id_callback_mutex, new_session_id_callback); + int64_t session_id = machine.getStore().getSessionID(30000); + + /// Build: /a -> /a/b, /a/b/c, /a/d + setNode(machine.getStore(), "a", "root", false, session_id); + setNode(machine.getStore(), "a/b", "child_b", false, session_id); + setNode(machine.getStore(), "a/b/c", "grandchild", false, session_id); + setNode(machine.getStore(), "a/d", "child_d", false, session_id); + ASSERT_TRUE(machine.getStore().getNode("/a") != nullptr); + ASSERT_TRUE(machine.getStore().getNode("/a/b/c") != nullptr); + + /// Record parent (/) child count before removal + auto root_before = machine.getStore().getNode("/"); + int32_t root_children_before = root_before->stat.numChildren; + + auto req = cs_new(); + req->path = "/a"; + req->xid = 1; + + KeeperStore::KeeperResponsesQueue response_queue; + int64_t time = std::chrono::system_clock::now().time_since_epoch() / std::chrono::milliseconds(1); + machine.getStore().processRequest( + response_queue, {req, session_id, time}, {}, true, false); + + ASSERT_EQ(machine.getStore().getNode("/a"), nullptr); + ASSERT_EQ(machine.getStore().getNode("/a/b"), nullptr); + ASSERT_EQ(machine.getStore().getNode("/a/b/c"), nullptr); + ASSERT_EQ(machine.getStore().getNode("/a/d"), nullptr); + + /// Parent stat must reflect the removed child (regression: issue #1) + auto root_after = machine.getStore().getNode("/"); + ASSERT_EQ(root_after->stat.numChildren, root_children_before - 1); + + ResponseForSession r; + ASSERT_TRUE(response_queue.tryPop(r)); + ASSERT_EQ(r.response->error, Error::ZOK); + + machine.shutdown(); + cleanDirectory(snap_dir); + cleanDirectory(log_dir); +} + +TEST(RaftStateMachine, TryRemove) +{ + String snap_dir(SNAP_DIR + "/tryrem"); + String log_dir(LOG_DIR + "/tryrem"); + cleanDirectory(snap_dir); + cleanDirectory(log_dir); + + KeeperResponsesQueue queue; + RaftSettingsPtr setting_ptr = RaftSettings::getDefault(); + std::mutex new_session_id_callback_mutex; + std::unordered_map> new_session_id_callback; + + NuRaftStateMachine machine(queue, setting_ptr, snap_dir, log_dir, 10, 3, new_session_id_callback_mutex, new_session_id_callback); + int64_t session_id = machine.getStore().getSessionID(30000); + setNode(machine.getStore(), "exists_node", "data", false, session_id); + + /// TryRemove existing node + { + auto req = cs_new(); + req->path = "/exists_node"; + req->try_remove = true; + req->xid = 1; + + KeeperStore::KeeperResponsesQueue response_queue; + int64_t time = std::chrono::system_clock::now().time_since_epoch() / std::chrono::milliseconds(1); + machine.getStore().processRequest(response_queue, {req, session_id, time}, {}, true, false); + + ResponseForSession r; + ASSERT_TRUE(response_queue.tryPop(r)); + ASSERT_EQ(r.response->error, Error::ZOK); + ASSERT_EQ(machine.getStore().getNode("/exists_node"), nullptr); + } + + /// TryRemove nonexistent — succeeds + { + auto req = cs_new(); + req->path = "/nonexistent"; + req->try_remove = true; + req->xid = 2; + + /// Register a data watch on the missing node, then verify TryRemove + /// on a nonexistent path does NOT fire it. + uint64_t watches_before = machine.getStore().getTotalWatchesCount(); + { + /// Exists (unlike Get) registers a data watch even on a missing path + auto exists_req = cs_new(); + exists_req->path = "/nonexistent"; + exists_req->has_watch = true; + exists_req->xid = 3; + + KeeperStore::KeeperResponsesQueue watch_queue; + int64_t time = std::chrono::system_clock::now().time_since_epoch() / std::chrono::milliseconds(1); + machine.getStore().processRequest(watch_queue, {exists_req, session_id, time}, {}, true, false); + ResponseForSession reg; + ASSERT_TRUE(watch_queue.tryPop(reg)); + } + ASSERT_EQ(machine.getStore().getTotalWatchesCount(), watches_before + 1); + + KeeperStore::KeeperResponsesQueue response_queue; + int64_t time = std::chrono::system_clock::now().time_since_epoch() / std::chrono::milliseconds(1); + machine.getStore().processRequest(response_queue, {req, session_id, time}, {}, true, false); + + ResponseForSession r; + ASSERT_TRUE(response_queue.tryPop(r)); + ASSERT_EQ(r.response->error, Error::ZOK); + + /// No watch event fired and the watch survives: the node was never deleted + ASSERT_FALSE(response_queue.tryPop(r)); + ASSERT_EQ(machine.getStore().getTotalWatchesCount(), watches_before + 1); + } + + machine.shutdown(); + cleanDirectory(snap_dir); + cleanDirectory(log_dir); +} + +TEST(RaftStateMachine, CheckStat) +{ + String snap_dir(SNAP_DIR + "/chkstat"); + String log_dir(LOG_DIR + "/chkstat"); + cleanDirectory(snap_dir); + cleanDirectory(log_dir); + + KeeperResponsesQueue queue; + RaftSettingsPtr setting_ptr = RaftSettings::getDefault(); + std::mutex new_session_id_callback_mutex; + std::unordered_map> new_session_id_callback; + + NuRaftStateMachine machine(queue, setting_ptr, snap_dir, log_dir, 10, 3, new_session_id_callback_mutex, new_session_id_callback); + int64_t session_id = machine.getStore().getSessionID(30000); + setNode(machine.getStore(), "check_node", "data", false, session_id); + + auto node = machine.getStore().getNode("/check_node"); + int32_t v = node->stat.version; + int32_t cv = node->stat.cversion; + int32_t av = node->stat.aversion; + + /// Matching stat + { + auto req = cs_new(); + req->path = "/check_node"; + req->version = v; + req->cversion = cv; + req->aversion = av; + req->xid = 1; + + KeeperStore::KeeperResponsesQueue response_queue; + int64_t time = std::chrono::system_clock::now().time_since_epoch() / std::chrono::milliseconds(1); + machine.getStore().processRequest(response_queue, {req, session_id, time}, {}, true, false); + + ResponseForSession r; + ASSERT_TRUE(response_queue.tryPop(r)); + ASSERT_EQ(r.response->error, Error::ZOK); + } + + /// Wrong version + { + auto req = cs_new(); + req->path = "/check_node"; + req->version = v + 1; + req->cversion = -1; + req->aversion = -1; + req->xid = 2; + + KeeperStore::KeeperResponsesQueue response_queue; + int64_t time = std::chrono::system_clock::now().time_since_epoch() / std::chrono::milliseconds(1); + machine.getStore().processRequest(response_queue, {req, session_id, time}, {}, true, false); + + ResponseForSession r; + ASSERT_TRUE(response_queue.tryPop(r)); + ASSERT_EQ(r.response->error, Error::ZBADVERSION); + } + + machine.shutdown(); + cleanDirectory(snap_dir); + cleanDirectory(log_dir); +} + +TEST(RaftStateMachine, ListRecursive) +{ + String snap_dir(SNAP_DIR + "/listrec"); + String log_dir(LOG_DIR + "/listrec"); + cleanDirectory(snap_dir); + cleanDirectory(log_dir); + + KeeperResponsesQueue queue; + RaftSettingsPtr setting_ptr = RaftSettings::getDefault(); + std::mutex new_session_id_callback_mutex; + std::unordered_map> new_session_id_callback; + + NuRaftStateMachine machine(queue, setting_ptr, snap_dir, log_dir, 10, 3, new_session_id_callback_mutex, new_session_id_callback); + int64_t session_id = machine.getStore().getSessionID(30000); + + /// Build: /subtree -> /subtree/x, /subtree/y, /subtree/y/z + setNode(machine.getStore(), "subtree", "root", false, session_id); + setNode(machine.getStore(), "subtree/x", "x_data", false, session_id); + setNode(machine.getStore(), "subtree/y", "y_data", false, session_id); + setNode(machine.getStore(), "subtree/y/z", "z_data", false, session_id); + + auto req = cs_new(); + req->path = "/subtree"; + req->xid = 1; + + /// ListRecursive is read-only: zxid must not advance (regression: issue #2) + int64_t zxid_before = machine.getStore().getZxid(); + + KeeperStore::KeeperResponsesQueue response_queue; + int64_t time = std::chrono::system_clock::now().time_since_epoch() / std::chrono::milliseconds(1); + machine.getStore().processRequest(response_queue, {req, session_id, time}, {}, true, false); + + ASSERT_EQ(machine.getStore().getZxid(), zxid_before); + + ResponseForSession r; + ASSERT_TRUE(response_queue.tryPop(r)); + ASSERT_EQ(r.response->error, Error::ZOK); + + auto & list_resp = dynamic_cast(*r.response); + std::vector names; + for (auto it = list_resp.names.begin(); it != list_resp.names.end(); ++it) + names.emplace_back(*it); + std::sort(names.begin(), names.end()); + ASSERT_EQ(names.size(), 3u); + ASSERT_EQ(names[0], "/subtree/x"); + ASSERT_EQ(names[1], "/subtree/y"); + ASSERT_EQ(names[2], "/subtree/y/z"); + + machine.shutdown(); + cleanDirectory(snap_dir); + cleanDirectory(log_dir); +} + +TEST(RaftStateMachine, FilteredListWithStatsAndData) +{ + String snap_dir(SNAP_DIR + "/flist_stats"); + String log_dir(LOG_DIR + "/flist_stats"); + cleanDirectory(snap_dir); + cleanDirectory(log_dir); + + KeeperResponsesQueue queue; + RaftSettingsPtr setting_ptr = RaftSettings::getDefault(); + std::mutex new_session_id_callback_mutex; + std::unordered_map> new_session_id_callback; + + NuRaftStateMachine machine(queue, setting_ptr, snap_dir, log_dir, 10, 3, new_session_id_callback_mutex, new_session_id_callback); + int64_t session_id = machine.getStore().getSessionID(30000); + + setNode(machine.getStore(), "flist", "root", false, session_id); + setNode(machine.getStore(), "flist/x", "data_x", false, session_id); + setNode(machine.getStore(), "flist/y", "data_y", false, session_id); + + auto req = cs_new(); + req->path = "/flist"; + req->xid = 1; + req->list_with_stats_and_data = true; + req->with_stat = true; + req->with_data = true; + ASSERT_EQ(req->getOpNum(), OpNum::FilteredListWithStatsAndData); + + int64_t zxid_before = machine.getStore().getZxid(); + + KeeperStore::KeeperResponsesQueue response_queue; + int64_t time = std::chrono::system_clock::now().time_since_epoch() / std::chrono::milliseconds(1); + machine.getStore().processRequest(response_queue, {req, session_id, time}, {}, true, false); + + ASSERT_EQ(machine.getStore().getZxid(), zxid_before); + + ResponseForSession r; + ASSERT_TRUE(response_queue.tryPop(r)); + ASSERT_EQ(r.response->error, Error::ZOK); + + auto & resp = dynamic_cast(*r.response); + ASSERT_EQ(resp.names.size(), 2u); + ASSERT_EQ(resp.stats.size(), 2u); + ASSERT_EQ(resp.data.size(), 2u); + + size_t i = 0; + for (auto it = resp.names.begin(); it != resp.names.end(); ++it, ++i) + { + String name = (*it).toString(); + if (name == "x") + ASSERT_EQ(resp.data[i], "data_x"); + else if (name == "y") + ASSERT_EQ(resp.data[i], "data_y"); + else + FAIL() << "unexpected child " << name; + } + + machine.shutdown(); + cleanDirectory(snap_dir); + cleanDirectory(log_dir); +} + +TEST(RaftStateMachine, MultiWriteWithCheckStatAndTryRemove) +{ + String snap_dir(SNAP_DIR + "/multi_checkstat_tryremove"); + String log_dir(LOG_DIR + "/multi_checkstat_tryremove"); + cleanDirectory(snap_dir); + cleanDirectory(log_dir); + + KeeperResponsesQueue queue; + RaftSettingsPtr setting_ptr = RaftSettings::getDefault(); + std::mutex new_session_id_callback_mutex; + std::unordered_map> new_session_id_callback; + + NuRaftStateMachine machine(queue, setting_ptr, snap_dir, log_dir, 10, 3, new_session_id_callback_mutex, new_session_id_callback); + int64_t session_id = machine.getStore().getSessionID(30000); + + setNode(machine.getStore(), "mnode", "data", false, session_id); + auto node = machine.getStore().getNode("/mnode"); + int32_t ver = node->stat.version; + + /// Multi (write): CheckStat(correct version) + TryRemove(existing) must both succeed. + auto multi = cs_new(); + multi->operation_type = ZooKeeperMultiRequest::OperationType::Write; + multi->xid = 10; + { + auto cs = cs_new(); + cs->path = "/mnode"; + cs->version = ver; + cs->cversion = -1; + cs->aversion = -1; + multi->requests.push_back(cs); + } + { + auto tr = cs_new(); + tr->path = "/mnode"; + tr->try_remove = true; + tr->version = -1; + multi->requests.push_back(tr); + } + + KeeperStore::KeeperResponsesQueue response_queue; + int64_t time = std::chrono::system_clock::now().time_since_epoch() / std::chrono::milliseconds(1); + machine.getStore().processRequest(response_queue, {multi, session_id, time}, {}, true, false); + + ResponseForSession r; + ASSERT_TRUE(response_queue.tryPop(r)); + auto & multi_resp = dynamic_cast(*r.response); + ASSERT_EQ(multi_resp.responses.size(), 2u); + ASSERT_EQ(multi_resp.responses[0]->error, Error::ZOK); + ASSERT_EQ(multi_resp.responses[1]->error, Error::ZOK); + /// Node removed by the TryRemove sub-op + ASSERT_EQ(machine.getStore().getNode("/mnode"), nullptr); + + machine.shutdown(); + cleanDirectory(snap_dir); + cleanDirectory(log_dir); +} + +TEST(RaftStateMachine, MultiRemoveRecursiveRollback) +{ + String snap_dir(SNAP_DIR + "/multi_remrec_rollback"); + String log_dir(LOG_DIR + "/multi_remrec_rollback"); + cleanDirectory(snap_dir); + cleanDirectory(log_dir); + + KeeperResponsesQueue queue; + RaftSettingsPtr setting_ptr = RaftSettings::getDefault(); + std::mutex new_session_id_callback_mutex; + std::unordered_map> new_session_id_callback; + + NuRaftStateMachine machine(queue, setting_ptr, snap_dir, log_dir, 10, 3, new_session_id_callback_mutex, new_session_id_callback); + int64_t session_id = machine.getStore().getSessionID(30000); + + /// Build subtree /r -> /r/a, /r/a/b, /r/c + setNode(machine.getStore(), "r", "root", false, session_id); + setNode(machine.getStore(), "r/a", "a_data", false, session_id); + setNode(machine.getStore(), "r/a/b", "b_data", false, session_id); + setNode(machine.getStore(), "r/c", "c_data", false, session_id); + + auto root_parent = machine.getStore().getNode("/"); + int32_t parent_children_before = root_parent->stat.numChildren; + + /// Multi (write): RemoveRecursive(/r) then a Check that FAILS (wrong version on a + /// node that no longer needs to exist) -> whole multi must roll back, restoring /r. + auto multi = cs_new(); + multi->operation_type = ZooKeeperMultiRequest::OperationType::Write; + multi->xid = 20; + { + auto rr = cs_new(); + rr->path = "/r"; + multi->requests.push_back(rr); + } + { + /// Check on a nonexistent path -> ZNONODE -> triggers rollback + auto ck = cs_new(); + ck->path = "/does_not_exist"; + ck->version = -1; + multi->requests.push_back(ck); + } + + KeeperStore::KeeperResponsesQueue response_queue; + int64_t time = std::chrono::system_clock::now().time_since_epoch() / std::chrono::milliseconds(1); + machine.getStore().processRequest(response_queue, {multi, session_id, time}, {}, true, false); + + ResponseForSession r; + ASSERT_TRUE(response_queue.tryPop(r)); + /// Multi failed, so the whole subtree must be restored intact. + ASSERT_NE(machine.getStore().getNode("/r"), nullptr); + ASSERT_NE(machine.getStore().getNode("/r/a"), nullptr); + ASSERT_NE(machine.getStore().getNode("/r/a/b"), nullptr); + ASSERT_NE(machine.getStore().getNode("/r/c"), nullptr); + /// Data preserved + ASSERT_EQ(machine.getStore().getNode("/r/a/b")->data, "b_data"); + /// Parent link + stat restored + ASSERT_TRUE(machine.getStore().getNode("/")->children.count("r") == 1); + ASSERT_EQ(machine.getStore().getNode("/")->stat.numChildren, parent_children_before); + /// /r still has both children + ASSERT_EQ(machine.getStore().getNode("/r")->children.size(), 2u); + + machine.shutdown(); + cleanDirectory(snap_dir); + cleanDirectory(log_dir); +} diff --git a/src/ZooKeeper/IKeeper.cpp b/src/ZooKeeper/IKeeper.cpp index 851c87582c..48ce8edfe0 100644 --- a/src/ZooKeeper/IKeeper.cpp +++ b/src/ZooKeeper/IKeeper.cpp @@ -145,6 +145,9 @@ void ListRequest::addRootPath(const String & root_path) { Coordination::addRootP void CheckRequest::addRootPath(const String & root_path) { Coordination::addRootPath(path, root_path); } void SetACLRequest::addRootPath(const String & root_path) { Coordination::addRootPath(path, root_path); } void GetACLRequest::addRootPath(const String & root_path) { Coordination::addRootPath(path, root_path); } +void RemoveRecursiveRequest::addRootPath(const String & root_path) { Coordination::addRootPath(path, root_path); } +void CheckStatRequest::addRootPath(const String & root_path) { Coordination::addRootPath(path, root_path); } +void ListRecursiveRequest::addRootPath(const String & root_path) { Coordination::addRootPath(path, root_path); } void MultiRequest::addRootPath(const String & root_path) { diff --git a/src/ZooKeeper/IKeeper.h b/src/ZooKeeper/IKeeper.h index b3a860218f..921b84bc74 100644 --- a/src/ZooKeeper/IKeeper.h +++ b/src/ZooKeeper/IKeeper.h @@ -246,6 +246,8 @@ struct RemoveRequest : virtual Request { String path; int32_t version = -1; + /// If true, don't error when the node doesn't exist (TryRemove semantics) + bool try_remove = false; void addRootPath(const String & root_path) override; String getPath() const override { return path; } @@ -309,6 +311,10 @@ struct ListResponse : virtual Response { CompactStrings names; Stat stat; + + /// Optional per-child fields for FilteredListWithStatsAndData (populated only for OpNum 506). + std::vector stats; + std::vector data; }; struct SimpleListResponse : virtual Response @@ -352,6 +358,50 @@ struct ErrorResponse : virtual Response { }; +struct RemoveRecursiveRequest : virtual Request +{ + String path; + /// Limit on the number of nodes removed in one call. 0 means unlimited. + uint32_t remove_nodes_limit = 0; + + void addRootPath(const String & root_path) override; + String getPath() const override { return path; } +}; + +struct RemoveRecursiveResponse : virtual Response +{ +}; + +struct CheckStatRequest : virtual Request +{ + String path; + int32_t version = -1; + int32_t cversion = -1; + int32_t aversion = -1; + + void addRootPath(const String & root_path) override; + String getPath() const override { return path; } +}; + +struct CheckStatResponse : virtual Response +{ +}; + +struct ListRecursiveRequest : virtual Request +{ + String path; + /// Maximum number of entries to return. 0 means unlimited. + uint32_t max_entries = 0; + + void addRootPath(const String & root_path) override; + String getPath() const override { return path; } +}; + +struct ListRecursiveResponse : virtual Response +{ + CompactStrings names; +}; + using CreateCallback = std::function; using RemoveCallback = std::function; diff --git a/src/ZooKeeper/ZooKeeperCommon.cpp b/src/ZooKeeper/ZooKeeperCommon.cpp index 8d8e29640b..d004c33476 100644 --- a/src/ZooKeeper/ZooKeeperCommon.cpp +++ b/src/ZooKeeper/ZooKeeperCommon.cpp @@ -158,6 +158,71 @@ void ZooKeeperRemoveRequest::readImpl(ReadBuffer & in) Coordination::read(version, in); } +void ZooKeeperRemoveRecursiveRequest::writeImpl(WriteBuffer & out) const +{ + Coordination::write(path, out); + Coordination::write(remove_nodes_limit, out); +} + +void ZooKeeperRemoveRecursiveRequest::readImpl(ReadBuffer & in) +{ + Coordination::read(path, in); + Coordination::read(remove_nodes_limit, in); +} + +ZooKeeperResponsePtr ZooKeeperRemoveRecursiveRequest::makeResponse() const +{ + return std::make_shared(); +} + +void ZooKeeperCheckStatRequest::writeImpl(WriteBuffer & out) const +{ + Coordination::write(path, out); + Coordination::write(version, out); + Coordination::write(cversion, out); + Coordination::write(aversion, out); +} + +void ZooKeeperCheckStatRequest::readImpl(ReadBuffer & in) +{ + Coordination::read(path, in); + Coordination::read(version, in); + Coordination::read(cversion, in); + Coordination::read(aversion, in); +} + +ZooKeeperResponsePtr ZooKeeperCheckStatRequest::makeResponse() const +{ + return std::make_shared(); +} + +void ZooKeeperListRecursiveRequest::writeImpl(WriteBuffer & out) const +{ + Coordination::write(path, out); + Coordination::write(max_entries, out); +} + +void ZooKeeperListRecursiveRequest::readImpl(ReadBuffer & in) +{ + Coordination::read(path, in); + Coordination::read(max_entries, in); +} + +ZooKeeperResponsePtr ZooKeeperListRecursiveRequest::makeResponse() const +{ + return std::make_shared(); +} + +void ZooKeeperListRecursiveResponse::readImpl(ReadBuffer & in) +{ + Coordination::read(names, in); +} + +void ZooKeeperListRecursiveResponse::writeImpl(WriteBuffer & out) const +{ + Coordination::write(names, out); +} + void ZooKeeperExistsRequest::writeImpl(WriteBuffer & out) const { Coordination::write(path, out); @@ -251,6 +316,11 @@ void ZooKeeperFilteredListRequest::writeImpl(WriteBuffer & out) const Coordination::write(path, out); Coordination::write(has_watch, out); Coordination::write(static_cast(list_request_type), out); + if (list_with_stats_and_data) + { + Coordination::write(with_stat, out); + Coordination::write(with_data, out); + } } void ZooKeeperFilteredListRequest::readImpl(ReadBuffer & in) @@ -261,6 +331,11 @@ void ZooKeeperFilteredListRequest::readImpl(ReadBuffer & in) uint8_t read_request_type{0}; Coordination::read(read_request_type, in); list_request_type = static_cast(read_request_type); + if (list_with_stats_and_data) + { + Coordination::read(with_stat, in); + Coordination::read(with_data, in); + } } void ZooKeeperListResponse::writeImpl(WriteBuffer & out) const @@ -269,6 +344,22 @@ void ZooKeeperListResponse::writeImpl(WriteBuffer & out) const Coordination::write(stat, out); } +void ZooKeeperFilteredListWithStatsAndDataResponse::readImpl(ReadBuffer & in) +{ + Coordination::read(names, in); + Coordination::read(stat, in); + Coordination::read(stats, in); + Coordination::read(data, in); +} + +void ZooKeeperFilteredListWithStatsAndDataResponse::writeImpl(WriteBuffer & out) const +{ + Coordination::write(names, out); + Coordination::write(stat, out); + Coordination::write(stats, out); + Coordination::write(data, out); +} + void ZooKeeperSimpleListResponse::readImpl(ReadBuffer & in) { Coordination::read(names, in); @@ -580,11 +671,18 @@ ZooKeeperResponsePtr ZooKeeperSetWatchesRequest::makeResponse() const { return s ZooKeeperResponsePtr ZooKeeperSyncRequest::makeResponse() const { return std::make_shared(); } ZooKeeperResponsePtr ZooKeeperAuthRequest::makeResponse() const { return std::make_shared(); } ZooKeeperResponsePtr ZooKeeperCreateRequest::makeResponse() const { return std::make_shared(); } -ZooKeeperResponsePtr ZooKeeperRemoveRequest::makeResponse() const { return std::make_shared(); } +ZooKeeperResponsePtr ZooKeeperRemoveRequest::makeResponse() const { return try_remove ? std::static_pointer_cast(std::make_shared()) : std::static_pointer_cast(std::make_shared()); } ZooKeeperResponsePtr ZooKeeperExistsRequest::makeResponse() const { return std::make_shared(); } ZooKeeperResponsePtr ZooKeeperGetRequest::makeResponse() const { return std::make_shared(); } ZooKeeperResponsePtr ZooKeeperSetRequest::makeResponse() const { return std::make_shared(); } ZooKeeperResponsePtr ZooKeeperListRequest::makeResponse() const { return std::make_shared(); } + +ZooKeeperResponsePtr ZooKeeperFilteredListRequest::makeResponse() const +{ + if (list_with_stats_and_data) + return std::make_shared(); + return std::make_shared(); +} ZooKeeperResponsePtr ZooKeeperSimpleListRequest::makeResponse() const { return std::make_shared(); } ZooKeeperResponsePtr ZooKeeperCheckRequest::makeResponse() const { return not_exists ? std::make_shared() : std::make_shared(); } ZooKeeperResponsePtr ZooKeeperCloseRequest::makeResponse() const { return std::make_shared(); } @@ -725,6 +823,10 @@ void registerZooKeeperRequest(ZooKeeperRequestFactory & factory) res->operation_type = ZooKeeperMultiRequest::OperationType::Write; else if constexpr (num == OpNum::CheckNotExists || num == OpNum::CreateIfNotExists) res->not_exists = true; + else if constexpr (num == OpNum::TryRemove) + res->try_remove = true; + else if constexpr (num == OpNum::FilteredListWithStatsAndData) + res->list_with_stats_and_data = true; return res; }); @@ -754,6 +856,11 @@ ZooKeeperRequestFactory::ZooKeeperRequestFactory() registerZooKeeperRequest(*this); registerZooKeeperRequest(*this); registerZooKeeperRequest(*this); + registerZooKeeperRequest(*this); + registerZooKeeperRequest(*this); + registerZooKeeperRequest(*this); + registerZooKeeperRequest(*this); + registerZooKeeperRequest(*this); } } diff --git a/src/ZooKeeper/ZooKeeperCommon.h b/src/ZooKeeper/ZooKeeperCommon.h index 66fe8869a7..d4e31bc0f3 100644 --- a/src/ZooKeeper/ZooKeeperCommon.h +++ b/src/ZooKeeper/ZooKeeperCommon.h @@ -263,7 +263,7 @@ struct ZooKeeperRemoveRequest final : RemoveRequest, ZooKeeperRequest ZooKeeperRemoveRequest() = default; explicit ZooKeeperRemoveRequest(const RemoveRequest & base) : RemoveRequest(base) { } - OpNum getOpNum() const override { return OpNum::Remove; } + OpNum getOpNum() const override { return try_remove ? OpNum::TryRemove : OpNum::Remove; } void writeImpl(WriteBuffer & out) const override; void readImpl(ReadBuffer & in) override; @@ -278,13 +278,37 @@ struct ZooKeeperRemoveRequest final : RemoveRequest, ZooKeeperRequest } }; -struct ZooKeeperRemoveResponse final : RemoveResponse, ZooKeeperResponse +struct ZooKeeperRemoveResponse : RemoveResponse, ZooKeeperResponse { void readImpl(ReadBuffer &) override { } void writeImpl(WriteBuffer &) const override { } OpNum getOpNum() const override { return OpNum::Remove; } }; +struct ZooKeeperTryRemoveResponse final : ZooKeeperRemoveResponse +{ + OpNum getOpNum() const override { return OpNum::TryRemove; } +}; + +struct ZooKeeperRemoveRecursiveRequest final : RemoveRecursiveRequest, ZooKeeperRequest +{ + ZooKeeperRemoveRecursiveRequest() = default; + explicit ZooKeeperRemoveRecursiveRequest(const RemoveRecursiveRequest & base) : RemoveRecursiveRequest(base) { } + + OpNum getOpNum() const override { return OpNum::RemoveRecursive; } + void writeImpl(WriteBuffer & out) const override; + void readImpl(ReadBuffer & in) override; + ZooKeeperResponsePtr makeResponse() const override; + bool isReadRequest() const override { return false; } +}; + +struct ZooKeeperRemoveRecursiveResponse final : RemoveRecursiveResponse, ZooKeeperResponse +{ + void readImpl(ReadBuffer &) override { } + void writeImpl(WriteBuffer &) const override { } + OpNum getOpNum() const override { return OpNum::RemoveRecursive; } +}; + struct ZooKeeperExistsRequest final : ExistsRequest, ZooKeeperRequest { ZooKeeperExistsRequest() = default; @@ -441,10 +465,21 @@ struct ZooKeeperFilteredListRequest final : ZooKeeperListRequest } ListRequestType list_request_type{ListRequestType::ALL}; + bool with_stat = false; + bool with_data = false; + /// Set by the request factory for OpNum::FilteredListWithStatsAndData (506). + /// Controls whether with_stat/with_data are present on the wire. + bool list_with_stats_and_data = false; - OpNum getOpNum() const override { return OpNum::FilteredList; } + OpNum getOpNum() const override + { + if (list_with_stats_and_data) + return OpNum::FilteredListWithStatsAndData; + return OpNum::FilteredList; + } void writeImpl(WriteBuffer & out) const override; void readImpl(ReadBuffer & in) override; + ZooKeeperResponsePtr makeResponse() const override; String toString() const override { return Coordination::toString(getOpNum()) + ", xid " + std::to_string(xid) + ", path " + path @@ -480,6 +515,13 @@ struct ZooKeeperListResponse final : ListResponse, ZooKeeperResponse } }; +struct ZooKeeperFilteredListWithStatsAndDataResponse final : ListResponse, ZooKeeperResponse +{ + void readImpl(ReadBuffer & in) override; + void writeImpl(WriteBuffer & out) const override; + OpNum getOpNum() const override { return OpNum::FilteredListWithStatsAndData; } +}; + struct ZooKeeperSimpleListResponse final : SimpleListResponse, ZooKeeperResponse { void readImpl(ReadBuffer & in) override; @@ -542,6 +584,51 @@ struct ZooKeeperCheckNotExistsResponse : public ZooKeeperCheckResponse using ZooKeeperCheckResponse::ZooKeeperCheckResponse; }; +struct ZooKeeperCheckStatRequest final : CheckStatRequest, ZooKeeperRequest +{ + ZooKeeperCheckStatRequest() = default; + explicit ZooKeeperCheckStatRequest(const CheckStatRequest & base) : CheckStatRequest(base) { } + + OpNum getOpNum() const override { return OpNum::CheckStat; } + void writeImpl(WriteBuffer & out) const override; + void readImpl(ReadBuffer & in) override; + ZooKeeperResponsePtr makeResponse() const override; + bool isReadRequest() const override { return false; } + String toString() const override + { + return Coordination::toString(getOpNum()) + ", xid " + std::to_string(xid) + ", path " + path; + } +}; + +struct ZooKeeperCheckStatResponse final : ZooKeeperCheckResponse +{ + OpNum getOpNum() const override { return OpNum::CheckStat; } + using ZooKeeperCheckResponse::ZooKeeperCheckResponse; +}; + +struct ZooKeeperListRecursiveRequest final : ListRecursiveRequest, ZooKeeperRequest +{ + ZooKeeperListRecursiveRequest() = default; + explicit ZooKeeperListRecursiveRequest(const ListRecursiveRequest & base) : ListRecursiveRequest(base) { } + + OpNum getOpNum() const override { return OpNum::ListRecursive; } + void writeImpl(WriteBuffer & out) const override; + void readImpl(ReadBuffer & in) override; + ZooKeeperResponsePtr makeResponse() const override; + bool isReadRequest() const override { return true; } + String toString() const override + { + return Coordination::toString(getOpNum()) + ", xid " + std::to_string(xid) + ", path " + path; + } +}; + +struct ZooKeeperListRecursiveResponse final : ListRecursiveResponse, ZooKeeperResponse +{ + void readImpl(ReadBuffer & in) override; + void writeImpl(WriteBuffer & out) const override; + OpNum getOpNum() const override { return OpNum::ListRecursive; } +}; + /// This response may be received only as an element of responses in MultiResponse. struct ZooKeeperErrorResponse final : ErrorResponse, ZooKeeperResponse { diff --git a/src/ZooKeeper/ZooKeeperConstants.cpp b/src/ZooKeeper/ZooKeeperConstants.cpp index 1730c512de..2f4645adc0 100644 --- a/src/ZooKeeper/ZooKeeperConstants.cpp +++ b/src/ZooKeeper/ZooKeeperConstants.cpp @@ -30,6 +30,11 @@ static const std::unordered_set VALID_OPERATIONS = static_cast(OpNum::FilteredList), static_cast(OpNum::CheckNotExists), static_cast(OpNum::CreateIfNotExists), + static_cast(OpNum::RemoveRecursive), + static_cast(OpNum::CheckStat), + static_cast(OpNum::TryRemove), + static_cast(OpNum::FilteredListWithStatsAndData), + static_cast(OpNum::ListRecursive), static_cast(OpNum::UpdateSession), }; @@ -87,6 +92,16 @@ std::string toString(OpNum op_num) return "CheckNotExists"; case OpNum::CreateIfNotExists: return "CreateIfNotExists"; + case OpNum::RemoveRecursive: + return "RemoveRecursive"; + case OpNum::CheckStat: + return "CheckStat"; + case OpNum::TryRemove: + return "TryRemove"; + case OpNum::FilteredListWithStatsAndData: + return "FilteredListWithStatsAndData"; + case OpNum::ListRecursive: + return "ListRecursive"; } int32_t raw_op = static_cast(op_num); throw Exception("Operation " + std::to_string(raw_op) + " is unknown", Error::ZUNIMPLEMENTED); diff --git a/src/ZooKeeper/ZooKeeperConstants.h b/src/ZooKeeper/ZooKeeperConstants.h index fb13182219..8c8415a2d2 100644 --- a/src/ZooKeeper/ZooKeeperConstants.h +++ b/src/ZooKeeper/ZooKeeperConstants.h @@ -46,6 +46,11 @@ enum class OpNum : int32_t FilteredList = 500, CheckNotExists = 501, CreateIfNotExists = 502, + RemoveRecursive = 503, + CheckStat = 504, + TryRemove = 505, + FilteredListWithStatsAndData = 506, + ListRecursive = 507, UpdateSession = 998, /// Special internal request. Used to session reconnect. }; diff --git a/tests/integration/helpers/utils.py b/tests/integration/helpers/utils.py index 164985b165..ef22262748 100644 --- a/tests/integration/helpers/utils.py +++ b/tests/integration/helpers/utils.py @@ -198,6 +198,111 @@ def deserialize(cls, bytes, offset): return children, stat +class RemoveRecursive(namedtuple('RemoveRecursive', 'path remove_nodes_limit')): + type = 503 + + def serialize(self): + b = bytearray() + b.extend(write_string(self.path)) + b.extend(struct.pack('!I', self.remove_nodes_limit)) + return b + + @classmethod + def deserialize(cls, bytes, offset): + return True + + +class TryRemove(namedtuple('TryRemove', 'path version')): + type = 505 + + def serialize(self): + b = bytearray() + b.extend(write_string(self.path)) + b.extend(struct.pack('!i', self.version)) + return b + + @classmethod + def deserialize(cls, bytes, offset): + return True + + +class ListRecursive(namedtuple('ListRecursive', 'path max_entries')): + type = 507 + + def serialize(self): + b = bytearray() + b.extend(write_string(self.path)) + b.extend(struct.pack('!I', self.max_entries)) + return b + + @classmethod + def deserialize(cls, bytes, offset): + count = int_struct.unpack_from(bytes, offset)[0] + offset += int_struct.size + children = [] + for c in range(count): + child, offset = read_string(bytes, offset) + children.append(child) + return children + + +class CheckStat(namedtuple('CheckStat', 'path version cversion aversion')): + type = 504 + + def serialize(self): + b = bytearray() + b.extend(write_string(self.path)) + b.extend(int_struct.pack(self.version)) + b.extend(int_struct.pack(self.cversion)) + b.extend(int_struct.pack(self.aversion)) + return b + + @classmethod + def deserialize(cls, bytes, offset): + return True + + +class FilteredListWithStatsAndData(namedtuple('FilteredListWithStatsAndData', 'path watcher list_type with_stat with_data')): + type = 506 + + def serialize(self): + b = bytearray() + b.extend(write_string(self.path)) + b.extend([1 if self.watcher else 0]) + b.extend(struct.pack('B', self.list_type)) + b.extend([1 if self.with_stat else 0]) + b.extend([1 if self.with_data else 0]) + return b + + @classmethod + def deserialize(cls, bytes, offset): + # names + count = int_struct.unpack_from(bytes, offset)[0] + offset += int_struct.size + children = [] + for _ in range(count): + child, offset = read_string(bytes, offset) + children.append(child) + # parent stat + stat = ZnodeStat._make(stat_struct.unpack_from(bytes, offset)) + offset += stat_struct.size + # per-child stats + stats_count = int_struct.unpack_from(bytes, offset)[0] + offset += int_struct.size + stats = [] + for _ in range(stats_count): + stats.append(ZnodeStat._make(stat_struct.unpack_from(bytes, offset))) + offset += stat_struct.size + # per-child data + data_count = int_struct.unpack_from(bytes, offset)[0] + offset += int_struct.size + data = [] + for _ in range(data_count): + d, offset = read_buffer(bytes, offset) + data.append(d) + return children, stat, stats, data + + class KeeperFeatureClient(KazooClient): """A Zookeeper Python client extends from Kazoo.KazooClient, Kazoo is a Python library working with Zookeeper. @@ -239,6 +344,52 @@ def multi_read(self): """ return MultiReadRequest(self) + def remove_recursive(self, path, remove_nodes_limit=0): + """Remove a node and all its descendants atomically. + + :returns: True on success. + :raises NoNodeError: if the node doesn't exist. + """ + async_result = self.handler.async_result() + self._call(RemoveRecursive(_prefix_root(self.chroot, path), remove_nodes_limit), async_result) + return async_result.get() + + def try_remove(self, path, version=-1): + """Remove a node if it exists. No error if the node is absent. + + :returns: True on success. + """ + async_result = self.handler.async_result() + self._call(TryRemove(_prefix_root(self.chroot, path), version), async_result) + return async_result.get() + + def list_recursive(self, path, max_entries=0): + """List all descendants of a node recursively. + + :returns: List of full paths under the given node. + """ + async_result = self.handler.async_result() + self._call(ListRecursive(_prefix_root(self.chroot, path), max_entries), async_result) + return async_result.get() + + def check_stat(self, path, version=-1, cversion=-1, aversion=-1): + """Check a node's version/cversion/aversion (OpNum 504). + + :returns: True on match. Raises on mismatch/missing node. + """ + async_result = self.handler.async_result() + self._call(CheckStat(_prefix_root(self.chroot, path), version, cversion, aversion), async_result) + return async_result.get() + + def list_children_with_stats_and_data(self, path, list_type=0, with_stat=True, with_data=True, watch=None): + """FilteredListWithStatsAndData (OpNum 506). + + :returns: (children, stat, per_child_stats, per_child_data) + """ + async_result = self.handler.async_result() + self._call(FilteredListWithStatsAndData(_prefix_root(self.chroot, path), watch, list_type, with_stat, with_data), async_result) + return async_result.get() + def get_filtered_children(self, path, watch=None, list_type=None, include_data=False): """Get a list of child nodes of a path. @@ -576,6 +727,18 @@ def check_if_not_exists(self, path, version): CheckIfNotExistsVersion(_prefix_root(self.client.chroot, path), version) ) + def check_stat(self, path, version=-1, cversion=-1, aversion=-1): + """Add a CheckStat (OpNum 504) condition to the transaction.""" + self._add(CheckStat(_prefix_root(self.client.chroot, path), version, cversion, aversion)) + + def try_remove(self, path, version=-1): + """Add a TryRemove (OpNum 505) to the transaction.""" + self._add(TryRemove(_prefix_root(self.client.chroot, path), version)) + + def remove_recursive(self, path, remove_nodes_limit=0): + """Add a RemoveRecursive (OpNum 503) to the transaction.""" + self._add(RemoveRecursive(_prefix_root(self.client.chroot, path), remove_nodes_limit)) + def create_if_not_exist(self, path, value=b"", acl=None, ephemeral=False, sequence=False): """Add a create ZNode ops to the operations. @@ -615,6 +778,7 @@ def create_if_not_exist(self, path, value=b"", acl=None, ephemeral=False, self._add(CreateIfNotExists(_prefix_root(self.client.chroot, path), value, acl, flags), None) + class MultiRead(namedtuple('MultiRead', 'operations')): type = 22 diff --git a/tests/integration/test_back_to_back/test.py b/tests/integration/test_back_to_back/test.py index 45d168e427..4ea51e0c8c 100644 --- a/tests/integration/test_back_to_back/test.py +++ b/tests/integration/test_back_to_back/test.py @@ -585,7 +585,9 @@ def read_func2(zk, path=path, callback=callback): def test_random_requests(started_cluster): genuine_zk = fake_zk = None try: - requests = generate_requests("/test_random_requests", 10) + # ponytail: 10 iters (~3000 serial round-trips x2 clients) times out under + # sanitizers (~10x slowdown). 3 iters keeps broad randomized coverage within 300s. + requests = generate_requests("/test_random_requests", 3) print("Generated", len(requests), "requests") genuine_zk = get_genuine_zk() fake_zk = get_fake_zk() @@ -947,4 +949,255 @@ def test_multi_read_subrequest_watch(started_cluster): "Exists watch from MultiRead subrequest not registered on server" finally: - close_zk_clients([fake_zk]) \ No newline at end of file + close_zk_clients([fake_zk]) + + +def test_remove_recursive(started_cluster): + fake_zk = None + try: + fake_zk = get_fake_zk(True) + fake_zk.start() + + fake_zk.create('/test_rem_rec') + fake_zk.create('/test_rem_rec/a', b'data_a') + fake_zk.create('/test_rem_rec/a/aa', b'data_aa') + fake_zk.create('/test_rem_rec/b', b'data_b') + fake_zk.create('/test_rem_rec/c', b'data_c') + + assert fake_zk.exists('/test_rem_rec/a/aa') is not None + + fake_zk.remove_recursive('/test_rem_rec') + + assert fake_zk.exists('/test_rem_rec') is None + assert fake_zk.exists('/test_rem_rec/a') is None + assert fake_zk.exists('/test_rem_rec/a/aa') is None + assert fake_zk.exists('/test_rem_rec/b') is None + + finally: + close_zk_clients([fake_zk]) + + +def test_try_remove(started_cluster): + fake_zk = None + try: + fake_zk = get_fake_zk(True) + fake_zk.start() + + fake_zk.create('/test_tryrem') + fake_zk.create('/test_tryrem/exists', b'data') + + # try_remove on existing node — succeeds + fake_zk.try_remove('/test_tryrem/exists') + assert fake_zk.exists('/test_tryrem/exists') is None + + # try_remove on nonexistent node — succeeds (no error) + fake_zk.try_remove('/test_tryrem/nonexistent') + + finally: + close_zk_clients([fake_zk]) + + +def test_list_recursive(started_cluster): + fake_zk = None + try: + fake_zk = get_fake_zk(True) + fake_zk.start() + + fake_zk.create('/test_listrec') + fake_zk.create('/test_listrec/x', b'x_data') + fake_zk.create('/test_listrec/y', b'y_data') + fake_zk.create('/test_listrec/y/z', b'z_data') + + names = fake_zk.list_recursive('/test_listrec') + assert len(names) == 3 + assert '/test_listrec/x' in names + assert '/test_listrec/y' in names + assert '/test_listrec/y/z' in names + + finally: + close_zk_clients([fake_zk]) + +def test_check_stat(started_cluster): + fake_zk = None + try: + fake_zk = get_fake_zk(True) + fake_zk.start() + + fake_zk.create('/test_check_stat', b'data') + stat = fake_zk.exists('/test_check_stat') + + # Matching version/cversion/aversion -> succeeds + fake_zk.check_stat('/test_check_stat', version=stat.version, + cversion=stat.cversion, aversion=stat.aversion) + + # Wrong version -> BadVersionError + from kazoo.exceptions import BadVersionError, NoNodeError + got = False + try: + fake_zk.check_stat('/test_check_stat', version=stat.version + 1) + except BadVersionError: + got = True + assert got, "CheckStat with wrong version should raise BadVersionError" + + # Nonexistent node -> NoNodeError + got = False + try: + fake_zk.check_stat('/test_check_stat_missing', version=-1) + except NoNodeError: + got = True + assert got, "CheckStat on missing node should raise NoNodeError" + finally: + close_zk_clients([fake_zk]) + + +def test_filtered_list_with_stats_and_data(started_cluster): + fake_zk = None + try: + fake_zk = get_fake_zk(True) + fake_zk.start() + + fake_zk.create('/test_flist_sd') + fake_zk.create('/test_flist_sd/x', b'data_x') + fake_zk.create('/test_flist_sd/y', b'data_y') + + children, stat, stats, data = fake_zk.list_children_with_stats_and_data( + '/test_flist_sd', list_type=0, with_stat=True, with_data=True) + + assert sorted(children) == ['x', 'y'] + assert len(stats) == 2 + assert len(data) == 2 + # data entries correspond to children by index + by_child = dict(zip(children, data)) + assert by_child['x'] == b'data_x' + assert by_child['y'] == b'data_y' + finally: + close_zk_clients([fake_zk]) + + +def test_try_remove_fires_watch(started_cluster): + fake_zk = None + try: + fake_zk = get_fake_zk(True) + fake_zk.start() + + fake_zk.create('/test_tryrem_watch', b'data') + + events = [] + def watch_cb(event): + events.append(event) + + # data watch via get + fake_zk.get('/test_tryrem_watch', watch=watch_cb) + fake_zk.try_remove('/test_tryrem_watch') + time.sleep(3) + + assert len(events) >= 1, "TryRemove on existing node must fire a watch event" + assert events[0].type == 'DELETED', f"expected DELETED, got {events[0].type}" + finally: + close_zk_clients([fake_zk]) + + +def test_remove_recursive_fires_watches(started_cluster): + fake_zk = None + try: + fake_zk = get_fake_zk(True) + fake_zk.start() + + fake_zk.create('/test_remrec_watch') + fake_zk.create('/test_remrec_watch/a', b'a') + fake_zk.create('/test_remrec_watch/a/b', b'b') + + deleted = [] + def cb_a(event): + if event.type == 'DELETED': + deleted.append(event.path) + def cb_b(event): + if event.type == 'DELETED': + deleted.append(event.path) + + fake_zk.get('/test_remrec_watch/a', watch=cb_a) + fake_zk.get('/test_remrec_watch/a/b', watch=cb_b) + + fake_zk.remove_recursive('/test_remrec_watch') + time.sleep(3) + + assert '/test_remrec_watch/a' in deleted + assert '/test_remrec_watch/a/b' in deleted + finally: + close_zk_clients([fake_zk]) + + +def test_remove_recursive_limit(started_cluster): + fake_zk = None + try: + fake_zk = get_fake_zk(True) + fake_zk.start() + + fake_zk.create('/test_remrec_limit') + fake_zk.create('/test_remrec_limit/a') + fake_zk.create('/test_remrec_limit/b') + fake_zk.create('/test_remrec_limit/c') + + # 4 nodes total (root + 3 children); limit of 2 must reject with NotEmpty + from kazoo.exceptions import NotEmptyError + got = False + try: + fake_zk.remove_recursive('/test_remrec_limit', remove_nodes_limit=2) + except NotEmptyError: + got = True + assert got, "RemoveRecursive over the node limit must raise NotEmptyError" + # Nothing removed + assert fake_zk.exists('/test_remrec_limit') is not None + assert fake_zk.exists('/test_remrec_limit/a') is not None + + # Sufficient limit succeeds + fake_zk.remove_recursive('/test_remrec_limit', remove_nodes_limit=10) + assert fake_zk.exists('/test_remrec_limit') is None + finally: + close_zk_clients([fake_zk]) + + +def test_list_recursive_max_entries(started_cluster): + fake_zk = None + try: + fake_zk = get_fake_zk(True) + fake_zk.start() + + fake_zk.create('/test_listrec_max') + for i in range(5): + fake_zk.create(f'/test_listrec_max/n{i}') + + # Unlimited returns all 5 + all_names = fake_zk.list_recursive('/test_listrec_max') + assert len(all_names) == 5 + + # max_entries caps the result + limited = fake_zk.list_recursive('/test_listrec_max', max_entries=2) + assert len(limited) == 2 + finally: + close_zk_clients([fake_zk]) + + +def test_multi_remove_recursive_rollback(started_cluster): + fake_zk = None + try: + fake_zk = get_fake_zk(True) + fake_zk.start() + + fake_zk.create('/test_multi_rr') + fake_zk.create('/test_multi_rr/a', b'a') + fake_zk.create('/test_multi_rr/a/b', b'b') + + # Multi: RemoveRecursive(/test_multi_rr) + a Check that fails -> rollback + t = fake_zk.transaction() + t.remove_recursive('/test_multi_rr') + t.check('/does_not_exist_xyz', version=-1) # fails -> whole multi rolls back + results = t.commit() + + # Subtree must be fully restored + assert fake_zk.exists('/test_multi_rr') is not None + assert fake_zk.exists('/test_multi_rr/a') is not None + assert fake_zk.exists('/test_multi_rr/a/b') is not None + assert fake_zk.get('/test_multi_rr/a/b')[0] == b'b' + finally: + close_zk_clients([fake_zk]) diff --git a/tests/integration/test_four_word_command/configs/enable_keeper1.xml b/tests/integration/test_four_word_command/configs/enable_keeper1.xml index 7b62695cc0..05e3b1a10d 100644 --- a/tests/integration/test_four_word_command/configs/enable_keeper1.xml +++ b/tests/integration/test_four_word_command/configs/enable_keeper1.xml @@ -7,7 +7,7 @@ 16 - 1000 + 10000 1000 30000 3000000 diff --git a/tests/integration/test_four_word_command/configs/enable_keeper2.xml b/tests/integration/test_four_word_command/configs/enable_keeper2.xml index 09ce14236d..c8b2bef7c4 100644 --- a/tests/integration/test_four_word_command/configs/enable_keeper2.xml +++ b/tests/integration/test_four_word_command/configs/enable_keeper2.xml @@ -6,7 +6,7 @@ * - 1000 + 10000 1000 30000 debug diff --git a/tests/integration/test_four_word_command/configs/enable_keeper3.xml b/tests/integration/test_four_word_command/configs/enable_keeper3.xml index 178eff38f2..a0b1ca226c 100644 --- a/tests/integration/test_four_word_command/configs/enable_keeper3.xml +++ b/tests/integration/test_four_word_command/configs/enable_keeper3.xml @@ -6,7 +6,7 @@ * - 1000 + 10000 1000 30000 debug diff --git a/tests/integration/test_four_word_command/test.py b/tests/integration/test_four_word_command/test.py index d6cca4bb55..8ceb070456 100644 --- a/tests/integration/test_four_word_command/test.py +++ b/tests/integration/test_four_word_command/test.py @@ -294,7 +294,10 @@ def test_cmd_conf(started_cluster): assert result["client_req_timeout_ms"] == result["operation_timeout_ms"] assert result["min_session_timeout_ms"] == "1000" assert result["max_session_timeout_ms"] == "30000" - assert result["operation_timeout_ms"] == "1000" + # 10s, not the 3s default: under sanitizer slowdown a 1s client_req_timeout + # makes commits exceed it, hitting NuRaft's commit-callback timeout path + # (which has an unsynchronized read that tsan reports as a data race). + assert result["operation_timeout_ms"] == "10000" assert result["dead_session_check_period_ms"] == "500" assert result["heart_beat_interval_ms"] == "500" assert result["election_timeout_lower_bound_ms"] == "3000" diff --git a/tests/integration/test_nodes_replace/test.py b/tests/integration/test_nodes_replace/test.py index 09590b6acb..ca1001a780 100644 --- a/tests/integration/test_nodes_replace/test.py +++ b/tests/integration/test_nodes_replace/test.py @@ -28,7 +28,7 @@ def started_cluster(): def start(node): - node.start_raftkeeper(start_wait=True) + node.start_raftkeeper(start_wait=False) def test_node_replace(started_cluster): diff --git a/tests/integration/test_session_fake_client/test.py b/tests/integration/test_session_fake_client/test.py index 15134b254e..5f162d2d2b 100644 --- a/tests/integration/test_session_fake_client/test.py +++ b/tests/integration/test_session_fake_client/test.py @@ -185,14 +185,14 @@ def test_session_max_min_session_timeout(started_cluster): assert len(heartbeat(client2)) > 0 assert len(heartbeat(client3)) > 0 - time.sleep(2) - # 2s after the first heartbeat, client1 session should expire + time.sleep(3) + # 3s after the first heartbeat, client1 session should expire assert len(heartbeat(client1)) == 0 assert len(heartbeat(client2)) > 0 assert len(heartbeat(client3)) > 0 - time.sleep(5) - # 5s after the second heartbeat, client2 session should expire + time.sleep(6) + # 6s after the second heartbeat, client2 session should expire assert len(heartbeat(client2)) == 0 assert len(heartbeat(client3)) > 0 @@ -218,7 +218,7 @@ def test_invalid_timeout_setting(started_cluster): node1.stop_raftkeeper() time.sleep(3) node1.replace_in_config('/etc/raftkeeper-server/config.d/enable_keeper1.xml', '12000', '200') - node1.start_raftkeeper() + node1.start_raftkeeper(start_wait=True) data = send_4lw_cmd(node1.name, cmd='conf') reader = csv.reader(data.split('\n'), delimiter='=') diff --git a/tests/integration/test_snapshots/test.py b/tests/integration/test_snapshots/test.py index 3375862793..b2d344dff6 100644 --- a/tests/integration/test_snapshots/test.py +++ b/tests/integration/test_snapshots/test.py @@ -159,6 +159,21 @@ def get_snapshots(node): return snapshots +def wait_for_snapshots(node, expected_count, timeout=30): + # ponytail: poll instead of a fixed sleep(1) — under sanitizers the snapshot dir + # may not exist yet right after csnp, which made get_snapshots' ls fail intermittently. + deadline = time.time() + timeout + while time.time() < deadline: + try: + snapshots = get_snapshots(node) + if len(snapshots) == expected_count: + return snapshots + except Exception: + pass + time.sleep(0.5) + return get_snapshots(node) + + @pytest.mark.parametrize( 'node', [ @@ -175,8 +190,7 @@ def test_snapshot_clear(started_cluster, node): node_zk.create(f"/test_node_clear_{i}", b"test") node.send_4lw_cmd(cmd="csnp") # wait for snapshot to be taken - time.sleep(1) - snapshots = get_snapshots(node) + snapshots = wait_for_snapshots(node, 1) assert (len(snapshots) == 1) finally: close_zk_clients([node_zk, node_zk2]) \ No newline at end of file