feat(catalog): add create-only PutCollectionIfAbsent entry variant

farhan-syah · farhan-syah · commit 32c07f9bc8dc · 2026-07-02T07:10:05.000+08:00
Adds a CatalogEntry variant that upserts a collection only when absent,
threading it through apply, descriptor versioning, post-apply sync/async
dispatch, gateway cache invalidation, drain lease descriptor resolution,
and integrity verification. Lets CRDT sync materialize an announced
collection without racing or clobbering a locally-authored definition
of the same name.
diff --git a/nodedb-cluster-tests/tests/catalog_put_if_absent.rs b/nodedb-cluster-tests/tests/catalog_put_if_absent.rs
@@ -0,0 +1,154 @@
+// SPDX-License-Identifier: BUSL-1.1
+//! End-to-end cluster tests for the create-only
+//! `CatalogEntry::PutCollectionIfAbsent` primitive.
+//!
+//! `PutCollectionIfAbsent` materializes a collection through the
+//! metadata raft group ONLY when no collection of the same
+//! `(database_id, tenant_id, name)` already exists — it never
+//! clobbers an existing schema. This is the durable primitive CRDT
+//! sync will use to announce collections without racing a
+//! locally-authored definition.
+//!
+//! No SQL DDL emits this variant yet, so the tests propose the entry
+//! directly through `metadata_proposer::propose_catalog_entry` (which
+//! forwards to the metadata-group leader) and assert idempotency +
+//! no-clobber by reading the replicated record on every node.
+
+mod common;
+
+use std::time::Duration;
+
+use common::cluster_harness::{TestCluster, wait_for};
+
+use nodedb::control::catalog_entry::CatalogEntry;
+use nodedb::control::metadata_proposer::propose_catalog_entry;
+use nodedb::control::security::catalog::StoredCollection;
+use nodedb_types::DatabaseId;
+
+const TENANT: u64 = 1;
+const COLL: &str = "if_absent_coll";
+
+/// Read the distinguishing fields `(bitemporal, declared_primary_key)`
+/// of a collection from a node's local `SystemCatalog` redb — the same
+/// record every node's applier writes.
+fn coll_fields(
+    node: &common::cluster_harness::TestClusterNode,
+    name: &str,
+) -> Option<(bool, Option<String>)> {
+    node.shared
+        .credentials
+        .catalog()
+        .as_ref()
+        .and_then(|c| {
+            c.get_collection(DatabaseId::DEFAULT, TENANT, name)
+                .ok()
+                .flatten()
+        })
+        .map(|c| (c.bitemporal, c.declared_primary_key))
+}
+
+/// Propose a `PutCollectionIfAbsent` for `coll`, trying each node
+/// until one accepts (the proposer forwards to the metadata leader,
+/// so any node works — the loop mirrors `exec_ddl_on_any_leader`).
+fn propose_if_absent(cluster: &TestCluster, coll: StoredCollection) -> Result<(), String> {
+    let entry = CatalogEntry::PutCollectionIfAbsent(Box::new(coll));
+    let mut last_err = String::new();
+    for node in &cluster.nodes {
+        match propose_catalog_entry(&node.shared, &entry) {
+            Ok(_) => return Ok(()),
+            Err(e) => last_err = e.to_string(),
+        }
+    }
+    Err(format!(
+        "no node accepted PutCollectionIfAbsent: {last_err}"
+    ))
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
+async fn put_if_absent_creates_then_no_clobbers_then_idempotent() {
+    let cluster = TestCluster::spawn_three().await.expect("3-node cluster");
+
+    // A: the winning definition — bitemporal + a distinct PRIMARY KEY.
+    let mut a = StoredCollection::new(TENANT, COLL, "tester");
+    a.bitemporal = true;
+    a.declared_primary_key = Some("a_key".to_string());
+
+    // 1. Create via PutCollectionIfAbsent (collection is absent).
+    propose_if_absent(&cluster, a.clone()).expect("propose A");
+
+    // Assert A materialized on all three nodes with A's fields.
+    wait_for(
+        "all 3 nodes see collection with A's distinguishing fields",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || {
+            cluster
+                .nodes
+                .iter()
+                .all(|n| coll_fields(n, COLL) == Some((true, Some("a_key".to_string()))))
+        },
+    )
+    .await;
+
+    // 2. Propose a DIFFERENT definition B under the same tenant+name.
+    //    Because the collection already exists, this must be a no-op.
+    let mut b = StoredCollection::new(TENANT, COLL, "tester");
+    b.bitemporal = false;
+    b.declared_primary_key = Some("b_key".to_string());
+    propose_if_absent(&cluster, b).expect("propose B");
+
+    // Wait for B's proposal to have applied cluster-wide, then assert
+    // every node STILL shows A's fields — B was skipped, no clobber.
+    wait_for(
+        "no-clobber: every node still shows A after B proposal",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || {
+            cluster
+                .nodes
+                .iter()
+                .all(|n| coll_fields(n, COLL) == Some((true, Some("a_key".to_string()))))
+        },
+    )
+    .await;
+    for node in &cluster.nodes {
+        assert_eq!(
+            coll_fields(node, COLL),
+            Some((true, Some("a_key".to_string()))),
+            "B must not clobber A on node {}",
+            node.node_id
+        );
+    }
+
+    // 3. Re-propose A verbatim — idempotent no-op. Still exactly one
+    //    collection, unchanged.
+    propose_if_absent(&cluster, a).expect("re-propose A");
+    wait_for(
+        "idempotent: still exactly one collection with A's fields",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || {
+            cluster.nodes.iter().all(|n| {
+                n.cached_collection_count() == 1
+                    && coll_fields(n, COLL) == Some((true, Some("a_key".to_string())))
+            })
+        },
+    )
+    .await;
+    for node in &cluster.nodes {
+        assert_eq!(
+            node.cached_collection_count(),
+            1,
+            "exactly one collection on node {}",
+            node.node_id
+        );
+        assert_eq!(
+            coll_fields(node, COLL),
+            Some((true, Some("a_key".to_string()))),
+            "A unchanged on node {}",
+            node.node_id
+        );
+    }
+
+    cluster.shutdown().await;
+}
diff --git a/nodedb/src/control/catalog_entry/apply/collection.rs b/nodedb/src/control/catalog_entry/apply/collection.rs
@@ -26,6 +26,49 @@ pub fn put(stored: &StoredCollection, catalog: &SystemCatalog) {
     );
 }
 
+/// Create-only variant of [`put`]: writes the collection (and its
+/// owner row) exactly as `put` does, but ONLY when no collection with
+/// the same `(database_id, tenant_id, name)` already exists. If one is
+/// present, this is a no-op — the existing schema is never clobbered.
+///
+/// The existence check is the ONLY behavioral difference from `put`;
+/// the create path mirrors `put`'s body precisely so replay/snapshot
+/// re-application stays idempotent.
+pub fn put_if_absent(stored: &StoredCollection, catalog: &SystemCatalog) {
+    match catalog.get_collection(stored.database_id, stored.tenant_id, &stored.name) {
+        Ok(Some(_)) => {
+            debug!(
+                collection = %stored.name,
+                tenant = stored.tenant_id,
+                "catalog_entry: put_collection_if_absent skipped existing collection"
+            );
+        }
+        Ok(None) => {
+            if let Err(e) = catalog.put_collection(stored.database_id, stored) {
+                warn!(
+                    collection = %stored.name,
+                    tenant = stored.tenant_id,
+                    error = %e,
+                    "catalog_entry: put_collection_if_absent put_collection failed"
+                );
+            }
+            super::owner::put_parent_owner(
+                object_type::COLLECTION,
+                stored.tenant_id,
+                &stored.name,
+                &stored.owner,
+                catalog,
+            );
+        }
+        Err(e) => warn!(
+            collection = %stored.name,
+            tenant = stored.tenant_id,
+            error = %e,
+            "catalog_entry: put_collection_if_absent get failed"
+        ),
+    }
+}
+
 /// Hard-delete the catalog metadata for a collection: primary
 /// `StoredCollection` row, owner row, and surrogate ↔ PK map.
 ///
diff --git a/nodedb/src/control/catalog_entry/apply/mod.rs b/nodedb/src/control/catalog_entry/apply/mod.rs
@@ -77,6 +77,7 @@ pub fn apply_to(entry: &CatalogEntry, catalog: &SystemCatalog) {
 fn apply_to_inner(entry: &CatalogEntry, catalog: &SystemCatalog) {
     match entry {
         CatalogEntry::PutCollection(stored) => collection::put(stored, catalog),
+        CatalogEntry::PutCollectionIfAbsent(stored) => collection::put_if_absent(stored, catalog),
         CatalogEntry::DeactivateCollection { tenant_id, name } => {
             collection::deactivate(*tenant_id, name, catalog)
         }
diff --git a/nodedb/src/control/catalog_entry/descriptor_stamp.rs b/nodedb/src/control/catalog_entry/descriptor_stamp.rs
@@ -78,6 +78,33 @@ pub fn stamp(entry: CatalogEntry, clock: &HlcClock, catalog: &SystemCatalog) ->
             stored.modification_hlc = hlc;
             CatalogEntry::PutCollection(stored)
         }
+        CatalogEntry::PutCollectionIfAbsent(mut stored) => {
+            let prior = catalog
+                .get_collection(stored.database_id, stored.tenant_id, &stored.name)
+                .ok()
+                .flatten();
+            let prior_descriptor = prior.as_ref().map(|c| c.descriptor_version).unwrap_or(0);
+            stored.descriptor_version = prior_descriptor.saturating_add(1);
+            // Constraint version bumps ONLY when the derived constraint set
+            // actually changes, so an unrelated ALTER never advances the
+            // apply-time fence key and never transiently rejects in-flight
+            // CRDT deltas. `Constraint: Eq` + name-sorted translator make the
+            // set comparison exact and order-stable.
+            let prior_constraint_version =
+                prior.as_ref().map(|c| c.constraint_version).unwrap_or(0);
+            let prior_set = prior
+                .as_ref()
+                .map(crate::control::security::catalog::collection_constraints)
+                .unwrap_or_default();
+            let new_set = crate::control::security::catalog::collection_constraints(&stored);
+            stored.constraint_version = if new_set != prior_set {
+                prior_constraint_version.saturating_add(1)
+            } else {
+                prior_constraint_version
+            };
+            stored.modification_hlc = hlc;
+            CatalogEntry::PutCollectionIfAbsent(stored)
+        }
         CatalogEntry::PutMaterializedView(mut stored) => {
             let prior = catalog
                 .get_materialized_view(stored.tenant_id, &stored.name)
diff --git a/nodedb/src/control/catalog_entry/entry.rs b/nodedb/src/control/catalog_entry/entry.rs
@@ -30,6 +30,11 @@ pub enum CatalogEntry {
     /// (strict schema changes, retention / legal_hold / LVC /
     /// append_only toggles, materialized_sum bindings).
     PutCollection(Box<StoredCollection>),
+    /// Create-only collection upsert: applies iff the collection is
+    /// absent, and never clobbers an existing schema. Used by CRDT
+    /// sync to materialize announced collections without racing or
+    /// overwriting a locally-authored definition of the same name.
+    PutCollectionIfAbsent(Box<StoredCollection>),
     /// Mark a collection as `is_active = false`. Record is
     /// preserved for audit + undrop. The soft-delete step in the
     /// two-step DROP → retention-expiry → PURGE flow.
@@ -332,6 +337,7 @@ impl CatalogEntry {
     pub fn kind(&self) -> &'static str {
         match self {
             Self::PutCollection(_) => "put_collection",
+            Self::PutCollectionIfAbsent(_) => "put_collection_if_absent",
             Self::DeactivateCollection { .. } => "deactivate_collection",
             Self::PurgeCollection { .. } => "purge_collection",
             Self::PutSequence(_) => "put_sequence",
diff --git a/nodedb/src/control/catalog_entry/post_apply/async_dispatch/dispatcher.rs b/nodedb/src/control/catalog_entry/post_apply/async_dispatch/dispatcher.rs
@@ -49,6 +49,17 @@ pub fn spawn_post_apply_async_side_effects(
                 });
             });
         }
+        CatalogEntry::PutCollectionIfAbsent(stored) => {
+            // SYNCHRONOUS: Register must complete before the applied-index
+            // watcher bumps so any subsequent scan on this node finds the
+            // collection in doc_configs. block_in_place is valid because
+            // the raft tick loop runs on a tokio worker thread.
+            tokio::task::block_in_place(|| {
+                tokio::runtime::Handle::current().block_on(async move {
+                    collection::put_async(*stored, shared).await;
+                });
+            });
+        }
         CatalogEntry::PurgeCollection { tenant_id, name } => {
             tokio::spawn(async move {
                 collection::purge_async(tenant_id, name, raft_index, shared).await;
diff --git a/nodedb/src/control/catalog_entry/post_apply/gateway_invalidation.rs b/nodedb/src/control/catalog_entry/post_apply/gateway_invalidation.rs
@@ -50,6 +50,9 @@ pub(crate) fn invalidate_gateway_cache_for_entry(entry: &CatalogEntry, shared: &
         CatalogEntry::PutCollection(stored) => {
             inv.invalidate(&stored.name, stored.descriptor_version.max(1));
         }
+        CatalogEntry::PutCollectionIfAbsent(stored) => {
+            inv.invalidate(&stored.name, stored.descriptor_version.max(1));
+        }
         CatalogEntry::DeactivateCollection { name, .. } => {
             // Treat deactivation as version 0 (collection gone — any cached
             // plan for it is stale).
diff --git a/nodedb/src/control/catalog_entry/post_apply/sync.rs b/nodedb/src/control/catalog_entry/post_apply/sync.rs
@@ -44,6 +44,11 @@ pub fn apply_post_apply_side_effects_sync(entry: &CatalogEntry, shared: &Arc<Sha
             // the async part, handled by `spawn_post_apply_async_side_effects`.
             collection::put_owner_sync(stored, Arc::clone(shared));
         }
+        CatalogEntry::PutCollectionIfAbsent(stored) => {
+            // Owner record install is sync; Data Plane register is
+            // the async part, handled by `spawn_post_apply_async_side_effects`.
+            collection::put_owner_sync(stored, Arc::clone(shared));
+        }
         CatalogEntry::DeactivateCollection { tenant_id, name } => {
             collection::deactivate(*tenant_id, name.clone(), Arc::clone(shared));
         }
diff --git a/nodedb/src/control/cluster/metadata_applier_audit.rs b/nodedb/src/control/cluster/metadata_applier_audit.rs
@@ -138,6 +138,11 @@ pub(super) fn describe_entry(e: &catalog_entry::CatalogEntry) -> (String, u64, S
             c.descriptor_version,
             format!("{:?}", c.modification_hlc),
         ),
+        E::PutCollectionIfAbsent(c) => (
+            c.name.clone(),
+            c.descriptor_version,
+            format!("{:?}", c.modification_hlc),
+        ),
         E::DeactivateCollection { name, .. } => (name.clone(), 0, String::new()),
         E::PurgeCollection { name, .. } => (name.clone(), 0, String::new()),
         E::RecordWalTombstone { collection, .. } => (collection.clone(), 0, String::new()),
diff --git a/nodedb/src/control/lease/drain_propose.rs b/nodedb/src/control/lease/drain_propose.rs
@@ -256,6 +256,11 @@ pub fn descriptor_id_for_implicit_clear(entry: &CatalogEntry) -> Option<Descript
             DescriptorKind::Collection,
             stored.name.clone(),
         )),
+        CatalogEntry::PutCollectionIfAbsent(stored) => Some(DescriptorId::new(
+            stored.tenant_id,
+            DescriptorKind::Collection,
+            stored.name.clone(),
+        )),
         CatalogEntry::PutMaterializedView(stored) => Some(DescriptorId::new(
             stored.tenant_id,
             DescriptorKind::MaterializedView,
@@ -319,6 +324,22 @@ pub fn descriptor_id_and_prior_version(
                 prior,
             ))
         }
+        CatalogEntry::PutCollectionIfAbsent(stored) => {
+            let prior = catalog
+                .get_collection(DatabaseId::DEFAULT, stored.tenant_id, &stored.name)
+                .ok()
+                .flatten()
+                .map(|c| c.descriptor_version)
+                .unwrap_or(0);
+            Some((
+                DescriptorId::new(
+                    stored.tenant_id,
+                    DescriptorKind::Collection,
+                    stored.name.clone(),
+                ),
+                prior,
+            ))
+        }
         CatalogEntry::PutMaterializedView(stored) => {
             let prior = catalog
                 .get_materialized_view(stored.tenant_id, &stored.name)
diff --git a/nodedb/tests/catalog_integrity_verify.rs b/nodedb/tests/catalog_integrity_verify.rs
@@ -159,6 +159,7 @@ enum VariantClass {
 fn classify(entry: &CatalogEntry) -> VariantClass {
     match entry {
         CatalogEntry::PutCollection(_) => VariantClass::ParentReplicated,
+        CatalogEntry::PutCollectionIfAbsent(_) => VariantClass::ParentReplicated,
         CatalogEntry::PutFunction(_) => VariantClass::ParentReplicated,
         CatalogEntry::PutProcedure(_) => VariantClass::ParentReplicated,
         CatalogEntry::PutTrigger(_) => VariantClass::ParentReplicated,

Original file line number	Diff line number	Diff line change
`@@ -77,6 +77,7 @@ pub fn apply_to(entry: &CatalogEntry, catalog: &SystemCatalog) {`
`77`	`77`	`fn apply_to_inner(entry: &CatalogEntry, catalog: &SystemCatalog) {`
`78`	`78`	`match entry {`
`79`	`79`	`CatalogEntry::PutCollection(stored) => collection::put(stored, catalog),`
	`80`	`+ CatalogEntry::PutCollectionIfAbsent(stored) => collection::put_if_absent(stored, catalog),`
`80`	`81`	`CatalogEntry::DeactivateCollection { tenant_id, name } => {`
`81`	`82`	`collection::deactivate(*tenant_id, name, catalog)`
`82`	`83`	`}`
Original file line number	Diff line number	Diff line change
`@@ -44,6 +44,11 @@ pub fn apply_post_apply_side_effects_sync(entry: &CatalogEntry, shared: &Arc<Sha`
`44`	`44`	// the async part, handled by `spawn_post_apply_async_side_effects`.
`45`	`45`	`collection::put_owner_sync(stored, Arc::clone(shared));`
`46`	`46`	`}`
	`47`	`+ CatalogEntry::PutCollectionIfAbsent(stored) => {`
	`48`	`+ // Owner record install is sync; Data Plane register is`
	`49`	+ // the async part, handled by `spawn_post_apply_async_side_effects`.
	`50`	`+ collection::put_owner_sync(stored, Arc::clone(shared));`
	`51`	`+ }`
`47`	`52`	`CatalogEntry::DeactivateCollection { tenant_id, name } => {`
`48`	`53`	`collection::deactivate(*tenant_id, name.clone(), Arc::clone(shared));`
`49`	`54`	`}`