diff --git a/CHANGELOG.md b/CHANGELOG.md index a87d8bd9c02..aef89a0541d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. [7.0.14]: https://github.com/microsoft/CCF/releases/tag/ccf-7.0.14 +### Added + +- Pending node entries are now automatically removed when they stop retrying joins for the configurable `pending_node_timeout` delay (1 hour by default). Set it to `0s` to disable automatic removal (#8173). + ### Fixed - If the view changed while a transaction was committing, the transaction could apply its writes to the local key-value store and then fail to replicate, leaving state that never reached consensus. The transaction's view is now validated atomically with the allocation of its version, so it is rejected before any map is modified, and `ccf::kv::CommitResult::FAIL_NO_REPLICATE` no longer implies a locally applied write (#8242). diff --git a/doc/host_config_schema/host_config.json b/doc/host_config_schema/host_config.json index 781cad58269..1edc325a72b 100644 --- a/doc/host_config_schema/host_config.json +++ b/doc/host_config_schema/host_config.json @@ -575,6 +575,11 @@ "description": "This section includes configuration for periodic cleanup of old files (snapshots, ledger chunks)", "additionalProperties": false }, + "pending_node_timeout": { + "type": "string", + "default": "1h", + "description": "Delay after which the primary automatically removes a Pending node that has stopped sending join requests. Cleanup runs at least once per minute, so removal may occur up to one minute after this delay. Set to 0s to disable automatic removal." + }, "identity_history_fetch": { "type": "object", "properties": { diff --git a/doc/operations/start_network.rst b/doc/operations/start_network.rst index 022a5b298bd..4f5cea1b403 100644 --- a/doc/operations/start_network.rst +++ b/doc/operations/start_network.rst @@ -62,6 +62,8 @@ If the network has not yet been opened by members (see :ref:`governance/open_net The ``Pending`` joining node automatically polls the service (interval configurable via ``join.retry_timeout`` configuration entry) until the members have successfully transitioned the node to the ``Trusted`` state. It is only then that the joining node transitions to the ``PartOfNetwork`` state and starts updating its ledger. +The primary automatically removes a ``Pending`` node that has stopped sending join requests for longer than the ``pending_node_timeout`` configuration entry (1 hour by default). Cleanup runs at least once per minute. Set ``pending_node_timeout`` to ``0s`` to disable automatic removal. + .. tip:: After the node has been trusted by members, operators should poll the :http:GET:`/node/state` endpoint on the newly added node, using the node's self-signed certificate as TLS CA, until the ``{"state": "PartOfNetwork"}`` is reported. This status confirms that the replication of the ledger has started on this node. .. note:: To accelerate the joining procedure, it is possible for new nodes to join from a snapshot. More information on snapshots :ref:`here `. diff --git a/include/ccf/node/startup_config.h b/include/ccf/node/startup_config.h index 190fa236c28..13bfd774ee9 100644 --- a/include/ccf/node/startup_config.h +++ b/include/ccf/node/startup_config.h @@ -29,6 +29,8 @@ namespace ccf ccf::ds::SizeString historical_cache_soft_limit = {"512MB"}; + ccf::ds::TimeString pending_node_timeout = {"1h"}; + ccf::consensus::Configuration consensus = {}; ccf::NodeInfoNetwork network; diff --git a/include/ccf/service/node_info.h b/include/ccf/service/node_info.h index ac913289aa3..a485ce2e422 100644 --- a/include/ccf/service/node_info.h +++ b/include/ccf/service/node_info.h @@ -72,6 +72,10 @@ namespace ccf * compatibility. */ bool retired_committed = false; + + /** Time when this Pending node last sent a join request, in milliseconds + * since the Unix epoch. Optional for backward compatibility. */ + std::optional pending_last_seen = std::nullopt; }; DECLARE_JSON_TYPE_WITH_BASE_AND_OPTIONAL_FIELDS(NodeInfo, NodeInfoNetwork); DECLARE_JSON_REQUIRED_FIELDS( @@ -84,7 +88,8 @@ namespace ccf certificate_signing_request, public_key, node_data, - retired_committed); + retired_committed, + pending_last_seen); } FMT_BEGIN_NAMESPACE diff --git a/src/common/configuration.h b/src/common/configuration.h index bbf9c53ce82..39c84eb4d35 100644 --- a/src/common/configuration.h +++ b/src/common/configuration.h @@ -145,6 +145,7 @@ namespace ccf attestation, snapshots, files_cleanup, + pending_node_timeout, node_to_node_message_limit, historical_cache_soft_limit, identity_history_fetch); diff --git a/src/node/node_state.h b/src/node/node_state.h index 8e27555ceb4..a8fa8bee4e0 100644 --- a/src/node/node_state.h +++ b/src/node/node_state.h @@ -44,6 +44,7 @@ #include "node/local_sealing.h" #include "node/node_inbound_message.h" #include "node/node_to_node_channel_manager.h" +#include "node/pending_node_cleanup.h" #include "node/recovery_decision_protocol.h" #include "node/recovery_snapshot_ledger.h" #include "node/signature_cache_subsystem.h" @@ -485,6 +486,7 @@ namespace ccf // JWT key auto-refresh // std::shared_ptr jwt_key_auto_refresh; + std::shared_ptr pending_node_cleanup; std::unique_ptr startup_snapshot_info = nullptr; // Set to the snapshot seqno when a node starts from one and remembered for @@ -3599,6 +3601,12 @@ namespace ccf commit_callbacks, public_only); + pending_node_cleanup = std::make_shared( + node_client, + consensus, + std::chrono::milliseconds(config.pending_node_timeout)); + pending_node_cleanup->start(); + network.tables->set_consensus(consensus); network.tables->set_snapshotter(snapshotter); diff --git a/src/node/pending_node_cleanup.h b/src/node/pending_node_cleanup.h new file mode 100644 index 00000000000..4d2917c50bc --- /dev/null +++ b/src/node/pending_node_cleanup.h @@ -0,0 +1,92 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the Apache 2.0 License. +#pragma once + +#include "kv/kv_types.h" +#include "node/node_client.h" +#include "tasks/basic_task.h" +#include "tasks/task_system.h" + +#include +#include +#include + +namespace ccf +{ + class PendingNodeCleanup + : public std::enable_shared_from_this + { + private: + static constexpr auto max_cleanup_interval = std::chrono::minutes(1); + + std::shared_ptr node_client; + std::shared_ptr consensus; + std::chrono::milliseconds cleanup_interval; + ccf::tasks::Task periodic_cleanup_task; + + void send_cleanup_request() + { + ::http::Request request( + fmt::format( + "/{}/{}", + ccf::get_actor_prefix(ccf::ActorsType::nodes), + "network/nodes/remove_expired_pending"), + HTTP_POST); + request.set_header(http::headers::CONTENT_LENGTH, "0"); + + node_client->make_request(request); + } + + public: + PendingNodeCleanup( + std::shared_ptr node_client_, + std::shared_ptr consensus_, + std::chrono::milliseconds pending_node_timeout) : + node_client(std::move(node_client_)), + consensus(std::move(consensus_)), + cleanup_interval(std::min( + pending_node_timeout, + std::chrono::duration_cast( + max_cleanup_interval))) + {} + + ~PendingNodeCleanup() + { + stop(); + } + + void start() + { + if (cleanup_interval <= std::chrono::milliseconds::zero()) + { + return; + } + + const auto self = weak_from_this(); + periodic_cleanup_task = ccf::tasks::make_basic_task([self]() { + const auto self_sp = self.lock(); + if (self_sp == nullptr) + { + return; + } + + if (self_sp->consensus->can_replicate()) + { + self_sp->send_cleanup_request(); + } + }); + + ccf::tasks::add_periodic_task( + periodic_cleanup_task, cleanup_interval, cleanup_interval); + } + + void stop() + { + if (periodic_cleanup_task != nullptr) + { + periodic_cleanup_task->cancel_task(); + periodic_cleanup_task = nullptr; + } + } + }; +} diff --git a/src/node/rpc/node_frontend.h b/src/node/rpc/node_frontend.h index 963cfb13111..6b6bc8e3eff 100644 --- a/src/node/rpc/node_frontend.h +++ b/src/node/rpc/node_frontend.h @@ -9,6 +9,7 @@ #include "ccf/http_query.h" #include "ccf/js/core/context.h" #include "ccf/json_handler.h" +#include "ccf/node/node_configuration_interface.h" #include "ccf/node/quote.h" #include "ccf/odata_error.h" #include "ccf/pal/attestation.h" @@ -36,6 +37,7 @@ #include "service/tables/snapshot_status.h" #include "snapshots/filenames.h" +#include #include #include @@ -267,6 +269,26 @@ namespace ccf return duplicate_node_id; } + static int64_t current_time_ms() + { + return std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()) + .count(); + } + + std::optional get_pending_node_timeout() + { + const auto node_configuration_subsystem = + this->context.get_subsystem(); + if (node_configuration_subsystem == nullptr) + { + return std::nullopt; + } + + return std::chrono::milliseconds( + node_configuration_subsystem->get().node_config.pending_node_timeout); + } + auto add_node( ccf::kv::Tx& tx, const std::vector& node_der, @@ -362,6 +384,11 @@ namespace ccf client_public_key_pem, in.node_data}; + if (node_status == NodeStatus::PENDING) + { + node_info.pending_last_seen = current_time_ms(); + } + nodes->put(joining_node_id, node_info); if (in.sealing_recovery_data.has_value()) @@ -489,6 +516,49 @@ namespace ccf payload); } + auto* current_consensus = get_consensus(); + const auto should_redirect_to_primary = + current_consensus != nullptr && !this->node_operation.can_replicate(); + auto redirect_to_primary = [&]() { + auto primary_id = current_consensus->primary(); + if (primary_id.has_value()) + { + const auto address = node::get_redirect_address_for_node( + args, args.tx, primary_id.value()); + if (!address.has_value()) + { + LOG_INFO_FMT( + "Join request rejected: no redirect address for " + "primary {}", + primary_id.value()); + return already_populated_response(); + } + + args.rpc_ctx->set_response_header( + http::headers::LOCATION, + fmt::format("https://{}/node/join", address.value())); + + const std::string payload = + "Node is not primary; cannot handle write"; + LOG_INFO_FMT( + "Join request redirected to primary {} at {}: {}", + primary_id.value(), + address.value(), + payload); + return make_error( + HTTP_STATUS_PERMANENT_REDIRECT, + ccf::errors::NodeCannotHandleRequest, + payload); + } + + const std::string payload = "Primary unknown"; + LOG_INFO_FMT("Join request rejected: {}", payload); + return make_error( + HTTP_STATUS_INTERNAL_SERVER_ERROR, + ccf::errors::InternalError, + payload); + }; + auto nodes = args.tx.ro(network.nodes); // If already joined => return equivalent response @@ -524,6 +594,38 @@ namespace ccf if (node_status == NodeStatus::PENDING) { + const auto pending_node_timeout = get_pending_node_timeout(); + if (!pending_node_timeout.has_value()) + { + return make_error( + HTTP_STATUS_INTERNAL_SERVER_ERROR, + ccf::errors::InternalError, + "NodeConfiguration subsystem is not available"); + } + + if ( + pending_node_timeout.value() > std::chrono::milliseconds::zero()) + { + const auto now = current_time_ms(); + const auto refresh_interval = + pending_node_timeout.value().count() / 2; + if ( + !node_info->pending_last_seen.has_value() || + node_info->pending_last_seen.value() < 0 || + node_info->pending_last_seen.value() > now || + now - node_info->pending_last_seen.value() >= refresh_interval) + { + if (should_redirect_to_primary) + { + return redirect_to_primary(); + } + + node_info->pending_last_seen = now; + args.tx.rw(network.nodes) + ->put(existing_node_info->node_id, node_info.value()); + } + } + // Only return node status and ID LOG_DEBUG_FMT( "Join request accepted: {} already marked as PENDING", @@ -539,47 +641,9 @@ namespace ccf } // Not the primary => Redirect if possible to primary - auto* current_consensus = get_consensus(); - if ( - current_consensus != nullptr && !this->node_operation.can_replicate()) + if (should_redirect_to_primary) { - auto primary_id = current_consensus->primary(); - if (primary_id.has_value()) - { - const auto address = node::get_redirect_address_for_node( - args, args.tx, primary_id.value()); - if (!address.has_value()) - { - LOG_INFO_FMT( - "Join request rejected: no redirect address for " - "primary {}", - primary_id.value()); - return already_populated_response(); - } - - args.rpc_ctx->set_response_header( - http::headers::LOCATION, - fmt::format("https://{}/node/join", address.value())); - - const std::string payload = - "Node is not primary; cannot handle write"; - LOG_INFO_FMT( - "Join request redirected to primary {} at {}: {}", - primary_id.value(), - address.value(), - payload); - return make_error( - HTTP_STATUS_PERMANENT_REDIRECT, - ccf::errors::NodeCannotHandleRequest, - "payload"); - } - - const std::string payload = "Primary unknown"; - LOG_INFO_FMT("Join request rejected: {}", payload); - return make_error( - HTTP_STATUS_INTERNAL_SERVER_ERROR, - ccf::errors::InternalError, - payload); + return redirect_to_primary(); } // Joiner's snapshot too old => StartupSeqnoIsOld @@ -665,6 +729,72 @@ namespace ccf .set_openapi_hidden(true) .install(); + auto remove_expired_pending = [this](auto& ctx, nlohmann::json&&) { + const auto pending_node_timeout = get_pending_node_timeout(); + if (!pending_node_timeout.has_value()) + { + return make_error( + HTTP_STATUS_INTERNAL_SERVER_ERROR, + ccf::errors::InternalError, + "NodeConfiguration subsystem is not available"); + } + + if (pending_node_timeout.value() <= std::chrono::milliseconds::zero()) + { + return make_success(true); + } + + const auto now = current_time_ms(); + auto nodes = ctx.tx.rw(network.nodes); + std::map pending_nodes_to_timestamp; + std::vector expired_pending_nodes; + nodes->foreach([&](const auto& node_id, const auto& node_info) { + if (node_info.status != NodeStatus::PENDING) + { + return true; + } + + if ( + !node_info.pending_last_seen.has_value() || + node_info.pending_last_seen.value() < 0 || + node_info.pending_last_seen.value() > now) + { + auto updated_node_info = node_info; + updated_node_info.pending_last_seen = now; + pending_nodes_to_timestamp.emplace( + node_id, std::move(updated_node_info)); + } + else if ( + now - node_info.pending_last_seen.value() >= + pending_node_timeout.value().count()) + { + expired_pending_nodes.push_back(node_id); + } + + return true; + }); + + for (const auto& [node_id, node_info] : pending_nodes_to_timestamp) + { + nodes->put(node_id, node_info); + } + + for (const auto& node_id : expired_pending_nodes) + { + LOG_INFO_FMT("Removing expired Pending node {}", node_id); + InternalTablesAccess::remove_node(ctx.tx, node_id); + } + + return make_success(true); + }; + make_endpoint( + "network/nodes/remove_expired_pending", + HTTP_POST, + json_adapter(remove_expired_pending), + {std::make_shared()}) + .set_openapi_hidden(true) + .install(); + auto set_retired_committed = [this](auto& ctx, nlohmann::json&&) { auto nodes = ctx.tx.rw(network.nodes); nodes->foreach([&nodes](const auto& node_id, auto node_info) { diff --git a/src/node/rpc/test/frontend_test.cpp b/src/node/rpc/test/frontend_test.cpp index f4035d791fd..49ee6443c9c 100644 --- a/src/node/rpc/test/frontend_test.cpp +++ b/src/node/rpc/test/frontend_test.cpp @@ -473,26 +473,6 @@ UserId user_id; MemberId member_id; MemberId invalid_member_id; -class TestNodeConfiguration : public NodeConfigurationInterface -{ -private: - StartupConfig config; - NodeConfigurationState state; - -public: - TestNodeConfiguration() : state{config, {}, true} - { - NodeInfoNetwork_v2::NetInterface interface; - interface.redirections = NodeInfoNetwork_v2::NetInterface::Redirections{}; - config.network.rpc_interfaces.emplace("test_interface", interface); - } - - const NodeConfigurationState& get() override - { - return state; - } -}; - class BlockingUserEndpointRegistry : public UserEndpointRegistry { std::latch& init_started; @@ -635,7 +615,10 @@ TEST_CASE("Redirect resolution handles unpublished consensus") NetworkState network; prepare_callers(network); TestUserFrontend frontend(*network.tables); - frontend.context.install_subsystem(std::make_shared()); + NodeInfoNetwork_v2::NetInterface interface; + interface.redirections = NodeInfoNetwork_v2::NetInterface::Redirections{}; + frontend.context.node_configuration->config.network.rpc_interfaces.emplace( + "test_interface", interface); const auto request = create_simple_request("/empty_function_no_auth"); const auto serialised_request = request.build_request(); diff --git a/src/node/rpc/test/node_frontend_test.cpp b/src/node/rpc/test/node_frontend_test.cpp index 46ec41e5d7a..a2ff2a0ba0c 100644 --- a/src/node/rpc/test/node_frontend_test.cpp +++ b/src/node/rpc/test/node_frontend_test.cpp @@ -35,8 +35,8 @@ TResponse frontend_process( r.set_body(body); auto serialise_request = r.build_request(); - auto session = - std::make_shared(ccf::InvalidSessionId, caller.raw()); + auto session = std::make_shared( + ccf::InvalidSessionId, ccf::crypto::cert_pem_to_der(caller)); auto rpc_ctx = ccf::make_rpc_context(session, serialise_request); frontend.process(rpc_ctx); @@ -375,6 +375,141 @@ TEST_CASE("Add a node to an open service") response.network_info->endorsed_certificate.value() == dummy_endorsed_certificate); } + + INFO("Expired Pending nodes are removed"); + { + CHECK( + std::chrono::milliseconds(StartupConfig{}.pending_node_timeout) == + std::chrono::hours(1)); + + ccf::crypto::ECKeyPairPtr expired_node_kp = ccf::crypto::make_ec_key_pair(); + const auto expired_node_caller = + expired_node_kp->self_sign("CN=Expired Joiner", valid_from, valid_to); + const auto expired_node_id = ccf::compute_node_id_from_kp(expired_node_kp); + + JoinNetworkNodeToNode::In expired_join_input; + expired_join_input.public_encryption_key = + ccf::crypto::make_ec_key_pair()->public_key_pem(); + expired_join_input.certificate_signing_request = + expired_node_kp->create_csr("CN=Expired Joiner"); + expired_join_input.node_info_network.node_to_node_interface + .published_address = "localhost:1234"; + + auto http_response = frontend_process( + frontend, expired_join_input, "join", expired_node_caller); + CHECK(http_response.status == HTTP_STATUS_OK); + + { + auto verify_tx = network.tables->create_tx(); + auto nodes = verify_tx.ro(network.nodes); + const auto node_info = nodes->get(expired_node_id); + REQUIRE(node_info.has_value()); + REQUIRE(node_info->pending_last_seen.has_value()); + + nlohmann::json node_info_json = node_info.value(); + CHECK( + node_info_json.get().pending_last_seen == + node_info->pending_last_seen); + node_info_json.erase("pending_last_seen"); + CHECK(!node_info_json.get().pending_last_seen.has_value()); + } + + { + auto age_tx = network.tables->create_tx(); + auto nodes = age_tx.rw(network.nodes); + auto node_info = nodes->get(expired_node_id); + REQUIRE(node_info.has_value()); + node_info->pending_last_seen.reset(); + nodes->put(expired_node_id, node_info.value()); + REQUIRE(age_tx.commit() == ccf::kv::CommitResult::SUCCESS); + } + + http_response = frontend_process( + frontend, + nullptr, + "network/nodes/remove_expired_pending", + expired_node_caller); + CHECK(http_response.status == HTTP_STATUS_OK); + { + auto verify_tx = network.tables->create_tx(); + const auto node_info = verify_tx.ro(network.nodes)->get(expired_node_id); + REQUIRE(node_info.has_value()); + CHECK(node_info->pending_last_seen.has_value()); + } + + const auto stale_pending_last_seen = + std::chrono::duration_cast( + (std::chrono::system_clock::now() - std::chrono::minutes(31)) + .time_since_epoch()) + .count(); + { + auto age_tx = network.tables->create_tx(); + auto nodes = age_tx.rw(network.nodes); + auto node_info = nodes->get(expired_node_id); + REQUIRE(node_info.has_value()); + node_info->pending_last_seen = stale_pending_last_seen; + nodes->put(expired_node_id, node_info.value()); + REQUIRE(age_tx.commit() == ccf::kv::CommitResult::SUCCESS); + } + + auto consensus = std::make_shared(); + consensus->state = ccf::kv::test::StubConsensus::Backup; + frontend.set_consensus_and_history(consensus.get(), nullptr); + context.node_operation->can_replicate_result = false; + + http_response = frontend_process( + frontend, expired_join_input, "join", expired_node_caller); + CHECK(http_response.status != HTTP_STATUS_OK); + { + auto verify_tx = network.tables->create_tx(); + const auto node_info = verify_tx.ro(network.nodes)->get(expired_node_id); + REQUIRE(node_info.has_value()); + CHECK(node_info->pending_last_seen == stale_pending_last_seen); + } + + consensus->state = ccf::kv::test::StubConsensus::Primary; + context.node_operation->can_replicate_result = true; + http_response = frontend_process( + frontend, expired_join_input, "join", expired_node_caller); + CHECK(http_response.status == HTTP_STATUS_OK); + { + auto verify_tx = network.tables->create_tx(); + const auto node_info = verify_tx.ro(network.nodes)->get(expired_node_id); + REQUIRE(node_info.has_value()); + REQUIRE(node_info->pending_last_seen.has_value()); + CHECK( + node_info->pending_last_seen.value() > + std::chrono::duration_cast( + (std::chrono::system_clock::now() - std::chrono::minutes(1)) + .time_since_epoch()) + .count()); + } + + { + auto age_tx = network.tables->create_tx(); + auto nodes = age_tx.rw(network.nodes); + auto node_info = nodes->get(expired_node_id); + REQUIRE(node_info.has_value()); + node_info->pending_last_seen = + std::chrono::duration_cast( + (std::chrono::system_clock::now() - std::chrono::hours(2)) + .time_since_epoch()) + .count(); + nodes->put(expired_node_id, node_info.value()); + REQUIRE(age_tx.commit() == ccf::kv::CommitResult::SUCCESS); + } + + http_response = frontend_process( + frontend, + nullptr, + "network/nodes/remove_expired_pending", + expired_node_caller); + CHECK(http_response.status == HTTP_STATUS_OK); + { + auto verify_tx = network.tables->create_tx(); + CHECK(!verify_tx.ro(network.nodes)->has(expired_node_id)); + } + } } int main(int argc, char** argv) diff --git a/src/node/rpc/test/node_stub.h b/src/node/rpc/test/node_stub.h index 48ffe018e68..3655e5fc835 100644 --- a/src/node/rpc/test/node_stub.h +++ b/src/node/rpc/test/node_stub.h @@ -3,6 +3,7 @@ #pragma once #include "ccf/historical_queries_interface.h" +#include "ccf/node/node_configuration_interface.h" #include "kv/test/stub_consensus.h" #include "node/recovery_decision_protocol.h" #include "node/rpc/gov_effects_interface.h" @@ -17,6 +18,7 @@ namespace ccf { public: bool is_public = false; + bool can_replicate_result = true; ccf::COSESignaturesConfig cose_signatures_config = {}; ExtendedState state() override @@ -66,7 +68,7 @@ namespace ccf bool can_replicate() override { - return true; + return can_replicate_result; } std::optional get_primary() override @@ -277,12 +279,25 @@ namespace ccf } }; + class StubNodeConfiguration : public ccf::NodeConfigurationInterface + { + public: + ccf::StartupConfig config = {}; + ccf::NodeConfigurationState state = {config, {}, true}; + + const ccf::NodeConfigurationState& get() override + { + return state; + } + }; + struct StubNodeContext : public ccf::AbstractNodeContext { public: std::shared_ptr node_operation = nullptr; std::shared_ptr gov_effects = nullptr; std::shared_ptr cache = nullptr; + std::shared_ptr node_configuration = nullptr; StubNodeContext() { @@ -294,6 +309,9 @@ namespace ccf cache = std::make_shared(); install_subsystem(cache); + + node_configuration = std::make_shared(); + install_subsystem(node_configuration); } ccf::NodeId get_node_id() const override diff --git a/tests/config.jinja b/tests/config.jinja index 7de8e0084f7..d4bba84898a 100644 --- a/tests/config.jinja +++ b/tests/config.jinja @@ -74,7 +74,9 @@ "target_rpc_interface": "{{ backup_snapshot_fetch_target_rpc_interface }}"{% endif %}{% if backup_snapshot_fetch_max_size %}, "max_size": "{{ backup_snapshot_fetch_max_size }}"{% endif %} }{% endif %} - },{% if files_cleanup_max_snapshots or files_cleanup_max_committed_ledger_chunks or files_cleanup_interval %} + },{% if pending_node_timeout %} + "pending_node_timeout": "{{ pending_node_timeout }}", + {% endif %}{% if files_cleanup_max_snapshots or files_cleanup_max_committed_ledger_chunks or files_cleanup_interval %} "files_cleanup": { {% if files_cleanup_max_snapshots %}"max_snapshots": {{ files_cleanup_max_snapshots }}{% endif %}{% if files_cleanup_max_snapshots and (files_cleanup_max_committed_ledger_chunks or files_cleanup_interval) %}, diff --git a/tests/e2e_operations.py b/tests/e2e_operations.py index e5257f3af49..17d2664c301 100644 --- a/tests/e2e_operations.py +++ b/tests/e2e_operations.py @@ -4801,6 +4801,57 @@ def run_ledger_chunk_cleanup_tests(const_args): test_ledger_chunk_cleanup_digest_mismatch(network, args) +@reqs.description("Pending node entries expire after the configured timeout") +def test_pending_node_expiration(network, args): + primary, _ = network.find_primary() + pending_node = network.create_node() + network.join_node( + pending_node, + args.package, + args, + target_node=primary, + from_snapshot=False, + ) + + with primary.client() as c: + r = c.get(f"/node/network/nodes/{pending_node.node_id}") + assert r.status_code == http.HTTPStatus.OK, r + assert r.body.json()["status"] == "Pending", r.body.json() + + pending_node.stop() + + end_time = time.time() + 30 + with primary.client() as c: + while time.time() < end_time: + r = c.get(f"/node/network/nodes/{pending_node.node_id}") + if r.status_code == http.HTTPStatus.NOT_FOUND: + break + assert r.status_code == http.HTTPStatus.OK, r + assert r.body.json()["status"] == "Pending", r.body.json() + time.sleep(0.1) + else: + raise TimeoutError(f"Pending node {pending_node.node_id} was not removed") + + return network + + +def run_pending_node_expiration(const_args): + args = copy.deepcopy(const_args) + args.label += "_pending_node_expiration" + args.nodes = infra.e2e_args.min_nodes(args, f=0) + args.pending_node_timeout = "10s" + + with infra.network.network( + args.nodes, + args.binary_dir, + args.debug_nodes, + pdb=args.pdb, + txs=app.LoggingTxs("user0"), + ) as network: + network.start_and_open(args) + test_pending_node_expiration(network, args) + + # The operations tests below are split into groups which are run # concurrently, as separate ConcurrentRunner sub-tests (see tests/schema.py). # Each group runs its own tests sequentially, so tests which share a workspace @@ -4835,6 +4886,7 @@ def run_node_config_checks(args): run_tls_san_checks(args) run_tls_san_join_mismatch(args) run_config_timeout_check(args) + run_pending_node_expiration(args) run_configuration_file_checks(args) run_pid_file_check(args) run_preopen_readiness_check(args) diff --git a/tests/infra/network.py b/tests/infra/network.py index b9c481aa30d..cd80439aec7 100644 --- a/tests/infra/network.py +++ b/tests/infra/network.py @@ -191,6 +191,7 @@ class Network: "log_format_json", "constitution", "join_timer_s", + "pending_node_timeout", "worker_threads", "ledger_chunk_bytes", "ledger_max_transaction_bytes", diff --git a/tests/infra/remote.py b/tests/infra/remote.py index 31522f11a2c..ab075880ea2 100644 --- a/tests/infra/remote.py +++ b/tests/infra/remote.py @@ -405,6 +405,7 @@ def __init__( node_address=None, config_file=None, join_timer_s=None, + pending_node_timeout=None, sig_ms_interval=None, jwt_key_refresh_interval_s=None, jwt_key_refresh_max_response_size="1MB", @@ -611,6 +612,7 @@ def __init__( curve_id=curve_id.name.title(), host_log_level=log_level.title(), join_timer=f"{join_timer_s}s" if join_timer_s else None, + pending_node_timeout=pending_node_timeout, signature_interval_duration=f"{sig_ms_interval}ms", jwt_key_refresh_interval=f"{jwt_key_refresh_interval_s}s", jwt_key_refresh_max_response_size=jwt_key_refresh_max_response_size,