itest: ensure rbf works

itest/tests/swapd.py

@@ -31,6 +31,7 @@
         "chain-poll-interval-seconds": "1",
         "redeem-poll-interval-seconds": "1",
         "preimage-poll-interval-seconds": "1",
+        "whatthefee-poll-interval-seconds": "1",
         "max-swap-amount-sat": "4000000",
         "lock-time": "288",
         "min-confirmations": "1",
@@ -257,7 +258,7 @@ def stop(self, timeout=10):
             return self.rc
 
     def restart(self, timeout=10, clean=True):
-        """Stop and restart the lightning node.
+        """Stop and restart the swapd node.
 
         Keyword arguments:
         timeout: number of seconds to wait for a shutdown
@@ -430,6 +431,7 @@ def __init__(self, port=0):
         self.app.add_url_rule("/", "API entrypoint", self.get_fees, methods=["GET"])
         self.port = port
         self.request_count = 0
+        self.quotient = 1
 
     def get_fees(self):
         self.request_count += 1
@@ -440,7 +442,13 @@ def get_fees(self):
             response.status_code = 500
             return response
 
-        fees = request.args.get("fees")
+        # multiply the caller fees by the quotient.
+        fees = ",".join(
+            map(
+                lambda x: str(int(x) * self.quotient),
+                request.args.get("fees").split(","),
+            )
+        )
         content = (
             '{"index": [3, 6, 9, 12, 18, 24, 36, 48, 72, 96, 144], '
             '"columns": ["0.0500", "0.2000", "0.5000", "0.8000", "0.9500"], '
@@ -476,3 +484,6 @@ def stop(self):
                 self.request_count
             )
         )
+
+    def magnify(self, quotient=2):
+        self.quotient = quotient

itest/tests/test_redeem_rbf.py

@@ -0,0 +1,175 @@
+from binascii import hexlify
+from bitcoin.wallet import CBitcoinSecret
+from pyln.testing.fixtures import (
+    bitcoind,
+    directory,
+    db_provider,
+    executor,
+    jsonschemas,
+    node_cls,
+    node_factory,
+    setup_logging,
+    teardown_checks,
+    test_base_dir,
+    test_name,
+)
+from decimal import Decimal
+from pyln.testing.utils import wait_for
+from fixtures import whatthefee, swapd_factory, postgres_factory
+import hashlib
+import os
+
+
+def test_redeem_rbf_close_to_deadline(node_factory, swapd_factory):
+    user = node_factory.get_node()
+
+    # slow down the redeem poll interval, so the replacement transaction is not
+    # mined during the creation of new blocks.
+    swapper = swapd_factory.get_swapd(
+        options={
+            "redeem-poll-interval-seconds": "4",
+        }
+    )
+    swapper.lightning_node.openchannel(user, 1000000)
+    wait_for(
+        lambda: all(
+            channel["state"] == "CHANNELD_NORMAL"
+            for channel in swapper.lightning_node.rpc.listpeerchannels()["channels"]
+        )
+    )
+    expected_outputs = len(swapper.lightning_node.rpc.listfunds()["outputs"]) + 1
+    user_node_id = user.info["id"]
+    secret_key = CBitcoinSecret.from_secret_bytes(os.urandom(32))
+    public_key = secret_key.pub
+    preimage = os.urandom(32)
+    h = hashlib.sha256(preimage).digest()
+    add_fund_resp = swapper.rpc.add_fund_init(user, public_key, h)
+    txid = user.bitcoin.rpc.sendtoaddress(add_fund_resp.address, 100_000 / 10**8)
+    user.bitcoin.generate_block(1)
+
+    wait_for(
+        lambda: len(swapper.internal_rpc.get_swap(add_fund_resp.address).outputs) > 0
+    )
+
+    payment_request = user.rpc.invoice(
+        100_000_000,
+        "redeem-success",
+        "redeem-success",
+        preimage=hexlify(preimage).decode("ASCII"),
+    )["bolt11"]
+    swapper.rpc.get_swap_payment(payment_request)
+    wait_for(
+        lambda: user.rpc.listinvoices(payment_hash=hexlify(h).decode("ASCII"))[
+            "invoices"
+        ][0]["status"]
+        == "paid"
+    )
+
+    wait_for(lambda: swapper.lightning_node.bitcoin.rpc.getmempoolinfo()["size"] == 1)
+    redeem_txid1 = swapper.lightning_node.bitcoin.rpc.getrawmempool()[0]
+    redeem_raw1 = swapper.lightning_node.bitcoin.rpc.getrawtransaction(
+        redeem_txid1, True
+    )
+    assert redeem_raw1["vin"][0]["txid"] == txid
+    assert redeem_raw1["vout"][0]["value"] == Decimal("0.00099755")
+
+    # Set the effective fee rate of the mempool tx to 0, so it won't be mined
+    swapper.lightning_node.bitcoin.rpc.prioritisetransaction(
+        redeem_txid1, None, -1000000
+    )
+    orig_len = swapper.lightning_node.bitcoin.rpc.getblockcount()
+    swapper.lightning_node.bitcoin.generate_block(288)
+
+    def check_bumped():
+        memp = swapper.lightning_node.bitcoin.rpc.getrawmempool()
+        if len(memp) == 0:
+            return False
+        return memp[0] != redeem_txid1
+
+    wait_for(check_bumped)
+    redeem_txid2 = swapper.lightning_node.bitcoin.rpc.getrawmempool()[0]
+    redeem_raw2 = swapper.lightning_node.bitcoin.rpc.getrawtransaction(
+        redeem_txid2, True
+    )
+    assert redeem_raw2["vin"][0]["txid"] == txid
+    assert redeem_raw2["vout"][0]["value"] == Decimal("0.00099635")
+
+    swapper.lightning_node.bitcoin.generate_block(1)
+    wait_for(
+        lambda: len(swapper.lightning_node.rpc.listfunds()["outputs"])
+        == expected_outputs
+    )
+
+
+def test_redeem_rbf_new_feerate(node_factory, swapd_factory):
+    user = node_factory.get_node()
+    swapper = swapd_factory.get_swapd()
+    swapper.lightning_node.openchannel(user, 1000000)
+    wait_for(
+        lambda: all(
+            channel["state"] == "CHANNELD_NORMAL"
+            for channel in swapper.lightning_node.rpc.listpeerchannels()["channels"]
+        )
+    )
+    expected_outputs = len(swapper.lightning_node.rpc.listfunds()["outputs"]) + 1
+    user_node_id = user.info["id"]
+    secret_key = CBitcoinSecret.from_secret_bytes(os.urandom(32))
+    public_key = secret_key.pub
+    preimage = os.urandom(32)
+    h = hashlib.sha256(preimage).digest()
+    add_fund_resp = swapper.rpc.add_fund_init(user, public_key, h)
+    txid = user.bitcoin.rpc.sendtoaddress(add_fund_resp.address, 100_000 / 10**8)
+    user.bitcoin.generate_block(1)
+
+    wait_for(
+        lambda: len(swapper.internal_rpc.get_swap(add_fund_resp.address).outputs) > 0
+    )
+
+    payment_request = user.rpc.invoice(
+        100_000_000,
+        "redeem-success",
+        "redeem-success",
+        preimage=hexlify(preimage).decode("ASCII"),
+    )["bolt11"]
+    swapper.rpc.get_swap_payment(payment_request)
+    wait_for(
+        lambda: user.rpc.listinvoices(payment_hash=hexlify(h).decode("ASCII"))[
+            "invoices"
+        ][0]["status"]
+        == "paid"
+    )
+
+    wait_for(lambda: swapper.lightning_node.bitcoin.rpc.getmempoolinfo()["size"] == 1)
+    redeem_txid1 = swapper.lightning_node.bitcoin.rpc.getrawmempool()[0]
+    redeem_raw1 = swapper.lightning_node.bitcoin.rpc.getrawtransaction(
+        redeem_txid1, True
+    )
+    assert redeem_raw1["vin"][0]["txid"] == txid
+    assert redeem_raw1["vout"][0]["value"] == Decimal("0.00099755")
+    # Set the effective fee rate of the mempool tx to 0, so it won't be mined
+    swapper.lightning_node.bitcoin.rpc.prioritisetransaction(
+        redeem_txid1, None, -1000000
+    )
+
+    # increase the current feerates by 10x (it's an exponent, so won't be 10x)
+    swapper.whatthefee.magnify(5)
+
+    def check_bumped():
+        memp = swapper.lightning_node.bitcoin.rpc.getrawmempool()
+        if len(memp) == 0:
+            return False
+        return memp[0] != redeem_txid1
+
+    wait_for(check_bumped)
+    redeem_txid2 = swapper.lightning_node.bitcoin.rpc.getrawmempool()[0]
+    redeem_raw2 = swapper.lightning_node.bitcoin.rpc.getrawtransaction(
+        redeem_txid2, True
+    )
+    assert redeem_raw2["vin"][0]["txid"] == txid
+    assert redeem_raw2["vout"][0]["value"] == Decimal("0.00097303")
+
+    swapper.lightning_node.bitcoin.generate_block(1)
+    wait_for(
+        lambda: len(swapper.lightning_node.rpc.listfunds()["outputs"])
+        == expected_outputs
+    )

swapd/src/main.rs

@@ -140,6 +140,10 @@ struct Args {
     #[arg(long, default_value = "60")]
     pub preimage_poll_interval_seconds: u64,
 
+    /// Polling interval between checking whatthefee.io fees.
+    #[arg(long, default_value = "60")]
+    pub whatthefee_poll_interval_seconds: u64,
+
     /// Automatically apply migrations to the database.
     #[arg(long)]
     pub auto_migrate: bool,
@@ -206,7 +210,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         Arc::clone(&chain_client),
         Arc::clone(&chain_filter_repository),
     ));
-    let fee_estimator_1 = WhatTheFeeEstimator::new(args.whatthefee_url, args.lock_time);
+    let fee_estimator_1 = WhatTheFeeEstimator::new(
+        args.whatthefee_url,
+        args.lock_time,
+        Duration::from_secs(args.whatthefee_poll_interval_seconds),
+    );
     fee_estimator_1.start().await?;
     let fee_estimator_2 = bitcoind::FeeEstimator::new(Arc::clone(&chain_client));
     let fee_estimator = Arc::new(FallbackFeeEstimator::new(fee_estimator_1, fee_estimator_2));

swapd/src/public_server.rs

@@ -426,6 +426,12 @@ where
             })
             .await?;
 
+        debug!(
+            label = field::display(&label),
+            hash = field::display(hash),
+            "successfully paid"
+        );
+
         // Persist the preimage right away. There's also a background service
         // checking for preimages, in case the `pay` call failed, but the
         // payment did succeed.

swapd/src/redeem/monitor.rs

@@ -99,10 +99,10 @@ where
         }
 
         loop {
-            debug!("starting chain sync task");
-            match self.do_sync().await {
-                Ok(_) => debug!("chain sync task completed succesfully"),
-                Err(e) => error!("chain sync task failed with: {:?}", e),
+            debug!("starting redeem task");
+            match self.do_redeem().await {
+                Ok(_) => debug!("redeem task completed succesfully"),
+                Err(e) => error!("redeem task failed with: {:?}", e),
             }
 
             tokio::select! {
@@ -118,7 +118,7 @@ where
     }
 
     #[instrument(skip(self), level = "trace")]
-    async fn do_sync(&self) -> Result<(), RedeemError> {
+    async fn do_redeem(&self) -> Result<(), RedeemError> {
         let current_height = self.chain_client.get_blockheight().await?;
         let redeemables = self.redeem_service.list_redeemable().await?;
         let redeemables: HashMap<_, _> = redeemables

swapd/src/whatthefee/estimator.rs

@@ -7,7 +7,7 @@ use reqwest::Url;
 use serde::Deserialize;
 use thiserror::Error;
 use tokio::{sync::Mutex, time::MissedTickBehavior};
-use tracing::error;
+use tracing::{error, field, trace};
 
 use crate::chain::{FeeEstimate, FeeEstimateError, FeeEstimator};
 
@@ -31,6 +31,7 @@ pub struct WhatTheFeeEstimator {
     url: Url,
     lock_time: u32,
     last_response: Arc<Mutex<Option<LastResponse>>>,
+    poll_interval: Duration,
 }
 
 #[derive(Debug, Error)]
@@ -40,11 +41,12 @@ pub enum WhatTheFeeError {
 }
 
 impl WhatTheFeeEstimator {
-    pub fn new(url: Url, lock_time: u32) -> Self {
+    pub fn new(url: Url, lock_time: u32, poll_interval: Duration) -> Self {
         Self {
             url,
             lock_time,
             last_response: Arc::new(Mutex::new(None)),
+            poll_interval,
         }
     }
 
@@ -58,8 +60,9 @@ impl WhatTheFeeEstimator {
     fn run_forever(&self) {
         let last_response = Arc::clone(&self.last_response);
         let url = self.url.clone();
+        let poll_interval = self.poll_interval;
         tokio::spawn(async move {
-            let mut interval = tokio::time::interval(Duration::from_secs(60));
+            let mut interval = tokio::time::interval(poll_interval);
             interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
             loop {
                 interval.tick().await;
@@ -157,6 +160,10 @@ async fn get_fees(url: &Url) -> Result<LastResponse, WhatTheFeeError> {
         .await?
         .json::<WhatTheFeeResponse>()
         .await?;
+    trace!(
+        response = field::debug(&response),
+        "got whatthefee response"
+    );
     Ok(LastResponse {
         response,
         timestamp: now,