Force cli to cancel and broadcast fallback when available mid loop

Previously the cli only checked for expiry at program start during long
polling loops there was not an expiry check leading to the possiblity of
an indefinite poll past expiry where the counterparty had already given
up.

This also adds an `--expire-in` flag for the receiver to both test this
functionality and demonstrate that the receiver is not actually fixed to
a 24hr expiration and is in controll of this expiry time.

Author: benalleng

payjoin-cli/src/app/config.rs

@@ -61,6 +61,9 @@ pub struct Config {
     pub root_certificate: Option<PathBuf>,
     #[cfg(feature = "_manual-tls")]
     pub certificate_key: Option<PathBuf>,
+    #[cfg(feature = "v2")]
+    #[serde(skip)]
+    pub expire_in_secs: Option<u64>,
 }
 
 impl Config {
@@ -156,6 +159,8 @@ impl Config {
             max_fee_rate: built_config.get("max_fee_rate").ok(),
             bitcoind: built_config.get("bitcoind")?,
             version: None,
+            #[cfg(feature = "v2")]
+            expire_in_secs: None,
             #[cfg(feature = "_manual-tls")]
             root_certificate: built_config.get("root_certificate").ok(),
             #[cfg(feature = "_manual-tls")]
@@ -243,6 +248,11 @@ impl Config {
             ));
         }
 
+        #[cfg(feature = "v2")]
+        if let Commands::Receive { expire_in, .. } = &cli.command {
+            config.expire_in_secs = *expire_in;
+        }
+
         tracing::trace!("App config: {config:?}");
         Ok(config)
     }

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

@@ -296,6 +296,10 @@ impl AppTrait for App {
         if let Some(max_fee_rate) = self.config.max_fee_rate {
             receiver_builder = receiver_builder.with_max_fee_rate(max_fee_rate);
         }
+        if let Some(expire_in_secs) = self.config.expire_in_secs {
+            let expiration = std::time::Duration::from_secs(expire_in_secs);
+            receiver_builder = receiver_builder.with_expiration(expiration);
+        }
         let session = receiver_builder.build().save(&persister)?;
         println!("Receive session established: {}", persister.session_id());
 
@@ -337,6 +341,12 @@ impl AppTrait for App {
                 Err(e) => {
                     if e.is_expired() {
                         println!("Session {session_id} receiver expired.");
+                        match e.fallback_tx() {
+                            Some(tx) => self.broadcast_fallback_tx(tx, &session_id, "receiver"),
+                            None => println!(
+                                        "No fallback transaction available for expired receiver session {session_id}."
+                            ),
+                        }
                     } else {
                         tracing::error!(
                             "An error {:?} occurred while replaying receiver session",
@@ -365,6 +375,12 @@ impl AppTrait for App {
                 Err(e) => {
                     if e.is_expired() {
                         println!("Session {session_id} sender expired.");
+                        match e.fallback_tx() {
+                            Some(tx) => self.broadcast_fallback_tx(tx, &session_id, "sender"),
+                            None => println!(
+                                "No fallback transaction available for expired sender session {session_id}."
+                            ),
+                        }
                     } else {
                         tracing::error!("An error {:?} occurred while replaying Sender session", e);
                         println!("Session {session_id} sender failed to replay -  {e}");
@@ -688,7 +704,31 @@ impl App {
         if let Err(close_err) = SessionPersister::close(persister) {
             tracing::error!("Failed to close {} session {}: {:?}", role, session_id, close_err);
         } else {
-            tracing::error!("Closed failed {} session: {}", role, session_id);
+            tracing::debug!("Closed failed {} session: {}", role, session_id);
+        }
+    }
+
+    /// Broadcast a fallback transaction with graceful error handling. Logs
+    /// success/failure, never propagates the broadcast error.
+    fn broadcast_fallback_tx(
+        &self,
+        tx: &payjoin::bitcoin::Transaction,
+        session_id: &SessionId,
+        role: &str,
+    ) {
+        match self.wallet().broadcast_tx(tx) {
+            Ok(txid) =>
+                println!("Broadcasted fallback transaction for {role} session {session_id}: {txid}"),
+            Err(err) => {
+                println!(
+                    "Expired {role} session {session_id} has a fallback transaction \
+                     but it failed to broadcast: {err}\n\
+                     Recover it manually with `payjoin-cli history` / `cancel`."
+                );
+                tracing::error!(
+                    "Fallback broadcast failed for {role} session {session_id}: {err:?}"
+                );
+            }
         }
     }
 
@@ -739,26 +779,53 @@ impl App {
         persister: &SenderPersister,
     ) -> Result<()> {
         let mut session = sender.clone();
-        // Long poll until we get a response
+        // Long poll until we get a response. The session's expiration is enforced
+        // here: once `create_poll_request` reports expiry, the session is driven to
+        // a terminal `Closed(Aborted)` state and the fallback transaction is surfaced
+        // so it can be broadcast manually. The persisted terminal outcome prevents
+        // the session from being mistakenly resumed.
         loop {
-            let (response, ctx) =
-                self.post_via_relay(|relay| session.create_poll_request(relay)).await?;
-            let res = session.process_response(&response.bytes().await?, ctx).save(persister);
-            match res {
-                Ok(OptionalTransitionOutcome::Progress(psbt)) => {
-                    println!("Proposal received. Processing...");
-                    self.process_pj_response(psbt)?;
-                    return Ok(());
+            let relay = self.mailroom_manager.choose_relay()?;
+            let (req, ctx) = match session.create_poll_request(relay.as_str()) {
+                Ok(r) => r,
+                Err(e) if e.is_expired() => {
+                    let pending = session.cancel().save(persister)?;
+                    self.broadcast_fallback_tx(
+                        pending.fallback_tx(),
+                        &persister.session_id(),
+                        "sender",
+                    );
+                    pending.close().save(persister)?;
+                    return Err(anyhow!("Sender session expired"));
                 }
-                Ok(OptionalTransitionOutcome::Stasis(current_state)) => {
-                    println!("No response yet.");
-                    session = current_state;
-                    continue;
+                Err(e) => return Err(e.into()),
+            };
+            match self.post_request(req).await {
+                Ok(response) => {
+                    let bytes = response.bytes().await?;
+                    let res = session.process_response(&bytes, ctx).save(persister);
+                    match res {
+                        Ok(OptionalTransitionOutcome::Progress(psbt)) => {
+                            println!("Proposal received. Processing...");
+                            self.process_pj_response(psbt)?;
+                            return Ok(());
+                        }
+                        Ok(OptionalTransitionOutcome::Stasis(current_state)) => {
+                            println!("No response yet.");
+                            session = current_state;
+                            continue;
+                        }
+                        Err(re) => {
+                            println!("{re}");
+                            tracing::debug!("{re:?}");
+                            return Err(anyhow!("Response error").context(re));
+                        }
+                    }
                 }
-                Err(re) => {
-                    println!("{re}");
-                    tracing::debug!("{re:?}");
-                    return Err(anyhow!("Response error").context(re));
+                Err(e) => {
+                    tracing::debug!("Request to relay {relay} failed: {e:?}");
+                    self.mailroom_manager.add_failed_relay(relay);
+                    continue;
                 }
             }
         }
@@ -771,22 +838,44 @@ impl App {
     ) -> Result<Receiver<UncheckedOriginalPayload>> {
         let mut session = session;
         loop {
+            let relay = self.mailroom_manager.choose_relay()?;
+            let (req, context) = match session.create_poll_request(relay.as_str()) {
+                Ok(r) => r,
+                Err(e) if e.is_expired() => {
+                    session.cancel().save(persister)?;
+                    println!(
+                        "Receiver session expired before any proposal was received; \
+                         no fallback transaction to broadcast."
+                    );
+                    return Err(anyhow!("Receiver session expired"));
+                }
+                Err(e) => return Err(e.into()),
+            };
             println!("Polling receive request...");
-            let (ohttp_response, context) =
-                self.post_via_relay(|relay| session.create_poll_request(relay)).await?;
-            let state_transition = session
-                .process_response(ohttp_response.bytes().await?.to_vec().as_slice(), context)
-                .save(persister);
-            match state_transition {
-                Ok(OptionalTransitionOutcome::Progress(next_state)) => {
-                    println!("Got a request from the sender. Responding with a Payjoin proposal.");
-                    return Ok(next_state);
+            match self.post_request(req).await {
+                Ok(ohttp_response) => {
+                    let bytes = ohttp_response.bytes().await?.to_vec();
+                    let state_transition =
+                        session.process_response(bytes.as_slice(), context).save(persister);
+                    match state_transition {
+                        Ok(OptionalTransitionOutcome::Progress(next_state)) => {
+                            println!(
+                                "Got a request from the sender. Responding with a Payjoin proposal."
+                            );
+                            return Ok(next_state);
+                        }
+                        Ok(OptionalTransitionOutcome::Stasis(current_state)) => {
+                            session = current_state;
+                            continue;
+                        }
+                        Err(e) => return Err(e.into()),
+                    }
                 }
-                Ok(OptionalTransitionOutcome::Stasis(current_state)) => {
-                    session = current_state;
+                Err(e) => {
+                    tracing::debug!("Request to relay {relay} failed: {e:?}");
+                    self.mailroom_manager.add_failed_relay(relay);
                     continue;
                 }
-                Err(e) => return Err(e.into()),
             }
         }
     }

payjoin-cli/src/cli/mod.rs

@@ -125,6 +125,11 @@ pub enum Commands {
         /// The path to the ohttp keys file
         #[arg(long = "ohttp-keys", value_parser = value_parser!(PathBuf))]
         ohttp_keys: Option<PathBuf>,
+
+        #[cfg(feature = "v2")]
+        /// Session expiration in seconds from now
+        #[arg(long = "expire-in")]
+        expire_in: Option<u64>,
     },
     /// Resume pending payjoins (BIP77/v2 only)
     #[cfg(feature = "v2")]