Add v2 fallback subcommand

arminsabouri · arminsabouri · commit 5f292b1c7815 · 2026-05-05T09:44:58.000-04:00
Co-authored-by: zealsham &lt;shammahagwor@gmail.com&gt;
Co-authored-by: DanGould &lt;d@ngould.dev&gt;

The command broadcasts the fallback tx for a specific sender session.
If a session fails or is canceled, pj-cli outputs the fallback command
for that particular session.
diff --git a/payjoin-cli/src/app/config.rs b/payjoin-cli/src/app/config.rs
@@ -336,6 +336,8 @@ fn handle_subcommands(config: Builder, cli: &Cli) -> Result<Builder, ConfigError
         Commands::Resume => Ok(config),
         #[cfg(feature = "v2")]
         Commands::History => Ok(config),
+        #[cfg(feature = "v2")]
+        Commands::Fallback { .. } => Ok(config),
     }
 }
 
diff --git a/payjoin-cli/src/app/mod.rs b/payjoin-cli/src/app/mod.rs
@@ -10,6 +10,8 @@ pub mod config;
 pub mod wallet;
 use crate::app::config::Config;
 use crate::app::wallet::BitcoindWallet;
+#[cfg(feature = "v2")]
+use crate::db::v2::SessionId;
 
 #[cfg(feature = "v1")]
 pub(crate) mod v1;
@@ -28,6 +30,8 @@ pub trait App: Send + Sync {
     async fn resume_payjoins(&self) -> Result<()>;
     #[cfg(feature = "v2")]
     async fn history(&self) -> Result<()>;
+    #[cfg(feature = "v2")]
+    async fn fallback_sender(&self, session_id: SessionId) -> Result<()>;
 
     fn create_original_psbt(
         &self,
diff --git a/payjoin-cli/src/app/v1.rs b/payjoin-cli/src/app/v1.rs
@@ -116,6 +116,11 @@ impl AppTrait for App {
     async fn history(&self) -> Result<()> {
         unimplemented!("history not implemented for v1");
     }
+
+    #[cfg(feature = "v2")]
+    async fn fallback_sender(&self, _session_id: crate::db::v2::SessionId) -> Result<()> {
+        anyhow::bail!("fallback is only supported for v2 (BIP77) sessions")
+    }
 }
 
 impl App {
diff --git a/payjoin-cli/src/app/v2/mod.rs b/payjoin-cli/src/app/v2/mod.rs
@@ -241,9 +241,21 @@ 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) => {
+                                let id = persister.session_id();
+                                println!("Session {id} failed. Run `payjoin-cli fallback {id}` to broadcast the original transaction.");
+                                return Err(err);
+                            }
+                        }
+                    },
                     _ = interrupt.changed() => {
-                        println!("Interrupted. Call `send` with the same arguments to resume this session or `resume` to resume all sessions.");
+                        let id = persister.session_id();
+                        println!(
+                            "Session {id} interrupted. Call `send` again to resume, `resume` to resume all sessions, or `payjoin-cli fallback {id}` to broadcast the original transaction."
+                        );
                         return Err(anyhow!("Interrupted"))
                     }
                 }
@@ -461,6 +473,32 @@ impl AppTrait for App {
 
         Ok(())
     }
+
+    async fn fallback_sender(&self, session_id: SessionId) -> Result<()> {
+        let persister = SenderPersister::from_id(self.db.clone(), session_id.clone());
+        let (session, history) = replay_sender_event_log(&persister)?;
+
+        if let SendSession::Closed(SenderSessionOutcome::Success(proposal)) = session {
+            let txid = proposal.clone().extract_tx_unchecked_fee_rate().compute_txid();
+            println!(
+                "Session {session_id} already produced payjoin transaction {txid}. \
+                 Broadcasting the original now would double-spend against it. \
+                 If the payjoin tx needs re-broadcast, run \
+                 `bitcoin-cli gettransaction {txid}` to fetch the hex, then \
+                 `bitcoin-cli sendrawtransaction <hex>`."
+            );
+            return Ok(());
+        }
+
+        let fallback_tx = history.fallback_tx();
+        self.wallet().broadcast_tx(&fallback_tx)?;
+        println!("Broadcasted fallback transaction txid: {}", fallback_tx.compute_txid());
+
+        if let Err(e) = SessionPersister::close(&persister) {
+            tracing::warn!("Failed to close session {session_id} after fallback: {e}");
+        }
+        Ok(())
+    }
 }
 
 impl App {
@@ -489,7 +527,14 @@ impl App {
                 self.process_pj_response(proposal)?;
                 return Ok(());
             }
-            _ => return Err(anyhow!("Unexpected sender state")),
+            SendSession::Closed(SenderSessionOutcome::Failure)
+            | SendSession::Closed(SenderSessionOutcome::Cancel) => {
+                let id = persister.session_id();
+                println!(
+                    "Session {id} ended without payjoin. Run `payjoin-cli fallback {id}` to broadcast the original transaction."
+                );
+                return Ok(());
+            }
         }
         Ok(())
     }
diff --git a/payjoin-cli/src/cli/mod.rs b/payjoin-cli/src/cli/mod.rs
@@ -133,6 +133,13 @@ pub enum Commands {
     #[cfg(feature = "v2")]
     /// Show payjoin session history
     History,
+    #[cfg(feature = "v2")]
+    /// Broadcast the original transaction for a sender session (BIP77/v2 only)
+    Fallback {
+        /// The session ID to broadcast the fallback transaction for
+        #[arg(required = true)]
+        session_id: i64,
+    },
 }
 
 pub fn parse_amount_in_sat(s: &str) -> Result<Amount, ParseAmountError> {
diff --git a/payjoin-cli/src/db/v2.rs b/payjoin-cli/src/db/v2.rs
@@ -9,7 +9,7 @@ use rusqlite::params;
 use super::*;
 
 #[derive(Debug, Clone)]
-pub(crate) struct SessionId(i64);
+pub(crate) struct SessionId(pub(crate) i64);
 
 impl core::ops::Deref for SessionId {
     type Target = i64;
@@ -61,6 +61,8 @@ impl SenderPersister {
     }
 
     pub fn from_id(db: Arc<Database>, id: SessionId) -> Self { Self { db, session_id: id } }
+
+    pub fn session_id(&self) -> SessionId { self.session_id.clone() }
 }
 impl SessionPersister for SenderPersister {
     type SessionEvent = SenderSessionEvent;
diff --git a/payjoin-cli/src/main.rs b/payjoin-cli/src/main.rs
@@ -6,6 +6,9 @@ use cli::{Cli, Commands};
 use tracing_subscriber::filter::LevelFilter;
 use tracing_subscriber::EnvFilter;
 
+#[cfg(feature = "v2")]
+use crate::db::v2::SessionId;
+
 mod app;
 mod cli;
 mod db;
@@ -75,6 +78,10 @@ async fn main() -> Result<()> {
         Commands::History => {
             app.history().await?;
         }
+        #[cfg(feature = "v2")]
+        Commands::Fallback { session_id } => {
+            app.fallback_sender(SessionId(*session_id)).await?;
+        }
     };
 
     Ok(())
diff --git a/payjoin-cli/tests/e2e.rs b/payjoin-cli/tests/e2e.rs
@@ -1,13 +1,15 @@
 #[cfg(feature = "_manual-tls")]
 mod e2e {
     use std::process::{ExitStatus, Stdio};
+    use std::str::FromStr;
 
     use nix::sys::signal::{kill, Signal};
     use nix::unistd::Pid;
+    use payjoin::bitcoin::Txid;
     use payjoin_test_utils::{init_bitcoind_sender_receiver, BoxError};
     use tempfile::tempdir;
     use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
-    use tokio::process::Command;
+    use tokio::process::{Child, Command};
 
     async fn terminate(mut child: tokio::process::Child) -> tokio::io::Result<ExitStatus> {
         let pid = child.id().expect("Failed to get child PID");
@@ -64,6 +66,20 @@ mod e2e {
         res
     }
 
+    async fn send_until_request_timeout(mut cli_sender: Child) -> Result<(), BoxError> {
+        let mut stdout = cli_sender.stdout.take().expect("failed to take stdout of child process");
+        let timeout = tokio::time::Duration::from_secs(35);
+        let res = tokio::time::timeout(
+            timeout,
+            wait_for_stdout_match(&mut stdout, |line| line.contains("No response yet.")),
+        )
+        .await?;
+
+        terminate(cli_sender).await.expect("Failed to kill payjoin-cli initial sender");
+        assert!(res.is_some(), "Fallback send was not detected");
+        Ok(())
+    }
+
     #[cfg(feature = "v1")]
     #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
     async fn send_receive_payjoin_v1() -> Result<(), BoxError> {
@@ -201,7 +217,6 @@ mod e2e {
     async fn send_receive_payjoin_v2() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         use payjoin_test_utils::{init_tracing, TestServices};
         use tempfile::TempDir;
-        use tokio::process::Child;
 
         type Result<T> = std::result::Result<T, BoxError>;
 
@@ -385,21 +400,6 @@ mod e2e {
             Ok(())
         }
 
-        async fn send_until_request_timeout(mut cli_sender: Child) -> Result<()> {
-            let mut stdout =
-                cli_sender.stdout.take().expect("failed to take stdout of child process");
-            let timeout = tokio::time::Duration::from_secs(35);
-            let res = tokio::time::timeout(
-                timeout,
-                wait_for_stdout_match(&mut stdout, |line| line.contains("No response yet.")),
-            )
-            .await?;
-
-            terminate(cli_sender).await.expect("Failed to kill payjoin-cli initial sender");
-            assert!(res.is_some(), "Fallback send was not detected");
-            Ok(())
-        }
-
         async fn respond_with_payjoin(mut cli_receive_resumer: Child) -> Result<()> {
             let mut stdout =
                 cli_receive_resumer.stdout.take().expect("Failed to take stdout of child process");
@@ -621,4 +621,142 @@ mod e2e {
 
         Ok(())
     }
+
+    #[cfg(feature = "v2")]
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn sender_fallback_v2() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        use payjoin_test_utils::{init_tracing, TestServices};
+        use tempfile::TempDir;
+
+        type Result<T> = std::result::Result<T, BoxError>;
+
+        init_tracing();
+        let mut services = TestServices::initialize().await?;
+        let temp_dir = tempdir()?;
+
+        let result = tokio::select! {
+            res = services.take_ohttp_relay_handle() => Err(format!("Ohttp relay is long running: {res:?}").into()),
+            res = services.take_directory_handle() => Err(format!("Directory server is long running: {res:?}").into()),
+            res = fallback_cli_async(&services, &temp_dir) => res,
+        };
+
+        assert!(result.is_ok(), "sender_fallback_v2 failed: {:#?}", result.unwrap_err());
+
+        async fn fallback_cli_async(services: &TestServices, temp_dir: &TempDir) -> Result<()> {
+            let sender_db_path = temp_dir.path().join("sender_db");
+            let (bitcoind, sender, _receiver) = init_bitcoind_sender_receiver(None, None)?;
+            let cert_path = &temp_dir.path().join("localhost.der");
+            tokio::fs::write(cert_path, services.cert()).await?;
+            services.wait_for_services_ready().await?;
+            let ohttp_keys = services.fetch_ohttp_keys().await?;
+            let ohttp_keys_path = temp_dir.path().join("ohttp_keys");
+            tokio::fs::write(&ohttp_keys_path, ohttp_keys.encode()?).await?;
+
+            let receiver_db_path = temp_dir.path().join("receiver_db");
+            let receiver_rpchost = format!("http://{}/wallet/receiver", bitcoind.params.rpc_socket);
+            let sender_rpchost = format!("http://{}/wallet/sender", bitcoind.params.rpc_socket);
+            let cookie_file = &bitcoind.params.cookie_file;
+            let payjoin_cli = env!("CARGO_BIN_EXE_payjoin-cli");
+            let directory = &services.directory_url();
+            let ohttp_relay = &services.ohttp_relay_url();
+
+            // Get a BIP21 from a receiver then kill it so the sender can never complete payjoin
+            let cli_receiver = Command::new(payjoin_cli)
+                .arg("--root-certificate")
+                .arg(cert_path)
+                .arg("--rpchost")
+                .arg(&receiver_rpchost)
+                .arg("--cookie-file")
+                .arg(cookie_file)
+                .arg("--db-path")
+                .arg(&receiver_db_path)
+                .arg("--ohttp-relays")
+                .arg(ohttp_relay)
+                .arg("receive")
+                .arg(RECEIVE_SATS)
+                .arg("--pj-directory")
+                .arg(directory)
+                .arg("--ohttp-keys")
+                .arg(&ohttp_keys_path)
+                .stdout(Stdio::piped())
+                .stderr(Stdio::inherit())
+                .spawn()
+                .expect("Failed to execute payjoin-cli receiver");
+            let bip21 = get_bip21_from_receiver(cli_receiver).await;
+
+            // Start sender and capture the session-id from the hint line, then interrupt
+            let cli_sender = Command::new(payjoin_cli)
+                .arg("--root-certificate")
+                .arg(cert_path)
+                .arg("--rpchost")
+                .arg(&sender_rpchost)
+                .arg("--cookie-file")
+                .arg(cookie_file)
+                .arg("--db-path")
+                .arg(&sender_db_path)
+                .arg("--ohttp-relays")
+                .arg(ohttp_relay)
+                .arg("send")
+                .arg(&bip21)
+                .arg("--fee-rate")
+                .arg("1")
+                .stdout(Stdio::piped())
+                .stderr(Stdio::inherit())
+                .spawn()
+                .expect("Failed to execute payjoin-cli sender");
+
+            send_until_request_timeout(cli_sender).await?;
+
+            // There is only one sender session in progress.
+            let session_id = 1i64;
+            // Ensure the fallback was not broadcast yet
+            let mempool_size =
+                sender.get_mempool_info().expect("should be able to get mempool").unbroadcast_count;
+            assert_eq!(mempool_size, 0, "fallback should not be in mempool");
+
+            // Run `payjoin-cli fallback <session-id>` and assert broadcast
+            let mut cli_fallback = Command::new(payjoin_cli)
+                .arg("--root-certificate")
+                .arg(cert_path)
+                .arg("--rpchost")
+                .arg(&sender_rpchost)
+                .arg("--cookie-file")
+                .arg(cookie_file)
+                .arg("--db-path")
+                .arg(&sender_db_path)
+                .arg("--ohttp-relays")
+                .arg(ohttp_relay)
+                .arg("fallback")
+                .arg(session_id.to_string())
+                .stdout(Stdio::piped())
+                .stderr(Stdio::inherit())
+                .spawn()
+                .expect("Failed to execute payjoin-cli fallback");
+
+            let mut fallback_stdout =
+                cli_fallback.stdout.take().expect("failed to take stdout of fallback");
+            let timeout = tokio::time::Duration::from_secs(10);
+            let broadcast_line = tokio::time::timeout(
+                timeout,
+                wait_for_stdout_match(&mut fallback_stdout, |l| {
+                    l.contains("Broadcasted fallback transaction txid")
+                }),
+            )
+            .await?;
+
+            terminate(cli_fallback).await.expect("Failed to kill payjoin-cli fallback");
+            let subcommand_output = broadcast_line.expect("fallback should broadcast");
+            let fallback_txid = subcommand_output.split_whitespace().nth(4).unwrap_or("");
+            let fallback_txid = Txid::from_str(fallback_txid).expect("valid txid");
+
+            assert!(
+                sender.get_raw_transaction(fallback_txid).is_ok(),
+                "fallback tx should be in the mempool"
+            );
+
+            Ok(())
+        }
+
+        Ok(())
+    }
 }

Original file line number	Diff line number	Diff line change
`@@ -336,6 +336,8 @@ fn handle_subcommands(config: Builder, cli: &Cli) -> Result<Builder, ConfigError`
`336`	`336`	`Commands::Resume => Ok(config),`
`337`	`337`	`#[cfg(feature = "v2")]`
`338`	`338`	`Commands::History => Ok(config),`
	`339`	`+ #[cfg(feature = "v2")]`
	`340`	`+ Commands::Fallback { .. } => Ok(config),`
`339`	`341`	`}`
`340`	`342`	`}`
`341`	`343`
Original file line number	Diff line number	Diff line change
`@@ -116,6 +116,11 @@ impl AppTrait for App {`
`116`	`116`	`async fn history(&self) -> Result<()> {`
`117`	`117`	`unimplemented!("history not implemented for v1");`
`118`	`118`	`}`
	`119`	`+`
	`120`	`+ #[cfg(feature = "v2")]`
	`121`	`+ async fn fallback_sender(&self, _session_id: crate::db::v2::SessionId) -> Result<()> {`
	`122`	`+ anyhow::bail!("fallback is only supported for v2 (BIP77) sessions")`
	`123`	`+ }`
`119`	`124`	`}`
`120`	`125`
`121`	`126`	`impl App {`
Original file line number	Diff line number	Diff line change
`@@ -9,7 +9,7 @@ use rusqlite::params;`
`9`	`9`	`use super::*;`
`10`	`10`
`11`	`11`	`#[derive(Debug, Clone)]`
`12`		`-pub(crate) struct SessionId(i64);`
	`12`	`+pub(crate) struct SessionId(pub(crate) i64);`
`13`	`13`
`14`	`14`	`impl core::ops::Deref for SessionId {`
`15`	`15`	`type Target = i64;`
`@@ -61,6 +61,8 @@ impl SenderPersister {`
`61`	`61`	`}`
`62`	`62`
`63`	`63`	`pub fn from_id(db: Arc<Database>, id: SessionId) -> Self { Self { db, session_id: id } }`
	`64`	`+`
	`65`	`+ pub fn session_id(&self) -> SessionId { self.session_id.clone() }`
`64`	`66`	`}`
`65`	`67`	`impl SessionPersister for SenderPersister {`
`66`	`68`	`type SessionEvent = SenderSessionEvent;`