From b87cd9a3db2b18c4231f0bef59ac914428aeb3c0 Mon Sep 17 00:00:00 2001 From: jolah1 Date: Mon, 24 Aug 2026 14:24:39 +0100 Subject: [PATCH] Test session persister load and close paths The in-memory persisters used by the existing tests are infallible, so the error paths where replay maps a storage failure into ReplayError were never exercised, and neither load() nor close() had direct coverage of their successful behavior. Add fault-injecting sync and async persisters under a single test-only module, and assert that sender and receiver replay surface a failure from load() and from the close() that replay performs when the event log is invalid. Assertions match on the error variant and downcast to the injected storage error rather than comparing Display strings, so rewording a message does not break them. Also cover the happy paths directly: a fresh persister loads nothing, load() replays events in save order without consuming the log, and close() marks the session closed while leaving the log readable. --- payjoin/src/core/error.rs | 9 ++ payjoin/src/core/persist.rs | 190 +++++++++++++++++++++++++ payjoin/src/core/receive/v2/session.rs | 36 +++++ payjoin/src/core/send/v2/session.rs | 35 +++++ 4 files changed, 270 insertions(+) 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); + } }