Test replay persistence failure paths

payjoin/src/core/error.rs

@@ -89,6 +89,15 @@ impl<SessionState, SessionEvent> ReplayError<SessionState, SessionEvent> {
             _ => 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")]

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<V> {
+        events: Vec<V>,
+        failure: PersisterFailure,
+    }
+
+    impl<V> FailingPersister<V> {
+        pub(crate) fn on_load() -> Self { Self { events: vec![], failure: PersisterFailure::Load } }
+
+        pub(crate) fn on_close(events: Vec<V>) -> Self {
+            Self { events, failure: PersisterFailure::Close }
+        }
+    }
+
+    impl<V> SessionPersister for FailingPersister<V>
+    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<Box<dyn Iterator<Item = Self::SessionEvent>>, 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<V>(FailingPersister<V>);
+
+    impl<V> FailingAsyncPersister<V> {
+        pub(crate) fn on_load() -> Self { Self(FailingPersister::on_load()) }
+
+        pub(crate) fn on_close(events: Vec<V>) -> Self { Self(FailingPersister::on_close(events)) }
+    }
+
+    impl<V> AsyncSessionPersister for FailingAsyncPersister<V>
+    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<Box<dyn Iterator<Item = Self::SessionEvent> + 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<SessionState, SessionEvent>(
+        err: &ReplayError<SessionState, SessionEvent>,
+        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::<TestStorageError>())
+            .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<InMemoryTestEvent>| {
+            persister.load().expect("load should not fail").map(|event| event.0).collect::<Vec<_>>()
+        };
+        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<InMemoryTestEvent>) -> Vec<String> {
+            persister
+                .load()
+                .await
+                .expect("load should not fail")
+                .map(|event| event.0)
+                .collect::<Vec<_>>()
+        }
+        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");
+    }
 }

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::<SessionEvent>::on_load())
+            .expect_err("load failure should fail replay");
+        assert_persistence_failure(&err, PersisterFailure::Load);
+
+        let err = replay_event_log_async(&FailingAsyncPersister::<SessionEvent>::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();

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::<SessionEvent>::on_load())
+            .expect_err("load failure should fail replay");
+        assert_persistence_failure(&err, PersisterFailure::Load);
+
+        let err = replay_event_log_async(&FailingAsyncPersister::<SessionEvent>::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);
+    }
 }