Add redb, rocksdb and fs OpenDAL services to persistable layer; tests and benches.

- Initialize optional operators behind features and include in dual-write/read-fastest selection
- Test that all configured profiles receive blobs and are enumerated
- Add Criterion benches for op write/read per operator and key paths

Author: AlexMikhalev

lib/Cargo.toml

@@ -22,7 +22,7 @@ directories = {version = ">= 2, < 5", optional = true}
 html2md = {version = "0.2.14", optional = true}
 kuchiki = {version = "0.8.1", optional = true}
 lol_html = {version = "0.3.1", optional = true}
-opendal = { version = "0.39", features = ["services-sled", "services-dashmap"] }
+opendal = { version = "0.39", features = ["services-sled", "services-dashmap", "services-rocksdb", "services-redb", "services-fs"] }
 rand = {version = "0.8"}
 regex = "1"
 ring = "0.16.19"
@@ -50,3 +50,6 @@ config = ["directories", "toml"]
 db = ["sled", "bincode"]
 html = ["kuchiki", "lol_html", "html2md"]
 rdf = ["rio_api", "rio_turtle"]
+persist-rocksdb = ["opendal/services-rocksdb"]
+persist-redb = ["opendal/services-redb"]
+persist-fs = ["opendal/services-fs"]

lib/benches/benchmarks.rs

@@ -4,7 +4,8 @@
 
 use atomic_lib::utils::random_string;
 use atomic_lib::*;
-use criterion::{criterion_group, criterion_main, Criterion};
+use criterion::{black_box, criterion_group, criterion_main, Criterion};
+use std::time::Duration;
 
 fn random_atom() -> Atom {
     Atom::new(
@@ -21,17 +22,21 @@ fn random_resource(atom: &Atom) -> Resource {
 }
 
 fn criterion_benchmark(c: &mut Criterion) {
+    let mut g = c.benchmark_group("persistable");
+    g.sample_size(20);
+    g.measurement_time(Duration::from_secs(10));
+
     let store = Db::init_temp("bench").unwrap();
 
-    c.bench_function("add_atom_to_index", |b| {
+    g.bench_function("add_atom_to_index", |b| {
         b.iter(|| {
             let atom = random_atom();
             let resource = random_resource(&random_atom());
             store.add_atom_to_index(&atom, &resource).unwrap();
         })
     });
 
-    c.bench_function("add_resource", |b| {
+    g.bench_function("add_resource", |b| {
         b.iter(|| {
             let resource = random_resource(&random_atom());
             store
@@ -40,7 +45,7 @@ fn criterion_benchmark(c: &mut Criterion) {
         })
     });
 
-    c.bench_function("resource.save()", |b| {
+    g.bench_function("resource.save()", |b| {
         b.iter(|| {
             let mut resource = random_resource(&random_atom());
             resource.save(&store).unwrap();
@@ -51,35 +56,55 @@ fn criterion_benchmark(c: &mut Criterion) {
         .get_resource_extended("https://localhost/collections", false, None)
         .unwrap();
 
-    c.bench_function("resource.to_json_ad()", |b| {
+    g.bench_function("resource.to_json_ad()", |b| {
         b.iter(|| {
             big_resource.to_json_ad().unwrap();
         })
     });
 
-    c.bench_function("resource.to_json_ld()", |b| {
+    g.bench_function("resource.to_json_ld()", |b| {
         b.iter(|| {
             big_resource.to_json_ld(&store).unwrap();
         })
     });
 
-    c.bench_function("resource.to_json()", |b| {
+    g.bench_function("resource.to_json()", |b| {
         b.iter(|| {
             big_resource.to_json(&store).unwrap();
         })
     });
 
-    c.bench_function("resource.to_n_triples()", |b| {
+    g.bench_function("resource.to_n_triples()", |b| {
         b.iter(|| {
             big_resource.to_n_triples(&store).unwrap();
         })
     });
 
-    c.bench_function("all_resources()", |b| {
+    g.bench_function("all_resources()", |b| {
         b.iter(|| {
-            let _all = store.all_resources(false).collect::<Vec<Resource>>();
+            let _all = black_box(store.all_resources(false).collect::<Vec<Resource>>());
         })
     });
+
+    // Persistable operator benchmarks: write/read single blob via each configured operator
+    for (name, op) in store.dal_ops.iter() {
+        let key = format!("bench_{}", name);
+        let data = vec![0u8; 16 * 1024];
+        g.bench_function(&format!("op_write_{}", name), |b| {
+            b.iter(|| {
+                // We ignore errors because operators may share keys across iterations
+                let _ = store
+                    .runtime
+                    .block_on(async { op.write(&key, data.clone()).await });
+            })
+        });
+        g.bench_function(&format!("op_read_{}", name), |b| {
+            b.iter(|| {
+                let _ = store.runtime.block_on(async { op.read(&key).await });
+            })
+        });
+    }
+    g.finish();
 }
 
 criterion_group!(benches, criterion_benchmark);

lib/src/db.rs

@@ -117,6 +117,39 @@ impl Db {
             .layer(opendal::layers::LoggingLayer::default())
             .finish();
 
+        // OpenDAL operator: RocksDB (on-disk kv)
+        #[cfg(feature = "persist-rocksdb")]
+        let rocks_op = {
+            let mut b = opendal::services::Rocksdb::default();
+            b.datadir(path.join("opendal_rocksdb").to_str().expect("rocks path"));
+            opendal::Operator::new(b)
+                .map_err(|_e| format!("Error operator: {}", _e))?
+                .layer(opendal::layers::LoggingLayer::default())
+                .finish()
+        };
+
+        // OpenDAL operator: ReDB (embedded db)
+        #[cfg(feature = "persist-redb")]
+        let redb_op = {
+            let mut b = opendal::services::Redb::default();
+            b.datadir(path.join("opendal_redb").to_str().expect("redb path"));
+            opendal::Operator::new(b)
+                .map_err(|_e| format!("Error operator: {}", _e))?
+                .layer(opendal::layers::LoggingLayer::default())
+                .finish()
+        };
+
+        // OpenDAL operator: FS (filesystem flat files)
+        #[cfg(feature = "persist-fs")]
+        let fs_op = {
+            let mut b = opendal::services::Fs::default();
+            b.root(path.join("opendal_fs").to_str().expect("fs path"));
+            opendal::Operator::new(b)
+                .map_err(|_e| format!("Error operator: {}", _e))?
+                .layer(opendal::layers::LoggingLayer::default())
+                .finish()
+        };
+
         // Simple speed check: write+read small payload and measure read latency
         fn measure_read_ns(
             rt: &tokio::runtime::Runtime,
@@ -144,16 +177,40 @@ impl Db {
 
         let sled_ns = measure_read_ns(&rt, &sled_op)?;
         let dash_ns = measure_read_ns(&rt, &dash_op)?;
-
-        let fastest_op = if dash_ns <= sled_ns {
+        #[cfg(feature = "persist-rocksdb")]
+        let rocks_ns = measure_read_ns(&rt, &rocks_op)?;
+        #[cfg(feature = "persist-redb")]
+        let redb_ns = measure_read_ns(&rt, &redb_op)?;
+        #[cfg(feature = "persist-fs")]
+        let fs_ns = measure_read_ns(&rt, &fs_op)?;
+
+        let mut fastest_op = if dash_ns <= sled_ns {
             dash_op.clone()
         } else {
             sled_op.clone()
         };
+        #[cfg(feature = "persist-rocksdb")]
+        if rocks_ns < measure_read_ns(&rt, &fastest_op)? {
+            fastest_op = rocks_op.clone();
+        }
+        #[cfg(feature = "persist-redb")]
+        if redb_ns < measure_read_ns(&rt, &fastest_op)? {
+            fastest_op = redb_op.clone();
+        }
+        #[cfg(feature = "persist-fs")]
+        if fs_ns < measure_read_ns(&rt, &fastest_op)? {
+            fastest_op = fs_op.clone();
+        }
 
         let mut dal_ops = std::collections::HashMap::new();
         dal_ops.insert("sled".to_string(), sled_op);
         dal_ops.insert("dashmap".to_string(), dash_op);
+        #[cfg(feature = "persist-rocksdb")]
+        dal_ops.insert("rocksdb".to_string(), rocks_op);
+        #[cfg(feature = "persist-redb")]
+        dal_ops.insert("redb".to_string(), redb_op);
+        #[cfg(feature = "persist-fs")]
+        dal_ops.insert("fs".to_string(), fs_op);
 
         let db = sled::open(path).map_err(|e|format!("Failed opening DB at this location: {:?} . Is another instance of Atomic Server running? {}", path, e))?;
         let resources = db.open_tree("resources_v1").map_err(|e|format!("Failed building resources. Your DB might be corrupt. Go back to a previous version and export your data. {}", e))?;

lib/src/db/test.rs

@@ -434,17 +434,19 @@ fn persistable_multi_store_dual_write_and_fallback() {
     r.save(&store).unwrap();
     let subject = r.get_subject().to_string();
 
-    // Both OpenDAL operators should have the blob
-    let dash = store.dal_ops.get("dashmap").expect("dashmap op");
-    let sled = store.dal_ops.get("sled").expect("sled op");
-    let dash_bytes = store
-        .runtime
-        .block_on(async { dash.read(&subject).await.expect("dash read") });
-    let sled_bytes = store
-        .runtime
-        .block_on(async { sled.read(&subject).await.expect("sled read") });
-    assert!(dash_bytes.len() > 0);
-    assert_eq!(dash_bytes, sled_bytes, "stores should contain same bytes");
+    // All configured OpenDAL operators should have the blob
+    let mut blobs: Vec<Vec<u8>> = Vec::new();
+    for (name, op) in store.dal_ops.iter() {
+        let bytes = store
+            .runtime
+            .block_on(async { op.read(&subject).await.expect("op read") });
+        assert!(bytes.len() > 0, "{} empty", name);
+        blobs.push(bytes);
+    }
+    // Compare all blobs to the first
+    for b in blobs.iter() {
+        assert_eq!(b, &blobs[0], "stores should contain same bytes");
+    }
 
     // Delete from the fastest store, ensure fallback (sled tree) still serves resource
     store
@@ -453,7 +455,7 @@ fn persistable_multi_store_dual_write_and_fallback() {
         .ok();
     // Should still be able to fetch via sled resources fallback
     let fetched = store.get_resource(&subject).expect("fallback should work");
-    assert_eq!(fetched.get_subject(), subject);
+    assert_eq!(fetched.get_subject(), &subject);
 
     // Remove from sled resources tree and ensure not found now
     store
@@ -506,6 +508,21 @@ fn persistable_collections_still_work() {
     assert!(res.count >= 8, "count should include all members");
 }
 
+#[test]
+fn persistable_profiles_available_and_benchmarked() {
+    let store = Db::init_temp("persistable_profiles").unwrap();
+    // Expect core profiles
+    assert!(store.dal_ops.contains_key("sled"));
+    assert!(store.dal_ops.contains_key("dashmap"));
+    // Optional ones if features enabled
+    #[cfg(feature = "persist-rocksdb")]
+    assert!(store.dal_ops.contains_key("rocksdb"));
+    #[cfg(feature = "persist-redb")]
+    assert!(store.dal_ops.contains_key("redb"));
+    #[cfg(feature = "persist-fs")]
+    assert!(store.dal_ops.contains_key("fs"));
+}
+
 #[test]
 /// Changing these values actually correctly updates the index.
 fn index_invalidate_cache() {