diff --git a/payjoin/src/core/error.rs b/payjoin/src/core/error.rs index 35857fd1f..034d8b4e2 100644 --- a/payjoin/src/core/error.rs +++ b/payjoin/src/core/error.rs @@ -89,6 +89,15 @@ impl ReplayError { _ => None, } } + + /// Returns the application storage error that failed the replay, if any. + #[cfg(test)] + pub(crate) fn persistence_failure(&self) -> Option<&ImplementationError> { + match &self.0 { + InternalReplayError::PersistenceFailure(e) => Some(e), + _ => None, + } + } } #[cfg(feature = "v2")] diff --git a/payjoin/src/core/persist.rs b/payjoin/src/core/persist.rs index c6be11cec..bdfc0dce8 100644 --- a/payjoin/src/core/persist.rs +++ b/payjoin/src/core/persist.rs @@ -1099,6 +1099,136 @@ where } } +#[cfg(test)] +pub(crate) mod test_support { + use std::error::Error; + + use super::*; + use crate::error::ReplayError; + + /// Which [`SessionPersister`] method a fault-injecting persister fails on. + #[derive(Debug, Clone, Copy, PartialEq, Eq)] + pub(crate) enum PersisterFailure { + Load, + Close, + } + + /// Storage error reported by the fault-injecting persisters. + #[derive(Debug)] + pub(crate) struct TestStorageError(PersisterFailure); + + impl fmt::Display for TestStorageError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self.0 { + PersisterFailure::Load => write!(f, "load failed"), + PersisterFailure::Close => write!(f, "close failed"), + } + } + } + + impl Error for TestStorageError {} + + /// Session persister that fails on exactly one of `load` or `close`, so + /// replay's persistence error paths can be exercised. + pub(crate) struct FailingPersister { + events: Vec, + failure: PersisterFailure, + } + + impl FailingPersister { + pub(crate) fn on_load() -> Self { Self { events: vec![], failure: PersisterFailure::Load } } + + pub(crate) fn on_close(events: Vec) -> Self { + Self { events, failure: PersisterFailure::Close } + } + } + + impl SessionPersister for FailingPersister + where + V: Clone + 'static, + { + type InternalStorageError = TestStorageError; + type SessionEvent = V; + + fn save_event(&self, _event: Self::SessionEvent) -> Result<(), Self::InternalStorageError> { + Ok(()) + } + + fn load( + &self, + ) -> Result>, Self::InternalStorageError> + { + match self.failure { + PersisterFailure::Load => Err(TestStorageError(PersisterFailure::Load)), + PersisterFailure::Close => Ok(Box::new(self.events.clone().into_iter())), + } + } + + fn close(&self) -> Result<(), Self::InternalStorageError> { + match self.failure { + PersisterFailure::Load => Ok(()), + PersisterFailure::Close => Err(TestStorageError(PersisterFailure::Close)), + } + } + } + + /// Async counterpart of [`FailingPersister`]. + pub(crate) struct FailingAsyncPersister(FailingPersister); + + impl FailingAsyncPersister { + pub(crate) fn on_load() -> Self { Self(FailingPersister::on_load()) } + + pub(crate) fn on_close(events: Vec) -> Self { Self(FailingPersister::on_close(events)) } + } + + impl AsyncSessionPersister for FailingAsyncPersister + where + V: Clone + Send + Sync + 'static, + { + type InternalStorageError = TestStorageError; + type SessionEvent = V; + + async fn save_event( + &self, + _event: Self::SessionEvent, + ) -> Result<(), Self::InternalStorageError> { + Ok(()) + } + + async fn load( + &self, + ) -> Result + Send>, Self::InternalStorageError> + { + match self.0.failure { + PersisterFailure::Load => Err(TestStorageError(PersisterFailure::Load)), + PersisterFailure::Close => Ok(Box::new(self.0.events.clone().into_iter())), + } + } + + async fn close(&self) -> Result<(), Self::InternalStorageError> { + match self.0.failure { + PersisterFailure::Load => Ok(()), + PersisterFailure::Close => Err(TestStorageError(PersisterFailure::Close)), + } + } + } + + /// Assert that replay failed on persistence, and that the failure came from + /// the injected fault rather than from some other storage error. + pub(crate) fn assert_persistence_failure( + err: &ReplayError, + expected: PersisterFailure, + ) { + let implementation_error = + err.persistence_failure().expect("replay should fail on persistence"); + let storage_error = implementation_error + .source() + .and_then(|source| source.downcast_ref::()) + .expect("persistence failure should carry the injected storage error"); + assert_eq!(storage_error.0, expected); + } +} + #[cfg(test)] mod tests { use serde::{Deserialize, Serialize}; @@ -1844,4 +1974,64 @@ mod tests { assert!(!transient_error.is_fatal()); assert_eq!(transient_error.transient_state(), Some("Current state".to_string())); } + + #[test] + fn in_memory_persister_load_and_close() { + let persister = InMemoryPersister::default(); + assert!( + persister.load().expect("empty load should not fail").next().is_none(), + "a fresh persister should load no events" + ); + + let saved = ["first", "second", "third"]; + for event in saved { + persister + .save_event(InMemoryTestEvent(event.to_string())) + .expect("save should not fail"); + } + + let loaded = |persister: &InMemoryPersister| { + persister.load().expect("load should not fail").map(|event| event.0).collect::>() + }; + assert_eq!(loaded(&persister), saved, "load should replay events in save order"); + assert_eq!(loaded(&persister), saved, "load should not consume the event log"); + + assert!(!persister.inner.lock().expect("lock should not be poisoned").is_closed); + persister.close().expect("close should not fail"); + assert!(persister.inner.lock().expect("lock should not be poisoned").is_closed); + assert_eq!(loaded(&persister), saved, "closing should not discard the event log"); + } + + #[tokio::test] + async fn in_memory_async_persister_load_and_close() { + let persister = InMemoryAsyncPersister::default(); + assert!( + persister.load().await.expect("empty load should not fail").next().is_none(), + "a fresh persister should load no events" + ); + + let saved = ["first", "second", "third"]; + for event in saved { + persister + .save_event(InMemoryTestEvent(event.to_string())) + .await + .expect("save should not fail"); + } + + async fn loaded(persister: &InMemoryAsyncPersister) -> Vec { + persister + .load() + .await + .expect("load should not fail") + .map(|event| event.0) + .collect::>() + } + assert_eq!(loaded(&persister).await, saved, "load should replay events in save order"); + assert_eq!(loaded(&persister).await, saved, "load should not consume the event log"); + + assert!(!persister.inner.lock().await.is_closed); + persister.close().await.expect("close should not fail"); + assert!(persister.inner.lock().await.is_closed); + assert_eq!(loaded(&persister).await, saved, "closing should not discard the event log"); + } } diff --git a/payjoin/src/core/receive/v2/session.rs b/payjoin/src/core/receive/v2/session.rs index 63fd3c3da..300aa9235 100644 --- a/payjoin/src/core/receive/v2/session.rs +++ b/payjoin/src/core/receive/v2/session.rs @@ -252,6 +252,9 @@ mod tests { use payjoin_test_utils::{BoxError, EXAMPLE_URL}; use super::*; + use crate::persist::test_support::{ + assert_persistence_failure, FailingAsyncPersister, FailingPersister, PersisterFailure, + }; use crate::persist::{InMemoryAsyncPersister, InMemoryPersister}; use crate::receive::tests::original_from_test_vector; use crate::receive::v2::test::{mock_err, SHARED_CONTEXT}; @@ -784,6 +787,39 @@ mod tests { assert!(persister.inner.lock().await.is_closed); } + /// A `load` that fails aborts replay before any event is examined. + #[tokio::test] + async fn replay_surfaces_load_failure() { + let err = replay_event_log(&FailingPersister::::on_load()) + .expect_err("load failure should fail replay"); + assert_persistence_failure(&err, PersisterFailure::Load); + + let err = replay_event_log_async(&FailingAsyncPersister::::on_load()) + .await + .expect_err("async load failure should fail replay"); + assert_persistence_failure(&err, PersisterFailure::Load); + } + + /// Replay closes the session when the event log is invalid, so an event log + /// that both replays badly and fails to close reports the close failure: + /// storage is broken, which is the more actionable of the two errors. The + /// invalid first event below is what drives replay into that close call. + #[tokio::test] + async fn replay_surfaces_close_failure() { + let err = replay_event_log(&FailingPersister::on_close(vec![ + SessionEvent::CheckedBroadcastSuitability(), + ])) + .expect_err("close failure should replace the invalid-event error"); + assert_persistence_failure(&err, PersisterFailure::Close); + + let err = replay_event_log_async(&FailingAsyncPersister::on_close(vec![ + SessionEvent::CheckedBroadcastSuitability(), + ])) + .await + .expect_err("async close failure should replace the invalid-event error"); + assert_persistence_failure(&err, PersisterFailure::Close); + } + #[tokio::test] async fn test_replaying_unchecked_proposal() { let session_context = SHARED_CONTEXT.clone(); diff --git a/payjoin/src/core/send/v2/session.rs b/payjoin/src/core/send/v2/session.rs index 78126bced..c5bdb0d9f 100644 --- a/payjoin/src/core/send/v2/session.rs +++ b/payjoin/src/core/send/v2/session.rs @@ -186,6 +186,9 @@ mod tests { use super::*; use crate::core::Url; use crate::output_substitution::OutputSubstitution; + use crate::persist::test_support::{ + assert_persistence_failure, FailingAsyncPersister, FailingPersister, PersisterFailure, + }; use crate::persist::{InMemoryAsyncPersister, InMemoryPersister}; use crate::send::v2::{Sender, SenderBuilder, SessionContext, WithReplyKey}; use crate::send::PsbtContext; @@ -538,4 +541,36 @@ mod tests { assert_eq!(err.to_string(), expected_err.to_string()); assert!(persister.inner.lock().await.is_closed); } + + /// A `load` that fails aborts replay before any event is examined. + #[tokio::test] + async fn replay_surfaces_load_failure() { + let err = replay_event_log(&FailingPersister::::on_load()) + .expect_err("load failure should fail replay"); + assert_persistence_failure(&err, PersisterFailure::Load); + + let err = replay_event_log_async(&FailingAsyncPersister::::on_load()) + .await + .expect_err("async load failure should fail replay"); + assert_persistence_failure(&err, PersisterFailure::Load); + } + + /// Replay closes the session when the event log is invalid, so an event log + /// that both replays badly and fails to close reports the close failure: + /// storage is broken, which is the more actionable of the two errors. The + /// invalid first event below is what drives replay into that close call. + #[tokio::test] + async fn replay_surfaces_close_failure() { + let err = + replay_event_log(&FailingPersister::on_close(vec![SessionEvent::PostedOriginalPsbt()])) + .expect_err("close failure should replace the invalid-event error"); + assert_persistence_failure(&err, PersisterFailure::Close); + + let err = replay_event_log_async(&FailingAsyncPersister::on_close(vec![ + SessionEvent::PostedOriginalPsbt(), + ])) + .await + .expect_err("async close failure should replace the invalid-event error"); + assert_persistence_failure(&err, PersisterFailure::Close); + } }