Optimize signature aggregation

src/bls/bls_aggregator.h

@@ -7,15 +7,19 @@
 #include <string>
 #include <vector>
 
+#include "bls/bls_crypto.h"
+#include "crypto/eth.h"
+
 namespace oxenmq {
 class OxenMq;
 }
 
 namespace eth {
 struct bls_aggregate_signed {
     std::vector<uint8_t> msg_to_sign;
-    std::vector<bls_public_key> signers_bls_pubkeys;
-    bls_signature signature;
+    std::unordered_map<bls_public_key, bls_signature> signatures;  // Individual signatures
+    bls_public_key aggregate_pubkey;  // Aggregate pubkey of the pubkeys in `signatures`
+    bls_signature signature;          // Aggregate of all the signatures in `signatures`
 };
 
 enum class bls_exit_type {
@@ -47,32 +51,46 @@ struct bls_registration_response {
     crypto::ed25519_signature ed_signature;
 };
 
-struct bls_response {
-    service_nodes::service_node_address sn;
-    bool success;
-};
+using request_callback = std::function<void(
+        const service_nodes::service_node_address& sn,
+        bool success,
+        std::vector<std::string> data)>;
 
 class bls_aggregator {
   public:
-    using request_callback =
-            std::function<void(const bls_response& response, const std::vector<std::string>& data)>;
-
     explicit bls_aggregator(cryptonote::core& core);
 
-    // Request the service node network to sign the requested amount of
-    // `rewards` for the given Ethereum `address` if by consensus they agree
-    // that the amount is valid. This node (the aggregator) will aggregate the
-    // signatures into the response.
+    // Request the service node network to sign the requested amount of `rewards` for the given
+    // Ethereum `address` if by consensus they agree that the amount is valid. Once all requests
+    // finish, this node (the aggregator) will aggregate the signatures into the response and pass
+    // it to `callback`.
+    //
+    // This function is asychronous: it returns immediately, calling `callback` once the final
+    // aggregate response is available.
     //
     // This function throws an `invalid_argument` exception if `address` is zero or, the `rewards`
     // amount is `0` or height is greater than the current blockchain height.
-    bls_rewards_response rewards_request(const address& addr, uint64_t height);
-
-    // Request the service node network to sign a request to remove the node specified by
-    // `pubkey` from the network. The nature of this exit is set by `type`. This node (the
-    // aggregator) will aggregate the signatures into the response.
-    bls_exit_liquidation_response exit_liquidation_request(
-            const crypto::public_key& pubkey, bls_exit_type type);
+    void rewards_request(
+            const address& addr,
+            uint64_t height,
+            std::function<void(std::shared_ptr<const bls_rewards_response>)> callback);
+
+    // Request the service node network to sign a request to remove the node specified by `pubkey`
+    // (either SN pubkey or BLS pubkey) from the network. The nature of this exit is set by `type`.
+    // This node (the aggregator) will aggregate the signatures into the response, and call
+    // `callback` with it once available.
+    //
+    // For normal (non-liqudiation) exits, the pubkey must exist in the recently removed nodes list.
+    // For liquidations via bls pubkey, this is not required (that is: liquidations can be issued
+    // for BLS pubkeys that oxend does not know about, to be able to remove bad registrations that
+    // oxend doesn't accept for whatever reason from the contract side).
+    //
+    // This function is asychronous: it returns immediately, later calling `callback` once the
+    // final aggregate response is available.
+    void exit_liquidation_request(
+            const std::variant<crypto::public_key, eth::bls_public_key>& pubkey,
+            bls_exit_type type,
+            std::function<void(std::shared_ptr<const bls_exit_liquidation_response>)> callback);
 
     bls_registration_response registration(
             const address& sender, const crypto::public_key& sn_pubkey) const;
@@ -82,13 +100,27 @@ class bls_aggregator {
 
     void get_exit_liquidation(oxenmq::Message& m, bls_exit_type type) const;
 
-    // Goes out to the nodes on the network and makes oxenmq requests to all of them, when getting
-    // the reply `callback` will be called to process their reply
-    // Returns the number of nodes that we dispatched a request to
+    template <typename Result>
+    struct node_req_data {
+        Result result;
+        size_t remaining;
+        std::mutex signers_mutex;
+        eth::signature_aggregator agg_sig;
+    };
+
+    // Initiates asynchronous requests to service nodes on the network, making oxenmq requests to
+    // each of them; as each reply is returned the the reply `callback` will be called to process
+    // the reply.  Once the callback has been invoked for all requests (successful or failed) the
+    // `final_callback` is invoked (if non-null).
+    //
+    // Returns the total number of nodes to which requests will be sent.
+    //
+    // Note that this function is asychronous: it returns immediately without waiting for replies.
     uint64_t nodes_request(
-            std::string_view request_name,
-            std::string_view message,
-            const request_callback& callback);
+            std::string request_name,
+            std::string message,
+            request_callback callback,
+            std::function<void(int total_requests)> final_callback);
 
     cryptonote::core& core;
 
@@ -100,10 +132,10 @@ class bls_aggregator {
     std::mutex exit_liquidation_response_cache_mutex;
 
     // Cache the aggregate signature response for updating rewards to avoid requerying the network
-    std::unordered_map<address, bls_rewards_response> rewards_response_cache;
+    std::unordered_map<address, std::shared_ptr<bls_rewards_response>> rewards_response_cache;
 
     // The cache for exits and liquidation signature aggregations. See `rewards_response_cache`
-    std::unordered_map<bls_public_key, bls_exit_liquidation_response>
+    std::unordered_map<bls_public_key, std::shared_ptr<bls_exit_liquidation_response>>
             exit_liquidation_response_cache;
 };
 }  // namespace eth

src/cryptonote_core/cryptonote_core.cpp

@@ -2640,15 +2640,21 @@ uint64_t core::get_free_space() const {
     return fs::space(m_config_folder).available;
 }
 //-----------------------------------------------------------------------------------------------
-eth::bls_rewards_response core::bls_rewards_request(const eth::address& address, uint64_t height) {
-    return m_bls_aggregator->rewards_request(address, height);
+void core::bls_rewards_request(
+        const eth::address& address,
+        uint64_t height,
+        std::function<void(std::shared_ptr<const eth::bls_rewards_response>)> callback) {
+    m_bls_aggregator->rewards_request(address, height, std::move(callback));
 }
 //-----------------------------------------------------------------------------------------------
-eth::bls_exit_liquidation_response core::bls_exit_liquidation_request(
-        const crypto::public_key& pubkey, bool liquidate) {
-    eth::bls_exit_type type =
-            liquidate ? eth::bls_exit_type::liquidate : eth::bls_exit_type::normal;
-    return m_bls_aggregator->exit_liquidation_request(pubkey, type);
+void core::bls_exit_liquidation_request(
+        const std::variant<crypto::public_key, eth::bls_public_key>& pubkey,
+        bool liquidate,
+        std::function<void(std::shared_ptr<const eth::bls_exit_liquidation_response>)> callback) {
+    m_bls_aggregator->exit_liquidation_request(
+            pubkey,
+            liquidate ? eth::bls_exit_type::liquidate : eth::bls_exit_type::normal,
+            std::move(callback));
 }
 //-----------------------------------------------------------------------------------------------
 eth::bls_registration_response core::bls_registration(const eth::address& address) const {

src/cryptonote_core/cryptonote_core.h

@@ -31,6 +31,7 @@
 #include "common/command_line.h"
 #include "common/exception.h"
 #include "crypto/crypto.h"
+#include "crypto/eth.h"
 #include "crypto/hash.h"
 #include "cryptonote_basic/connection_context.h"
 #include "cryptonote_basic/hardfork.h"
@@ -594,9 +595,21 @@ class core final {
      */
     bool offline() const { return m_offline; }
 
-    eth::bls_rewards_response bls_rewards_request(const eth::address& address, uint64_t height);
-    eth::bls_exit_liquidation_response bls_exit_liquidation_request(
-            const crypto::public_key& pubkey, bool liquidate);
+    // Initiates an asynchronous rewards signing request to the network's nodes and returns
+    // immediately.  `callback` gets invoked once the requests are finished.
+    void bls_rewards_request(
+            const eth::address& address,
+            uint64_t height,
+            std::function<void(std::shared_ptr<const eth::bls_rewards_response>)> callback);
+
+    // Initiates an asynchronous exit or liquidation signing request to the network's nodes and
+    // returns immediately.  `callback` gets invoked once the requests are finished.
+    void bls_exit_liquidation_request(
+            const std::variant<crypto::public_key, eth::bls_public_key>& pubkey,
+            bool liquidate,
+            std::function<void(std::shared_ptr<const eth::bls_exit_liquidation_response>)>
+                    callback);
+
     eth::bls_registration_response bls_registration(const eth::address& ethereum_address) const;
 
     /**

src/l2_tracker/l2_tracker.cpp

@@ -809,14 +809,16 @@ uint64_t L2Tracker::get_safe_height() const {
                  : latest_height - config.L2_TRACKER_SAFE_BLOCKS;
 }
 
-std::vector<uint64_t> L2Tracker::get_non_signers(
-        const std::unordered_set<bls_public_key>& bls_public_keys) {
-    return rewards_contract.get_non_signers(bls_public_keys);
+void L2Tracker::get_non_signers(
+        std::unordered_set<bls_public_key> bls_public_keys,
+        std::function<void(std::optional<NonSigners>)> callback) {
+    rewards_contract.get_non_signers(std::move(bls_public_keys), std::move(callback));
 }
 
-RewardsContract::ServiceNodeIDs L2Tracker::get_all_service_node_ids(
-        std::optional<uint64_t> height) {
-    return rewards_contract.all_service_node_ids(height);
+void L2Tracker::get_all_service_node_ids(
+        std::optional<uint64_t> height,
+        std::function<void(std::optional<ServiceNodeIDs>)> callback) {
+    rewards_contract.all_service_node_ids(height, std::move(callback));
 }
 
 std::optional<bool> L2Tracker::is_in_contract(const eth::bls_public_key& pubkey) const {

src/l2_tracker/l2_tracker.h

@@ -7,6 +7,7 @@
 #include <ethyl/provider.hpp>
 #include <forward_list>
 #include <iterator>
+#include <ranges>
 #include <shared_mutex>
 #include <unordered_set>
 
@@ -169,24 +170,35 @@ class L2Tracker {
     // Returns the age of the last successful height response we got from an L2 RPC provider.
     std::optional<std::chrono::nanoseconds> latest_height_age() const;
 
-    // Initiates a synchronous request to the current L2 contract to retrieve the current list of
-    // BLS pubkeys, and returns the numeric contract IDs of any contract pubkeys that are not in the
-    // `[begin..end)` input range.
+    // Performs an asynchronous request to the current L2 contract to retrieve the current list of
+    // BLS pubkeys, and returns a NonSigners struct containing the numeric contract
+    // IDs of any contract pubkeys that are not in the `[begin..end)` input range, and any rejected
+    // pubkeys that are not in the contract and should be remove before submission to the contract.
     //
-    // TODO: make asychronous
-    std::vector<uint64_t> get_non_signers(
-            const std::unordered_set<bls_public_key>& bls_public_keys);
-    template <std::input_iterator It, std::sentinel_for<It> End>
-    std::vector<uint64_t> get_non_signers(It begin, End end) {
-        return get_non_signers(std::unordered_set<bls_public_key>{begin, end});
+    // This method returns immediately: when the data is returned from the contract the given
+    // callback is invoked with the result.
+    //
+    // Invokes with nullopt on failure.
+    void get_non_signers(
+            std::unordered_set<bls_public_key> bls_public_keys,
+            std::function<void(std::optional<NonSigners>)> callback);
+
+    // Versions of the above that take a pubkey range and construct the unordered_set on the fly
+    template <std::ranges::input_range R>
+        requires std::same_as<const std::ranges::range_reference_t<R>, const bls_public_key&>
+    void get_non_signers(R&& r, std::function<void(std::optional<NonSigners>)> callback) {
+        return get_non_signers(
+                std::unordered_set<bls_public_key>{r.begin(), r.end()}, std::move(callback));
     }
 
-    // Initiates a synchronous request to the current L2 contract to retrieve the list of contract
-    // numeric IDs and pubkeys of all contract service nodes as of the given height (or as of the
-    // latest height if height is nullopt).
-    //
-    // TODO: make asychronous
-    RewardsContract::ServiceNodeIDs get_all_service_node_ids(std::optional<uint64_t> height);
+    // Initiates an asynchronous request for all current L2 contract numeric IDs and pubkeys of
+    // contract service nodes at the given height (current height if nullopt).  This method returns
+    // immediately, with a followup call passing the result to given lambda (likely from a different
+    // thread) when the request completes, or a call to the lambda with `std::nullopt` if the
+    // request fails or times out.
+    void get_all_service_node_ids(
+            std::optional<uint64_t> height,
+            std::function<void(std::optional<ServiceNodeIDs>)> callback);
 
     // Returns true if the given pubkey is in the L2 contract, as of the last contract list update,
     // false if not in the list.  (This method does not make a new L2 provider request).  Returns