Implement multi-store persistence (OpenDAL) with sled + dashmap.

Write resource blobs to both stores; read from fastest (benchmarked at init) with sled fallback. Add delete across stores. Keeps existing Sled indexes untouched. Preps for full Persistable pattern across profiles.

Author: AlexMikhalev

lessons-learned.md

@@ -0,0 +1,10 @@
+Lessons learned from OpenDAL integration and terraphim_persistence pattern
+
+- Consistency first: Switching only reads to OpenDAL while writes remain on Sled breaks read-after-write; dual-write or single-path is required for correctness.
+- One abstraction boundary: Use OpenDAL as the single storage interface; let Sled/DashMap/RocksDB be OpenDAL services instead of directly coupling to them.
+- Fastest-read via benchmarking: Measuring operator latency at startup and selecting the fastest improves read performance; still need write-all for durability.
+- Tokio runtime scope: Avoid constructing runtimes deep inside libraries. Expose async APIs or use appropriate blocking adaptors.
+- Migration strategy: Plan backfill and deletion symmetry. When introducing a new backend, provide tools/tests to migrate and keep stores in sync.
+- Feature gating services: Keep backend choices behind features to reduce dependency surface and compile times.
+- Key normalization: Stable, normalized keys (e.g., `document_<id>.json`) avoid cross-backend issues.
+- Testing breadth: Memory-only configs are invaluable for CI; integration tests for each optional backend help catch configurational drift.

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"] }
+opendal = { version = "0.39", features = ["services-sled", "services-dashmap"] }
 rand = {version = "0.8"}
 regex = "1"
 ring = "0.16.19"
@@ -32,6 +32,7 @@ serde = {version = "1", features = ["derive"]}
 serde_json = "1"
 sled = {version = "0.34", optional = true, features = ["no_logs"]}
 tokio = "1.29.1"
+uuid = { version = "1", features = ["v4"] }
 toml = {version = "0.7", optional = true}
 tracing = "0.1"
 ureq = "2"

lib/src/db.rs

@@ -41,6 +41,9 @@ use self::{
     val_prop_sub_index::{add_atom_to_reference_index, remove_atom_from_reference_index},
 };
 
+// External crates used in persistable multi-store setup
+use uuid;
+
 // A function called by the Store when a Commit is accepted
 type HandleCommit = Box<dyn Fn(&CommitResponse) + Send + Sync>;
 
@@ -63,8 +66,10 @@ pub struct Db {
     /// Try not to use this directly, but use the Trees.
     db: sled::Db,
     default_agent: Arc<Mutex<Option<crate::agents::Agent>>>,
-    /// Stores all resources in OpenDAL. The Key is the Subject as a `string.as_bytes()`, the value a [PropVals]. Propvals must be serialized using [bincode].
-    dal_resources: opendal::Operator,
+    /// OpenDAL operators by profile name (e.g., "sled", "dashmap") for resource blobs
+    dal_ops: std::collections::HashMap<String, opendal::Operator>,
+    /// Fastest OpenDAL operator chosen via a simple read benchmark
+    dal_fastest: opendal::Operator,
     /// Stores all resources. The Key is the Subject as a `string.as_bytes()`, the value a [PropVals]. Propvals must be serialized using [bincode].
     resources: sled::Tree,
     /// Index of all Atoms, sorted by {Value}-{Property}-{Subject}.
@@ -92,26 +97,64 @@ impl Db {
     /// The server_url is the domain where the db will be hosted, e.g. http://localhost/
     /// It is used for distinguishing locally defined items from externally defined ones.
     pub fn init(path: &std::path::Path, server_url: String) -> AtomicResult<Db> {
-        let mut dal_sled = opendal::services::Sled::default();
+        // Local runtime for async OpenDAL ops during initialization
+        let rt = tokio::runtime::Runtime::new()?;
 
+        // OpenDAL operator: Sled (on-disk)
+        let mut dal_sled = opendal::services::Sled::default();
         let dal_path = path.clone().join("opendal");
         dal_sled
             .datadir(dal_path.to_str().expect("wrong data dir string"))
             .tree("resources_v1");
+        let sled_op = opendal::Operator::new(dal_sled)
+            .map_err(|_e| format!("Error operator: {}", _e))?
+            .layer(opendal::layers::LoggingLayer::default())
+            .finish();
 
-        let dal_op = opendal::Operator::new(dal_sled)
+        // OpenDAL operator: DashMap (in-memory, fast)
+        let dash_op = opendal::Operator::new(opendal::services::Dashmap::default())
             .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, op: &opendal::Operator) -> AtomicResult<u128> {
+            use std::time::Instant;
+            rt.block_on(async {
+                let key = format!("bench_{}", uuid::Uuid::new_v4());
+                let payload = b"atomic-bench".to_vec();
+                op.write(&key, payload).await.map_err(|e| format!("bench write error: {e}"))?;
+                let start = Instant::now();
+                let _ = op.read(&key).await.map_err(|e| format!("bench read error: {e}"))?;
+                let ns = start.elapsed().as_nanos();
+                // best-effort cleanup
+                let _ = op.delete(&key).await;
+                Ok::<u128, String>(ns)
+            }).map_err(|e| e.into())
+        }
+
+        let sled_ns = measure_read_ns(&rt, &sled_op)?;
+        let dash_ns = measure_read_ns(&rt, &dash_op)?;
+
+        let (fastest_name, fastest_op) = if dash_ns <= sled_ns {
+            ("dashmap".to_string(), dash_op.clone())
+        } else {
+            ("sled".to_string(), sled_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);
+
         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))?;
         let reference_index = db.open_tree("reference_index_v1")?;
         let query_index = db.open_tree("members_index")?;
         let prop_val_sub_index = db.open_tree("prop_val_sub_index")?;
         let watched_queries = db.open_tree("watched_queries")?;
         let store = Db {
-            dal_resources: dal_op,
+            dal_ops,
+            dal_fastest: fastest_op,
             db,
             default_agent: Arc::new(Mutex::new(None)),
             resources,
@@ -122,7 +165,7 @@ impl Db {
             watched_queries,
             endpoints: default_endpoints(),
             on_commit: None,
-            runtime: Arc::new(tokio::runtime::Runtime::new()?),
+            runtime: Arc::new(rt),
         };
         migrate_maybe(&store).map(|e| format!("Error during migration of database: {:?}", e))?;
         crate::populate::populate_base_models(&store)
@@ -165,6 +208,13 @@ impl Db {
     #[instrument(skip(self))]
     fn set_propvals(&self, subject: &str, propvals: &PropVals) -> AtomicResult<()> {
         let resource_bin = bincode::serialize(propvals)?;
+        // Write to all OpenDAL operators (multi-store), then to primary Sled tree used by indexes
+        self.runtime.block_on(async {
+            for (_name, op) in self.dal_ops.iter() {
+                // Ignore per-operator errors to allow others to succeed; surface first error if needed later
+                let _ = op.write(subject, resource_bin.clone()).await;
+            }
+        });
         self.resources.insert(subject.as_bytes(), resource_bin)?;
         Ok(())
     }
@@ -179,17 +229,21 @@ impl Db {
     /// Deals with the binary API of Sled
     #[instrument(skip(self))]
     fn get_propvals(&self, subject: &str) -> AtomicResult<PropVals> {
-        let propval_maybe = self.runtime.block_on(async {
-            self.dal_resources
+        // Try fastest OpenDAL operator first
+        let mut propval_maybe = self.runtime.block_on(async {
+            self.dal_fastest
                 .read(subject)
                 .await
-                .map_err(|e| format!("Can't open {} from store: {}", subject, e))
+                .map_err(|e| format!("Can't open {} from fastest store: {}", subject, e))
                 .ok()
         });
-        // let propval_maybe = self
-        //     .resources
-        //     .get(subject.as_bytes())
-        //     .map_err(|e| format!("Can't open {} from store: {}", subject, e))?;
+        // Fallback to Sled resources tree if OpenDAL miss
+        if propval_maybe.is_none() {
+            propval_maybe = self
+                .resources
+                .get(subject.as_bytes())
+                .map_err(|e| format!("Can't open {} from sled resources: {}", subject, e))?;
+        }
         match propval_maybe.as_ref() {
             Some(binpropval) => {
                 let propval: PropVals = bincode::deserialize(binpropval).map_err(|e| {
@@ -612,6 +666,12 @@ impl Storelike for Db {
                 let remove_atom = crate::Atom::new(subject.into(), prop.clone(), val.clone());
                 self.remove_atom_from_index(&remove_atom, &resource)?;
             }
+            // Remove from OpenDAL operators (best-effort)
+            self.runtime.block_on(async {
+                for (_name, op) in self.dal_ops.iter() {
+                    let _ = op.delete(subject).await;
+                }
+            });
             let _found = self.resources.remove(subject.as_bytes())?;
         } else {
             return Err(format!(

memories.md

@@ -0,0 +1,21 @@
+OpenDAL integration review (atomic-server)
+
+- Branch: origin/opendal; commits include "WIP OpenDAL #433" and "WIP dal".
+- Changes: adds OpenDAL (services-sled) to `lib/Cargo.toml`; in `lib/src/db.rs` adds `dal_resources: opendal::Operator`, initializes it with OpenDAL Sled at `<db_path>/opendal` tree `resources_v1`, and switches `get_propvals` to read via `dal_resources.read(subject)` using a new Tokio runtime.
+- Gaps / issues:
+  - Writes still go to Sled via `set_propvals` (OpenDAL not used for write). Reads now use OpenDAL, so read-after-write breaks (OpenDAL store is empty).
+  - Deletion still removes from Sled tree; OpenDAL items (if any) won’t be deleted.
+  - New Tokio runtime inside the DB struct risks nested runtimes and increases complexity; prefer async surface or OpenDAL blocking API.
+  - Parallel index/other code still relies on Sled trees; storage is now split across two backends.
+- Recommendation: unify I/O through OpenDAL (including Sled via `services-sled`) OR dual-write until migration completes. Remove in-DB runtime; make read path async or use blocking layer. Add migration/backfill from Sled to OpenDAL and tests for CRUD consistency.
+
+Multi-store pattern (terraphim-ai/terraphim_persistence)
+
+- `Persistable` trait: save to all profiles, load from fastest. Key derived by implementor (e.g., `document_<id>.json`).
+- `settings::parse_profiles`: builds OpenDAL operators for profiles (memory, dashmap, rocksdb, redb, sqlite, s3, atomicserver, etc.), benchmarks read latency and picks fastest.
+- Implementations provided for `Thesaurus` and `Document`; includes memory-only helpers and tests for memory/rocksdb/redb/sqlite.
+- This matches the intended architecture: write-multiplex + read-from-fastest.
+
+Action items alignment
+
+- Port the terraphim pattern to atomic-server: single abstraction via OpenDAL; write to all configured profiles (or at least primary+replica), read from fastest; ensure index/storage consistency and clear migration path.

scratchpad.md

@@ -0,0 +1,29 @@
+Atomic-server OpenDAL PR review checklist
+
+- Code changes
+  - [x] Added opendal dep w/ services-sled
+  - [x] Added `dal_resources: Operator` to Db
+  - [x] Init OpenDAL Sled under `<db>/opendal` tree `resources_v1`
+  - [x] Read path switched to `dal_resources.read(subject)`
+  - [ ] Write path still uses Sled `resources.insert`
+  - [ ] Delete path still uses Sled remove
+  - [ ] Indexing and queries use Sled trees only
+  - [ ] Concurrency/runtime: embedded tokio runtime introduced
+
+- Risks
+  - Read-after-write inconsistency (OpenDAL store empty)
+  - Double storage cost; unclear source of truth
+  - Migration undefined; no backfill from existing sled tree
+  - Runtime nesting problems in servers already using Tokio
+
+- Proposed plan
+  1) Decide strategy: a) OpenDAL-only with `services-sled` as one profile; or b) dual-write with single-read (fastest) via profiles like terraphim.
+  2) If (a): wrap all CRUD in OpenDAL; drop direct sled access for propvals; use sled only for indexes until ported.
+  3) If (b): implement `save_to_all` concept in atomic Db for propvals (write to N stores); pick fastest for read; keep indexes coherent on one canonical store.
+  4) Remove embedded runtime; move read calls to async surface or blocking shim.
+  5) Add migration: backfill OpenDAL from Sled existing items. One-time tool or lazy-on-read write-through.
+  6) Tests: CRUD, consistency across stores, performance benchmark selecting fastest profile.
+
+- Notes
+  - terraphim-persistence crate already models profiles + speed test + save_to_all + read_fastest. Could reuse patterns and helper code or even the crate.
+  - Consider features for memory/dashmap/rocksdb/redb/sqlite - align with atomic-server env.