Skip to content
This repository was archived by the owner on Sep 21, 2024. It is now read-only.

Commit 4facbd9

Browse files
jsantelljsantell
committed
feat: Introduce Importable/Exportable Storage traits.
1 parent 7155f86 commit 4facbd9

File tree

7 files changed

+217
-36
lines changed

7 files changed

+217
-36
lines changed

‎rust/noosphere-storage/examples/bench/main.rs

+1-3
Original file line numberDiff line numberDiff line change
@@ -128,9 +128,7 @@ impl BenchmarkStorage {
128128
))]
129129
let (storage, storage_name) = {
130130
(
131-
noosphere_storage::SledStorage::new(noosphere_storage::SledStorageInit::Path(
132-
storage_path.into(),
133-
))?,
131+
noosphere_storage::SledStorage::new(storage_path.into())?,
134132
"SledDbStorage",
135133
)
136134
};

‎rust/noosphere-storage/src/exportable.rs

+176
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,176 @@
1+
use crate::{Storage, Store, SPHERE_DB_STORE_NAMES};
2+
use anyhow::Result;
3+
use async_trait::async_trait;
4+
use noosphere_common::ConditionalSync;
5+
use std::pin::Pin;
6+
use tokio_stream::{Stream, StreamExt};
7+
8+
#[cfg(target_arch = "wasm32")]
9+
pub type IterableStream<'a> = Pin<Box<dyn Stream<Item = Result<(Vec<u8>, Option<Vec<u8>>)>> + 'a>>;
10+
#[cfg(not(target_arch = "wasm32"))]
11+
pub type IterableStream<'a> =
12+
Pin<Box<dyn Stream<Item = Result<(Vec<u8>, Option<Vec<u8>>)>> + Send + 'a>>;
13+
14+
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
15+
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
16+
pub trait Iterable: Store {
17+
fn get_all_entries<'a>(&'a self) -> IterableStream<'a>;
18+
}
19+
20+
/// A blanket implementation for [Storage]s that can be
21+
/// imported via [Importable::import].
22+
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
23+
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
24+
pub trait Exportable
25+
where
26+
Self: Storage,
27+
<Self as Storage>::KeyValueStore: Iterable,
28+
{
29+
async fn get_all_store_names(&self) -> Result<Vec<String>> {
30+
let mut names = vec![];
31+
names.extend(SPHERE_DB_STORE_NAMES.iter().map(|name| String::from(*name)));
32+
Ok(names)
33+
}
34+
}
35+
36+
impl<S> Exportable for S
37+
where
38+
S: Storage,
39+
S::KeyValueStore: Iterable,
40+
{
41+
}
42+
43+
/// A blanket implementation for all [Storage]s to import
44+
/// an [Exportable] storage.
45+
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
46+
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
47+
pub trait Importable<'a, E>
48+
where
49+
Self: Storage,
50+
Self::KeyValueStore: Store,
51+
E: Exportable + ConditionalSync + 'a,
52+
<E as Storage>::KeyValueStore: Iterable,
53+
{
54+
async fn import(&'a mut self, exportable: E) -> Result<()> {
55+
for store_name in exportable.get_all_store_names().await? {
56+
let mut store = self.get_key_value_store(&store_name).await?;
57+
let export_store = exportable.get_key_value_store(&store_name).await?;
58+
{
59+
let mut stream = export_store.get_all_entries();
60+
while let Some((key, value)) = stream.try_next().await? {
61+
if let Some(value) = value {
62+
Store::write(&mut store, key.as_ref(), value.as_ref()).await?;
63+
}
64+
}
65+
}
66+
drop(export_store)
67+
}
68+
Ok(())
69+
}
70+
}
71+
72+
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
73+
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
74+
impl<'a, T, E> Importable<'a, E> for T
75+
where
76+
T: Storage,
77+
T::KeyValueStore: Store,
78+
E: Exportable + ConditionalSync + 'a,
79+
<E as Storage>::KeyValueStore: Iterable,
80+
{
81+
}
82+
83+
#[cfg(test)]
84+
mod test {
85+
use super::*;
86+
use std::path::PathBuf;
87+
88+
#[cfg(target_arch = "wasm32")]
89+
use wasm_bindgen_test::{wasm_bindgen_test, wasm_bindgen_test_configure};
90+
#[cfg(target_arch = "wasm32")]
91+
wasm_bindgen_test_configure!(run_in_browser);
92+
93+
struct TempPath {
94+
#[cfg(target_arch = "wasm32")]
95+
pub path: String,
96+
#[cfg(not(target_arch = "wasm32"))]
97+
pub path: PathBuf,
98+
#[cfg(not(target_arch = "wasm32"))]
99+
#[allow(unused)]
100+
temp_dir: tempfile::TempDir,
101+
}
102+
103+
impl TempPath {
104+
pub fn new() -> Result<Self> {
105+
#[cfg(target_arch = "wasm32")]
106+
let path: String = witty_phrase_generator::WPGen::new()
107+
.with_words(3)
108+
.unwrap()
109+
.into_iter()
110+
.map(|word| String::from(word))
111+
.collect();
112+
113+
#[cfg(not(target_arch = "wasm32"))]
114+
let temp_dir = tempfile::TempDir::new()?;
115+
#[cfg(not(target_arch = "wasm32"))]
116+
let path = temp_dir.path().to_owned();
117+
118+
#[cfg(target_arch = "wasm32")]
119+
let obj = Self { path };
120+
#[cfg(not(target_arch = "wasm32"))]
121+
let obj = Self { temp_dir, path };
122+
123+
Ok(obj)
124+
}
125+
}
126+
127+
/// wasm32: MemoryStorage -> IndexedDbStorage
128+
/// native: SledStorage -> MemoryStorage
129+
/// native+rocks: SledStorage -> RocksDbStorage
130+
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test)]
131+
#[cfg_attr(not(target_arch = "wasm32"), tokio::test)]
132+
pub async fn it_can_migrate_to_new_storage_backend() -> Result<()> {
133+
noosphere_core::tracing::initialize_tracing(None);
134+
135+
#[allow(unused)]
136+
let from_temp_path = TempPath::new()?;
137+
#[allow(unused)]
138+
let to_temp_path = TempPath::new()?;
139+
140+
#[cfg(target_arch = "wasm32")]
141+
let from_storage = crate::MemoryStorage::default();
142+
#[cfg(not(target_arch = "wasm32"))]
143+
let from_storage = crate::SledStorage::new(from_temp_path.path.clone())?;
144+
145+
#[cfg(target_arch = "wasm32")]
146+
let mut to_storage = crate::IndexedDbStorage::new(&to_temp_path.path).await?;
147+
#[cfg(all(feature = "rocksdb", not(target_arch = "wasm32")))]
148+
let mut to_storage = crate::RocksDbStorage::new(to_temp_path.path.clone())?;
149+
#[cfg(all(not(feature = "rocksdb"), not(target_arch = "wasm32")))]
150+
let mut to_storage = crate::MemoryStorage::default();
151+
152+
{
153+
let mut store = from_storage.get_key_value_store("links").await?;
154+
for n in 0..10 {
155+
let bytes = vec![n; 10];
156+
store.write(&[n], bytes.as_ref()).await?;
157+
}
158+
}
159+
160+
to_storage.import(from_storage).await?;
161+
162+
{
163+
let store = to_storage.get_key_value_store("links").await?;
164+
for n in 0..10 {
165+
let expected_bytes = vec![n; 10];
166+
167+
if let Some(bytes) = store.read(&[n]).await? {
168+
assert_eq!(bytes, expected_bytes);
169+
} else {
170+
panic!("Expected key `{n}` to exist in new db");
171+
}
172+
}
173+
}
174+
Ok(())
175+
}
176+
}

‎rust/noosphere-storage/src/helpers.rs

+2-2
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@ use crate::Storage;
22
use anyhow::Result;
33

44
#[cfg(not(target_arch = "wasm32"))]
5-
use crate::{SledStorage, SledStorageInit, SledStore};
5+
use crate::{SledStorage, SledStore};
66

77
#[cfg(not(target_arch = "wasm32"))]
88
pub async fn make_disposable_store() -> Result<SledStore> {
@@ -13,7 +13,7 @@ pub async fn make_disposable_store() -> Result<SledStore> {
1313
.into_iter()
1414
.map(String::from)
1515
.collect();
16-
let provider = SledStorage::new(SledStorageInit::Path(temp_dir.join(temp_name)))?;
16+
let provider = SledStorage::new(temp_dir.join(temp_name))?;
1717
provider.get_block_store("foo").await
1818
}
1919

‎rust/noosphere-storage/src/implementation/memory.rs

+14
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
use anyhow::{anyhow, Result};
2+
use async_stream::try_stream;
23
use async_trait::async_trait;
34
use cid::Cid;
45
use std::{collections::HashMap, sync::Arc};
@@ -131,6 +132,19 @@ impl Store for MemoryStore {
131132
}
132133
}
133134

135+
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
136+
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
137+
impl crate::Iterable for MemoryStore {
138+
fn get_all_entries<'a>(&'a self) -> crate::IterableStream<'a> {
139+
Box::pin(try_stream! {
140+
let dags = self.entries.lock().await;
141+
for key in dags.keys() {
142+
yield (key.to_owned(), dags.get(key).cloned());
143+
}
144+
})
145+
}
146+
}
147+
134148
#[cfg(feature = "performance")]
135149
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
136150
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]

‎rust/noosphere-storage/src/implementation/sled.rs

+18-23
Original file line numberDiff line numberDiff line change
@@ -4,32 +4,22 @@ use crate::storage::Storage;
44
use crate::store::Store;
55

66
use anyhow::Result;
7+
use async_stream::try_stream;
78
use async_trait::async_trait;
89
use sled::{Db, Tree};
910

10-
pub enum SledStorageInit {
11-
Path(PathBuf),
12-
Db(Db),
13-
}
14-
1511
#[derive(Clone, Debug)]
1612
pub struct SledStorage {
1713
db: Db,
1814
#[allow(unused)]
19-
path: Option<PathBuf>,
15+
path: PathBuf,
2016
}
2117

2218
impl SledStorage {
23-
pub fn new(init: SledStorageInit) -> Result<Self> {
24-
let mut db_path = None;
25-
let db: Db = match init {
26-
SledStorageInit::Path(path) => {
27-
std::fs::create_dir_all(&path)?;
28-
db_path = Some(path.clone().canonicalize()?);
29-
sled::open(path)?
30-
}
31-
SledStorageInit::Db(db) => db,
32-
};
19+
pub fn new(path: PathBuf) -> Result<Self> {
20+
std::fs::create_dir_all(&path)?;
21+
let db_path = path.canonicalize()?;
22+
let db = sled::open(&db_path)?;
3323

3424
Ok(SledStorage { db, path: db_path })
3525
}
@@ -104,16 +94,21 @@ impl Drop for SledStorage {
10494
}
10595
}
10696

97+
impl crate::Iterable for SledStore {
98+
fn get_all_entries<'a>(&'a self) -> crate::IterableStream<'a> {
99+
Box::pin(try_stream! {
100+
for entry in self.db.iter() {
101+
let (key, value) = entry?;
102+
yield (Vec::from(key.as_ref()), Some(Vec::from(value.as_ref())));
103+
}
104+
})
105+
}
106+
}
107+
107108
#[cfg(feature = "performance")]
108109
#[async_trait]
109110
impl crate::Space for SledStorage {
110111
async fn get_space_usage(&self) -> Result<u64> {
111-
if let Some(path) = &self.path {
112-
crate::get_dir_size(path).await
113-
} else {
114-
Err(anyhow::anyhow!(
115-
"Could not calculate storage space, requires usage of a path constructor."
116-
))
117-
}
112+
crate::get_dir_size(&self.path).await
118113
}
119114
}

‎rust/noosphere-storage/src/lib.rs

+2
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ mod key_value;
1212

1313
mod db;
1414
mod encoding;
15+
mod exportable;
1516
mod retry;
1617
mod storage;
1718
mod store;
@@ -22,6 +23,7 @@ pub use crate::ucan::*;
2223
pub use block::*;
2324
pub use db::*;
2425
pub use encoding::*;
26+
pub use exportable::*;
2527
pub use implementation::*;
2628
pub use key_value::*;
2729
pub use retry::*;

‎rust/noosphere/src/storage.rs

+4-8
Original file line numberDiff line numberDiff line change
@@ -45,15 +45,11 @@ impl From<StorageLayout> for PathBuf {
4545
impl StorageLayout {
4646
pub async fn to_storage(&self) -> Result<PrimitiveStorage> {
4747
#[cfg(sled)]
48-
{
49-
noosphere_storage::SledStorage::new(noosphere_storage::SledStorageInit::Path(
50-
PathBuf::from(self),
51-
))
52-
}
48+
let storage = noosphere_storage::SledStorage::new(PathBuf::from(self));
5349
#[cfg(rocksdb)]
54-
{
55-
noosphere_storage::RocksDbStorage::new(PathBuf::from(self))
56-
}
50+
let storage = noosphere_storage::RocksDbStorage::new(PathBuf::from(self));
51+
52+
storage
5753
}
5854
}
5955

0 commit comments

Comments
 (0)