diff --git a/include/session/config/contacts.hpp b/include/session/config/contacts.hpp index ff09704e..a3c4b61a 100644 --- a/include/session/config/contacts.hpp +++ b/include/session/config/contacts.hpp @@ -132,13 +132,33 @@ struct contact_info { /// API: contacts/contact_info::set_name /// - /// Sets a name or nickname; this is exactly the same as assigning to .name/.nickname directly, - /// except that we throw an exception if the given name is longer than MAX_NAME_LENGTH. + /// Sets the contact's own name, as assigning to .name does, except that an over-long one is + /// put through `fixup_contact_name` rather than rejected: this is their name, arriving in their + /// profile, and refusing it would leave us unable to hold the contact at all. /// /// Inputs: /// - `name` -- Name to assign to the contact void set_name(std::string name); + + /// API: contacts/contact_info::set_nickname + /// + /// Sets our own name for the contact, as assigning to .nickname does, except that it throws if + /// the nickname is longer than MAX_NAME_LENGTH. It throws rather than truncating because a + /// nickname is something a person here typed, and storing a prefix of what they wrote changes + /// what they said. Check it with `validate_contact_name` before calling this -- or as it is + /// typed -- and put the refusal in front of them. + /// + /// Inputs: + /// - `nickname` -- Nickname to assign to the contact void set_nickname(std::string nickname); + + /// API: contacts/contact_info::set_nickname_truncated + /// + /// As `set_nickname`, but truncating rather than throwing. Only for a caller that has already + /// decided truncation is acceptable for what it holds. + /// + /// Inputs: + /// - `nickname` -- Nickname to assign to the contact void set_nickname_truncated(std::string nickname); private: @@ -146,6 +166,34 @@ struct contact_info { void load(const dict& info_dict); }; +/// API: contacts/validate_contact_name +/// +/// Reports what is wrong with `name` as a contact name, nickname or profile name, or nullopt when +/// nothing is. The same limits apply to all three, so one check serves them. +/// +/// A free function needing no account or config, so an application can call it on each keystroke +/// while a name is being typed, rather than finding out when it tries to store one. That is the +/// intended use: anything a person typed should be refused here and corrected by them, because the +/// alternative -- keeping a prefix of what they wrote -- changes what they said. +/// +/// Inputs: +/// - `name` -- the candidate name +/// +/// Outputs: +/// - `std::optional` -- what is wrong with it, or nullopt if nothing is +std::optional validate_contact_name(std::string_view name); + +/// API: contacts/fixup_contact_name +/// +/// Makes `name` storable as a contact name, changing as little as it can -- today that is +/// truncating it to MAX_NAME_LENGTH on a utf8 boundary, but it is the place for whatever else +/// storing a name comes to require. +/// +/// For names we are *given* rather than told: a peer's profile name has to be stored whatever they +/// set it to, and refusing it would leave us unable to hold their contact at all. Never use it on +/// something a person typed here -- see `validate_contact_name`. +void fixup_contact_name(std::string& name); + struct blinded_contact_info { const std::string session_id() const; // in hex std::string name; diff --git a/include/session/core.hpp b/include/session/core.hpp index f68faf17..9d9efa71 100644 --- a/include/session/core.hpp +++ b/include/session/core.hpp @@ -19,6 +19,7 @@ #include "core/schema/schema_registry.hpp" #include "session/network/key_types.hpp" #include "session/network/service_node.hpp" +#include "session/network/session_network_types.hpp" /// The "Core" class holds a Session account's own state, in an encrypted sqlite database: its keys, /// its device group, its configs, and the bookkeeping needed to talk to the network on its behalf. @@ -205,6 +206,15 @@ namespace detail { } } // namespace detail +/// Thrown by `set_network` when a Network is already attached. See the TODO on that method for +/// what a replacement would have to do first. +struct network_already_attached : std::logic_error { + network_already_attached() : + std::logic_error{ + "This Core already has a Network attached; replacing it is not yet " + "supported"} {} +}; + /// Wraps a predefined 32-byte account seed to pass to the Core constructor, overriding any seed /// already stored in the database. Used when restoring an existing account from a seed. struct predefined_seed { @@ -309,6 +319,10 @@ class Core { sqlite::Database db; friend class detail::CoreComponent; + // Friendship does not reach a component through its base, and Configs pushes to the swarm, so + // it needs `_swarm_request` by name. + friend class Configs; + core::callbacks callbacks; // Called during the constructor: the database is opened and all members are constructed, but @@ -332,16 +346,83 @@ class Core { void _update_polling(); void _poll(); + /// The outcome of a swarm request. + struct SwarmResponse { + bool timeout; + + /// The storage server's status, or one of the negative ERROR_ values when the request did + /// not get far enough to have one. A batch whose subrequests all failed identically + /// reports that failure rather than the 200 the batch itself returned. + int16_t status_code; + std::optional body; + + /// Which member this came from: the one that answered, or the last one tried. Not + /// necessarily the one the operation started with -- a request can be re-aimed at another + /// member several times before it succeeds, and anything recorded per-node has to be + /// recorded against *this* one. + network::service_node node; + + /// Whether the storage server answered, and answered with a 2xx. + bool ok() const { return !timeout && status_code >= 200 && status_code <= 299; } + explicit operator bool() const { return ok(); } + }; + + // Sends `endpoint` to a member of `swarm_pubkey`'s swarm, re-aiming it as needed, and reports + // which member finally answered. + // + // Re-aiming is here rather than in Network because it is a decision, not a mechanism: only the + // caller knows whether a substitution matters to it, and a substitution made below Core is + // invisible to the bookkeeping that depends on it. Two things move a request: + // + // - a 421, meaning this member does not hold the account. Network will have taken the + // corrected swarm out of the rejection by the time we see it, so re-resolving gets the new + // membership rather than the stale one that misdirected us. Bounded by + // SWARM_REDIRECT_LIMIT, since a server that keeps saying no is not going to stop. + // - an unreachable member, which says nothing about the swarm. Keep the swarm and walk to a + // member not already spent, until they are exhausted. + // + // `make_body` is given the member the attempt will use, because a body can depend on it: a + // retrieve carries that node's cursor, and sending one node's cursor to another asks the wrong + // question. + // `prefer` names a member to go back to rather than choosing afresh, for an operation that + // has to continue against the one it started with; it is dropped as soon as that member turns + // out to be wrong or unreachable. + void _swarm_request( + network::x25519_pubkey swarm_pubkey, + std::string endpoint, + std::function(const network::service_node&)> make_body, + std::function on_done, + std::optional prefer = std::nullopt); + + struct SwarmOp; + void _swarm_attempt(std::shared_ptr op); + void _swarm_send(std::shared_ptr op, network::service_node node); + + // What `_swarm_send`'s reply does, once it is back on our own queue. Split out rather than + // written inline because the network hands it to us on its loop, and everything it does -- + // re-resolving a swarm, running `on_done` -- reaches Core state that only this thread may + // touch. Every swarm operation answers through here, so this is the single hop for all of + // them. + void _swarm_response( + std::shared_ptr op, + network::service_node node, + bool timeout, + int16_t status, + std::optional body); + // Sends one round of retrieves to `node` for `namespaces`. A retrieve is capped by the storage // server, so one round may not exhaust a namespace; `round` counts continuations and bounds // them. Every round goes to the same node: the retrieve cursor is stored per (namespace, // node), so continuing against a different swarm member would resume from that member's // position. void _send_poll( - network::Network* net, - network::service_node node, std::vector namespaces, - int round); + int round, + std::optional node); + + // The batch of retrieves to send `node`, carrying that node's cursor for each namespace. + std::vector _build_poll_body( + const network::service_node& node, const std::vector& namespaces); void _handle_poll_response( network::service_node node, std::vector namespaces, @@ -371,6 +452,49 @@ class Core { const network::ed25519_pubkey& node, std::span messages); void _conclude_profile_fetch(ProfileFanOut& state); + // Swarm push subscription. All of this is touched only on the loop. + // + // Having subscribed with a swarm member, that member pushes each new message to us instead of + // our asking for them, and the poll ticker stops. The subscription belongs to the connection, + // so it does not survive one being rebuilt and there is no notice from the far end when it + // lapses -- it simply stops pushing. Hence: renew on a timer well inside the server's expiry, + // and treat losing the connection as having lost the subscription. + // + // `_sub_node` is not a preference to be restored. It is only the member we happen to be + // talking to, held for as long as its connection lasts because there is no reason to move; a + // fresh one is chosen the ordinary way -- a new `get_swarm`, whatever it hands back first -- + // once this one is gone. + // The swarm member currently carrying our messages: the one being polled, or the one a + // subscription is held with. Not a preference -- each poll re-picks at random, and this only + // stops moving because a subscription stops the polling. + std::optional _swarm_node; + + std::optional _sub_node; + bool _subscribed = false; + std::shared_ptr _sub_ticker; + std::shared_ptr _probe_ticker; + + // Subscribes to `node` if a subscription is possible and we do not already have one. Called + // when a poll of `node` drains, which is what makes it the node we subscribe with: it has an + // established connection and its cursors are current. + void _maybe_subscribe(const network::service_node& node); + void _send_subscribe(network::Network* net, network::service_node node); + + // Re-sends the subscribe, so that the server's expiry never elapses on a connection that is + // still up. + void _subscription_renew(); + + // Asks the subscribed node a question whose only purpose is the 421 we get if it has stopped + // holding our swarm. Nothing else would notice: a subscription that has stopped applying is + // silent, not an error. Temporary -- see the comment on the definition. + void _subscription_probe(); + + // Gives up the subscription and returns to polling. + void _drop_subscription(std::string_view why); + + // Feeds one pushed message in as though it had been retrieved. + void _handle_server_push(std::string_view endpoint, std::span body); + // Decrypts and dispatches one-to-one messages from Namespace::Default. void _handle_direct_messages(std::span messages); @@ -489,8 +613,28 @@ class Core { init(); } + /// Detaches from the Network before letting anything be destroyed; see the definition. + ~Core(); + /// Set an optional network interface that can be used to make network requests to swarm /// members. Ownership is taken: nothing else may hold on to the Network. + /// + /// May only be called once, and only from a thread that is not Core's loop; replacing an + /// already-attached Network (including with nullptr) throws `network_already_attached`. + /// + /// TODO: allow the Network to be replaced. A client that lets the user choose a routing mode + /// needs it, and so does anything that has to re-establish swarm state across the swap. Two + /// things block it today: + /// + /// - This calls `_update_polling()` on the caller's thread, which creates and stops the + /// libevent poll ticker. `set_poll_interval` marshals onto the loop for exactly that reason. + /// - Tearing down a Network *invokes* the callbacks it is holding: failing the requests queued + /// in its router and transport is part of `~Network`. Those callbacks are Core's, they hold + /// a raw `Network*` (see `_poll`), and a poll continuation among them will call back into a + /// Network whose router has already been destroyed. + /// + /// So a fix is not a `_loop.call` around this body: polling has to be stopped and in-flight + /// swarm work quiesced before the old Network is dropped. void set_network(std::unique_ptr network); /// Constructs the network in place and attaches it, forwarding the arguments to its @@ -647,6 +791,21 @@ class Core { /// must not keep this beyond the point where the network could be replaced or dropped. network::Network* network() const { return _network.get(); } + /// The swarm member currently carrying our messages -- the one being polled, or the one a + /// subscription is held with -- or nullopt before there is one. + /// + /// Which member that is changes on its own: each poll picks a fresh one at random, and a + /// subscription holds one only for as long as its connection lasts. Read it to show what is + /// happening now, not to depend on it. + std::optional swarm_node() const { return _swarm_node; } + + /// The route our traffic to `swarm_node()` is taking right now, for showing a user where it + /// goes. Nullopt when there is no member yet, no network attached, or no route to report. + /// + /// A snapshot rather than a commitment: paths rotate and subscriptions move, so asking again + /// later can legitimately give a different answer. + std::optional current_swarm_path() const; + /// The event loop this account's work runs on. /// /// Everything Core does off the caller's thread — polling, send completion, and therefore every diff --git a/include/session/core/configs.hpp b/include/session/core/configs.hpp index 1cadcd6d..003b855c 100644 --- a/include/session/core/configs.hpp +++ b/include/session/core/configs.hpp @@ -96,6 +96,11 @@ class Configs : public detail::CoreComponent { size_t count; }; + // Queues a flush for the next turn of the loop, once however many times it is called before + // then. Every accessor that hands out a config calls this; see the note above them. + void _schedule_settle(); + bool _settle_scheduled = false; + void _schedule_push(); void _arm_push_timer(std::chrono::milliseconds delay); void _push_if_due(); diff --git a/include/session/core/globals.hpp b/include/session/core/globals.hpp index 96c76f0b..cc429035 100644 --- a/include/session/core/globals.hpp +++ b/include/session/core/globals.hpp @@ -127,7 +127,6 @@ class Globals final : detail::CoreComponent { void restore_account(predefined_seed seed, failable_function cb); void restore_account(const predefined_seed& seed, await_t); - public: // Retrieval methods. These query for the given key and, if the type matches, return the given // value. You get back nullopt if the database key does not exist, or if it contains a value // of some other type. diff --git a/include/session/network/network_config.hpp b/include/session/network/network_config.hpp index f1745296..bf0905d2 100644 --- a/include/session/network/network_config.hpp +++ b/include/session/network/network_config.hpp @@ -38,7 +38,6 @@ struct Config { bool increase_no_file_limit = false; uint8_t path_length = 3; bool enforce_subnet_diversity = true; - uint8_t redirect_retry_count = 1; opt::retry_delay retry_delay = opt::retry_delay(200ms, 5s); uint8_t num_nodes_to_check_for_network_offset = 3; std::chrono::minutes min_resume_clock_resync_interval = 10min; @@ -68,7 +67,20 @@ struct Config { std::chrono::days onionreq_edge_node_cache_duration = std::chrono::days{10}; // Quic Transport Options - std::chrono::milliseconds quic_handshake_timeout{3s}; + + /// How long a QUIC handshake straight out to a node's own address gets: the guard node of an + /// onion request, a direct-mode destination, a connectivity check. One internet round trip and + /// a little slack. + std::chrono::milliseconds quic_handshake_timeout{5s}; + + /// How long a QUIC handshake gets when its packets go through a Session Router tunnel. + /// + /// A separate figure because it measures something else entirely: the connection is nominally + /// to ::1, but every packet of it crosses the whole tunnel, so the budget has to cover a + /// multi-hop round trip rather than a direct one. That it currently sits at a value other + /// constants here also happen to use means nothing -- move either one on its own merits. + std::chrono::milliseconds quic_tunnel_handshake_timeout{10s}; + std::chrono::seconds quic_keep_alive{10s}; std::optional quic_max_udp_payload; @@ -102,7 +114,6 @@ struct Config { void handle_config_opt(opt::increase_no_file_limit infl); void handle_config_opt(opt::path_length pl); void handle_config_opt(opt::disable_subnet_diversity dsd); - void handle_config_opt(opt::redirect_retry_count rrc); void handle_config_opt(opt::retry_delay rd); void handle_config_opt(opt::num_nodes_to_check_for_network_offset nncno); void handle_config_opt(opt::min_resume_clock_resync_interval mrcri); @@ -125,6 +136,7 @@ struct Config { // Quic transport options void handle_config_opt(opt::quic_handshake_timeout qht); + void handle_config_opt(opt::quic_tunnel_handshake_timeout qtht); void handle_config_opt(opt::quic_keep_alive qka); void handle_config_opt(opt::quic_max_udp_payload qmup); diff --git a/include/session/network/network_opt.hpp b/include/session/network/network_opt.hpp index c2592081..cfaf2323 100644 --- a/include/session/network/network_opt.hpp +++ b/include/session/network/network_opt.hpp @@ -126,6 +126,18 @@ namespace opt { } }; + /// Storage server version from which a service node is expected to be reachable over Session + /// Router, and so worth preferring when a swarm member is chosen in that mode. + /// + /// The storage server does not run the relay and says nothing about it; this is an inference + /// from how releases are packaged. Every storage server at or above this version on the + /// network is paired with oxend 11.6.0, which requires a session-router relay beside it, so an + /// older one almost certainly cannot be reached that way. + /// + /// A preference rather than a requirement, because the relay is not enforced: a node can be new + /// enough and still not answer. Revisit once it is, and whenever the packaging changes. + inline constexpr std::array MIN_SESSION_ROUTER_SS_VERSION = {2, 11, 1}; + /// Can be used to override the default (onion_requests) routing method for requests. struct router { enum class Type { @@ -230,14 +242,6 @@ namespace opt { /// included in the same path when building onion request or session router paths. struct disable_subnet_diversity {}; - /// Can be used to override the default (1) number of request retries that will occur when - /// receiving a 421 error. - struct redirect_retry_count { - uint8_t count; - - redirect_retry_count(uint8_t count) : count{count} {} - }; - struct retry_delay { std::chrono::milliseconds base_delay; std::chrono::milliseconds max_delay; @@ -379,13 +383,22 @@ namespace opt { // MARK: Quic Transport Options - /// Can be used to override the default (10s) handshake timeout duration for Quic connections. + /// Can be used to override the default (5s) handshake timeout duration for Quic connections + /// made directly to a node's own address. struct quic_handshake_timeout { std::chrono::milliseconds duration; quic_handshake_timeout(std::chrono::milliseconds duration) : duration{duration} {} }; - /// Can be used to override the default (0ms) keep alive duration for Quic connections. + /// Can be used to override the default (10s) handshake timeout duration for Quic connections + /// whose packets travel through a Session Router tunnel, which have a multi-hop round trip to + /// complete rather than a direct one. + struct quic_tunnel_handshake_timeout { + std::chrono::milliseconds duration; + quic_tunnel_handshake_timeout(std::chrono::milliseconds duration) : duration{duration} {} + }; + + /// Can be used to override the default (10s) keep alive duration for Quic connections. struct quic_keep_alive { std::chrono::seconds duration; quic_keep_alive(std::chrono::seconds duration) : duration{duration} {} @@ -467,7 +480,6 @@ namespace opt { increase_no_file_limit, path_length, disable_subnet_diversity, - redirect_retry_count, retry_delay, num_nodes_to_check_for_network_offset, min_resume_clock_resync_interval, @@ -490,6 +502,7 @@ namespace opt { // Quic transport options quic_handshake_timeout, + quic_tunnel_handshake_timeout, quic_keep_alive, quic_max_udp_payload, diff --git a/include/session/network/routing/direct_router.hpp b/include/session/network/routing/direct_router.hpp index 5d5efbe4..ef52a5e4 100644 --- a/include/session/network/routing/direct_router.hpp +++ b/include/session/network/routing/direct_router.hpp @@ -58,6 +58,11 @@ class DirectRouter : public IRouter, public std::enable_shared_from_this seed) override; void download(DownloadRequest request) override; + /// Sending direct, the route to a node is the node: one hop, no relays, nothing hidden. + std::optional get_path_to(const service_node& node) override { + return PathInfo{{{node.remote_pubkey, node.ip}}}; + } + private: std::atomic _status{ConnectionStatus::unknown}; void _close_connections(); diff --git a/include/session/network/routing/network_router.hpp b/include/session/network/routing/network_router.hpp index a4a3c67c..b8e90031 100644 --- a/include/session/network/routing/network_router.hpp +++ b/include/session/network/routing/network_router.hpp @@ -16,7 +16,15 @@ class IRouter { virtual void clear_cache() = 0; virtual ConnectionStatus get_status() const = 0; - virtual std::vector get_active_paths() { return {}; }; + /// The route traffic to `node` is taking right now, for showing a user where it goes. + /// + /// A snapshot, not a commitment: a router may rotate away from it at any time, and asking + /// again a moment later can legitimately give a different answer. Nullopt when there is + /// nothing to report -- nothing has been sent to that node yet, or no route to it exists. + /// + /// Takes the whole node rather than its pubkey because sending direct has no route to look + /// up: the answer is the node itself, and that needs its address. + virtual std::optional get_path_to(const service_node&) { return std::nullopt; }; virtual std::vector get_all_used_nodes() { return {}; }; virtual void send_request(Request request, network_response_callback_t callback) = 0; [[deprecated("use upload_file() instead")]] diff --git a/include/session/network/routing/onion_request_router.hpp b/include/session/network/routing/onion_request_router.hpp index 42885d48..3891ff83 100644 --- a/include/session/network/routing/onion_request_router.hpp +++ b/include/session/network/routing/onion_request_router.hpp @@ -96,6 +96,9 @@ inline PathCategory to_path_category(RequestCategory category) { case RequestCategory::standard_small: return PathCategory::standard; case RequestCategory::file: return PathCategory::file; case RequestCategory::file_small: return PathCategory::file; + // Nothing to distinguish here: the stream a config would get is a Session Router notion, + // and an onion path carries it like any other standard request. + case RequestCategory::config: return PathCategory::standard; } return PathCategory::standard; // Should not be reached } @@ -161,7 +164,7 @@ class OnionRequestRouter : public IRouter, public std::enable_shared_from_this get_active_paths() override; + std::optional get_path_to(const service_node& node) override; std::vector get_all_used_nodes() override; void send_request(Request request, network_response_callback_t callback) override; void upload(UploadRequest request) override; // deprecated: use upload_file() @@ -213,6 +216,15 @@ class OnionRequestRouter : public IRouter, public std::enable_shared_from_this desired_path_index, + std::string_view request_id); + void _send_on_path(OnionPath& path, Request request, network_response_callback_t callback); void _handle_transport_response( std::string path_id, diff --git a/include/session/network/routing/session_router_router.hpp b/include/session/network/routing/session_router_router.hpp index 0f02c068..a9258a52 100644 --- a/include/session/network/routing/session_router_router.hpp +++ b/include/session/network/routing/session_router_router.hpp @@ -82,7 +82,7 @@ class SessionRouter : public IRouter, public std::enable_shared_from_this get_active_paths() override; + std::optional get_path_to(const service_node& node) override; void send_request(Request request, network_response_callback_t callback) override; void upload(UploadRequest request) override; // deprecated: use upload_file() void upload_file(FileUploadRequest request, std::span seed) override; diff --git a/include/session/network/session_network.h b/include/session/network/session_network.h index d3a63568..722211dc 100644 --- a/include/session/network/session_network.h +++ b/include/session/network/session_network.h @@ -58,7 +58,6 @@ typedef struct session_network_config { bool increase_no_file_limit; uint8_t path_length; bool enforce_subnet_diversity; - uint8_t redirect_retry_count; uint64_t min_retry_delay_ms; uint64_t max_retry_delay_ms; uint8_t num_nodes_to_check_for_network_offset; @@ -92,6 +91,9 @@ typedef struct session_network_config { // Quic transport options (for transport == SESSION_NETWORK_TRANSPORT_QUIC) uint32_t quic_handshake_timeout_seconds; + /// Handshake timeout for connections whose packets travel through a Session Router tunnel, + /// which have a multi-hop round trip to complete rather than a direct one. + uint32_t quic_tunnel_handshake_timeout_seconds; uint32_t quic_keep_alive_seconds; bool quic_disable_mtu_discovery; // deprecated: use quic_max_udp_payload instead /// Maximum QUIC UDP payload size for PMTUD; 0 for default (no cap). @@ -219,11 +221,6 @@ LIBSESSION_EXPORT void session_network_callbacks_respond( LIBSESSION_EXPORT CONNECTION_STATUS session_network_get_status(network_object* network); -LIBSESSION_EXPORT void session_network_get_active_paths( - network_object* network, session_path_info** out_paths, size_t* out_paths_len); - -LIBSESSION_EXPORT void session_network_paths_free(session_path_info* paths); - LIBSESSION_EXPORT void session_network_get_swarm( network_object* network, const char* swarm_pubkey_hex, diff --git a/include/session/network/session_network.hpp b/include/session/network/session_network.hpp index 33ee4476..3f42fa2b 100644 --- a/include/session/network/session_network.hpp +++ b/include/session/network/session_network.hpp @@ -69,6 +69,28 @@ class Network { std::function on_network_info_changed; + /// Hook to be notified when a storage server sends us something we did not ask for, on a + /// connection we already hold -- which is how a swarm subscription delivers messages. `node` + /// names the swarm member; `endpoint` and `body` are the pushed request's, unparsed. + /// + /// Only reachable with a routing mode that gives the storage server a connection to us, which + /// means `session_router` or `direct`. Under `onion_requests` the server has nothing to push + /// down: the connection it can see belongs to the last relay rather than to us, so it would + /// key the subscription to that relay. This is not a property of onion routing in general -- + /// `session_router` is onion-routed too, and is the mode this exists for. + std::function body)> + on_server_push; + + /// Hook to be notified once a connection to `node` is usable, including when it comes back + /// after having been lost. Per-connection state the far end holds for us -- a subscription -- + /// does not survive that, so this is where it has to be established again. + std::function on_connection_established; + + /// Hook to be notified when a connection to `node` is gone, for any reason. A subscription + /// held on it is gone too, and the far end will not say so: it simply stops pushing. + std::function on_connection_lost; + template requires(!std::is_same_v< std::decay_t>>, @@ -87,13 +109,24 @@ class Network { uint16_t hardfork() const { return _fork_versions.load().hardfork; }; uint16_t softfork() const { return _fork_versions.load().softfork; }; + /// Whether a storage server can push to us on this network, i.e. whether `on_server_push` can + /// ever fire and a subscription is worth making. + /// + /// False for onion requests, and not as a matter of it being unimplemented: the storage server + /// keys a subscription to the connection the request arrived on, which for an onion request is + /// the last relay's rather than ours, so subscribing over one would register a relay as the + /// subscriber. Anything relying on pushed messages has to keep polling in that mode. + bool supports_server_push() const { return config.router != opt::router::Type::onion_requests; } + void suspend(); void resume(bool automatically_reconnect = true); void close_connections(); void clear_cache(); ConnectionStatus get_status(); - std::vector get_active_paths(); + /// The route traffic to `node` is taking right now, for showing a user where it goes. A + /// snapshot rather than a commitment; see IRouter::get_path_to. + std::optional get_path_to(const service_node& node); /// API: network/get_swarm /// @@ -142,20 +175,9 @@ class Network { 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); - - // Re-sends a request to the next member of the same swarm, after the one it was sent to could - // not be reached. Distinct from the 421 path: there the swarm information was wrong and is - // thrown away, here it is right and only one member of it is unusable. Gives up when - // selection has no member left that has not already failed, reporting the original failure - // rather than one of its own invention. - void _retry_next_swarm_node( - Request original_request, - bool timeout, - int16_t status_code, - std::vector> headers, - std::optional body, - network_response_callback_t final_callback); + // 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 _resync_clock( std::optional original_request, network_response_callback_t request_callback); diff --git a/include/session/network/session_network_types.h b/include/session/network/session_network_types.h index a830cdeb..13a259dc 100644 --- a/include/session/network/session_network_types.h +++ b/include/session/network/session_network_types.h @@ -22,6 +22,7 @@ typedef enum { SESSION_NETWORK_REQUEST_CATEGORY_STANDARD_SMALL, SESSION_NETWORK_REQUEST_CATEGORY_FILE, SESSION_NETWORK_REQUEST_CATEGORY_FILE_SMALL, + SESSION_NETWORK_REQUEST_CATEGORY_CONFIG, } SESSION_NETWORK_REQUEST_CATEGORY; typedef enum { @@ -75,25 +76,6 @@ typedef struct { } session_request_params; -typedef struct { - SESSION_NETWORK_PATH_CATEGORY category; -} session_onion_path_metadata; - -typedef struct { - char destination_pubkey[65]; // The 64-byte ed25519 pubkey in hex + null terminator. - char destination_snode_address[65]; // The 64-byte .snode address + null terminator. -} session_router_tunnel_metadata; - -typedef struct { - const network_service_node* nodes; - size_t nodes_count; - - // Only ONE of these pointers should be set, the other should be left null - const session_onion_path_metadata* onion_metadata; - const session_router_tunnel_metadata* session_router_metadata; - -} session_path_info; - #ifdef __cplusplus } #endif diff --git a/include/session/network/session_network_types.hpp b/include/session/network/session_network_types.hpp index 620ea33d..d5d43990 100644 --- a/include/session/network/session_network_types.hpp +++ b/include/session/network/session_network_types.hpp @@ -70,11 +70,32 @@ enum class ConnectionStatus { disconnected = CONNECTION_STATUS_DISCONNECTED, }; +/// What a request is for, which decides how it is carried. +/// +/// The `_small` distinction is a QUIC stream choice: a small request goes on the connection's +/// reserved stream 0, sharing it with everything else small, while the rest take a stream of their +/// own from the connection's pool. Ordering is per-stream, so what shares a stream waits for what +/// is ahead of it. +/// +/// Two of these are meaningful only under one routing mode, because the thing they distinguish does +/// not exist under the other. enum class RequestCategory { standard = SESSION_NETWORK_REQUEST_CATEGORY_STANDARD, standard_small = SESSION_NETWORK_REQUEST_CATEGORY_STANDARD_SMALL, + + /// A file transfer. Only means anything under `onion_requests`, where a file goes to the file + /// server through a storage node like everything else and so shares that node's connection. + /// Session Router reaches the file server directly rather than through a snode, so there is no + /// shared connection for a file to be separated from. file = SESSION_NETWORK_REQUEST_CATEGORY_FILE, file_small = SESSION_NETWORK_REQUEST_CATEGORY_FILE_SMALL, + + /// A config push. Only means anything under Session Router, which holds a real QUIC connection + /// per storage node and can therefore put this on a stream of its own -- so a large config does + /// not delay a small store queued behind it on the reserved stream. Under `onion_requests` + /// there is no such connection to open a second stream on, and this behaves as + /// `standard_small`. + config = SESSION_NETWORK_REQUEST_CATEGORY_CONFIG, }; enum class PathCategory { @@ -88,6 +109,7 @@ inline std::string to_string(RequestCategory category) { case RequestCategory::standard_small: return "standard_small"; case RequestCategory::file: return "file"; case RequestCategory::file_small: return "file_small"; + case RequestCategory::config: return "config"; } return "unknown"; // Should not be reached } @@ -168,6 +190,12 @@ struct Request { /// behaviour. std::optional desired_path_index; + /// True when `destination` is the local end of a Session Router tunnel rather than an address + /// out on the internet. The transport cannot tell from the address -- it is a loopback port + /// either way -- and it needs to know, because a handshake whose packets cross a whole tunnel + /// gets a different budget than one that does not. + bool tunnelled = false; + /// Any extra request details which may modify the structure of the request. RequestDetails details; @@ -176,33 +204,16 @@ struct Request { /// 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. /// - /// 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. + /// 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 + /// `Network::send_request`. std::optional swarm_pubkey; /// The time the request was created, this is used primarily for determining whether the /// `overall_timeout` has been exceeded. std::chrono::steady_clock::time_point creation_time = std::chrono::steady_clock::now(); - /// How many times this request has been redirected after a 421, bounded by - /// `config.redirect_retry_count`. Counts redirects only -- a 421 means our swarm information - /// was wrong, so recovery is to re-resolve the swarm from scratch. It has nothing to do with - /// `failed_nodes` below, which is the opposite situation. - int retry_421_count = 0; - - /// Swarm members that could not be reached for this request, in the order they were tried. - /// - /// A node that cannot be reached says nothing about the swarm -- unlike a 421, which says the - /// swarm itself is wrong -- so recovery is to keep the swarm and move to the next-best member, - /// excluding these. Running out of members is what ends it, so this is a set rather than a - /// count: "once per node" cannot be expressed as a number, since choosing the next one has to - /// know which have already been spent. - /// - /// Empty for anything not addressed to a swarm; a request with no `swarm_pubkey` has no other - /// member to move to. - std::vector failed_nodes; - Request(std::string request_id, network_destination destination, std::string endpoint, @@ -315,19 +326,29 @@ namespace response { std::optional find_uniform_batch_error(std::string_view body); } // namespace response -struct OnionPathMetadata { - PathCategory category; -}; -struct SessionRouterTunnelMetadata { - std::string destination_pubkey; - std::string destination_snode_address; +/// One hop of a route, for showing a user where their traffic goes. Identity and location, which +/// is all a diagnostic needs -- deliberately not a `service_node`, because a hop is not always one: +/// a Session Router relay has no ports, no storage server version and no swarm, and filling those +/// in with zeros would make a swarm id of 0 that means something else. +struct PathHop { + ed25519_pubkey pubkey; + oxen::quic::ipv4 ip; }; -using PathMetadata = std::variant; - +/// A route, as shown to a user. Deliberately separate from how a router represents the paths it +/// actually sends on: those carry pool membership, strike counts and other bookkeeping that means +/// nothing outside the router, and mixing the two ends up publishing one to serve the other. struct PathInfo { - std::vector nodes; - PathMetadata metadata; + /// The hops we can see, in order from the one nearest us. + /// + /// Whether the last of them is the destination depends on the route, and the caller cannot + /// tell from here. Sending direct, the only hop *is* the destination. A Session Router + /// tunnel to a `.snode` terminates at the storage node, so it is; a path to a client + /// terminates at the pivot relay, with the rest belonging to the other side and invisible to + /// us; and an `onion_requests` path is built before any destination is chosen, so its last + /// hop is a relay that will forward to whatever the request names. Reporting what is known + /// beats a shape that promises a destination which is sometimes a guess. + std::vector hops; }; } // namespace session::network diff --git a/include/session/network/snode_pool.hpp b/include/session/network/snode_pool.hpp index 2352f78d..22d750d2 100644 --- a/include/session/network/snode_pool.hpp +++ b/include/session/network/snode_pool.hpp @@ -35,6 +35,13 @@ namespace config { uint8_t cache_num_nodes_to_use_for_refresh; uint8_t cache_min_num_refresh_presence_to_include_node; uint16_t cache_node_strike_threshold; + + /// Storage server version at or above which a swarm member is preferred, or nullopt to + /// treat every member alike. + /// + /// Deliberately a version rather than a reason: what the version *means* belongs to + /// whoever sets this, and this layer only has to order by it. + std::optional> prefer_min_version; }; } // namespace config @@ -87,6 +94,17 @@ class SnodePool : public std::enable_shared_from_this { bool ignore_strike_count, std::function)> 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 nodes); + virtual std::vector get_unused_nodes( size_t count, const std::vector& exclude = {}); diff --git a/include/session/network/transport/network_transport.hpp b/include/session/network/transport/network_transport.hpp index 12426267..3ec9cd9a 100644 --- a/include/session/network/transport/network_transport.hpp +++ b/include/session/network/transport/network_transport.hpp @@ -8,6 +8,42 @@ class ITransport { public: std::function on_status_changed; + /// Called when the far end sends us something we did not ask for, on a connection we are + /// already holding. A swarm subscription is delivered this way: having subscribed, the + /// storage server pushes each matching message as a request of its own rather than as a + /// response to anything. + /// + /// `node` is the far end's ed25519 pubkey, which is the key the connection is addressed by + /// whether it reached the node directly or through a tunnel, so it names the swarm member + /// either way. `endpoint` and `body` are the pushed request's, unparsed: the transport does + /// not know what any of them mean. + /// + /// No reply is sent. Nothing that pushes to us expects one, and answering a request the far + /// end is not tracking would only be discarded. + /// + /// Runs on the network loop and must not throw. + std::function body)> + on_server_push; + + /// Called once a connection to `node` is up and requests can be sent on it, including when it + /// comes back after having been lost. The counterpart to `add_failure_listener`, and the + /// point at which per-connection state the far end holds -- a subscription, say -- has to be + /// established again, since the far end keys that state on the connection and the old one is + /// gone. + /// + /// Runs on the network loop and must not throw. + std::function on_connection_established; + + /// Called when a connection to `node` is gone, for any reason: closed, failed, or timed out. + /// Whatever the far end was holding for that connection is gone with it. + /// + /// Unlike `add_failure_listener` this is not one-shot and not per-node: it reports every + /// connection this transport loses, and stays registered. + /// + /// Runs on the network loop and must not throw. + std::function on_connection_lost; + virtual ~ITransport() = default; virtual void suspend() = 0; diff --git a/include/session/network/transport/quic_transport.hpp b/include/session/network/transport/quic_transport.hpp index bbfc7cea..6fd4acd4 100644 --- a/include/session/network/transport/quic_transport.hpp +++ b/include/session/network/transport/quic_transport.hpp @@ -22,6 +22,7 @@ namespace session::network { namespace config { struct QuicTransport { std::chrono::milliseconds handshake_timeout; + std::chrono::milliseconds tunnel_handshake_timeout; std::chrono::seconds keep_alive; std::optional max_udp_payload; @@ -92,7 +93,8 @@ class QuicTransport : public ITransport { void _establish_connection( const oxen::quic::RemoteAddress& address, const std::string& initiating_req_id, - const RequestCategory category); + const RequestCategory category, + bool tunnelled); void _send_on_connection( oxen::quic::ConnectionID conn_id, const std::string remote_pubkey_hex, diff --git a/src/client/client.cpp b/src/client/client.cpp index 7b72f550..79d40270 100644 --- a/src/client/client.cpp +++ b/src/client/client.cpp @@ -1727,6 +1727,17 @@ void Client::_set_auto_download(const ConversationId& id, AutoDownload mode) { } void Client::_set_nickname(const ConversationId& id, std::string_view nickname) { + // Before the row is written rather than after. The config refuses a nickname this long, and + // the row is what the config is rebuilt from -- so a committed over-long one could never be + // carried, and because a sync rebuilds the whole entry it would take every later change to that + // contact down with it: a block, an approval, a priority, a delete-before instruction. + // + // Reported rather than corrected: `_async` turns this into the error the caller's handler is + // given, which is what an application wants to put in front of whoever typed it. Quietly + // keeping the first hundred bytes would change what they wrote. + if (auto problem = config::validate_contact_name(nickname)) + throw std::invalid_argument{"set_nickname: {}"_format(*problem)}; + bool changed = false; { auto c = core.database().conn(); diff --git a/src/config/contacts.cpp b/src/config/contacts.cpp index be8b5330..d0bf579c 100644 --- a/src/config/contacts.cpp +++ b/src/config/contacts.cpp @@ -40,11 +40,21 @@ contact_info::contact_info(std::string sid) : session_id{std::move(sid)} { check_session_id(session_id); } +std::optional session::config::validate_contact_name(std::string_view name) { + if (name.size() > contact_info::MAX_NAME_LENGTH) + return "name is too long: {} bytes, and the maximum is {}"_format( + name.size(), contact_info::MAX_NAME_LENGTH); + return std::nullopt; +} + +void session::config::fixup_contact_name(std::string& name) { + if (name.size() > contact_info::MAX_NAME_LENGTH) + name = session::utf8_truncate(std::move(name), contact_info::MAX_NAME_LENGTH); +} + void contact_info::set_name(std::string n) { - if (n.size() > MAX_NAME_LENGTH) - name = utf8_truncate(std::move(n), MAX_NAME_LENGTH); - else - name = std::move(n); + fixup_contact_name(n); + name = std::move(n); } void contact_info::set_nickname(std::string n) { @@ -413,10 +423,8 @@ const std::string blinded_contact_info::session_id() const { } void blinded_contact_info::set_name(std::string n) { - if (n.size() > contact_info::MAX_NAME_LENGTH) - name = utf8_truncate(std::move(n), contact_info::MAX_NAME_LENGTH); - else - name = std::move(n); + fixup_contact_name(n); + name = std::move(n); } void blinded_contact_info::set_base_url(std::string_view base_url) { diff --git a/src/core.cpp b/src/core.cpp index 335a85a1..cd49950f 100644 --- a/src/core.cpp +++ b/src/core.cpp @@ -4,6 +4,7 @@ #include #include #include +#include #include #include #include @@ -100,6 +101,59 @@ quic::Loop& Core::loop() { return _loop; } +Core::~Core() { + // Tearing a Network down fails every request its transport is still holding, and failing them + // fires the hooks installed in set_network -- which marshal onto our loop and reach members + // that are already gone. Members are destroyed in reverse declaration order and every ticker + // is declared after `_network`, so by the time ~Network runs they have been released while + // `_loop`, declared first, is still alive to run the job: `stop()` on a freed Ticker, at every + // exit that had a subscription. + // + // So detach before anything is torn down. ~Network pays its own router and transport the + // same courtesy for the same reason, and doing it here rather than by shuffling the member + // declarations leaves the requirement stated instead of resting on where a field sits. + if (_network) { + _network->on_server_push = nullptr; + _network->on_connection_established = nullptr; + _network->on_connection_lost = nullptr; + } + + // Stopped while they are certainly still alive. Releasing them is left to the members + // themselves, which happens before `_loop` goes and so can still reach it. + for (auto* ticker : {&_poll_ticker, &_sub_ticker, &_probe_ticker}) + if (*ticker) + (*ticker)->stop(); + + // Blocking, and it must not run on the Network's own loop -- it does not, because a Core is + // destroyed by whoever owns it. Before the queue is stopped rather than after: tearing the + // Network down fails whatever it still holds, and those completions marshal onto our queue, + // which throws if it has already been stopped. Queued here they are simply cancelled below. + _network.reset(); + + // One job that settles and then shuts the queue down from inside itself, which is what + // `process_job_queue` is written for -- it re-checks the running flag between jobs, so nothing + // else in the batch runs once this returns. `stop()` clears the queue and deletes every armed + // `call_later`, so by the time the members below are destroyed there is no job, no timer and no + // deferred deleter left that could reach one of them. None of what a component holds is + // thread-safe, so that guarantee is the point rather than a tidiness. + // + // The dump goes first because a change settles a turn of the loop after it is made, and one + // made just before this is still queued ahead of us -- so it runs, and then this writes what it + // could not. + // + // Swallowed rather than propagated: this is a destructor, and a database that cannot be written + // is not something the caller tearing Core down can act on. + try { + call_get([this] { + if (globals.have_account()) + configs.store_dumps(); + _jq.stop(); + }); + } catch (const std::exception& e) { + log::warning(cat, "Could not shut Core's queue down cleanly: {}", e.what()); + } +} + void Core::set_network(std::unique_ptr network) { // Polling signs its retrieve requests with the account key, so attaching a network before the // account has an identity would fail inside a background poll rather than here. Refuse at the @@ -107,11 +161,57 @@ void Core::set_network(std::unique_ptr network) { if (network && !globals.have_account()) throw no_account{}; + // Replacing an attached Network is unsupported, and unsupported here means unsafe rather than + // merely unimplemented: see the TODO in core.hpp. Refuse rather than corrupt. + if (_network) + throw network_already_attached{}; + // Ownership moves in via release() because the two pointer types differ deliberately: the // parameter is a plain unique_ptr so callers can hand over a std::make_unique, while the // member's deleter (which is just `delete`) is what keeps Network an incomplete type in // core.hpp -- including session_network.hpp there costs ~6x the compile time per file. _network.reset(network.release()); + + if (_network) { + // These fire on the network's loop; each hops onto ours before touching subscription + // state. Safe to capture `this` bare: the Network is declared after `_loop` so it is + // destroyed first, and ~Network does not return until no callback of its is still in + // flight. + // The body is copied because it has to be: the span is the transport's buffer and is only + // valid for the duration of this call, so anything deferred must own its bytes. That cost + // is what a push is worth -- a poll response arrives the same way and is copied too. + // + // Which node sent it is deliberately not checked: `_sub_node` is our loop's, and by the + // time this runs the answer could have changed anyway. It does not need to be, either -- + // everything here is authenticated downstream, replays dedup on the swarm hash, and + // configs merge by seqno, so the worst a connected node achieves by pushing us something + // is making us do work we would have done anyway. + _network->on_server_push = [this](const network::ed25519_pubkey& /*node*/, + std::string_view endpoint, + std::span body) { + call([this, endpoint = std::string{endpoint}, body = to_vector(body)] { + _handle_server_push(endpoint, body); + }); + }; + + _network->on_connection_lost = [this](const network::ed25519_pubkey& node) { + call([this, node] { + if (_sub_node && _sub_node->remote_pubkey == node) + _drop_subscription("connection lost"); + }); + }; + + _network->on_connection_established = [this](const network::ed25519_pubkey& node) { + call([this, node] { + // A rebuilt connection carries no subscription: the far end keyed the old one to + // the connection that just went away. Losing it should already have dropped us, + // so this is the case where it somehow did not. + if (_subscribed && _sub_node && _sub_node->remote_pubkey == node) + _drop_subscription("connection was re-established"); + }); + }; + } + _update_polling(); } @@ -262,25 +362,238 @@ struct Core::ProfileFanOut { bool concluded = false; }; -void Core::_poll() { - // Non-owning: the Network is ours alone, and callbacks below must not keep it alive -- doing so - // could make the loop thread the last owner and run ~Network there. +// The namespaces we ask a storage server to push to us: the ones we poll, in ascending order, +// which is what the server requires (it rejects an unordered `n=` list). Derived from +// POLL_NAMESPACES rather than written out so that adding a namespace to the poll cannot leave the +// subscription silently not covering it. +static constexpr auto SUBSCRIBE_NAMESPACES = [] { + std::array ns{}; + for (size_t i = 0; i < POLL_NAMESPACES.size(); i++) + ns[i] = static_cast(POLL_NAMESPACES[i]); + std::ranges::sort(ns); + return ns; +}(); + +// How often a live subscription is re-sent. The storage server expires one 65 minutes after the +// last renewal, so this leaves four times the headroom it needs -- and the expiry only ever +// matters on a connection that has stayed up that long, since losing the connection loses the +// subscription outright. +static constexpr auto SUBSCRIPTION_RENEW_INTERVAL = 15min; + +// How often the subscribed node is probed; see _subscription_probe. +static constexpr auto SUBSCRIPTION_PROBE_INTERVAL = 30s; + +// The namespace the probe asks about: negative and of the form -(20n+1), which is what makes a +// retrieve of it need no signature (oxenss/common/namespace.h, is_noauth_retrieve_namespace), and +// otherwise unassigned, so it is permanently empty and the reply is a fixed 57 bytes. Deliberately +// not a memorable number: it should not look like it means something. +static constexpr int16_t PROBE_NAMESPACE = -3741; + +// A probe is answered or it is not; there is no reason to spend the swarm budget on it. +static constexpr auto PROBE_TIMEOUT = 10s; + +// What the storage server pushes a subscribed client, as the endpoint of a request of its own. +static constexpr auto NOTIFY_ENDPOINT = "notify"sv; + +// How many times a swarm request is re-aimed after a 421 before it is given up on. +// +// One redirect is the ordinary case: our membership was stale, the rejection corrected it, the +// next member answers. More than that means the corrected swarm is also being rejected, and +// asking a fourth time will not change that. +static constexpr int SWARM_REDIRECT_LIMIT = 3; + +// The least time worth starting another attempt with. A request given a second or two cannot +// resolve a node, connect and get an answer, so spending the remainder of the budget on it only +// delays telling the caller what we already know. +static constexpr auto MIN_RETRY_BUDGET = 2s; + +struct Core::SwarmOp { + network::x25519_pubkey swarm_pubkey; + std::string endpoint; + std::function(const network::service_node&)> make_body; + std::function on_done; + + // Members already spent on this operation, in the order they were tried: ones that could not + // be reached, and ones that said the account is not theirs. Both are reasons not to ask + // again, and a list rather than a count because "once per member" cannot be expressed as a + // number -- choosing the next one has to know which are gone. + std::vector spent; + + // A member to go back to rather than choosing afresh, for an operation that has to continue + // against the one it started with -- a retrieve continuation resumes from a cursor that is + // that member's alone. Dropped the moment that member turns out to be wrong or unusable, + // which puts us back to choosing normally. + std::optional prefer; + + int redirects = 0; + std::chrono::steady_clock::time_point started = std::chrono::steady_clock::now(); +}; + +void Core::_swarm_request( + network::x25519_pubkey swarm_pubkey, + std::string endpoint, + std::function(const network::service_node&)> make_body, + std::function on_done, + std::optional prefer) { + auto op = std::make_shared( + swarm_pubkey, std::move(endpoint), std::move(make_body), std::move(on_done)); + op->prefer = std::move(prefer); + _swarm_attempt(std::move(op)); +} + +void Core::_swarm_attempt(std::shared_ptr op) { + // Non-owning, as everywhere else here: keeping the Network alive from a callback could make + // the loop thread its last owner and run ~Network there. auto* net = _network.get(); - if (!net) { - log::debug(cat, "Not polling: no network attached"); - return; + if (!net) + return op->on_done({false, network::ERROR_NO_ROUTING_LAYER, "no network attached", {}}); + + // Read out before the call: `op` is moved into the callback below, and the order of those two + // against each other is unspecified, so reading through `op` in the argument list can happen + // after it has been emptied. + auto swarm_pubkey = op->swarm_pubkey; + + net->get_swarm( + swarm_pubkey, + false, + [this, op = std::move(op), net]( + network::swarm::swarm_id_t, std::vector swarm) mutable { + auto fail = [&op](int16_t status, std::string why) { + op->on_done( + {false, + status, + std::move(why), + op->spent.empty() ? network::service_node{} : op->spent.back()}); + }; + + if (swarm.empty()) + return fail(network::ERROR_NO_SNODE_POOL, "no swarm members available"); + + if (op->prefer) { + auto pinned = *op->prefer; + return _swarm_send(std::move(op), std::move(pinned)); + } + + // The first member not already spent. get_swarm shuffles and partitions by strike + // count, so this is the least-struck members first in a random order among equals + // -- the right preference anyway; what matters is only that a member already tried + // is never chosen again, which is what ends the walk. + auto next = std::ranges::find_if(swarm, [&op](const network::service_node& n) { + return std::ranges::find(op->spent, n) == op->spent.end(); + }); + + if (next == swarm.end()) { + log::warning( + cat, + "No swarm member left to try for '{}': all {} are spent.", + op->endpoint, + op->spent.size()); + return fail(network::ERROR_INVALID_DESTINATION, "no usable swarm member"); + } + + _swarm_send(std::move(op), *next); + }); +} + +void Core::_swarm_send(std::shared_ptr op, network::service_node node) { + auto* net = _network.get(); + if (!net) + return op->on_done({false, network::ERROR_NO_ROUTING_LAYER, "no network attached", node}); + + auto elapsed = std::chrono::duration_cast( + std::chrono::steady_clock::now() - op->started); + if (SWARM_OVERALL_TIMEOUT - elapsed < MIN_RETRY_BUDGET) { + log::warning(cat, "Out of time to try another member for '{}'.", op->endpoint); + return op->on_done( + {false, network::ERROR_REQUEST_TIMEOUT, "swarm request budget exhausted", node}); } - log::debug(cat, "Polling swarm for {}", globals.session_id_hex()); + // Rebuilt per attempt: a retrieve carries the chosen node's cursor, so reusing the body built + // for a previous member would resume from a position that member never gave us. + net->send_request( + swarm_request(node, op->swarm_pubkey, op->endpoint, op->make_body(node)), + [this, op, node]( + bool /*success*/, + bool timeout, + int16_t status, + std::vector> /*headers*/, + std::optional body) mutable { + // Onto our own queue: this runs on the *network's* loop, a different thread + // entirely -- Network builds its own quic::Loop -- and everything below reaches + // Core's state, from re-resolving a swarm to whatever `on_done` does with the + // answer. Marshalling here rather than in each caller covers every swarm + // operation at once, since they all come back through this one handler. It also + // means a response landing after Core has gone is dropped rather than run against + // a Core that is being torn down. + call([this, + op = std::move(op), + node = std::move(node), + timeout, + status, + body = std::move(body)]() mutable { + _swarm_response( + std::move(op), std::move(node), timeout, status, std::move(body)); + }); + }); +} - net->get_swarm(globals.pubkey_x25519(), false, [this, net](auto, auto swarm) { - if (swarm.empty()) { - log::warning(cat, "Cannot poll: no swarm nodes available"); - return; +void Core::_swarm_response( + std::shared_ptr op, + network::service_node node, + bool timeout, + int16_t status, + std::optional 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. + 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. + op->spent.push_back(node); + + if (++op->redirects > SWARM_REDIRECT_LIMIT) + log::warning( + cat, + "Giving up on '{}': redirected {} times.", + op->endpoint, + op->redirects - 1); + else { + log::info( + cat, + "{} does not hold {}; re-resolving its swarm.", + node.remote_pubkey.hex(), + op->swarm_pubkey.hex()); + return _swarm_attempt(std::move(op)); } + } - _send_poll(net, swarm.front(), {POLL_NAMESPACES.begin(), POLL_NAMESPACES.end()}, 0); - }); + // The member itself could not be reached. The swarm is not in question, so move to another + // one rather than failing. + else if (status == network::ERROR_INVALID_DESTINATION) { + log::info( + cat, + "{} unreachable for '{}'; trying another member.", + node.remote_pubkey.hex(), + op->endpoint); + op->prefer.reset(); + op->spent.push_back(node); + return _swarm_attempt(std::move(op)); + } + + op->on_done({timeout, status, std::move(body), node}); +} + +void Core::_poll() { + if (!_network) { + log::debug(cat, "Not polling: no network attached"); + return; + } + + log::debug(cat, "Polling swarm for {}", globals.session_id_hex()); + _send_poll({POLL_NAMESPACES.begin(), POLL_NAMESPACES.end()}, 0, std::nullopt); } void Core::fetch_user_profile(std::function done) { @@ -451,11 +764,38 @@ void Core::_conclude_profile_fetch(ProfileFanOut& state) { } void Core::_send_poll( - network::Network* net, - network::service_node node, std::vector namespaces, - int round) { + int round, + std::optional node) { + _swarm_request( + globals.pubkey_x25519(), + "batch", + [this, namespaces](const network::service_node& n) { + return _build_poll_body(n, namespaces); + }, + [this, namespaces, round](SwarmResponse res) mutable { + if (!res.ok() || !res.body) { + log::warning( + cat, + "Swarm poll request failed: {}", + res.timeout ? "timed out" + : res.body ? *res.body + : "request failed"); + return; + } + + // The member that actually answered, which is not necessarily the one the attempt + // started with. Everything below records against it -- the retrieve cursors, and + // the subscription that a drained poll goes on to make. + _swarm_node = res.node; + _handle_poll_response(res.node, std::move(namespaces), std::move(*res.body), round); + }, + std::move(node)); +} + +std::vector Core::_build_poll_body( + const network::service_node& node, const std::vector& namespaces) { auto now_ms = epoch_ms(clock_now_ms()); auto ed25519_hex = globals.pubkey_ed25519().hex(); @@ -522,44 +862,12 @@ SELECT h.hash FROM swarm_hashes h JOIN swarm_nodes n ON n.id = h.node log::debug( cat, - "Retrieving {} namespaces from {} (round {}): {}", + "Retrieving {} namespaces from {}: {}", namespaces.size(), node.remote_pubkey.hex(), - round, body_str); - net->send_request( - swarm_request(node, globals.pubkey_x25519(), "batch", to_vector(body_str)), - [this, node, namespaces = std::move(namespaces), round]( - bool success, - bool timeout, - int16_t /*status_code*/, - std::vector> /*headers*/, - std::optional body) mutable { - if (!success || !body) { - log::warning( - cat, - "Swarm poll request failed: {}", - timeout ? "timed out" - : body ? *body - : "request failed"); - return; - } - - // Onto our own queue: this handler runs on the *network's* loop, which is a - // different thread entirely -- Network builds its own quic::Loop -- and handling a - // poll response merges configs and flushes their dumps, which is only safe on - // ours. It also means a response landing after Core has gone is dropped rather - // than run against a Core that is being torn down. - _jq.call([this, - node = std::move(node), - namespaces = std::move(namespaces), - body = std::move(*body), - round]() mutable { - _handle_poll_response( - std::move(node), std::move(namespaces), std::move(body), round); - }); - }); + return to_vector(body_str); } /// Records where this node's next retrieve of `ns` should resume from. @@ -674,8 +982,13 @@ void Core::_handle_poll_response( return; } - if (unfinished.empty()) + // Nothing reports more, so this node's namespaces are drained and its cursors are current -- + // which is exactly the state a subscription has to start from, or the gap between the last + // retrieve and the subscription taking effect would be lost. + if (unfinished.empty()) { + _maybe_subscribe(node); return; + } if (round + 1 >= POLL_MAX_ROUNDS) { log::warning( @@ -687,10 +1000,304 @@ void Core::_handle_poll_response( return; } - // Deliberately not re-fetching the swarm: the cursor these resume from is this node's, so the - // continuation has to go back to the same one. + // Named rather than chosen afresh: the cursor these resume from is this node's, so the + // continuation has to go back to the same one. If it has since become unusable the swarm + // request falls back to choosing normally, and the round starts over from that member's + // cursor rather than resuming from one it never issued. + _send_poll(std::move(unfinished), round + 1, std::move(node)); +} + +std::optional Core::current_swarm_path() const { + if (!_swarm_node || !_network) + return std::nullopt; + + return _network->get_path_to(*_swarm_node); +} + +void Core::_maybe_subscribe(const network::service_node& node) { + // On our queue because everything below -- the tickers especially -- is Core's loop state. + // Inline once the caller is already there, which the poll response now is, so this costs a + // check rather than a turn of the loop. + call([this, node] { + // Already have one, or are waiting on one: a subscription is with a single node, and + // there is no reason to move while it works. + if (_sub_node) + return; + + auto* net = _network.get(); + if (!net) + return; + + if (!net->supports_server_push()) { + log::debug(cat, "Not subscribing: this routing mode cannot receive pushes"); + return; + } + + _sub_node = node; + _send_subscribe(net, node); + }); +} + +void Core::_send_subscribe(network::Network* net, network::service_node node) { + auto now_s = epoch_ms(clock_now_ms()) / 1000; + + // Mirrors the storage server's sig_msg in handle_monitor_message_single: the literal + // "MONITOR", the 33-byte account pubkey in hex, the timestamp in *seconds*, the want-data + // flag as 0/1, and the namespaces comma-joined in the same order they are sent. + auto to_sign = "MONITOR{}{}{}{}"_format( + globals.session_id_hex(), now_s, 1, fmt::join(SUBSCRIBE_NAMESPACES, ",")); + + b64 sig; + { + auto seed = globals.account_seed(); + ed25519::sign(sig, seed.ed25519_secret(), to_span(to_sign)); + } + + // bt dict keys have to be appended in sorted order: P < d < n < s < t. + oxenc::bt_dict_producer d; + d.append("P", to_string_view(globals.pubkey_ed25519().view())); + // Ask for the message body, not just its metadata. These are the same bytes a retrieve would + // have returned, so carrying them costs nothing over fetching them and saves the round trip: + // a notification is then self-sufficient. + d.append("d", 1); + { + auto ns_list = d.append_list("n"); + for (auto ns : SUBSCRIBE_NAMESPACES) + ns_list.append(ns); + } + d.append("s", to_string_view(sig)); + d.append("t", now_s); + + log::debug(cat, "Subscribing to {} for {}", node.remote_pubkey.hex(), globals.session_id_hex()); + + net->send_request( + swarm_request(node, globals.pubkey_x25519(), "monitor", to_vector(std::move(d).str())), + [this, node]( + bool success, + bool timeout, + int16_t /*status_code*/, + std::vector> /*headers*/, + std::optional body) { + call([this, node, success, timeout, body = std::move(body)] { + // We gave this subscription up while the request was in flight. + if (!_sub_node || _sub_node->remote_pubkey != node.remote_pubkey) + return; + + if (!success || !body) + return _drop_subscription( + timeout ? "subscribe timed out" : "subscribe request failed"); + + // The reply is bt, not the JSON every other storage server endpoint answers + // with: `monitor` is handled outside the RPC dispatch and replies with what + // handle_monitor built. + try { + oxenc::bt_dict_consumer d{*body}; + + if (d.skip_until("errcode")) { + auto code = d.consume_integer(); + std::string err; + if (d.skip_until("error")) + err = d.consume_string(); + return _drop_subscription( + "subscribe rejected (code {}): {}"_format(code, err)); + } + + if (!d.skip_until("success") || d.consume_integer() != 1) + return _drop_subscription("subscribe reply did not report success"); + } catch (const std::exception& e) { + return _drop_subscription( + "could not parse subscribe reply: {}"_format(e.what())); + } + + if (!_subscribed) { + _subscribed = true; + + // Stop polling: from here the node pushes what arrives, and the only + // requests we make are the renew tick's. + if (_poll_ticker) { + _poll_ticker->stop(); + _poll_ticker.reset(); + } + _sub_ticker = _loop.call_every( + SUBSCRIPTION_RENEW_INTERVAL, [this] { _subscription_renew(); }); + _probe_ticker = _loop.call_every( + SUBSCRIPTION_PROBE_INTERVAL, [this] { _subscription_probe(); }); + + log::info( + cat, "Subscribed to {}; polling stopped", node.remote_pubkey.hex()); + + // One last poll, against the member we just subscribed with. + // + // Draining ran before the subscription existed, so a message stored + // between the last retrieve's snapshot and the subscription taking effect + // was in neither: too late for the retrieve, too early to be pushed. This + // is the only thing that closes that window -- the renew sends no + // retrieve, and the probe asks about a namespace that is empty by design. + _send_poll({POLL_NAMESPACES.begin(), POLL_NAMESPACES.end()}, 0, node); + } + }); + }); +} + +void Core::_subscription_renew() { + if (!_sub_node) + return; + if (auto* net = _network.get()) - _send_poll(net, std::move(node), std::move(unfinished), round + 1); + _send_subscribe(net, *_sub_node); +} + +// Asks the subscribed node to retrieve a namespace that is always empty, purely for the 421 we get +// back if it has stopped holding our swarm. +// +// This exists because a subscription that has stopped applying is *silent*. The storage server +// runs no swarm check when subscribing and none when a swarm changes underneath one: +// `get_notifiers` simply stops matching, so a client that has given up polling cannot tell "nothing +// has been sent to me" from "I am subscribed to a node that no longer holds my messages". +// +// It is deliberately the cheapest question that still produces a 421. The storage server decides +// that from the pubkey alone, on the first two lines of its retrieve handler -- before the +// signature-required check and before verifying anything -- so the request needs no signature, no +// ed25519 pubkey and no timestamp, and PROBE_NAMESPACE has nothing in it so it needs no cursor +// either. 97 bytes out, 57 back. +// +// Two things this leans on, neither of them a promised contract: +// +// - that the swarm check precedes the auth check. If the server ever reorders them this stops +// working *silently*, answering 200-with-nothing where it used to answer 421. +// - that no swarm_pubkey is set on the request, which is what stops the network layer from +// quietly retrying a 421 on some other swarm member and reporting success. We are asking about +// this node specifically; an answer from a different one would defeat the point. +// +// Temporary. The storage server is gaining a notification that tells a subscriber outright when +// its subscription has stopped applying, and carries the replacement swarm with it. Once that has +// been deployed widely enough this can go, though it has to outlive the last un-upgraded node. +void Core::_subscription_probe() { + if (!_sub_node) + return; + + auto* net = _network.get(); + if (!net) + return _drop_subscription("network detached"); + + // Logged even though it is uneventful: this and the renewal are the only traffic a subscribed + // client makes, so their absence from a log is the first thing worth checking when pushes + // stop arriving -- and a subscription that has quietly stopped applying looks exactly like a + // conversation nobody is talking in. + log::debug(cat, "Probing {} for a swarm change", _sub_node->remote_pubkey.hex()); + + auto body = + nlohmann::json{ + {"pubkey", globals.session_id_hex()}, + {"namespace", PROBE_NAMESPACE}, + } + .dump(); + + auto node = *_sub_node; + network::Request req{ + node, + "retrieve", + to_vector(body), + network::RequestCategory::standard_small, + PROBE_TIMEOUT}; + + net->send_request( + std::move(req), + [this, node]( + bool success, + bool /*timeout*/, + int16_t status_code, + std::vector> /*headers*/, + std::optional /*body*/) { + if (success) + return; + + call([this, node, status_code] { + if (!_sub_node || _sub_node->remote_pubkey != node.remote_pubkey) + return; + + if (status_code == network::ERROR_MISDIRECTED_REQUEST) + return _drop_subscription("node no longer holds our swarm"); + + // Anything else is the node being unreachable or unwell. A dead connection + // reaches us through on_connection_lost instead, so getting here means it is + // notionally up but not answering, which is no better for a client that has + // nothing else to fall back on. + _drop_subscription("probe failed (status {})"_format(status_code)); + }); + }); +} + +void Core::_drop_subscription(std::string_view why) { + if (!_sub_node) + return; + + log::info(cat, "Dropping subscription with {}: {}", _sub_node->remote_pubkey.hex(), why); + + _sub_node.reset(); + _subscribed = false; + + // Stopped now, but released on a later turn of the loop. This is reachable from inside one of + // these tickers' own callbacks, and a Ticker's deleter runs inline once we are already on the + // loop (Loop::call_every hands out a shared_ptr whose deleter is a `call_get` of the delete, + // and call_get runs inline when inside) -- so dropping the last reference here would free the + // std::function we are currently executing. Stopping is safe from within; freeing is not. + for (auto* ticker : {&_sub_ticker, &_probe_ticker}) { + if (*ticker) { + (*ticker)->stop(); + _loop.reset_soon(std::move(*ticker)); + } + } + + // Back to polling, which is also what picks the next node: the swarm member a fresh + // `get_swarm` happens to hand back first. + _update_polling(); +} + +void Core::_handle_server_push(std::string_view endpoint, std::span body) { + if (endpoint != NOTIFY_ENDPOINT) { + log::debug(cat, "Ignoring pushed '{}': not a notification", endpoint); + return; + } + + std::string hash; + int16_t ns_val; + int64_t timestamp, expiry; + std::string_view data; + + // Keys in the order the server writes them, which is also sorted: @ h n t z ~. `@` (the + // account the message is for) is skipped: we subscribed for one account only. + try { + oxenc::bt_dict_consumer d{to_string_view(body)}; + + hash = d.require("h"); + ns_val = d.require("n"); + timestamp = d.require("t"); + expiry = d.require("z"); + + if (!d.skip_until("~")) { + // We subscribe with d=1, so a notification without a body is the server disagreeing + // with us about what we asked for rather than something to go and fetch. + log::warning(cat, "Pushed notification for {} carried no message data", hash); + return; + } + data = d.consume_string_view(); + } catch (const std::exception& e) { + log::warning(cat, "Could not parse pushed notification: {}", e.what()); + return; + } + + log::debug(cat, "Pushed message {} in namespace {}", hash, ns_val); + + // No cursor is written for a pushed message. The retrieve cursor is per (namespace, node) and + // means "the newest hash that node handed us"; a push did not come from a retrieve, and + // recording it would move the cursor past messages an interrupted retrieve had not yet + // reached. Re-fetching a pushed message after a reconnect is harmless -- delivery is + // at-least-once and Client dedups on the hash -- whereas skipping one is not. + SwarmMessage msg{ + to_span(data), std::move(hash), from_epoch_ms(timestamp), from_epoch_ms(expiry)}; + + receive_messages({&msg, 1}, static_cast(ns_val), true); } PfsKeyStatus Core::prefetch_pfs_keys(std::span session_id) { @@ -762,39 +1369,24 @@ PfsKeyStatus Core::prefetch_pfs_keys(std::span session_id) {"namespace", static_cast(config::Namespace::AccountPubkeys)}, }; - net->get_swarm( + _swarm_request( x25519_pub, - false, - [this, net, sid = std::move(sid), params, x25519_pub](auto, auto swarm) { - if (swarm.empty()) { - log::debug(cat, "prefetch_pfs_keys: get_swarm returned empty swarm"); + "retrieve", + [body = params.dump()](const network::service_node&) { return to_vector(body); }, + [this, sid = std::move(sid)](SwarmResponse res) { + if (!res.ok() || !res.body) { + log::warning( + cat, + "Failed to fetch PFS keys for {}: {}", + sid, + res.timeout ? "timed out" + : res.body ? *res.body + : "request failed"); _pfs_fetch_done(sid, PfsKeyFetch::failed); return; } - auto body_str = params.dump(); - net->send_request( - swarm_request(swarm.front(), x25519_pub, "retrieve", to_vector(body_str)), - [this, sid = std::move(sid)]( - bool success, - bool timeout, - int16_t /*status_code*/, - std::vector> /*headers*/, - std::optional body) { - if (!success || !body) { - log::warning( - cat, - "Failed to fetch PFS keys for {}: {}", - sid, - timeout ? "timed out" - : body ? *body - : "request failed"); - _pfs_fetch_done(sid, PfsKeyFetch::failed); - return; - } - - return _handle_pfs_response(sid, std::move(*body)); - }); + return _handle_pfs_response(sid, std::move(*res.body)); }); return status; } @@ -950,52 +1542,34 @@ void Core::delete_from_swarm( }; auto body = to_vector(params.dump()); - net->get_swarm( + _swarm_request( globals.pubkey_x25519(), - false, - [this, net, hashes = std::move(hashes), body = std::move(body), on_complete]( - auto, auto swarm) mutable { - if (swarm.empty()) { - log::warning(cat, "Cannot delete from swarm: no swarm nodes available"); + "delete", + [body = std::move(body)](const network::service_node&) { return body; }, + [this, hashes = std::move(hashes), on_complete](SwarmResponse res) { + if (!res.ok()) { + log::warning( + cat, + "Swarm delete failed ({}): {}", + res.timeout ? "timed out" : "status {}"_format(res.status_code), + res.body.value_or("no response body")); if (on_complete) on_complete(false); return; } - net->send_request( - swarm_request( - swarm.front(), globals.pubkey_x25519(), "delete", std::move(body)), - [this, hashes = std::move(hashes), on_complete]( - bool success, - bool timeout, - int16_t status, - auto, - std::optional resp) { - if (!success) { - log::warning( - cat, - "Swarm delete failed ({}): {}", - timeout ? "timed out" : "status {}"_format(status), - resp.value_or("no response body")); - if (on_complete) - on_complete(false); - return; - } - - // Forget the cursors naming what we just deleted, so the next retrieve - // measures from the newest hash the node still holds. Done on success - // only: a failed delete leaves the messages there, and dropping the - // cursor would replay the retention window for nothing. - { - auto conn = db.conn(); - for (const auto& h : hashes) - conn.prepared_exec( - "DELETE FROM swarm_hashes WHERE hash = ?", h); - } - - if (on_complete) - on_complete(true); - }); + // Forget the cursors naming what we just deleted, so the next retrieve measures + // from the newest hash the node still holds. Done on success only: a failed + // delete leaves the messages there, and dropping the cursor would replay the + // retention window for nothing. + { + auto conn = db.conn(); + for (const auto& h : hashes) + conn.prepared_exec("DELETE FROM swarm_hashes WHERE hash = ?", h); + } + + if (on_complete) + on_complete(true); }); } @@ -1064,62 +1638,38 @@ void Core::_send_to_swarm( network::x25519_pubkey x25519_pub; std::memcpy(x25519_pub.data(), dest_pubkey.data() + 1, 32); - net->get_swarm( + _swarm_request( x25519_pub, - false, - [net, body = std::move(body), on_complete = std::move(on_complete), x25519_pub]( - auto, auto swarm) mutable { - if (swarm.empty()) { - log::warning(cat, "Cannot store: no swarm nodes available"); - if (on_complete) - on_complete(false, std::nullopt); - return; - } - // The two values a 421 turns on: which swarm we resolved, and which of its nodes we - // picked. Read this against the "Storing ... of " line above -- a store - // rejected as misdirected means those two pubkeys are not the same account. + "store", + [body = std::move(body), x25519_pub](const network::service_node& node) { + // Read this against the "Storing ... of " line above: a store rejected as + // misdirected means the pubkey in the body and the swarm we resolved are not the + // same account, which no amount of trying other members will fix. log::debug( - cat, - "Storing to swarm of {} via {} ({} nodes)", - x25519_pub.hex(), - swarm.front().to_string(), - swarm.size()); + cat, "Storing to swarm of {} via {}", x25519_pub.hex(), node.to_string()); + return body; + }, + [on_complete = std::move(on_complete)](SwarmResponse res) { + if (!res.ok()) + log::warning( + cat, + "Store request failed ({}): {}", + res.timeout ? "timed out" : "status {}"_format(res.status_code), + res.body.value_or("no response body")); + if (!on_complete) + return; - net->send_request( - swarm_request(swarm.front(), x25519_pub, "store", std::move(body)), - [on_complete = std::move(on_complete)]( - bool success, - bool timeout, - int16_t status, - auto, - std::optional resp) { - if (!success) - log::warning( - cat, - "Store request failed ({}): {}", - timeout ? "timed out" : "status {}"_format(status), - resp.value_or("no response body")); - if (!on_complete) - return; - - std::optional hash; - if (success && resp) { - try { - auto json = nlohmann::json::parse(*resp); - if (auto h = json.find("hash"); - h != json.end() && h->is_string()) - hash = h->get(); - } catch (const std::exception& e) { - log::warning( - cat, - "Could not read stored message hash: {}", - e.what()); - } - } - on_complete( - success, - hash ? std::optional{*hash} : std::nullopt); - }); + std::optional hash; + if (res.ok() && res.body) { + try { + auto json = nlohmann::json::parse(*res.body); + if (auto h = json.find("hash"); h != json.end() && h->is_string()) + hash = h->get(); + } catch (const std::exception& e) { + log::warning(cat, "Could not read stored message hash: {}", e.what()); + } + } + on_complete(res.ok(), hash ? std::optional{*hash} : std::nullopt); }); } diff --git a/src/core/configs.cpp b/src/core/configs.cpp index 1e00b9fc..754b8efe 100644 --- a/src/core/configs.cpp +++ b/src/core/configs.cpp @@ -89,28 +89,43 @@ std::vector Configs::all() { _local.get()}; } +// Each of these schedules a settle, because handing out the reference is the last thing that +// happens before a caller may change what it points at, and it is the only thing this layer sees. +// A config is mutated through that reference; nothing tells us afterwards, and asking the config to +// tell us does not work either -- its own "needs dump" flag is set *before* the assignment that +// follows, so anything acting on it would serialise a change that has not happened yet. +// +// Reads schedule one too, since a reader and a writer ask the same question. That costs a job that +// finds every config clean and does nothing: `store_dumps` skips what has not changed and +// `needs_push` is five state reads. + config::UserProfile& Configs::user_profile() { _load(); + _schedule_settle(); return *_user_profile; } config::Contacts& Configs::contacts() { _load(); + _schedule_settle(); return *_contacts; } config::ConvoInfoVolatile& Configs::convo_info_volatile() { _load(); + _schedule_settle(); return *_convo_info_volatile; } config::UserGroups& Configs::user_groups() { _load(); + _schedule_settle(); return *_user_groups; } config::Local& Configs::local() { _load(); + _schedule_settle(); return *_local; } @@ -149,8 +164,17 @@ Configs::Batch::Batch(Configs& configs) : _configs{configs} { } Configs::Batch::~Batch() { - if (--_configs._batch_depth == 0) + if (--_configs._batch_depth != 0) + return; + + // `_flush` writes to the database, so it can throw -- and this runs during unwinding whenever + // the work inside the batch threw, where a second exception is a call to std::terminate. The + // changes stay dirty, so the next thing to settle writes them. + try { _configs._flush(); + } catch (const std::exception& e) { + log::warning(cat, "Could not flush configs at the end of a batch: {}", e.what()); + } } void Configs::_flush() { @@ -159,8 +183,9 @@ void Configs::_flush() { store_dumps(); - // Unconditional rather than only after a merge, so that the batch a poll holds doubles as a - // sweep: a config changed locally without one gets noticed here rather than sitting unpushed. + // Unconditional rather than only after a merge, so that this is the one place that decides a + // push is owed: a settle scheduled by an accessor lands here knowing only that someone held a + // config, and whether that left anything to publish is this check's to answer. if (needs_push()) _schedule_push(); @@ -257,6 +282,34 @@ bool Configs::needs_push() { // is nothing to be gained by expiring it sooner. static constexpr auto CONFIG_TTL = 30 * 24h; +void Configs::_schedule_settle() { + // Not while Core is still being built. The loop thread is already running by then, but the + // constructing thread is legitimately off it -- that is what `on_loop()` allows for -- so a job + // posted here would serialise a config on the loop while construction is still writing to it. + // Nothing is owed: construction flushes what it changes itself. + if (!core._constructed) + return; + + // Once per turn of the loop, however many configs were handed out and however many fields were + // touched in each. + if (_settle_scheduled) + return; + _settle_scheduled = true; + + // `call_soon` rather than running it here, and that is the whole of why this works: the caller + // is holding a reference it has not written through yet -- `dirty()` bumps the seqno and marks + // the config before the assignment that follows it -- so there is no moment during the accessor + // at which the config is whole. Once the job that took the reference has returned, there is. + jq().call_soon([this] { + _settle_scheduled = false; + try { + _flush(); + } catch (const std::exception& e) { + log::warning(cat, "Could not settle config changes: {}", e.what()); + } + }); +} + void Configs::_schedule_push() { auto now = std::chrono::steady_clock::now(); _last_change = now; @@ -393,55 +446,79 @@ void Configs::_send_push() { _push_in_flight = true; - net->get_swarm( + core._swarm_request( core.globals.pubkey_x25519(), - false, - [this, - net, - alive = std::weak_ptr{_alive}, - pending = std::move(pending), - body = std::move(body)](auto, auto swarm) mutable { + "sequence", + [body = std::move(body)](const network::service_node&) { return body; }, + [this, alive = std::weak_ptr{_alive}, pending = std::move(pending)]( + Core::SwarmResponse res) { if (alive.expired()) return; - if (swarm.empty()) { - log::warning(cat, "Cannot push configs: no swarm nodes available"); - // Onto Core's queue like everything else here: this handler runs on the - // Network's own loop, which is a different thread, and the flag is ours. - jq().call([this] { _push_in_flight = false; }); + _push_in_flight = false; + + if (!res.ok() || !res.body) { + log::warning( + cat, + "Config push failed ({}): {}", + res.timeout ? "timed out" : "status {}"_format(res.status_code), + res.body.value_or("no response body")); + return; + } + + // A config is confirmed only if *every* message it split into was stored. + // Confirming a partial push would drop the parts that did land from the obsolete + // list while leaving the config believing it is clean, so the missing part would + // never be sent again. + try { + auto json = nlohmann::json::parse(*res.body); + auto results = json.find("results"); + if (results == json.end() || !results->is_array()) { + log::warning(cat, "Config push response carried no results"); + return; + } + + for (const auto& p : pending) { + std::unordered_set hashes; + bool stored = true; + for (size_t i = p.first; stored && i < p.first + p.count; i++) { + if (i >= results->size()) { + stored = false; + break; + } + const auto& r = (*results)[i]; + auto code = r.find("code"); + auto b = r.find("body"); + if (code == r.end() || code->get() != 200 || b == r.end()) { + stored = false; + break; + } + auto h = b->find("hash"); + if (h == b->end() || !h->is_string()) { + stored = false; + break; + } + hashes.insert(h->get()); + } + + if (!stored) { + log::warning( + cat, + "Config push: {} was not stored, leaving it dirty", + p.conf->encryption_domain()); + continue; + } + p.conf->confirm_pushed(p.seqno, std::move(hashes)); + } + } catch (const std::exception& e) { + log::warning(cat, "Could not read config push response: {}", e.what()); return; } - net->send_request( - swarm_request( - swarm.front(), - core.globals.pubkey_x25519(), - "sequence", - std::move(body)), - [this, alive, pending = std::move(pending)]( - bool success, - bool timeout, - int16_t status, - auto, - std::optional resp) mutable { - // The canary first, because reaching `jq()` at all means touching - // this: the Network owns this callback, so it can outlive Core, and - // cancelling the queue cannot reach something that was never on it. - if (alive.expired()) - return; - jq().call([this, - pending = std::move(pending), - success, - timeout, - status, - resp = std::move(resp)]() mutable { - _handle_push_response( - std::move(pending), - success, - timeout, - status, - std::move(resp)); - }); - }); + // Confirming changes the configs' state, and a change that arrived while this was + // in flight has re-dirtied them. + store_dumps(); + if (needs_push()) + _schedule_push(); }); } diff --git a/src/network/network_config.cpp b/src/network/network_config.cpp index 73b3b25b..55c50bfc 100644 --- a/src/network/network_config.cpp +++ b/src/network/network_config.cpp @@ -141,11 +141,6 @@ void Config::handle_config_opt(opt::disable_subnet_diversity) { log::debug(cat, "Network config disabled subnet diversity"); } -void Config::handle_config_opt(opt::redirect_retry_count rrc) { - redirect_retry_count = rrc.count; - log::debug(cat, "Network config redirect retry count set to {}", rrc.count); -} - void Config::handle_config_opt(opt::retry_delay rd) { retry_delay = std::move(rd); log::debug( @@ -249,6 +244,14 @@ void Config::handle_config_opt(opt::quic_handshake_timeout qht) { log::debug(cat, "Network config quic handshake timeout set to {}ms", qht.duration.count()); } +void Config::handle_config_opt(opt::quic_tunnel_handshake_timeout qtht) { + quic_tunnel_handshake_timeout = qtht.duration; + log::debug( + cat, + "Network config quic tunnelled handshake timeout set to {}ms", + qtht.duration.count()); +} + void Config::handle_config_opt(opt::quic_keep_alive qka) { quic_keep_alive = qka.duration; log::debug(cat, "Network config quic keep alive set to {}s", qka.duration.count()); diff --git a/src/network/routing/onion_request_router.cpp b/src/network/routing/onion_request_router.cpp index 5b1b4e02..08ae8b81 100644 --- a/src/network/routing/onion_request_router.cpp +++ b/src/network/routing/onion_request_router.cpp @@ -534,16 +534,25 @@ void OnionRequestRouter::clear_cache() { }); } -std::vector OnionRequestRouter::get_active_paths() { - return _jq.call_get([this] { - std::vector result; - result.reserve(_paths.size()); - - for (const auto& [category, path_list] : _paths) - for (const auto& p : path_list) - result.push_back({p.nodes, OnionPathMetadata{category}}); - - return result; +std::optional OnionRequestRouter::get_path_to(const service_node& node) { + return _jq.call_get([this, &node]() -> std::optional { + // An onion-request path is built before any destination is chosen and carries requests to + // all of them, so the answerable question is which path a swarm request to this node + // would go down were one sent now. Asked of the selection the sending path itself uses, + // rather than reimplemented: the choice turns on strike counts, how busy each path is, + // and skipping any path that contains the destination, and a second copy of that would + // drift into reporting a path requests do not take. + auto* path = _find_valid_path( + &node, RequestCategory::standard_small, std::nullopt, "path query"); + if (!path) + return std::nullopt; + + PathInfo info; + info.hops.reserve(path->nodes.size()); + for (const auto& n : path->nodes) + info.hops.push_back({n.remote_pubkey, n.ip}); + + return info; }); } @@ -1487,6 +1496,18 @@ void OnionRequestRouter::_on_edge_connectivity_response( } OnionPath* OnionRequestRouter::_find_valid_path(const Request& request) { + return _find_valid_path( + std::get_if(&request.destination), + request.category, + request.desired_path_index, + request.request_id); +} + +OnionPath* OnionRequestRouter::_find_valid_path( + const service_node* target_node, + RequestCategory category, + std::optional desired_path_index, + std::string_view request_id) { // If we are in `single_path_mode` then just return the first path we have (don't care about // category as there should only be one path) if (_config.single_path_mode) { @@ -1496,7 +1517,7 @@ OnionPath* OnionRequestRouter::_find_valid_path(const Request& request) { return nullptr; } - auto it = _paths.find(to_path_category(request.category)); + auto it = _paths.find(to_path_category(category)); if (it == _paths.end() || it->second.empty()) return nullptr; @@ -1504,15 +1525,13 @@ OnionPath* OnionRequestRouter::_find_valid_path(const Request& request) { std::vector suitable_paths; suitable_paths.reserve(candidate_paths.size()); - auto target_node = std::get_if(&request.destination); - // We want to allow explicit path selection for client-side automated tests so if a // `desired_path_index` has been specified then use it - if (request.desired_path_index) { - if (candidate_paths.size() < *request.desired_path_index) + if (desired_path_index) { + if (candidate_paths.size() < *desired_path_index) return nullptr; - return &candidate_paths[*request.desired_path_index]; + return &candidate_paths[*desired_path_index]; } for (OnionPath& path : candidate_paths) { @@ -1536,7 +1555,7 @@ OnionPath* OnionRequestRouter::_find_valid_path(const Request& request) { cat, "[Request {}]: Path destination conflicts with the only available path, " "but single_path_mode is enabled, proceeding.", - request.request_id); + request_id); else if (conflict) continue; } @@ -1558,7 +1577,7 @@ OnionPath* OnionRequestRouter::_find_valid_path(const Request& request) { }); OnionPath* best_path = suitable_paths.front(); - const auto min_paths_for_type = _config.min_path_counts[to_path_category(request.category)]; + const auto min_paths_for_type = _config.min_path_counts[to_path_category(category)]; // Return the path with the fewest active requests if we had one with no requests, or // already have the minimum number of paths for this type diff --git a/src/network/routing/session_router_router.cpp b/src/network/routing/session_router_router.cpp index 52e00431..41ab2ba7 100644 --- a/src/network/routing/session_router_router.cpp +++ b/src/network/routing/session_router_router.cpp @@ -12,6 +12,7 @@ #include #include "session/crypto/ed25519.hpp" +#include "session/format.hpp" #include "session/network/network_opt.hpp" #include "session/onionreq/builder.hpp" #include "session/onionreq/response_parser.hpp" @@ -37,6 +38,11 @@ struct ActiveTunnel { static std::optional pubkey_from_srouter_address(std::string_view address); +// The name Session Router knows a storage node by: its ed25519 pubkey in base32z, plus ".snode". +static std::string srouter_address(std::span remote_pubkey) { + return "{:a}.snode"_format(remote_pubkey); +} + // The inner QUIC connection's UDP payload size, fixed rather than derived from the tunnel's // suggestion. // @@ -236,9 +242,35 @@ void SessionRouter::clear_cache() { // TODO: Implement this. } -std::vector SessionRouter::get_active_paths() { - // TODO: Implement this. - return {}; +std::optional SessionRouter::get_path_to(const service_node& node) { + if (!srouter) + return std::nullopt; + + // Deliberately the single-session lookup rather than get_all_session_paths(): we hold a + // session to every swarm member we have spoken to, to the file server, and to every group's + // swarm, and reporting all of them answers a question nobody asked. + auto hops = srouter->get_path_for_session(srouter_address(node.remote_pubkey)); + if (!hops) + return std::nullopt; + + PathInfo info; + info.hops.reserve(hops->size()); + + for (const auto& [address, ip] : *hops) { + auto pubkey = pubkey_from_srouter_address(address); + if (!pubkey) { + log::warning(cat, "Omitting path hop with an unparseable address: {}", address); + continue; + } + + try { + info.hops.push_back({*pubkey, oxen::quic::ipv4{ip}}); + } catch (const std::exception& e) { + log::warning(cat, "Omitting path hop {} with an unparseable ip {}", address, ip); + } + } + + return info; } void SessionRouter::send_request(Request request, network_response_callback_t callback) { @@ -1258,12 +1290,6 @@ void SessionRouter::_establish_tunnel( // return r; // } - std::string srouter_address; - srouter_address.reserve(oxenc::to_base32z_size(remote_pubkey.size()) + ".snode"sv.size()); - oxenc::to_base32z( - remote_pubkey.begin(), remote_pubkey.end(), std::back_inserter(srouter_address)); - srouter_address += ".snode"sv; - // srouter::RouterID router_id{remote_pubkey.first<32>()}; // auto snode_address = "34d9udo9ethfcrcaxcgdyxsi1w8gr79jzornsytcfgdw5rpmif8y.loki";// // address.to_network_address(true); @@ -1277,7 +1303,7 @@ void SessionRouter::_establish_tunnel( initiating_req_id, address_pubkey_hex); auto tunnel = srouter->establish_udp( - srouter_address, + srouter_address(remote_pubkey), test_port, [weak_self = weak_from_this(), this, address_pubkey_hex, initiating_req_id]( router::tunnel_info info) mutable { @@ -1432,6 +1458,7 @@ void SessionRouter::_send_via_tunnel( request.category, request.time_remaining(), remaining_overall_timeout}; + router_request.tunnelled = true; transport->send_request(std::move(router_request), std::move(callback)); } diff --git a/src/network/session_network.cpp b/src/network/session_network.cpp index cce07479..1a6cbd36 100644 --- a/src/network/session_network.cpp +++ b/src/network/session_network.cpp @@ -89,11 +89,19 @@ namespace { main_config.cache_min_swarm_size, main_config.cache_num_nodes_to_use_for_refresh, main_config.cache_min_num_refresh_presence_to_include_node, - main_config.cache_node_strike_threshold}; + main_config.cache_node_strike_threshold, + // Session Router only. Onion requests reach a storage server through relays that + // do not care what it runs, and `direct` talks to it straight, so in neither case + // does the version predict anything -- preferring on it there would narrow the + // swarm for no reason. + main_config.router == opt::router::Type::session_router + ? std::optional{opt::MIN_SESSION_ROUTER_SS_VERSION} + : std::nullopt}; } config::QuicTransport build_quic_transport_config(const config::Config& main_config) { return {main_config.quic_handshake_timeout, + main_config.quic_tunnel_handshake_timeout, main_config.quic_keep_alive, main_config.quic_max_udp_payload}; } @@ -192,6 +200,16 @@ Network::Network(config::Config _conf) : // The SnodePool is needed regardless of the transport layer as it includes swarm information // which is needed by the clients in order to send requests + // + // This fetcher goes straight to the transport, bypassing the router: it is what fills an empty + // snode cache, and onion requests cannot be built before there is one. Hence the seed list -- + // a bootstrap that has to leak the client's IP is at least aimed at nodes chosen in advance. + // + // TODO: session-router mode does not need this exemption. It bootstraps itself onto the + // network before it can carry anything of ours, so a tunnel to a seed node is available at the + // point this runs, and routing the bootstrap would close the one hole in that mode's IP + // guarantee. Left direct for now because the routed fetcher is installed further down, after + // the router exists. auto bootstrap_fetcher = [bt = std::weak_ptr{_transport}]( Request req, network_response_callback_t on_complete) { if (auto transport = bt.lock()) @@ -267,6 +285,24 @@ Network::Network(config::Config _conf) : _router->on_status_changed = [this] { _recalculate_status(); }; _transport->on_status_changed = [this] { _recalculate_status(); }; + // Pass the transport's inbound signals through to whoever owns us. Read each hook once into + // a local: our owner may replace it, and these fire on the loop rather than from the thread + // that would be doing the replacing. + _transport->on_server_push = [this](const ed25519_pubkey& node, + std::string_view endpoint, + std::span body) { + if (auto cb = on_server_push) + cb(node, endpoint, body); + }; + _transport->on_connection_established = [this](const ed25519_pubkey& node) { + if (auto cb = on_connection_established) + cb(node); + }; + _transport->on_connection_lost = [this](const ed25519_pubkey& node) { + if (auto cb = on_connection_lost) + cb(node); + }; + // Perform a clock resync _jq->call_soon([this] { _resync_clock(std::nullopt, nullptr); }); } @@ -409,11 +445,11 @@ ConnectionStatus Network::get_status() { return _status.load(); } -std::vector Network::get_active_paths() { +std::optional Network::get_path_to(const service_node& node) { if (_router) - return _router->get_active_paths(); + return _router->get_path_to(node); - return {}; + return std::nullopt; } void Network::get_swarm( @@ -490,71 +526,52 @@ void Network::send_request(Request request, network_response_callback_t callback auto processed_request = _preprocess_request(std::move(request)); // Bare `this`: the router is destroyed by ~Network before our queue is stopped, so it // cannot still be holding this callback by the time any of our state goes. - auto router_callback = - [this, original_req = processed_request, cb = std::move(callback)]( - bool success, bool timeout, int16_t status_code, auto headers, auto body) { - const auto dest_is_snode = - std::holds_alternative(original_req.destination); - - // 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); - - int16_t final_status_code = status_code; - - if (body) - if (auto uniform_error = response::find_uniform_batch_error(*body)) - final_status_code = *uniform_error; - - // 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) || - (final_status_code == ERROR_TOO_EARLY && !dest_is_snode)) { - _resync_clock(std::move(original_req), std::move(cb)); - return; - } - - // If we got a 421 then our swarm info is out of data so we need to refresh our - // 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)); - return; - } - - // The node itself could not be reached -- no relay contact for it, so session - // router cannot carry anything there. The swarm is not in question, so the - // request moves to the next member rather than being failed. Without this the - // first send to a node that does not participate dies, and the node is only - // struck out *afterwards*, so the cost is one dead request per such node. - if (final_status_code == ERROR_INVALID_DESTINATION && dest_is_snode && - original_req.swarm_pubkey) { - _retry_next_swarm_node( - std::move(original_req), - timeout, - status_code, - std::move(headers), - std::move(body), - std::move(cb)); - return; - } - - // 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_421_count > 0) - log::info( - cat, - "[Request {}] Received valid response after 421 retry.", - original_req.request_id); + auto router_callback = [this, original_req = processed_request, cb = std::move(callback)]( + bool success, + bool timeout, + int16_t status_code, + auto headers, + auto body) { + const auto dest_is_snode = + std::holds_alternative(original_req.destination); + + // 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); + + int16_t final_status_code = status_code; + + if (body) + if (auto uniform_error = response::find_uniform_batch_error(*body)) + final_status_code = *uniform_error; + + // 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) || + (final_status_code == ERROR_TOO_EARLY && !dest_is_snode)) { + _resync_clock(std::move(original_req), std::move(cb)); + return; + } - auto final_success = - (success && final_status_code >= 200 && final_status_code <= 299); - cb(final_success, timeout, status_code, std::move(headers), std::move(body)); - }; + // A 421 says this node does not hold the account we asked about, and its body + // carries the swarm that does. Take the correction -- it is our cache and the + // answer is authoritative -- but do not act on it: which member to ask next, + // and whether to ask at all, is the caller's to decide, and only the caller + // can know which node it ended up talking to. + if (final_status_code == 421 && dest_is_snode && original_req.swarm_pubkey && body) + _adopt_swarm_from_421(*original_req.swarm_pubkey, *body); + + // `final_status_code`, not the raw one: a batch whose subrequests all failed + // the same way arrives here as a transport-level 200, and reporting that + // would leave the caller unable to tell a misdirected request from any other + // failure -- which is exactly the decision it is now responsible for making. + auto final_success = (success && final_status_code >= 200 && final_status_code <= 299); + cb(final_success, timeout, final_status_code, std::move(headers), std::move(body)); + }; _router->send_request(std::move(processed_request), std::move(router_callback)); } catch (const std::exception& e) { @@ -769,6 +786,13 @@ Request Network::_preprocess_request(Request request) { } void Network::_update_network_state(const std::string& body) { + // Not every storage server endpoint answers in JSON: `monitor` is handled outside the RPC + // dispatch and replies with bt, which carries no clock or fork versions to read anyway. + // Recognised rather than parsed and complained about, since a subscription renews on a timer + // and would otherwise log a warning every time. + if (!body.empty() && (body.front() == 'd' || body.front() == 'l')) + return; + try { auto json = nlohmann::json::parse(body); const nlohmann::json* target_json = &json; @@ -847,209 +871,46 @@ void Network::_update_network_state(const std::string& body) { // MARK: Specific Error Handling -// The least time worth starting another attempt with. A request given a second or two cannot -// realistically resolve a node, connect and get an answer, so spending the remainder of the budget -// on it only delays telling the caller what we already know. -static constexpr auto MIN_RETRY_BUDGET = 2s; - -void Network::_retry_next_swarm_node( - Request original_request, - bool timeout, - int16_t status_code, - std::vector> headers, - std::optional body, - network_response_callback_t final_callback) { - - auto* failed_node = std::get_if(&original_request.destination); - if (!failed_node || !original_request.swarm_pubkey) - return final_callback(false, timeout, status_code, std::move(headers), std::move(body)); - - original_request.failed_nodes.push_back(*failed_node); - auto swarm_pubkey = *original_request.swarm_pubkey; - - // Deliberately not refreshing the snode cache first, which is what the 421 path does: nothing - // here suggests our swarm information is stale, only that one member of it is unreachable. The - // swarm comes back from the cache, so this costs nothing and returns the same members in the - // same order. - // - // The failure that got us here is carried into the callback rather than referenced from out - // here: this returns before get_swarm answers, so anything left behind would be gone by then. - _snode_pool->get_swarm( - swarm_pubkey, - false, - [this, - req = std::move(original_request), - cb = std::move(final_callback), - timeout, - status_code, - headers = std::move(headers), - body = std::move(body)]( - swarm::swarm_id_t, std::vector swarm_nodes) mutable { - // Reports the failure that got us here rather than one of our own invention: the - // caller wants to know why the request did not go through, and "no members left" - // says less than the reason each of them was unusable. - auto give_up = [&] { - cb(false, timeout, status_code, std::move(headers), std::move(body)); - }; - - // The first member that has not already failed. get_swarm shuffles and then - // partitions by strike count, so this is not a fixed order -- what it gives is the - // least-struck members first, in a random order among equals. That is the right - // preference anyway; what matters here is only that a member already spent is - // never chosen again, which is what ends the walk. - auto next = std::ranges::find_if(swarm_nodes, [&](const service_node& node) { - return std::ranges::find(req.failed_nodes, node) == req.failed_nodes.end(); - }); +// Takes the swarm a 421 hands back and writes it into the cache, so that whoever decides to try +// again resolves against the corrected membership rather than the stale one that misdirected us. +// +// Only the cache is touched. Choosing another member, or giving up, is the caller's: it is the +// only party that can know which node it ended up talking to, and a substitution made down here +// is invisible to it. +void Network::_adopt_swarm_from_421(const x25519_pubkey& swarm_pubkey, std::string_view body) { + try { + auto json = nlohmann::json::parse(body); - if (next == swarm_nodes.end()) { - log::warning( - cat, - "[Request {}] No swarm member left to try: all {} were unreachable.", - req.request_id, - req.failed_nodes.size()); - return give_up(); - } + // A batch collapses to a uniform 421, in which case the swarm sits inside the first + // subrequest's body rather than at the top level. + if (auto results = json.find("results"); + results != json.end() && results->is_array() && !results->empty()) + if (auto b = results->front().find("body"); b != results->front().end()) + json = *b; - auto chosen = next->to_string(); - auto retry = std::move(req); - retry.destination = std::move(*next); - - // Each attempt gets the per-request timeout or whatever is left of the operation's - // overall budget, whichever is shorter -- so walking the swarm cannot outlive what - // the caller asked for, however many members turn out to be unusable. The budget - // runs from the *original* request's creation, which a re-send carries with it, so - // time spent on earlier members counts against later ones. - if (retry.overall_timeout) { - auto spent = std::chrono::duration_cast( - std::chrono::steady_clock::now() - retry.creation_time); - auto left = *retry.overall_timeout - spent; - - if (left < MIN_RETRY_BUDGET) { - log::warning( - cat, - "[Request {}] Out of time to try another swarm member ({}ms left " - "of {}ms).", - retry.request_id, - left.count(), - retry.overall_timeout->count()); - return give_up(); - } + auto snodes = json.find("snodes"); + if (snodes == json.end() || !snodes->is_array() || snodes->empty()) + return; - retry.request_timeout = std::min(retry.request_timeout, left); - } + std::vector nodes; + nodes.reserve(snodes->size()); + for (const auto& n : *snodes) + nodes.push_back(service_node::from_json(n)); - log::info( - cat, - "[Request {}] Node unreachable, retrying on {} with {}ms ({} already " - "tried).", - retry.request_id, - chosen, - retry.request_timeout.count(), - retry.failed_nodes.size()); - - send_request(std::move(retry), std::move(cb)); - }); -} + swarm::swarm_id_t swarm_id = swarm::INVALID_SWARM_ID; + if (auto s = json.find("swarm"); s != json.end() && s->is_string()) + swarm_id = std::stoull(s->get(), nullptr, 16); -void Network::_handle_421_retry( - Request original_request, network_response_callback_t final_callback) { - if (original_request.retry_421_count >= config.redirect_retry_count) { - log::error( + log::info( cat, - "Request {} received 421 but exceeded max retry count.", - original_request.request_id); - return final_callback( - false, - false, - ERROR_MISDIRECTED_REQUEST, - {content_type_plain_text}, - "Exceeded retry limit for 421 error"); - } + "Adopting the {}-node swarm a 421 reported for {}", + nodes.size(), + swarm_pubkey.hex()); - // Shouldn't automatically retry if the destination isn't a node (we on'y want to auto-retry due - // to a node being in the wrong swarm) - auto* original_dest_node = std::get_if(&original_request.destination); - if (!original_dest_node) - return final_callback( - false, - false, - ERROR_MISDIRECTED_REQUEST, - {content_type_plain_text}, - "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) { - log::warning( - cat, - "Request {} received 421 but carries no swarm pubkey to re-resolve.", - original_request.request_id); - return final_callback( - false, - false, - ERROR_MISDIRECTED_REQUEST, - {content_type_plain_text}, - "421 Misdirected Request for a request with no swarm"); + _snode_pool->set_swarm(swarm_pubkey, swarm_id, std::move(nodes)); + } catch (const std::exception& e) { + log::warning(cat, "Could not read the swarm out of a 421 response: {}", e.what()); } - - // 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) - log::info( - cat, - "Request {} received 421 from node {}, refreshing swarm if stale.", - 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_421_count++; - final_request.destination = std::move(swarm_nodes[new_target]); - this->send_request(std::move(final_request), std::move(cb)); - }); - }); } void Network::_resync_clock( @@ -1358,7 +1219,6 @@ LIBSESSION_C_API session_network_config session_network_config_default() { config.increase_no_file_limit = cpp_defaults.increase_no_file_limit; config.path_length = cpp_defaults.path_length; config.enforce_subnet_diversity = cpp_defaults.enforce_subnet_diversity; - config.redirect_retry_count = cpp_defaults.redirect_retry_count; config.min_retry_delay_ms = cpp_defaults.retry_delay.base_delay.count(); config.max_retry_delay_ms = cpp_defaults.retry_delay.max_delay.count(); config.num_nodes_to_check_for_network_offset = @@ -1401,6 +1261,10 @@ LIBSESSION_C_API session_network_config session_network_config_default() { config.quic_handshake_timeout_seconds = std::chrono::duration_cast(cpp_defaults.quic_handshake_timeout) .count(); + config.quic_tunnel_handshake_timeout_seconds = + std::chrono::duration_cast( + cpp_defaults.quic_tunnel_handshake_timeout) + .count(); config.quic_keep_alive_seconds = std::chrono::duration_cast(cpp_defaults.quic_keep_alive).count(); config.quic_disable_mtu_discovery = cpp_defaults.quic_max_udp_payload.has_value(); @@ -1488,9 +1352,6 @@ LIBSESSION_C_API bool session_network_init( std::chrono::milliseconds{config->min_retry_delay_ms}, std::chrono::milliseconds{config->max_retry_delay_ms}}); - // A `0` value is valid for this option - cpp_opts.emplace_back(opt::redirect_retry_count{config->redirect_retry_count}); - if (config->num_nodes_to_check_for_network_offset > 0) cpp_opts.emplace_back(opt::num_nodes_to_check_for_network_offset{ config->num_nodes_to_check_for_network_offset}); @@ -1586,6 +1447,10 @@ LIBSESSION_C_API bool session_network_init( cpp_opts.emplace_back(opt::quic_handshake_timeout{ std::chrono::seconds{config->quic_handshake_timeout_seconds}}); + if (config->quic_tunnel_handshake_timeout_seconds > 0) + cpp_opts.emplace_back(opt::quic_tunnel_handshake_timeout{ + std::chrono::seconds{config->quic_tunnel_handshake_timeout_seconds}}); + if (config->quic_keep_alive_seconds > 0) cpp_opts.emplace_back(opt::quic_keep_alive{ std::chrono::seconds{config->quic_keep_alive_seconds}}); @@ -1718,110 +1583,6 @@ LIBSESSION_C_API CONNECTION_STATUS session_network_get_status(network_object* ne return static_cast(unbox(network)->get_status()); } -LIBSESSION_C_API void session_network_get_active_paths( - network_object* network, session_path_info** out_paths, size_t* out_paths_len) { - if (!network || !out_paths || !out_paths_len) - return; - - *out_paths = nullptr; - *out_paths_len = 0; - - try { - std::vector cpp_paths = unbox(network)->get_active_paths(); - if (cpp_paths.empty()) - return; - - // Calculate the size of the data - size_t total_size = cpp_paths.size() * sizeof(session_path_info); - size_t total_nodes = 0; - for (const auto& path : cpp_paths) - total_nodes += path.nodes.size(); - total_size += total_nodes * sizeof(network_service_node); - - size_t total_metadata_size = 0; - for (const auto& p : cpp_paths) { - std::visit( - [&](const T&) { - if constexpr (std::is_same_v) - total_metadata_size += sizeof(session_onion_path_metadata); - else { - static_assert(std::is_same_v); - total_metadata_size += sizeof(session_router_tunnel_metadata); - } - }, - p.metadata); - } - total_size += total_metadata_size; - - // Allocate and assign the memory - unsigned char* buffer = static_cast(std::malloc(total_size)); - if (!buffer) - return; - - auto* c_paths_array = reinterpret_cast(buffer); - auto* current_node_ptr = - reinterpret_cast(c_paths_array + cpp_paths.size()); - unsigned char* current_metadata_ptr = - reinterpret_cast(current_node_ptr + total_nodes); - - for (size_t i = 0; i < cpp_paths.size(); ++i) { - const auto& cpp_path = cpp_paths[i]; - auto& c_path = c_paths_array[i]; - - new (&c_path) session_path_info{}; - - c_path.nodes = current_node_ptr; - c_path.nodes_count = cpp_path.nodes.size(); - for (const auto& cpp_node : cpp_path.nodes) { - new (current_node_ptr) network_service_node{}; - cpp_node.into(*current_node_ptr); - current_node_ptr++; - } - - // Copy metadata - std::visit( - [&](const T& m) { - if constexpr (std::is_same_v) { - auto* meta = reinterpret_cast( - current_metadata_ptr); - new (meta) session_onion_path_metadata{}; - meta->category = static_cast(m.category); - c_path.onion_metadata = meta; - current_metadata_ptr += sizeof(session_onion_path_metadata); - } else { - static_assert(std::is_same_v); - auto* meta = reinterpret_cast( - current_metadata_ptr); - new (meta) session_router_tunnel_metadata{}; - strncpy(meta->destination_pubkey, - m.destination_pubkey.c_str(), - sizeof(meta->destination_pubkey) - 1); - meta->destination_pubkey[sizeof(meta->destination_pubkey) - 1] = '\0'; - strncpy(meta->destination_snode_address, - m.destination_snode_address.c_str(), - sizeof(meta->destination_snode_address) - 1); - meta->destination_snode_address - [sizeof(meta->destination_snode_address) - 1] = '\0'; - c_path.session_router_metadata = meta; - current_metadata_ptr += sizeof(session_router_tunnel_metadata); - } - }, - cpp_path.metadata); - } - - *out_paths = c_paths_array; - *out_paths_len = cpp_paths.size(); - } catch (...) { - *out_paths = nullptr; - *out_paths_len = 0; - } -} - -LIBSESSION_C_API void session_network_paths_free(session_path_info* paths) { - if (paths) - std::free(paths); -} - LIBSESSION_C_API void session_network_get_swarm( network_object* network, const char* swarm_pubkey_hex, diff --git a/src/network/snode_pool.cpp b/src/network/snode_pool.cpp index a38ee693..e23f99d1 100644 --- a/src/network/snode_pool.cpp +++ b/src/network/snode_pool.cpp @@ -1174,6 +1174,20 @@ std::vector SnodePool::get_unused_nodes( }); } +void SnodePool::set_swarm( + session::network::x25519_pubkey swarm_pubkey, + swarm_id_t swarm_id, + std::vector nodes) { + _jq.call([this, swarm_pubkey, swarm_id, nodes = std::move(nodes)]() mutable { + log::info( + cat, + "Overriding cached swarm for {} with {} authoritative node(s).", + swarm_pubkey.hex(), + nodes.size()); + _swarm_cache[swarm_pubkey] = {swarm_id, std::move(nodes)}; + }); +} + void SnodePool::get_swarm( session::network::x25519_pubkey swarm_pubkey, bool ignore_strike_count, @@ -1198,6 +1212,20 @@ void SnodePool::get_swarm( return get_strike_count(node) < _config.cache_node_strike_threshold; }); + // Within the ones we would use, put the preferred versions first -- again stable, so + // each subset keeps its shuffled order. Strikes stay the outer split: one is a node + // that has actually failed us, whereas the version is only a prediction about whether + // it can be reached at all. + // + // Ordering rather than filtering, and only the retained set is touched, so everything + // below about adopting struck nodes to make up the numbers is unaffected -- a swarm + // with no preferred members still hands back exactly what it did before. + if (_config.prefer_min_version) + std::ranges::stable_partition( + nodes.begin(), over_nodes.begin(), [&](const auto& node) { + return node.storage_server_version >= *_config.prefer_min_version; + }); + auto under_count = nodes.size() - over_nodes.size(); if (over_nodes.empty()) { // Nothing we can do even if we want more diff --git a/src/network/transport/quic_transport.cpp b/src/network/transport/quic_transport.cpp index 6f27681d..9235e959 100644 --- a/src/network/transport/quic_transport.cpp +++ b/src/network/transport/quic_transport.cpp @@ -26,6 +26,10 @@ namespace { case RequestCategory::standard_small: return true; case RequestCategory::file: return false; case RequestCategory::file_small: return true; + // Small enough to qualify for the reserved stream, and deliberately kept off it: a + // config push is the one swarm request that can be large, and stream 0 also carries + // the polls, the stores and the server's pushes back to us. + case RequestCategory::config: return false; } return false; // Shouldn't happen } @@ -109,7 +113,10 @@ void QuicTransport::verify_connectivity( if (_pending_requests.count(pubkey_hex) == 0 && _pending_verification_callbacks.at(pubkey_hex).size() == 1) _establish_connection( - {node.remote_pubkey.view(), node.host(), node.omq_port}, request_id, category); + {node.remote_pubkey.view(), node.host(), node.omq_port}, + request_id, + category, + false); }); } @@ -267,15 +274,19 @@ void QuicTransport::_send_request_internal(Request request, network_response_cal "[Request {}] No connection to {}, initiating new connection.", request.request_id, remote_pubkey_hex); + // Everything the connect needs has to be read before the request is moved into the queue. std::string initiating_req_id = request.request_id; + auto category = request.category; + bool tunnelled = request.tunnelled; _pending_requests[remote_pubkey_hex].emplace_back(std::move(request), std::move(callback)); - _establish_connection(*remote, initiating_req_id, request.category); + _establish_connection(*remote, initiating_req_id, category, tunnelled); } void QuicTransport::_establish_connection( const oxen::quic::RemoteAddress& address, const std::string& initiating_req_id, - const RequestCategory /*category*/) { + const RequestCategory /*category*/, + bool tunnelled) { const auto address_pubkey_hex = oxenc::to_hex(address.view_remote_key()); try { @@ -302,7 +313,8 @@ void QuicTransport::_establish_connection( address, creds, oxen::quic::opt::outbound_alpn(ALPN), - oxen::quic::opt::handshake_timeout{_config.handshake_timeout}, + oxen::quic::opt::handshake_timeout{ + tunnelled ? _config.tunnel_handshake_timeout : _config.handshake_timeout}, oxen::quic::opt::keep_alive{_config.keep_alive}, // libquic hands these a live Connection, so they run inline on the loop rather than // as jobs of ours. ~QuicTransport destroys the endpoint on the loop before the @@ -317,6 +329,32 @@ void QuicTransport::_establish_connection( auto stream = conn.open_stream(); auto conn_id = conn.reference_id(); auto stream_id = stream->stream_id(); + + // Anything the far end sends us of its own accord arrives here. Registered + // generically rather than per endpoint name because what those names mean is + // the storage server's business, not the transport's. + // + // Caught rather than left to propagate: this runs inside libquic's stream + // machinery, where an exception would tear down the connection for a fault in + // a consumer's handler. + stream->register_generic_handler( + [this, address_pubkey_hex](oxen::quic::message msg) { + if (!on_server_push) + return; + try { + on_server_push( + ed25519_pubkey::from_hex(address_pubkey_hex), + msg.endpoint(), + msg.body()); + } catch (const std::exception& e) { + log::error( + cat, + "Handler for pushed '{}' from {} threw: {}", + msg.endpoint(), + address_pubkey_hex, + e.what()); + } + }); auto it = _pending_verification_callbacks.find(address_pubkey_hex); decltype(it->second) verification_callbacks; if (it != _pending_verification_callbacks.end()) { @@ -346,6 +384,21 @@ void QuicTransport::_establish_connection( _send_on_connection( conn_id, address_pubkey_hex, std::move(req), std::move(cb)); } + + // Last, so that anything already waiting on this connection goes out ahead of + // whatever the listener sends, and so that the connection is in + // `_active_connection_ids` by the time it does. + if (on_connection_established) { + try { + on_connection_established(ed25519_pubkey::from_hex(address_pubkey_hex)); + } catch (const std::exception& e) { + log::error( + cat, + "Connection-established listener for {} threw: {}", + address_pubkey_hex, + e.what()); + } + } }, [this, address_pubkey_hex, initiating_req_id]( oxen::quic::Connection&, uint64_t error_code) { @@ -608,10 +661,19 @@ void QuicTransport::_fail_connection( auto to_fail = std::move(it->second); _failure_listeners.erase(it); - for (const auto& listener : it->second) + for (const auto& listener : to_fail) listener(); } + if (on_connection_lost) { + try { + on_connection_lost(ed25519_pubkey::from_hex(address_pubkey_hex)); + } catch (const std::exception& e) { + log::error( + cat, "Connection-lost listener for {} threw: {}", address_pubkey_hex, e.what()); + } + } + // If we have no longer have any active connections then we are disconnected if (_active_connection_ids.empty()) _update_status(ConnectionStatus::disconnected); diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 4a6a1443..211d83a6 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -51,6 +51,7 @@ set(LIB_SESSION_UTESTS_SOURCES test_pro_backend.cpp test_random.cpp test_session_encrypt.cpp + test_network_teardown.cpp test_swarm_retry.cpp test_utils.cpp test_session_protocol.cpp diff --git a/tests/test_client/conversation_api.cpp b/tests/test_client/conversation_api.cpp index fcba4c25..7057b2f4 100644 --- a/tests/test_client/conversation_api.cpp +++ b/tests/test_client/conversation_api.cpp @@ -56,6 +56,37 @@ TEST_CASE("Client: a conversation reports the settings it carries", "[client][co CHECK(convo().exp_timer() == 0s); } +TEST_CASE("Client: a nickname too long to sync is refused rather than stored", "[client][convos]") { + TempClient c; + auto them = "05" + std::string(64, 'a'); + auto id = dm_from_hex(them); + c->open_dm(id, await); + + c->dm(id, await)->set_nickname("Bilbo", await); + + // One byte over is enough. What must not happen is the row being written and the config then + // refusing it: a sync rebuilds the whole entry from that row, so it would never carry again -- + // taking every later change to this contact with it. + std::string too_long(config::contact_info::MAX_NAME_LENGTH + 1, 'x'); + REQUIRE(config::validate_contact_name(too_long).has_value()); + CHECK_THROWS_AS(c->dm(id, await)->set_nickname(too_long, await), std::invalid_argument); + + // Nothing moved: not the database, and not the config it is reconciled into. + CHECK(c->conversation(id, await)->dm()->nickname == "Bilbo"); + CHECK(in_configs(*c, [&](auto& cfg) { return cfg.contacts().get(them); })->nickname == "Bilbo"); + + // And the contact still syncs, which is the part that would have been lost quietly. + c->set_blocked(id, true, await); + CHECK(in_configs(*c, [&](auto& cfg) { return cfg.contacts().get(them); })->blocked); + + // Exactly at the limit is fine: the check is the config's own, not a stricter one. + std::string at_limit(config::contact_info::MAX_NAME_LENGTH, 'y'); + CHECK_FALSE(config::validate_contact_name(at_limit).has_value()); + c->dm(id, await)->set_nickname(at_limit, await); + CHECK(in_configs(*c, [&](auto& cfg) { return cfg.contacts().get(them); })->nickname == + at_limit); +} + TEST_CASE("Client: settings from another device reach the conversation", "[client][configs]") { TempClient c; auto them = "05" + std::string(64, 'b'); diff --git a/tests/test_core_configs.cpp b/tests/test_core_configs.cpp index 61ac2ecf..bd4eb45b 100644 --- a/tests/test_core_configs.cpp +++ b/tests/test_core_configs.cpp @@ -633,6 +633,104 @@ TEST_CASE("Configs: a change schedules a push rather than sending one", "[core][ }); } +TEST_CASE( + "Configs: a change nobody announced is still written and pushed", "[core][configs][push]") { + TempCore c; + + // An account's own creation writes the defaults, which is a change like any other. Let that + // reach disk first, so the only thing left to dump is what this test does. + TestHelper::drain(*c); + + // All in one job, because nothing may run between clearing the pending push and making the + // change: with no network attached the defaults can never be confirmed, so `needs_push()` stays + // true and any settle in between would arm the timer again. + TestHelper::on_loop(*c, [&] { + TestHelper::backdate_push_state(c->configs, 3s, 4s); + TestHelper::push_if_due(c->configs); + REQUIRE_FALSE(TestHelper::push_scheduled(c->configs)); + REQUIRE_FALSE(c->configs.user_profile().needs_dump()); + + // A bare change: no batch around it, nothing merged, nothing polling. That is the shape + // every Client setter makes, and it used to reach neither disk nor the swarm -- the only + // thing that armed the timer was a caller announcing a run of changes was over, and no + // local caller ever did. It survived because a poll held a batch every few seconds and + // swept it up. + c->configs.user_profile().set_name("Leia"); + + // Still nothing: the settle is queued rather than run, because this job is still holding + // the reference it changed through and the config is mid-change until it returns. + CHECK(c->configs.user_profile().needs_dump()); + CHECK_FALSE(TestHelper::push_scheduled(c->configs)); + }); + + // One turn of the loop, with nothing else prompting it. + TestHelper::drain(*c); + + TestHelper::on_loop(*c, [&] { + CHECK_FALSE(c->configs.user_profile().needs_dump()); + CHECK(TestHelper::push_scheduled(c->configs)); + }); +} + +TEST_CASE("Configs: a locally made change survives a restart", "[core][configs]") { + TempCore c; + + TestHelper::on_loop(*c, [&] { c->configs.user_profile().set_name("Leia"); }); + TestHelper::drain(*c); + + reopen(c); + + CHECK(TestHelper::on_loop(*c, [&] { + return std::string{c->configs.user_profile().get_name().value_or("")}; + }) == "Leia"); +} + +TEST_CASE( + "Configs: changes in quick succession make one push, not several", + "[core][configs][push]") { + PushableCore c; + c->configs.push_debounce = 2s; + c->configs.push_max_delay = 10s; + + // Three changes, each in its own job, the way three calls from an application arrive. Each one + // settles separately; what must not happen is three pushes, or three timers racing each other. + for (std::string_view name : {"Leia", "Leia Organa", "General Organa"}) + TestHelper::on_loop(*c.core, [&] { c->configs.user_profile().set_name(name); }); + TestHelper::drain(*c.core); + + TestHelper::on_loop(*c.core, [&] { + REQUIRE(TestHelper::push_scheduled(c->configs)); + REQUIRE(c.net->sent_requests.empty()); + }); + + SECTION("a further change pushes the deadline out again") { + TestHelper::on_loop(*c.core, [&] { + // Quiet long enough that it would go out right now -- and then touched again, which is + // what has to move the deadline. Backdated past the threshold deliberately: at 1.9s it + // would be held back whether or not the change reset anything, and the test would pass + // without testing. + TestHelper::backdate_push_state(c->configs, 3s, 4s); + c->configs.user_profile().set_name("Leia Skywalker"); + }); + TestHelper::drain(*c.core); + + TestHelper::on_loop(*c.core, [&] { + TestHelper::push_if_due(c->configs); + CHECK(c.net->sent_requests.empty()); + CHECK(TestHelper::push_scheduled(c->configs)); + }); + } + + SECTION("quiet for long enough sends one request carrying all of them") { + TestHelper::on_loop(*c.core, [&] { + TestHelper::backdate_push_state(c->configs, 3s, 4s); + TestHelper::push_if_due(c->configs); + CHECK(c.net->sent_requests.size() == 1); + CHECK(c->configs.user_profile().get_name() == "General Organa"); + }); + } +} + TEST_CASE("Configs: the debounce waits for quiet, up to a limit", "[core][configs][push]") { PushableCore c; c->configs.push_debounce = 2s; diff --git a/tests/test_helper.hpp b/tests/test_helper.hpp index 47122a6f..4567e064 100644 --- a/tests/test_helper.hpp +++ b/tests/test_helper.hpp @@ -50,9 +50,51 @@ class MockNetwork : public network::Network { // `current_node` when set. std::vector current_swarm; + /// Set to answer requests as they are sent rather than leaving them in `sent_requests` for the + /// test to fire by hand. Return nullopt to leave one pending. + /// + /// Called with the request; the tuple is (success, timeout, status, body). Requests are still + /// recorded either way, so a test can assert on what was sent as well as script the answer. + using Reply = std::tuple>; + std::function(const network::Request&)> auto_reply; + + /// The Core this is attached to, set by attach_mock_network, so that a response driven by hand + /// can be delivered on Core's loop as a real one is. Null only for a MockNetwork built + /// directly, which is what the Network-level tests do. + core::Core* core = nullptr; + void send_request( network::Request request, network::network_response_callback_t callback) override { - sent_requests.push_back({std::move(request), std::move(callback)}); + std::optional scripted; + if (auto_reply) + scripted = auto_reply(request); + + // Wrapped so that answering lands on Core's loop, which is where a real response is + // handled: the Network delivers on its own thread and Core marshals it across. Done here + // rather than at each answering helper because a test may fire `sent_requests[i].callback` + // itself, and what that prompts -- continuing a swarm walk, writing a cache entry -- would + // otherwise sit on the queue until something else happened to run it. `call_get` is inline + // once already on the loop, so scripted replies, which are sent from there, cost nothing. + network::network_response_callback_t on_loop = + [this, callback = std::move(callback)]( + bool ok, + bool timeout, + int16_t status, + std::vector> headers, + std::optional body) { + if (!core) + return callback(ok, timeout, status, std::move(headers), std::move(body)); + core->call_get([&] { + callback(ok, timeout, status, std::move(headers), std::move(body)); + }); + }; + + sent_requests.push_back({std::move(request), on_loop}); + + if (scripted) { + auto [ok, timeout, status, body] = std::move(*scripted); + on_loop(ok, timeout, status, {}, std::move(body)); + } } void get_swarm( @@ -110,7 +152,9 @@ class MockNetwork : public network::Network { /// Network outright -- nothing else may hold it alive -- so a test that goes on poking at the mock /// keeps a raw pointer rather than a second reference. inline MockNetwork* attach_mock_network(core::Core& core) { - return &core.make_network(); + auto& net = core.make_network(); + net.core = &core; + return &net; } /// Answers every captured download with `data`, delivered in chunks as a transport would rather @@ -326,7 +370,15 @@ class FakeRouter : public network::IRouter { class TestHelper { public: - static void poll(core::Core& core) { core._poll(); } + /// Polls the way the ticker does: on the loop. + /// + /// Not `core._poll()` on the caller's thread. A poll reaches component state, which is the + /// loop's alone, and everything the walk does afterwards keys off being there already -- + /// `JobQueue::call` runs inline inside the loop and defers outside it, so a poll driven from a + /// test's own thread would answer its first member and leave the rest of the walk queued. + static void poll(core::Core& core) { + core.call_get([&core] { core._poll(); }); + } /// Runs `f` on Core's loop and hands back what it returned. /// diff --git a/tests/test_network_teardown.cpp b/tests/test_network_teardown.cpp new file mode 100644 index 00000000..d672d552 --- /dev/null +++ b/tests/test_network_teardown.cpp @@ -0,0 +1,64 @@ +#include +#include +#include +#include + +#include "test_helper.hpp" + +using namespace session; +using namespace session::network; +using namespace std::literals; + +TEST_CASE( + "Network: an owner reference dropped mid-callback does not tear the Network down from its " + "own loop", + "[network]") { + auto net = std::make_shared(network::config::Config{}); + + // get_swarm answers from the SnodePool's loop, so the callback below runs on the loop thread + // rather than on this one. (This used to be reached through the swarm retry, which answered + // from the loop for the same reason; that has moved to Core, so the vehicle is different but + // the interleaving under test is the same.) + auto swarm_pubkey = x25519_pubkey::from_hex(std::string(64, 'a')); + TestHelper::seed_swarm( + TestHelper::snode_pool(*net), + swarm_pubkey, + {service_node{ + ed25519_pubkey::from_hex(std::string(64, 'b')), + oxen::quic::ipv4{127, 0, 0, 1}, + 1001, + 2001, + {2, 8, 0}, + 0, + 0}}); + + auto reached_callback = std::promise{}; + auto in_callback = reached_callback.get_future(); + std::atomic answered = false; + + net->get_swarm(swarm_pubkey, false, [&reached_callback, &answered](auto, auto swarm) { + answered = !swarm.empty(); + reached_callback.set_value(); + + // Stay on the loop thread while the reference below goes, which is the interleaving that + // used to abort: a callback holding a shared_ptr of its own meant dropping the + // owner's left the loop thread as the last owner, and ~Network joins that thread. + std::this_thread::sleep_for(50ms); + }); + + REQUIRE(in_callback.wait_for(5s) == std::future_status::ready); + + auto observer = std::weak_ptr{net}; + net.reset(); + + // Waits for the Network to actually be gone rather than merely unreferenced from here: the + // teardown is what fails, so it has to happen while this test is still running. Nothing else + // holds a reference, so this returns as soon as the callback has finished. + for (int i = 0; i < 500 && !observer.expired(); i++) + std::this_thread::sleep_for(10ms); + + // Surviving to here is the assertion: the failure was an abort out of a destructor rather than + // a wrong answer. + CHECK(observer.expired()); + CHECK(answered); +} diff --git a/tests/test_snode_pool.cpp b/tests/test_snode_pool.cpp index fccc0628..e7446b13 100644 --- a/tests/test_snode_pool.cpp +++ b/tests/test_snode_pool.cpp @@ -65,6 +65,14 @@ class TestSnodePool : public SnodePool { // loop thread void update_cache(std::vector nodes) { _update_cache("test", std::move(nodes)); } + // Puts a swarm straight into the cache so get_swarm answers from it rather than resolving one. + void seed_swarm(const x25519_pubkey& pubkey, std::vector nodes) { + _jq.call_get([&] { + _swarm_cache[pubkey] = {swarm::swarm_id_t{0}, std::move(nodes)}; + return 0; + }); + } + void debug_on_refresh_complete(std::vector> raw_results) { auto total_requests = static_cast(raw_results.size()); _jq.call_get([&] { @@ -332,3 +340,103 @@ 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][swarm_version_preference]") { + constexpr std::array old_ss{2, 11, 0}; + constexpr std::array new_ss{2, 11, 1}; + + auto make_config = [](std::optional> prefer) { + session::network::config::SnodePool config = { + std::nullopt, + std::nullopt, + std::chrono::minutes{5}, + std::chrono::minutes{5}, + false, + network::opt::retry_delay{50ms, 200ms}, + opt::netid::Target::testnet, + {}, + 0, + 0, + 3, + 0, + 3}; + config.prefer_min_version = prefer; + return config; + }; + + // Alternating, so that grouping by version can only come from the preference rather than from + // the order they were put in. + auto member = [](uint8_t n, std::array version) { + return service_node{ + ed25519_pubkey::from_hex("{:02x}{}"_format(n, std::string(62, '0'))), + oxen::quic::ipv4{"192.168.0.{}"_format(n)}, + static_cast(20000 + n), + static_cast(30000 + n), + version, + 0}; + }; + std::vector swarm; + for (uint8_t i = 0; i < 6; i++) + swarm.push_back(member(i, i % 2 ? new_ss : old_ss)); + + auto pubkey = x25519_pubkey::from_hex(std::string(64, 'a')); + + auto loop = std::make_shared(); + auto disk_loop = std::make_shared(); + + auto ordered = [&](TestSnodePool& pool) { + std::vector got; + pool.get_swarm(pubkey, false, [&got](auto, std::vector nodes) { + got = std::move(nodes); + }); + // get_swarm answers from the cache on the pool's own queue; this waits for that to drain. + pool.pending_post_refresh_callbacks(); + return got; + }; + + SECTION("preferred versions come first") { + auto pool = std::make_shared(make_config(new_ss), *loop, *disk_loop); + pool->seed_swarm(pubkey, swarm); + + auto got = ordered(*pool); + REQUIRE(got.size() == swarm.size()); + + // Asserted as a boundary rather than a fixed order: each subset keeps its shuffled order, + // so which preferred node comes first is deliberately not fixed. + for (size_t i = 0; i < got.size(); i++) + CHECK((got[i].storage_server_version >= new_ss) == (i < 3)); + } + + SECTION("a swarm with nothing preferred is still usable") { + std::vector all_old; + for (uint8_t i = 0; i < 4; i++) + all_old.push_back(member(i, old_ss)); + + auto pool = std::make_shared(make_config(new_ss), *loop, *disk_loop); + pool->seed_swarm(pubkey, all_old); + + // Ordering, not filtering: preferring what none of them are must not empty the swarm. + CHECK(ordered(*pool).size() == all_old.size()); + } + + SECTION("without a preference nothing is lost") { + auto pool = std::make_shared(make_config(std::nullopt), *loop, *disk_loop); + pool->seed_swarm(pubkey, swarm); + CHECK(ordered(*pool).size() == swarm.size()); + } + + SECTION("strikes still outrank the version") { + auto pool = std::make_shared(make_config(new_ss), *loop, *disk_loop); + pool->seed_swarm(pubkey, swarm); + + // A node that has actually failed us is worse than one merely predicted to be unreachable, + // so striking out every preferred member puts them behind the rest. + for (const auto& n : swarm) + if (n.storage_server_version >= new_ss) + pool->record_node_failure(n, /*permanent=*/true); + + auto got = ordered(*pool); + REQUIRE(!got.empty()); + CHECK(got.front().storage_server_version == old_ss); + } +} diff --git a/tests/test_swarm_retry.cpp b/tests/test_swarm_retry.cpp index 1a07419d..2eee3c30 100644 --- a/tests/test_swarm_retry.cpp +++ b/tests/test_swarm_retry.cpp @@ -1,7 +1,6 @@ #include -#include +#include #include -#include #include "test_helper.hpp" @@ -9,10 +8,13 @@ using namespace session; using namespace session::network; using namespace std::literals; +// Re-aiming a swarm request is Core's, not Network's: only Core can know whether being answered by +// a different member matters to it, and a substitution made below Core is invisible to the +// bookkeeping that depends on it. These drive it through the poll, which is a real caller rather +// than a harness, so what is asserted is the behaviour a caller actually gets. + namespace { -/// A swarm member. Only the pubkey distinguishes them here; the addresses are never dialled, -/// because FakeRouter answers without going anywhere. std::string key_hex(uint8_t n) { return fmt::format("{:02x}{}", n, std::string(62, '0')); } @@ -28,192 +30,133 @@ service_node node_at(uint8_t n) { 0}; } -/// A Network with its router replaced and one swarm primed, which is the least a test needs to -/// exercise anything Network does above routing. -struct ScriptedNetwork { - std::shared_ptr net; - std::shared_ptr router = std::make_shared(); - x25519_pubkey swarm_pubkey; - std::vector swarm; +/// Who each request was addressed to, in order. +std::vector tried(const MockNetwork& net) { + std::vector out; + for (const auto& s : net.sent_requests) + out.push_back(std::get(s.request.destination).remote_pubkey); + return out; +} - explicit ScriptedNetwork(size_t members) { - net = std::make_shared(network::config::Config{}); - swarm_pubkey = x25519_pubkey::from_hex(key_hex(0xAA)); +/// Whether every entry is distinct -- what "once per member" means, given get_swarm hands members +/// back in a shuffled order rather than a fixed one. +bool all_distinct(std::vector keys) { + std::ranges::sort(keys, [](const auto& a, const auto& b) { return a.hex() < b.hex(); }); + return std::ranges::adjacent_find(keys) == keys.end(); +} - for (size_t i = 0; i < members; i++) - swarm.push_back(node_at(static_cast(i + 1))); +/// A batch response that says nothing was found, so a poll treats the member as drained. +std::string empty_batch(const Request& req) { + auto batch = parse_json(*req.body); + auto results = nlohmann::json::array(); + for (size_t i = 0; i < batch["requests"].size(); i++) + results.push_back({{"code", 200}, {"body", {{"messages", nlohmann::json::array()}}}}); + return nlohmann::json{{"results", std::move(results)}}.dump(); +} - TestHelper::set_router(*net, router); - TestHelper::seed_swarm(TestHelper::snode_pool(*net), swarm_pubkey, swarm); - } +struct PollFixture { + TempCore core; + MockNetwork* net; - /// A request addressed to the swarm, starting at whichever member the caller would have picked. - Request to(const service_node& first, std::optional overall = 60s) { - Request req{first, "store", std::vector{}, RequestCategory::standard_small, 10s}; - req.swarm_pubkey = swarm_pubkey; - req.overall_timeout = overall; - return req; + explicit PollFixture(size_t members) : net{attach_mock_network(*core)} { + for (size_t i = 0; i < members; i++) + net->current_swarm.push_back(node_at(static_cast(i + 1))); } - /// The answer is delivered from the loop, not from send_request, so this waits for it. The - /// promise is shared rather than captured by reference: if the callback never comes, a - /// reference to a local here would dangle rather than merely time out. - std::pair send(Request req) { - auto done = std::make_shared>>(); - auto waiter = done->get_future(); - net->send_request(std::move(req), [done](bool ok, bool, int16_t status, auto, auto) { - done->set_value({ok, status}); - }); - REQUIRE(waiter.wait_for(5s) == std::future_status::ready); - return waiter.get(); - } + void poll() { TestHelper::poll(*core); } }; } // namespace -/// Whether every entry is distinct -- what "once per node" means, given get_swarm hands members -/// back in a shuffled order rather than a fixed one. -bool all_distinct(const std::vector& tried) { - auto sorted = tried; - std::ranges::sort(sorted, [](const auto& a, const auto& b) { return a.hex() < b.hex(); }); - return std::ranges::adjacent_find(sorted) == sorted.end(); +TEST_CASE("Core: an unreachable member moves the request to the next one", "[core][swarm]") { + PollFixture f{4}; + + // Only one member is reachable; the rest have no relay contact, which is what session routing + // reports as an invalid destination rather than as a failure of the request. + auto reachable = f.net->current_swarm[2].remote_pubkey; + f.net->auto_reply = [&](const Request& req) -> std::optional { + if (std::get(req.destination).remote_pubkey == reachable) + return MockNetwork::Reply{true, false, 200, empty_batch(req)}; + return MockNetwork::Reply{false, false, ERROR_INVALID_DESTINATION, "unreachable"}; + }; + + f.poll(); + + // It reached the one that works, spending no member twice on the way. Which it tried first is + // deliberately not asserted: get_swarm shuffles, so the order is not fixed. + auto attempts = tried(*f.net); + REQUIRE(attempts.size() >= 2); + CHECK(attempts.size() <= f.net->current_swarm.size()); + CHECK(attempts.back() == reachable); + CHECK(all_distinct(attempts)); } -TEST_CASE("Network: an unreachable node moves the request to the next swarm member", "[network]") { - ScriptedNetwork n{4}; - - // Only one member participates in session routing; the rest have no relay contact. - n.router->replies[n.swarm[2].remote_pubkey] = {}; - - auto [ok, status] = n.send(n.to(n.swarm[0])); - CHECK(ok); - CHECK(status == 200); +TEST_CASE("Core: running out of members ends the walk", "[core][swarm]") { + PollFixture f{3}; - // It reached the one that works, having spent no member twice on the way. Which members it - // tried first is not asserted: get_swarm shuffles, so the order is deliberately not fixed. - REQUIRE(n.router->tried.size() >= 2); - CHECK(n.router->tried.size() <= n.swarm.size()); - CHECK(n.router->tried.back() == n.swarm[2].remote_pubkey); - CHECK(all_distinct(n.router->tried)); -} - -TEST_CASE("Network: running out of swarm members reports the original failure", "[network]") { - ScriptedNetwork n{3}; - // Nobody answers. + f.net->auto_reply = [](const Request&) -> std::optional { + return MockNetwork::Reply{false, false, ERROR_INVALID_DESTINATION, "unreachable"}; + }; - auto [ok, status] = n.send(n.to(n.swarm[0])); - CHECK_FALSE(ok); - // The reason each member was unusable, not "no members left" -- which would tell the caller - // less than what it already had. - CHECK(status == ERROR_INVALID_DESTINATION); + f.poll(); // Every member tried, once each: it ends when selection has nothing left rather than at a // fixed count, and never revisits one already spent. - REQUIRE(n.router->tried.size() == 3); - CHECK(all_distinct(n.router->tried)); + auto attempts = tried(*f.net); + CHECK(attempts.size() == 3); + CHECK(all_distinct(attempts)); } -TEST_CASE("Network: a failure that is not the node's fault is not retried elsewhere", "[network]") { - ScriptedNetwork n{3}; +TEST_CASE( + "Core: a failure that is not the member's fault is not retried elsewhere", + "[core][swarm]") { + PollFixture f{3}; // A 500 says the request was carried and the server disliked it. Asking a different member of // the same swarm the same question gets the same answer, so this is not what the walk is for. - n.router->replies[n.swarm[0].remote_pubkey] = {false, false, 500, "nope"}; - - auto [ok, status] = n.send(n.to(n.swarm[0])); - CHECK_FALSE(ok); - CHECK(status == 500); - CHECK(n.router->tried.size() == 1); -} - -TEST_CASE("Network: a request with no swarm has nowhere else to go", "[network]") { - ScriptedNetwork n{3}; + f.net->auto_reply = [](const Request&) -> std::optional { + return MockNetwork::Reply{false, false, 500, "nope"}; + }; - // Something aimed at a node rather than at an account -- a cache refresh, a clock resync -- - // has no swarm to walk, so the failure is simply reported. - auto req = n.to(n.swarm[0]); - req.swarm_pubkey.reset(); + f.poll(); - auto [ok, status] = n.send(std::move(req)); - CHECK_FALSE(ok); - CHECK(status == ERROR_INVALID_DESTINATION); - CHECK(n.router->tried.size() == 1); + CHECK(tried(*f.net).size() == 1); } -TEST_CASE("Network: attempts are bounded by the overall budget", "[network]") { - SECTION("each attempt gets the per-request timeout while there is budget for it") { - ScriptedNetwork n{3}; - n.send(n.to(n.swarm[0], 60s)); - - REQUIRE(n.router->timeouts.size() == 3); - for (auto t : n.router->timeouts) - CHECK(t == 10s); +TEST_CASE("Core: a misdirected request is re-aimed at another member", "[core][swarm]") { + PollFixture f{3}; + + // A 421 says this member does not hold the account. Unlike an unreachable member it says our + // swarm information was wrong, so Core re-resolves rather than merely stepping along -- but + // either way the member that said it must not be asked again. + auto wrong = f.net->current_swarm[0].remote_pubkey; + f.net->auto_reply = [&](const Request& req) -> std::optional { + if (std::get(req.destination).remote_pubkey == wrong) + return MockNetwork::Reply{false, false, ERROR_MISDIRECTED_REQUEST, "wrong swarm"}; + return MockNetwork::Reply{true, false, 200, empty_batch(req)}; + }; + + f.poll(); + + auto attempts = tried(*f.net); + REQUIRE(attempts.size() >= 1); + if (attempts.front() == wrong) { + REQUIRE(attempts.size() == 2); + CHECK(attempts.back() != wrong); } +} - SECTION("a shrinking budget shortens the retry rather than overrunning it") { - ScriptedNetwork n{3}; - // Less than one full attempt's worth of budget, but more than the minimum worth starting. - n.send(n.to(n.swarm[0], 6s)); - - REQUIRE(n.router->timeouts.size() >= 2); - // The first attempt is the caller's own request, untouched -- the budget only governs what - // this layer *adds*. - CHECK(n.router->timeouts[0] == 10s); - // Every retry after it is capped by what remains of the operation. - for (size_t i = 1; i < n.router->timeouts.size(); i++) - CHECK(n.router->timeouts[i] <= 6s); - } +TEST_CASE("Core: redirects are bounded", "[core][swarm]") { + PollFixture f{3}; - SECTION("too little left to be worth starting stops the walk early") { - ScriptedNetwork n{4}; - // Below MIN_RETRY_BUDGET, so the first failure ends it rather than starting an attempt - // that cannot finish. - n.send(n.to(n.swarm[0], 1s)); + // Every member insists the account is not theirs. Re-resolving cannot help -- the corrected + // swarm is the one now rejecting us -- so this has to stop rather than loop. + f.net->auto_reply = [](const Request&) -> std::optional { + return MockNetwork::Reply{false, false, ERROR_MISDIRECTED_REQUEST, "wrong swarm"}; + }; - CHECK(n.router->tried.size() == 1); - } -} + f.poll(); -TEST_CASE( - "Network: an owner reference dropped mid-callback does not tear the Network down from its " - "own loop", - "[network]") { - // Two members so that the first, unreachable one sends the request through - // _retry_next_swarm_node: that goes via SnodePool::get_swarm, which answers from the loop, so - // the second attempt -- and the callback below -- run on the loop thread rather than on this - // one. - ScriptedNetwork n{2}; - n.router->replies[n.swarm[1].remote_pubkey] = {}; - - auto reached_callback = std::promise{}; - auto in_callback = reached_callback.get_future(); - std::atomic answered = false; - - n.net->send_request( - n.to(n.swarm[0]), [&reached_callback, &answered](bool ok, bool, int16_t, auto, auto) { - answered = ok; - reached_callback.set_value(); - - // Stay on the loop thread while the reference below goes, which is the interleaving - // that used to abort: the callback held a shared_ptr of its own, so - // dropping the owner's left the loop thread as the last owner, and ~Network joins - // that thread. - std::this_thread::sleep_for(50ms); - }); - - REQUIRE(in_callback.wait_for(5s) == std::future_status::ready); - - auto observer = std::weak_ptr{n.net}; - n.net.reset(); - - // Waits for the Network to actually be gone rather than merely unreferenced from here: the - // teardown is what fails, so it has to happen while this test is still running. Nothing else - // holds a reference, so this returns as soon as the callback has finished. - for (int i = 0; i < 500 && !observer.expired(); i++) - std::this_thread::sleep_for(10ms); - - // Surviving to here is the assertion: the failure was an abort out of a destructor rather than - // a wrong answer. - CHECK(observer.expired()); - CHECK(answered); + // Bounded, and bounded low: a handful of attempts, not one per member per round. + CHECK(tried(*f.net).size() <= 5); }