From e942c37dc5ca2af3a34b84e78b6eac1b33adc2ff Mon Sep 17 00:00:00 2001 From: Morgan Pretty Date: Fri, 11 Sep 2026 10:46:00 +1000 Subject: [PATCH] Tell callers when the snode pool could not be refreshed `refresh_if_needed` had three ways to do nothing and say nothing: a suspended pool, no candidate nodes, and no fetcher. The callback was simply never invoked, and every caller sets its own state up before calling, so "never" is not a delay - it is permanent: - `_resync_clock` sets `_current_clock_resync_id` first, so a dropped callback makes every later clock resync short-circuit as "already in progress" and strands everything queued behind it. - both routers finish setup from the callback, so they never finish it. - `get_random_nodes` and the path builder retry from the callback, so their callers wait on something that will not arrive. The callback now takes whether a refresh actually happened, which is the smallest thing that lets a caller tell "the pool is fine" from "we could not find out". Nothing needing a refresh reports true - that is the answer the caller wanted, not a failure to get one. Each caller had to be given a failing branch, and two of them would have spun rather than hung if the callback had simply always been invoked: `get_random_nodes` re-enters itself, and the path builder rebuilds into the same too-few-nodes branch. `SnodePool` has no C API surface, so this is internal only. --- include/session/network/session_network.hpp | 2 +- include/session/network/snode_pool.hpp | 11 ++- src/network/routing/onion_request_router.cpp | 52 +++++++++++-- src/network/routing/session_router_router.cpp | 20 ++++- src/network/session_network.cpp | 46 +++++++++-- src/network/snode_pool.cpp | 68 +++++++++++------ tests/test_onion_request_router.cpp | 2 +- tests/test_snode_pool.cpp | 76 +++++++++++++++++-- 8 files changed, 231 insertions(+), 46 deletions(-) diff --git a/include/session/network/session_network.hpp b/include/session/network/session_network.hpp index 9c8f07eb9..a713c7b52 100644 --- a/include/session/network/session_network.hpp +++ b/include/session/network/session_network.hpp @@ -125,7 +125,7 @@ class Network : public std::enable_shared_from_this { const uint8_t index, const service_node& node, const uint8_t total_requests); - void _on_clock_resync_complete(const uint8_t total_requests); + void _on_clock_resync_complete(); Request _preprocess_request(Request request); }; diff --git a/include/session/network/snode_pool.hpp b/include/session/network/snode_pool.hpp index 787e2c582..7bfd45550 100644 --- a/include/session/network/snode_pool.hpp +++ b/include/session/network/snode_pool.hpp @@ -42,6 +42,12 @@ class empty_file_exception : public std::runtime_error { class SnodePool : public std::enable_shared_from_this { public: using network_fetcher_t = std::function; + + // `refreshed` is false when the pool could not be refreshed at all - suspended, no candidate + // nodes, no fetcher. Callers that retry their work on this callback have to check it: without + // it the only way to signal "that didn't happen" was to never call back, which leaves whatever + // the caller set up beforehand set up forever. + using refresh_callback_t = std::function; using fetcher_connectivity_check_t = std::function; SnodePool( @@ -74,7 +80,7 @@ class SnodePool : public std::enable_shared_from_this { // Checks if the pool is empty or stale and triggers a refresh if needed virtual void refresh_if_needed( const std::vector& in_use_nodes, - std::function on_refresh_complete = nullptr); + refresh_callback_t on_refresh_complete = nullptr); virtual void get_swarm( session::network::x25519_pubkey swarm_pubkey, @@ -113,7 +119,7 @@ class SnodePool : public std::enable_shared_from_this { int _snode_cache_refresh_failure_count = 0; std::vector _refresh_candidate_nodes; std::vector> _snode_refresh_results; - std::vector> _after_snode_cache_refresh; + std::vector _after_snode_cache_refresh; // Disk I/O functions void _load_from_disk(); @@ -145,6 +151,7 @@ class SnodePool : public std::enable_shared_from_this { const bool use_direct_fetcher, const uint8_t total_requests); void _update_cache(std::string refresh_id, std::vector nodes); + void _run_pending_refresh_callbacks(bool refreshed); }; } // namespace session::network diff --git a/src/network/routing/onion_request_router.cpp b/src/network/routing/onion_request_router.cpp index 99f6c4549..9917543a3 100644 --- a/src/network/routing/onion_request_router.cpp +++ b/src/network/routing/onion_request_router.cpp @@ -331,7 +331,12 @@ OnionRequestRouter::OnionRequestRouter( } if (snode_pool->size() == 0) - snode_pool->refresh_if_needed({}, [weak_self = weak_from_this()] { + snode_pool->refresh_if_needed({}, [weak_self = weak_from_this()](bool refreshed) { + // Setup finishes either way: not finishing leaves the router permanently unusable, + // where finishing with an empty pool just means the first request refreshes again + if (!refreshed) + log::warning(cat, "Finishing router setup without a refreshed snode pool."); + if (auto self = weak_self.lock()) self->_loop->call([weak_self] { if (auto self = weak_self.lock()) @@ -1160,15 +1165,50 @@ void OnionRequestRouter::_build_path( snode_pool->refresh_if_needed( nodes_to_exclude, - [weak_self = weak_from_this(), - this, - category, - initiating_req_id, - nodes_to_exclude]() { + [weak_self = weak_from_this(), this, category, initiating_req_id, nodes_to_exclude]( + bool refreshed) { auto self = weak_self.lock(); if (!self) return; + // Rebuilding without a refresh lands back in this same branch with the same + // too-few nodes, so the queued requests would wait on a loop that cannot end + if (!refreshed) { + log::error( + cat, + "[Request {}]: Cannot build a path, the snode pool could not be " + "refreshed.", + initiating_req_id.value_or("internal")); + _update_status(); + + auto queue_it = _request_queues.find(category); + if (queue_it == _request_queues.end()) { + log::critical( + cat, + "No request queue for category '{}'.", + to_string(category, _config.single_path_mode)); + return; + } + + if (!queue_it->second->is_empty()) { + auto to_fail = queue_it->second->pop_all(); + log::error( + cat, + "Failing {} queued requests for '{}' paths; the snode pool " + "could not be refreshed.", + to_fail.size(), + to_string(category, _config.single_path_mode)); + + for (const auto& [req, cb] : to_fail) + cb(false, + false, + -1, + {content_type_plain_text}, + "Failed to refresh the snode pool to build a path."); + } + return; + } + log::info( cat, "[Request {}]: SnodePool refresh complete, retrying path build.", diff --git a/src/network/routing/session_router_router.cpp b/src/network/routing/session_router_router.cpp index cf1606d05..5dd80ecbd 100644 --- a/src/network/routing/session_router_router.cpp +++ b/src/network/routing/session_router_router.cpp @@ -131,11 +131,19 @@ void SessionRouter::_init() { return; if (snode_pool->size() == 0) - snode_pool->refresh_if_needed({}, [weak_self, this] { + snode_pool->refresh_if_needed({}, [weak_self, this](bool refreshed) { auto self = weak_self.lock(); if (!self) return; + // Setup finishes either way: not finishing leaves the router + // permanently unusable, where finishing with an empty pool just means + // the first request refreshes again + if (!refreshed) + log::warning( + cat, + "Finishing router setup without a refreshed snode pool."); + _loop->call([weak_self] { if (auto self = weak_self.lock()) self->_finish_setup(); @@ -458,11 +466,19 @@ void SessionRouter::_send_proxy_request(Request request, network_response_callba [weak_self = weak_from_this(), this, req = std::move(request), - cb = std::move(callback)]() { + cb = std::move(callback)](bool refreshed) { auto self = weak_self.lock(); if (!self) return; + if (!refreshed) + return cb( + false, + false, + -1, + {content_type_plain_text}, + "Failed to refresh the snode pool to find a proxy."); + auto snode_pool = _snode_pool.lock(); if (!snode_pool) return cb( diff --git a/src/network/session_network.cpp b/src/network/session_network.cpp index 7d160fc41..eb231bbe0 100644 --- a/src/network/session_network.cpp +++ b/src/network/session_network.cpp @@ -409,8 +409,19 @@ void Network::get_random_nodes( std::vector nodes_to_exclude = _router->get_all_used_nodes(); return _snode_pool->refresh_if_needed( - nodes_to_exclude, - [this, count, cb = std::move(cb)] { get_random_nodes(count, cb); }); + nodes_to_exclude, [this, count, cb = std::move(cb)](bool refreshed) { + // Retrying without a refresh would re-enter this same branch and recurse + // until the stack gives out + if (!refreshed) { + log::warning( + cat, + "Cannot get {} random nodes: the pool could not be refreshed.", + count); + return cb({}); + } + + get_random_nodes(count, cb); + }); } cb(unused_nodes); }); @@ -836,7 +847,17 @@ void Network::_handle_421_retry( [this, req_to_retry = std::move(original_request), cb = std::move(final_callback), - failed_node = failed_node_copy] { + failed_node = failed_node_copy](bool refreshed) { + // Without a refresh the swarm cache still holds the mapping the 421 disproved, so + // the retry would go straight back to the swarm that just rejected us + if (!refreshed) + return cb( + false, + false, + ERROR_MISDIRECTED_REQUEST, + {content_type_plain_text}, + "421 Misdirected Request, and the snode pool could not be refreshed"); + auto swarm_pubkey = *req_to_retry.swarm_pubkey; _snode_pool->get_swarm( @@ -933,7 +954,20 @@ void Network::_resync_clock( // Refresh the snode pool if needed to ensure we have the most up-to-date cache std::vector nodes_to_exclude = _router->get_all_used_nodes(); - _snode_pool->refresh_if_needed(std::move(nodes_to_exclude), [this, request_id] { + _snode_pool->refresh_if_needed(std::move(nodes_to_exclude), [this, request_id](bool refreshed) { + // `_current_clock_resync_id` was set before this call, so giving up without clearing it + // blocks every later resync attempt for the life of the process and strands everything + // queued behind it. Completing with no results takes the existing "resync finished, + // successful or not" path, which does both and leaves the offset we already had alone - + // an old offset beats none, and this says nothing about whether it was right. + if (!refreshed) { + log::warning( + cat, + "[Request {}] Abandoning clock resync: the snode pool could not be refreshed.", + request_id); + return _on_clock_resync_complete(); + } + // Pick the random nodes we want to use for retrying (these won't change for this resync // attempt) auto resync_nodes = @@ -1040,11 +1074,11 @@ void Network::_launch_next_clock_out_of_sync_request( // If we've received all the results then we need to process them and complete the // resync if (_clock_resync_results.size() >= total_requests) - _on_clock_resync_complete(total_requests); + _on_clock_resync_complete(); }); } -void Network::_on_clock_resync_complete(const uint8_t /*total_requests*/) { +void Network::_on_clock_resync_complete() { auto raw_results = std::move(_clock_resync_results); auto refresh_id = std::move(*_current_clock_resync_id); diff --git a/src/network/snode_pool.cpp b/src/network/snode_pool.cpp index d4f61ad9a..f4416b341 100644 --- a/src/network/snode_pool.cpp +++ b/src/network/snode_pool.cpp @@ -339,6 +339,11 @@ void SnodePool::_refresh_snode_cache(std::optional request_id_opt) _loop->call([this, request_id_opt] { if (_suspended) { log::info(cat, "Ignoring refresh as pool is suspended."); + + // Anything queued against a refresh we are declining to start has nothing left to wait + // for, unless one is already in flight and still owns it + if (!_current_snode_cache_refresh_id) + _run_pending_refresh_callbacks(false); return; } @@ -416,6 +421,7 @@ void SnodePool::_refresh_snode_cache(std::optional request_id_opt) (use_seed_nodes ? "No seed nodes are configured!" : "Found no nodes and decided not to use seed nodes!")); _current_snode_cache_refresh_id.reset(); + _run_pending_refresh_callbacks(false); return; } @@ -570,6 +576,7 @@ void SnodePool::_launch_next_refresh_request( cat, "[Request {}] No fetcher available, aborting refresh.", target_request_id); _current_snode_cache_refresh_id.reset(); _refresh_candidate_nodes.clear(); + _run_pending_refresh_callbacks(false); return; } @@ -889,28 +896,35 @@ void SnodePool::_update_cache(std::string refresh_id, std::vector SnodePool::_perform_cache_write(path, cache); }); - // Trigger any callbacks - // - // These must be moved out of the member before being run: a callback can re-enter the pool - // and register another post-refresh callback (`get_swarm` does exactly that if the cache is - // still empty), which would reallocate the vector we're iterating and leave us calling - // through a freed `std::function`. Anything registered while we're here accumulates in the - // now-empty member and runs after the next refresh instead of being silently discarded. - if (!_after_snode_cache_refresh.empty()) { - auto callbacks = std::move(_after_snode_cache_refresh); - _after_snode_cache_refresh.clear(); // A moved-from vector is valid but unspecified + _run_pending_refresh_callbacks(true); + }); +} - log::debug(cat, "Executing {} post-refresh callbacks.", callbacks.size()); +void SnodePool::_run_pending_refresh_callbacks(bool refreshed) { + if (_after_snode_cache_refresh.empty()) + return; - for (const auto& cb : callbacks) { - try { - cb(); - } catch (const std::exception& e) { - log::error(cat, "Exception thrown in a post-refresh callback: {}", e.what()); - } - } + // These must be moved out of the member before being run: a callback can re-enter the pool and + // register another post-refresh callback (`get_swarm` does exactly that if the cache is still + // empty), which would reallocate the vector we're iterating and leave us calling through a + // freed `std::function`. Anything registered while we're here accumulates in the now-empty + // member and runs after the next refresh instead of being silently discarded. + auto callbacks = std::move(_after_snode_cache_refresh); + _after_snode_cache_refresh.clear(); // A moved-from vector is valid but unspecified + + log::debug( + cat, + "Executing {} post-refresh callbacks ({}).", + callbacks.size(), + (refreshed ? "refreshed" : "refresh did not happen")); + + for (const auto& cb : callbacks) { + try { + cb(refreshed); + } catch (const std::exception& e) { + log::error(cat, "Exception thrown in a post-refresh callback: {}", e.what()); } - }); + } } // MARK: Public Functions @@ -1040,10 +1054,13 @@ void SnodePool::clear_node_strikes() { } void SnodePool::refresh_if_needed( - const std::vector& in_use_nodes, std::function on_refresh_complete) { + const std::vector& in_use_nodes, refresh_callback_t on_refresh_complete) { _loop->call([this, in_use_nodes, cb = std::move(on_refresh_complete)] { if (_suspended) { log::info(cat, "Ignoring refresh as pool is suspended."); + + if (cb) + cb(false); return; } @@ -1109,7 +1126,9 @@ void SnodePool::refresh_if_needed( } else _refresh_snode_cache(); else if (!already_running && cb) - cb(); + // Nothing needed refreshing, which is the answer the caller wanted rather than a + // failure to get one + cb(true); }); } @@ -1240,7 +1259,12 @@ void SnodePool::get_swarm( // Queue this entire function call to be re-run after the refresh. _after_snode_cache_refresh.push_back( - [this, swarm_pubkey, ignore_strike_count, cb = std::move(cb)]() { + [this, swarm_pubkey, ignore_strike_count, cb = std::move(cb)](bool refreshed) { + // Re-deferring when the refresh never happened would just queue against the + // next one that doesn't either; there is no swarm to give back + if (!refreshed) + return cb(swarm::INVALID_SWARM_ID, {}); + this->get_swarm(swarm_pubkey, ignore_strike_count, std::move(cb)); }); diff --git a/tests/test_onion_request_router.cpp b/tests/test_onion_request_router.cpp index 8913315d8..3f12c4f67 100644 --- a/tests/test_onion_request_router.cpp +++ b/tests/test_onion_request_router.cpp @@ -154,7 +154,7 @@ namespace { void refresh_if_needed( const std::vector& /*in_use_nodes*/, - std::function /*on_refresh_complete*/ = nullptr) override { + refresh_callback_t /*on_refresh_complete*/ = nullptr) override { func_called("refresh_if_needed"); // Do nothing (don't want to trigger a cache refresh) } diff --git a/tests/test_snode_pool.cpp b/tests/test_snode_pool.cpp index d470ed193..33dacf5b9 100644 --- a/tests/test_snode_pool.cpp +++ b/tests/test_snode_pool.cpp @@ -32,11 +32,19 @@ class TestSnodePool : public SnodePool { void refresh_if_needed( const std::vector& /*in_use_nodes*/, - std::function /*on_refresh_complete*/ = nullptr) override { + refresh_callback_t /*on_refresh_complete*/ = nullptr) override { // Do nothing (don't want to trigger a cache refresh) } - void debug_queue_post_refresh_callback(std::function cb) { + // Reaches past the no-op override above to the real age-based policy, on the loop thread so it + // resolves inline + void debug_refresh_if_needed(refresh_callback_t cb) { + _loop->call_get([this, cb = std::move(cb)]() mutable { + SnodePool::refresh_if_needed({}, std::move(cb)); + }); + } + + void debug_queue_post_refresh_callback(refresh_callback_t cb) { _loop->call_get([this, cb = std::move(cb)]() mutable { _after_snode_cache_refresh.push_back(std::move(cb)); }); @@ -263,12 +271,12 @@ TEST_CASE("Network", "[network][update_cache]") { // what a deferred `get_swarm` does when the refresh left the cache empty) rather than // invalidating the vector it's iterating std::vector callbacks_run; - snode_pool->debug_queue_post_refresh_callback([&] { + snode_pool->debug_queue_post_refresh_callback([&](bool) { callbacks_run.push_back(0); - snode_pool->debug_queue_post_refresh_callback([&] { callbacks_run.push_back(3); }); + snode_pool->debug_queue_post_refresh_callback([&](bool) { callbacks_run.push_back(3); }); }); - snode_pool->debug_queue_post_refresh_callback([&] { callbacks_run.push_back(1); }); - snode_pool->debug_queue_post_refresh_callback([&] { callbacks_run.push_back(2); }); + snode_pool->debug_queue_post_refresh_callback([&](bool) { callbacks_run.push_back(1); }); + snode_pool->debug_queue_post_refresh_callback([&](bool) { callbacks_run.push_back(2); }); REQUIRE(snode_pool->debug_remove_post_refresh_callback_spare_capacity()); snode_pool->update_cache({}); CHECK(callbacks_run == std::vector{0, 1, 2}); @@ -336,3 +344,59 @@ 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][refresh_callback_contract]") { + session::network::config::SnodePool pool_config{ + .cache_expiration = 2h, + .cache_min_lifetime = 2s, + .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 = 0, + .cache_min_num_refresh_presence_to_include_node = 0, + .cache_node_strike_threshold = 3}; + auto ed_pk = "4cb76fdc6d32278e3f83dbf608360ecc6b65727934b85d2fb86862ff98c46ab7"_hexbytes; + std::vector snode_cache; + + for (uint16_t i = 0; i < 5; ++i) + snode_cache.emplace_back(service_node{ + ed25519_pubkey::from_bytes(ed_pk), + 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); + + auto run = [&snode_pool] { + std::pair result{false, true}; // {called, refreshed} + snode_pool->debug_refresh_if_needed( + [&result](bool refreshed) { result = {true, refreshed}; }); + return result; + }; + + // An empty cache and no seed nodes means the refresh cannot even start. The callback still has + // to run: callers set state up before calling (`_resync_clock` sets its in-progress id) and + // never hearing back leaves that set for the life of the process. + auto [called, refreshed] = run(); + CHECK(called); + CHECK_FALSE(refreshed); + + // Nothing needing a refresh is the answer the caller asked for, not a failure to get one + snode_pool->update_cache(snode_cache); + std::tie(called, refreshed) = run(); + CHECK(called); + CHECK(refreshed); + + // ... and a suspended pool reports that it didn't refresh rather than going quiet + snode_pool->suspend(); + std::tie(called, refreshed) = run(); + CHECK(called); + CHECK_FALSE(refreshed); +}