feat(electrum): add blockchain.transaction.broadcast_package

src/daemon.rs

@@ -177,6 +177,46 @@ pub struct TxResult {
     error: Option<String>,
 }
 
+impl SubmitPackageResult {
+    /// Txids of the transactions accepted into the daemon's mempool by this package submission
+    /// (those without a per-transaction error).
+    pub fn accepted_txids(&self) -> Vec<Txid> {
+        self.tx_results
+            .values()
+            .filter(|tx| tx.error.is_none())
+            .filter_map(|tx| Txid::from_str(&tx.txid).ok())
+            .collect()
+    }
+
+    /// Build the response for the Electrum `blockchain.transaction.broadcast_package` method.
+    ///
+    /// When `verbose` is true, the full `submitpackage` result is returned. Otherwise a compact
+    /// `{ success, errors? }` object is returned, where `success` is whether the package was
+    /// accepted and `errors` lists any per-transaction errors.
+    ///
+    /// Ported from romanz/electrs (https://github.com/romanz/electrs).
+    pub fn into_electrum_response(self, verbose: bool) -> Value {
+        if verbose {
+            return json!(self);
+        }
+        let success = self.package_msg == "success";
+        let errors: Vec<Value> = self
+            .tx_results
+            .values()
+            .filter_map(|tx| {
+                tx.error
+                    .as_ref()
+                    .map(|error| json!({ "error": error, "txid": tx.txid }))
+            })
+            .collect();
+        if errors.is_empty() {
+            json!({ "success": success })
+        } else {
+            json!({ "success": success, "errors": errors })
+        }
+    }
+}
+
 pub trait CookieGetter: Send + Sync {
     fn get(&self) -> Result<Vec<u8>>;
 }

src/electrum/server.rs

@@ -441,6 +441,27 @@ impl Connection {
         Ok(json!(txid))
     }
 
+    // Ported from romanz/electrs (https://github.com/romanz/electrs).
+    fn blockchain_transaction_broadcast_package(&self, params: &[Value]) -> Result<Value> {
+        let txhexes: Vec<String> = params
+            .get(0)
+            .ok_or_else(|| invalid_params("missing transactions"))
+            .and_then(|txs| {
+                serde_json::from_value(txs.clone())
+                    .map_err(|_| invalid_params("non-array transactions"))
+            })?;
+        let verbose = bool_from_value_or(params.get(1), "verbose", false)?;
+
+        let result = self.query.submit_package(txhexes, None, None)?;
+        if let Err(e) = self.sender.try_send(Message::PeriodicUpdate) {
+            warn!(
+                "failed to issue PeriodicUpdate after broadcast_package: {}",
+                e
+            );
+        }
+        Ok(result.into_electrum_response(verbose))
+    }
+
     fn blockchain_transaction_get(&self, params: &[Value]) -> Result<Value> {
         let tx_hash = Txid::from(hash_from_value(params.get(0))?);
         let verbose = match params.get(1) {
@@ -521,6 +542,9 @@ impl Connection {
             "blockchain.scripthash.subscribe" => self.blockchain_scripthash_subscribe(&params),
             "blockchain.scripthash.unsubscribe" => self.blockchain_scripthash_unsubscribe(&params),
             "blockchain.transaction.broadcast" => self.blockchain_transaction_broadcast(&params),
+            "blockchain.transaction.broadcast_package" => {
+                self.blockchain_transaction_broadcast_package(&params)
+            }
             "blockchain.transaction.get" => self.blockchain_transaction_get(&params),
             "blockchain.transaction.get_merkle" => self.blockchain_transaction_get_merkle(&params),
             "blockchain.transaction.id_from_pos" => {

src/new_index/mempool.rs

@@ -322,6 +322,28 @@ impl Mempool {
         }
     }
 
+    /// Add multiple transactions (e.g. an accepted package) to the mempool in a single batch,
+    /// so that interdependent parent/child txs are linked within the same `add` call. Txids that
+    /// are already present or cannot be fetched from the daemon are skipped.
+    pub fn add_by_txids(&mut self, daemon: &Daemon, txids: &[Txid]) -> Result<()> {
+        let mut txs_map = HashMap::new();
+        for &txid in txids {
+            if self.txstore.get(&txid).is_some() {
+                continue;
+            }
+            match daemon.getmempooltx(&txid) {
+                Ok(tx) => {
+                    txs_map.insert(txid, tx);
+                }
+                Err(e) => warn!("add_by_txids cannot find txid='{}': e='{}'", txid, e),
+            }
+        }
+        if txs_map.is_empty() {
+            return Ok(());
+        }
+        self.add(txs_map)
+    }
+
     #[trace]
     fn add(&mut self, txs_map: HashMap<Txid, Transaction>) -> Result<()> {
         self.delta

src/new_index/query.rs

@@ -87,7 +87,18 @@ impl Query {
         maxfeerate: Option<f64>,
         maxburnamount: Option<f64>,
     ) -> Result<SubmitPackageResult> {
-        self.daemon.submit_package(txhex, maxfeerate, maxburnamount)
+        let result = self.daemon.submit_package(txhex, maxfeerate, maxburnamount)?;
+        // Add accepted txs to the local mempool so subscription status updates reflect them
+        // immediately (they read from the local mempool), mirroring broadcast_raw() above.
+        let accepted_txids = result.accepted_txids();
+        if !accepted_txids.is_empty() {
+            let _ = self
+                .mempool
+                .write()
+                .unwrap()
+                .add_by_txids(&self.daemon, &accepted_txids);
+        }
+        Ok(result)
     }
 
     #[trace]

tests/electrum.rs

@@ -264,6 +264,134 @@ fn test_electrum_jsonrpc_errors() {
     assert_eq!(s, expected);
 }
 
+/// Test blockchain.transaction.broadcast_package submits a package via bitcoind's submitpackage
+#[cfg(not(feature = "liquid"))]
+#[test]
+fn test_electrum_broadcast_package() -> Result<()> {
+    use bitcoin::consensus::encode::serialize_hex;
+
+    let (_electrum_server, electrum_addr, mut tester) = common::init_electrum_tester()?;
+
+    let addr = tester.newaddress()?;
+    // create a tx; it stays in the mempool, so re-submitting it as a 1-tx package succeeds
+    let txid = tester.send(&addr, "0.1 BTC".parse().unwrap())?;
+    let tx_hex = serialize_hex(&tester.get_raw_transaction(txid)?);
+
+    let mut stream = TcpStream::connect(electrum_addr).unwrap();
+
+    // non-verbose: returns { "success": true }
+    let s = write_and_read(
+        &mut stream,
+        &format!(
+            "{{\"jsonrpc\": \"2.0\", \"method\": \"blockchain.transaction.broadcast_package\", \"params\": [[\"{}\"]], \"id\": 1}}",
+            tx_hex
+        ),
+    );
+    let v: electrumd::jsonrpc::serde_json::Value =
+        electrumd::jsonrpc::serde_json::from_str(&s).unwrap();
+    assert_eq!(
+        v["result"]["success"].as_bool(),
+        Some(true),
+        "unexpected response: {}",
+        s
+    );
+
+    // verbose: returns the full submitpackage result, including package_msg
+    let s = write_and_read(
+        &mut stream,
+        &format!(
+            "{{\"jsonrpc\": \"2.0\", \"method\": \"blockchain.transaction.broadcast_package\", \"params\": [[\"{}\"], true], \"id\": 2}}",
+            tx_hex
+        ),
+    );
+    let v: electrumd::jsonrpc::serde_json::Value =
+        electrumd::jsonrpc::serde_json::from_str(&s).unwrap();
+    assert_eq!(
+        v["result"]["package_msg"].as_str(),
+        Some("success"),
+        "unexpected verbose response: {}",
+        s
+    );
+
+    Ok(())
+}
+
+/// Regression test: a package broadcast must add accepted txs to electrs' local mempool, so a
+/// subscribed/queried scripthash sees them immediately (not only after the next background sync).
+#[cfg(not(feature = "liquid"))]
+#[test]
+fn test_electrum_broadcast_package_updates_mempool() -> Result<()> {
+    use bitcoin::consensus::encode::serialize_hex;
+    use bitcoin::hashes::{sha256, Hash};
+    use bitcoin::hex::DisplayHex;
+
+    let (_electrum_server, electrum_addr, tester) = common::init_electrum_tester()?;
+
+    let addr = tester.newaddress()?;
+    let mut hash = sha256::Hash::hash(addr.script_pubkey().as_bytes()).to_byte_array();
+    hash.reverse();
+    let scripthash = hash.to_lower_hex_string();
+
+    // Create a tx via the node directly, WITHOUT tester.send() -- so electrs does not sync it into
+    // its local mempool. The tx is in bitcoind's mempool but unknown to electrs.
+    let amount = bitcoin::Amount::from_btc(0.1).unwrap();
+    let txid: bitcoin::Txid = tester
+        .node_client()
+        .call(
+            "sendtoaddress",
+            &[addr.to_string().into(), json!(amount.to_btc())],
+        )
+        .unwrap();
+    let tx_hex = serialize_hex(&tester.get_raw_transaction(txid)?);
+
+    let mut stream = TcpStream::connect(electrum_addr).unwrap();
+
+    let history = |stream: &mut TcpStream, id: u32| -> Vec<String> {
+        let s = write_and_read(
+            stream,
+            &format!(
+                "{{\"jsonrpc\": \"2.0\", \"method\": \"blockchain.scripthash.get_history\", \"params\": [\"{}\"], \"id\": {}}}",
+                scripthash, id
+            ),
+        );
+        let v: electrumd::jsonrpc::serde_json::Value =
+            electrumd::jsonrpc::serde_json::from_str(&s).unwrap();
+        v["result"]
+            .as_array()
+            .expect("result array")
+            .iter()
+            .map(|e| e["tx_hash"].as_str().unwrap().to_string())
+            .collect()
+    };
+
+    // electrs has not synced the tx yet, so its history is empty
+    assert!(
+        history(&mut stream, 1).is_empty(),
+        "tx should not be in electrs' mempool before broadcast_package"
+    );
+
+    // broadcast the (already-in-bitcoind-mempool) tx as a 1-tx package
+    let s = write_and_read(
+        &mut stream,
+        &format!(
+            "{{\"jsonrpc\": \"2.0\", \"method\": \"blockchain.transaction.broadcast_package\", \"params\": [[\"{}\"]], \"id\": 2}}",
+            tx_hex
+        ),
+    );
+    let v: electrumd::jsonrpc::serde_json::Value =
+        electrumd::jsonrpc::serde_json::from_str(&s).unwrap();
+    assert_eq!(v["result"]["success"].as_bool(), Some(true), "response: {}", s);
+
+    // the accepted tx must now be visible in the scripthash history immediately
+    assert_eq!(
+        history(&mut stream, 3),
+        vec![txid.to_string()],
+        "broadcast_package did not add the accepted tx to electrs' local mempool"
+    );
+
+    Ok(())
+}
+
 fn write_and_read(stream: &mut TcpStream, write: &str) -> String {
     stream.write_all(write.as_bytes()).unwrap();
     stream.write(b"\n").unwrap();

tests/rest.rs

@@ -1241,6 +1241,57 @@ fn test_rest_submit_package() -> Result<()> {
     Ok(())
 }
 
+/// Regression test: POST /txs/package must add accepted txs to electrs' local mempool, so they are
+/// immediately visible via the address/scripthash endpoints (not only after the next background sync).
+#[cfg(not(feature = "liquid"))]
+#[test]
+fn test_rest_package_updates_mempool() -> Result<()> {
+    use bitcoin::consensus::encode::serialize_hex;
+
+    let (rest_handle, rest_addr, tester) = common::init_rest_tester().unwrap();
+
+    let addr = tester.newaddress()?;
+
+    // Create a tx via the node directly, WITHOUT tester.send() -- so electrs does not sync it into
+    // its local mempool. The tx is in bitcoind's mempool but unknown to electrs.
+    let txid: Txid = tester
+        .node_client()
+        .call("sendtoaddress", &[addr.to_string().into(), 0.1.into()])
+        .unwrap();
+    let tx_hex = serialize_hex(&tester.get_raw_transaction(txid)?);
+
+    let addr_txids = |rest_addr: net::SocketAddr| -> Vec<String> {
+        get_json(rest_addr, &format!("/address/{}/txs", addr))
+            .unwrap()
+            .as_array()
+            .expect("txs array")
+            .iter()
+            .map(|tx| tx["txid"].as_str().unwrap().to_string())
+            .collect()
+    };
+
+    // electrs has not synced the tx yet, so the address has no txs
+    assert!(
+        addr_txids(rest_addr).is_empty(),
+        "tx should not be in electrs' mempool before the package is submitted"
+    );
+
+    // submit the (already-in-bitcoind-mempool) tx as a 1-tx package
+    let package_resp =
+        ureq::post(&format!("http://{}/txs/package", rest_addr)).send_json([tx_hex])?;
+    assert_eq!(package_resp.status(), 200);
+
+    // the accepted tx must now be visible immediately via the address endpoint
+    assert_eq!(
+        addr_txids(rest_addr),
+        vec![txid.to_string()],
+        "POST /txs/package did not add the accepted tx to electrs' local mempool"
+    );
+
+    rest_handle.stop();
+    Ok(())
+}
+
 // Elements-only tests
 
 #[cfg(feature = "liquid")]