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.

Author: benalleng

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

@@ -337,6 +337,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 +371,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,10 +700,52 @@ 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:?}"
+                );
+            }
         }
     }
 
+    // / Best-effort broadcast of a fallback transaction recovered from an expired
+    // / session's `ReplayError`. A sender always has a fallback (its original
+    // / PSBT); a receiver only has one if it had already received and validated
+    // / the sender's original proposal.
+    // fn broadcast_fallback_on_expiry<SessionState, SessionEvent>(
+    //     &self,
+    //     e: &payjoin::error::ReplayError<SessionState, SessionEvent>,
+    //     session_id: &SessionId,
+    //     role: &str,
+    // ) {
+    //     match e.fallback_tx() {
+    //         Some(tx) => self.broadcast_fallback_tx(tx, session_id, role),
+    //         None => println!(
+    //             "No fallback transaction available for expired {role} session {session_id}."
+    //         ),
+    //     }
+    // }
+
     async fn process_sender_session(
         &self,
         session: SendSession,
@@ -739,26 +793,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 +852,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/src/core/error.rs

@@ -51,7 +51,7 @@ impl<SessionState: Debug, SessionEvent: Debug> std::fmt::Display
                 Some(session) => write!(f, "Invalid event ({event:?}) for session ({session:?})",),
                 None => write!(f, "Invalid first event ({event:?}) for session",),
             },
-            Expired(time) => write!(f, "Session expired at {time:?}"),
+            Expired(time, _) => write!(f, "Session expired at {time:?}"),
             PersistenceFailure(e) => write!(f, "Persistence failure: {e}"),
         }
     }
@@ -73,7 +73,20 @@ impl<SessionState: Debug, SessionEvent: Debug> From<InternalReplayError<SessionS
 impl<SessionState, SessionEvent> ReplayError<SessionState, SessionEvent> {
     /// Returns `true` if the event log could not be replayed because the
     /// session has expired.
-    pub fn is_expired(&self) -> bool { matches!(self.0, InternalReplayError::Expired(_)) }
+    pub fn is_expired(&self) -> bool { matches!(self.0, InternalReplayError::Expired(..)) }
+
+    /// Returns the fallback transaction that should be broadcast when the
+    /// session has expired, if one is recoverable from the event log.
+    ///
+    /// Returns `None` for non-expired errors, or for an expired session that
+    /// never produced a fallback transaction (e.g. a receiver that expired
+    /// before receiving the sender's original proposal).
+    pub fn fallback_tx(&self) -> Option<&bitcoin::Transaction> {
+        match &self.0 {
+            InternalReplayError::Expired(_, tx) => tx.as_ref(),
+            _ => None,
+        }
+    }
 }
 
 #[cfg(feature = "v2")]
@@ -83,8 +96,8 @@ pub(crate) enum InternalReplayError<SessionState, SessionEvent> {
     NoEvents,
     /// Invalid initial event
     InvalidEvent(Box<SessionEvent>, Option<Box<SessionState>>),
-    /// Session is expired
-    Expired(crate::time::Time),
+    /// Session is expired. Carries the recoverable fallback transaction, if any.
+    Expired(crate::time::Time, Option<bitcoin::Transaction>),
     /// Application storage error
     PersistenceFailure(ImplementationError),
 }
@@ -96,8 +109,10 @@ mod tests {
 
     #[test]
     fn replay_error_is_expired() {
-        let expired: ReplayError<(), ()> = ReplayError(InternalReplayError::Expired(Time::now()));
+        let expired: ReplayError<(), ()> =
+            ReplayError(InternalReplayError::Expired(Time::now(), None));
         assert!(expired.is_expired());
+        assert!(expired.fallback_tx().is_none());
 
         let other: ReplayError<(), ()> = ReplayError(InternalReplayError::NoEvents);
         assert!(!other.is_expired());

payjoin/src/core/receive/v2/session.rs

@@ -33,7 +33,7 @@ fn construct_history(
     if !matches!(receiver, ReceiveSession::Closed(_)) {
         let ctx = history.session_context();
         if ctx.expiration.elapsed() {
-            return Err(InternalReplayError::Expired(ctx.expiration).into());
+            return Err(InternalReplayError::Expired(ctx.expiration, history.fallback_tx()).into());
         }
     }
     Ok(history)
@@ -395,7 +395,7 @@ mod tests {
             .expect("in memory persister save should not fail");
         let err = replay_event_log(&persister).expect_err("session should be expired");
         let expected_err: ReplayError<ReceiveSession, SessionEvent> =
-            InternalReplayError::Expired(expiration).into();
+            InternalReplayError::Expired(expiration, None).into();
         assert_eq!(err.to_string(), expected_err.to_string());
 
         let persister = InMemoryAsyncPersister::<SessionEvent>::default();
@@ -405,7 +405,7 @@ mod tests {
             .expect("in memory async persister save should not fail");
         let err = replay_event_log_async(&persister).await.expect_err("session should be expired");
         let expected_err: ReplayError<ReceiveSession, SessionEvent> =
-            InternalReplayError::Expired(expiration).into();
+            InternalReplayError::Expired(expiration, None).into();
         assert_eq!(err.to_string(), expected_err.to_string());
     }
 

payjoin/src/core/send/v2/session.rs

@@ -30,7 +30,11 @@ fn construct_history(
     if !matches!(sender, SendSession::Closed(_)) {
         let pj_param = history.pj_param();
         if pj_param.expiration().elapsed() {
-            return Err(InternalReplayError::Expired(pj_param.expiration()).into());
+            return Err(InternalReplayError::Expired(
+                pj_param.expiration(),
+                Some(history.fallback_tx()),
+            )
+            .into());
         }
     }
     Ok(history)