Broadcast v2 fallback when session closed

Co-authored-by: zealsham <shammahagwor@gmail.com>

After a replay, if the session is closed due to being cancled or failure
we broadcast the fallback.

Author: arminsabouri

payjoin-cli/src/app/v2/mod.rs

@@ -249,7 +249,23 @@ impl AppTrait for App {
                 };
                 let mut interrupt = self.interrupt.clone();
                 tokio::select! {
-                    res = self.process_sender_session(sender_state, &persister) => return res,
+                    res = self.process_sender_session(sender_state, &persister) => {
+                        match res {
+                            Ok(()) => return Ok(()),
+                            Err(err) => {
+                                match self.maybe_broadcast_sender_fallback(&persister) {
+                                    Ok(true) => return Ok(()),
+                                    Ok(false) => return Err(err),
+                                    Err(broadcast_err) => {
+                                        return Err(err.context(format!(
+                                            "Failed to auto-broadcast fallback after sender session error: {}",
+                                            broadcast_err
+                                        )));
+                                    }
+                                }
+                            }
+                        }
+                    },
                     _ = interrupt.changed() => {
                         println!("Interrupted. Call `send` with the same arguments to resume this session or `resume` to resume all sessions.");
                         return Err(anyhow!("Interrupted"))
@@ -472,6 +488,20 @@ impl AppTrait for App {
 }
 
 impl App {
+    fn maybe_broadcast_sender_fallback(&self, persister: &SenderPersister) -> Result<bool> {
+        let (session, history) = replay_sender_event_log(persister)?;
+        match session {
+            SendSession::Closed(SenderSessionOutcome::Failure)
+            | SendSession::Closed(SenderSessionOutcome::Cancel) => {
+                let fallback_tx = history.fallback_tx();
+                self.wallet().broadcast_tx(&fallback_tx)?;
+                println!("Broadcasted fallback transaction txid: {}", fallback_tx.compute_txid());
+                Ok(true)
+            }
+            _ => Ok(false),
+        }
+    }
+
     fn close_failed_session<P>(persister: &P, session_id: &SessionId, role: &str)
     where
         P: SessionPersister,
@@ -497,7 +527,11 @@ impl App {
                 self.process_pj_response(proposal)?;
                 return Ok(());
             }
-            _ => return Err(anyhow!("Unexpected sender state")),
+            SendSession::Closed(SenderSessionOutcome::Failure)
+            | SendSession::Closed(SenderSessionOutcome::Cancel) => {
+                self.maybe_broadcast_sender_fallback(persister)?;
+                return Ok(());
+            }
         }
         Ok(())
     }
@@ -872,3 +906,131 @@ impl App {
             .context("HTTP request failed")
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::sync::{Arc, Mutex};
+
+    use payjoin::bitcoin::{Amount, FeeRate};
+    use payjoin::persist::{NoopSessionPersister, SessionPersister};
+    use payjoin::receive::v2::ReceiverBuilder;
+    use payjoin::send::v2::{
+        SendSession, SenderBuilder, SessionEvent as SenderSessionEvent,
+        SessionOutcome as SenderSessionOutcome,
+    };
+    use payjoin::{PjParam, Uri, UriExt};
+    use payjoin_test_utils::{corepc_node, TestServices};
+    use tempfile::tempdir;
+    use tokio::sync::watch;
+
+    use super::ohttp::RelayManager;
+    use super::App;
+    use crate::app::config::{BitcoindConfig, Config};
+    use crate::app::wallet::BitcoindWallet;
+    use crate::app::App as AppTrait;
+    use crate::db::v2::SenderPersister;
+    use crate::db::Database;
+
+    async fn test_app(
+        bitcoind: &corepc_node::Node,
+        wallet_name: &str,
+    ) -> (App, BitcoindWallet, tempfile::TempDir) {
+        let rpchost = payjoin::Url::parse(&format!(
+            "http://{}/wallet/{}",
+            bitcoind.params.rpc_socket, wallet_name
+        ))
+        .expect("valid url");
+        let bitcoind_config = BitcoindConfig {
+            rpchost: rpchost.clone(),
+            cookie: Some(bitcoind.params.cookie_file.clone()),
+            rpcuser: String::new(),
+            rpcpassword: String::new(),
+        };
+        let wallet = BitcoindWallet::new(&bitcoind_config).await.expect("wallet should connect");
+        let temp_dir = tempdir().expect("temp dir");
+        let db = Arc::new(Database::create(temp_dir.path().join("test.db")).expect("db created"));
+        let (_, interrupt_rx) = watch::channel(());
+        let config = Config {
+            db_path: temp_dir.path().to_path_buf(),
+            max_fee_rate: None,
+            bitcoind: bitcoind_config,
+            version: None,
+            #[cfg(feature = "_manual-tls")]
+            root_certificate: None,
+            #[cfg(feature = "_manual-tls")]
+            certificate_key: None,
+        };
+        let app = App {
+            config,
+            db,
+            wallet: wallet.clone(),
+            interrupt: interrupt_rx,
+            relay_manager: Arc::new(Mutex::new(RelayManager::new())),
+        };
+        (app, wallet, temp_dir)
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn sender_processes_fallback_on_failure() {
+        let (bitcoind, _sender, receiver) =
+            payjoin_test_utils::init_bitcoind_sender_receiver(None, None)
+                .expect("bitcoind should start");
+        let (app, wallet, _temp_dir) = test_app(&bitcoind, "sender").await;
+        let mut services = TestServices::initialize().await.expect("services should initialize");
+        services.wait_for_services_ready().await.expect("services should be healthy");
+
+        let receive_address = receiver.new_address().expect("receiver address should be created");
+        let ohttp_keys = services.fetch_ohttp_keys().await.expect("ohttp keys should be fetched");
+        let receiver_session =
+            ReceiverBuilder::new(receive_address, services.directory_url().as_str(), ohttp_keys)
+                .expect("receiver builder should initialize")
+                .with_amount(Amount::from_sat(50_000))
+                .build()
+                .save(&NoopSessionPersister::default())
+                .expect("receiver session should persist");
+        let pj_uri = receiver_session.pj_uri();
+
+        let uri = Uri::try_from(pj_uri.to_string())
+            .expect("pj uri string should parse")
+            .assume_checked()
+            .check_pj_supported()
+            .expect("pj uri should support payjoin");
+        let (pj_param, amount) = match (uri.extras.pj_param(), uri.amount) {
+            (PjParam::V2(pj_param), Some(amount)) => (pj_param, amount),
+            _ => panic!("receiver should produce a v2 URI with amount"),
+        };
+
+        let original_psbt = app
+            .create_original_psbt(&uri.address, amount, FeeRate::BROADCAST_MIN)
+            .expect("original psbt should be created");
+        let sender =
+            SenderBuilder::from_parts(original_psbt, &pj_param, &uri.address, Some(amount))
+                .build_recommended(FeeRate::BROADCAST_MIN)
+                .expect("sender should be created");
+
+        let sender_persister =
+            SenderPersister::new(app.db.clone(), &pj_uri.to_string(), pj_param.receiver_pubkey())
+                .expect("persister created");
+        sender.save(&sender_persister).expect("sender should persist");
+
+        // Manually mark the sender session as failed to trigger the fallback
+        sender_persister
+            .save_event(SenderSessionEvent::Closed(SenderSessionOutcome::Failure))
+            .expect("failure event should be saved");
+        app.process_sender_session(
+            SendSession::Closed(SenderSessionOutcome::Failure),
+            &sender_persister,
+        )
+        .await
+        .expect("sender failure should process");
+
+        let (_, history) =
+            payjoin::send::v2::replay_event_log(&sender_persister).expect("sender history replays");
+        let fallback_tx = history.fallback_tx();
+        wallet
+            .get_raw_transaction(&fallback_tx.compute_txid())
+            .expect("fallback tx should be broadcast");
+        let _ = services.take_ohttp_relay_handle();
+        let _ = services.take_directory_handle();
+    }
+}