From a8bfc17dd718ff5ff2b7256be96e28e83a842d64 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 18 Aug 2026 23:43:28 +0000 Subject: [PATCH 1/5] Expire stale pending node entries Co-authored-by: achamayou <4016369+achamayou@users.noreply.github.com> Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 1 + doc/host_config_schema/host_config.json | 5 ++ doc/operations/start_network.rst | 2 + include/ccf/node/startup_config.h | 2 + include/ccf/service/node_info.h | 7 +- src/common/configuration.h | 1 + src/node/node_state.h | 8 +++ src/node/pending_node_cleanup.h | 92 ++++++++++++++++++++++++ src/node/rpc/node_frontend.h | 83 +++++++++++++++++++++ src/node/rpc/test/node_frontend_test.cpp | 87 +++++++++++++++++++++- src/node/rpc/test/node_stub.h | 17 +++++ tests/config.jinja | 4 +- tests/e2e_operations.py | 52 ++++++++++++++ tests/infra/network.py | 1 + tests/infra/remote.py | 2 + tests/nodes.py | 7 ++ 16 files changed, 367 insertions(+), 4 deletions(-) create mode 100644 src/node/pending_node_cleanup.h diff --git a/CHANGELOG.md b/CHANGELOG.md index c5efa4fb2639..a1d72949b791 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Added - C++ endpoints can now use `ccf::endpoints::Endpoint::add_openapi_response()` to document additional HTTP responses in their generated OpenAPI schema without changing the endpoint's primary success response (#8115). +- Pending node entries are now automatically removed after the configurable `pending_node_timeout` delay (1 hour by default). Set it to `0s` to disable automatic removal (#8173). ### Changed diff --git a/doc/host_config_schema/host_config.json b/doc/host_config_schema/host_config.json index 8183c55d664a..6c55e36ef2be 100644 --- a/doc/host_config_schema/host_config.json +++ b/doc/host_config_schema/host_config.json @@ -570,6 +570,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 node that remains Pending. Cleanup runs at least once per minute, so removal may occur up to one minute after this delay. A still-running joining node may register itself again on its next retry. 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 022a5b298bd4..1aeb1328c4fd 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 node entry that remains ``Pending`` for longer than the ``pending_node_timeout`` configuration entry (1 hour by default). Cleanup runs at least once per minute. A joining node that is still running will register itself again on its next retry. 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 3d6cef413798..b78224ecceef 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 ac913289aa35..e82c0b8e47c7 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 node was added as Pending, in milliseconds since the Unix + * epoch. Optional for backward compatibility. */ + std::optional pending_since = 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_since); } FMT_BEGIN_NAMESPACE diff --git a/src/common/configuration.h b/src/common/configuration.h index d11b775b26e6..fb731213b341 100644 --- a/src/common/configuration.h +++ b/src/common/configuration.h @@ -141,6 +141,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 eb4f64e7a911..889646bf03ad 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 000000000000..4d2917c50bcc --- /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 2f90dc0e26a3..df7ff729f1b6 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" @@ -35,6 +36,7 @@ #include "service/tables/snapshot_status.h" #include "snapshots/filenames.h" +#include #include #include @@ -266,6 +268,13 @@ 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(); + } + auto add_node( ccf::kv::Tx& tx, const std::vector& node_der, @@ -361,6 +370,11 @@ namespace ccf client_public_key_pem, in.node_data}; + if (node_status == NodeStatus::PENDING) + { + node_info.pending_since = current_time_ms(); + } + nodes->put(joining_node_id, node_info); if (in.sealing_recovery_data.has_value()) @@ -660,6 +674,75 @@ namespace ccf .set_openapi_hidden(true) .install(); + auto remove_expired_pending = [this](auto& ctx, nlohmann::json&&) { + const auto node_configuration_subsystem = + this->context.get_subsystem(); + if (node_configuration_subsystem == nullptr) + { + return make_error( + HTTP_STATUS_INTERNAL_SERVER_ERROR, + ccf::errors::InternalError, + "NodeConfiguration subsystem is not available"); + } + + const auto pending_node_timeout = std::chrono::milliseconds( + node_configuration_subsystem->get().node_config.pending_node_timeout); + if (pending_node_timeout <= 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_since.has_value() || + node_info.pending_since.value() < 0 || + node_info.pending_since.value() > now) + { + auto updated_node_info = node_info; + updated_node_info.pending_since = now; + pending_nodes_to_timestamp.emplace( + node_id, std::move(updated_node_info)); + } + else if ( + now - node_info.pending_since.value() >= + pending_node_timeout.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/node_frontend_test.cpp b/src/node/rpc/test/node_frontend_test.cpp index 3ff019796af9..ebd0edc3f8b9 100644 --- a/src/node/rpc/test/node_frontend_test.cpp +++ b/src/node/rpc/test/node_frontend_test.cpp @@ -30,8 +30,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); @@ -313,6 +313,89 @@ TEST_CASE("Add a node to an open service") response.network_info->endorsed_certificate.value() == dummy_endorsed_certificate); } + + INFO("Expired Pending nodes are removed"); + { + 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_since.has_value()); + + nlohmann::json node_info_json = node_info.value(); + CHECK( + node_info_json.get().pending_since == + node_info->pending_since); + node_info_json.erase("pending_since"); + CHECK(!node_info_json.get().pending_since.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_since.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_since.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_since = + std::chrono::duration_cast( + (std::chrono::system_clock::now() - std::chrono::hours(25)) + .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 48ffe018e681..16b912b7e20b 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" @@ -277,12 +278,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 +308,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 b6a0bf18b51d..9b4f9612cb2e 100644 --- a/tests/config.jinja +++ b/tests/config.jinja @@ -73,7 +73,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 6edf7dcb9c9f..9d59c81d0da6 100644 --- a/tests/e2e_operations.py +++ b/tests/e2e_operations.py @@ -4782,6 +4782,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() + 10 + 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 = "2s" + + 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) + + def run(args): run_ledger_viz_test(args) run_split_ledger_test(args) @@ -4798,6 +4849,7 @@ def run(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 a4b7b5befbb8..dc816303dc71 100644 --- a/tests/infra/network.py +++ b/tests/infra/network.py @@ -192,6 +192,7 @@ class Network: "log_format_json", "constitution", "join_timer_s", + "pending_node_timeout", "worker_threads", "ledger_chunk_bytes", "subject_alt_names", diff --git a/tests/infra/remote.py b/tests/infra/remote.py index 49278bb682b4..26e0b1d9c7e4 100644 --- a/tests/infra/remote.py +++ b/tests/infra/remote.py @@ -327,6 +327,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", @@ -533,6 +534,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, diff --git a/tests/nodes.py b/tests/nodes.py index 966fd17c048d..b5794ac8af8d 100644 --- a/tests/nodes.py +++ b/tests/nodes.py @@ -6,6 +6,7 @@ import time import committable +import e2e_operations import infra.checker import infra.e2e_args import infra.network @@ -285,6 +286,12 @@ def add(parser): cr = ConcurrentRunner(add) args = copy.deepcopy(cr.args) + cr.add( + "pending_node_expiration", + e2e_operations.run_pending_node_expiration, + package="samples/apps/logging/logging", + ) + cr.add( "rotation", run_rotations, From 93297444c14cf85230098a7e7818209c72a69a08 Mon Sep 17 00:00:00 2001 From: Amaury Chamayou Date: Mon, 24 Aug 2026 15:33:38 +0100 Subject: [PATCH 2/5] Test one-hour pending node expiry Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/node/rpc/test/node_frontend_test.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/node/rpc/test/node_frontend_test.cpp b/src/node/rpc/test/node_frontend_test.cpp index ebd0edc3f8b9..4f994b4d371e 100644 --- a/src/node/rpc/test/node_frontend_test.cpp +++ b/src/node/rpc/test/node_frontend_test.cpp @@ -378,7 +378,7 @@ TEST_CASE("Add a node to an open service") REQUIRE(node_info.has_value()); node_info->pending_since = std::chrono::duration_cast( - (std::chrono::system_clock::now() - std::chrono::hours(25)) + (std::chrono::system_clock::now() - std::chrono::hours(2)) .time_since_epoch()) .count(); nodes->put(expired_node_id, node_info.value()); From f8929e79576ca6effbf30e9f07015799eb1e9ad0 Mon Sep 17 00:00:00 2001 From: Amaury Chamayou Date: Mon, 24 Aug 2026 16:05:01 +0100 Subject: [PATCH 3/5] Address pending node cleanup review Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/node/rpc/test/frontend_test.cpp | 25 ++++-------------------- src/node/rpc/test/node_frontend_test.cpp | 4 ++++ tests/nodes.py | 7 ------- 3 files changed, 8 insertions(+), 28 deletions(-) diff --git a/src/node/rpc/test/frontend_test.cpp b/src/node/rpc/test/frontend_test.cpp index e1fe0365593e..0f8815fa0ad4 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 4f994b4d371e..855ba5a6aac4 100644 --- a/src/node/rpc/test/node_frontend_test.cpp +++ b/src/node/rpc/test/node_frontend_test.cpp @@ -316,6 +316,10 @@ TEST_CASE("Add a node to an open service") 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); diff --git a/tests/nodes.py b/tests/nodes.py index b5794ac8af8d..966fd17c048d 100644 --- a/tests/nodes.py +++ b/tests/nodes.py @@ -6,7 +6,6 @@ import time import committable -import e2e_operations import infra.checker import infra.e2e_args import infra.network @@ -286,12 +285,6 @@ def add(parser): cr = ConcurrentRunner(add) args = copy.deepcopy(cr.args) - cr.add( - "pending_node_expiration", - e2e_operations.run_pending_node_expiration, - package="samples/apps/logging/logging", - ) - cr.add( "rotation", run_rotations, From e63e7853679cafc9143f1a42830a5034b7e86418 Mon Sep 17 00:00:00 2001 From: Amaury Chamayou Date: Wed, 26 Aug 2026 14:39:36 +0100 Subject: [PATCH 4/5] Keep active pending nodes alive Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 2 +- doc/host_config_schema/host_config.json | 2 +- doc/operations/start_network.rst | 2 +- include/ccf/service/node_info.h | 8 +- src/node/rpc/node_frontend.h | 153 +++++++++++++++-------- src/node/rpc/test/node_frontend_test.cpp | 64 ++++++++-- src/node/rpc/test/node_stub.h | 3 +- tests/e2e_operations.py | 4 +- 8 files changed, 167 insertions(+), 71 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a1d72949b791..1833d70d9df0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,7 +16,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Added - C++ endpoints can now use `ccf::endpoints::Endpoint::add_openapi_response()` to document additional HTTP responses in their generated OpenAPI schema without changing the endpoint's primary success response (#8115). -- Pending node entries are now automatically removed after the configurable `pending_node_timeout` delay (1 hour by default). Set it to `0s` to disable automatic removal (#8173). +- 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). ### Changed diff --git a/doc/host_config_schema/host_config.json b/doc/host_config_schema/host_config.json index 6c55e36ef2be..1bbab57cd3f9 100644 --- a/doc/host_config_schema/host_config.json +++ b/doc/host_config_schema/host_config.json @@ -573,7 +573,7 @@ "pending_node_timeout": { "type": "string", "default": "1h", - "description": "Delay after which the primary automatically removes a node that remains Pending. Cleanup runs at least once per minute, so removal may occur up to one minute after this delay. A still-running joining node may register itself again on its next retry. Set to 0s to disable automatic removal." + "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", diff --git a/doc/operations/start_network.rst b/doc/operations/start_network.rst index 1aeb1328c4fd..4f5cea1b403f 100644 --- a/doc/operations/start_network.rst +++ b/doc/operations/start_network.rst @@ -62,7 +62,7 @@ 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 node entry that remains ``Pending`` for longer than the ``pending_node_timeout`` configuration entry (1 hour by default). Cleanup runs at least once per minute. A joining node that is still running will register itself again on its next retry. Set ``pending_node_timeout`` to ``0s`` to disable automatic removal. +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. diff --git a/include/ccf/service/node_info.h b/include/ccf/service/node_info.h index e82c0b8e47c7..a485ce2e422a 100644 --- a/include/ccf/service/node_info.h +++ b/include/ccf/service/node_info.h @@ -73,9 +73,9 @@ namespace ccf */ bool retired_committed = false; - /** Time when this node was added as Pending, in milliseconds since the Unix - * epoch. Optional for backward compatibility. */ - std::optional pending_since = std::nullopt; + /** 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( @@ -89,7 +89,7 @@ namespace ccf public_key, node_data, retired_committed, - pending_since); + pending_last_seen); } FMT_BEGIN_NAMESPACE diff --git a/src/node/rpc/node_frontend.h b/src/node/rpc/node_frontend.h index df7ff729f1b6..848a8203dc1f 100644 --- a/src/node/rpc/node_frontend.h +++ b/src/node/rpc/node_frontend.h @@ -275,6 +275,19 @@ namespace ccf .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, @@ -372,7 +385,7 @@ namespace ccf if (node_status == NodeStatus::PENDING) { - node_info.pending_since = current_time_ms(); + node_info.pending_last_seen = current_time_ms(); } nodes->put(joining_node_id, node_info); @@ -498,6 +511,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 @@ -533,6 +589,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", @@ -548,47 +636,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 @@ -675,9 +725,8 @@ namespace ccf .install(); auto remove_expired_pending = [this](auto& ctx, nlohmann::json&&) { - const auto node_configuration_subsystem = - this->context.get_subsystem(); - if (node_configuration_subsystem == nullptr) + const auto pending_node_timeout = get_pending_node_timeout(); + if (!pending_node_timeout.has_value()) { return make_error( HTTP_STATUS_INTERNAL_SERVER_ERROR, @@ -685,9 +734,7 @@ namespace ccf "NodeConfiguration subsystem is not available"); } - const auto pending_node_timeout = std::chrono::milliseconds( - node_configuration_subsystem->get().node_config.pending_node_timeout); - if (pending_node_timeout <= std::chrono::milliseconds::zero()) + if (pending_node_timeout.value() <= std::chrono::milliseconds::zero()) { return make_success(true); } @@ -703,18 +750,18 @@ namespace ccf } if ( - !node_info.pending_since.has_value() || - node_info.pending_since.value() < 0 || - node_info.pending_since.value() > now) + !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_since = now; + 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_since.value() >= - pending_node_timeout.count()) + now - node_info.pending_last_seen.value() >= + pending_node_timeout.value().count()) { expired_pending_nodes.push_back(node_id); } diff --git a/src/node/rpc/test/node_frontend_test.cpp b/src/node/rpc/test/node_frontend_test.cpp index 855ba5a6aac4..53dad41a4b24 100644 --- a/src/node/rpc/test/node_frontend_test.cpp +++ b/src/node/rpc/test/node_frontend_test.cpp @@ -342,14 +342,14 @@ TEST_CASE("Add a node to an open service") 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_since.has_value()); + REQUIRE(node_info->pending_last_seen.has_value()); nlohmann::json node_info_json = node_info.value(); CHECK( - node_info_json.get().pending_since == - node_info->pending_since); - node_info_json.erase("pending_since"); - CHECK(!node_info_json.get().pending_since.has_value()); + 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()); } { @@ -357,7 +357,7 @@ TEST_CASE("Add a node to an open service") auto nodes = age_tx.rw(network.nodes); auto node_info = nodes->get(expired_node_id); REQUIRE(node_info.has_value()); - node_info->pending_since.reset(); + node_info->pending_last_seen.reset(); nodes->put(expired_node_id, node_info.value()); REQUIRE(age_tx.commit() == ccf::kv::CommitResult::SUCCESS); } @@ -372,15 +372,63 @@ TEST_CASE("Add a node to an open service") 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_since.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_since = + 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()) diff --git a/src/node/rpc/test/node_stub.h b/src/node/rpc/test/node_stub.h index 16b912b7e20b..3655e5fc8358 100644 --- a/src/node/rpc/test/node_stub.h +++ b/src/node/rpc/test/node_stub.h @@ -18,6 +18,7 @@ namespace ccf { public: bool is_public = false; + bool can_replicate_result = true; ccf::COSESignaturesConfig cose_signatures_config = {}; ExtendedState state() override @@ -67,7 +68,7 @@ namespace ccf bool can_replicate() override { - return true; + return can_replicate_result; } std::optional get_primary() override diff --git a/tests/e2e_operations.py b/tests/e2e_operations.py index 9d59c81d0da6..720614fc41aa 100644 --- a/tests/e2e_operations.py +++ b/tests/e2e_operations.py @@ -4801,7 +4801,7 @@ def test_pending_node_expiration(network, args): pending_node.stop() - end_time = time.time() + 10 + 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}") @@ -4820,7 +4820,7 @@ 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 = "2s" + args.pending_node_timeout = "10s" with infra.network.network( args.nodes, From 359c9aa59c57dce25eb686be688402e04bc2b50e Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 3 Sep 2026 14:16:19 +0000 Subject: [PATCH 5/5] Move pending node timeout entry to 7.0.14 Co-authored-by: achamayou <4016369+achamayou@users.noreply.github.com> --- CHANGELOG.md | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c09c8fd72ca3..aef89a0541d6 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). @@ -33,7 +37,6 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Added - C++ endpoints can now use `ccf::endpoints::Endpoint::add_openapi_response()` to document additional HTTP responses in their generated OpenAPI schema without changing the endpoint's primary success response (#8115). -- 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). - New `ledger.max_transaction_size` node configuration option (default `32MB`), which caps the total serialised size of transactions written to the ledger. The limit covers the whole ledger entry: the fixed 8-byte ledger entry header, the ledger encryption header, public domain size field, public domain and encrypted private domain. It is checked before a transaction is applied, so an oversized transaction is now rejected with `413 Payload Too Large` and error code `TransactionTooLarge`, and subsequent transactions are unaffected, where previously an excessively large transaction could terminate the node. Reserved internal signature transactions are exempt because they must fill their reserved ledger version. The limit applies only to newly serialised non-reserved transactions; deserialising existing entries (including during recovery), historical queries and snapshots are unaffected, so entries written under a larger or unset limit remain readable. It must be smaller than `memory.max_msg_size` by at least the ring-buffer range response overhead, which is validated at node startup and by `--check` (#7992). ### Changed