Echo broadcast transactions to subscribed Electrum clients.

include/bitcoin/server/protocols/protocol_electrum.hpp

@@ -78,6 +78,10 @@ class BCS_API protocol_electrum
     bool handle_chase(const code&, node::chase event_,
         node::event_value) NOEXCEPT;
 
+    bool handle_broadcast_transaction(const code& ec,
+        const network::messages::peer::transaction::cptr& message,
+        uint64_t sender) NOEXCEPT;
+
     /// Handlers (headers).
     void handle_blockchain_number_of_blocks_subscribe(const code& ec,
         rpc_interface::blockchain_number_of_blocks_subscribe) NOEXCEPT;
@@ -289,6 +293,9 @@ class BCS_API protocol_electrum
     void do_outpoint(node::header_t link) NOEXCEPT;
     void do_scripthash(node::header_t link) NOEXCEPT;
     void do_reorganized(node::header_t link) NOEXCEPT;
+    void do_broadcast_scripthash(
+        const system::chain::transaction::cptr& tx,
+        const retained_txs& retained_snapshot) NOEXCEPT;
 
     /// Address.
     /// -----------------------------------------------------------------------
@@ -306,7 +313,8 @@ class BCS_API protocol_electrum
         notify_t type) NOEXCEPT;
 
     code get_scripthash_history(address_subscription& sub,
-        const hash_digest& hash, size_t limit) NOEXCEPT;
+        const hash_digest& hash, size_t limit,
+        const retained_txs& extra={}) NOEXCEPT;
 
     /// Outpoint.
     /// -----------------------------------------------------------------------

src/protocols/electrum/protocol_electrum.cpp

@@ -53,6 +53,10 @@ void protocol_electrum::start() NOEXCEPT
     // Chaser subscription is asynchronous, events may be missed.
     subscribe_chase(BIND(handle_chase, _1, _2, _3));
 
+    // Echoes broadcast txs to subscribers, as the tx/block protocols do.
+    SUBSCRIBE_BROADCAST(network::messages::peer::transaction,
+        handle_broadcast_transaction, _1, _2, _3);
+
     // Header methods.
     SUBSCRIBE_RPC(handle_blockchain_number_of_blocks_subscribe, _1, _2);
     SUBSCRIBE_RPC(handle_blockchain_block_get_chunk, _1, _2, _3);
@@ -125,6 +129,7 @@ void protocol_electrum::stopping(const code& ec) NOEXCEPT
     BC_ASSERT(stranded());
     stopping_.store(true);
     unsubscribe_chase();
+    UNSUBSCRIBE_BROADCAST();
     ping_timer_->stop();
     protocol_rpc<interface::electrum>::stopping(ec);
 }

src/protocols/electrum/protocol_electrum_subscribe.cpp

@@ -194,6 +194,34 @@ void protocol_electrum::complete_scripthash_unsubscribe(bool found) NOEXCEPT
 // notify
 // ----------------------------------------------------------------------------
 
+bool protocol_electrum::handle_broadcast_transaction(const code& ec,
+    const network::messages::peer::transaction::cptr& message,
+    uint64_t) NOEXCEPT
+{
+    BC_ASSERT(stranded());
+
+    if (stopped(ec))
+        return false;
+
+    // blockchain.transaction.get also depends on this (see broadcast_tx).
+    retain_tx(message->transaction_ptr);
+
+    // Notifications require a full duplex transport, as with handle_chase.
+    if (!channel_->websocket() && !channel_->downgraded())
+        return true;
+
+    if (subscribed_address_.load(relaxed))
+    {
+        BC_ASSERT(archive().address_enabled());
+
+        // retained() must be copied here, on the channel strand.
+        POST_NOTIFY(do_broadcast_scripthash, message->transaction_ptr,
+            retained());
+    }
+
+    return true;
+}
+
 // Notifier for blockchain_scripthash_subscribe events.
 void protocol_electrum::do_scripthash(node::header_t) NOEXCEPT
 {
@@ -219,6 +247,36 @@ void protocol_electrum::do_scripthash(node::header_t) NOEXCEPT
     }
 }
 
+// Recomputes and echoes status for the subscriptions the tx touches.
+void protocol_electrum::do_broadcast_scripthash(
+    const chain::transaction::cptr& tx,
+    const retained_txs& retained_snapshot) NOEXCEPT
+{
+    BC_ASSERT(notification_strand_.running_in_this_thread());
+
+    for (auto& [key, sub]: address_subscriptions_)
+    {
+        if (!touches(*tx, key))
+            continue;
+
+        const auto previous = sub.status;
+        if (const auto ec = get_scripthash_history(sub, key, max_size_t,
+            retained_snapshot))
+        {
+            if (ec != database::error::query_canceled &&
+                ec != error::not_found)
+            {
+                LOGF("Electrum::do_broadcast_scripthash, " << ec.message());
+            }
+
+            continue;
+        }
+
+        if (sub.status != previous)
+            POST(scripthash_notify, sub.status, key, sub.type);
+    }
+}
+
 void protocol_electrum::scripthash_notify(const hash_digest& status,
     const hash_digest& hash, notify_t type) NOEXCEPT
 {
@@ -261,8 +319,11 @@ void protocol_electrum::write_status(midstate& accumulator,
 }
 
 // protected
+// extra is a channel-strand snapshot of retained(), passed by the caller,
+// as this runs on notification_strand_ (see handle_broadcast_transaction).
 code protocol_electrum::get_scripthash_history(address_subscription& sub,
-    const hash_digest& hash, size_t limit) NOEXCEPT
+    const hash_digest& hash, size_t limit,
+    const retained_txs& extra) NOEXCEPT
 {
     BC_ASSERT(notification_strand_.running_in_this_thread());
 
@@ -272,9 +333,26 @@ code protocol_electrum::get_scripthash_history(address_subscription& sub,
         limit, turbo_))
         return ec;
 
+    // Same criteria as append_retained.
+    for (const auto& [tx_hash, tx]: extra)
+    {
+        if (!query.to_tx(tx_hash).is_terminal() || !touches(*tx, hash))
+            continue;
+
+        history.push_back(database::history
+        {
+            { tx_hash, database::history::rooted_height },
+            tx->fee(),
+            database::history::unconfirmed_position
+        });
+    }
+
     if (history.empty())
         return error::success;
 
+    if (!extra.empty())
+        database::history::filter_sort_and_dedup(history);
+
     auto it = history.cbegin();
     while (it != history.cend() && it->confirmed())
         write_status(sub.accumulator, *it++);