Port dev's swarm invalidation on 421 (#154) to client

include/session/network/routing/onion_request_router.hpp

@@ -183,6 +183,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/service_node.h

@@ -13,7 +13,7 @@ typedef struct network_service_node {
     uint16_t omq_port;
     uint16_t version[3];
     uint64_t swarm_id;
-    uint64_t requested_unlock_height;
+    int64_t requested_unlock_height;  // -1 if unknown; 0 if no unlock has been requested.
 } network_service_node;
 
 #ifdef __cplusplus

include/session/network/service_node.hpp

@@ -41,7 +41,10 @@ struct service_node {
     uint16_t omq_port;
     std::array<uint16_t, 3> storage_server_version;
     swarm_id_t swarm_id;
-    uint64_t requested_unlock_height;
+
+    // -1 when we don't know: most sources (the binary pool format, the on-disk cache, swarm
+    // responses) carry no unlock information at all, and 0 would claim they said "not requested".
+    int64_t requested_unlock_height = -1;
 
     oxen::quic::RemoteAddress to_https_address() const {
         return oxen::quic::RemoteAddress{remote_pubkey.view(), ip, https_port};

include/session/network/session_network.hpp

@@ -174,10 +174,21 @@ class Network {
     void _close_connections();
     void _recalculate_status();
     void _update_status(ConnectionStatus new_status);
-    void _update_network_state(const std::string& body);
-    // Writes the swarm a 421 reported into the cache.  Does not retry: choosing another member is
-    // the caller's, since only the caller can know which node it ended up talking to.
-    void _adopt_swarm_from_421(const x25519_pubkey& swarm_pubkey, std::string_view body);
+    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);
+
+    // Corrects the swarm a 421 disproved, then reports the 421.  Does not retry: choosing another
+    // member is the caller's, since only the caller can know which node it ended up talking to.
+    void _handle_421(
+            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);
@@ -186,7 +197,7 @@ class 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

@@ -67,7 +67,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

@@ -4,7 +4,9 @@
 #include <filesystem>
 #include <functional>
 #include <optional>
+#include <span>
 #include <string>
+#include <unordered_map>
 #include <vector>
 
 #include "session/attachments.hpp"
@@ -32,9 +34,14 @@ 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_FILE_SERVER_UNAVAILABLE = -10300;
+// A 421 whose swarm could not be corrected, so re-resolving would return the one that rejected us;
+// distinct from ERROR_MISDIRECTED_REQUEST, which means re-resolving now gives a better answer.
+constexpr int16_t ERROR_SWARM_UNRESOLVED = -10400;
 constexpr int16_t ERROR_UNKNOWN = -11000;
 
 const std::pair<std::string, std::string> content_type_plain_text = {
@@ -199,16 +206,17 @@ 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).
     ///
-    /// Set it to have a 421's correction applied: the swarm the storage server reports in the
-    /// rejection is written into the cache for *this account*, which cannot be derived from the
-    /// node we happened to ask.  Recovering from the 421 is still the caller's -- see
+    /// 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.  Network corrects the swarm but does not retry; that is the caller's -- see
     /// `Network::send_request`.
-    std::optional<session::network::x25519_pubkey> swarm_pubkey;
+    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.
@@ -321,9 +329,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 std::byte> 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
 
 /// One hop of a route, for showing a user where their traffic goes.  Identity and location, which

include/session/network/snode_pool.hpp

@@ -55,6 +55,12 @@ 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(
@@ -82,29 +88,44 @@ 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,
             bool ignore_strike_count,
             std::function<void(swarm::swarm_id_t, std::vector<service_node>)> callback);
 
-    /// Replaces the cached swarm for an account with one a storage server told us authoritatively,
-    /// as a 421 body does.
-    ///
-    /// Overrides the locally computed membership, which is only as fresh as the snode cache
-    /// (`cache_expiration`, hours) and is exactly what a 421 says was wrong.  The override lives
-    /// until the next snode cache refresh recomputes everything.
-    virtual void set_swarm(
-            session::network::x25519_pubkey swarm_pubkey,
-            swarm::swarm_id_t swarm_id,
-            std::vector<service_node> nodes);
-
     virtual std::vector<service_node> get_unused_nodes(
             size_t count, const std::vector<service_node>& exclude = {});
 
@@ -125,6 +146,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;
 
@@ -134,11 +162,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;
 
     /// This pool's own jobs, rather than the loop's shared queue, so that ~SnodePool can take them
     /// away from the loop before anything is torn down: `stop()` waits out whatever is running and
@@ -178,6 +213,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

src/core.cpp

@@ -544,15 +544,15 @@ void Core::_swarm_response(
         bool timeout,
         int16_t status,
         std::optional<std::string> body) {
-    // Not this member's swarm.  Network has already taken the corrected membership
-    // out of the rejection, so resolving again gets the new one -- and whatever we
-    // were sticking to is exactly what was wrong.
+    // Not this member's swarm.  Network reports this only once the swarm has been corrected -- from
+    // the rejection's redirect, or by a refresh -- so resolving again gets the new one, and
+    // whatever we were sticking to is exactly what was wrong.  A rejection it could not correct
+    // arrives as ERROR_SWARM_UNRESOLVED instead, and ends the operation below.
     if (status == network::ERROR_MISDIRECTED_REQUEST) {
         op->prefer.reset();
 
         // Spent, not merely wrong to stick to: a member that says the account is not its own will
-        // say so again, and the corrected swarm may not have arrived -- an older server sends no
-        // swarm with the rejection, and then re-resolving returns the very same membership.
+        // say so again, and a refreshed pool can still place it in the account's swarm.
         op->spent.push_back(node);
 
         if (++op->redirects > SWARM_REDIRECT_LIMIT)

src/core/swarm_request.hpp

@@ -99,7 +99,9 @@ inline network::Request swarm_request(
             std::move(body),
             network::RequestCategory::standard_small,
             SWARM_REQUEST_TIMEOUT};
-    req.swarm_pubkey = swarm_pubkey;
+    // A single entry: every Core request is about one account, including its batches, whose
+    // sub-requests all address the same swarm
+    req.swarm_pubkeys = {swarm_pubkey};
     req.overall_timeout = SWARM_OVERALL_TIMEOUT;
     return req;
 }