WIP refactor and partial impl for electrum subscribers.

Author: evoskuil

include/bitcoin/server/protocols/protocol_electrum.hpp

@@ -204,22 +204,29 @@ class BCS_API protocol_electrum
         rpc_interface::mempool_get_info) NOEXCEPT;
 
 protected:
+    using hash_digest = system::hash_digest;
+    enum class notify_t { address, scripthash, scriptpubkey };
+    typedef std::function<void(const code&, const hash_digest&,
+        const hash_digest&)> status_handler;
+
     /// Common implementation for block_header/s.
     void blockchain_block_headers(size_t starting, size_t quantity,
         size_t waypoint, bool multiplicity) NOEXCEPT;
 
     /// Completion handlers (for long-running address queries).
     /// -----------------------------------------------------------------------
 
-    void get_balance(const system::hash_digest& hash) NOEXCEPT;
-    void get_history(const system::hash_digest& hash) NOEXCEPT;
-    void get_mempool(const system::hash_digest& hash) NOEXCEPT;
-    void list_unspent(const system::hash_digest& hash) NOEXCEPT;
+    void get_balance(const hash_digest& hash) NOEXCEPT;
+    void get_history(const hash_digest& hash) NOEXCEPT;
+    void get_mempool(const hash_digest& hash) NOEXCEPT;
+    void list_unspent(const hash_digest& hash) NOEXCEPT;
 
-    void do_get_balance(const system::hash_digest& hash) NOEXCEPT;
-    void do_get_history(const system::hash_digest& hash) NOEXCEPT;
-    void do_get_mempool(const system::hash_digest& hash) NOEXCEPT;
-    void do_list_unspent(const system::hash_digest& hash) NOEXCEPT;
+    void do_get_balance(const hash_digest& hash) NOEXCEPT;
+    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;
@@ -229,16 +236,22 @@ class BCS_API protocol_electrum
         const database::histories& histories) NOEXCEPT;
     void complete_list_unspent(const code& ec,
         const database::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::address_t link) NOEXCEPT;
 
     /// Notification senders and send 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_address(node::address_t link) NOEXCEPT;
     void do_scripthash(node::address_t link) NOEXCEPT;
-    void do_scriptpubkey(node::address_t link) NOEXCEPT;
 
     /// Utilities.
     /// -----------------------------------------------------------------------
@@ -254,19 +267,32 @@ class BCS_API protocol_electrum
     }
 
 private:
+    // Status hash optimization (~200 bytes).
+    struct midstate
+    {
+        hash_digest status{};
+        system::stream::out::fast stream{ status };
+        system::hash::sha256::fast writer{ stream };
+    };
+
+    // 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;
 
+    // Address transformations.
+    static array_t transform(const database::histories& histories) NOEXCEPT;
+    static array_t transform(const database::unspents& unspents) NOEXCEPT;
+    static std::string to_method_name(notify_t type) NOEXCEPT;
+
     // Compute server.features.hosts value from config.
     object_t self_hosts() const NOEXCEPT;
     array_t more_hosts() const NOEXCEPT;
 
     // Convert between legacy bitcoin payment address and scripthash.
-    system::hash_digest extract_scripthash(
-        const std::string& address) const NOEXCEPT;
+    hash_digest extract_scripthash(const std::string& address) const NOEXCEPT;
     system::wallet::payment_address extract_address(
         const system::chain::script& script) const NOEXCEPT;
 
@@ -275,15 +301,13 @@ 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;
 
-    // Address transformations.
-    array_t transform(const database::histories& histories) NOEXCEPT;
-    array_t transform(const database::unspents& unspents) NOEXCEPT;
-
-    // Shared implementations.
-    void send_scripthash_unsubscribe(const system::hash_digest& hash) NOEXCEPT;
-    void send_scripthash_subscribe(const system::hash_digest& hash) NOEXCEPT;
-    bool send_get_status(const system::chain::point& prevout,
+    // 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_;
@@ -294,15 +318,13 @@ class BCS_API protocol_electrum
     std::atomic_bool subscribed_height_{};
     std::atomic_bool subscribed_header_{};
     std::atomic_bool subscribed_outpoint_{};
-    std::atomic_bool subscribed_address_{};
     std::atomic_bool subscribed_scripthash_{};
-    std::atomic_bool subscribed_scriptpubkey_{};
 
     // 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<system::hash_digest> subscriptions_{};
+    std::unordered_set<hash_digest> subscriptions_{};
 };
 
 } // namespace server

src/protocols/electrum/protocol_electrum.cpp

@@ -165,23 +165,14 @@ bool protocol_electrum::handle_event(const code&, node::chase event_,
                 POST(do_outpoint, std::get<node::address_t>(value));
             }
 
-            if (subscribed_address_.load(relaxed))
-            {
-                BC_ASSERT(std::holds_alternative<node::address_t>(value));
-                POST(do_address, std::get<node::address_t>(value));
-            }
-
             if (subscribed_scripthash_.load(relaxed))
             {
+                BC_ASSERT(archive().address_enabled());
                 BC_ASSERT(std::holds_alternative<node::address_t>(value));
                 POST(do_scripthash, std::get<node::address_t>(value));
             }
 
-            if (subscribed_scriptpubkey_.load(relaxed))
-            {
-                BC_ASSERT(std::holds_alternative<node::address_t>(value));
-                POST(do_scriptpubkey, std::get<node::address_t>(value));
-            }
+            break;
         }
         default:
         {
@@ -206,7 +197,7 @@ void protocol_electrum::do_height(node::header_t link) NOEXCEPT
 
     if (height.is_terminal())
     {
-        LOGF("Electrum::do_height, object not found (" << link << ").");
+        LOGF("Electrum::do_height, height not found (" << link << ").");
         return;
     }
 
@@ -227,7 +218,7 @@ void protocol_electrum::do_header(node::header_t link) NOEXCEPT
 
     if (height.is_terminal())
     {
-        LOGF("Electrum::do_header, object not found (" << link << ").");
+        LOGF("Electrum::do_header, header not found (" << link << ").");
         return;
     }
 
@@ -239,77 +230,79 @@ void protocol_electrum::do_header(node::header_t link) NOEXCEPT
 }
 
 // Notifier for blockchain_outpoint_subscribe events.
-void protocol_electrum::do_outpoint(node::address_t) NOEXCEPT
+void protocol_electrum::do_outpoint(node::address_t link) NOEXCEPT
 {
-    chain::point point{};
+    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;
+    }
 
-    // TODO: compute and return outpoint status from a store query.
-    // TODO: unlike scripthash, this is not a cumulative "status" hash.
     send_notification("blockchain.outpoint.subscribe", array_t
     {
-        array_t{ encode_hash(point.hash()), point.index() },
-        object_t{}
+        array_t{ encode_hash(prevout.hash()), prevout.index() },
+        std::move(status)
     }, 128, BIND(handle_send, _1));
 }
 
-// This struct is small and stack allocated (208 bytes).
-// Writer holds stream ref to status and midstate via accumulator.
-// Writer flush rewrites status via stream and resets accumulator.
-// Pass writer to store method to accumulate confirmed address midstate.
-// Once initialized, copy writer after adding state and flushing for status.
-struct midstate
-{
-    hash_digest status{};
-    stream::out::fast stream{ status };
-    hash::sha256::fast writer{ stream };
-};
-
-// Notifier for blockchain_address_subscribe events.
-void protocol_electrum::do_address(node::address_t ) NOEXCEPT
+// Notifier for blockchain_scripthash_subscribe events.
+void protocol_electrum::do_scripthash(node::address_t link) NOEXCEPT
 {
-    std::string status_hash{};
-    std::string script_hash{};
+    BC_ASSERT(stranded());
 
-    midstate value{};
-    auto copy = value;
+    // TODO: get hash/type from event.
+    ///////////////////////////////////////////////////////////////////////////
+    hash_digest hash{};
+    constexpr auto type = notify_t::scripthash;
+    ///////////////////////////////////////////////////////////////////////////
 
-    // EXAMPLE:
-    // script_hash is a payment address for address.
-    send_notification("blockchain.address.subscribe", array_t
-    {
-        script_hash,
-        status_hash
-    }, 128, BIND(handle_send, _1));
+    // Address status is long-running, so cannot tie up strand.
+    PARALLEL(do_status, hash, BIND(notify_status, _1, _2, _3, type, link));
 }
 
-// Notifier for blockchain_scripthash_subscribe events.
-void protocol_electrum::do_scripthash(node::address_t) NOEXCEPT
+void protocol_electrum::notify_status(const code& ec, const hash_digest& hash,
+    const hash_digest& status, notify_t type, node::address_t link) NOEXCEPT
 {
-    std::string status_hash{};
-    std::string script_hash{};
+    BC_ASSERT(stranded());
 
-    // EXAMPLE:
-    // script_hash is a payment address for address.
-    send_notification("blockchain.scripthash.subscribe", array_t
+    if (ec)
     {
-        script_hash,
-        status_hash
-    }, 128, BIND(handle_send, _1));
+        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));
 }
 
-// Notifier for blockchain_scriptpubkey_subscribe events.
-void protocol_electrum::do_scriptpubkey(node::address_t) NOEXCEPT
-{
-    std::string status_hash{};
-    std::string script_hash{};
+// utilities
+// ----------------------------------------------------------------------------
+// private/static
 
-    // EXAMPLE:
-    // script_hash is a payment address for address.
-    send_notification("blockchain.scriptpubkey.subscribe", array_t
+// Convert enumeration to json-rpc notification method name.
+std::string protocol_electrum::to_method_name(notify_t type) NOEXCEPT
+{
+    switch (type)
     {
-        script_hash,
-        status_hash
-    }, 128, BIND(handle_send, _1));
+        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()