feat(control): support WITH (crdt=true) on document collections

Adds a persisted `crdt` flag to StoredCollection/CollectionDescriptor
so document collections can opt into CRDT (Loro) storage at CREATE
time and have that choice sync across nodes. DDL validates that
`crdt=true` is document-engine-only, the descriptor conversion carries
the flag over sync, and the doc-config builder now honors it instead
of hardcoding CRDT off.

Author: farhan-syah

nodedb-types/src/sync/wire/collection_schema.rs

@@ -35,6 +35,9 @@ pub struct CollectionDescriptor {
     /// Whether the collection tracks system-time + valid-time versions.
     #[msgpack(default)]
     pub bitemporal: bool,
+    /// Whether this collection uses CRDT (Loro) storage for offline-first sync.
+    #[msgpack(default)]
+    pub crdt: bool,
     /// Lightweight field type hints, e.g. `[("email", "string")]`.
     #[msgpack(default)]
     pub fields: Vec<(String, String)>,
@@ -83,6 +86,7 @@ mod tests {
             name: "users".into(),
             collection_type: CollectionType::document(),
             bitemporal: true,
+            crdt: true,
             fields: vec![("email".into(), "string".into())],
             primary: PrimaryEngine::Document,
             vector_primary: None,
@@ -105,6 +109,7 @@ mod tests {
         assert_eq!(decoded_msg.descriptor.name, "users");
         assert_eq!(decoded_msg.descriptor.tenant_id, 7);
         assert!(decoded_msg.descriptor.bitemporal);
+        assert!(decoded_msg.descriptor.crdt);
         assert_eq!(decoded_msg.descriptor.primary, PrimaryEngine::Document);
     }
 

nodedb/src/control/security/catalog/collection.rs

@@ -146,6 +146,12 @@ pub struct StoredCollection {
     /// other engines ignore it.
     #[msgpack(default)]
     pub bitemporal: bool,
+    /// Whether this collection uses CRDT (Loro) storage for offline-first
+    /// sync. Document-engine-only: `CREATE COLLECTION ... WITH (crdt=true)`
+    /// is rejected at DDL time for any other collection type, so a `true`
+    /// value here always implies a document collection.
+    #[msgpack(default)]
+    pub crdt: bool,
     /// Durable CRDT conflict-resolution policy (JSON-serialized
     /// `nodedb_crdt::policy::CollectionPolicy`), set via
     /// `ALTER COLLECTION ... SET ON CONFLICT ... FOR ...`. `None` = no
@@ -292,6 +298,7 @@ impl StoredCollection {
             materialized_sums: Vec::new(),
             lvc_enabled: false,
             bitemporal: false,
+            crdt: false,
             conflict_policy: None,
             permission_tree_def: None,
             indexes: Vec::new(),

nodedb/src/control/security/catalog/collection_descriptor_convert.rs

@@ -5,7 +5,7 @@
 //! `CREATE COLLECTION` and CRDT sync produce the same descriptor shape.
 //!
 //! `From<&StoredCollection> for CollectionDescriptor` is the emit side: it
-//! reads only the 11 fields the descriptor carries, dropping ownership,
+//! reads only the 12 fields the descriptor carries, dropping ownership,
 //! timestamps, and enforcement/index/constraint state that never travels
 //! over sync. [`stored_from_descriptor`] is the receive side: it starts
 //! from [`StoredCollection::new`] (which sets sane enforcement defaults
@@ -24,6 +24,7 @@ impl From<&StoredCollection> for CollectionDescriptor {
             name: stored.name.clone(),
             collection_type: stored.collection_type.clone(),
             bitemporal: stored.bitemporal,
+            crdt: stored.crdt,
             fields: stored.fields.clone(),
             primary: stored.primary,
             vector_primary: stored.vector_primary.clone(),
@@ -49,6 +50,7 @@ pub(crate) fn stored_from_descriptor(
     stored.database_id = descriptor.database_id;
     stored.collection_type = descriptor.collection_type.clone();
     stored.bitemporal = descriptor.bitemporal;
+    stored.crdt = descriptor.crdt;
     stored.fields = descriptor.fields.clone();
     stored.primary = descriptor.primary;
     stored.vector_primary = descriptor.vector_primary.clone();
@@ -73,6 +75,7 @@ mod tests {
         assert_eq!(back.database_id, stored.database_id);
         assert_eq!(back.collection_type, stored.collection_type);
         assert_eq!(back.bitemporal, stored.bitemporal);
+        assert_eq!(back.crdt, stored.crdt);
         assert_eq!(back.fields, stored.fields);
         assert_eq!(back.primary, stored.primary);
         assert_eq!(back.vector_primary, stored.vector_primary);
@@ -87,6 +90,7 @@ mod tests {
             PartitionStrategy::default_for_collection_type(&collection_type);
         stored.collection_type = collection_type;
         stored.bitemporal = true;
+        stored.crdt = true;
         stored.declared_primary_key = Some("id".to_string());
         stored.descriptor_version = 5;
         stored
@@ -188,6 +192,16 @@ mod tests {
         assert!(back.bitemporal);
     }
 
+    #[test]
+    fn crdt_flag_round_trips() {
+        let mut stored = base_stored("synced_docs", CollectionType::document());
+        stored.crdt = true;
+        let descriptor = CollectionDescriptor::from(&stored);
+        assert!(descriptor.crdt);
+        let back = stored_from_descriptor(&descriptor, "sync");
+        assert!(back.crdt);
+    }
+
     #[test]
     fn owner_assigned_on_receive() {
         let stored = base_stored("owned", CollectionType::document());

nodedb/src/control/server/shared/ddl/neutral/collection/create/build.rs

@@ -37,6 +37,43 @@ fn err(sqlstate: &str, message: String) -> DdlError {
     }
 }
 
+/// Parse a `WITH (crdt=...)` option value as a boolean, accepting
+/// `"true"`/`"false"` case-insensitively. Any other value is a
+/// user error surfaced as a typed DDL error (SQLSTATE 42601).
+fn parse_crdt_flag(value: &str) -> Result<bool, DdlError> {
+    match value.trim() {
+        v if v.eq_ignore_ascii_case("true") => Ok(true),
+        v if v.eq_ignore_ascii_case("false") => Ok(false),
+        other => Err(err(
+            "42601",
+            format!("invalid value for WITH (crdt=...): '{other}'; expected 'true' or 'false'"),
+        )),
+    }
+}
+
+/// Resolve the CRDT storage flag from the `WITH (...)` option list.
+///
+/// A missing `crdt` option defaults to `false`. CRDT (Loro) storage is a
+/// document-engine capability, so `crdt=true` is rejected with SQLSTATE
+/// 42601 on any non-document collection rather than persisting a flag no
+/// engine would honor.
+fn resolve_crdt_flag(
+    options: &[(String, String)],
+    collection_type: &nodedb_types::CollectionType,
+) -> Result<bool, DdlError> {
+    let crdt = match options.iter().find(|(k, _)| k.eq_ignore_ascii_case("crdt")) {
+        Some((_, v)) => parse_crdt_flag(v)?,
+        None => false,
+    };
+    if crdt && !matches!(collection_type, nodedb_types::CollectionType::Document(_)) {
+        return Err(err(
+            "42601",
+            "WITH (crdt=true) is only supported on document collections".to_string(),
+        ));
+    }
+    Ok(crdt)
+}
+
 /// Per-surface configuration. The fields are the entire surface-level
 /// difference between `CREATE COLLECTION` and `CREATE TABLE`.
 pub struct Variant {
@@ -172,6 +209,8 @@ pub async fn build_and_persist(
     if hash_chain && !append_only {
         return Err(err("42601", "HASH_CHAIN requires APPEND_ONLY".to_string()));
     }
+
+    let crdt = resolve_crdt_flag(options, &collection_type)?;
     let balanced =
         parse_balanced_clause_from_raw(balanced_raw.unwrap_or("")).map_err(|e| err("42601", e))?;
 
@@ -217,6 +256,7 @@ pub async fn build_and_persist(
         materialized_sums: Vec::new(),
         lvc_enabled: false,
         bitemporal,
+        crdt,
         permission_tree_def: None,
         indexes: Vec::new(),
         size_bytes_estimate: 0,
@@ -389,6 +429,47 @@ mod tests {
     //! Collection name validation tests. Relocated verbatim from the pgwire
     //! `pgwire::ddl::collection::create::tests` module (now deleted).
 
+    use super::resolve_crdt_flag;
+
+    fn opts(pairs: &[(&str, &str)]) -> Vec<(String, String)> {
+        pairs
+            .iter()
+            .map(|(k, v)| (k.to_string(), v.to_string()))
+            .collect()
+    }
+
+    #[test]
+    fn crdt_true_on_document_collection_resolves_true() {
+        let options = opts(&[("crdt", "true")]);
+        let flag = resolve_crdt_flag(&options, &nodedb_types::CollectionType::document())
+            .expect("crdt=true on a document collection must resolve");
+        assert!(flag);
+    }
+
+    #[test]
+    fn crdt_true_on_non_document_collection_rejected() {
+        let options = opts(&[("crdt", "true")]);
+        let err = resolve_crdt_flag(&options, &nodedb_types::CollectionType::columnar())
+            .expect_err("crdt=true on a non-document collection must be rejected");
+        assert_eq!(err.sqlstate, "42601");
+    }
+
+    #[test]
+    fn crdt_garbage_value_rejected() {
+        let options = opts(&[("crdt", "maybe")]);
+        let err = resolve_crdt_flag(&options, &nodedb_types::CollectionType::document())
+            .expect_err("a non-boolean crdt value must be rejected");
+        assert_eq!(err.sqlstate, "42601");
+    }
+
+    #[test]
+    fn crdt_absent_defaults_false() {
+        let options = opts(&[("engine", "kv")]);
+        let flag = resolve_crdt_flag(&options, &nodedb_types::CollectionType::document())
+            .expect("absent crdt option must resolve to a default");
+        assert!(!flag);
+    }
+
     /// Collection name validation: allowed chars are `[a-zA-Z0-9_-]`.
     fn validate_name(name: &str) -> bool {
         !name.is_empty()

nodedb/src/control/server/shared/ddl/neutral/collection/register.rs

@@ -57,7 +57,11 @@ pub async fn dispatch_register_if_needed(
         crate::control::server::shared::ddl::schema_validation::parse_fields_clause(parts);
     let mut indexes = derive_auto_indexes(fields.iter().map(|(n, _)| n.as_str()));
     extend_with_catalog_indexes(&mut indexes, &coll);
-    let _ = sql; // Reserved for future CRDT detection from SQL.
+    // `sql` is unused on this leader-side path: index derivation reads
+    // `parts`/the catalog row directly, and the `crdt` flag already
+    // travels on `StoredCollection` (set at CREATE time from `WITH
+    // (crdt=...)`), so no SQL re-parsing is needed here.
+    let _ = sql;
     dispatch_register_from_stored_inner(state, tenant_id, &coll, indexes).await
 }
 
@@ -244,7 +248,7 @@ pub(crate) fn build_doc_config_from_stored(
     };
 
     let mut config = crate::engine::document::store::CollectionConfig::new(&name);
-    config.crdt_enabled = false;
+    config.crdt_enabled = coll.crdt;
     config.storage_mode = storage_mode;
     config.enforcement = enforcement;
     config.bitemporal = coll.bitemporal;

nodedb/src/control/server/shared/ddl/neutral/continuous_agg/create.rs

@@ -209,6 +209,7 @@ pub async fn create_continuous_aggregate(
             materialized_sums: Vec::new(),
             lvc_enabled: false,
             bitemporal: false,
+            crdt: false,
             permission_tree_def: None,
             indexes: Vec::new(),
             size_bytes_estimate: 0,

nodedb/src/control/server/shared/ddl/neutral/materialized_view/create.rs

@@ -138,6 +138,7 @@ pub async fn create_materialized_view(
             materialized_sums: Vec::new(),
             lvc_enabled: false,
             bitemporal: false,
+            crdt: false,
             permission_tree_def: None,
             indexes: Vec::new(),
             size_bytes_estimate: 0,

nodedb/src/control/server/shared/ddl/neutral/timeseries/create.rs

@@ -86,6 +86,7 @@ pub fn create_timeseries(
         materialized_sums: Vec::new(),
         lvc_enabled: false,
         bitemporal: false,
+        crdt: false,
         permission_tree_def: None,
         indexes: Vec::new(),
         size_bytes_estimate: 0,