feat(sync): materialize peer-announced collections into local catalog

farhan-syah · farhan-syah · commit f0bbfb63969e · 2026-07-02T07:43:58.000+08:00
Add the receive-side handler for the CollectionSchema wire message: a
peer's announced descriptor is converted to a stored collection and
proposed as a create-only PutCollectionIfAbsent entry, then applied
locally, mirroring the pgwire CREATE path.

Post-apply side effects for PutCollectionIfAbsent now read the
canonical collection back from the catalog instead of trusting the
carried entry, so a no-op re-announce (collection already existed)
can't clobber the pre-existing owner or register a divergent config
into the Data Plane.
diff --git a/nodedb-cluster-tests/tests/sync_collection_schema.rs b/nodedb-cluster-tests/tests/sync_collection_schema.rs
@@ -0,0 +1,146 @@
+// SPDX-License-Identifier: BUSL-1.1
+//! A peer-announced `CollectionSchema` (opcode 0x13) materializes the
+//! collection into the cluster catalog on EVERY node.
+//!
+//! ## What this guards
+//!
+//! When a sync peer announces a `CollectionSchemaSyncMsg`, the receiving
+//! node must materialize the collection into the system catalog — create-only,
+//! never clobbering an existing collection — and, via the shared post-apply
+//! path, register the Data-Plane engine state on every node that applies the
+//! Raft entry. The end result: the synced collection is catalog-visible and
+//! queryable cluster-wide, carrying the correct engine type and bitemporal
+//! flag.
+//!
+//! The test drives the sync WebSocket end-to-end: it connects to ONE node's
+//! sync listener, announces a descriptor for each supported engine, then
+//! asserts the collection appears with the right `collection_type` +
+//! `bitemporal` on a DIFFERENT node — proving Raft propagation plus per-node
+//! catalog materialization, not just a local write on the receiving node.
+
+mod common;
+use common::cluster_harness::{TestCluster, TestClusterNode};
+
+use std::time::{Duration, Instant};
+
+use nodedb::control::server::sync::listener::{SyncListenerConfig, start_sync_listener};
+use nodedb_test_support::sync_client::SyncTestClient;
+use nodedb_types::collection_config::{PartitionStrategy, PrimaryEngine};
+use nodedb_types::columnar::{ColumnDef, ColumnType, StrictSchema};
+use nodedb_types::sync::wire::CollectionDescriptor;
+use nodedb_types::{CollectionType, DatabaseId, Hlc};
+
+/// Trust-mode sync sessions authenticate as tenant 1 (see the sync
+/// handshake), so announced descriptors must carry tenant 1.
+const TENANT: u64 = 1;
+
+/// A single strict-schema column set reused by strict + kv descriptors.
+fn pk_schema() -> StrictSchema {
+    StrictSchema {
+        columns: vec![ColumnDef::required("id", ColumnType::Int64).with_primary_key()],
+        version: 1,
+        dropped_columns: Vec::new(),
+        bitemporal: false,
+    }
+}
+
+/// Build a descriptor for `name` with `collection_type` and `bitemporal`,
+/// mirroring the shape a local `CREATE COLLECTION` would emit over sync.
+fn descriptor(
+    name: &str,
+    collection_type: CollectionType,
+    bitemporal: bool,
+) -> CollectionDescriptor {
+    CollectionDescriptor {
+        tenant_id: TENANT,
+        database_id: DatabaseId::DEFAULT,
+        name: name.to_string(),
+        partition_strategy: PartitionStrategy::default_for_collection_type(&collection_type),
+        collection_type,
+        bitemporal,
+        fields: Vec::new(),
+        primary: PrimaryEngine::Document,
+        vector_primary: None,
+        declared_primary_key: None,
+        descriptor_version: 0,
+    }
+}
+
+/// Poll `node`'s local catalog until the collection is visible, then return
+/// its `(collection_type, bitemporal)`. Panics if it does not converge —
+/// a bounded retry loop, not a single fixed sleep.
+async fn await_collection(node: &TestClusterNode, name: &str) -> (CollectionType, bool) {
+    let deadline = Instant::now() + Duration::from_secs(30);
+    loop {
+        let found = node.shared.credentials.catalog().as_ref().and_then(|c| {
+            c.get_collection(DatabaseId::DEFAULT, TENANT, name)
+                .ok()
+                .flatten()
+        });
+        if let Some(coll) = found {
+            return (coll.collection_type, coll.bitemporal);
+        }
+        if Instant::now() >= deadline {
+            panic!("collection '{name}' did not become catalog-visible on the follower within 30s");
+        }
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn announced_schema_materializes_on_every_node() {
+    let cluster = TestCluster::spawn_three()
+        .await
+        .expect("spawn three-node cluster");
+
+    // Start the sync listener on node 0; the client connects here, so node 0
+    // is the RECEIVING node. Assertions run on node 1 (a different node) to
+    // prove Raft propagation + per-node catalog materialization.
+    let cfg = SyncListenerConfig {
+        listen_addr: std::net::SocketAddr::from(([127, 0, 0, 1], 0)),
+        ..Default::default()
+    };
+    let state = start_sync_listener(cfg, Some(std::sync::Arc::clone(&cluster.nodes[0].shared)))
+        .await
+        .expect("start sync listener on node 0");
+    let addr = state.config.listen_addr;
+
+    let mut client = SyncTestClient::connect(addr)
+        .await
+        .expect("sync handshake with node 0");
+
+    // One descriptor per engine, plus a bitemporal document variant.
+    let cases: Vec<(&str, CollectionType, bool)> = vec![
+        ("sync_doc", CollectionType::document(), false),
+        ("sync_strict", CollectionType::strict(pk_schema()), false),
+        ("sync_kv", CollectionType::kv(pk_schema()), false),
+        ("sync_columnar", CollectionType::columnar(), false),
+        ("sync_ts", CollectionType::timeseries("ts", "1m"), false),
+        ("sync_spatial", CollectionType::spatial("geom"), false),
+        ("sync_bitemporal", CollectionType::document(), true),
+    ];
+
+    let hlc = Hlc::new(1, 0);
+    for (name, ct, bitemporal) in &cases {
+        client
+            .push_collection_schema(descriptor(name, ct.clone(), *bitemporal), hlc)
+            .await
+            .unwrap_or_else(|e| panic!("push CollectionSchema for '{name}': {e}"));
+    }
+
+    // Assert on node 1 — a node the client never talked to.
+    let follower = &cluster.nodes[1];
+    for (name, expected_ct, expected_bitemporal) in &cases {
+        let (got_ct, got_bitemporal) = await_collection(follower, name).await;
+        assert_eq!(
+            &got_ct, expected_ct,
+            "collection '{name}' materialized with the wrong engine type on the follower"
+        );
+        assert_eq!(
+            got_bitemporal, *expected_bitemporal,
+            "collection '{name}' materialized with the wrong bitemporal flag on the follower"
+        );
+    }
+
+    cluster.shutdown().await;
+}
diff --git a/nodedb-test-support/src/sync_client.rs b/nodedb-test-support/src/sync_client.rs
@@ -16,9 +16,10 @@ use futures::{SinkExt, StreamExt};
 use tokio::net::TcpStream;
 use tokio_tungstenite::{MaybeTlsStream, WebSocketStream, connect_async, tungstenite::Message};
 
+use nodedb_types::Hlc;
 use nodedb_types::sync::wire::{
-    DeltaAckMsg, DeltaPushMsg, DeltaRejectMsg, HandshakeAckMsg, HandshakeMsg, SyncFrame,
-    SyncMessageType,
+    CollectionDescriptor, CollectionSchemaSyncMsg, DeltaAckMsg, DeltaPushMsg, DeltaRejectMsg,
+    HandshakeAckMsg, HandshakeMsg, SyncFrame, SyncMessageType,
 };
 
 /// Bound on how long a single frame receive may take before the test client
@@ -133,6 +134,27 @@ impl SyncTestClient {
             }
         }
     }
+
+    /// Announce a collection descriptor to the server. Fire-and-forget:
+    /// the CollectionSchema announce has no ack frame, so this only sends
+    /// the frame and returns once it is flushed onto the socket.
+    pub async fn push_collection_schema(
+        &mut self,
+        descriptor: CollectionDescriptor,
+        creation_hlc: Hlc,
+    ) -> Result<(), String> {
+        let msg = CollectionSchemaSyncMsg {
+            descriptor,
+            creation_hlc,
+        };
+        let frame = SyncFrame::new_msgpack(SyncMessageType::CollectionSchema, &msg)
+            .ok_or_else(|| "failed to encode CollectionSchemaSyncMsg".to_string())?;
+        self.ws
+            .send(Message::Binary(frame.to_bytes().into()))
+            .await
+            .map_err(|e| format!("failed to send CollectionSchema frame: {e}"))?;
+        Ok(())
+    }
 }
 
 /// Receive the next binary frame, decode it as a `SyncFrame`, and error out
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
@@ -50,15 +50,43 @@ 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;
-                });
+            // Register from the CANONICAL collection read back from the
+            // catalog after apply — never from the carried entry. On the
+            // no-op path (the collection already existed) the carried
+            // `stored` may hold a divergent incoming config; the catalog
+            // holds the authoritative pre-existing one. Post-apply the
+            // collection always exists (created or pre-existing), so the
+            // read-back is always Some; a None here would mean the redb
+            // write silently failed, so warn and skip rather than register
+            // a divergent config.
+            let canonical = shared.credentials.catalog().as_ref().and_then(|catalog| {
+                catalog
+                    .get_collection(stored.database_id, stored.tenant_id, &stored.name)
+                    .ok()
+                    .flatten()
             });
+            match canonical {
+                Some(canonical) => {
+                    // 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(canonical, shared).await;
+                        });
+                    });
+                }
+                None => {
+                    tracing::warn!(
+                        collection = %stored.name,
+                        tenant = stored.tenant_id,
+                        "PutCollectionIfAbsent post-apply: canonical collection not found in \
+                         catalog after apply; skipping Data Plane register"
+                    );
+                }
+            }
         }
         CatalogEntry::PurgeCollection { tenant_id, name } => {
             tokio::spawn(async move {
diff --git a/nodedb/src/control/catalog_entry/post_apply/sync.rs b/nodedb/src/control/catalog_entry/post_apply/sync.rs
@@ -45,9 +45,23 @@ pub fn apply_post_apply_side_effects_sync(entry: &CatalogEntry, shared: &Arc<Sha
             collection::put_owner_sync(stored, Arc::clone(shared));
         }
         CatalogEntry::PutCollectionIfAbsent(stored) => {
+            // Install owner from the CANONICAL catalog collection, not the
+            // carried entry: a no-op re-announce must not overwrite the
+            // pre-existing owner. Post-apply the collection always exists,
+            // so the read-back is Some; the carried entry is only a
+            // best-effort fallback if the redb write silently failed.
             // 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));
+            let canonical = shared.credentials.catalog().as_ref().and_then(|catalog| {
+                catalog
+                    .get_collection(stored.database_id, stored.tenant_id, &stored.name)
+                    .ok()
+                    .flatten()
+            });
+            match canonical {
+                Some(canonical) => collection::put_owner_sync(&canonical, Arc::clone(shared)),
+                None => 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/security/catalog/collection_descriptor_convert.rs b/nodedb/src/control/security/catalog/collection_descriptor_convert.rs
@@ -41,9 +41,6 @@ impl From<&StoredCollection> for CollectionDescriptor {
 /// sync-wire one. All fields not carried by the descriptor (field_defs,
 /// event_defs, indexes, constraints, `is_active`, etc.) are left at the
 /// [`StoredCollection::new`] defaults.
-// Consumed by the sync receive handler that materializes announced
-// collections into the catalog; only the round-trip tests call it today.
-#[allow(dead_code)]
 pub(crate) fn stored_from_descriptor(
     descriptor: &CollectionDescriptor,
     owner: &str,
diff --git a/nodedb/src/control/server/sync/session/collection_schema.rs b/nodedb/src/control/server/sync/session/collection_schema.rs
@@ -0,0 +1,95 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! CollectionSchema receive handler.
+//!
+//! When a sync peer announces a [`CollectionSchemaSyncMsg`], the receiving
+//! cluster materializes the collection into its system catalog (create-only,
+//! via `PutCollectionIfAbsent` — never clobbering an existing collection).
+//! The Data-Plane engine register happens in the shared post-apply path on
+//! **every** node that applies the Raft entry, exactly as it does for a
+//! local `CREATE COLLECTION`. This handler is therefore symmetric with the
+//! pgwire CREATE handler: it only `stored_from_descriptor` → propose
+//! `PutCollectionIfAbsent` → `apply_locally_if_needed`, and never dispatches
+//! the register itself.
+
+use std::sync::Arc;
+
+use tracing::warn;
+
+use crate::control::catalog_entry::CatalogEntry;
+use crate::control::state::SharedState;
+
+use super::super::wire::{CollectionSchemaSyncMsg, SyncFrame};
+use super::state::SyncSession;
+
+impl SyncSession {
+    /// Materialize a peer-announced collection descriptor into the local
+    /// catalog. Returns `None`: this is a fire-and-forget announce with no
+    /// ack frame (mirrors `ShapeUnsubscribe`).
+    pub fn handle_collection_schema(
+        &mut self,
+        msg: &CollectionSchemaSyncMsg,
+        shared: Option<&Arc<SharedState>>,
+    ) -> Option<SyncFrame> {
+        let Some(shared) = shared else {
+            warn!(
+                session = %self.session_id,
+                collection = %msg.descriptor.name,
+                "CollectionSchema received without SharedState (permissive/test path); dropping"
+            );
+            return None;
+        };
+
+        let Some(tenant) = self.tenant_id else {
+            warn!(
+                session = %self.session_id,
+                collection = %msg.descriptor.name,
+                "CollectionSchema received before handshake established a tenant; dropping"
+            );
+            return None;
+        };
+
+        // Security: a peer must not materialize a collection into a tenant
+        // other than the one it authenticated as.
+        if msg.descriptor.tenant_id != tenant.as_u64() {
+            warn!(
+                session = %self.session_id,
+                collection = %msg.descriptor.name,
+                descriptor_tenant = msg.descriptor.tenant_id,
+                session_tenant = tenant.as_u64(),
+                "CollectionSchema tenant mismatch; refusing to materialize"
+            );
+            return None;
+        }
+
+        // Owner is the receiving peer's authenticated principal — the same
+        // identity a local CREATE records as owner.
+        let owner = self.username.as_deref().unwrap_or("sync");
+
+        let stored =
+            crate::control::security::catalog::collection_descriptor_convert::stored_from_descriptor(
+                &msg.descriptor,
+                owner,
+            );
+
+        let entry = CatalogEntry::PutCollectionIfAbsent(Box::new(stored));
+        let log_index =
+            match crate::control::metadata_proposer::propose_catalog_entry(shared, &entry) {
+                Ok(idx) => idx,
+                Err(e) => {
+                    warn!(
+                        session = %self.session_id,
+                        collection = %msg.descriptor.name,
+                        error = %e,
+                        "CollectionSchema: failed to propose PutCollectionIfAbsent; \
+                         collection not materialized"
+                    );
+                    return None;
+                }
+            };
+        crate::control::catalog_entry::apply::local::apply_locally_if_needed(
+            shared, &entry, log_index,
+        );
+        None
+    }
+}
diff --git a/nodedb/src/control/server/sync/session/dispatch.rs b/nodedb/src/control/server/sync/session/dispatch.rs
@@ -216,6 +216,14 @@ impl SyncSession {
                 );
                 None
             }
+            SyncMessageType::CollectionSchema => {
+                // A peer announces a collection descriptor. Materialize it
+                // into the local catalog (create-only) so it becomes
+                // catalog-visible and queryable cluster-wide; the handler
+                // handles the permissive `shared == None` path by warn+skip.
+                let msg: CollectionSchemaSyncMsg = frame.decode_body()?;
+                self.handle_collection_schema(&msg, shared)
+            }
             _ => {
                 warn!(
                     session = %self.session_id,
diff --git a/nodedb/src/control/server/sync/session/mod.rs b/nodedb/src/control/server/sync/session/mod.rs
@@ -15,6 +15,7 @@
 //! - `dispatch.rs` — `process_frame` (match on `msg_type`, route).
 
 pub mod clock_ping;
+pub mod collection_schema;
 pub mod delta;
 pub mod dispatch;
 pub mod handshake;
diff --git a/nodedb/src/control/server/sync/wire.rs b/nodedb/src/control/server/sync/wire.rs
