Introduce HasReplyableError typestate

This enforces proper handling within the receiver state machine, such
that they must attempt replying to the sender with an error response
before the session can be considered closed.

Co-authored-by: DanGould <d@ngould.dev>

Author: spacebear21

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

@@ -6,9 +6,9 @@ use payjoin::bitcoin::consensus::encode::serialize_hex;
 use payjoin::bitcoin::{Amount, FeeRate};
 use payjoin::persist::OptionalTransitionOutcome;
 use payjoin::receive::v2::{
-    process_err_res, replay_event_log as replay_receiver_event_log, Initialized, MaybeInputsOwned,
-    MaybeInputsSeen, OutputsUnknown, PayjoinProposal, ProvisionalProposal, ReceiveSession,
-    Receiver, ReceiverBuilder, SessionHistory, UncheckedOriginalPayload, WantsFeeRange,
+    replay_event_log as replay_receiver_event_log, HasReplyableError, Initialized,
+    MaybeInputsOwned, MaybeInputsSeen, OutputsUnknown, PayjoinProposal, ProvisionalProposal,
+    ReceiveSession, Receiver, ReceiverBuilder, UncheckedOriginalPayload, WantsFeeRange,
     WantsInputs, WantsOutputs,
 };
 use payjoin::send::v2::{
@@ -70,6 +70,7 @@ impl StatusText for ReceiveSession {
             | ReceiveSession::WantsFeeRange(_)
             | ReceiveSession::ProvisionalProposal(_) => "Processing original proposal",
             ReceiveSession::PayjoinProposal(_) => "Payjoin proposal sent",
+            ReceiveSession::HasReplyableError(_) => "Session failure",
             ReceiveSession::TerminalFailure => "Session failure",
         }
     }
@@ -521,22 +522,13 @@ impl App {
                     self.finalize_proposal(proposal, persister).await,
                 ReceiveSession::PayjoinProposal(proposal) =>
                     self.send_payjoin_proposal(proposal, persister).await,
+                ReceiveSession::HasReplyableError(error) =>
+                    self.handle_error(error, persister).await,
                 ReceiveSession::TerminalFailure =>
                     return Err(anyhow!("Terminal receiver session")),
             }
         };
-
-        match res {
-            Ok(_) => Ok(()),
-            Err(e) => {
-                let (_, session_history) = replay_receiver_event_log(persister)?;
-                let pj_uri = session_history.pj_uri().extras.endpoint().clone();
-                let ohttp_relay = self.unwrap_relay_or_else_fetch(Some(pj_uri)).await?;
-                self.handle_recoverable_error(&ohttp_relay, &session_history).await?;
-
-                Err(e)
-            }
-        }
+        res
     }
 
     #[allow(clippy::incompatible_msrv)]
@@ -702,20 +694,14 @@ impl App {
         Ok(ohttp_relay)
     }
 
-    /// Handle request error by sending an error response over the directory
-    async fn handle_recoverable_error(
+    /// Handle error by attempting to send an error response over the directory
+    async fn handle_error(
         &self,
-        ohttp_relay: &payjoin::Url,
-        session_history: &SessionHistory,
+        session: Receiver<HasReplyableError>,
+        persister: &ReceiverPersister,
     ) -> Result<()> {
-        let e = match session_history.terminal_error() {
-            Some(e) => e,
-            _ => return Ok(()),
-        };
-        let (err_req, err_ctx) = session_history
-            .extract_err_req(ohttp_relay.as_str())?
-            .expect("If JsonReply is Some, then err_req and err_ctx should be Some");
-        let to_return = anyhow!("Replied with error: {}", e.to_json());
+        let (err_req, err_ctx) =
+            session.create_error_request(self.unwrap_relay_or_else_fetch(None).await?.as_str())?;
 
         let err_response = match self.post_request(err_req).await {
             Ok(response) => response,
@@ -727,11 +713,11 @@ impl App {
             Err(e) => return Err(anyhow!("Failed to get error response bytes: {}", e)),
         };
 
-        if let Err(e) = process_err_res(&err_bytes, err_ctx) {
+        if let Err(e) = session.process_error_response(&err_bytes, err_ctx).save(persister) {
             return Err(anyhow!("Failed to process error response: {}", e));
         }
 
-        Err(to_return)
+        Ok(())
     }
 
     async fn post_request(&self, req: payjoin::Request) -> Result<reqwest::Response> {

payjoin-ffi/src/receive/mod.rs

@@ -78,6 +78,7 @@ pub enum ReceiveSession {
     WantsFeeRange { inner: Arc<WantsFeeRange> },
     ProvisionalProposal { inner: Arc<ProvisionalProposal> },
     PayjoinProposal { inner: Arc<PayjoinProposal> },
+    HasReplyableError { inner: Arc<HasReplyableError> },
     TerminalFailure,
 }
 
@@ -105,6 +106,8 @@ impl From<payjoin::receive::v2::ReceiveSession> for ReceiveSession {
                 Self::ProvisionalProposal { inner: Arc::new(inner.into()) },
             ReceiveSession::PayjoinProposal(inner) =>
                 Self::PayjoinProposal { inner: Arc::new(inner.into()) },
+            ReceiveSession::HasReplyableError(inner) =>
+                Self::HasReplyableError { inner: Arc::new(inner.into()) },
             ReceiveSession::TerminalFailure => Self::TerminalFailure,
         }
     }
@@ -1003,6 +1006,80 @@ impl PayjoinProposal {
     }
 }
 
+#[derive(Clone, uniffi::Object)]
+pub struct HasReplyableError(
+    pub payjoin::receive::v2::Receiver<payjoin::receive::v2::HasReplyableError>,
+);
+
+impl From<HasReplyableError>
+    for payjoin::receive::v2::Receiver<payjoin::receive::v2::HasReplyableError>
+{
+    fn from(value: HasReplyableError) -> Self { value.0 }
+}
+
+impl From<payjoin::receive::v2::Receiver<payjoin::receive::v2::HasReplyableError>>
+    for HasReplyableError
+{
+    fn from(
+        value: payjoin::receive::v2::Receiver<payjoin::receive::v2::HasReplyableError>,
+    ) -> Self {
+        Self(value)
+    }
+}
+
+#[derive(uniffi::Object)]
+pub struct HasReplyableErrorTransition(
+    Arc<
+        RwLock<
+            Option<
+                payjoin::persist::MaybeSuccessTransition<
+                    payjoin::receive::v2::SessionEvent,
+                    (),
+                    payjoin::receive::Error,
+                >,
+            >,
+        >,
+    >,
+);
+
+#[uniffi::export]
+impl HasReplyableErrorTransition {
+    pub fn save(
+        &self,
+        persister: Arc<dyn JsonReceiverSessionPersister>,
+    ) -> Result<(), ReceiverPersistedError> {
+        let adapter = CallbackPersisterAdapter::new(persister);
+        let mut inner = self.0.write().expect("Lock should not be poisoned");
+
+        let value = inner.take().expect("Already saved or moved");
+
+        value.save(&adapter).map_err(ReceiverPersistedError::from)?;
+        Ok(())
+    }
+}
+
+#[uniffi::export]
+impl HasReplyableError {
+    pub fn create_error_request(
+        &self,
+        ohttp_relay: String,
+    ) -> Result<RequestResponse, SessionError> {
+        self.0.clone().create_error_request(ohttp_relay).map_err(Into::into).map(|(req, ctx)| {
+            RequestResponse { request: req.into(), client_response: Arc::new(ctx.into()) }
+        })
+    }
+
+    pub fn process_error_response(
+        &self,
+        body: &[u8],
+        ohttp_context: &ClientResponse,
+    ) -> PayjoinProposalTransition {
+        PayjoinProposalTransition(Arc::new(RwLock::new(Some(
+            self.0.clone().process_error_response(body, ohttp_context.into()),
+        ))))
+    }
+}
+
 /// Session persister that should save and load events as JSON strings.
 #[uniffi::export(with_foreign)]
 pub trait JsonReceiverSessionPersister: Send + Sync {

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

@@ -138,6 +138,7 @@ pub enum ReceiveSession {
     WantsFeeRange(Receiver<WantsFeeRange>),
     ProvisionalProposal(Receiver<ProvisionalProposal>),
     PayjoinProposal(Receiver<PayjoinProposal>),
+    HasReplyableError(Receiver<HasReplyableError>),
     TerminalFailure,
 }
 
@@ -188,7 +189,25 @@ impl ReceiveSession {
                 SessionEvent::FinalizedProposal(payjoin_proposal),
             ) => Ok(state.apply_finalized_proposal(payjoin_proposal)),
 
-            (_, SessionEvent::GotReplyableError(_)) => Ok(ReceiveSession::TerminalFailure),
+            (session, SessionEvent::GotReplyableError(error)) =>
+                Ok(ReceiveSession::HasReplyableError(Receiver {
+                    state: HasReplyableError { error_reply: error.clone() },
+                    session_context: match session {
+                        ReceiveSession::Initialized(r) => r.session_context,
+                        ReceiveSession::UncheckedOriginalPayload(r) => r.session_context,
+                        ReceiveSession::MaybeInputsOwned(r) => r.session_context,
+                        ReceiveSession::MaybeInputsSeen(r) => r.session_context,
+                        ReceiveSession::OutputsUnknown(r) => r.session_context,
+                        ReceiveSession::WantsOutputs(r) => r.session_context,
+                        ReceiveSession::WantsInputs(r) => r.session_context,
+                        ReceiveSession::WantsFeeRange(r) => r.session_context,
+                        ReceiveSession::ProvisionalProposal(r) => r.session_context,
+                        ReceiveSession::PayjoinProposal(r) => r.session_context,
+                        ReceiveSession::HasReplyableError(r) => r.session_context,
+                        ReceiveSession::TerminalFailure =>
+                            return Ok(ReceiveSession::TerminalFailure),
+                    },
+                })),
 
             (current_state, SessionEvent::Closed(_)) => Ok(current_state),
 
@@ -214,6 +233,7 @@ mod sealed {
     impl State for super::WantsFeeRange {}
     impl State for super::ProvisionalProposal {}
     impl State for super::PayjoinProposal {}
+    impl State for super::HasReplyableError {}
 }
 
 /// Sealed trait for V2 receive session states.
@@ -1083,6 +1103,42 @@ impl Receiver<PayjoinProposal> {
     }
 }
 
+#[derive(Debug, Clone, PartialEq)]
+pub struct HasReplyableError {
+    error_reply: JsonReply,
+}
+
+impl Receiver<HasReplyableError> {
+    pub fn create_error_request(
+        &self,
+        ohttp_relay: impl IntoUrl,
+    ) -> Result<(Request, ohttp::ClientResponse), SessionError> {
+        extract_err_req(&self.error_reply, ohttp_relay, &self.session_context)
+    }
+
+    pub fn process_error_response(
+        &self,
+        res: &[u8],
+        ohttp_context: ohttp::ClientResponse,
+    ) -> MaybeSuccessTransition<SessionEvent, (), ProtocolError> {
+        match process_post_res(res, ohttp_context) {
+            Ok(_) =>
+                MaybeSuccessTransition::success(SessionEvent::Closed(SessionOutcome::Failure), ()),
+            Err(e) =>
+                if e.is_fatal() {
+                    MaybeSuccessTransition::fatal(
+                        SessionEvent::Closed(SessionOutcome::Failure),
+                        ProtocolError::V2(InternalSessionError::DirectoryResponse(e).into()),
+                    )
+                } else {
+                    MaybeSuccessTransition::transient(ProtocolError::V2(
+                        InternalSessionError::DirectoryResponse(e).into(),
+                    ))
+                },
+        }
+    }
+}
+
 /// Derive a mailbox endpoint on a directory given a [`ShortId`].
 /// It consists of a directory URL and the session ShortID in the path.
 fn mailbox_endpoint(directory: &Url, id: &ShortId) -> Url {