Skip to content

Commit 436749f

Browse files
spacebear21spacebear21
authored
Return current state in transient errors (#1724)
2 parents 6cd4e5c + a7c000e commit 436749f

8 files changed

Lines changed: 742 additions & 294 deletions

File tree

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

Lines changed: 97 additions & 57 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,10 @@ const W_ID: usize = 36;
3535
const W_ROLE: usize = 25;
3636
const W_STATUS: usize = 15;
3737

38+
/// Delay before retrying a transiently failed state transition, so a
39+
/// misbehaving directory or relay is not hammered in a tight loop.
40+
const TRANSIENT_RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(5);
41+
3842
#[derive(Clone)]
3943
pub(crate) struct App {
4044
config: Config,
@@ -697,27 +701,37 @@ impl App {
697701

698702
async fn post_original_proposal(
699703
&self,
700-
sender: Sender<WithReplyKey>,
704+
mut sender: Sender<WithReplyKey>,
701705
persister: &SenderPersister,
702706
) -> Result<()> {
703-
let (response, ctx) =
704-
self.post_via_relay(|relay| sender.create_v2_post_request(relay)).await?;
705-
let sender = sender.process_response(&response.bytes().await?, ctx).save(persister)?;
706-
println!("Posted Original PSBT...");
707-
self.get_proposed_payjoin_psbt(sender, persister).await
707+
loop {
708+
let (response, ctx) =
709+
self.post_via_relay(|relay| sender.create_v2_post_request(relay)).await?;
710+
match sender.process_response(&response.bytes().await?, ctx).save(persister) {
711+
Ok(sender) => {
712+
println!("Posted Original PSBT...");
713+
return self.get_proposed_payjoin_psbt(sender, persister).await;
714+
}
715+
Err(e) if e.is_transient() => {
716+
tracing::debug!("Transient error posting original proposal, retrying: {e:?}");
717+
sender = e.transient_state().expect("transient error carries current state");
718+
tokio::time::sleep(TRANSIENT_RETRY_DELAY).await;
719+
}
720+
Err(e) => return Err(e.into()),
721+
}
722+
}
708723
}
709724

710725
async fn get_proposed_payjoin_psbt(
711726
&self,
712-
sender: Sender<PollingForProposal>,
727+
mut sender: Sender<PollingForProposal>,
713728
persister: &SenderPersister,
714729
) -> Result<()> {
715-
let mut session = sender.clone();
716730
// Long poll until we get a response
717731
loop {
718732
let (response, ctx) =
719-
self.post_via_relay(|relay| session.create_poll_request(relay)).await?;
720-
let res = session.process_response(&response.bytes().await?, ctx).save(persister);
733+
self.post_via_relay(|relay| sender.create_poll_request(relay)).await?;
734+
let res = sender.process_response(&response.bytes().await?, ctx).save(persister);
721735
match res {
722736
Ok(OptionalTransitionOutcome::Progress(psbt)) => {
723737
println!("Proposal received. Processing...");
@@ -726,8 +740,12 @@ impl App {
726740
}
727741
Ok(OptionalTransitionOutcome::Stasis(current_state)) => {
728742
println!("No response yet.");
729-
session = current_state;
730-
continue;
743+
sender = current_state;
744+
}
745+
Err(e) if e.is_transient() => {
746+
tracing::debug!("Transient error polling for proposal, retrying: {e:?}");
747+
sender = e.transient_state().expect("transient error carries current state");
748+
tokio::time::sleep(TRANSIENT_RETRY_DELAY).await;
731749
}
732750
Err(re) => {
733751
println!("{re}");
@@ -740,10 +758,9 @@ impl App {
740758

741759
async fn long_poll_fallback(
742760
&self,
743-
session: Receiver<Initialized>,
761+
mut session: Receiver<Initialized>,
744762
persister: &ReceiverPersister,
745763
) -> Result<Receiver<UncheckedOriginalPayload>> {
746-
let mut session = session;
747764
loop {
748765
println!("Polling receive request...");
749766
let (ohttp_response, context) =
@@ -758,7 +775,11 @@ impl App {
758775
}
759776
Ok(OptionalTransitionOutcome::Stasis(current_state)) => {
760777
session = current_state;
761-
continue;
778+
}
779+
Err(e) if e.is_transient() => {
780+
tracing::debug!("Transient error polling for request, retrying: {e:?}");
781+
session = e.transient_state().expect("transient error carries current state");
782+
tokio::time::sleep(TRANSIENT_RETRY_DELAY).await;
762783
}
763784
Err(e) => return Err(e.into()),
764785
}
@@ -946,24 +967,34 @@ impl App {
946967

947968
async fn send_payjoin_proposal(
948969
&self,
949-
proposal: Receiver<PayjoinProposal>,
970+
mut proposal: Receiver<PayjoinProposal>,
950971
persister: &ReceiverPersister,
951972
) -> Result<()> {
952-
let (res, ohttp_ctx) = self
953-
.post_via_relay(|relay| {
954-
proposal
955-
.create_post_request(relay)
956-
.map_err(|e| anyhow!("v2 req extraction failed {}", e))
957-
})
958-
.await?;
959-
let payjoin_psbt = proposal.psbt().clone();
960-
let session = proposal.process_response(&res.bytes().await?, ohttp_ctx).save(persister)?;
961-
println!(
962-
"Response successful. Watch mempool for successful Payjoin. TXID: {}",
963-
payjoin_psbt.extract_tx_unchecked_fee_rate().compute_txid()
964-
);
965-
966-
return self.monitor_payjoin_proposal(session, persister).await;
973+
loop {
974+
let (res, ohttp_ctx) = self
975+
.post_via_relay(|relay| {
976+
proposal
977+
.create_post_request(relay)
978+
.map_err(|e| anyhow!("v2 req extraction failed {}", e))
979+
})
980+
.await?;
981+
let payjoin_psbt = proposal.psbt().clone();
982+
match proposal.process_response(&res.bytes().await?, ohttp_ctx).save(persister) {
983+
Ok(session) => {
984+
println!(
985+
"Response successful. Watch mempool for successful Payjoin. TXID: {}",
986+
payjoin_psbt.extract_tx_unchecked_fee_rate().compute_txid()
987+
);
988+
return self.monitor_payjoin_proposal(session, persister).await;
989+
}
990+
Err(e) if e.is_transient() => {
991+
tracing::debug!("Transient error sending payjoin proposal, retrying: {e:?}");
992+
proposal = e.transient_state().expect("transient error carries current state");
993+
tokio::time::sleep(TRANSIENT_RETRY_DELAY).await;
994+
}
995+
Err(e) => return Err(e.into()),
996+
}
997+
}
967998
}
968999

9691000
async fn monitor_payjoin_proposal(
@@ -1014,38 +1045,47 @@ impl App {
10141045
/// Handle error by attempting to send an error response over the directory
10151046
async fn handle_error(
10161047
&self,
1017-
session: Receiver<HasReplyableError>,
1048+
mut session: Receiver<HasReplyableError>,
10181049
persister: &ReceiverPersister,
10191050
) -> Result<()> {
1020-
let (err_response, err_ctx) = self
1021-
.post_via_relay(|relay| {
1022-
session
1023-
.create_error_request(relay)
1024-
.map_err(|e| anyhow!("Failed to post error request: {}", e))
1025-
})
1026-
.await?;
1027-
1028-
let err_bytes = match err_response.bytes().await {
1029-
Ok(bytes) => bytes,
1030-
Err(e) => return Err(anyhow!("Failed to get error response bytes: {}", e)),
1031-
};
1051+
loop {
1052+
let (err_response, err_ctx) = self
1053+
.post_via_relay(|relay| {
1054+
session
1055+
.create_error_request(relay)
1056+
.map_err(|e| anyhow!("Failed to post error request: {}", e))
1057+
})
1058+
.await?;
1059+
1060+
let err_bytes = match err_response.bytes().await {
1061+
Ok(bytes) => bytes,
1062+
Err(e) => return Err(anyhow!("Failed to get error response bytes: {}", e)),
1063+
};
10321064

1033-
if let Err(e) = session.process_error_response(&err_bytes, err_ctx).save(persister) {
1034-
if let Some(api_err) = e.api_error_ref() {
1035-
tracing::warn!("Failed to confirm error response delivery: {api_err}");
1036-
}
1037-
match e.error_state() {
1038-
Some(_) => {
1039-
let id = persister.session_id();
1040-
println!(
1041-
"Session {id} failed. Run `payjoin-cli cancel {id}` to cancel and broadcast the fallback transaction."
1042-
);
1065+
match session.process_error_response(&err_bytes, err_ctx).save(persister) {
1066+
Ok(_) => return Ok(()),
1067+
Err(e) if e.is_transient() => {
1068+
tracing::debug!("Transient error posting error response, retrying: {e:?}");
1069+
session = e.transient_state().expect("transient error carries current state");
1070+
tokio::time::sleep(TRANSIENT_RETRY_DELAY).await;
1071+
}
1072+
Err(e) => {
1073+
if let Some(api_err) = e.api_error_ref() {
1074+
tracing::warn!("Failed to confirm error response delivery: {api_err}");
1075+
}
1076+
match e.error_state() {
1077+
Some(_) => {
1078+
let id = persister.session_id();
1079+
println!(
1080+
"Session {id} failed. Run `payjoin-cli cancel {id}` to cancel and broadcast the fallback transaction."
1081+
);
1082+
return Ok(());
1083+
}
1084+
None => return Err(anyhow!("Failed to process error response")),
1085+
}
10431086
}
1044-
None => return Err(anyhow!("Failed to process error response")),
10451087
}
10461088
}
1047-
1048-
Ok(())
10491089
}
10501090

10511091
async fn post_request(&self, req: payjoin::Request) -> Result<reqwest::Response> {

‎payjoin-ffi/src/receive/error.rs‎

Lines changed: 64 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -59,9 +59,13 @@ impl ReceiverCreateRequestError {
5959
#[derive(Debug, thiserror::Error, uniffi::Error)]
6060
#[error(transparent)]
6161
pub enum ReceiverPersistedError {
62-
/// rust-payjoin receiver error
62+
/// A transient error: nothing was persisted, and the failed transition
63+
/// can be retried by calling it again on the same receiver object
6364
#[error(transparent)]
64-
Receiver(ReceiverError),
65+
Transient(ReceiverError),
66+
/// A fatal error: the session is closed or has moved to an error state
67+
#[error(transparent)]
68+
Fatal(ReceiverError),
6569
/// Storage error that could occur at application storage layer
6670
#[error(transparent)]
6771
Storage(Arc<ImplementationError>),
@@ -76,23 +80,30 @@ macro_rules! impl_persisted_error_from {
7680
$api_error_ty:ty,
7781
$receiver_arm:expr
7882
) => {
79-
impl<S, E> From<payjoin::persist::PersistedError<$api_error_ty, S, E>>
83+
impl<S, E, C> From<payjoin::persist::PersistedError<$api_error_ty, S, E, C>>
8084
for ReceiverPersistedError
8185
where
8286
S: std::error::Error + Send + Sync + 'static,
8387
E: std::fmt::Debug,
88+
C: std::fmt::Debug,
8489
{
85-
fn from(err: payjoin::persist::PersistedError<$api_error_ty, S, E>) -> Self {
90+
fn from(err: payjoin::persist::PersistedError<$api_error_ty, S, E, C>) -> Self {
8691
if err.storage_error_ref().is_some() {
8792
if let Some(storage_err) = err.storage_error() {
8893
return ReceiverPersistedError::from(ImplementationError::new(storage_err));
8994
}
90-
return ReceiverPersistedError::Receiver(ReceiverError::Unexpected);
95+
return ReceiverPersistedError::Fatal(ReceiverError::Unexpected);
9196
}
97+
let is_transient = err.is_transient();
9298
if let Some(api_err) = err.api_error() {
93-
return ReceiverPersistedError::Receiver($receiver_arm(api_err));
99+
let receiver_err = $receiver_arm(api_err);
100+
return if is_transient {
101+
ReceiverPersistedError::Transient(receiver_err)
102+
} else {
103+
ReceiverPersistedError::Fatal(receiver_err)
104+
};
94105
}
95-
ReceiverPersistedError::Receiver(ReceiverError::Unexpected)
106+
ReceiverPersistedError::Fatal(ReceiverError::Unexpected)
96107
}
97108
}
98109
};
@@ -320,4 +331,50 @@ mod tests {
320331
receiver.create_poll_request(EXAMPLE_URL).map(|_| ()).expect_err("session is expired");
321332
assert!(ReceiverCreateRequestError::from(expired).is_expired());
322333
}
334+
335+
#[cfg(feature = "_test-utils")]
336+
#[test]
337+
fn persisted_error_classifies_transient_and_fatal() {
338+
use std::str::FromStr;
339+
340+
use payjoin::bitcoin::Address;
341+
use payjoin::directory::ENCAPSULATED_MESSAGE_BYTES;
342+
use payjoin::persist::InMemoryPersister;
343+
use payjoin::receive::v2::{ReceiverBuilder, SessionEvent};
344+
use payjoin::OhttpKeys;
345+
use payjoin_test_utils::EXAMPLE_URL;
346+
347+
let address = Address::from_str("tb1q6d3a2w975yny0asuvd9a67ner4nks58ff0q8g4")
348+
.expect("valid address")
349+
.assume_checked();
350+
let ohttp_keys = OhttpKeys::decode(&payjoin_test_utils::ohttp_key_config_bytes())
351+
.expect("valid ohttp keys");
352+
let persister = InMemoryPersister::<SessionEvent>::default();
353+
let receiver = ReceiverBuilder::new(address, EXAMPLE_URL, ohttp_keys)
354+
.expect("valid builder")
355+
.build()
356+
.save(&persister)
357+
.expect("in-memory persister is infallible");
358+
359+
// An undersized directory response fails the size check, which is
360+
// retryable, so the binding error classifies it as transient.
361+
let (_req, ctx) = receiver.create_poll_request(EXAMPLE_URL).expect("valid poll request");
362+
let err = receiver
363+
.clone()
364+
.process_response(&[0u8; 1], ctx)
365+
.save(&persister)
366+
.expect_err("undersized response should fail");
367+
assert!(err.is_transient());
368+
assert!(matches!(ReceiverPersistedError::from(err), ReceiverPersistedError::Transient(_)));
369+
370+
// A right-sized garbage body fails OHTTP decapsulation, which is
371+
// fatal, so the binding error classifies it as fatal.
372+
let (_req, ctx) = receiver.create_poll_request(EXAMPLE_URL).expect("valid poll request");
373+
let err = receiver
374+
.process_response(&[0u8; ENCAPSULATED_MESSAGE_BYTES], ctx)
375+
.save(&persister)
376+
.expect_err("garbage response should fail");
377+
assert!(err.is_fatal());
378+
assert!(matches!(ReceiverPersistedError::from(err), ReceiverPersistedError::Fatal(_)));
379+
}
323380
}

‎payjoin-ffi/src/receive/mod.rs‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -748,6 +748,7 @@ pub struct UncheckedOriginalPayloadTransition(
748748
payjoin::receive::v2::Receiver<payjoin::receive::v2::MaybeInputsOwned>,
749749
payjoin::receive::Error,
750750
payjoin::receive::v2::Receiver<payjoin::receive::v2::HasReplyableError>,
751+
payjoin::receive::v2::Receiver<payjoin::receive::v2::UncheckedOriginalPayload>,
751752
>,
752753
>,
753754
>,
@@ -829,6 +830,7 @@ pub struct MaybeInputsOwnedTransition(
829830
payjoin::receive::v2::Receiver<payjoin::receive::v2::MaybeInputsSeen>,
830831
payjoin::receive::Error,
831832
payjoin::receive::v2::Receiver<payjoin::receive::v2::HasReplyableError>,
833+
payjoin::receive::v2::Receiver<payjoin::receive::v2::MaybeInputsOwned>,
832834
>,
833835
>,
834836
>,
@@ -884,6 +886,7 @@ pub struct MaybeInputsSeenTransition(
884886
payjoin::receive::v2::Receiver<payjoin::receive::v2::OutputsUnknown>,
885887
payjoin::receive::Error,
886888
payjoin::receive::v2::Receiver<payjoin::receive::v2::HasReplyableError>,
889+
payjoin::receive::v2::Receiver<payjoin::receive::v2::MaybeInputsSeen>,
887890
>,
888891
>,
889892
>,
@@ -937,6 +940,7 @@ pub struct OutputsUnknownTransition(
937940
payjoin::receive::v2::Receiver<payjoin::receive::v2::WantsOutputs>,
938941
payjoin::receive::Error,
939942
payjoin::receive::v2::Receiver<payjoin::receive::v2::HasReplyableError>,
943+
payjoin::receive::v2::Receiver<payjoin::receive::v2::OutputsUnknown>,
940944
>,
941945
>,
942946
>,
@@ -1137,6 +1141,8 @@ pub struct WantsFeeRangeTransition(
11371141
payjoin::receive::v2::SessionEvent,
11381142
payjoin::receive::v2::Receiver<payjoin::receive::v2::ProvisionalProposal>,
11391143
payjoin::receive::ProtocolError,
1144+
(),
1145+
payjoin::receive::v2::Receiver<payjoin::receive::v2::WantsFeeRange>,
11401146
>,
11411147
>,
11421148
>,
@@ -1217,6 +1223,7 @@ pub struct ProvisionalProposalTransition(
12171223
payjoin::receive::v2::SessionEvent,
12181224
payjoin::receive::v2::Receiver<payjoin::receive::v2::PayjoinProposal>,
12191225
payjoin::ImplementationError,
1226+
payjoin::receive::v2::Receiver<payjoin::receive::v2::ProvisionalProposal>,
12201227
>,
12211228
>,
12221229
>,
@@ -1279,6 +1286,7 @@ pub struct PayjoinProposalTransition(
12791286
payjoin::receive::v2::Receiver<payjoin::receive::v2::Monitor>,
12801287
payjoin::receive::ProtocolError,
12811288
payjoin::receive::v2::Receiver<payjoin::receive::v2::PendingFallback>,
1289+
payjoin::receive::v2::Receiver<payjoin::receive::v2::PayjoinProposal>,
12821290
>,
12831291
>,
12841292
>,
@@ -1368,6 +1376,7 @@ pub struct HasReplyableErrorTransition(
13681376
payjoin::receive::v2::SessionEvent,
13691377
payjoin::receive::v2::Receiver<payjoin::receive::v2::PendingFallback>,
13701378
payjoin::receive::ProtocolError,
1379+
payjoin::receive::v2::Receiver<payjoin::receive::v2::HasReplyableError>,
13711380
>,
13721381
>,
13731382
>,

0 commit comments

Comments
 (0)