nectar_primitives/file/
mod.rs

1//! File splitting and joining for arbitrary-size data.
2//!
3//! Joining is async; splitting is CPU-bound and runs on rayon. `Splitter`
4//! implements `Write` for streaming producers.
5//!
6//! # Store-centric API (extension traits)
7//!
8//! ```
9//! use futures::executor::block_on;
10//! use nectar_primitives::file::{ChunkGetExt, ChunkPutExt};
11//! use nectar_primitives::store::MemoryStore;
12//! use nectar_primitives::DEFAULT_BODY_SIZE;
13//!
14//! let store = MemoryStore::<DEFAULT_BODY_SIZE>::new();
15//! let addr = block_on(store.write_file(b"hello swarm".to_vec())).unwrap();
16//! let data = block_on(store.read_file(addr)).unwrap();
17//! assert_eq!(data, b"hello swarm");
18//! ```
19//!
20//! # Free function API
21//!
22//! ```
23//! use futures::executor::block_on;
24//! use nectar_primitives::file::{split, join};
25//! use nectar_primitives::DEFAULT_BODY_SIZE;
26//!
27//! let data = b"Hello, Swarm!";
28//! let (root, store) = split::<DEFAULT_BODY_SIZE>(data).unwrap();
29//! let recovered = block_on(join(&store, root)).unwrap();
30//! assert_eq!(recovered, data);
31//! ```
32//!
33//! # Encrypted split and join
34//!
35//! ```
36//! # #[cfg(feature = "encryption")] {
37//! use futures::executor::block_on;
38//! use nectar_primitives::file::{split_encrypted, join};
39//! use nectar_primitives::DEFAULT_BODY_SIZE;
40//!
41//! let data = b"secret data";
42//! let (root_ref, store) = split_encrypted::<DEFAULT_BODY_SIZE>(data).unwrap();
43//! let recovered = block_on(join(&store, root_ref)).unwrap();
44//! assert_eq!(recovered, data);
45//! # }
46//! ```
47
48mod constants;
49pub mod entry_ref;
50pub mod error;
51mod fold;
52mod frontier;
53mod helpers;
54#[cfg(test)]
55#[macro_use]
56mod joiner_tests;
57#[cfg(test)]
58#[macro_use]
59mod splitter_tests;
60mod joiner;
61pub mod mode;
62mod read_at;
63mod splitter;
64mod splitter_parallel;
65mod tree;
66mod windowed;
67mod write_at;
68
69use crate::chunk::ChunkAddress;
70#[cfg(feature = "encryption")]
71use crate::chunk::encryption::EncryptedChunkRef;
72use crate::store::{ChunkPut, MaybeSend, MaybeSync};
73
74// Async (primary) re-exports
75#[cfg(feature = "encryption")]
76pub use joiner::EncryptedJoiner;
77#[cfg(feature = "tokio")]
78pub use joiner::JoinerReader;
79pub use joiner::{GenericJoiner, Joiner};
80#[cfg(feature = "tokio")]
81pub use windowed::WindowedJoinerReader;
82pub use windowed::WindowedReader;
83// Splitter re-exports
84pub use read_at::ReadAt;
85#[cfg(feature = "encryption")]
86pub use splitter::EncryptedSplitter;
87pub use splitter::Splitter;
88#[cfg(feature = "encryption")]
89pub use splitter_parallel::EncryptedParallelSplitter;
90pub use splitter_parallel::ParallelSplitter;
91pub use write_at::WriteAt;
92
93pub use entry_ref::EntryRef;
94pub use error::FileError;
95pub use tree::{ChunkRange, TreeParams};
96
97mod join_ref_sealed {
98    pub trait Sealed {}
99    impl Sealed for crate::ChunkAddress {}
100    #[cfg(feature = "encryption")]
101    impl Sealed for crate::EncryptedChunkRef {}
102}
103
104/// Maps a reference type to its join mode.
105/// Sealed; implemented for `ChunkAddress` and `EncryptedChunkRef`.
106pub trait JoinRef: join_ref_sealed::Sealed + Clone + Send + Sync + 'static {
107    /// The join mode associated with this reference type.
108    type Mode: mode::JoinMode + Send + Sync;
109
110    /// Convert into the root reference expected by the joiner.
111    fn into_root_ref(self) -> <Self::Mode as mode::JoinMode>::RootRef;
112}
113
114impl JoinRef for ChunkAddress {
115    type Mode = mode::PlainMode;
116
117    fn into_root_ref(self) -> Self {
118        self
119    }
120}
121
122#[cfg(feature = "encryption")]
123impl JoinRef for EncryptedChunkRef {
124    type Mode = mode::EncryptedMode;
125
126    fn into_root_ref(self) -> Self {
127        self
128    }
129}
130
131/// Resolve a `SeekFrom` position to an absolute byte offset.
132pub(crate) fn resolve_seek_position(
133    pos: std::io::SeekFrom,
134    current: u64,
135    span: u64,
136) -> std::io::Result<u64> {
137    use std::io::{Error, ErrorKind::InvalidInput, SeekFrom};
138    let to_i64 = |v: u64, msg: &str| i64::try_from(v).map_err(|_| Error::new(InvalidInput, msg));
139    let new_pos = match pos {
140        SeekFrom::Start(off) => to_i64(off, "seek offset exceeds i64::MAX")?,
141        SeekFrom::End(off) => to_i64(span, "file span exceeds i64::MAX")? + off,
142        SeekFrom::Current(off) => to_i64(current, "current position exceeds i64::MAX")? + off,
143    };
144    if new_pos < 0 {
145        return Err(Error::new(InvalidInput, "seek to negative position"));
146    }
147    Ok(new_pos as u64)
148}
149
150// ---- Primary async API ----
151
152/// Join chunks asynchronously. Dispatches plain/encrypted via [`JoinRef`].
153pub async fn join<R, G, const BODY_SIZE: usize>(getter: G, root: R) -> error::Result<Vec<u8>>
154where
155    R: JoinRef,
156    G: crate::store::ChunkGet<BODY_SIZE>,
157{
158    GenericJoiner::<G, R::Mode, BODY_SIZE>::new(getter, root.into_root_ref())
159        .await?
160        .read_all()
161        .await
162}
163
164// ---- Splitting (CPU-bound, rayon) ----
165
166/// Split data into chunks, returning root address and chunk store.
167///
168/// Uses `ParallelSplitter` for best performance on in-memory data.
169pub fn split<const BODY_SIZE: usize>(
170    data: &[u8],
171) -> error::Result<(ChunkAddress, crate::store::MemoryStore<BODY_SIZE>)> {
172    let (root, chunks) = ParallelSplitter::<BODY_SIZE>::split_to_vec(&data)?;
173    Ok((root, crate::store::MemoryStore::from_chunks(chunks)))
174}
175
176/// Split data into encrypted chunks.
177#[cfg(feature = "encryption")]
178pub fn split_encrypted<const BODY_SIZE: usize>(
179    data: &[u8],
180) -> error::Result<(EncryptedChunkRef, crate::store::MemoryStore<BODY_SIZE>)> {
181    let (root_ref, chunks) = EncryptedParallelSplitter::<BODY_SIZE>::split_to_vec(&data)?;
182    Ok((root_ref, crate::store::MemoryStore::from_chunks(chunks)))
183}
184
185/// Calculate tree depth for a given file size (plain mode).
186#[cfg(test)]
187pub(crate) const fn levels(length: u64, chunk_size: usize) -> usize {
188    constants::tree_depth(length, chunk_size, constants::REF_SIZE)
189}
190
191// ---- Extension traits ----
192
193/// Extension methods for async chunk getters.
194///
195/// Uses [`JoinRef`] for unified plain/encrypted dispatch.
196pub trait ChunkGetExt<const BODY_SIZE: usize>: crate::store::ChunkGet<BODY_SIZE> {
197    /// Open a file for async reading.
198    fn joiner<R: JoinRef>(
199        self,
200        root: R,
201    ) -> impl std::future::Future<Output = error::Result<GenericJoiner<Self, R::Mode, BODY_SIZE>>>
202    + MaybeSend
203    where
204        Self: Sized + Clone + MaybeSend + MaybeSync + 'static,
205    {
206        GenericJoiner::new(self, root.into_root_ref())
207    }
208
209    /// Read entire file into memory asynchronously.
210    fn read_file<R: JoinRef>(
211        self,
212        root: R,
213    ) -> impl std::future::Future<Output = error::Result<Vec<u8>>> + MaybeSend
214    where
215        Self: Sized + Clone + MaybeSend + MaybeSync + 'static,
216    {
217        join(self, root)
218    }
219}
220
221impl<T, const BODY_SIZE: usize> ChunkGetExt<BODY_SIZE> for T where
222    T: crate::store::ChunkGet<BODY_SIZE>
223{
224}
225
226/// Extension methods for chunk putters.
227///
228/// Splits data on rayon, then stores the chunks via the async [`ChunkPut`].
229/// Uses [`JoinRef`] for unified plain/encrypted dispatch on read-back.
230///
231/// ```
232/// use futures::executor::block_on;
233/// use nectar_primitives::file::{ChunkGetExt, ChunkPutExt};
234/// use nectar_primitives::store::MemoryStore;
235/// use nectar_primitives::DEFAULT_BODY_SIZE;
236///
237/// let store = MemoryStore::<DEFAULT_BODY_SIZE>::new();
238/// let addr = block_on(store.write_file(b"hello swarm".to_vec())).unwrap();
239/// let recovered = block_on(store.read_file(addr)).unwrap();
240/// assert_eq!(recovered, b"hello swarm");
241/// ```
242pub trait ChunkPutExt<const BODY_SIZE: usize>: ChunkPut<BODY_SIZE> {
243    /// Split `data` and store every produced chunk, returning the root address.
244    ///
245    /// Splitting runs on rayon up front; `data` is dropped before the first
246    /// store await, so the returned future never holds the source.
247    fn write_file<D: ReadAt + Sync>(
248        &self,
249        data: D,
250    ) -> impl std::future::Future<Output = error::Result<ChunkAddress>> + MaybeSend
251    where
252        Self: Sized + MaybeSync,
253    {
254        let split = ParallelSplitter::<BODY_SIZE>::split_to_vec(&data);
255        async move {
256            let (root, chunks) = split?;
257            for chunk in chunks {
258                self.put(chunk).await.map_err(FileError::store)?;
259            }
260            Ok(root)
261        }
262    }
263
264    /// Split `data` into encrypted chunks and store them, returning the root reference.
265    #[cfg(feature = "encryption")]
266    fn write_encrypted_file<D: ReadAt + Sync>(
267        &self,
268        data: D,
269    ) -> impl std::future::Future<Output = error::Result<EncryptedChunkRef>> + MaybeSend
270    where
271        Self: Sized + MaybeSync,
272    {
273        let split = EncryptedParallelSplitter::<BODY_SIZE>::split_to_vec(&data);
274        async move {
275            let (root_ref, chunks) = split?;
276            for chunk in chunks {
277                self.put(chunk).await.map_err(FileError::store)?;
278            }
279            Ok(root_ref)
280        }
281    }
282}
283
284impl<T, const BODY_SIZE: usize> ChunkPutExt<BODY_SIZE> for T where T: ChunkPut<BODY_SIZE> {}
285
286#[cfg(test)]
287mod tests;
nectar_primitives/file/mod.rs

nectar_primitives/file/
mod.rs
