Snode pool: 421 recovery on evidence, strike expiry, and reporting failed refreshes

include/session/network/routing/onion_request_router.hpp

@@ -168,6 +168,7 @@ class OnionRequestRouter : public IRouter, public std::enable_shared_from_this<O
     // All of the below functions should only be called from within `_loop`
     void _finish_setup();
     void _pre_build_paths_if_needed();
+    void _drop_struck_cached_edge_nodes();
     void _close_connections();
     void _update_status();
     void _send_request_internal(Request request, network_response_callback_t callback);

include/session/network/session_network.hpp

@@ -115,8 +115,18 @@ class Network : public std::enable_shared_from_this<Network> {
     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<std::pair<std::string, std::string>> headers,
+            std::optional<std::string> body,
+            network_response_callback_t final_callback);
 
     void _resync_clock(
             std::optional<Request> original_request, network_response_callback_t request_callback);
@@ -125,7 +135,7 @@ class Network : public std::enable_shared_from_this<Network> {
             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);
 };

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

include/session/network/session_network_types.hpp

@@ -2,7 +2,9 @@
 
 #include <functional>
 #include <optional>
+#include <span>
 #include <string>
+#include <unordered_map>
 #include <vector>
 
 #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<session::network::x25519_pubkey> 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<std::optional<session::network::x25519_pubkey>> 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<void(
         std::vector<std::pair<std::string, std::string>> headers,
         std::optional<std::string> 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<std::vector<std::optional<x25519_pubkey>>> batch_request_accounts(
+        std::string_view endpoint,
+        std::span<const unsigned char> body,
+        std::optional<x25519_pubkey> fallback = std::nullopt);
+
 namespace response {
     std::optional<std::pair<int16_t, bool>> parse_text_error(std::string_view body);
     std::optional<int16_t> find_uniform_batch_error(std::string_view body);
+
+    /// As `find_uniform_batch_error`, for a body that has already been parsed.
+    std::optional<int16_t> 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<int16_t> 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<subresponse> 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<x25519_pubkey, std::vector<ed25519_pubkey>> 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<const std::optional<x25519_pubkey>> accounts);
 }  // namespace response
 
 struct OnionPathMetadata {

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<SnodePool> {
   public:
     using network_fetcher_t = std::function<void(Request, network_response_callback_t)>;
+
+    // `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<void(bool refreshed)>;
     using fetcher_connectivity_check_t = std::function<bool()>;
 
     SnodePool(
@@ -69,12 +75,38 @@ class SnodePool : public std::enable_shared_from_this<SnodePool> {
     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<service_node>& in_use_nodes,
-            std::function<void()> 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<ed25519_pubkey>& 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<SnodePool> {
     std::vector<std::pair<swarm::swarm_id_t, std::vector<service_node>>> _all_swarms;
     std::unordered_map<x25519_pubkey, std::pair<swarm::swarm_id_t, std::vector<service_node>>>
             _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<x25519_pubkey, std::pair<std::vector<service_node>, uint8_t>>
+            _swarm_overrides;
     std::map<ed25519_pubkey, std::vector<std::chrono::sys_seconds>> _snode_strikes;
     bool _strikes_flush_scheduled = false;
 
@@ -109,11 +148,18 @@ class SnodePool : public std::enable_shared_from_this<SnodePool> {
 
     // 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<std::string> _current_snode_cache_refresh_id;
     int _snode_cache_refresh_failure_count = 0;
     std::vector<service_node> _refresh_candidate_nodes;
     std::vector<std::vector<std::byte>> _snode_refresh_results;
-    std::vector<std::function<void()>> _after_snode_cache_refresh;
+    std::vector<refresh_callback_t> _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<SnodePool> {
             const bool use_direct_fetcher,
             const uint8_t total_requests);
     void _update_cache(std::string refresh_id, std::vector<service_node> nodes);
+    void _run_pending_refresh_callbacks(bool refreshed);
 };
 
 }  // namespace session::network