Skip to content

Return current state in transient errors #1724

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
spacebear21 merged 4 commits into from
Jul 10, 2026
Merged

Return current state in transient errors #1724

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
7f41f46
Return current state in transient errors
spacebear21 Jul 9, 2026
c7dc8d6
Expose transient vs fatal on PersistedError
spacebear21 Jul 9, 2026
7f7003d
Classify FFI persisted errors as transient/fatal
spacebear21 Jul 9, 2026
a7c000e
Handle transient errors in CLI send/receive loops
DanGould Jul 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
154 changes: 97 additions & 57 deletions payjoin-cli/src/app/v2/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,10 @@ const W_ID: usize = 36;
const W_ROLE: usize = 25;
const W_STATUS: usize = 15;

/// Delay before retrying a transiently failed state transition, so a
/// misbehaving directory or relay is not hammered in a tight loop.
const TRANSIENT_RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(5);

#[derive(Clone)]
pub(crate) struct App {
config: Config,
Expand Down Expand Up @@ -697,27 +701,37 @@ impl App {

async fn post_original_proposal(
&self,
sender: Sender<WithReplyKey>,
mut sender: Sender<WithReplyKey>,
persister: &SenderPersister,
) -> Result<()> {
let (response, ctx) =
self.post_via_relay(|relay| sender.create_v2_post_request(relay)).await?;
let sender = sender.process_response(&response.bytes().await?, ctx).save(persister)?;
println!("Posted Original PSBT...");
self.get_proposed_payjoin_psbt(sender, persister).await
loop {

@xstoicunicornx xstoicunicornx Jul 10, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Rather than introducing a loop couldn't we just call process_receiver_session with the returned transient current state?

let (response, ctx) =
self.post_via_relay(|relay| sender.create_v2_post_request(relay)).await?;
match sender.process_response(&response.bytes().await?, ctx).save(persister) {
Ok(sender) => {
println!("Posted Original PSBT...");
return self.get_proposed_payjoin_psbt(sender, persister).await;
}
Err(e) if e.is_transient() => {
tracing::debug!("Transient error posting original proposal, retrying: {e:?}");
sender = e.transient_state().expect("transient error carries current state");
tokio::time::sleep(TRANSIENT_RETRY_DELAY).await;
}
Err(e) => return Err(e.into()),
}
}
}

async fn get_proposed_payjoin_psbt(
&self,
sender: Sender<PollingForProposal>,
mut sender: Sender<PollingForProposal>,
persister: &SenderPersister,
) -> Result<()> {
let mut session = sender.clone();
// Long poll until we get a response
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);
self.post_via_relay(|relay| sender.create_poll_request(relay)).await?;
let res = sender.process_response(&response.bytes().await?, ctx).save(persister);
match res {
Ok(OptionalTransitionOutcome::Progress(psbt)) => {
println!("Proposal received. Processing...");
Expand All @@ -726,8 +740,12 @@ impl App {
}
Ok(OptionalTransitionOutcome::Stasis(current_state)) => {
println!("No response yet.");
session = current_state;
continue;
sender = current_state;
}
Err(e) if e.is_transient() => {
tracing::debug!("Transient error polling for proposal, retrying: {e:?}");
sender = e.transient_state().expect("transient error carries current state");
tokio::time::sleep(TRANSIENT_RETRY_DELAY).await;
}
Err(re) => {
println!("{re}");
Expand All @@ -740,10 +758,9 @@ impl App {

async fn long_poll_fallback(
&self,
session: Receiver<Initialized>,
mut session: Receiver<Initialized>,
persister: &ReceiverPersister,
) -> Result<Receiver<UncheckedOriginalPayload>> {
let mut session = session;
loop {
println!("Polling receive request...");
let (ohttp_response, context) =
Expand All @@ -758,7 +775,11 @@ impl App {
}
Ok(OptionalTransitionOutcome::Stasis(current_state)) => {
session = current_state;
continue;
}
Err(e) if e.is_transient() => {
tracing::debug!("Transient error polling for request, retrying: {e:?}");
session = e.transient_state().expect("transient error carries current state");
tokio::time::sleep(TRANSIENT_RETRY_DELAY).await;
}
Err(e) => return Err(e.into()),
}
Expand Down Expand Up @@ -946,24 +967,34 @@ impl App {

async fn send_payjoin_proposal(
&self,
proposal: Receiver<PayjoinProposal>,
mut proposal: Receiver<PayjoinProposal>,
persister: &ReceiverPersister,
) -> Result<()> {
let (res, ohttp_ctx) = self
.post_via_relay(|relay| {
proposal
.create_post_request(relay)
.map_err(|e| anyhow!("v2 req extraction failed {}", e))
})
.await?;
let payjoin_psbt = proposal.psbt().clone();
let session = proposal.process_response(&res.bytes().await?, ohttp_ctx).save(persister)?;
println!(
"Response successful. Watch mempool for successful Payjoin. TXID: {}",
payjoin_psbt.extract_tx_unchecked_fee_rate().compute_txid()
);

return self.monitor_payjoin_proposal(session, persister).await;
loop {
let (res, ohttp_ctx) = self
.post_via_relay(|relay| {
proposal
.create_post_request(relay)
.map_err(|e| anyhow!("v2 req extraction failed {}", e))
})
.await?;
let payjoin_psbt = proposal.psbt().clone();
match proposal.process_response(&res.bytes().await?, ohttp_ctx).save(persister) {
Ok(session) => {
println!(
"Response successful. Watch mempool for successful Payjoin. TXID: {}",
payjoin_psbt.extract_tx_unchecked_fee_rate().compute_txid()
);
return self.monitor_payjoin_proposal(session, persister).await;
}
Err(e) if e.is_transient() => {
tracing::debug!("Transient error sending payjoin proposal, retrying: {e:?}");
proposal = e.transient_state().expect("transient error carries current state");
tokio::time::sleep(TRANSIENT_RETRY_DELAY).await;
}
Err(e) => return Err(e.into()),
}
}
}

async fn monitor_payjoin_proposal(
Expand Down Expand Up @@ -1014,38 +1045,47 @@ impl App {
/// Handle error by attempting to send an error response over the directory
async fn handle_error(
&self,
session: Receiver<HasReplyableError>,
mut session: Receiver<HasReplyableError>,
persister: &ReceiverPersister,
) -> Result<()> {
let (err_response, err_ctx) = self
.post_via_relay(|relay| {
session
.create_error_request(relay)
.map_err(|e| anyhow!("Failed to post error request: {}", e))
})
.await?;

let err_bytes = match err_response.bytes().await {
Ok(bytes) => bytes,
Err(e) => return Err(anyhow!("Failed to get error response bytes: {}", e)),
};
loop {
let (err_response, err_ctx) = self
.post_via_relay(|relay| {
session
.create_error_request(relay)
.map_err(|e| anyhow!("Failed to post error request: {}", e))
})
.await?;

let err_bytes = match err_response.bytes().await {
Ok(bytes) => bytes,
Err(e) => return Err(anyhow!("Failed to get error response bytes: {}", e)),
};

if let Err(e) = session.process_error_response(&err_bytes, err_ctx).save(persister) {
if let Some(api_err) = e.api_error_ref() {
tracing::warn!("Failed to confirm error response delivery: {api_err}");
}
match e.error_state() {
Some(_) => {
let id = persister.session_id();
println!(
"Session {id} failed. Run `payjoin-cli cancel {id}` to cancel and broadcast the fallback transaction."
);
match session.process_error_response(&err_bytes, err_ctx).save(persister) {
Ok(_) => return Ok(()),
Err(e) if e.is_transient() => {
tracing::debug!("Transient error posting error response, retrying: {e:?}");
session = e.transient_state().expect("transient error carries current state");
tokio::time::sleep(TRANSIENT_RETRY_DELAY).await;
}
Err(e) => {
if let Some(api_err) = e.api_error_ref() {
tracing::warn!("Failed to confirm error response delivery: {api_err}");
}
match e.error_state() {
Some(_) => {
let id = persister.session_id();
println!(
"Session {id} failed. Run `payjoin-cli cancel {id}` to cancel and broadcast the fallback transaction."
);
return Ok(());
}
None => return Err(anyhow!("Failed to process error response")),
}
}
None => return Err(anyhow!("Failed to process error response")),
}
}

Ok(())
}

async fn post_request(&self, req: payjoin::Request) -> Result<reqwest::Response> {
Expand Down
71 changes: 64 additions & 7 deletions payjoin-ffi/src/receive/error.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,9 +59,13 @@ impl ReceiverCreateRequestError {
#[derive(Debug, thiserror::Error, uniffi::Error)]
#[error(transparent)]
pub enum ReceiverPersistedError {
/// rust-payjoin receiver error
/// A transient error: nothing was persisted, and the failed transition
/// can be retried by calling it again on the same receiver object
#[error(transparent)]
Receiver(ReceiverError),
Transient(ReceiverError),
/// A fatal error: the session is closed or has moved to an error state
#[error(transparent)]
Fatal(ReceiverError),
/// Storage error that could occur at application storage layer
#[error(transparent)]
Storage(Arc<ImplementationError>),
Expand All @@ -76,23 +80,30 @@ macro_rules! impl_persisted_error_from {
$api_error_ty:ty,
$receiver_arm:expr
) => {
impl<S, E> From<payjoin::persist::PersistedError<$api_error_ty, S, E>>
impl<S, E, C> From<payjoin::persist::PersistedError<$api_error_ty, S, E, C>>
for ReceiverPersistedError
where
S: std::error::Error + Send + Sync + 'static,
E: std::fmt::Debug,
C: std::fmt::Debug,
{
fn from(err: payjoin::persist::PersistedError<$api_error_ty, S, E>) -> Self {
fn from(err: payjoin::persist::PersistedError<$api_error_ty, S, E, C>) -> Self {
if err.storage_error_ref().is_some() {
if let Some(storage_err) = err.storage_error() {
return ReceiverPersistedError::from(ImplementationError::new(storage_err));
}
return ReceiverPersistedError::Receiver(ReceiverError::Unexpected);
return ReceiverPersistedError::Fatal(ReceiverError::Unexpected);
}
let is_transient = err.is_transient();
if let Some(api_err) = err.api_error() {
return ReceiverPersistedError::Receiver($receiver_arm(api_err));
let receiver_err = $receiver_arm(api_err);
return if is_transient {
ReceiverPersistedError::Transient(receiver_err)
} else {
ReceiverPersistedError::Fatal(receiver_err)
};
}
ReceiverPersistedError::Receiver(ReceiverError::Unexpected)
ReceiverPersistedError::Fatal(ReceiverError::Unexpected)
}
}
};
Expand Down Expand Up @@ -320,4 +331,50 @@ mod tests {
receiver.create_poll_request(EXAMPLE_URL).map(|_| ()).expect_err("session is expired");
assert!(ReceiverCreateRequestError::from(expired).is_expired());
}

#[cfg(feature = "_test-utils")]
#[test]
fn persisted_error_classifies_transient_and_fatal() {
use std::str::FromStr;

use payjoin::bitcoin::Address;
use payjoin::directory::ENCAPSULATED_MESSAGE_BYTES;
use payjoin::persist::InMemoryPersister;
use payjoin::receive::v2::{ReceiverBuilder, SessionEvent};
use payjoin::OhttpKeys;
use payjoin_test_utils::EXAMPLE_URL;

let address = Address::from_str("tb1q6d3a2w975yny0asuvd9a67ner4nks58ff0q8g4")
.expect("valid address")
.assume_checked();
let ohttp_keys = OhttpKeys::decode(&payjoin_test_utils::ohttp_key_config_bytes())
.expect("valid ohttp keys");
let persister = InMemoryPersister::<SessionEvent>::default();
let receiver = ReceiverBuilder::new(address, EXAMPLE_URL, ohttp_keys)
.expect("valid builder")
.build()
.save(&persister)
.expect("in-memory persister is infallible");

// An undersized directory response fails the size check, which is
// retryable, so the binding error classifies it as transient.
let (_req, ctx) = receiver.create_poll_request(EXAMPLE_URL).expect("valid poll request");
let err = receiver
.clone()
.process_response(&[0u8; 1], ctx)
.save(&persister)
.expect_err("undersized response should fail");
assert!(err.is_transient());
assert!(matches!(ReceiverPersistedError::from(err), ReceiverPersistedError::Transient(_)));

// A right-sized garbage body fails OHTTP decapsulation, which is
// fatal, so the binding error classifies it as fatal.
let (_req, ctx) = receiver.create_poll_request(EXAMPLE_URL).expect("valid poll request");
let err = receiver
.process_response(&[0u8; ENCAPSULATED_MESSAGE_BYTES], ctx)
.save(&persister)
.expect_err("garbage response should fail");
assert!(err.is_fatal());
assert!(matches!(ReceiverPersistedError::from(err), ReceiverPersistedError::Fatal(_)));
}
}
9 changes: 9 additions & 0 deletions payjoin-ffi/src/receive/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -748,6 +748,7 @@ pub struct UncheckedOriginalPayloadTransition(
payjoin::receive::v2::Receiver<payjoin::receive::v2::MaybeInputsOwned>,
payjoin::receive::Error,
payjoin::receive::v2::Receiver<payjoin::receive::v2::HasReplyableError>,
payjoin::receive::v2::Receiver<payjoin::receive::v2::UncheckedOriginalPayload>,
>,
>,
>,
Expand Down Expand Up @@ -829,6 +830,7 @@ pub struct MaybeInputsOwnedTransition(
payjoin::receive::v2::Receiver<payjoin::receive::v2::MaybeInputsSeen>,
payjoin::receive::Error,
payjoin::receive::v2::Receiver<payjoin::receive::v2::HasReplyableError>,
payjoin::receive::v2::Receiver<payjoin::receive::v2::MaybeInputsOwned>,
>,
>,
>,
Expand Down Expand Up @@ -884,6 +886,7 @@ pub struct MaybeInputsSeenTransition(
payjoin::receive::v2::Receiver<payjoin::receive::v2::OutputsUnknown>,
payjoin::receive::Error,
payjoin::receive::v2::Receiver<payjoin::receive::v2::HasReplyableError>,
payjoin::receive::v2::Receiver<payjoin::receive::v2::MaybeInputsSeen>,
>,
>,
>,
Expand Down Expand Up @@ -937,6 +940,7 @@ pub struct OutputsUnknownTransition(
payjoin::receive::v2::Receiver<payjoin::receive::v2::WantsOutputs>,
payjoin::receive::Error,
payjoin::receive::v2::Receiver<payjoin::receive::v2::HasReplyableError>,
payjoin::receive::v2::Receiver<payjoin::receive::v2::OutputsUnknown>,
>,
>,
>,
Expand Down Expand Up @@ -1137,6 +1141,8 @@ pub struct WantsFeeRangeTransition(
payjoin::receive::v2::SessionEvent,
payjoin::receive::v2::Receiver<payjoin::receive::v2::ProvisionalProposal>,
payjoin::receive::ProtocolError,
(),
payjoin::receive::v2::Receiver<payjoin::receive::v2::WantsFeeRange>,
>,
>,
>,
Expand Down Expand Up @@ -1217,6 +1223,7 @@ pub struct ProvisionalProposalTransition(
payjoin::receive::v2::SessionEvent,
payjoin::receive::v2::Receiver<payjoin::receive::v2::PayjoinProposal>,
payjoin::ImplementationError,
payjoin::receive::v2::Receiver<payjoin::receive::v2::ProvisionalProposal>,
>,
>,
>,
Expand Down Expand Up @@ -1279,6 +1286,7 @@ pub struct PayjoinProposalTransition(
payjoin::receive::v2::Receiver<payjoin::receive::v2::Monitor>,
payjoin::receive::ProtocolError,
payjoin::receive::v2::Receiver<payjoin::receive::v2::PendingFallback>,
payjoin::receive::v2::Receiver<payjoin::receive::v2::PayjoinProposal>,
>,
>,
>,
Expand Down Expand Up @@ -1368,6 +1376,7 @@ pub struct HasReplyableErrorTransition(
payjoin::receive::v2::SessionEvent,
payjoin::receive::v2::Receiver<payjoin::receive::v2::PendingFallback>,
payjoin::receive::ProtocolError,
payjoin::receive::v2::Receiver<payjoin::receive::v2::HasReplyableError>,
>,
>,
>,
Expand Down
Loading
Loading