feat(sync): announce collection schema before shape and delta data

farhan-syah · farhan-syah · commit 676df0097996 · 2026-07-02T08:03:33.000+08:00
Emit side of the announce-precedes-data invariant: a session now tracks
which collections it has already announced, and a shared helper builds
the CollectionSchema frame from the catalog descriptor on demand. Both
the shape-subscribe path and the delta-push path send this announce
frame before their first data frame for a collection, so peers always
receive the descriptor ahead of any data referencing it.
diff --git a/nodedb-cluster-tests/tests/sync_collection_schema_emit.rs b/nodedb-cluster-tests/tests/sync_collection_schema_emit.rs
@@ -0,0 +1,143 @@
+// SPDX-License-Identifier: BUSL-1.1
+//! The EMIT side of collection-schema sync: a `CollectionSchema` frame
+//! (opcode 0x13) is written to the peer STRICTLY BEFORE the first data frame
+//! for a collection.
+//!
+//! ## What this guards
+//!
+//! The announce-precedes-data invariant. Before the server ships a shape
+//! snapshot (or a delta) for a collection to a sync peer, it must first send
+//! that collection's `CollectionDescriptor` so the peer can materialize the
+//! collection with the correct engine before any rows arrive. The announce is
+//! idempotent per session.
+//!
+//! The test drives the sync WebSocket end-to-end: it connects to node 0's
+//! sync listener, materializes a document collection on the server (via a
+//! client-side `CollectionSchema` announce, which does NOT mark the server
+//! session's emit-announce set), then subscribes to a Document shape for that
+//! collection and asserts the FIRST frame back is `CollectionSchema` and the
+//! NEXT is `ShapeSnapshot`.
+
+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::sync::wire::{CollectionDescriptor, SyncMessageType};
+use nodedb_types::{CollectionType, DatabaseId, Hlc};
+
+/// Trust-mode sync sessions authenticate as tenant 1.
+const TENANT: u64 = 1;
+
+/// Build a plain-document descriptor for `name`, mirroring what a local
+/// `CREATE COLLECTION` would emit over sync.
+fn document_descriptor(name: &str) -> CollectionDescriptor {
+    let collection_type = CollectionType::document();
+    CollectionDescriptor {
+        tenant_id: TENANT,
+        database_id: DatabaseId::DEFAULT,
+        name: name.to_string(),
+        partition_strategy: PartitionStrategy::default_for_collection_type(&collection_type),
+        collection_type,
+        bitemporal: false,
+        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. Panics if it
+/// does not converge — a bounded retry loop, not a single fixed sleep.
+async fn await_collection(node: &TestClusterNode, name: &str) {
+    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 found.is_some() {
+            return;
+        }
+        if Instant::now() >= deadline {
+            panic!("collection '{name}' did not become catalog-visible within 30s");
+        }
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn collection_schema_precedes_shape_snapshot() {
+    let cluster = TestCluster::spawn_three()
+        .await
+        .expect("spawn three-node cluster");
+
+    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");
+
+    // Materialize the collection on the server. This is the client→server
+    // announce (receive side); it does NOT mark the server session's
+    // emit-announce set, so the subscription below still triggers an emit.
+    let name = "emit_doc";
+    client
+        .push_collection_schema(document_descriptor(name), Hlc::new(1, 0))
+        .await
+        .expect("push CollectionSchema to materialize collection");
+
+    // Wait until the receiving node's catalog can resolve the collection —
+    // the emit path resolves the descriptor from exactly this catalog.
+    await_collection(&cluster.nodes[0], name).await;
+
+    // Subscribe to a Document shape for the collection.
+    client
+        .subscribe_document_shape("emit_shape", name, TENANT as u32)
+        .await
+        .expect("send ShapeSubscribe");
+
+    // The first frame back MUST be the collection-schema announce.
+    let first = client
+        .recv_next_frame()
+        .await
+        .expect("receive first frame after subscribe");
+    assert_eq!(
+        first.msg_type,
+        SyncMessageType::CollectionSchema,
+        "expected CollectionSchema announce before any shape data, got {:?}",
+        first.msg_type
+    );
+    let announced: nodedb_types::sync::wire::CollectionSchemaSyncMsg =
+        first.decode_body().expect("decode CollectionSchemaSyncMsg");
+    assert_eq!(
+        announced.descriptor.name, name,
+        "announced descriptor is for the wrong collection"
+    );
+
+    // The next frame MUST be the shape snapshot — data strictly after schema.
+    let second = client
+        .recv_next_frame()
+        .await
+        .expect("receive second frame after subscribe");
+    assert_eq!(
+        second.msg_type,
+        SyncMessageType::ShapeSnapshot,
+        "expected ShapeSnapshot after the CollectionSchema announce, got {:?}",
+        second.msg_type
+    );
+
+    cluster.shutdown().await;
+}
diff --git a/nodedb-test-support/src/sync_client.rs b/nodedb-test-support/src/sync_client.rs
@@ -155,6 +155,47 @@ impl SyncTestClient {
             .map_err(|e| format!("failed to send CollectionSchema frame: {e}"))?;
         Ok(())
     }
+
+    /// Subscribe to a Document shape covering all documents in `collection`
+    /// (empty predicate). Fire-and-forget: the server's response frames — a
+    /// `CollectionSchema` announce followed by the `ShapeSnapshot` — are read
+    /// by the caller via `recv_next_frame`.
+    pub async fn subscribe_document_shape(
+        &mut self,
+        shape_id: &str,
+        collection: &str,
+        tenant_id: u32,
+    ) -> Result<(), String> {
+        use nodedb_types::sync::shape::{ShapeDefinition, ShapeType};
+        use nodedb_types::sync::wire::ShapeSubscribeMsg;
+
+        let shape = ShapeDefinition {
+            shape_id: shape_id.to_string(),
+            tenant_id,
+            shape_type: ShapeType::Document {
+                collection: collection.to_string(),
+                predicate: Vec::new(),
+            },
+            description: String::new(),
+            field_filter: Vec::new(),
+        };
+        let msg = ShapeSubscribeMsg { shape };
+        let frame = SyncFrame::new_msgpack(SyncMessageType::ShapeSubscribe, &msg)
+            .ok_or_else(|| "failed to encode ShapeSubscribeMsg".to_string())?;
+        self.ws
+            .send(Message::Binary(frame.to_bytes().into()))
+            .await
+            .map_err(|e| format!("failed to send ShapeSubscribe frame: {e}"))?;
+        Ok(())
+    }
+
+    /// Receive the next binary sync frame from the server, decoded as a
+    /// `SyncFrame`. Non-binary frames are skipped. Bounded internally by
+    /// `RECV_TIMEOUT`, so callers do not need to wrap this in a timeout to
+    /// avoid hanging on a silent server.
+    pub async fn recv_next_frame(&mut self) -> Result<SyncFrame, String> {
+        recv_any_frame(&mut self.ws).await
+    }
 }
 
 /// Receive the next binary frame, decode it as a `SyncFrame`, and error out
diff --git a/nodedb/src/control/server/sync/session/state.rs b/nodedb/src/control/server/sync/session/state.rs
@@ -48,6 +48,11 @@ pub struct SyncSession {
     /// Origin `CollectionPurged` broadcast to decide which sessions
     /// need to be notified when a collection is hard-deleted.
     pub tracked_collections: std::collections::HashSet<(u64, String)>,
+    /// Collections whose descriptor has already been announced to the peer
+    /// this session via a `CollectionSchema` frame. Enforces the
+    /// announce-precedes-data guard: a collection's schema is emitted at most
+    /// once per session, strictly before its first shape snapshot or delta.
+    pub announced_collections: std::collections::HashSet<String>,
     /// Last WAL LSN the client advertised in its vector clock at
     /// handshake. Used by offline-client replay to identify
     /// `CollectionPurged` events that committed while the client
@@ -86,6 +91,7 @@ impl SyncSession {
             rate_limiter: SyncRateLimiter::new(rate_config),
             device_metadata: DeviceMetadata::default(),
             tracked_collections: std::collections::HashSet::new(),
+            announced_collections: std::collections::HashSet::new(),
             last_seen_lsn: 0,
             producer_id: 0,
             accepted_epoch: 0,
diff --git a/nodedb/src/control/server/sync/session_handler/announce.rs b/nodedb/src/control/server/sync/session_handler/announce.rs
@@ -0,0 +1,65 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! Collection-schema announce helper.
+//!
+//! Builds the `CollectionSchema` frame that must precede the first shape
+//! snapshot or delta for a collection on a sync session. The emit side of
+//! the announce-precedes-data invariant: any peer that receives a delta or
+//! snapshot for a collection has already received its descriptor.
+
+use nodedb_types::sync::wire::{CollectionDescriptor, CollectionSchemaSyncMsg};
+
+use tracing::warn;
+
+use crate::control::state::SharedState;
+
+use super::super::session::SyncSession;
+use super::super::wire::{SyncFrame, SyncMessageType};
+
+/// Build a `CollectionSchema` announce frame for `collection` if it has not
+/// already been announced to the peer this session.
+///
+/// Returns `None` when the collection was already announced (idempotent) or
+/// cannot be resolved in the catalog (logged; the caller then proceeds to
+/// send the data frame anyway so delivery never regresses). The caller is
+/// responsible for marking the collection announced only after the returned
+/// frame has been sent successfully.
+pub(in crate::control::server::sync) fn build_collection_schema_frame(
+    shared: &SharedState,
+    session: &SyncSession,
+    tenant_id: u64,
+    collection: &str,
+) -> Option<SyncFrame> {
+    if session.announced_collections.contains(collection) {
+        return None;
+    }
+
+    let stored = shared
+        .credentials
+        .catalog()
+        .as_ref()
+        .and_then(|c| {
+            c.get_collection(crate::types::DatabaseId::DEFAULT, tenant_id, collection)
+                .ok()
+        })
+        .flatten();
+
+    let Some(stored) = stored else {
+        warn!(
+            session = %session.session_id,
+            collection,
+            tenant_id,
+            "sync: collection not found in catalog; cannot announce schema before data"
+        );
+        return None;
+    };
+
+    let msg = CollectionSchemaSyncMsg {
+        descriptor: CollectionDescriptor::from(&stored),
+        // Not correctness-load-bearing: the receive side materializes the
+        // collection create-only and does not consume this timestamp.
+        creation_hlc: nodedb_types::hlc::Hlc::ZERO,
+    };
+
+    SyncFrame::new_msgpack(SyncMessageType::CollectionSchema, &msg)
+}
diff --git a/nodedb/src/control/server/sync/session_handler/mod.rs b/nodedb/src/control/server/sync/session_handler/mod.rs
@@ -1,5 +1,6 @@
 // SPDX-License-Identifier: BUSL-1.1
 
+mod announce;
 mod array;
 mod engine_dispatch;
 mod session_loop;
diff --git a/nodedb/src/control/server/sync/session_handler/session_loop.rs b/nodedb/src/control/server/sync/session_handler/session_loop.rs
@@ -129,18 +129,46 @@ pub(in crate::control::server::sync) async fn handle_sync_session(
                             )
                             .await
                     {
-                        if session.authenticated
-                            && presence_registered
-                            && let Some(sub_msg) =
-                                frame.decode_body::<super::super::wire::ShapeSubscribeMsg>()
+                        // Decode once, reused for both the presence-channel
+                        // subscribe and the schema-announce below (avoids a
+                        // redundant second msgpack decode of the same body).
+                        let shape_sub_msg = if session.authenticated {
+                            frame.decode_body::<super::super::wire::ShapeSubscribeMsg>()
+                        } else {
+                            None
+                        };
+
+                        if let Some(sub_msg) = shape_sub_msg.as_ref()
                             && let Some(coll) = sub_msg.shape.collection()
                         {
-                            let channel = format!("shape:{coll}");
-                            shared
-                                .presence
-                                .write()
-                                .await
-                                .subscribe_to_channel(&session_id, &channel);
+                            if presence_registered {
+                                let channel = format!("shape:{coll}");
+                                shared
+                                    .presence
+                                    .write()
+                                    .await
+                                    .subscribe_to_channel(&session_id, &channel);
+                            }
+
+                            // Announce the collection descriptor before the shape
+                            // snapshot so schema strictly precedes data on the
+                            // subscription path. Idempotent per session; skips shape
+                            // variants that carry no single collection.
+                            let tenant_id = session.tenant_id.map(|t| t.as_u64()).unwrap_or(0);
+                            if let Some(schema_frame) =
+                                super::announce::build_collection_schema_frame(
+                                    shared, &session, tenant_id, coll,
+                                )
+                            {
+                                if ws
+                                    .send(Message::Binary(schema_frame.to_bytes().into()))
+                                    .await
+                                    .is_err()
+                                {
+                                    break;
+                                }
+                                session.announced_collections.insert(coll.to_string());
+                            }
                         }
 
                         if ws
@@ -351,6 +379,29 @@ pub(in crate::control::server::sync) async fn handle_sync_session(
 
         if let Some(ref mut rx) = crdt_delivery_rx {
             while let Ok(delta) = rx.try_recv() {
+                // Announce the collection descriptor before its first delta so
+                // schema strictly precedes data on the peer. Idempotent per
+                // session; a lookup miss warns and proceeds without marking.
+                if let Some(shared) = shared.as_ref()
+                    && let Some(schema_frame) = super::announce::build_collection_schema_frame(
+                        shared,
+                        &session,
+                        delta.tenant_id,
+                        &delta.collection,
+                    )
+                {
+                    if ws
+                        .send(Message::Binary(schema_frame.to_bytes().into()))
+                        .await
+                        .is_err()
+                    {
+                        break;
+                    }
+                    session
+                        .announced_collections
+                        .insert(delta.collection.clone());
+                }
+
                 let push_msg = nodedb_types::sync::wire::DeltaPushMsg {
                     collection: delta.collection,
                     document_id: delta.document_id,
