Skip to content

Commit 0aff1dd

Browse files
committed
feat(crdt): add movable-list ops and a declared crdt collection flag
Add list_insert/list_delete/list_move dispatch backed by the existing CRDT list executor, a set_fields partial-merge write for UPDATE SET semantics on CRDT document rows, and a declared `crdt` flag on CollectionMeta so `WITH (crdt=true)` collections announce their DML routing to Origin.
1 parent 6b5b675 commit 0aff1dd

16 files changed

Lines changed: 663 additions & 3 deletions

File tree

nodedb-lite/src/engine/crdt/engine.rs

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -173,6 +173,46 @@ impl CrdtEngine {
173173
Ok(mutation_id)
174174
}
175175

176+
/// Partial-merge write: set exactly the provided scalar fields on a row,
177+
/// leaving untouched keys intact.
178+
///
179+
/// This is `upsert` without its full-projection prune — the UPDATE SET
180+
/// semantic behind `CrdtOp::DocUpsert { partial: true }`. Delta export and
181+
/// pending-sync accounting are identical to `upsert`.
182+
pub fn set_fields(
183+
&mut self,
184+
collection: &str,
185+
doc_id: &str,
186+
fields: &[(&str, LoroValue)],
187+
) -> Result<u64, LiteError> {
188+
let version_before = self.state.doc().oplog_vv();
189+
190+
self.state
191+
.set_fields(collection, doc_id, fields)
192+
.map_err(|e| LiteError::Storage {
193+
detail: format!("CRDT set_fields failed: {e}"),
194+
})?;
195+
196+
let delta_bytes = self
197+
.state
198+
.doc()
199+
.export(loro::ExportMode::updates(&version_before))
200+
.map_err(|e| LiteError::Storage {
201+
detail: format!("delta export failed: {e}"),
202+
})?;
203+
204+
let mutation_id = self.next_mutation_id.fetch_add(1, Ordering::Relaxed);
205+
self.pending_deltas.push(PendingDelta {
206+
mutation_id,
207+
collection: collection.to_string(),
208+
document_id: doc_id.to_string(),
209+
delta_bytes,
210+
seq: 0,
211+
});
212+
213+
Ok(mutation_id)
214+
}
215+
176216
/// Delete a document/row.
177217
pub fn delete(&mut self, collection: &str, doc_id: &str) -> Result<u64, LiteError> {
178218
let version_before = self.state.doc().oplog_vv();

nodedb-lite/src/nodedb/collection/ddl.rs

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,14 @@ pub struct CollectionMeta {
2727
/// Whether the collection tracks system-time + valid-time versions.
2828
#[serde(default)]
2929
pub bitemporal: bool,
30+
/// Whether the collection was declared with `WITH (crdt=true)`, i.e. its
31+
/// DML is routed through the CRDT (Loro) path rather than the plain
32+
/// document path. Announced to Origin in the collection descriptor so the
33+
/// peer applies the same routing. A declared property, not an inferred one:
34+
/// Lite's schemaless store happens to be Loro-backed, but that alone does
35+
/// NOT make a collection CRDT-declared.
36+
#[serde(default)]
37+
pub crdt: bool,
3038
}
3139

3240
impl<S: StorageEngine> NodeDbLite<S> {
@@ -47,6 +55,7 @@ impl<S: StorageEngine> NodeDbLite<S> {
4755
config_json: None,
4856
descriptor_json: None,
4957
bitemporal: false,
58+
crdt: false,
5059
};
5160
let key = format!("collection:{name}");
5261
let bytes = sonic_rs::to_vec(&meta).map_err(|e| NodeDbError::storage(e.to_string()))?;
@@ -83,6 +92,7 @@ impl<S: StorageEngine> NodeDbLite<S> {
8392
config_json: Some(config_json),
8493
descriptor_json: None,
8594
bitemporal: false,
95+
crdt: false,
8696
};
8797
let key = format!("collection:{name}");
8898
let bytes = sonic_rs::to_vec(&meta).map_err(|e| NodeDbError::storage(e.to_string()))?;
@@ -120,6 +130,7 @@ impl<S: StorageEngine> NodeDbLite<S> {
120130
config_json: None,
121131
descriptor_json: None,
122132
bitemporal: false,
133+
crdt: false,
123134
})
124135
}
125136

@@ -175,6 +186,7 @@ impl<S: StorageEngine> NodeDbLite<S> {
175186
config_json: None,
176187
descriptor_json: None,
177188
bitemporal: false,
189+
crdt: false,
178190
});
179191
}
180192
}

nodedb-lite/src/nodedb/collection/import.rs

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -155,7 +155,10 @@ impl<S: StorageEngine> NodeDbLite<S> {
155155
}
156156

157157
/// Convert serde_json::Value to LoroValue for COPY FROM import.
158-
fn json_to_loro(v: &serde_json::Value) -> loro::LoroValue {
158+
///
159+
/// Also used by the CRDT document-row ops, which receive their fields as a
160+
/// Control-Plane-built JSON object.
161+
pub(crate) fn json_to_loro(v: &serde_json::Value) -> loro::LoroValue {
159162
match v {
160163
serde_json::Value::Null => loro::LoroValue::Null,
161164
serde_json::Value::Bool(b) => loro::LoroValue::Bool(*b),

nodedb-lite/src/nodedb/sync_delegate/import_collection_schema.rs

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,7 @@ impl<S: StorageEngine> NodeDbLite<S> {
8585
config_json,
8686
descriptor_json: Some(descriptor_json),
8787
bitemporal: descriptor.bitemporal,
88+
crdt: descriptor.crdt,
8889
};
8990
let bytes = sonic_rs::to_vec(&meta).map_err(|e| NodeDbError::storage(e.to_string()))?;
9091
self.storage
@@ -117,6 +118,7 @@ mod tests {
117118
name: name.into(),
118119
collection_type: ct,
119120
bitemporal,
121+
crdt: false,
120122
fields: vec![("email".into(), "TEXT".into())],
121123
primary: PrimaryEngine::Document,
122124
vector_primary: None,
Lines changed: 131 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,131 @@
1+
// SPDX-License-Identifier: Apache-2.0
2+
3+
//! CRDT movable-list operation helpers for `NodeDbLite`.
4+
//!
5+
//! Reuses the already-wired execution helpers in
6+
//! `crate::query::crdt_ops::list` — the same handlers the native-protocol
7+
//! dispatch path (`query::physical_visitor::adapter::crdt`) drives for
8+
//! `PhysicalPlan::Crdt(CrdtOp::List*)` — instead of duplicating the Loro
9+
//! movable-list logic here.
10+
11+
use std::collections::HashMap;
12+
13+
use nodedb_types::error::{NodeDbError, NodeDbResult};
14+
use nodedb_types::value::Value;
15+
16+
use crate::nodedb::NodeDbLite;
17+
use crate::query::crdt_ops::list;
18+
use crate::storage::engine::StorageEngine;
19+
20+
/// Extract the field map from `fields`, rejecting anything that is not a
21+
/// JSON-object-shaped `Value::Object`.
22+
///
23+
/// `handle_list_insert` treats `fields_json` as a JSON object whose
24+
/// top-level keys become fields on the new list block; a scalar or array
25+
/// here would silently produce a malformed block.
26+
fn require_object_fields(fields: &Value) -> NodeDbResult<&HashMap<String, Value>> {
27+
match fields {
28+
Value::Object(obj) => Ok(obj),
29+
other => Err(NodeDbError::bad_request(format!(
30+
"list_insert: fields must be Value::Object, got {other:?}"
31+
))),
32+
}
33+
}
34+
35+
impl<S: StorageEngine> NodeDbLite<S> {
36+
pub(super) async fn list_insert_impl(
37+
&self,
38+
collection: &str,
39+
document_id: &str,
40+
list_path: &str,
41+
index: usize,
42+
fields: &Value,
43+
) -> NodeDbResult<()> {
44+
let obj = require_object_fields(fields)?;
45+
let fields_json = sonic_rs::to_string(obj)
46+
.map_err(|e| NodeDbError::serialization("json", format!("list_insert fields: {e}")))?;
47+
48+
list::handle_list_insert(
49+
&self.query_engine,
50+
collection,
51+
document_id,
52+
list_path,
53+
index,
54+
&fields_json,
55+
)
56+
.await?;
57+
Ok(())
58+
}
59+
60+
pub(super) async fn list_delete_impl(
61+
&self,
62+
collection: &str,
63+
document_id: &str,
64+
list_path: &str,
65+
index: usize,
66+
) -> NodeDbResult<()> {
67+
list::handle_list_delete(
68+
&self.query_engine,
69+
collection,
70+
document_id,
71+
list_path,
72+
index,
73+
)
74+
.await?;
75+
Ok(())
76+
}
77+
78+
pub(super) async fn list_move_impl(
79+
&self,
80+
collection: &str,
81+
document_id: &str,
82+
list_path: &str,
83+
from_index: usize,
84+
to_index: usize,
85+
) -> NodeDbResult<()> {
86+
list::handle_list_move(
87+
&self.query_engine,
88+
collection,
89+
document_id,
90+
list_path,
91+
from_index,
92+
to_index,
93+
)
94+
.await?;
95+
Ok(())
96+
}
97+
}
98+
99+
#[cfg(test)]
100+
mod tests {
101+
use super::*;
102+
103+
fn obj_fields() -> Value {
104+
let mut map = HashMap::new();
105+
map.insert("title".to_string(), Value::String("hello".to_string()));
106+
map.insert("done".to_string(), Value::Bool(false));
107+
Value::Object(map)
108+
}
109+
110+
#[test]
111+
fn require_object_fields_accepts_object() {
112+
let fields = obj_fields();
113+
let obj = require_object_fields(&fields).expect("object fields must pass");
114+
assert_eq!(obj.get("title"), Some(&Value::String("hello".to_string())));
115+
assert_eq!(obj.get("done"), Some(&Value::Bool(false)));
116+
}
117+
118+
#[test]
119+
fn require_object_fields_rejects_scalar() {
120+
let err =
121+
require_object_fields(&Value::Integer(1)).expect_err("scalar fields must be rejected");
122+
assert!(format!("{err}").to_lowercase().contains("object"));
123+
}
124+
125+
#[test]
126+
fn require_object_fields_rejects_array() {
127+
let err = require_object_fields(&Value::Array(vec![Value::Integer(1)]))
128+
.expect_err("array fields must be rejected");
129+
assert!(format!("{err}").to_lowercase().contains("object"));
130+
}
131+
}

nodedb-lite/src/nodedb/trait_impl/dispatch.rs

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -164,6 +164,43 @@ impl<S: StorageEngine> NodeDb for NodeDbLite<S> {
164164
.await
165165
}
166166

167+
// ─── CRDT List Operations (Movable List) ───────────────────────────
168+
169+
async fn list_insert(
170+
&self,
171+
collection: &str,
172+
document_id: &str,
173+
list_path: &str,
174+
index: usize,
175+
fields: &Value,
176+
) -> NodeDbResult<()> {
177+
self.list_insert_impl(collection, document_id, list_path, index, fields)
178+
.await
179+
}
180+
181+
async fn list_delete(
182+
&self,
183+
collection: &str,
184+
document_id: &str,
185+
list_path: &str,
186+
index: usize,
187+
) -> NodeDbResult<()> {
188+
self.list_delete_impl(collection, document_id, list_path, index)
189+
.await
190+
}
191+
192+
async fn list_move(
193+
&self,
194+
collection: &str,
195+
document_id: &str,
196+
list_path: &str,
197+
from_index: usize,
198+
to_index: usize,
199+
) -> NodeDbResult<()> {
200+
self.list_move_impl(collection, document_id, list_path, from_index, to_index)
201+
.await
202+
}
203+
167204
// ─── Document Operations ─────────────────────────────────────────
168205

169206
async fn document_get(&self, collection: &str, id: &str) -> NodeDbResult<Option<Document>> {

nodedb-lite/src/nodedb/trait_impl/mod.rs

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,9 +3,10 @@
33
//! `NodeDb` trait implementation for `NodeDbLite`, split by concern.
44
//!
55
//! - `dispatch`: the single `impl NodeDb for NodeDbLite<S>` block.
6-
//! - `vector` / `graph` / `document` / `sql_lifecycle`: inherent helpers
7-
//! the dispatch block delegates to, one file per domain.
6+
//! - `vector` / `graph` / `document` / `sql_lifecycle` / `crdt_list`:
7+
//! inherent helpers the dispatch block delegates to, one file per domain.
88
9+
mod crdt_list;
910
mod dispatch;
1011
mod document;
1112
mod document_batch;

nodedb-lite/src/query/catalog.rs

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -378,6 +378,7 @@ mod tests {
378378
name: name.into(),
379379
collection_type: ct,
380380
bitemporal,
381+
crdt: false,
381382
fields: vec![("v".into(), "BIGINT".into())],
382383
primary: PrimaryEngine::Document,
383384
vector_primary: None,

0 commit comments

Comments
 (0)