diff --git a/include/session/network/routing/onion_request_router.hpp b/include/session/network/routing/onion_request_router.hpp index 1dbbbcee3..da17f3c40 100644 --- a/include/session/network/routing/onion_request_router.hpp +++ b/include/session/network/routing/onion_request_router.hpp @@ -168,6 +168,7 @@ class OnionRequestRouter : public IRouter, public std::enable_shared_from_this { virtual void record_node_failure(const ed25519_pubkey& key, bool permanent = false); uint16_t node_strike_count(const service_node& node); uint16_t node_strike_count(const ed25519_pubkey& key); + + // Whether the node has collected enough unexpired strikes to be kept out of node selection. + // Callers that pick a node by some other route - a cached one, say - need this to apply the + // same bar `get_unused_nodes` does, rather than comparing a raw count to a threshold they'd + // have to know about. + virtual bool node_struck_out(const service_node& node); + virtual bool node_struck_out(const ed25519_pubkey& key); void clear_node_strikes(); // Checks if the pool is empty or stale and triggers a refresh if needed @@ -115,6 +122,11 @@ class SnodePool : public std::enable_shared_from_this { std::vector> _snode_refresh_results; std::vector> _after_snode_cache_refresh; + // Counts a node's strikes that haven't expired yet. `record_node_failure` only appends, so the + // raw vector answers a different question - every strike the node has ever collected. + size_t _active_strike_count(const ed25519_pubkey& key) const; + bool _node_struck_out(const ed25519_pubkey& key) const; + // Disk I/O functions void _load_from_disk(); static void _clear_disk_cache(const std::filesystem::path& path); diff --git a/src/network/routing/onion_request_router.cpp b/src/network/routing/onion_request_router.cpp index 99f6c4549..9915dd453 100644 --- a/src/network/routing/onion_request_router.cpp +++ b/src/network/routing/onion_request_router.cpp @@ -595,9 +595,37 @@ void OnionRequestRouter::_finish_setup() { } } +// A cached edge node is handed to `_build_path` as a forced first hop, so it is the one node in a +// path that never passes the strike filter `get_unused_nodes` applies to the rest, and the only +// thing that dropped it was `edge_node_cache_duration` (10 days). +// +// Being struck out costs it the cached-edge-node role, not its place in the pool: it stays a node +// like any other and `get_unused_nodes` can pick it again once its strikes expire. Keeping the +// entry and merely skipping it would hand the role back at that expiry, by which point we have been +// running on a different edge node for two days - that is a second change of first hop, not a +// return to a stable one. +void OnionRequestRouter::_drop_struck_cached_edge_nodes() { + auto snode_pool = _snode_pool.lock(); + + if (!snode_pool) + return; + + std::erase_if(_cached_edge_nodes, [&snode_pool](const auto& cached) { + if (!snode_pool->node_struck_out(cached.node)) + return false; + + log::debug( + cat, + "Dropping cached edge node {}, it has been struck out.", + cached.node.to_string()); + return true; + }); +} + void OnionRequestRouter::_pre_build_paths_if_needed() { if (!_config.disable_pre_build_paths) { log::info(cat, "Pre-building initial paths."); + _drop_struck_cached_edge_nodes(); std::vector edge_nodes = _cached_edge_nodes; if (_config.single_path_mode) { @@ -2128,13 +2156,14 @@ void OnionRequestRouter::_rotate_path(const std::string& path_id, PathCategory c } // Get enough nodes for the path (if the edge node has been used for longer than the cache - // duration then we should create an entirely new path, otherwise we should try to reuse the - // edge node) + // duration, or has been struck out since we connected to it, then we should create an entirely + // new path, otherwise we should try to reuse the edge node) auto now = std::chrono::system_clock::now(); auto rotate_at = (std::chrono::steady_clock::now() + _config.path_rotation_frequency); std::vector rotated_path_nodes; - if (now > path.edge_first_connected_at + _config.edge_node_cache_duration) + if (now > path.edge_first_connected_at + _config.edge_node_cache_duration || + snode_pool->node_struck_out(edge_node)) rotated_path_nodes = snode_pool->get_unused_nodes(_config.path_length, nodes_to_exclude); else { rotated_path_nodes = diff --git a/src/network/snode_pool.cpp b/src/network/snode_pool.cpp index d4f61ad9a..9643078a8 100644 --- a/src/network/snode_pool.cpp +++ b/src/network/snode_pool.cpp @@ -963,6 +963,25 @@ void SnodePool::clear_cache() { }); } +size_t SnodePool::_active_strike_count(const ed25519_pubkey& key) const { + auto it = _snode_strikes.find(key); + + if (it == _snode_strikes.end()) + return 0; + + auto threshold = sysclock_now_s() - STRIKE_EXPIRY; + + return std::ranges::count_if(it->second, [threshold](auto t) { return t > threshold; }); +} + +bool SnodePool::_node_struck_out(const ed25519_pubkey& key) const { + auto strikes = _active_strike_count(key); + + // The `strikes > 0` is what keeps a threshold of 0 meaning "drop a node on its first strike"; + // comparing straight against 0 also drops every node that has never failed at all + return (strikes > 0 && strikes >= _config.cache_node_strike_threshold); +} + void SnodePool::record_node_failure(const service_node& node, bool permanent) { record_node_failure(node.remote_pubkey, permanent); } @@ -970,18 +989,17 @@ void SnodePool::record_node_failure(const service_node& node, bool permanent) { void SnodePool::record_node_failure(const ed25519_pubkey& key, bool permanent) { _loop->call([this, key, permanent] { auto now = sysclock_now_s(); + auto& stamps = _snode_strikes[key]; + std::erase_if(stamps, [threshold = now - STRIKE_EXPIRY](auto t) { return t <= threshold; }); - if (permanent) - for (int i = 0; i < _config.cache_node_strike_threshold; ++i) - _snode_strikes[key].push_back(now); - else - _snode_strikes[key].push_back(now); + // A permanent failure has to strike the node out whatever the threshold is - looping up to + // a threshold of 0 records nothing and leaves the node in rotation + auto strikes = (permanent ? std::max(1, _config.cache_node_strike_threshold) : 1); - log::trace( - cat, - "Recorded strike for node {}, total: {}", - key.hex(), - _snode_strikes[key].size()); + for (uint16_t i = 0; i < strikes; ++i) + stamps.push_back(now); + + log::trace(cat, "Recorded strike for node {}, total: {}", key.hex(), stamps.size()); // Throttle persisting the strikes to disk to at most every X minutes if (!_strikes_flush_scheduled && !_suspended) { @@ -1009,22 +1027,16 @@ uint16_t SnodePool::node_strike_count(const service_node& node) { } uint16_t SnodePool::node_strike_count(const ed25519_pubkey& key) { - return _loop->call_get([this, &key] { - auto it = _snode_strikes.find(key); - if (it == _snode_strikes.end()) - return uint16_t{0}; - - const auto& stamps = it->second; - - const auto threshold = sysclock_now_s() - STRIKE_EXPIRY; + return _loop->call_get( + [this, &key] { return static_cast(_active_strike_count(key)); }); +} - uint16_t count = 0; - for (auto t : stamps) - if (t > threshold) - count++; +bool SnodePool::node_struck_out(const service_node& node) { + return node_struck_out(node.remote_pubkey); +} - return count; - }); +bool SnodePool::node_struck_out(const ed25519_pubkey& key) { + return _loop->call_get([this, &key] { return _node_struck_out(key); }); } void SnodePool::clear_node_strikes() { @@ -1068,9 +1080,7 @@ void SnodePool::refresh_if_needed( in_use_keys.insert(node.remote_pubkey); for (const auto& node : _snode_cache) { - auto it = _snode_strikes.find(node.remote_pubkey); - if (it != _snode_strikes.end() && - it->second.size() >= _config.cache_node_strike_threshold) + if (_node_struck_out(node.remote_pubkey)) continue; // If the caller considers the node as already in use then it wouldn't be @@ -1158,9 +1168,7 @@ std::vector SnodePool::get_unused_nodes( continue; // Skip nodes with too many failures - auto it = _snode_strikes.find(node.remote_pubkey); - if (it != _snode_strikes.end() && - it->second.size() >= _config.cache_node_strike_threshold) + if (_node_struck_out(node.remote_pubkey)) continue; // Skip nodes whos IP addresses are in the exclusion list @@ -1196,8 +1204,7 @@ void SnodePool::get_swarm( std::ranges::shuffle(nodes, csrng); auto get_strike_count = [this](const service_node& node) -> size_t { - auto it = _snode_strikes.find(node.remote_pubkey); - return (it != _snode_strikes.end() ? it->second.size() : 0); + return _active_strike_count(node.remote_pubkey); }; // Partition into below-threshold and above-thresold. This keeps the shuffled order of diff --git a/tests/test_onion_request_router.cpp b/tests/test_onion_request_router.cpp index 8913315d8..910d0f217 100644 --- a/tests/test_onion_request_router.cpp +++ b/tests/test_onion_request_router.cpp @@ -50,6 +50,20 @@ class TestOnionRequestRouter { return 0; } + static void set_cached_edge_nodes( + std::shared_ptr router, std::vector nodes) { + router->_cached_edge_nodes = std::move(nodes); + } + + static std::vector cached_edge_nodes( + std::shared_ptr router) { + return router->_cached_edge_nodes; + } + + static void drop_struck_cached_edge_nodes(std::shared_ptr router) { + router->_drop_struck_cached_edge_nodes(); + } + static void build_path( std::shared_ptr router, PathCategory category, @@ -832,4 +846,82 @@ TEST_CASE("Network", "[network][onion_request_router][check_request_queue_timeou CHECK(result.timeout); } +TEST_CASE("Network", "[network][onion_request_router][cached_edge_nodes]") { + const auto node_strike_threshold = 3; + config::SnodePool pool_config = { + .cache_directory = std::nullopt, + .fallback_snode_pool_path = std::nullopt, + .cache_expiration = std::chrono::minutes{5}, + .cache_min_lifetime = std::chrono::minutes{5}, + .enforce_subnet_diversity = false, + .retry_delay = network::opt::retry_delay{50ms, 200ms}, + .netid = opt::netid::Target::testnet, + .seed_nodes = {}, + .cache_min_size = 0, + .cache_min_swarm_size = 0, + .cache_num_nodes_to_use_for_refresh = 3, + .cache_min_num_refresh_presence_to_include_node = 2, + .cache_node_strike_threshold = node_strike_threshold}; + config::OnionRequestRouter config = { + file_server::DEFAULT_CONFIG, + std::nullopt, + std::chrono::days{10}, + opt::netid::Target::testnet, + {}, + network::opt::retry_delay{50ms, 200ms}, + 3, + 3, + 10, + 10min, + node_strike_threshold, + true, + true, + {{PathCategory::standard, 1}}}; + + auto ed_pk = "4cb76fdc6d32278e3f83dbf608360ecc6b65727934b85d2fb86862ff98c46ab7"_hexbytes; + auto ed_pk2 = "5ea34e72bb044654a6a23675690ef5ffaaf1656b02f93fb76655f9cbdbe89876"_hexbytes; + auto healthy = service_node{ + ed25519_pubkey::from_bytes(ed_pk), + oxen::quic::ipv4{"127.0.0.1"}, + 20001, + 30001, + {2, 11, 0}, + 0}; + auto struck = service_node{ + ed25519_pubkey::from_bytes(ed_pk2), + oxen::quic::ipv4{"127.0.0.2"}, + 20002, + 30002, + {2, 11, 0}, + 0}; + + auto loop = std::make_shared(); + auto disk_loop = std::make_shared(); + auto snode_pool = std::make_shared(pool_config, loop, disk_loop); + auto transport = std::make_shared(); + auto router = + std::make_shared(config, loop, disk_loop, snode_pool, transport); + + auto now = std::chrono::system_clock::now(); + TestOnionRequestRouter::set_cached_edge_nodes( + router, {cached_edge_node{healthy, now}, cached_edge_node{struck, now}}); + + // Both are well inside `edge_node_cache_duration`, which is the only thing that used to matter + TestOnionRequestRouter::drop_struck_cached_edge_nodes(router); + CHECK(TestOnionRequestRouter::cached_edge_nodes(router).size() == 2); + + // A cached edge node is forced as a path's first hop without going through `get_unused_nodes`, + // so nothing else would apply the strike filter to it. Striking it out has to cost it the + // cached-edge role outright - skipping it while keeping the entry would hand the role back when + // the strikes expire, long after we moved to another edge node. + snode_pool->SnodePool::record_node_failure(struck, true); + REQUIRE(snode_pool->node_struck_out(struck)); + + // Only the cached-edge role is lost; nothing here touches `_snode_cache`, so it stays a node + // like any other and `get_unused_nodes` can pick it again when its strikes expire + TestOnionRequestRouter::drop_struck_cached_edge_nodes(router); + auto remaining = TestOnionRequestRouter::cached_edge_nodes(router); + REQUIRE(remaining.size() == 1); + CHECK(remaining.front().node == healthy); +} } // namespace session::network diff --git a/tests/test_snode_pool.cpp b/tests/test_snode_pool.cpp index d470ed193..57052dc43 100644 --- a/tests/test_snode_pool.cpp +++ b/tests/test_snode_pool.cpp @@ -69,6 +69,23 @@ class TestSnodePool : public SnodePool { // loop thread void update_cache(std::vector nodes) { _update_cache("test", std::move(nodes)); } + // Backdates every recorded strike so `STRIKE_EXPIRY` can be crossed without a 48h wait + void debug_age_strikes(std::chrono::seconds by) { + _loop->call_get([this, by] { + for (auto& [key, stamps] : _snode_strikes) + for (auto& stamp : stamps) + stamp -= by; + }); + } + + // Every strike on record, expired or not - which is what the decision sites used to count + size_t debug_recorded_strikes(const ed25519_pubkey& key) { + return _loop->call_get([this, &key] { + auto it = _snode_strikes.find(key); + return (it == _snode_strikes.end() ? size_t{0} : it->second.size()); + }); + } + void debug_on_refresh_complete(std::vector> raw_results) { auto total_requests = static_cast(raw_results.size()); _loop->call_get([&] { @@ -336,3 +353,57 @@ TEST_CASE("Network", "[network][refresh_min_cache_size]") { snode_pool->debug_on_refresh_complete({to_snode_cache_bin(enough)}); CHECK(snode_pool->size() == 12); } + +TEST_CASE("Network", "[network][strike_expiry]") { + session::network::config::SnodePool pool_config{ + .cache_expiration = 5min, + .cache_min_lifetime = 5min, + .enforce_subnet_diversity = false, + .retry_delay = network::opt::retry_delay{50ms, 200ms}, + .netid = opt::netid::Target::testnet, + .cache_min_size = 0, + .cache_min_swarm_size = 0, + .cache_num_nodes_to_use_for_refresh = 0, + .cache_min_num_refresh_presence_to_include_node = 0, + .cache_node_strike_threshold = 3}; + + // Strikes are keyed by node pubkey, so these have to differ per node or striking one strikes + // every node sharing its key + std::vector> ed_pks{ + "4cb76fdc6d32278e3f83dbf608360ecc6b65727934b85d2fb86862ff98c46ab7"_hexbytes, + "5ea34e72bb044654a6a23675690ef5ffaaf1656b02f93fb76655f9cbdbe89876"_hexbytes, + "e17a692033200ae41350df9709754edde7343e2cf2f23e88f993319e0720e5e5"_hexbytes, + "7b633fa6fb462b90db6f0f50384190ce7715e31b7aa93d87dbd7e94e33d4251f"_hexbytes}; + std::vector snode_cache; + + for (uint16_t i = 0; i < ed_pks.size(); ++i) + snode_cache.emplace_back(service_node{ + ed25519_pubkey::from_bytes(ed_pks[i]), + oxen::quic::ipv4{"192.168.0.{}"_format(i)}, + static_cast(20000 + i), + static_cast(30000 + i), + {2, 11, 0}, + 0}); + + auto loop = std::make_shared(); + auto disk_loop = std::make_shared(); + auto snode_pool = std::make_shared(pool_config, loop, disk_loop); + snode_pool->reset_state_with_cache(snode_cache); + + snode_pool->record_node_failure(snode_cache[0], true); + snode_pool->record_node_failure(snode_cache[1], true); + REQUIRE(snode_pool->node_strike_count(snode_cache[0]) == 3); + REQUIRE(snode_pool->get_unused_nodes(4).size() == 2); + + // Once the strikes are older than `STRIKE_EXPIRY` the node has to become selectable again; a + // node dropped for a blip two days ago is not evidence about the node now + snode_pool->debug_age_strikes(49h); + CHECK(snode_pool->node_strike_count(snode_cache[0]) == 0); + CHECK(snode_pool->get_unused_nodes(4).size() == 4); + + // ... and a fresh failure starts from one rather than stacking onto the expired ones, which is + // also what stops the vector growing for the life of the process + snode_pool->record_node_failure(snode_cache[0]); + CHECK(snode_pool->node_strike_count(snode_cache[0]) == 1); + CHECK(snode_pool->debug_recorded_strikes(snode_cache[0].remote_pubkey) == 1); +}