Implement electrum protocol address notifications.

Makefile.am

@@ -57,6 +57,7 @@ src_libbitcoin_server_la_SOURCES = \
     src/protocols/electrum/protocol_electrum_mempool.cpp \
     src/protocols/electrum/protocol_electrum_outputs.cpp \
     src/protocols/electrum/protocol_electrum_scripthash.cpp \
+    src/protocols/electrum/protocol_electrum_scripthash_subscribe.cpp \
     src/protocols/electrum/protocol_electrum_scriptpubkey.cpp \
     src/protocols/electrum/protocol_electrum_server.cpp \
     src/protocols/electrum/protocol_electrum_transactions.cpp \

builds/msvc/vs2022/libbitcoin-server/libbitcoin-server.vcxproj

@@ -139,6 +139,7 @@
     <ClCompile Include="..\..\..\..\src\protocols\electrum\protocol_electrum_mempool.cpp" />
     <ClCompile Include="..\..\..\..\src\protocols\electrum\protocol_electrum_outputs.cpp" />
     <ClCompile Include="..\..\..\..\src\protocols\electrum\protocol_electrum_scripthash.cpp" />
+    <ClCompile Include="..\..\..\..\src\protocols\electrum\protocol_electrum_scripthash_subscribe.cpp" />
     <ClCompile Include="..\..\..\..\src\protocols\electrum\protocol_electrum_scriptpubkey.cpp" />
     <ClCompile Include="..\..\..\..\src\protocols\electrum\protocol_electrum_server.cpp" />
     <ClCompile Include="..\..\..\..\src\protocols\electrum\protocol_electrum_transactions.cpp" />

builds/msvc/vs2022/libbitcoin-server/libbitcoin-server.vcxproj.filters

@@ -117,6 +117,9 @@
     <ClCompile Include="..\..\..\..\src\protocols\electrum\protocol_electrum_scripthash.cpp">
       <Filter>src\protocols\electrum</Filter>
     </ClCompile>
+    <ClCompile Include="..\..\..\..\src\protocols\electrum\protocol_electrum_scripthash_subscribe.cpp">
+      <Filter>src\protocols\electrum</Filter>
+    </ClCompile>
     <ClCompile Include="..\..\..\..\src\protocols\electrum\protocol_electrum_scriptpubkey.cpp">
       <Filter>src\protocols\electrum</Filter>
     </ClCompile>

include/bitcoin/server/error.hpp

@@ -61,6 +61,7 @@ enum error_t : uint8_t
     not_found,
     not_implemented,
     invalid_argument,
+    subscription_limit,
     unsupported_argument,
     unconfirmable_transaction,
     argument_overflow,

include/bitcoin/server/protocols/protocol_electrum.hpp

@@ -19,8 +19,9 @@
 #ifndef LIBBITCOIN_SERVER_PROTOCOLS_PROTOCOL_ELECTRUM_HPP
 #define LIBBITCOIN_SERVER_PROTOCOLS_PROTOCOL_ELECTRUM_HPP
 
+#include <map>
 #include <memory>
-#include <unordered_set>
+#include <set>
 #include <bitcoin/server/channels/channels.hpp>
 #include <bitcoin/server/define.hpp>
 #include <bitcoin/server/interfaces/interfaces.hpp>
@@ -52,6 +53,7 @@ class BCS_API protocol_electrum
             system::wallet::payment_address::mainnet_p2sh :
             system::wallet::payment_address::testnet_p2sh),
         channel_(std::dynamic_pointer_cast<channel_t>(channel)),
+        notification_strand_(channel_->service().get_executor()),
         network::tracker<protocol_electrum>(session->log)
     {
     }
@@ -227,48 +229,77 @@ class BCS_API protocol_electrum
     void do_get_history(const hash_digest& hash) NOEXCEPT;
     void do_get_mempool(const hash_digest& hash) NOEXCEPT;
     void do_list_unspent(const hash_digest& hash) NOEXCEPT;
-    void do_status(const hash_digest& hash,
-        const status_handler& sender) NOEXCEPT;
-
-    void complete_get_balance(const code& ec, uint64_t confirmed,
-        int64_t unconfirmed) NOEXCEPT;
-    void complete_get_history(const code& ec,
-        const histories& histories) NOEXCEPT;
-    void complete_get_mempool(const code& ec,
-        const histories& histories) NOEXCEPT;
-    void complete_list_unspent(const code& ec,
-        const unspents& unspents) NOEXCEPT;
-    void complete_status(const code& ec, const hash_digest& hash,
-        const hash_digest& status, const status_handler& sender) NOEXCEPT;
-
-    void send_status(const code& ec, const hash_digest& hash,
-        const hash_digest& status) NOEXCEPT;
-    void notify_status(const code& ec, const hash_digest& hash,
-        const hash_digest& status, notify_t type,
-        node::header_t link) NOEXCEPT;
-
-    /// Notification senders and send handlers.
+
+    void complete_get_balance(const code& ec, uint64_t confirmed, int64_t unconfirmed) NOEXCEPT;
+    void complete_get_history(const code& ec, const histories& histories) NOEXCEPT;
+    void complete_get_mempool(const code& ec, const histories& histories) NOEXCEPT;
+    void complete_list_unspent(const code& ec, const unspents& unspents) NOEXCEPT;
+
+    /// Notification event handlers.
     /// -----------------------------------------------------------------------
 
     void do_height(node::header_t link) NOEXCEPT;
     void do_header(node::header_t link) NOEXCEPT;
     void do_outpoint(node::header_t link) NOEXCEPT;
     void do_scripthash(node::header_t link) NOEXCEPT;
+    void do_regressed(node::header_t link) NOEXCEPT;
+
+    /// Address.
+    /// -----------------------------------------------------------------------
+
+    // subscription.
+    void scripthash_subscribe(const hash_digest& hash,
+        notify_t type) NOEXCEPT;
+    void do_scripthash_subscribe(const hash_digest& hash,
+        notify_t type) NOEXCEPT;
+    void complete_scripthash_subscribe(const code& ec,
+        hash_digest& status, const hash_digest& hash) NOEXCEPT;
+
+    // unsubscription.
+    void scripthash_unsubscribe(const hash_digest& hash) NOEXCEPT;
+    void do_scripthash_unsubscribe(const hash_digest& hash) NOEXCEPT;
+    void complete_scripthash_unsubscribe(bool found) NOEXCEPT;
+
+    // notification (do_scripthash()).
+    void scripthash_notify(const hash_digest& status, const hash_digest& hash,
+        notify_t type) NOEXCEPT;
+
+    /// Outpoint.
+    /// -----------------------------------------------------------------------
+
+    // subscription (do_outpoint()).
+    bool get_outpoint_status(interface::object_t& status,
+        const system::chain::point& prevout) const NOEXCEPT;
+    bool send_outpoint_status(const system::chain::point& prevout,
+        const std::string& spk_hint) NOEXCEPT;
+
+    // unsubscription.
+    // notification.
 
     /// Utilities.
     /// -----------------------------------------------------------------------
 
+    /// The negotiated version is at least the specified level.
     inline bool at_least(server::electrum::version version) const NOEXCEPT
     {
         return channel_->version() >= version;
     }
 
+    /// Configuration options.
     inline const options_t& options() const NOEXCEPT
     {
         return options_;
     }
 
 private:
+    // Post to notification strand.
+    template <class Derived, typename Method, typename... Args>
+    inline auto notify(Method&& method, Args&&... args) NOEXCEPT
+    {
+        return boost::asio::post(notification_strand_,
+            BIND_SAFE(BIND_SHARED(method, args)));
+    }
+
     // Status hash optimization (~200 bytes).
     struct midstate
     {
@@ -277,13 +308,25 @@ class BCS_API protocol_electrum
         system::hash::sha256::fast writer{ stream };
     };
 
+    // Subscription to address/scripthash/scruptpubkey.
+    struct subscription
+    {
+        notify_t type{};
+        midstate state{};
+        database::address_link cursor{};
+    };
+
     // Aliases.
     using array_t = network::rpc::array_t;
     using object_t = network::rpc::object_t;
     using version_t = protocol_electrum_version;
     static constexpr electrum::version minimum = version_t::minimum;
     static constexpr electrum::version maximum = version_t::maximum;
 
+    // Scripthash status.
+    code get_scripthash_status(hash_digest& out, subscription& sub,
+        const hash_digest& hash) NOEXCEPT;
+
     // Transformations.
     static std::string to_method_name(notify_t type) NOEXCEPT;
     static array_t transform(const unspents& unspents) NOEXCEPT;
@@ -304,14 +347,6 @@ class BCS_API protocol_electrum
     code validate_tx(const system::chain::transaction& tx) const NOEXCEPT;
     code broadcast_tx(const system::chain::transaction::cptr& tx) NOEXCEPT;
 
-    // Shared send/get implementations.
-    void send_scripthash_unsubscribe(const hash_digest& hash) NOEXCEPT;
-    void send_scripthash_subscribe(const hash_digest& hash) NOEXCEPT;
-    bool send_outpoint_status(const system::chain::point& prevout,
-        const std::string& spk_hint) NOEXCEPT;
-    bool get_outpoint_status(object_t& status,
-        const system::chain::point& prevout) const NOEXCEPT;
-
     // These are thread safe.
     const options_t& options_;
     const bool turbo_;
@@ -320,14 +355,18 @@ class BCS_API protocol_electrum
     std::atomic_bool stopping_{};
     std::atomic_bool subscribed_height_{};
     std::atomic_bool subscribed_header_{};
+    std::atomic_bool subscribed_address_{};
     std::atomic_bool subscribed_outpoint_{};
-    std::atomic_bool subscribed_scripthash_{};
 
     // This is mostly thread safe, and used in a thread safe manner.
     const channel_t::ptr channel_;
 
-    // This is protected by strand.
-    std::unordered_set<hash_digest> subscriptions_{};
+    // This is thread safe, uses network threadpool.
+    network::asio::strand notification_strand_;
+
+    // These are protected by notification strand.
+    std::map<hash_digest, subscription> address_subscriptions_{};
+    std::set<system::chain::point> outpoint_subscriptions_{};
 };
 
 } // namespace server

src/error.cpp

@@ -51,6 +51,7 @@ DEFINE_ERROR_T_MESSAGE_MAP(error)
     { not_found, "not_found" },
     { not_implemented, "not_implemented" },
     { invalid_argument, "invalid_argument" },
+    { subscription_limit, "subscription_limit" },
     { unsupported_argument, "unsupported_argument" },
     { unconfirmable_transaction, "unconfirmable_transaction" },
     { argument_overflow, "argument_overflow" },

src/protocols/electrum/protocol_electrum.cpp

@@ -28,6 +28,7 @@ namespace libbitcoin {
 namespace server {
 
 #define CLASS protocol_electrum
+#define NOTIFY(method, ...) notify<CLASS>(&CLASS::method, __VA_ARGS__)
 
 using namespace system;
 using namespace network::rpc;
@@ -47,7 +48,6 @@ BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT)
 void protocol_electrum::start() NOEXCEPT
 {
     BC_ASSERT(stranded());
-
     if (started())
         return;
 
@@ -158,18 +158,26 @@ bool protocol_electrum::handle_event(const code&, node::chase event_,
             if (subscribed_outpoint_.load(relaxed))
             {
                 BC_ASSERT(std::holds_alternative<node::header_t>(value));
-                POST(do_outpoint, std::get<node::header_t>(value));
+                NOTIFY(do_outpoint, std::get<node::header_t>(value));
             }
 
-            if (subscribed_scripthash_.load(relaxed))
+            if (subscribed_address_.load(relaxed))
             {
                 BC_ASSERT(archive().address_enabled());
                 BC_ASSERT(std::holds_alternative<node::header_t>(value));
-                POST(do_scripthash, std::get<node::header_t>(value));
+                NOTIFY(do_scripthash, std::get<node::header_t>(value));
             }
 
             break;
         }
+        case node::chase::regressed:
+        case node::chase::disorganized:
+        {
+            // value is regression branch_point.
+            BC_ASSERT(std::holds_alternative<node::height_t>(value));
+            NOTIFY(do_regressed, std::get<node::height_t>(value));
+            break;
+        }
         default:
         {
             break;
@@ -179,15 +187,14 @@ bool protocol_electrum::handle_event(const code&, node::chase event_,
     return true;
 }
 
-// notifications
+// height/header notifications.
 // ----------------------------------------------------------------------------
 // Each notification is an independent message.
 
 // Notifier for handle_blockchain_number_of_blocks_subscribe events.
 void protocol_electrum::do_height(node::header_t link) NOEXCEPT
 {
     BC_ASSERT(stranded());
-
     const auto& query = archive();
     const auto height = query.get_height(link);
 
@@ -207,7 +214,6 @@ void protocol_electrum::do_height(node::header_t link) NOEXCEPT
 void protocol_electrum::do_header(node::header_t link) NOEXCEPT
 {
     BC_ASSERT(stranded());
-
     const auto& query = archive();
     const auto height = query.get_height(link);
     const auto header = query.get_wire_header(link);
@@ -225,82 +231,6 @@ void protocol_electrum::do_header(node::header_t link) NOEXCEPT
     }, 64, BIND(complete, _1));
 }
 
-// Notifier for blockchain_outpoint_subscribe events.
-void protocol_electrum::do_outpoint(node::header_t link) NOEXCEPT
-{
-    BC_ASSERT(stranded());
-
-    // TODO: get prevout from event.
-    ///////////////////////////////////////////////////////////////////////////
-    chain::point prevout{};
-    ///////////////////////////////////////////////////////////////////////////
-
-    object_t status{};
-    if (!get_outpoint_status(status, prevout))
-    {
-        LOGF("Electrum::do_outpoint, outpoint not found (" << link << ").");
-        return;
-    }
-
-    send_notification("blockchain.outpoint.subscribe", array_t
-    {
-        array_t{ encode_hash(prevout.hash()), prevout.index() },
-        std::move(status)
-    }, 128, BIND(handle_send, _1));
-}
-
-// Notifier for blockchain_scripthash_subscribe events.
-void protocol_electrum::do_scripthash(node::header_t link) NOEXCEPT
-{
-    BC_ASSERT(stranded());
-
-    // TODO: get hash/type from event.
-    ///////////////////////////////////////////////////////////////////////////
-    hash_digest hash{};
-    constexpr auto type = notify_t::scripthash;
-    ///////////////////////////////////////////////////////////////////////////
-
-    // Address status is long-running, so cannot tie up strand.
-    PARALLEL(do_status, hash, BIND(notify_status, _1, _2, _3, type, link));
-}
-
-void protocol_electrum::notify_status(const code& ec, const hash_digest& hash,
-    const hash_digest& status, notify_t type, node::header_t link) NOEXCEPT
-{
-    BC_ASSERT(stranded());
-
-    if (ec)
-    {
-        LOGF("Electrum::do_scripthash, address not found (" << link << ").");
-        return;
-    }
-
-    send_notification(to_method_name(type), array_t
-    {
-        encode_hash(hash),
-        status == null_hash ? value_t{} : value_t{ encode_hash(status) }
-    }, 128, BIND(complete, _1));
-}
-
-// utilities
-// ----------------------------------------------------------------------------
-// private/static
-
-// Convert enumeration to json-rpc notification method name.
-std::string protocol_electrum::to_method_name(notify_t type) NOEXCEPT
-{
-    switch (type)
-    {
-        case notify_t::address:
-            return "blockchain.address.subscribe";
-        case notify_t::scripthash:
-            return "blockchain.scripthash.subscribe";
-        default:
-        case notify_t::scriptpubkey:
-            return "blockchain.scriptpubkey.subscribe";
-    }
-}
-
 BC_POP_WARNING()
 
 } // namespace server

src/protocols/electrum/protocol_electrum_addresses.cpp

@@ -118,7 +118,7 @@ void protocol_electrum::handle_blockchain_address_subscribe(const code& ec,
         return;
     }
 
-    send_scripthash_subscribe(hash);
+    scripthash_subscribe(hash, notify_t::address);
 }
 
 // utilities