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 { void _close_connections(); void _recalculate_status(); void _update_status(ConnectionStatus new_status); - void _update_network_state(const std::string& body); - void _handle_421_retry(Request original_request, network_response_callback_t final_callback); + void _update_network_state(const nlohmann::json& json); + + // An account's redirect is left empty where there was none to take. + response::swarm_rejections _take_swarm_redirects( + const Request& request, const nlohmann::json* json, int16_t status_code); + void _refresh_if_unredirected(const response::swarm_rejections& rejections); + void _handle_421_retry( + Request original_request, + response::swarm_rejections rejections, + std::vector> headers, + std::optional body, + network_response_callback_t final_callback); void _resync_clock( std::optional original_request, network_response_callback_t request_callback); @@ -125,7 +135,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/session_network_types.h b/include/session/network/session_network_types.h index a830cdebb..f312d5381 100644 --- a/include/session/network/session_network_types.h +++ b/include/session/network/session_network_types.h @@ -66,7 +66,9 @@ typedef struct { // The account whose swarm this request addresses, as 64 hex chars of its X25519 pubkey — the // same value passed to session_network_get_swarm() to choose the destination node. Null for a // request aimed at a node rather than about an account (a cache refresh, a clock resync) or at - // a server, which no swarm membership applies to. + // a server, which no swarm membership applies to. For a `batch` or `sequence` request each + // sub-request's account is read from the body instead, and this is used only if the body + // can't be parsed. // // Set it whenever there is one: a storage server rejects a request whose pubkey is outside its // swarm with a 421, and recovering means re-resolving this account's swarm. Left null, such a diff --git a/include/session/network/session_network_types.hpp b/include/session/network/session_network_types.hpp index 266cbf12b..efecd7e01 100644 --- a/include/session/network/session_network_types.hpp +++ b/include/session/network/session_network_types.hpp @@ -2,7 +2,9 @@ #include #include +#include #include +#include #include #include "session/network/key_types.hpp" @@ -28,6 +30,8 @@ constexpr int16_t ERROR_FAILED_TO_QUEUE_REQUEST = -10008; constexpr int16_t ERROR_INVALID_DESTINATION = -10009; constexpr int16_t ERROR_FAILED_GENERATE_ONION_PAYLOAD = -10010; constexpr int16_t ERROR_FAILED_TO_GET_STREAM = -10011; +constexpr int16_t ERROR_UNAUTHENTICATED_RESPONSE = -10012; +constexpr int16_t ERROR_INSUFFICIENT_NODES = -10013; constexpr int16_t ERROR_BUILD_TIMEOUT = -10100; constexpr int16_t ERROR_REQUEST_CANCELLED = -10200; constexpr int16_t ERROR_UNKNOWN = -11000; @@ -166,15 +170,16 @@ struct Request { /// Any extra request details which may modify the structure of the request. RequestDetails details; - /// The account whose swarm this request addresses, when it addresses one: the X25519 pubkey of - /// a session ID, group ID, etc. Set it for anything sent to a swarm member *about* that - /// account, and leave it unset for a request merely aimed at a node (a snode cache refresh, a - /// clock resync), which no swarm membership applies to. + /// The accounts this request addresses: one entry per sub-request of a `batch` or `sequence`, + /// in order, or a single entry for any other request. Each is the 32-byte pubkey an account's + /// swarm is computed from (a session ID or group ID without its prefix); an entry is empty for + /// a sub-request that isn't about an account. A single entry applies to every sub-request. + /// Leave it empty for a request merely aimed at a node (a snode cache refresh, a clock resync). /// - /// Required to recover from a 421: the storage server rejects a request whose pubkey is not in - /// its swarm, and recovering means re-resolving the swarm of *this account*, which cannot be - /// derived from the node we happened to ask. - std::optional swarm_pubkey; + /// Required to recover from a 421: it is what says whose swarm the redirect a 421 carries is + /// for. The account a storage server names in a 421 is only checked against it, never trusted + /// in its place. + std::vector> swarm_pubkeys; /// The time the request was created, this is used primarily for determining whether the /// `overall_timeout` has been exceeded. @@ -266,9 +271,61 @@ using network_response_callback_t = std::function> headers, std::optional response)>; +/// The accounts of each sub-request in a pregenerated `batch`/`sequence` body, in the form +/// `Request::swarm_pubkeys` takes, for a request whose body was built by the caller. A +/// sub-request whose params name no account gets `fallback` instead: the account the caller said +/// the request is for. Returns nullopt when `endpoint` is neither or the body can't be parsed. +std::optional>> batch_request_accounts( + std::string_view endpoint, + std::span body, + std::optional fallback = std::nullopt); + namespace response { std::optional> parse_text_error(std::string_view body); std::optional find_uniform_batch_error(std::string_view body); + + /// As `find_uniform_batch_error`, for a body that has already been parsed. + std::optional uniform_batch_error(const nlohmann::json& json); + + /// The status to report for a proxied response that could not be decrypted, whose status + /// therefore came from whichever node on the way answered rather than from the destination. + /// Statuses that are acted on as the destination's verdict (406, 421, 425) become + /// `ERROR_UNAUTHENTICATED_RESPONSE`, since any node on the path could have sent them; anything + /// else is returned unchanged. + int16_t undecrypted_status(int16_t status); + + /// One result of a `batch`/`sequence` response, or the whole of any other response. + struct subresponse { + /// The result's own status code; nullopt for a response that isn't a batch, whose status + /// is the one it came back with. + std::optional code; + + /// The result's body when it is a JSON object. Points into the parsed response. + const nlohmann::json* body = nullptr; + }; + + /// The sub-responses of a parsed response: one per entry of `results` for a batch or sequence, + /// otherwise the response itself as the only one. The returned bodies point into `json`. + std::vector subresponses(const nlohmann::json& json); + + /// The accounts a response's 421s rejected. + struct swarm_rejections { + /// Each rejected account, with the ed25519 pubkeys of the swarm its 421 redirected it to: + /// empty when the 421 named no swarm. An account with several 421s takes the first. + std::unordered_map> accounts; + + /// Whether a 421 could not be attributed to any account. + bool unattributed = false; + }; + + /// Finds the 421s in a storage server response. `json` is its parsed body, or null when the + /// body isn't JSON. `status_code` is the response's own status, which is the only status a + /// response that isn't a batch has. `accounts` is the request's `swarm_pubkeys`, which each + /// 421 is attributed by; a 421 naming a different account than its position does is ignored. + swarm_rejections find_swarm_rejections( + const nlohmann::json* json, + int16_t status_code, + std::span> accounts); } // namespace response struct OnionPathMetadata { diff --git a/include/session/network/snode_pool.hpp b/include/session/network/snode_pool.hpp index 787e2c582..4ebd252aa 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( @@ -69,12 +75,38 @@ class SnodePool : 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 virtual void refresh_if_needed( const std::vector& in_use_nodes, - std::function on_refresh_complete = nullptr); + refresh_callback_t on_refresh_complete = nullptr); + + // Re-resolves `swarm_pubkey` after a node we believed was in its swarm has told us the account + // isn't there. Where `refresh_if_needed` decides on the cache's age, this decides on that + // rejection, so a mapping we have direct proof is wrong can be replaced long before + // `cache_expiration`. `on_complete` runs once the swarm can be resolved again. + virtual void invalidate_swarm( + session::network::x25519_pubkey swarm_pubkey, refresh_callback_t on_complete = nullptr); + + // Records the swarm a node named when it rejected a request for `swarm_pubkey`, so the mapping + // can be corrected without refreshing the whole pool - a swarm change would otherwise have + // every client of that swarm fetch the full node list, where nothing made them before. + // + // Only the node pubkeys are taken from the response, and only ones that resolve against the + // pool we fetched ourselves, so a redirect can reach registered service nodes we already know + // and nothing else. Returns false when the redirect is unusable, which is the caller's cue to + // fall back to `invalidate_swarm`. + virtual bool record_swarm_redirect( + session::network::x25519_pubkey swarm_pubkey, + const std::vector& swarm_node_keys); virtual void get_swarm( session::network::x25519_pubkey swarm_pubkey, @@ -100,6 +132,13 @@ class SnodePool : public std::enable_shared_from_this { std::vector>> _all_swarms; std::unordered_map>> _swarm_cache; + + // Swarms a node has redirected us to, which `get_swarm` prefers over its own calculation. Kept + // apart from `_swarm_cache` deliberately: that is a memo of `_all_swarms` and stays one, where + // these are claims from outside that outrank it until the pool is refreshed. The count is the + // account's redirects since that refresh, which past a limit also ask for the next one. + std::unordered_map, uint8_t>> + _swarm_overrides; std::map> _snode_strikes; bool _strikes_flush_scheduled = false; @@ -109,11 +148,18 @@ class SnodePool : public std::enable_shared_from_this { // Refresh logic std::chrono::system_clock::time_point _last_snode_cache_update; + std::chrono::system_clock::time_point _last_evidence_refresh; + std::chrono::seconds _evidence_refresh_backoff{0}; std::optional _current_snode_cache_refresh_id; 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; + + // 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(); @@ -145,6 +191,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..0c379b173 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()) @@ -595,9 +600,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) { @@ -1160,15 +1193,61 @@ 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 lands back in this same branch unless the pool can now fill a + // path, and `refreshed` includes nothing having needed refreshing, which calls + // back inline - so rebuilding without checking first recurses until the stack + // gives out + std::optional cannot_build; + if (!refreshed) + cannot_build = "the snode pool could not be refreshed"; + else if (auto pool = _snode_pool.lock(); + !pool || + pool->get_unused_nodes(_config.path_length, nodes_to_exclude).size() < + _config.path_length) + cannot_build = "the snode pool has too few usable nodes"; + + if (cannot_build) { + log::error( + cat, + "[Request {}]: Cannot build a path, {}.", + initiating_req_id.value_or("internal"), + *cannot_build); + _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; {}.", + to_fail.size(), + to_string(category, _config.single_path_mode), + *cannot_build); + + for (const auto& [req, cb] : to_fail) + cb(false, + false, + ERROR_INSUFFICIENT_NODES, + {content_type_plain_text}, + "Cannot build a path: {}."_format(*cannot_build)); + } + return; + } + log::info( cat, "[Request {}]: SnodePool refresh complete, retrying path build.", @@ -1566,51 +1645,49 @@ void OnionRequestRouter::_send_on_path( return; } - auto decryption_callback = [weak_self = weak_from_this(), - this, - parser = std::move(parser), - path_id = path.id, - original_request = std::move(request), - cb = std::move(callback)]( - bool success, - bool timeout, - int16_t status, - auto headers, - auto response) { - auto self = weak_self.lock(); - if (!self) - return; + auto decryption_callback = + [weak_self = weak_from_this(), + this, + parser = std::move(parser), + path_id = path.id, + original_request = std::move(request), + cb = std::move(callback)]( + bool success, bool timeout, int16_t status, auto headers, auto response) { + auto self = weak_self.lock(); + if (!self) + return; - try { - if (!success) - throw std::runtime_error{response.value_or("Unknown request failure")}; - if (timeout) - throw std::runtime_error{response.value_or("Timed out")}; - if (!response) - throw std::runtime_error{"Unexpected empty response"}; - - onionreq::DecryptedResponse decrypted = parser->decrypted_response(*response); - _handle_transport_response( - path_id, - std::move(original_request), - true, - false, - decrypted.status_code, - std::move(decrypted.headers), - std::move(decrypted.body), - std::move(cb)); - } catch (const std::exception& e) { - _handle_transport_response( - path_id, - std::move(original_request), - false, - timeout, - status, - std::move(headers), - std::move("Failed to handle onion response due to error: {}"_format(e.what())), - std::move(cb)); - } - }; + try { + if (!success) + throw std::runtime_error{response.value_or("Unknown request failure")}; + if (timeout) + throw std::runtime_error{response.value_or("Timed out")}; + if (!response) + throw std::runtime_error{"Unexpected empty response"}; + + onionreq::DecryptedResponse decrypted = parser->decrypted_response(*response); + _handle_transport_response( + path_id, + std::move(original_request), + true, + false, + decrypted.status_code, + std::move(decrypted.headers), + std::move(decrypted.body), + std::move(cb)); + } catch (const std::exception& e) { + _handle_transport_response( + path_id, + std::move(original_request), + false, + timeout, + response::undecrypted_status(status), + std::move(headers), + "Failed to handle onion response (status {}) due to error: {}"_format( + status, e.what()), + std::move(cb)); + } + }; transport->send_request(std::move(onion_request), std::move(decryption_callback)); } @@ -2128,13 +2205,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/routing/session_router_router.cpp b/src/network/routing/session_router_router.cpp index cf1606d05..19240cfc1 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, + ERROR_INSUFFICIENT_NODES, + {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( @@ -476,7 +492,7 @@ void SessionRouter::_send_proxy_request(Request request, network_response_callba return cb( false, false, - -1, + ERROR_INSUFFICIENT_NODES, {content_type_plain_text}, "SnodePool refresh failed."); @@ -523,32 +539,35 @@ void SessionRouter::_send_proxy_request(Request request, network_response_callba request.time_remaining(), request.overall_timeout}; - auto proxy_callback = - [parser = std::move(parser), cb = std::move(callback)]( - bool success, bool timeout, int16_t status, auto headers, auto response) { - try { - if (!success) - throw std::runtime_error{response.value_or("Unknown request failure")}; - if (timeout) - throw std::runtime_error{response.value_or("Timed out")}; - if (!response) - throw std::runtime_error{"Unexpected empty response"}; - - onionreq::DecryptedResponse decrypted = parser->decrypted_response(*response); - cb(true, - false, - decrypted.status_code, - std::move(decrypted.headers), - std::move(decrypted.body)); - } catch (const std::exception& e) { - cb(false, - timeout, - status, - std::move(headers), - "Failed to handle proxied request response due to error: {}"_format( - e.what())); - } - }; + auto proxy_callback = [parser = std::move(parser), cb = std::move(callback)]( + bool success, + bool timeout, + int16_t status, + auto headers, + auto response) { + try { + if (!success) + throw std::runtime_error{response.value_or("Unknown request failure")}; + if (timeout) + throw std::runtime_error{response.value_or("Timed out")}; + if (!response) + throw std::runtime_error{"Unexpected empty response"}; + + onionreq::DecryptedResponse decrypted = parser->decrypted_response(*response); + cb(true, + false, + decrypted.status_code, + std::move(decrypted.headers), + std::move(decrypted.body)); + } catch (const std::exception& e) { + cb(false, + timeout, + response::undecrypted_status(status), + std::move(headers), + "Failed to handle proxied request response (status {}) due to error: {}"_format( + status, e.what())); + } + }; // Now that we have a service_node destination we can send a direct request _send_direct_request(std::move(proxy_request), std::move(proxy_callback)); diff --git a/src/network/session_network.cpp b/src/network/session_network.cpp index 7d160fc41..1270771db 100644 --- a/src/network/session_network.cpp +++ b/src/network/session_network.cpp @@ -1,6 +1,7 @@ #include "session/network/session_network.hpp" #include +#include #include #include @@ -292,12 +293,14 @@ void Network::suspend() { _loop->call_get([this] { _suspended = true; + // The router before the transport, so that it has let go of its path builds and failure + // listeners before the transport closes the connections they're attached to if (_snode_pool) _snode_pool->suspend(); - if (_transport) - _transport->suspend(); if (_router) _router->suspend(); + if (_transport) + _transport->suspend(); _close_connections(); }); @@ -409,8 +412,30 @@ 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) { + if (!refreshed) { + log::warning( + cat, + "Cannot get {} random nodes: the pool could not be refreshed.", + count); + return cb({}); + } + + // `refreshed` includes nothing having needed refreshing, which calls back + // inline, so asking again of a pool that can't supply `count` would + // re-enter this branch until the stack gives out + auto nodes = _snode_pool->get_unused_nodes(count); + if (nodes.size() < count) { + log::warning( + cat, + "Cannot get {} random nodes: the pool has only {} to give.", + count, + nodes.size()); + return cb({}); + } + + cb(std::move(nodes)); + }); } cb(unused_nodes); }); @@ -456,20 +481,34 @@ void Network::send_request(Request request, network_response_callback_t callback const auto dest_is_snode = std::holds_alternative(original_req.destination); + // Parsed once for everything below, since a retrieve's body can be large + std::optional json; + if (body) + if (auto parsed = nlohmann::json::parse(*body, nullptr, false); + !parsed.is_discarded()) + json = std::move(parsed); + // If we got a successful response (with a body) and the request was sent to a // service node then we should update the network state based on the response // (Note: we don't want to do this for server requests because they could // include values in different formats, eg. the "Session Network" API returns // `t` in seconds) - if (success && body && dest_is_snode) - _update_network_state(*body); + if (success && json && dest_is_snode) + _update_network_state(*json); int16_t final_status_code = status_code; - if (body) - if (auto uniform_error = response::find_uniform_batch_error(*body)) + if (json) + if (auto uniform_error = response::uniform_batch_error(*json)) final_status_code = *uniform_error; + // Taken from every snode response rather than only a 421 one: a batch can be + // rejected for some accounts and answered for others + response::swarm_rejections rejections; + if (dest_is_snode) + rejections = _take_swarm_redirects( + original_req, json ? &*json : nullptr, status_code); + // If we got a 406 from a snode, or a 425 from a server, then the device clock // is out of sync so we need to kick off a clock resync request if ((final_status_code == ERROR_NOT_ACCEPTABLE && dest_is_snode) || @@ -482,10 +521,17 @@ void Network::send_request(Request request, network_response_callback_t callback // cache, the original request might succeed after this refresh so we should // just automatically retry if (final_status_code == 421) { - _handle_421_retry(std::move(original_req), std::move(cb)); + _handle_421_retry( + std::move(original_req), + std::move(rejections), + std::move(headers), + std::move(body), + std::move(cb)); return; } + _refresh_if_unredirected(rejections); + // For debugging purposes we want to add a log if this was a successful request // after we did an automatic retry if (original_req.retry_count > 0) @@ -607,10 +653,13 @@ void Network::download(DownloadRequest request) { // MARK: Internal Logic void Network::_close_connections() { - if (_transport) - _transport->close_connections(); + // The router first, for the same reason as in `suspend`: otherwise each edge connection the + // transport closes fires a failure listener into a live router, which retires the path and + // starts rebuilding it if (_router) _router->close_connections(); + if (_transport) + _transport->close_connections(); _recalculate_status(); log::info(cat, "Closed all connections."); @@ -703,37 +752,22 @@ Request Network::_preprocess_request(Request request) { return request; } -void Network::_update_network_state(const std::string& body) { +void Network::_update_network_state(const nlohmann::json& json) { try { - auto json = nlohmann::json::parse(body); const nlohmann::json* target_json = &json; // If it was a batch/sequence request then take the one with the highest "t" value as that // would have been the one which was returned last - if (json.contains("results") && json["results"].is_array()) { - log::trace(cat, "Parsing batch response for latest network state."); - - int64_t max_t = -1; - const nlohmann::json* latest_body = nullptr; - - for (const auto& result : json["results"]) { - if (!result.is_object() || !result.contains("body") || !result["body"].is_object()) - continue; - - const auto& result_body = result["body"]; - - if (result_body.contains("t") && result_body["t"].is_number()) { - int64_t current_t = result_body["t"].get(); - - if (current_t > max_t) { - max_t = current_t; - latest_body = &result_body; - } + int64_t max_t = -1; + for (const auto& sub : response::subresponses(json)) { + if (!sub.body) + continue; + + if (auto t = sub.body->find("t"); t != sub.body->end() && t->is_number()) + if (auto current_t = t->get(); current_t > max_t) { + max_t = current_t; + target_json = sub.body; } - } - - if (latest_body) - target_json = latest_body; } // Update hardfork/softfork versions @@ -779,13 +813,40 @@ void Network::_update_network_state(const std::string& body) { // MARK: Specific Error Handling +// A node that rejects a request for an account usually names the swarm it actually belongs to. +// Taking its word (bounded - see `record_swarm_redirect`) fixes the one mapping we know is wrong, +// where refreshing has every client of a changed swarm fetch the full node list for it. +response::swarm_rejections Network::_take_swarm_redirects( + const Request& request, const nlohmann::json* json, int16_t status_code) { + auto rejections = response::find_swarm_rejections(json, status_code, request.swarm_pubkeys); + + for (auto& [account, keys] : rejections.accounts) + if (!keys.empty() && !_snode_pool->record_swarm_redirect(account, keys)) + keys.clear(); + + return rejections; +} + +// With no usable redirect the pool itself is the only thing that can put a rejection right, and +// one refresh re-resolves every account at once +void Network::_refresh_if_unredirected(const response::swarm_rejections& rejections) { + for (const auto& [account, keys] : rejections.accounts) + if (keys.empty()) + return _snode_pool->invalidate_swarm(account); +} + void Network::_handle_421_retry( - Request original_request, network_response_callback_t final_callback) { + Request original_request, + response::swarm_rejections rejections, + std::vector> headers, + std::optional body, + network_response_callback_t final_callback) { if (original_request.retry_count >= config.redirect_retry_count) { log::error( cat, "Request {} received 421 but exceeded max retry count.", original_request.request_id); + _refresh_if_unredirected(rejections); return final_callback( false, false, @@ -806,12 +867,12 @@ void Network::_handle_421_retry( "Received 421 from a non-service-node destination"); // A 421 says the account we asked about is not in this node's swarm, so recovering means - // re-resolving *that account's* swarm. Nothing about the node we asked can tell us which - // account that was, so a request that did not record one cannot be redirected. - if (!original_request.swarm_pubkey) { + // re-resolving *that account's* swarm; without knowing which account that was, there is + // nothing to re-resolve. + if (rejections.accounts.empty()) { log::warning( cat, - "Request {} received 421 but carries no swarm pubkey to re-resolve.", + "Request {} received 421 but names no account to re-resolve.", original_request.request_id); return final_callback( false, @@ -821,63 +882,86 @@ void Network::_handle_421_retry( "421 Misdirected Request for a request with no swarm"); } - // If we got a 421 it means our snode cache is outdated (because the swarm the destination node - // belongs to doesn't match our cache anymore) + // A retry goes to a single swarm, which a batch spanning several accounts doesn't have. Their + // redirects are already taken, so the next request for each goes to the right place. + if (rejections.accounts.size() > 1 || rejections.unattributed) { + log::info( + cat, + "Request {} received 421s for {} accounts; not retrying it as a whole.", + original_request.request_id, + rejections.accounts.size()); + _refresh_if_unredirected(rejections); + return final_callback( + false, false, ERROR_MISDIRECTED_REQUEST, std::move(headers), std::move(body)); + } + log::info( cat, - "Request {} received 421 from node {}, refreshing swarm if stale.", + "Request {} received 421 from node {}, re-resolving its swarm.", original_request.request_id, original_dest_node->to_string()); auto failed_node_copy = *original_dest_node; - std::vector nodes_to_exclude = _router->get_all_used_nodes(); - _snode_pool->refresh_if_needed( - std::move(nodes_to_exclude), - [this, - req_to_retry = std::move(original_request), - cb = std::move(final_callback), - failed_node = failed_node_copy] { - auto swarm_pubkey = *req_to_retry.swarm_pubkey; - - _snode_pool->get_swarm( - swarm_pubkey, - false, - [this, - req_to_retry = std::move(req_to_retry), - cb = std::move(cb), - failed_node](swarm::swarm_id_t, std::vector swarm_nodes) { - // Extract a single random index from the vector indices, but excluding - // the index of the failing node: - size_t new_target; - auto out = std::ranges::sample( - std::views::iota(0, static_cast(swarm_nodes.size())) | - std::views::filter([&](int i) { - return swarm_nodes[i] != failed_node; - }), - &new_target, - 1, - csrng); - - if (out == &new_target) - return cb( - false, - false, - ERROR_MISDIRECTED_REQUEST, - {content_type_plain_text}, - "421 Misdirected Request, but no other nodes in swarm to " - "retry"); - - log::info( - cat, - "Request {} retrying 421 error on new node {}.", - req_to_retry.request_id, - swarm_nodes[new_target].to_string()); - auto final_request = req_to_retry; - final_request.retry_count++; - final_request.destination = std::move(swarm_nodes[new_target]); - this->send_request(std::move(final_request), std::move(cb)); - }); - }); + auto [swarm_pubkey, redirect] = *rejections.accounts.begin(); + + auto retry_on_resolved_swarm = [this, + swarm_pubkey, + req_to_retry = std::move(original_request), + cb = std::move(final_callback), + failed_node = failed_node_copy](bool swarm_resolved) mutable { + // Without a re-resolve there is nothing better than the mapping the 421 disproved - a + // redirect naming any node we know would have been followed instead - so a retry would go + // straight back to the swarm that just rejected us. The refresh has been asked for, and + // the account's next rejection asks again. + if (!swarm_resolved) + return cb( + false, + false, + ERROR_MISDIRECTED_REQUEST, + {content_type_plain_text}, + "421 Misdirected Request, and the swarm could not be re-resolved"); + + _snode_pool->get_swarm( + swarm_pubkey, + false, + [this, req_to_retry = std::move(req_to_retry), cb = std::move(cb), failed_node]( + swarm::swarm_id_t, std::vector swarm_nodes) { + // Extract a single random index from the vector indices, but excluding + // the index of the failing node: + size_t new_target; + auto out = std::ranges::sample( + std::views::iota(0, static_cast(swarm_nodes.size())) | + std::views::filter( + [&](int i) { return swarm_nodes[i] != failed_node; }), + &new_target, + 1, + csrng); + + if (out == &new_target) + return cb( + false, + false, + ERROR_MISDIRECTED_REQUEST, + {content_type_plain_text}, + "421 Misdirected Request, but no other nodes in swarm to " + "retry"); + + log::info( + cat, + "Request {} retrying 421 error on new node {}.", + req_to_retry.request_id, + swarm_nodes[new_target].to_string()); + auto final_request = req_to_retry; + final_request.retry_count++; + final_request.destination = std::move(swarm_nodes[new_target]); + this->send_request(std::move(final_request), std::move(cb)); + }); + }; + + if (!redirect.empty()) + return retry_on_resolved_swarm(true); + + _snode_pool->invalidate_swarm(swarm_pubkey, std::move(retry_on_resolved_swarm)); } void Network::_resync_clock( @@ -933,7 +1017,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 +1137,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); @@ -1765,8 +1862,17 @@ LIBSESSION_C_API void session_network_send_request( std::nullopt, request_id}; + std::optional swarm_pubkey; if (params->swarm_pubkey_hex) - request.swarm_pubkey = x25519_pubkey::from_hex({params->swarm_pubkey_hex, 64}); + swarm_pubkey = x25519_pubkey::from_hex({params->swarm_pubkey_hex, 64}); + + std::optional>> batch_accounts; + if (request.body) + batch_accounts = batch_request_accounts(request.endpoint, *request.body, swarm_pubkey); + if (batch_accounts) + request.swarm_pubkeys = std::move(*batch_accounts); + else if (swarm_pubkey) + request.swarm_pubkeys = {*swarm_pubkey}; auto cpp_callback = [c_cb = callback, c_ctx = ctx]( bool success, diff --git a/src/network/session_network_types.cpp b/src/network/session_network_types.cpp index 4b40e1325..7f5af921a 100644 --- a/src/network/session_network_types.cpp +++ b/src/network/session_network_types.cpp @@ -73,35 +73,170 @@ std::optional> response::parse_text_error(std::string_v return std::nullopt; } +namespace { + inline auto cat = log::Cat("network"); + + const nlohmann::json* field( + const nlohmann::json& j, + std::string_view key, + bool (nlohmann::json::*is_kind)() const noexcept) { + auto it = j.find(key); + return it != j.end() && ((*it).*is_kind)() ? &*it : nullptr; + } + + std::optional account_in(const nlohmann::json& obj) { + auto pk = field(obj, "pubkey", &nlohmann::json::is_string); + if (!pk) + return std::nullopt; + + // Prefixed with the network id, except on testnet + auto hex = pk->get(); + if (hex.size() == 66) + hex.remove_prefix(2); + return x25519_pubkey::maybe_from_hex(hex); + } + + // Only the pubkeys: the rest of each record is contact information we deliberately don't read, + // so a redirect can't put us in touch with anything the pool hasn't already told us about. + std::vector redirect_keys(const nlohmann::json& body) { + std::vector keys; + + if (auto snodes = field(body, "snodes", &nlohmann::json::is_array)) + for (const auto& snode : *snodes) + if (auto pk = field(snode, "pubkey_ed25519", &nlohmann::json::is_string)) + if (auto key = ed25519_pubkey::maybe_from_hex(pk->get())) + keys.push_back(*key); + + return keys; + } + + std::optional account_by_position( + std::span> accounts, size_t index) { + if (accounts.size() == 1) + return accounts.front(); + return index < accounts.size() ? accounts[index] : std::nullopt; + } +} // namespace + +int16_t response::undecrypted_status(int16_t status) { + switch (status) { + case ERROR_NOT_ACCEPTABLE: + case ERROR_MISDIRECTED_REQUEST: + case ERROR_TOO_EARLY: return ERROR_UNAUTHENTICATED_RESPONSE; + default: return status; + } +} + std::optional response::find_uniform_batch_error(std::string_view body) { try { - auto json = nlohmann::json::parse(body); + return uniform_batch_error(nlohmann::json::parse(body)); + } catch (...) { /* Do nothing */ + } + + return std::nullopt; +} - // If it wasn't a batch response then just handle the non-batch status code - if (json.contains("results") && json["results"].is_array() && !json["results"].empty()) { - int16_t first_status_code = -1; +std::optional response::uniform_batch_error(const nlohmann::json& json) { + std::optional uniform; - for (const auto& result : json["results"]) { - if (!result.contains("code") || !result["code"].is_number()) - return std::nullopt; + // A response that isn't a batch has no per-result codes, so its own status stands + for (const auto& sub : subresponses(json)) { + if (!sub.code || (*sub.code >= 200 && *sub.code <= 299)) + return std::nullopt; + if (uniform && *uniform != *sub.code) + return std::nullopt; + uniform = sub.code; + } - // If we got a success then we can just use the original status code - int16_t code = result["code"].get(); - if (code >= 200 && code <= 299) - return std::nullopt; + return uniform; +} - if (first_status_code == -1) - first_status_code = code; - else if (first_status_code != code) - return std::nullopt; - } +std::vector response::subresponses(const nlohmann::json& json) { + auto results = field(json, "results", &nlohmann::json::is_array); + if (!results) + return {{std::nullopt, json.is_object() ? &json : nullptr}}; - return first_status_code; + std::vector subs; + subs.reserve(results->size()); + + for (const auto& result : *results) { + auto& sub = subs.emplace_back(); + if (auto code = field(result, "code", &nlohmann::json::is_number)) + sub.code = code->get(); + sub.body = field(result, "body", &nlohmann::json::is_object); + } + + return subs; +} + +std::optional>> batch_request_accounts( + std::string_view endpoint, + std::span body, + std::optional fallback) { + if (endpoint != "batch" && endpoint != "sequence") + return std::nullopt; + + auto json = nlohmann::json::parse(body.begin(), body.end(), nullptr, false); + auto requests = field(json, "requests", &nlohmann::json::is_array); + if (!requests) + return std::nullopt; + + std::vector> accounts; + accounts.reserve(requests->size()); + + for (const auto& req : *requests) { + auto& account = accounts.emplace_back(); + if (auto params = field(req, "params", &nlohmann::json::is_object)) + account = account_in(*params); + if (!account) + account = fallback; + } + + return accounts; +} + +response::swarm_rejections response::find_swarm_rejections( + const nlohmann::json* json, + int16_t status_code, + std::span> accounts) { + swarm_rejections result; + + // A body that isn't JSON still leaves the response's own status to act on, just with no + // redirect to go with it + auto subs = json ? subresponses(*json) : std::vector{{}}; + + for (size_t i = 0; i < subs.size(); ++i) { + const auto& [code, sub_body] = subs[i]; + if (code.value_or(status_code) != ERROR_MISDIRECTED_REQUEST) + continue; + + // Attributed by position in our own request, never by the account the 421 names: that is + // the responding node's word, and taking it would let any node we talk to redirect any + // account it likes - not just the ones it was asked about - onto nodes of its choosing + auto account = account_by_position(accounts, i); + if (!account) { + result.unattributed = true; + continue; } - } catch (...) { /* Do nothing */ + + if (auto echoed = sub_body ? account_in(*sub_body) : std::nullopt; + echoed && *echoed != *account) { + log::warning( + cat, + "Ignoring a 421 for {} that names {} as the rejected account.", + account->hex(), + echoed->hex()); + result.unattributed = true; + continue; + } + + // Several sub-requests for one account (a retrieve per namespace, say) all name its swarm, + // so the first is as good as any + if (auto [it, inserted] = result.accounts.try_emplace(*account); inserted && sub_body) + it->second = redirect_keys(*sub_body); } - return std::nullopt; + return result; } } // namespace session::network diff --git a/src/network/snode_pool.cpp b/src/network/snode_pool.cpp index d4f61ad9a..3dc99999e 100644 --- a/src/network/snode_pool.cpp +++ b/src/network/snode_pool.cpp @@ -8,6 +8,7 @@ #include #include #include +#include #include #include #include @@ -42,6 +43,25 @@ namespace { const std::chrono::seconds STRIKE_EXPIRY = 48h; const std::chrono::seconds SAVE_THROTTLE = 5min; + + // A swarm rejection is evidence against the pool snapshot the swarm was derived from, so it is + // only worth acting on once that snapshot is old enough for a refresh to plausibly return + // something different. + const std::chrono::seconds EVIDENCE_REFRESH_MIN_POOL_AGE = 1min; + + // A rejection arriving *after* we already refreshed on one is the refresh telling us it didn't + // help - the network hasn't settled, or the node is lying - so each successive one waits twice + // as long, up to `cache_expiration`. A single 3-node refresh per minute sustained indefinitely + // is not an acceptable resting state for a client on a bad connection, and without this it is + // exactly what a persistent rejection source would produce. + const std::chrono::seconds EVIDENCE_REFRESH_BASE_BACKOFF = 1min; + + // An account redirected more than this many times within one pool generation is being bounced + // between nodes that disagree, or its swarm has moved on too far for our pool to keep up. A + // redirect is still the best information we have, so past it each one is still followed, and + // also asks for the refresh that can settle it. Legitimate rearrangements count too (nothing + // resets this short of a refresh), hence the headroom. + constexpr uint8_t SWARM_REDIRECTS_BEFORE_REFRESH = 8; } // namespace SnodePool::SnodePool( @@ -339,6 +359,10 @@ 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: suspending abandons any refresh in progress, so none can answer it later + _run_pending_refresh_callbacks(false); return; } @@ -416,6 +440,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; } @@ -438,7 +463,9 @@ void SnodePool::_launch_next_refresh_request( refreshing_from_seed_nodes, use_direct_fetcher, total_requests] { - if (!_current_snode_cache_refresh_id) + // A retry scheduled for a refresh that has since been abandoned or replaced mustn't touch + // the one in its place - its candidates in particular + if (_current_snode_cache_refresh_id != request_id) return; const auto target_request_id = "{}-{}"_format(request_id, index); @@ -544,7 +571,7 @@ void SnodePool::_launch_next_refresh_request( "trying again in {}ms.", target_request_id, delay.count()); - _loop->call_later(delay, [weak_self = weak_from_this(), this] { + _loop->call_later(delay, [weak_self = weak_from_this(), this, request_id] { // We need to wait until after the `call_later` to reset the `refresh_id` (and clear // previous results) as if we don't then additional refreshes could be triggered // during the delay @@ -552,6 +579,11 @@ void SnodePool::_launch_next_refresh_request( if (!self) return; + // Abandoned (by a suspend) or replaced during the delay: resetting the id would + // wipe out whichever refresh has it now, and there is nothing of ours to restart + if (_current_snode_cache_refresh_id != request_id) + return; + _current_snode_cache_refresh_id.reset(); _snode_refresh_results.clear(); _refresh_snode_cache(); @@ -570,6 +602,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; } @@ -877,6 +910,11 @@ void SnodePool::_update_cache(std::string refresh_id, std::vector _snode_cache = std::move(nodes); _all_swarms = swarm::generate_swarms(_snode_cache); _swarm_cache.clear(); + + // The pool these redirects were correcting has been replaced, so they have outlived what + // they were for. Keeping them would let one node's claim outlast the ground truth that + // would have overruled it. + _swarm_overrides.clear(); _last_snode_cache_update = std::chrono::system_clock::now(); // Reset all failure and refresh-in-progress state @@ -889,28 +927,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 @@ -920,6 +965,22 @@ void SnodePool::suspend() { _loop->call_get([this] { _suspended = true; + // A refresh in progress can only fail from here on, since every fetch is refused while + // suspended, so it is abandoned rather than left to work its way through the candidates, + // and whatever waits on it is answered now. The failures it built up are dropped too: + // they say nothing about the network we resume on. + if (_current_snode_cache_refresh_id) { + log::info( + cat, + "[Request {}] Abandoning cache refresh for suspension.", + *_current_snode_cache_refresh_id); + _current_snode_cache_refresh_id.reset(); + _refresh_candidate_nodes.clear(); + _snode_refresh_results.clear(); + _snode_cache_refresh_failure_count = 0; + _run_pending_refresh_callbacks(false); + } + // Force a strike write immediately if we had one scheduled if (_strikes_flush_scheduled) _disk_loop->call([path = _strikes_file_path, strikes = _snode_strikes] { @@ -958,11 +1019,31 @@ void SnodePool::clear_cache() { _snode_cache = {}; _all_swarms = {}; _swarm_cache = {}; + _swarm_overrides = {}; _disk_loop->call([path = _snode_cache_file_path] { SnodePool::_clear_disk_cache(path); }); }); } +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 +1051,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 +1089,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() { @@ -1040,10 +1114,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; } @@ -1068,9 +1145,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 @@ -1109,7 +1184,159 @@ 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); + }); +} + +void SnodePool::invalidate_swarm(x25519_pubkey swarm_pubkey, refresh_callback_t on_complete) { + _loop->call([this, swarm_pubkey, cb = std::move(on_complete)]() mutable { + if (_suspended) { + log::info(cat, "Ignoring swarm invalidation as pool is suspended."); + + if (cb) + cb(false); + return; + } + + // The swarm cache only ever memoises `swarm::get_swarm(pubkey, _all_swarms)`, and + // `_all_swarms` is generated from the pool, so dropping the single entry would recompute + // the same rejected answer. What the rejection actually disproves is the pool snapshot, + // and a refresh is the only thing that replaces it - clearing the swarm cache as it goes. + auto now = std::chrono::system_clock::now(); + + if (!_current_snode_cache_refresh_id) { + // The backoff is deliberately global rather than per-swarm: a refresh replaces the pool + // every swarm is derived from, so a rejection for one swarm is exactly as unhelpful to + // act on right after a refresh as a second rejection for the swarm that prompted it + auto pool_age = now - _last_snode_cache_update; + auto since_last = now - _last_evidence_refresh; + + // Quiet for twice the interval we were holding means whatever caused the last run is + // over, so the next incident starts from the base delay rather than inheriting it + if (since_last > 2 * _evidence_refresh_backoff) + _evidence_refresh_backoff = 0s; + + if (pool_age < EVIDENCE_REFRESH_MIN_POOL_AGE || since_last < _evidence_refresh_backoff) + log::debug( + cat, + "Swarm {} was rejected, but the pool is {}s old and the last refresh on a " + "rejection was {}s ago (backoff {}s); leaving it alone.", + swarm_pubkey.hex(), + std::chrono::duration_cast(pool_age).count(), + std::chrono::duration_cast(since_last).count(), + _evidence_refresh_backoff.count()); + else { + _evidence_refresh_backoff = std::min( + std::chrono::seconds{_config.cache_expiration}, + (_evidence_refresh_backoff == 0s ? EVIDENCE_REFRESH_BASE_BACKOFF + : _evidence_refresh_backoff * 2)); + _last_evidence_refresh = now; + + log::info( + cat, + "Swarm {} was rejected by a node we had in it, refreshing the pool (next " + "rejection-driven refresh no sooner than {}s from now).", + swarm_pubkey.hex(), + _evidence_refresh_backoff.count()); + _refresh_snode_cache(); + } + } + + if (!cb) + return; + + // `_refresh_snode_cache` runs inline here (we are on the loop thread) and can decline to + // start, so it is the refresh id - not the call above - that says whether there is anything + // to wait for. A callback queued against a refresh that never started never runs, and on + // this path that would strand the request that triggered the rejection. + if (_current_snode_cache_refresh_id) + _after_snode_cache_refresh.push_back(std::move(cb)); + else + // No refresh started, so the caller is holding the same swarm answer it came in with + cb(false); + }); +} + +bool SnodePool::record_swarm_redirect( + x25519_pubkey swarm_pubkey, const std::vector& swarm_node_keys) { + return _loop->call_get([this, swarm_pubkey, &swarm_node_keys] { + if (_snode_cache.empty() || _all_swarms.empty()) + return false; + + // Resolving the names against our own pool is what makes a redirect safe to act on: the + // responding node chooses which registered nodes we end up talking to, but it cannot invent + // a node or point us at an address of its own choosing, because we never read the contact + // details it sent + std::unordered_set named{swarm_node_keys.begin(), swarm_node_keys.end()}; + std::vector resolved; + + for (const auto& node : _snode_cache) + if (named.count(node.remote_pubkey)) + resolved.push_back(node); + + if (resolved.empty()) { + log::debug( + cat, + "Ignoring redirect for {}: none of the {} named nodes are in our pool.", + swarm_pubkey.hex(), + swarm_node_keys.size()); + return false; + } + + // Too few of the named nodes resolved to call the result a swarm: either our pool is too + // old to know the rest, or the redirect isn't describing one. Only a refresh settles + // which, and the nodes we do have still beat the swarm that just rejected us, so they are + // followed meanwhile. Deliberately not also requiring that some names were unknown - a + // redirect naming one node we do know is the shape with no stale pool to explain it. + const bool too_few_resolved = resolved.size() < _config.cache_min_swarm_size; + + auto& [nodes, redirects] = _swarm_overrides[swarm_pubkey]; + nodes = std::move(resolved); + + // Saturating, because counting carries on past the limit for as long as the refresh it + // asks for is throttled, and wrapping would stop it asking + if (redirects < std::numeric_limits::max()) + ++redirects; + + log::debug( + cat, + "Redirected to a swarm of {} nodes for {} (redirect {} since the pool was " + "refreshed).", + nodes.size(), + swarm_pubkey.hex(), + redirects); + + if (too_few_resolved) { + log::warning( + cat, + "Asking for a refresh for {}: only {} of the {} nodes its redirect named are " + "ones we know.", + swarm_pubkey.hex(), + nodes.size(), + named.size()); + invalidate_swarm(swarm_pubkey); + } else if (redirects > SWARM_REDIRECTS_BEFORE_REFRESH) { + // Past the limit every redirect asks again, for as long as the refresh stays + // throttled, so only crossing it is worth a warning + if (redirects == SWARM_REDIRECTS_BEFORE_REFRESH + 1) + log::warning( + cat, + "Asking for a refresh for {}: {} redirects since the pool was refreshed.", + swarm_pubkey.hex(), + redirects); + else + log::debug( + cat, + "Asking again for a refresh for {}: {} redirects since the pool was " + "refreshed.", + swarm_pubkey.hex(), + redirects); + invalidate_swarm(swarm_pubkey); + } + + return true; }); } @@ -1158,9 +1385,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 @@ -1190,14 +1415,21 @@ void SnodePool::get_swarm( log::trace(cat, "{} called for {}.", __PRETTY_FUNCTION__, swarm_pubkey.hex()); _loop->call([this, swarm_pubkey, ignore_strike_count, cb = std::move(callback)] { + // Trigger a non-blocking background refresh if the data is stale. Up here so that it runs + // for every call: answers from an override or the swarm cache below return early, and + // they are most of them. + _loop->call_soon([weak_self = weak_from_this()] { + if (auto self = weak_self.lock()) + self->refresh_if_needed({}); + }); + auto filter_by_strikes = [this](std::vector nodes) -> std::vector { // Shuffle everything to start with 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 @@ -1224,6 +1456,17 @@ void SnodePool::get_swarm( return nodes; }; + // A node that rejected a request for this account told us where it belongs, which beats + // anything we can work out from a pool we now know is behind. There is no swarm id to give + // back - a redirect names members, not an id - and nothing downstream reads it. + if (auto it = _swarm_overrides.find(swarm_pubkey); it != _swarm_overrides.end()) { + const auto& swarm_nodes = it->second.first; + + return cb( + swarm::INVALID_SWARM_ID, + (ignore_strike_count ? swarm_nodes : filter_by_strikes(swarm_nodes))); + } + // Check the in-memory swarm cache first if (auto it = _swarm_cache.find(swarm_pubkey); it != _swarm_cache.end()) { const auto& swarm_nodes = it->second.second; @@ -1240,7 +1483,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)); }); @@ -1251,12 +1499,6 @@ void SnodePool::get_swarm( return; } - // Trigger a non-blocking background refresh if the data is stale - _loop->call_soon([weak_self = weak_from_this()] { - if (auto self = weak_self.lock()) - self->refresh_if_needed({}); - }); - // Perform the swarm calculation using our local copy of the data auto swarm = swarm::get_swarm(swarm_pubkey, _all_swarms); _swarm_cache[swarm_pubkey] = swarm; diff --git a/src/network/transport/quic_transport.cpp b/src/network/transport/quic_transport.cpp index f90f72007..a743944db 100644 --- a/src/network/transport/quic_transport.cpp +++ b/src/network/transport/quic_transport.cpp @@ -51,7 +51,7 @@ QuicTransport::~QuicTransport() { void QuicTransport::suspend() { // Use 'call_get' to force this to be synchronous _loop->call_get([this] { - if (!_suspended) + if (_suspended) return; _suspended = true; @@ -148,42 +148,62 @@ void QuicTransport::_recreate_endpoint() { // The optional must CONTAIN the option to have any effect: libquic's optional-taking // handle_ep_opt overload does nothing when the optional is empty, so a default-constructed // std::optional{} silently leaves discovery enabled. + // + // TODO: replace with quic::opt::max_udp_payload::minimum(), which is the same value under the + // name that isn't deprecated. That needs a newer libquic than this pin carries, and moving the + // pin needs the dependency handling in external/CMakeLists.txt moved onto session-deps first - + // the newer session-router defines oxen::quic itself, which collides with our own definition of + // it. Both of those come over with the Session client work, and this goes when they do. + // + // Until then the deprecated name is the only spelling that compiles against both this pin and + // the newer liboxenquic the Linux CI stages take from the system, which deprecates it under + // -Werror. The replacement does not exist in the pinned copy at all. +#pragma GCC diagnostic push +#pragma GCC diagnostic ignored "-Wdeprecated-declarations" _endpoint = quic::Endpoint::endpoint( *_loop, quic::Address{}, (_config.disable_mtu_discovery ? std::make_optional() : std::nullopt)); +#pragma GCC diagnostic pop } void QuicTransport::_close_connections() { - // Explicitly close all connections then reset the endpoint + // Taken out before the endpoint goes. Resetting it destroys it on the spot (close_conns() + // only defers), and its destructor closes every connection with code 0 through the handler + // that fails whatever is pending on it - so left in place, our own close would reach each + // pending verification as an error code, which is what gets a node struck. Answering copies + // also keeps a callback that calls back into the transport from changing the maps being walked. + auto verifications = std::exchange(_pending_verification_callbacks, {}); + auto requests = std::exchange(_pending_requests, {}); + if (_endpoint) _endpoint->close_conns(); _endpoint.reset(); - // Cancel any pending verifications (they can't succeed once the connection is closed) - for (const auto& [pubkey, callbacks] : _pending_verification_callbacks) + // Clear all storage of requests, paths and connections so that we are in a fresh state on + // relaunch + _active_connection_ids.clear(); + _available_stream_ids.clear(); + _update_status(ConnectionStatus::disconnected); + + // Answered only now the transport is fully closed, so anything a callback asks of it fails + // cleanly, without an error code, rather than meeting half-closed state. A verification gets + // no error code either: a close we asked for is no evidence against the node. + for (const auto& [pubkey, callbacks] : verifications) for (const auto& callback : callbacks) - callback(false, -1); + callback(false, std::nullopt); - // Cancel any pending requests (they can't succeed once the connection is closed) - for (const auto& [pubkey, pupkey_requests] : _pending_requests) - for (const auto& [info, callback] : pupkey_requests) + for (const auto& [pubkey, pubkey_requests] : requests) + for (const auto& [info, callback] : pubkey_requests) callback( false, false, - ERROR_NETWORK_SUSPENDED, + (_suspended ? ERROR_NETWORK_SUSPENDED : ERROR_CONNECTION_CLOSED), {content_type_plain_text}, - "QuickTransport is suspended."); + (_suspended ? "QuickTransport is suspended." + : "QuickTransport closed its connections.")); - // Clear all storage of requests, paths and connections so that we are in a fresh state on - // relaunch - _active_connection_ids.clear(); - _available_stream_ids.clear(); - _pending_verification_callbacks.clear(); - _pending_requests.clear(); - - _update_status(ConnectionStatus::disconnected); log::info(cat, "Closed all connections."); } diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 2ed0d1af5..3f0cd7f9a 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -44,6 +44,7 @@ if(ENABLE_NETWORKING) list(APPEND LIB_SESSION_UTESTS_SOURCES test_network_swarm.cpp) list(APPEND LIB_SESSION_UTESTS_SOURCES test_onionreq.cpp) list(APPEND LIB_SESSION_UTESTS_SOURCES test_onion_request_router.cpp) + list(APPEND LIB_SESSION_UTESTS_SOURCES test_quic_transport.cpp) list(APPEND LIB_SESSION_UTESTS_SOURCES test_snode_pool.cpp) endif() diff --git a/tests/test_config_userprofile.cpp b/tests/test_config_userprofile.cpp index 9133e38a8..1c72e2eb6 100644 --- a/tests/test_config_userprofile.cpp +++ b/tests/test_config_userprofile.cpp @@ -9,6 +9,7 @@ #include #include #include +#include #include "utils.hpp" diff --git a/tests/test_network_swarm.cpp b/tests/test_network_swarm.cpp index 887ea581e..fa9dc0019 100644 --- a/tests/test_network_swarm.cpp +++ b/tests/test_network_swarm.cpp @@ -1,6 +1,7 @@ #include #include #include +#include #include #include @@ -198,3 +199,188 @@ TEST_CASE("Swarm", "[network][swarm][get_swarm]") { "05000000000000000000000000000000000000000000000000fffffffffffffffe", swarms) == 0); } + +namespace { +using json = nlohmann::json; + +x25519_pubkey account(char c) { + return x25519_pubkey::from_hex(std::string(64, c)); +} + +std::vector nodes(std::string_view chars) { + std::vector keys; + for (char c : chars) + keys.push_back(ed25519_pubkey::from_hex(std::string(64, c))); + return keys; +} + +json rejection(std::optional echoed, std::string_view node_chars) { + auto snodes = json::array(); + for (char c : node_chars) + snodes.push_back({{"pubkey_ed25519", std::string(64, c)}, {"ip", "10.0.0.1"}}); + + json body = {{"snodes", std::move(snodes)}, {"swarm", "abc"}}; + if (echoed) + body["pubkey"] = *echoed; + return body; +} + +json batch(std::vector> results) { + auto list = json::array(); + for (auto& [code, body] : results) + list.push_back({{"code", code}, {"body", std::move(body)}}); + return {{"results", std::move(list)}}; +} + +response::swarm_rejections rejections( + const json& body, + int16_t status_code, + std::vector> accounts = {}) { + return response::find_swarm_rejections(&body, status_code, accounts); +} + +const std::string A = "05" + std::string(64, 'a'); +const std::string B = "03" + std::string(64, 'b'); +} // namespace + +TEST_CASE("Swarm", "[network][swarm][batch_error]") { + CHECK(response::find_uniform_batch_error(batch({{421, {}}, {421, {}}}).dump()) == 421); + CHECK_FALSE(response::find_uniform_batch_error(batch({{421, {}}, {406, {}}}).dump())); + CHECK_FALSE(response::find_uniform_batch_error(batch({{421, {}}, {200, {}}}).dump())); + CHECK_FALSE(response::find_uniform_batch_error(batch({}).dump())); + CHECK_FALSE(response::find_uniform_batch_error(rejection(A, "12").dump())); + CHECK_FALSE(response::find_uniform_batch_error("not json")); +} + +TEST_CASE("Swarm", "[network][swarm][rejections]") { + SECTION("A plain 421 is attributed to the request's account") { + auto r = rejections(rejection(A, "123"), 421, {account('a')}); + CHECK_FALSE(r.unattributed); + REQUIRE(r.accounts.size() == 1); + CHECK(r.accounts.at(account('a')) == nodes("123")); + } + + SECTION("An account echoed in testnet form, without a prefix, still matches") { + auto r = rejections(rejection(std::string(64, 'a'), "12"), 421, {account('a')}); + REQUIRE(r.accounts.count(account('a'))); + CHECK(r.accounts.at(account('a')) == nodes("12")); + } + + SECTION("A 421 needn't name its account") { + auto r = rejections( + batch({{421, rejection(std::nullopt, "12")}, {421, rejection(std::nullopt, "34")}}), + 421, + {account('a'), account('b')}); + CHECK_FALSE(r.unattributed); + REQUIRE(r.accounts.size() == 2); + CHECK(r.accounts.at(account('a')) == nodes("12")); + CHECK(r.accounts.at(account('b')) == nodes("34")); + } + + SECTION("Nothing is found in a response that isn't a 421") { + auto r = rejections(json{{"t", 1}}, 200, {account('a')}); + CHECK(r.accounts.empty()); + CHECK_FALSE(r.unattributed); + } + + SECTION("A batch can reject some accounts and answer others") { + auto r = rejections( + batch({{421, rejection(A, "12")}, + {200, json{{"messages", json::array()}}}, + {421, rejection(B, "34")}}), + 200, + {account('a'), account('c'), account('b')}); + CHECK_FALSE(r.unattributed); + REQUIRE(r.accounts.size() == 2); + CHECK(r.accounts.at(account('a')) == nodes("12")); + CHECK(r.accounts.at(account('b')) == nodes("34")); + } + + SECTION("Several rejections of one account are one redirect") { + auto r = rejections( + batch({{421, rejection(A, "12")}, {421, rejection(A, "12")}}), 421, {account('a')}); + REQUIRE(r.accounts.size() == 1); + CHECK(r.accounts.at(account('a')) == nodes("12")); + } + + SECTION("A single account applies to every sub-request") { + auto r = rejections( + batch({{421, rejection(std::nullopt, "12")}, {421, rejection(std::nullopt, "12")}}), + 421, + {account('a')}); + REQUIRE(r.accounts.size() == 1); + CHECK(r.accounts.at(account('a')) == nodes("12")); + } + + SECTION("A sequence that stopped early is still attributed by position") { + auto r = rejections( + batch({{200, json::object()}, {421, rejection(std::nullopt, "34")}}), + 200, + {account('a'), account('b'), account('c')}); + REQUIRE(r.accounts.size() == 1); + CHECK(r.accounts.at(account('b')) == nodes("34")); + } + + // The account a 421 names is the responding node's word; taking it would let any node redirect + // any account it likes, not just those it was asked about + SECTION("The account a 421 names is never trusted in place of the request's") { + auto r = rejections(rejection(A, "12"), 421, {account('b')}); + CHECK(r.accounts.empty()); + CHECK(r.unattributed); + + r = rejections( + batch({{421, rejection("05" + std::string(64, 'c'), "12")}, + {421, rejection(B, "34")}}), + 421, + {account('a'), account('b')}); + CHECK(r.unattributed); + REQUIRE(r.accounts.size() == 1); + CHECK(r.accounts.at(account('b')) == nodes("34")); + + CHECK(rejections(rejection(A, "12"), 421).accounts.empty()); + } + + SECTION("A sub-request with no account leaves its 421 unattributed") { + auto r = rejections( + batch({{200, json::object()}, {421, rejection(std::nullopt, "12")}}), + 200, + {account('a'), std::nullopt}); + CHECK(r.accounts.empty()); + CHECK(r.unattributed); + } + + SECTION("A 421 whose body isn't JSON has no redirect") { + std::vector> accounts{account('a')}; + auto r = response::find_swarm_rejections(nullptr, 421, accounts); + REQUIRE(r.accounts.size() == 1); + CHECK(r.accounts.at(account('a')).empty()); + CHECK(response::find_swarm_rejections(nullptr, 200, accounts).accounts.empty()); + } +} + +TEST_CASE("Swarm", "[network][swarm][batch_request_accounts]") { + SECTION("A pregenerated batch body gives each sub-request's account") { + auto body = json{{"requests", + {{{"method", "retrieve"}, {"params", {{"pubkey", A}}}}, + {{"method", "info"}, {"params", json::object()}}, + {{"method", "retrieve"}, {"params", {{"pubkey", B}}}}}}} + .dump(); + auto bytes = std::span{reinterpret_cast(body.data()), body.size()}; + + auto accounts = batch_request_accounts("batch", bytes); + REQUIRE(accounts); + CHECK(*accounts == + std::vector>{account('a'), std::nullopt, account('b')}); + CHECK(batch_request_accounts("sequence", bytes) == accounts); + CHECK_FALSE(batch_request_accounts("retrieve", bytes)); + + // A sub-request naming no account takes the one the caller gave for the whole request, + // while those that name their own keep it + CHECK(batch_request_accounts("batch", bytes, account('c')) == + std::vector>{account('a'), account('c'), account('b')}); + + std::string_view junk = "not json"; + CHECK_FALSE(batch_request_accounts( + "batch", {reinterpret_cast(junk.data()), junk.size()})); + } +} diff --git a/tests/test_onion_request_router.cpp b/tests/test_onion_request_router.cpp index 8913315d8..20ac44c7c 100644 --- a/tests/test_onion_request_router.cpp +++ b/tests/test_onion_request_router.cpp @@ -25,7 +25,7 @@ class TestOnionRequestRouter { std::shared_ptr router, PathCategory category, std::vector paths) { - router->_paths.emplace(category, paths); + router->_paths[category] = std::move(paths); } static std::vector get_paths( @@ -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, @@ -64,6 +78,21 @@ class TestOnionRequestRouter { return router->_find_valid_path(request); } + static void send_on_path( + std::shared_ptr router, + OnionPath& path, + Request request, + network_response_callback_t callback) { + router->_send_on_path(path, std::move(request), std::move(callback)); + } + + static void rotate_path( + std::shared_ptr router, + const std::string& path_id, + PathCategory category) { + router->_rotate_path(path_id, category); + } + static void handle_transport_response( std::shared_ptr router, std::string path_id, @@ -152,11 +181,21 @@ namespace { SnodePool::record_node_failure(key, permanent); } + // When set, `refresh_if_needed` calls back with it straight away - as the real one does, on + // the loop thread, when nothing needs refreshing - instead of never calling back. Only the + // first `max_refresh_callbacks` calls do, so code that asks again on every callback stops + // there instead of recursing until the stack gives out. + std::optional refresh_result; + int max_refresh_callbacks = 20; + 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) + // Don't trigger a real cache refresh + if (refresh_result && on_refresh_complete && + get_call_count("refresh_if_needed") <= max_refresh_callbacks) + on_refresh_complete(*refresh_result); } void get_swarm( @@ -168,8 +207,12 @@ namespace { // Do nothing (don't want to trigger a cache refresh) } + // The `count` of each `get_unused_nodes` call, which the mock result doesn't reflect + std::vector unused_node_counts; + std::vector get_unused_nodes( size_t count, const std::vector& exclude = {}) override { + unused_node_counts.push_back(count); if (check_should_ignore_and_log_call("get_unused_nodes")) return {}; @@ -204,8 +247,19 @@ namespace { func_called("remove_failure_listeners"); } - void send_request(Request /*request*/, network_response_callback_t /*callback*/) override { + // Answered the way a node on the path does when it fails a request itself: a status of its + // own and a plaintext body, which is nothing the destination encrypted + std::optional> plaintext_reply; + + void send_request(Request /*request*/, network_response_callback_t callback) override { func_called("send_request"); + if (plaintext_reply) + callback( + false, + false, + plaintext_reply->first, + {content_type_plain_text}, + plaintext_reply->second); } }; @@ -216,6 +270,60 @@ namespace { std::vector> headers; std::optional response; }; + + config::SnodePool test_pool_config() { + return {.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 = 3}; + } + + config::OnionRequestRouter test_router_config() { + return {.file_server_config = file_server::DEFAULT_CONFIG, + .cache_directory = std::nullopt, + .edge_node_cache_duration = std::chrono::days{10}, + .netid = opt::netid::Target::testnet, + .seed_nodes = {}, + .retry_delay = network::opt::retry_delay{50ms, 200ms}, + .path_length = 3, + .path_strike_threshold = 3, + .path_build_retry_limit = 10, + .path_rotation_frequency = 10min, + .node_strike_threshold = 3, + .disable_pre_build_paths = true, + .single_path_mode = true, + .min_path_counts = {{PathCategory::standard, 1}}}; + } + + service_node test_node(std::string_view pk_hex, uint16_t port) { + return service_node{ + ed25519_pubkey::from_hex(pk_hex), + oxen::quic::ipv4{"127.0.0.1"}, + port, + static_cast(port + 10000), + {2, 11, 0}, + 0}; + } + + network_response_callback_t record_into(std::optional& result) { + return [&result]( + bool success, + bool timeout, + int16_t status_code, + std::vector> headers, + std::optional response) { + result = {success, timeout, status_code, headers, response}; + }; + } } // namespace TEST_CASE("Network", "[network][onion_request_router][handle_errors]") { @@ -832,4 +940,228 @@ 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); +} + +TEST_CASE("Network", "[network][onion_request_router][undecrypted_response]") { + auto target = + test_node("4cb76fdc6d32278e3f83dbf608360ecc6b65727934b85d2fb86862ff98c46ab7", 20001); + auto path_nodes = std::vector{ + test_node("5ea34e72bb044654a6a23675690ef5ffaaf1656b02f93fb76655f9cbdbe89876", 20002), + test_node("e17a692033200ae41350df9709754edde7343e2cf2f23e88f993319e0720e5e5", 20003), + test_node("7b633fa6fb462b90db6f0f50384190ce7715e31b7aa93d87dbd7e94e33d4251f", 20004)}; + + auto loop = std::make_shared(); + auto disk_loop = std::make_shared(); + auto snode_pool = std::make_shared(test_pool_config(), loop, disk_loop); + auto transport = std::make_shared(); + auto router = std::make_shared( + test_router_config(), loop, disk_loop, snode_pool, transport); + + // The router is driven on the loop, as its own code is, and checked from here + loop->call_get([&] { + TestOnionRequestRouter::set_paths( + router, PathCategory::standard, {OnionPath{"Test", path_nodes}}); + }); + + auto send_with_reply = [&](int16_t status) { + transport->plaintext_reply.emplace(status, "Forged by a node on the path"); + auto request = Request{ + "AAAA", target, "retrieve", to_vector("{}"), RequestCategory::standard, 10s}; + + bool have_path = false; + std::optional result; + loop->call_get([&] { + if (auto* path = TestOnionRequestRouter::find_valid_path(router, request)) { + have_path = true; + TestOnionRequestRouter::send_on_path(router, *path, request, record_into(result)); + } + }); + REQUIRE(have_path); + REQUIRE(result); + return *result; + }; + + // Each of these is acted on as the destination's verdict (a swarm redirect, a clock resync), + // which only the destination can give; a node on the path sending one mustn't pass for it + for (int16_t status : {ERROR_MISDIRECTED_REQUEST, ERROR_NOT_ACCEPTABLE, ERROR_TOO_EARLY}) { + CAPTURE(status); + auto result = send_with_reply(status); + CHECK_FALSE(result.success); + CHECK(result.status_code == ERROR_UNAUTHENTICATED_RESPONSE); + CHECK(result.response.value_or("").find("status {}"_format(status)) != std::string::npos); + } + + // Anything else a node on the path says about its own failure is reported as it was + auto result = send_with_reply(502); + CHECK_FALSE(result.success); + CHECK(result.status_code == 502); +} + +TEST_CASE("Network", "[network][onion_request_router][build_path_too_few_nodes]") { + auto loop = std::make_shared(); + auto disk_loop = std::make_shared(); + auto snode_pool = std::make_shared(test_pool_config(), loop, disk_loop); + auto transport = std::make_shared(); + auto router = std::make_shared( + test_router_config(), loop, disk_loop, snode_pool, transport); + + // Run on the loop, as the router's own code is: that also puts it after the router's setup, + // which asks for the (empty) pool to be refreshed and mustn't be counted below + std::optional result; + loop->call_get([&] { + snode_pool->reset_calls(); + + auto queue = std::make_shared(loop); + TestOnionRequestRouter::set_request_queues(router, {{PathCategory::standard, queue}}); + queue->add( + Request{"AAAA", + test_node( + "4cb76fdc6d32278e3f83dbf608360ecc6b65727934b85d2fb86862ff98c46ab7", + 20001), + "retrieve", + to_vector("{}"), + RequestCategory::standard, + 10s}, + record_into(result)); + + // A path needs three nodes and the pool can only offer one, while saying nothing needs + // refreshing - which is a pool too concentrated in a few subnets to fill a path, say + snode_pool->mock_unused_nodes = std::vector{test_node( + "5ea34e72bb044654a6a23675690ef5ffaaf1656b02f93fb76655f9cbdbe89876", 20002)}; + snode_pool->refresh_result = true; + + TestOnionRequestRouter::build_path(router, PathCategory::standard, "AAAA"); + }); + + // Asking for a refresh once and then giving up; without checking that the pool can now fill a + // path, every "refreshed" rebuilt straight into another refresh request, and only the mock's + // limit stops it + CHECK(snode_pool->get_call_count("refresh_if_needed") == 1); + REQUIRE(result); + CHECK_FALSE(result->success); + CHECK(result->status_code == ERROR_INSUFFICIENT_NODES); + CHECK(result->response.value_or("").find("too few usable nodes") != std::string::npos); +} + +TEST_CASE("Network", "[network][onion_request_router][rotate_path]") { + auto key1 = "4cb76fdc6d32278e3f83dbf608360ecc6b65727934b85d2fb86862ff98c46ab7"sv; + auto key2 = "5ea34e72bb044654a6a23675690ef5ffaaf1656b02f93fb76655f9cbdbe89876"sv; + auto key3 = "e17a692033200ae41350df9709754edde7343e2cf2f23e88f993319e0720e5e5"sv; + auto key4 = "7b633fa6fb462b90db6f0f50384190ce7715e31b7aa93d87dbd7e94e33d4251f"sv; + auto edge = test_node(key1, 20001); + auto path_nodes = std::vector{edge, test_node(key2, 20002), test_node(key3, 20003)}; + + auto loop = std::make_shared(); + auto disk_loop = std::make_shared(); + auto snode_pool = std::make_shared(test_pool_config(), loop, disk_loop); + auto transport = std::make_shared(); + auto config = test_router_config(); + auto router = + std::make_shared(config, loop, disk_loop, snode_pool, transport); + snode_pool->mock_unused_nodes = + std::vector{test_node(key2, 20012), test_node(key3, 20013), test_node(key4, 20014)}; + + // Rotates a path whose edge node was connected just now, so well inside + // `edge_node_cache_duration`, and returns what the rotation asked the pool for. The first + // request says which it chose: a whole new path, or the rest of one that keeps the edge node. + auto rotate = [&] { + std::vector counts; + loop->call_get([&] { + auto now = std::chrono::system_clock::now(); + TestOnionRequestRouter::set_paths( + router, PathCategory::standard, {OnionPath{"Test", path_nodes, now, now}}); + snode_pool->unused_node_counts.clear(); + TestOnionRequestRouter::rotate_path(router, "Test", PathCategory::standard); + counts = snode_pool->unused_node_counts; + }); + return counts; + }; + + auto kept = rotate(); + REQUIRE_FALSE(kept.empty()); + CHECK(kept.front() == config.path_length - 1u); + + // Struck out since we connected to it, the edge node is replaced along with the rest + snode_pool->record_node_failure(edge, true); + auto replaced = rotate(); + REQUIRE_FALSE(replaced.empty()); + CHECK(replaced.front() == config.path_length); +} } // namespace session::network diff --git a/tests/test_quic_transport.cpp b/tests/test_quic_transport.cpp new file mode 100644 index 000000000..52464e236 --- /dev/null +++ b/tests/test_quic_transport.cpp @@ -0,0 +1,80 @@ +#include +#include +#include +#include + +using namespace session::network; +using namespace std::literals; + +namespace { +std::shared_ptr make_transport(std::shared_ptr loop) { + return std::make_shared( + config::QuicTransport{ + .handshake_timeout = 1s, .keep_alive = 10s, .disable_mtu_discovery = false}, + std::move(loop)); +} + +service_node test_node() { + auto pubkey = "4cb76fdc6d32278e3f83dbf608360ecc6b65727934b85d2fb86862ff98c46ab7"sv; + return service_node{ + ed25519_pubkey::from_hex(pubkey), + oxen::quic::ipv4{"127.0.0.1"}, + 20001, + 30001, + {2, 11, 0}, + 0}; +} +} // namespace + +TEST_CASE("Network", "[network][quic_transport][suspend]") { + auto loop = std::make_shared(); + auto transport = make_transport(loop); + transport->suspend(); + + std::promise status; + transport->send_request( + Request{"AAAA", test_node(), "info", std::nullopt, RequestCategory::standard, 1s}, + [&status, answered = false](bool, bool, int16_t status_code, auto, auto) mutable { + if (!std::exchange(answered, true)) + status.set_value(status_code); + }); + + // Refused outright rather than attempted: a suspended transport sends nothing + auto answer = status.get_future(); + REQUIRE(answer.wait_for(5s) == std::future_status::ready); + CHECK(answer.get() == ERROR_NETWORK_SUSPENDED); +} + +TEST_CASE("Network", "[network][quic_transport][close_connections]") { + auto loop = std::make_shared(); + auto transport = make_transport(loop); + + // Verified and closed within one loop job, so nothing from the network can settle the + // verification before the close does + int answers = 0; + bool success = true; + std::optional error_code; + loop->call_get([&] { + transport->verify_connectivity( + test_node(), + 3s, + "AAAA", + RequestCategory::standard, + [&](bool s, std::optional e) { + ++answers; + success = s; + error_code = e; + }); + transport->close_connections(); + }); + + // Anything the close deferred has run by the time a later job does + loop->call_get([] {}); + + // One answer, from the close, and with no error code: a close we asked for is no evidence + // against the node, which is what an error code would say - the router strikes a node whose + // verification fails with one + CHECK(answers == 1); + CHECK_FALSE(success); + CHECK_FALSE(error_code); +} diff --git a/tests/test_snode_pool.cpp b/tests/test_snode_pool.cpp index d470ed193..1a9eeb4b2 100644 --- a/tests/test_snode_pool.cpp +++ b/tests/test_snode_pool.cpp @@ -1,5 +1,6 @@ #include #include +#include #include "utils.hpp" @@ -30,13 +31,18 @@ class TestSnodePool : public SnodePool { }); } + // How often the pool was asked to check whether it needs refreshing; read it only after a + // loop job has run, since the checks are scheduled onto the loop + int refresh_checks = 0; + void refresh_if_needed( const std::vector& /*in_use_nodes*/, - std::function /*on_refresh_complete*/ = nullptr) override { - // Do nothing (don't want to trigger a cache refresh) + refresh_callback_t /*on_refresh_complete*/ = nullptr) override { + // Don't trigger a cache refresh + ++refresh_checks; } - void debug_queue_post_refresh_callback(std::function 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)); }); @@ -69,6 +75,86 @@ class TestSnodePool : public SnodePool { // loop thread void update_cache(std::vector nodes) { _update_cache("test", std::move(nodes)); } + bool debug_refresh_in_progress() { + return _loop->call_get([this] { return _current_snode_cache_refresh_id.has_value(); }); + } + + void debug_start_refresh() { + _loop->call_get([this] { _refresh_snode_cache(); }); + } + + std::optional debug_refresh_id() { + return _loop->call_get([this] { return _current_snode_cache_refresh_id; }); + } + + size_t debug_refresh_candidate_count() { + return _loop->call_get([this] { return _refresh_candidate_nodes.size(); }); + } + + void debug_clear_refresh_candidates() { + _loop->call_get([this] { _refresh_candidate_nodes.clear(); }); + } + + int debug_refresh_failure_count() { + return _loop->call_get([this] { return _snode_cache_refresh_failure_count; }); + } + + void debug_set_refresh_failure_count(int count) { + _loop->call_get([this, count] { _snode_cache_refresh_failure_count = count; }); + } + + // Launches one of the refresh `request_id`'s fetches, as its retries do + void debug_launch_refresh_request(const std::string& request_id) { + _loop->call_get([this, request_id] { + _launch_next_refresh_request(request_id, 0, false, true, 1); + }); + } + + // Backdates the pool snapshot so the age-based policies can be exercised without waiting + void debug_age_pool(std::chrono::seconds by) { + _loop->call_get([this, by] { _last_snode_cache_update -= by; }); + } + + void debug_age_evidence_refresh(std::chrono::seconds by) { + _loop->call_get([this, by] { _last_evidence_refresh -= by; }); + } + + std::chrono::seconds debug_evidence_backoff() { + return _loop->call_get([this] { return _evidence_refresh_backoff; }); + } + + // Runs `fn` on the loop thread, which is what makes `get_swarm` / `invalidate_swarm` resolve + // inline rather than being queued: a test observing them from off the loop has no ordering + // guarantee at all + void debug_run_on_loop(std::function fn) { + _loop->call_get([fn = std::move(fn)] { fn(); }); + } + + // 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()); + }); + } + + // 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_on_refresh_complete(std::vector> raw_results) { auto total_requests = static_cast(raw_results.size()); _loop->call_get([&] { @@ -77,6 +163,19 @@ class TestSnodePool : public SnodePool { } }; +// `TestSnodePool` stubs out `refresh_if_needed` so tests don't kick off real refreshes; the swarm +// invalidation test needs the real age-based policy to compare against, so it uses this instead +class TestSnodePoolAgePolicy : public TestSnodePool { + public: + using TestSnodePool::TestSnodePool; + + void refresh_if_needed( + const std::vector& in_use_nodes, + refresh_callback_t on_refresh_complete = nullptr) override { + SnodePool::refresh_if_needed(in_use_nodes, std::move(on_refresh_complete)); + } +}; + // Encodes nodes the way the storage server returns them, so they can be fed to // `_on_refresh_complete`: 51 bytes per node, all multi-byte fields big-endian std::vector to_snode_cache_bin(const std::vector& nodes) { @@ -107,18 +206,12 @@ std::vector to_snode_cache_bin(const std::vector& nodes TEST_CASE("Network", "[network][get_unused_nodes]") { session::network::config::SnodePool pool_config = { - std::nullopt, - std::nullopt, - std::chrono::minutes{5}, - std::chrono::minutes{5}, - false, // enforce_subnet_diversity - network::opt::retry_delay{50ms, 200ms}, - opt::netid::Target::testnet, - {}, - 0, - 0, - 3, // cache_node_strike_threshold - false}; + .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, + .cache_node_strike_threshold = 3}; auto ed_pk = "4cb76fdc6d32278e3f83dbf608360ecc6b65727934b85d2fb86862ff98c46ab7"_hexbytes; auto ed_pk2 = "5ea34e72bb044654a6a23675690ef5ffaaf1656b02f93fb76655f9cbdbe89876"_hexbytes; auto ed_pk3 = "e17a692033200ae41350df9709754edde7343e2cf2f23e88f993319e0720e5e5"_hexbytes; @@ -203,20 +296,31 @@ TEST_CASE("Network", "[network][get_unused_nodes]") { std::sort(unused_nodes.begin(), unused_nodes.end()); CHECK(unused_nodes == remaining); + // ... but only once they have passed it: ordinary strikes below the threshold leave a node in + // use (strikes are per pubkey, so these apply to every node sharing the struck node's key) + snode_pool->reset_state_with_cache(snode_cache); + auto struck_key = snode_cache[0].remote_pubkey; + auto uses_struck_key = [&struck_key](const service_node& n) { + return n.remote_pubkey == struck_key; + }; + auto with_struck_key = std::ranges::count_if(snode_cache, uses_struck_key); + + snode_pool->record_node_failure(snode_cache[0]); + snode_pool->record_node_failure(snode_cache[0]); + CHECK(std::ranges::count_if(snode_pool->get_unused_nodes(20), uses_struck_key) == + with_struck_key); + + snode_pool->record_node_failure(snode_cache[0]); + CHECK(std::ranges::none_of(snode_pool->get_unused_nodes(20), uses_struck_key)); + // Should exclude nodes which have the same subnet pool_config = { - std::nullopt, - std::nullopt, - std::chrono::minutes{5}, - std::chrono::minutes{5}, - true, // enforce_subnet_diversity - network::opt::retry_delay{50ms, 200ms}, - opt::netid::Target::testnet, - {}, - 0, - 0, - 3, // cache_node_strike_threshold - false}; + .cache_expiration = std::chrono::minutes{5}, + .cache_min_lifetime = std::chrono::minutes{5}, + .enforce_subnet_diversity = true, + .retry_delay = network::opt::retry_delay{50ms, 200ms}, + .netid = opt::netid::Target::testnet, + .cache_node_strike_threshold = 3}; snode_pool = std::make_shared(pool_config, loop, disk_loop); snode_pool->reset_state_with_cache(snode_cache); unused_nodes = snode_pool->get_unused_nodes(20); @@ -231,18 +335,12 @@ TEST_CASE("Network", "[network][get_unused_nodes]") { TEST_CASE("Network", "[network][update_cache]") { session::network::config::SnodePool pool_config = { - std::nullopt, - std::nullopt, - 5min, - 5min, - false, // enforce_subnet_diversity - network::opt::retry_delay{50ms, 200ms}, - opt::netid::Target::testnet, - {}, - 0, - 0, - 3, // cache_node_strike_threshold - false}; + .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_node_strike_threshold = 3}; auto ed_pk = "4cb76fdc6d32278e3f83dbf608360ecc6b65727934b85d2fb86862ff98c46ab7"_hexbytes; std::vector snode_cache; @@ -263,12 +361,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}); @@ -286,19 +384,13 @@ TEST_CASE("Network", "[network][update_cache]") { TEST_CASE("Network", "[network][refresh_min_cache_size]") { session::network::config::SnodePool pool_config = { - std::nullopt, - std::nullopt, - 5min, - 5min, - false, // enforce_subnet_diversity - network::opt::retry_delay{50ms, 200ms}, - opt::netid::Target::testnet, - {}, - 12, // cache_min_size - 0, - 0, - 3, // cache_node_strike_threshold - false}; + .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 = 12, + .cache_node_strike_threshold = 3}; auto ed_pk = "4cb76fdc6d32278e3f83dbf608360ecc6b65727934b85d2fb86862ff98c46ab7"_hexbytes; std::vector snode_cache; @@ -336,3 +428,531 @@ 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][invalidate_swarm]") { + 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, + .cache_min_size = 12, + .cache_min_swarm_size = 3, + .cache_num_nodes_to_use_for_refresh = 1, + .cache_min_num_refresh_presence_to_include_node = 1, + .cache_node_strike_threshold = 3}; + + auto ed_pk_before = "4cb76fdc6d32278e3f83dbf608360ecc6b65727934b85d2fb86862ff98c46ab7"_hexbytes; + auto ed_pk_after = "5ea34e72bb044654a6a23675690ef5ffaaf1656b02f93fb76655f9cbdbe89876"_hexbytes; + + // Two snapshots of the same two swarms, with an entirely different set of nodes serving them. + // Which swarm the pubkey lands in doesn't matter; what matters is that the correct answer + // changed and no node is common to both. + auto pool_snapshot = [](const std::vector& ed_pk, uint8_t subnet) { + std::vector nodes; + + for (uint16_t i = 0; i < 12; ++i) + nodes.emplace_back(service_node{ + ed25519_pubkey::from_bytes(ed_pk), + oxen::quic::ipv4{"192.168.{}.{}"_format(subnet, i)}, + static_cast(20000 + i), + static_cast(30000 + i), + {2, 11, 0}, + static_cast(i < 6 ? 0 : 1)}); + + return nodes; + }; + auto nodes_before = pool_snapshot(ed_pk_before, 0); + auto nodes_after = pool_snapshot(ed_pk_after, 1); + + auto loop = std::make_shared(); + auto disk_loop = std::make_shared(); + auto snode_pool = std::make_shared(pool_config, loop, disk_loop); + snode_pool->update_cache(nodes_before); + snode_pool->debug_age_pool(10min); + + auto swarm_pubkey = x25519_pubkey::from_hex( + "0000000000000000000000000000000000000000000000000000000000000000"); + + std::vector rejected_swarm, recovered_swarm; + bool age_policy_started_refresh = false, recovery_ran = false; + + snode_pool->debug_run_on_loop([&] { + snode_pool->get_swarm( + swarm_pubkey, true, [&](swarm::swarm_id_t, std::vector nodes) { + rejected_swarm = std::move(nodes); + }); + + // Asking by age is what the 421 path used to do, and on a cache this far short of + // `cache_expiration` it declines - which is what left the disproven mapping in place + snode_pool->refresh_if_needed({}); + age_policy_started_refresh = snode_pool->debug_refresh_in_progress(); + + snode_pool->invalidate_swarm(swarm_pubkey, [&](bool refreshed) { + recovery_ran = true; + CHECK(refreshed); + snode_pool->get_swarm( + swarm_pubkey, true, [&](swarm::swarm_id_t, std::vector nodes) { + recovered_swarm = std::move(nodes); + }); + }); + }); + + REQUIRE_FALSE(rejected_swarm.empty()); + CHECK_FALSE(age_policy_started_refresh); + + // The rejection has to produce a refresh, and the retry has to wait for it rather than being + // handed the answer that was just rejected + REQUIRE(snode_pool->debug_refresh_in_progress()); + CHECK_FALSE(recovery_ran); + + snode_pool->update_cache(nodes_after); + REQUIRE(recovery_ran); + REQUIRE_FALSE(recovered_swarm.empty()); + + for (const auto& node : recovered_swarm) + CHECK(std::ranges::find(rejected_swarm, node) == rejected_swarm.end()); + + // A rejection that arrives after we already refreshed on one has to be held off, or a node that + // rejects everything keeps a client refreshing from three nodes forever + CHECK(snode_pool->debug_evidence_backoff() == 1min); + snode_pool->debug_age_pool(10min); + snode_pool->debug_run_on_loop([&] { snode_pool->invalidate_swarm(swarm_pubkey); }); + CHECK_FALSE(snode_pool->debug_refresh_in_progress()); + + // Once it has elapsed the next one runs, and the one after it waits twice as long again + snode_pool->debug_age_evidence_refresh(90s); + snode_pool->debug_run_on_loop([&] { snode_pool->invalidate_swarm(swarm_pubkey); }); + CHECK(snode_pool->debug_refresh_in_progress()); + CHECK(snode_pool->debug_evidence_backoff() == 2min); + + // Staying quiet for twice the interval means the next incident starts from the base delay + // rather than inheriting an escalated one + snode_pool->update_cache(nodes_after); + snode_pool->debug_age_pool(10min); + snode_pool->debug_age_evidence_refresh(10min); + snode_pool->debug_run_on_loop([&] { snode_pool->invalidate_swarm(swarm_pubkey); }); + CHECK(snode_pool->debug_refresh_in_progress()); + CHECK(snode_pool->debug_evidence_backoff() == 1min); +} + +TEST_CASE("Network", "[network][swarm_redirect]") { + 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, + .cache_min_size = 12, + .cache_min_swarm_size = 3, + .cache_num_nodes_to_use_for_refresh = 1, + .cache_min_num_refresh_presence_to_include_node = 1, + .cache_node_strike_threshold = 3}; + + // A redirect is resolved by node pubkey, so every node needs its own + auto key_for = [](uint16_t i) { + return ed25519_pubkey::from_hex( + "4cb76fdc6d32278e3f83dbf608360ecc6b65727934b85d2fb86862ff98c46a{:02x}"_format(i)); + }; + std::vector snode_cache; + + for (uint16_t i = 0; i < 12; ++i) + snode_cache.emplace_back(service_node{ + key_for(i), + oxen::quic::ipv4{"192.168.0.{}"_format(i)}, + static_cast(20000 + i), + static_cast(30000 + i), + {2, 11, 0}, + static_cast(i < 6 ? 0 : 1)}); + + auto loop = std::make_shared(); + auto disk_loop = std::make_shared(); + auto snode_pool = std::make_shared(pool_config, loop, disk_loop); + snode_pool->update_cache(snode_cache); + + // Old enough that the fallback would refresh, so "no refresh happened" below means the redirect + // was taken rather than the pool being too fresh to bother + snode_pool->debug_age_pool(10min); + + auto swarm_pubkey = x25519_pubkey::from_hex( + "0000000000000000000000000000000000000000000000000000000000000000"); + + std::vector calculated; + snode_pool->debug_run_on_loop([&] { + snode_pool->get_swarm( + swarm_pubkey, true, [&](swarm::swarm_id_t, std::vector nodes) { + calculated = std::move(nodes); + }); + }); + REQUIRE(calculated.size() == 6); + + // Everything the calculation didn't pick, which is what a real redirect would be naming + std::vector elsewhere; + for (const auto& node : snode_cache) + if (std::ranges::find(calculated, node) == calculated.end()) + elsewhere.push_back(node); + REQUIRE(elsewhere.size() == 6); + + // Membership is the claim in every comparison below; the pool is reshuffled on each refresh and + // a redirect is resolved in pool order, so neither side has an order worth asserting + auto sorted = [](std::vector nodes) { + std::ranges::sort(nodes); + return nodes; + }; + auto keys_of = [](const std::vector& nodes, size_t count) { + std::vector keys; + for (size_t i = 0; i < count && i < nodes.size(); ++i) + keys.push_back(nodes[i].remote_pubkey); + return keys; + }; + + // Nodes we've never heard of can't redirect us anywhere - this is what stops a response putting + // us in touch with something that isn't a registered service node + CHECK_FALSE(snode_pool->record_swarm_redirect(swarm_pubkey, {key_for(200), key_for(201)})); + + auto current_swarm = [&] { + std::vector swarm; + snode_pool->debug_run_on_loop([&] { + snode_pool->get_swarm( + swarm_pubkey, true, [&](swarm::swarm_id_t, std::vector nodes) { + swarm = std::move(nodes); + }); + }); + return sorted(std::move(swarm)); + }; + auto overridden = sorted(std::vector(elsewhere.begin(), elsewhere.begin() + 4)); + + // A usable redirect is taken, and - the point of all this - no pool refresh is started for it + REQUIRE(snode_pool->record_swarm_redirect(swarm_pubkey, keys_of(elsewhere, 4))); + CHECK_FALSE(snode_pool->debug_refresh_in_progress()); + CHECK(current_swarm() == overridden); + CHECK_FALSE(snode_pool->debug_refresh_in_progress()); + + // How the named swarm relates to the one we sent to says nothing, since swarms can be + // rearranged any way at all: a node that has left the overridden swarm names its current + // members, and one in it naming the calculated swarm is how a wrong redirect gets put right + CHECK(snode_pool->record_swarm_redirect(swarm_pubkey, keys_of(elsewhere, 4))); + CHECK(current_swarm() == overridden); + CHECK(snode_pool->record_swarm_redirect(swarm_pubkey, keys_of(calculated, 6))); + CHECK(current_swarm() == sorted(calculated)); + + // Two sets of nodes that disagree, each redirecting to the other + auto bounce = [&](int times) { + for (int i = 0; i < times; ++i) + REQUIRE(snode_pool->record_swarm_redirect( + swarm_pubkey, i % 2 == 0 ? keys_of(elsewhere, 4) : keys_of(calculated, 6))); + }; + + // Every redirect counts, but none is refused for it - refusing would only fail the request. + // Up to the limit nothing else happens... + bounce(5); + CHECK(current_swarm() == overridden); + CHECK_FALSE(snode_pool->debug_refresh_in_progress()); + + // ... and past it the redirect is still followed, but also asks for a refresh, which the pool + // being old enough lets start + CHECK(snode_pool->record_swarm_redirect(swarm_pubkey, keys_of(calculated, 6))); + CHECK(current_swarm() == sorted(calculated)); + CHECK(snode_pool->debug_refresh_in_progress()); + + // A refreshed pool is ground truth again, so redirects correcting the old one are dropped + REQUIRE(snode_pool->record_swarm_redirect(swarm_pubkey, keys_of(elsewhere, 4))); + snode_pool->update_cache(snode_cache); + CHECK(current_swarm() == sorted(calculated)); + + // ... and the calculated swarm itself, or part of it, is followed like any other + CHECK(snode_pool->record_swarm_redirect(swarm_pubkey, keys_of(calculated, 6))); + CHECK(current_swarm() == sorted(calculated)); + auto remaining = std::vector(calculated.begin() + 1, calculated.end()); + CHECK(snode_pool->record_swarm_redirect(swarm_pubkey, keys_of(remaining, 5))); + CHECK(current_swarm() == sorted(remaining)); + + // Past the limit with the refresh throttled - the pool was only just refreshed - redirects are + // still followed, rather than leaving the account on a swarm already known to be wrong + bounce(6); + CHECK(snode_pool->record_swarm_redirect(swarm_pubkey, keys_of(elsewhere, 4))); + CHECK(current_swarm() == overridden); + CHECK_FALSE(snode_pool->debug_refresh_in_progress()); + + // A redirect naming nodes we mostly haven't heard of is followed with the ones we have - they + // are the right swarm, where the one we had just rejected us - and asks for the refresh that + // can fill in the rest (from a fresh count, so the limit plays no part) + snode_pool->update_cache(snode_cache); + snode_pool->debug_age_pool(10min); + snode_pool->debug_age_evidence_refresh(10min); + auto partial = keys_of(elsewhere, 2); + partial.push_back(key_for(200)); + partial.push_back(key_for(201)); + CHECK(snode_pool->record_swarm_redirect(swarm_pubkey, partial)); + CHECK(current_swarm() == + sorted(std::vector(elsewhere.begin(), elsewhere.begin() + 2))); + CHECK(snode_pool->debug_refresh_in_progress()); + + // A redirect naming too few nodes to be a swarm asks for the refresh whether or not any of the + // names were unknown. One naming a single node we do know has no stale pool to explain it, and + // is the only thing serving the account until a refresh drops it, so it is the case that most + // wants one + snode_pool->update_cache(snode_cache); + snode_pool->debug_age_pool(10min); + snode_pool->debug_age_evidence_refresh(10min); + CHECK(snode_pool->record_swarm_redirect(swarm_pubkey, keys_of(elsewhere, 1))); + CHECK(current_swarm() == + sorted(std::vector(elsewhere.begin(), elsewhere.begin() + 1))); + CHECK(snode_pool->debug_refresh_in_progress()); +} + +TEST_CASE("Network", "[network][get_swarm_checks_age]") { + 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, + .cache_min_swarm_size = 3, + .cache_node_strike_threshold = 3}; + + std::vector snode_cache; + for (uint16_t i = 0; i < 6; ++i) + snode_cache.emplace_back(service_node{ + ed25519_pubkey::from_hex( + "4cb76fdc6d32278e3f83dbf608360ecc6b65727934b85d2fb86862ff98c46a{:02x}"_format( + i)), + oxen::quic::ipv4{"192.168.0.{}"_format(i)}, + static_cast(20000 + i), + static_cast(30000 + i), + {2, 11, 0}, + static_cast(i < 3 ? 0 : 1)}); + + auto loop = std::make_shared(); + auto disk_loop = std::make_shared(); + auto snode_pool = std::make_shared(pool_config, loop, disk_loop); + snode_pool->update_cache(snode_cache); + + auto swarm_pubkey = x25519_pubkey::from_hex( + "0000000000000000000000000000000000000000000000000000000000000000"); + + // The check is scheduled onto the loop, so it has run by the time a later loop job does + auto checks_for_one_get_swarm = [&] { + snode_pool->debug_run_on_loop([&] { + snode_pool->refresh_checks = 0; + snode_pool->get_swarm(swarm_pubkey, true, [](auto, auto) {}); + }); + snode_pool->debug_run_on_loop([] {}); + return snode_pool->refresh_checks; + }; + + // Worked out from the pool and cached... + CHECK(checks_for_one_get_swarm() == 1); + + // ... then answered from that cache, and from an override, each of which returns before the + // swarm is worked out, and neither of which may skip checking whether the pool is stale + CHECK(checks_for_one_get_swarm() == 1); + + std::vector redirect; + for (const auto& node : snode_cache) + redirect.push_back(node.remote_pubkey); + REQUIRE(snode_pool->record_swarm_redirect(swarm_pubkey, redirect)); + CHECK(checks_for_one_get_swarm() == 1); +} + +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); +} + +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); +} + +TEST_CASE("Network", "[network][suspend_refresh]") { + 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, + .cache_min_size = 1, + .cache_min_swarm_size = 1, + .cache_num_nodes_to_use_for_refresh = 1, + .cache_min_num_refresh_presence_to_include_node = 1, + .cache_node_strike_threshold = 3}; + + std::vector snode_cache; + for (uint16_t i = 0; i < 4; ++i) + snode_cache.emplace_back(service_node{ + ed25519_pubkey::from_hex( + "4cb76fdc6d32278e3f83dbf608360ecc6b65727934b85d2fb86862ff98c46a{:02x}"_format( + i)), + oxen::quic::ipv4{"192.168.0.{}"_format(i)}, + static_cast(20000 + i), + static_cast(30000 + i), + {2, 11, 0}, + 0}); + + // The default fetcher never answers, so a refresh stays in progress until the test ends it + auto loop = std::make_shared(); + auto disk_loop = std::make_shared(); + auto snode_pool = std::make_shared(pool_config, loop, disk_loop); + snode_pool->update_cache(snode_cache); + + SECTION("Suspending abandons a refresh in progress, and answers what waits on it") { + snode_pool->debug_start_refresh(); + REQUIRE(snode_pool->debug_refresh_in_progress()); + + std::optional answer; + snode_pool->debug_queue_post_refresh_callback( + [&answer](bool refreshed) { answer = refreshed; }); + snode_pool->debug_set_refresh_failure_count(3); + + snode_pool->suspend(); + CHECK(answer == false); + CHECK_FALSE(snode_pool->debug_refresh_in_progress()); + CHECK(snode_pool->debug_refresh_failure_count() == 0); + } + + SECTION("A fetch belonging to an abandoned refresh leaves the next refresh alone") { + snode_pool->debug_start_refresh(); + auto abandoned = snode_pool->debug_refresh_id(); + REQUIRE(abandoned); + snode_pool->suspend(); + snode_pool->resume(); + + snode_pool->debug_start_refresh(); + auto current = snode_pool->debug_refresh_id(); + REQUIRE(current); + REQUIRE(current != abandoned); + auto candidates = snode_pool->debug_refresh_candidate_count(); + + snode_pool->debug_launch_refresh_request(*abandoned); + CHECK(snode_pool->debug_refresh_id() == current); + CHECK(snode_pool->debug_refresh_candidate_count() == candidates); + } + + SECTION("An abandoned refresh's retry after running out of candidates restarts nothing") { + // Running out schedules a retry of the whole refresh, after the retry delay + auto run_out_then_suspend = [&] { + snode_pool->debug_start_refresh(); + auto id = snode_pool->debug_refresh_id(); + REQUIRE(id); + snode_pool->debug_clear_refresh_candidates(); + snode_pool->debug_launch_refresh_request(*id); + snode_pool->suspend(); + snode_pool->resume(); + }; + auto past_retry_delay = [] { std::this_thread::sleep_for(500ms); }; + + // Once it was the refresh that used to restart itself here + run_out_then_suspend(); + past_retry_delay(); + CHECK_FALSE(snode_pool->debug_refresh_in_progress()); + + // ... and with another refresh started by the time it fires, that one is left running + run_out_then_suspend(); + snode_pool->debug_start_refresh(); + auto current = snode_pool->debug_refresh_id(); + REQUIRE(current); + past_retry_delay(); + CHECK(snode_pool->debug_refresh_id() == current); + } +} diff --git a/tests/utils.hpp b/tests/utils.hpp index 047a3ba77..120637abf 100644 --- a/tests/utils.hpp +++ b/tests/utils.hpp @@ -5,7 +5,9 @@ #include #include +#include #include +#include #include #include #include