9335703964

Committed 02 Jun 2024 04:43AM UTC coverage: 83.156% (-0.03%) from 83.186%

Build # 9335703964

Build Type

Pull #1454

github

Committed by

web-flow

Commit Message

Merge 96c758cf7 into 4a8452f9b

Pull Request Pull Request #1454: Refactor wallet and persist mod, remove bdk_persist crate

Run Details

221 of 312 new or added lines in 5 files covered. (70.83%)

6 existing lines in 2 files now uncovered.

11286 of 13572 relevant lines covered (83.16%)

16663.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

64.63

/crates/chain/src/persist.rs

//! This module is home to the [`Persist`] trait which defines the behavior of a data store
//! required to persist changes made to BDK data structures.
//!
//! The [`StagedPersist`] type provides a convenient wrapper around implementations of [`Persist`] that
//! allows changes to be staged before committing them.
//!
//! The [`CombinedChangeSet`] type encapsulates a combination of [`crate`] structures that are
//! typically persisted together.
//!
//! To use async versions [`PersistAsync`] and [`StagedPersistAsync`] enable the `async` feature.

use crate::{indexed_tx_graph, keychain, local_chain, Anchor, Append};
#[cfg(feature = "async")]
use alloc::boxed::Box;
#[cfg(feature = "async")]
use async_trait::async_trait;
use bitcoin::Network;
use core::convert::Infallible;
use core::default::Default;
use core::fmt::{Debug, Display};
use core::mem;

/// A changeset containing [`crate`] structures typically persisted together.
#[derive(Debug, Clone, PartialEq)]
#[cfg_attr(
    feature = "serde",
    derive(crate::serde::Deserialize, crate::serde::Serialize),
    serde(
        crate = "crate::serde",
        bound(
            deserialize = "A: Ord + crate::serde::Deserialize<'de>, K: Ord + crate::serde::Deserialize<'de>",
            serialize = "A: Ord + crate::serde::Serialize, K: Ord + crate::serde::Serialize",
        ),
    )
)]
pub struct CombinedChangeSet<K, A> {
    /// Changes to the [`LocalChain`](local_chain::LocalChain).
    pub chain: local_chain::ChangeSet,
    /// Changes to [`IndexedTxGraph`](indexed_tx_graph::IndexedTxGraph).
    pub indexed_tx_graph: indexed_tx_graph::ChangeSet<A, keychain::ChangeSet<K>>,
    /// Stores the network type of the transaction data.
    pub network: Option<Network>,
}

impl<K, A> Default for CombinedChangeSet<K, A> {
    fn default() -> Self {
        Self {
            chain: Default::default(),
            indexed_tx_graph: Default::default(),
            network: None,
        }
    }
}

impl<K: Ord, A: Anchor> Append for CombinedChangeSet<K, A> {
    fn append(&mut self, other: Self) {
        Append::append(&mut self.chain, other.chain);
        Append::append(&mut self.indexed_tx_graph, other.indexed_tx_graph);
        if other.network.is_some() {
            debug_assert!(
                self.network.is_none() || self.network == other.network,
                "network type must either be just introduced or remain the same"
            );
            self.network = other.network;
        }
    }

    fn is_empty(&self) -> bool {
        self.chain.is_empty() && self.indexed_tx_graph.is_empty() && self.network.is_none()
    }
}

impl<K, A> From<local_chain::ChangeSet> for CombinedChangeSet<K, A> {
    fn from(chain: local_chain::ChangeSet) -> Self {
        Self {
            chain,
            ..Default::default()
        }
    }
}

impl<K, A> From<indexed_tx_graph::ChangeSet<A, keychain::ChangeSet<K>>>
    for CombinedChangeSet<K, A>
{
    fn from(indexed_tx_graph: indexed_tx_graph::ChangeSet<A, keychain::ChangeSet<K>>) -> Self {
        Self {
            indexed_tx_graph,
            ..Default::default()
        }
    }
}

/// A persistence backend for writing and loading changesets.
///
/// `C` represents the changeset; a datatype that records changes made to in-memory data structures
/// that are to be persisted, or retrieved from persistence.
pub trait Persist<C> {
    /// The error the backend returns when it fails to write.
    type WriteError: Debug + Display;

    /// The error the backend returns when it fails to load changesets `C`.
    type LoadError: Debug + Display;

    /// Writes a changeset to the persistence backend.
    ///
    /// It is up to the backend what it does with this. It could store every changeset in a list or
    /// it inserts the actual changes into a more structured database. All it needs to guarantee is
    /// that [`load_from_persistence`] restores a keychain tracker to what it should be if all
    /// changesets had been applied sequentially.
    ///
    /// [`load_from_persistence`]: Self::load_changes
    fn write_changes(&mut self, changeset: &C) -> Result<(), Self::WriteError>;

    /// Return the aggregate changeset `C` from persistence.
    fn load_changes(&mut self) -> Result<Option<C>, Self::LoadError>;
}

impl<C> Persist<C> for () {
    type WriteError = Infallible;
    type LoadError = Infallible;

    fn write_changes(&mut self, _changeset: &C) -> Result<(), Self::WriteError> {
        Ok(())
    }

    fn load_changes(&mut self) -> Result<Option<C>, Self::LoadError> {
        Ok(None)
    }
}

#[cfg(feature = "async")]
/// An async persistence backend for writing and loading changesets.
///
/// `C` represents the changeset; a datatype that records changes made to in-memory data structures
/// that are to be persisted, or retrieved from persistence.
#[async_trait]
pub trait PersistAsync<C> {
    /// The error the backend returns when it fails to write.
    type WriteError: Debug + Display;

    /// The error the backend returns when it fails to load changesets `C`.
    type LoadError: Debug + Display;

    /// Writes a changeset to the persistence backend.
    ///
    /// It is up to the backend what it does with this. It could store every changeset in a list or
    /// it inserts the actual changes into a more structured database. All it needs to guarantee is
    /// that [`load_from_persistence`] restores a keychain tracker to what it should be if all
    /// changesets had been applied sequentially.
    ///
    /// [`load_from_persistence`]: Self::load_changes
    async fn write_changes(&mut self, changeset: &C) -> Result<(), Self::WriteError>;

    /// Return the aggregate changeset `C` from persistence.
    async fn load_changes(&mut self) -> Result<Option<C>, Self::LoadError>;
}

#[cfg(feature = "async")]
#[async_trait]
impl<C> PersistAsync<C> for () {
    type WriteError = Infallible;
    type LoadError = Infallible;

    async fn write_changes(&mut self, _changeset: &C) -> Result<(), Self::WriteError> {
        Ok(())
    }

    async fn load_changes(&mut self) -> Result<Option<C>, Self::LoadError> {
        Ok(None)
    }
}

/// `StagedPersist` adds a convenient staging area for changesets before they are persisted.
///
/// Not all changes to the in-memory representation needs to be written to disk right away, so
/// [`crate::persist::StagedPersist::stage`] can be used to *stage* changes first and then
/// [`crate::persist::StagedPersist::commit`] can be used to write changes to disk.
pub struct StagedPersist<C, P: Persist<C>> {
    inner: P,
    staged: C,
}

impl<C, P: Persist<C>> Persist<C> for StagedPersist<C, P> {
    type WriteError = P::WriteError;
    type LoadError = P::LoadError;

    fn write_changes(&mut self, changeset: &C) -> Result<(), Self::WriteError> {
        self.inner.write_changes(changeset)
    }

    fn load_changes(&mut self) -> Result<Option<C>, Self::LoadError> {
        self.inner.load_changes()
    }
}

impl<C, P> StagedPersist<C, P>
where
    C: Default + Append,
    P: Persist<C>,
{
    /// Create a new [`StagedPersist`] adding staging to an inner data store that implements
    /// [`Persist`].
    pub fn new(persist: P) -> Self {
        Self {
            inner: persist,
            staged: Default::default(),
        }
    }

    /// Stage a `changeset` to be committed later with [`commit`].
    ///
    /// [`commit`]: Self::commit
    pub fn stage(&mut self, changeset: C) {
        self.staged.append(changeset)
    }

    /// Get the changes that have not been committed yet.
    pub fn staged(&self) -> &C {
        &self.staged
    }

    /// Take the changes that have not been committed yet.
    ///
    /// New staged is set to default;
    pub fn take_staged(&mut self) -> C {
        mem::take(&mut self.staged)
    }

    /// Commit the staged changes to the underlying persistence backend.
    ///
    /// Changes that are committed (if any) are returned.
    ///
    /// # Error
    ///
    /// Returns a backend-defined error if this fails.
    pub fn commit(&mut self) -> Result<Option<C>, P::WriteError> {
        if self.staged().is_empty() {
            return Ok(None);
        }
        let staged = self.take_staged();
        self.write_changes(&staged)
            // if written successfully, take and return `self.stage`
            .map(|_| Some(staged))
    }

    /// Stages a new changeset and commits it (along with any other previously staged changes) to
    /// the persistence backend
    ///
    /// Convenience method for calling [`stage`] and then [`commit`].
    ///
    /// [`stage`]: Self::stage
    /// [`commit`]: Self::commit
    pub fn stage_and_commit(&mut self, changeset: C) -> Result<Option<C>, P::WriteError> {
        self.stage(changeset);
        self.commit()
    }
}

#[cfg(feature = "async")]
/// `StagedPersistAsync` adds a convenient async staging area for changesets before they are persisted.
///
/// Not all changes to the in-memory representation needs to be written to disk right away, so
/// [`StagedPersistAsync::stage`] can be used to *stage* changes first and then
/// [`StagedPersistAsync::commit`] can be used to write changes to disk.
pub struct StagedPersistAsync<C, P: PersistAsync<C>> {
    inner: P,
    staged: C,
}

#[cfg(feature = "async")]
#[async_trait]
impl<C: Send + Sync, P: PersistAsync<C> + Send> PersistAsync<C> for StagedPersistAsync<C, P> {
    type WriteError = P::WriteError;
    type LoadError = P::LoadError;

    async fn write_changes(&mut self, changeset: &C) -> Result<(), Self::WriteError> {
        self.inner.write_changes(changeset).await
    }

    async fn load_changes(&mut self) -> Result<Option<C>, Self::LoadError> {
        self.inner.load_changes().await
    }
}

#[cfg(feature = "async")]
impl<C, P> StagedPersistAsync<C, P>
where
    C: Default + Append + Send + Sync,
    P: PersistAsync<C> + Send,
{
    /// Create a new [`StagedPersistAsync`] adding staging to an inner data store that implements
    /// [`PersistAsync`].
    pub fn new(persist: P) -> Self {
        Self {
            inner: persist,
            staged: Default::default(),
        }
    }

    /// Stage a `changeset` to be committed later with [`commit`].
    ///
    /// [`commit`]: Self::commit
    pub fn stage(&mut self, changeset: C) {
        self.staged.append(changeset)
    }

    /// Get the changes that have not been committed yet.
    pub fn staged(&self) -> &C {
        &self.staged
    }

    /// Take the changes that have not been committed yet.
    ///
    /// New staged is set to default;
    pub fn take_staged(&mut self) -> C {
        mem::take(&mut self.staged)
    }

    /// Commit the staged changes to the underlying persistence backend.
    ///
    /// Changes that are committed (if any) are returned.
    ///
    /// # Error
    ///
    /// Returns a backend-defined error if this fails.
    pub async fn commit(&mut self) -> Result<Option<C>, P::WriteError> {
        if self.staged().is_empty() {
            return Ok(None);
        }
        let staged = self.take_staged();
        self.write_changes(&staged)
            .await
            // if written successfully, take and return `self.stage`
            .map(|_| Some(staged))
    }

    /// Stages a new changeset and commits it (along with any other previously staged changes) to
    /// the persistence backend
    ///
    /// Convenience method for calling [`stage`] and then [`commit`].
    ///
    /// [`stage`]: Self::stage
    /// [`commit`]: Self::commit
    pub async fn stage_and_commit(&mut self, changeset: C) -> Result<Option<C>, P::WriteError> {
        self.stage(changeset);
        self.commit().await
    }
}

#[cfg(test)]
mod test {
    extern crate core;

    use crate::persist::{Persist, StagedPersist};
    use crate::Append;
    use std::error::Error;
    use std::fmt::{self, Display, Formatter};
    use std::prelude::rust_2015::{String, ToString};
    use TestError::FailedWrite;

    struct TestBackend<C: Default + Append + Clone + ToString> {
        changeset: C,
    }

    #[derive(Debug, Eq, PartialEq)]
    enum TestError {
        FailedWrite,
        FailedLoad,
    }

    impl Display for TestError {
        fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
            write!(f, "{:?}", self)
        }
    }

    impl Error for TestError {}

    #[derive(Clone, Default)]
    struct TestChangeSet(Option<String>);

    impl fmt::Display for TestChangeSet {
        fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
            write!(f, "{}", self.clone().0.unwrap_or_default())
        }
    }

    impl Append for TestChangeSet {
        fn append(&mut self, other: Self) {
            if other.0.is_some() {
                self.0 = other.0
            }
        }

        fn is_empty(&self) -> bool {
            self.0.is_none()
        }
    }

    impl<C> Persist<C> for TestBackend<C>
    where
        C: Default + Append + Clone + ToString,
    {
        type WriteError = TestError;
        type LoadError = TestError;

        fn write_changes(&mut self, changeset: &C) -> Result<(), Self::WriteError> {
            if changeset.to_string() == "ERROR" {
                Err(FailedWrite)
            } else {
                self.changeset = changeset.clone();
                Ok(())
            }
        }

        fn load_changes(&mut self) -> Result<Option<C>, Self::LoadError> {
            if self.changeset.to_string() == "ERROR" {
                Err(Self::LoadError::FailedLoad)
            } else {
                Ok(Some(self.changeset.clone()))
            }
        }
    }

    #[test]
    fn test_persist_stage_commit() {
        let backend = TestBackend {
            changeset: TestChangeSet(None),
        };

        let mut staged_backend = StagedPersist::new(backend);
        staged_backend.stage(TestChangeSet(Some("ONE".to_string())));
        staged_backend.stage(TestChangeSet(None));
        staged_backend.stage(TestChangeSet(Some("TWO".to_string())));
        let result = staged_backend.commit();
        assert!(matches!(result, Ok(Some(TestChangeSet(Some(v)))) if v == *"TWO".to_string()));

        let result = staged_backend.commit();
        assert!(matches!(result, Ok(None)));

        staged_backend.stage(TestChangeSet(Some("TWO".to_string())));
        let result = staged_backend.stage_and_commit(TestChangeSet(Some("ONE".to_string())));
        assert!(matches!(result, Ok(Some(TestChangeSet(Some(v)))) if v == *"ONE".to_string()));
    }

    #[test]
    fn test_persist_commit_error() {
        let backend = TestBackend {
            changeset: TestChangeSet(None),
        };
        let mut staged_backend = StagedPersist::new(backend);
        staged_backend.stage(TestChangeSet(Some("ERROR".to_string())));
        let result = staged_backend.commit();
        assert!(matches!(result, Err(e) if e == FailedWrite));
    }
}

1	//! This module is home to the [`Persist`] trait which defines the behavior of a data store
2	//! required to persist changes made to BDK data structures.
3	//!
4	//! The [`StagedPersist`] type provides a convenient wrapper around implementations of [`Persist`] that
5	//! allows changes to be staged before committing them.
6	//!
7	//! The [`CombinedChangeSet`] type encapsulates a combination of [`crate`] structures that are
8	//! typically persisted together.
9	//!
10	//! To use async versions [`PersistAsync`] and [`StagedPersistAsync`] enable the `async` feature.
11
12	use crate::{indexed_tx_graph, keychain, local_chain, Anchor, Append};
13	#[cfg(feature = "async")]
14	use alloc::boxed::Box;
15	#[cfg(feature = "async")]
16	use async_trait::async_trait;
17	use bitcoin::Network;
18	use core::convert::Infallible;
19	use core::default::Default;
20	use core::fmt::{Debug, Display};
21	use core::mem;
22
23	/// A changeset containing [`crate`] structures typically persisted together.
24	#[derive(Debug, Clone, PartialEq)]
25	#[cfg_attr(
26	feature = "serde",
27	derive(crate::serde::Deserialize, crate::serde::Serialize),	12✔
28	serde(
29	crate = "crate::serde",
30	bound(
31	deserialize = "A: Ord + crate::serde::Deserialize<'de>, K: Ord + crate::serde::Deserialize<'de>",
32	serialize = "A: Ord + crate::serde::Serialize, K: Ord + crate::serde::Serialize",
33	),
34	)
35	)]
36	pub struct CombinedChangeSet<K, A> {
37	/// Changes to the [`LocalChain`](local_chain::LocalChain).
38	pub chain: local_chain::ChangeSet,
39	/// Changes to [`IndexedTxGraph`](indexed_tx_graph::IndexedTxGraph).
40	pub indexed_tx_graph: indexed_tx_graph::ChangeSet<A, keychain::ChangeSet<K>>,
41	/// Stores the network type of the transaction data.
42	pub network: Option<Network>,
43	}
44
45	impl<K, A> Default for CombinedChangeSet<K, A> {
46	fn default() -> Self {	11,283✔
47	Self {	11,283✔
48	chain: Default::default(),	11,283✔
49	indexed_tx_graph: Default::default(),	11,283✔
50	network: None,	11,283✔
51	}	11,283✔
52	}	11,283✔
53	}
54
55	impl<K: Ord, A: Anchor> Append for CombinedChangeSet<K, A> {
56	fn append(&mut self, other: Self) {	11,161✔
57	Append::append(&mut self.chain, other.chain);	11,161✔
58	Append::append(&mut self.indexed_tx_graph, other.indexed_tx_graph);	11,161✔
59	if other.network.is_some() {	11,161✔
60	debug_assert!(	3✔
61	self.network.is_none() \|\| self.network == other.network,	3✔
NEW 62	"network type must either be just introduced or remain the same"	×
63	);
64	self.network = other.network;	3✔
65	}	11,158✔
66	}	11,161✔
67
68	fn is_empty(&self) -> bool {	2,455✔
69	self.chain.is_empty() && self.indexed_tx_graph.is_empty() && self.network.is_none()	2,455✔
70	}	2,455✔
71	}
72
73	impl<K, A> From<local_chain::ChangeSet> for CombinedChangeSet<K, A> {
74	fn from(chain: local_chain::ChangeSet) -> Self {	2,158✔
75	Self {	2,158✔
76	chain,	2,158✔
77	..Default::default()	2,158✔
78	}	2,158✔
79	}	2,158✔
80	}
81
82	impl<K, A> From<indexed_tx_graph::ChangeSet<A, keychain::ChangeSet<K>>>
83	for CombinedChangeSet<K, A>
84	{
85	fn from(indexed_tx_graph: indexed_tx_graph::ChangeSet<A, keychain::ChangeSet<K>>) -> Self {	6,580✔
86	Self {	6,580✔
87	indexed_tx_graph,	6,580✔
88	..Default::default()	6,580✔
89	}	6,580✔
90	}	6,580✔
91	}
92
93	/// A persistence backend for writing and loading changesets.
94	///
95	/// `C` represents the changeset; a datatype that records changes made to in-memory data structures
96	/// that are to be persisted, or retrieved from persistence.
97	pub trait Persist<C> {
98	/// The error the backend returns when it fails to write.
99	type WriteError: Debug + Display;
100
101	/// The error the backend returns when it fails to load changesets `C`.
102	type LoadError: Debug + Display;
103
104	/// Writes a changeset to the persistence backend.
105	///
106	/// It is up to the backend what it does with this. It could store every changeset in a list or
107	/// it inserts the actual changes into a more structured database. All it needs to guarantee is
108	/// that [`load_from_persistence`] restores a keychain tracker to what it should be if all
109	/// changesets had been applied sequentially.
110	///
111	/// [`load_from_persistence`]: Self::load_changes
112	fn write_changes(&mut self, changeset: &C) -> Result<(), Self::WriteError>;
113
114	/// Return the aggregate changeset `C` from persistence.
115	fn load_changes(&mut self) -> Result<Option<C>, Self::LoadError>;
116	}
117
118	impl<C> Persist<C> for () {
119	type WriteError = Infallible;
120	type LoadError = Infallible;
121
NEW 122	fn write_changes(&mut self, _changeset: &C) -> Result<(), Self::WriteError> {	×
NEW 123	Ok(())	×
NEW 124	}	×
125
NEW 126	fn load_changes(&mut self) -> Result<Option<C>, Self::LoadError> {	×
NEW 127	Ok(None)	×
NEW 128	}	×
129	}
130
131	#[cfg(feature = "async")]
132	/// An async persistence backend for writing and loading changesets.
133	///
134	/// `C` represents the changeset; a datatype that records changes made to in-memory data structures
135	/// that are to be persisted, or retrieved from persistence.
136	#[async_trait]
137	pub trait PersistAsync<C> {
138	/// The error the backend returns when it fails to write.
139	type WriteError: Debug + Display;
140
141	/// The error the backend returns when it fails to load changesets `C`.
142	type LoadError: Debug + Display;
143
144	/// Writes a changeset to the persistence backend.
145	///
146	/// It is up to the backend what it does with this. It could store every changeset in a list or
147	/// it inserts the actual changes into a more structured database. All it needs to guarantee is
148	/// that [`load_from_persistence`] restores a keychain tracker to what it should be if all
149	/// changesets had been applied sequentially.
150	///
151	/// [`load_from_persistence`]: Self::load_changes
152	async fn write_changes(&mut self, changeset: &C) -> Result<(), Self::WriteError>;
153
154	/// Return the aggregate changeset `C` from persistence.
155	async fn load_changes(&mut self) -> Result<Option<C>, Self::LoadError>;
156	}
157
158	#[cfg(feature = "async")]
159	#[async_trait]
160	impl<C> PersistAsync<C> for () {
161	type WriteError = Infallible;
162	type LoadError = Infallible;
163
NEW 164	async fn write_changes(&mut self, _changeset: &C) -> Result<(), Self::WriteError> {	×
NEW 165	Ok(())	×
NEW 166	}	×
167
NEW 168	async fn load_changes(&mut self) -> Result<Option<C>, Self::LoadError> {	×
NEW 169	Ok(None)	×
NEW 170	}	×
171	}
172
173	/// `StagedPersist` adds a convenient staging area for changesets before they are persisted.
174	///
175	/// Not all changes to the in-memory representation needs to be written to disk right away, so
176	/// [`crate::persist::StagedPersist::stage`] can be used to stage changes first and then
177	/// [`crate::persist::StagedPersist::commit`] can be used to write changes to disk.
178	pub struct StagedPersist<C, P: Persist<C>> {
179	inner: P,
180	staged: C,
181	}
182
183	impl<C, P: Persist<C>> Persist<C> for StagedPersist<C, P> {
184	type WriteError = P::WriteError;
185	type LoadError = P::LoadError;
186
187	fn write_changes(&mut self, changeset: &C) -> Result<(), Self::WriteError> {	3✔
188	self.inner.write_changes(changeset)	3✔
189	}	3✔
190
NEW 191	fn load_changes(&mut self) -> Result<Option<C>, Self::LoadError> {	×
NEW 192	self.inner.load_changes()	×
NEW 193	}	×
194	}
195
196	impl<C, P> StagedPersist<C, P>
197	where
198	C: Default + Append,
199	P: Persist<C>,
200	{
201	/// Create a new [`StagedPersist`] adding staging to an inner data store that implements
202	/// [`Persist`].
203	pub fn new(persist: P) -> Self {	2✔
204	Self {	2✔
205	inner: persist,	2✔
206	staged: Default::default(),	2✔
207	}	2✔
208	}	2✔
209
210	/// Stage a `changeset` to be committed later with [`commit`].
211	///
212	/// [`commit`]: Self::commit
213	pub fn stage(&mut self, changeset: C) {	6✔
214	self.staged.append(changeset)	6✔
215	}	6✔
216
217	/// Get the changes that have not been committed yet.
218	pub fn staged(&self) -> &C {	4✔
219	&self.staged	4✔
220	}	4✔
221
222	/// Take the changes that have not been committed yet.
223	///
224	/// New staged is set to default;
225	pub fn take_staged(&mut self) -> C {	3✔
226	mem::take(&mut self.staged)	3✔
227	}	3✔
228
229	/// Commit the staged changes to the underlying persistence backend.
230	///
231	/// Changes that are committed (if any) are returned.
232	///
233	/// # Error
234	///
235	/// Returns a backend-defined error if this fails.
236	pub fn commit(&mut self) -> Result<Option<C>, P::WriteError> {	4✔
237	if self.staged().is_empty() {	4✔
238	return Ok(None);	1✔
239	}	3✔
240	let staged = self.take_staged();	3✔
241	self.write_changes(&staged)	3✔
242	// if written successfully, take and return `self.stage`	3✔
243	.map(\|_\| Some(staged))	3✔
244	}	4✔
245
246	/// Stages a new changeset and commits it (along with any other previously staged changes) to
247	/// the persistence backend
248	///
249	/// Convenience method for calling [`stage`] and then [`commit`].
250	///
251	/// [`stage`]: Self::stage
252	/// [`commit`]: Self::commit
253	pub fn stage_and_commit(&mut self, changeset: C) -> Result<Option<C>, P::WriteError> {	1✔
254	self.stage(changeset);	1✔
255	self.commit()	1✔
256	}	1✔
257	}
258
259	#[cfg(feature = "async")]
260	/// `StagedPersistAsync` adds a convenient async staging area for changesets before they are persisted.
261	///
262	/// Not all changes to the in-memory representation needs to be written to disk right away, so
263	/// [`StagedPersistAsync::stage`] can be used to stage changes first and then
264	/// [`StagedPersistAsync::commit`] can be used to write changes to disk.
265	pub struct StagedPersistAsync<C, P: PersistAsync<C>> {
266	inner: P,
267	staged: C,
268	}
269
270	#[cfg(feature = "async")]
271	#[async_trait]
272	impl<C: Send + Sync, P: PersistAsync<C> + Send> PersistAsync<C> for StagedPersistAsync<C, P> {
273	type WriteError = P::WriteError;
274	type LoadError = P::LoadError;
275
NEW 276	async fn write_changes(&mut self, changeset: &C) -> Result<(), Self::WriteError> {	×
NEW 277	self.inner.write_changes(changeset).await	×
NEW 278	}	×
279
NEW 280	async fn load_changes(&mut self) -> Result<Option<C>, Self::LoadError> {	×
NEW 281	self.inner.load_changes().await	×
NEW 282	}	×
283	}
284
285	#[cfg(feature = "async")]
286	impl<C, P> StagedPersistAsync<C, P>
287	where
288	C: Default + Append + Send + Sync,
289	P: PersistAsync<C> + Send,
290	{
291	/// Create a new [`StagedPersistAsync`] adding staging to an inner data store that implements
292	/// [`PersistAsync`].
NEW 293	pub fn new(persist: P) -> Self {	×
NEW 294	Self {	×
NEW 295	inner: persist,	×
NEW 296	staged: Default::default(),	×
NEW 297	}	×
NEW 298	}	×
299
300	/// Stage a `changeset` to be committed later with [`commit`].
301	///
302	/// [`commit`]: Self::commit
NEW 303	pub fn stage(&mut self, changeset: C) {	×
NEW 304	self.staged.append(changeset)	×
NEW 305	}	×
306
307	/// Get the changes that have not been committed yet.
NEW 308	pub fn staged(&self) -> &C {	×
NEW 309	&self.staged	×
NEW 310	}	×
311
312	/// Take the changes that have not been committed yet.
313	///
314	/// New staged is set to default;
NEW 315	pub fn take_staged(&mut self) -> C {	×
NEW 316	mem::take(&mut self.staged)	×
NEW 317	}	×
318
319	/// Commit the staged changes to the underlying persistence backend.
320	///
321	/// Changes that are committed (if any) are returned.
322	///
323	/// # Error
324	///
325	/// Returns a backend-defined error if this fails.
NEW 326	pub async fn commit(&mut self) -> Result<Option<C>, P::WriteError> {	×
NEW 327	if self.staged().is_empty() {	×
NEW 328	return Ok(None);	×
NEW 329	}	×
NEW 330	let staged = self.take_staged();	×
NEW 331	self.write_changes(&staged)	×
NEW 332	.await	×
333	// if written successfully, take and return `self.stage`
NEW 334	.map(\|_\| Some(staged))	×
NEW 335	}	×
336
337	/// Stages a new changeset and commits it (along with any other previously staged changes) to
338	/// the persistence backend
339	///
340	/// Convenience method for calling [`stage`] and then [`commit`].
341	///
342	/// [`stage`]: Self::stage
343	/// [`commit`]: Self::commit
NEW 344	pub async fn stage_and_commit(&mut self, changeset: C) -> Result<Option<C>, P::WriteError> {	×
NEW 345	self.stage(changeset);	×
NEW 346	self.commit().await	×
NEW 347	}	×
348	}
349
350	#[cfg(test)]
351	mod test {
352	extern crate core;
353
354	use crate::persist::{Persist, StagedPersist};
355	use crate::Append;
356	use std::error::Error;
357	use std::fmt::{self, Display, Formatter};
358	use std::prelude::rust_2015::{String, ToString};
359	use TestError::FailedWrite;
360
361	struct TestBackend<C: Default + Append + Clone + ToString> {
362	changeset: C,
363	}
364
365	#[derive(Debug, Eq, PartialEq)]
366	enum TestError {
367	FailedWrite,
368	FailedLoad,
369	}
370
371	impl Display for TestError {
NEW 372	fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {	×
NEW 373	write!(f, "{:?}", self)	×
NEW 374	}	×
375	}
376
377	impl Error for TestError {}
378
379	#[derive(Clone, Default)]
380	struct TestChangeSet(Option<String>);
381
382	impl fmt::Display for TestChangeSet {
383	fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {	3✔
384	write!(f, "{}", self.clone().0.unwrap_or_default())	3✔
385	}	3✔
386	}
387
388	impl Append for TestChangeSet {
389	fn append(&mut self, other: Self) {	6✔
390	if other.0.is_some() {	6✔
391	self.0 = other.0	5✔
392	}	1✔
393	}	6✔
394
395	fn is_empty(&self) -> bool {	4✔
396	self.0.is_none()	4✔
397	}	4✔
398	}
399
400	impl<C> Persist<C> for TestBackend<C>
401	where
402	C: Default + Append + Clone + ToString,
403	{
404	type WriteError = TestError;
405	type LoadError = TestError;
406
407	fn write_changes(&mut self, changeset: &C) -> Result<(), Self::WriteError> {	3✔
408	if changeset.to_string() == "ERROR" {	3✔
409	Err(FailedWrite)	1✔
410	} else {
411	self.changeset = changeset.clone();	2✔
412	Ok(())	2✔
413	}
414	}	3✔
415
NEW 416	fn load_changes(&mut self) -> Result<Option<C>, Self::LoadError> {	×
NEW 417	if self.changeset.to_string() == "ERROR" {	×
NEW 418	Err(Self::LoadError::FailedLoad)	×
419	} else {
NEW 420	Ok(Some(self.changeset.clone()))	×
421	}
NEW 422	}	×
423	}
424
425	#[test]
426	fn test_persist_stage_commit() {	1✔
427	let backend = TestBackend {	1✔
428	changeset: TestChangeSet(None),	1✔
429	};	1✔
430		1✔
431	let mut staged_backend = StagedPersist::new(backend);	1✔
432	staged_backend.stage(TestChangeSet(Some("ONE".to_string())));	1✔
433	staged_backend.stage(TestChangeSet(None));	1✔
434	staged_backend.stage(TestChangeSet(Some("TWO".to_string())));	1✔
435	let result = staged_backend.commit();	1✔
436	assert!(matches!(result, Ok(Some(TestChangeSet(Some(v)))) if v == *"TWO".to_string()));	1✔
437
438	let result = staged_backend.commit();	1✔
439	assert!(matches!(result, Ok(None)));	1✔
440
441	staged_backend.stage(TestChangeSet(Some("TWO".to_string())));	1✔
442	let result = staged_backend.stage_and_commit(TestChangeSet(Some("ONE".to_string())));	1✔
443	assert!(matches!(result, Ok(Some(TestChangeSet(Some(v)))) if v == *"ONE".to_string()));	1✔
444	}	1✔
445
446	#[test]
447	fn test_persist_commit_error() {	1✔
448	let backend = TestBackend {	1✔
449	changeset: TestChangeSet(None),	1✔
450	};	1✔
451	let mut staged_backend = StagedPersist::new(backend);	1✔
452	staged_backend.stage(TestChangeSet(Some("ERROR".to_string())));	1✔
453	let result = staged_backend.commit();	1✔
454	assert!(matches!(result, Err(e) if e == FailedWrite));	1✔
455	}	1✔
456	}

bitcoindevkit / bdk / 9335703964

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous