feat(sync): add CollectionSchema wire message for descriptor sync

Introduces an engine-agnostic CollectionDescriptor plus a
CollectionSchemaSyncMsg envelope (opcode 0x13) so any sync peer can
materialize a collection in its catalog with the correct engine and
config. Bumps the wire format version to reflect the new opcode.

Author: farhan-syah

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

@@ -0,0 +1,119 @@
+// SPDX-License-Identifier: Apache-2.0
+
+//! Standardized collection descriptor + wire envelope for sync.
+
+use serde::{Deserialize, Serialize};
+
+use crate::collection::CollectionType;
+use crate::collection_config::{PartitionStrategy, PrimaryEngine, VectorPrimaryConfig};
+use crate::hlc::Hlc;
+use crate::id::DatabaseId;
+
+/// Standardized, engine-agnostic descriptor of a collection's identity + engine
+/// config. This is the single unit that travels over sync so any peer can
+/// materialize the collection in its catalog with the correct engine. Reused by
+/// emit/receive/conversion paths in later units.
+#[derive(
+    Debug,
+    Clone,
+    PartialEq,
+    Serialize,
+    Deserialize,
+    zerompk::ToMessagePack,
+    zerompk::FromMessagePack,
+)]
+#[msgpack(map)]
+pub struct CollectionDescriptor {
+    /// Numeric tenant ID the collection belongs to.
+    pub tenant_id: u64,
+    /// Database the collection lives in.
+    pub database_id: DatabaseId,
+    /// Collection name.
+    pub name: String,
+    /// Storage engine + engine-specific configuration.
+    pub collection_type: CollectionType,
+    /// Whether the collection tracks system-time + valid-time versions.
+    #[msgpack(default)]
+    pub bitemporal: bool,
+    /// Lightweight field type hints, e.g. `[("email", "string")]`.
+    #[msgpack(default)]
+    pub fields: Vec<(String, String)>,
+    /// Which engine serves as the primary access path for this collection.
+    pub primary: PrimaryEngine,
+    /// Vector-primary configuration, present when `primary == PrimaryEngine::Vector`.
+    #[msgpack(default)]
+    pub vector_primary: Option<VectorPrimaryConfig>,
+    /// How rows are distributed across vShards.
+    pub partition_strategy: PartitionStrategy,
+    /// Explicitly declared primary key field, if any.
+    #[msgpack(default)]
+    pub declared_primary_key: Option<String>,
+    /// Monotonic version of this descriptor, bumped on schema-affecting change.
+    #[msgpack(default)]
+    pub descriptor_version: u64,
+}
+
+/// Wire envelope announcing a collection's descriptor to a sync peer (opcode 0x13).
+#[derive(
+    Debug,
+    Clone,
+    PartialEq,
+    Serialize,
+    Deserialize,
+    zerompk::ToMessagePack,
+    zerompk::FromMessagePack,
+)]
+pub struct CollectionSchemaSyncMsg {
+    /// The collection's engine-agnostic descriptor.
+    pub descriptor: CollectionDescriptor,
+    /// HLC timestamp at which this descriptor was created/announced.
+    pub creation_hlc: Hlc,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::sync::wire::{SyncFrame, SyncMessageType};
+
+    #[test]
+    fn collection_descriptor_msgpack_roundtrip() {
+        let descriptor = CollectionDescriptor {
+            tenant_id: 7,
+            database_id: DatabaseId::new(1024),
+            name: "users".into(),
+            collection_type: CollectionType::document(),
+            bitemporal: true,
+            fields: vec![("email".into(), "string".into())],
+            primary: PrimaryEngine::Document,
+            vector_primary: None,
+            partition_strategy: PartitionStrategy::CollectionHomed,
+            declared_primary_key: Some("id".into()),
+            descriptor_version: 3,
+        };
+        let msg = CollectionSchemaSyncMsg {
+            descriptor,
+            creation_hlc: Hlc {
+                wall_ns: 123,
+                logical: 1,
+            },
+        };
+        let frame = SyncFrame::new_msgpack(SyncMessageType::CollectionSchema, &msg).unwrap();
+        let bytes = frame.to_bytes();
+        let decoded = SyncFrame::from_bytes(&bytes).unwrap();
+        assert_eq!(decoded.msg_type, SyncMessageType::CollectionSchema);
+        let decoded_msg: CollectionSchemaSyncMsg = decoded.decode_body().unwrap();
+        assert_eq!(decoded_msg.descriptor.name, "users");
+        assert_eq!(decoded_msg.descriptor.tenant_id, 7);
+        assert!(decoded_msg.descriptor.bitemporal);
+        assert_eq!(decoded_msg.descriptor.primary, PrimaryEngine::Document);
+    }
+
+    #[test]
+    fn collection_schema_opcode() {
+        assert_eq!(
+            SyncMessageType::from_u8(0x13),
+            Some(SyncMessageType::CollectionSchema)
+        );
+        assert_eq!(SyncMessageType::CollectionSchema as u8, 0x13);
+    }
+}

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

@@ -26,6 +26,10 @@ pub enum SyncMessageType {
     DeltaPush = 0x10,
     DeltaAck = 0x11,
     DeltaReject = 0x12,
+    /// Collection schema descriptor announcement (bidirectional, 0x13).
+    /// Carries a `CollectionDescriptor` so any peer can materialize the
+    /// collection in its catalog with the correct engine + config.
+    CollectionSchema = 0x13,
     /// Collection purged notification (server → client, 0x14).
     /// Sent when an Origin collection is hard-deleted (UNDROP window
     /// expired or explicit `DROP COLLECTION ... PURGE`). The client
@@ -115,6 +119,7 @@ impl SyncMessageType {
             0x10 => Some(Self::DeltaPush),
             0x11 => Some(Self::DeltaAck),
             0x12 => Some(Self::DeltaReject),
+            0x13 => Some(Self::CollectionSchema),
             0x14 => Some(Self::CollectionPurged),
             0x20 => Some(Self::ShapeSubscribe),
             0x21 => Some(Self::ShapeSnapshot),

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

@@ -10,6 +10,7 @@
 //! - `0x10` DeltaPush (client → server)
 //! - `0x11` DeltaAck (server → client)
 //! - `0x12` DeltaReject (server → client)
+//! - `0x13` CollectionSchema (bidirectional)
 //! - `0x14` CollectionPurged (server → client)
 //! - `0x20` ShapeSubscribe (client → server)
 //! - `0x21` ShapeSnapshot (server → client)
@@ -53,6 +54,7 @@
 pub mod ack_result;
 pub mod ack_status;
 pub mod array;
+pub mod collection_schema;
 pub mod columnar;
 pub mod delta;
 pub mod frame;
@@ -76,6 +78,7 @@ pub use array::{
     ArrayAckMsg, ArrayCatchupRequestMsg, ArrayDeltaBatchMsg, ArrayDeltaMsg, ArrayRejectMsg,
     ArrayRejectReason, ArraySchemaSyncMsg, ArraySnapshotChunkMsg, ArraySnapshotMsg,
 };
+pub use collection_schema::{CollectionDescriptor, CollectionSchemaSyncMsg};
 pub use columnar::{ColumnarInsertAckMsg, ColumnarInsertMsg};
 pub use delta::{CollectionPurgedMsg, DeltaAckMsg, DeltaPushMsg, DeltaRejectMsg};
 pub use frame::{SyncFrame, SyncMessageType};

nodedb-types/src/wire_version.rs

@@ -16,7 +16,7 @@
 
 /// Cluster-wide wire format version. Stamped on every `NodeInfo` and
 /// returned by `nodedb::version::WIRE_FORMAT_VERSION` (a re-export).
-pub const WIRE_FORMAT_VERSION: u16 = 6;
+pub const WIRE_FORMAT_VERSION: u16 = 7;
 
 /// Minimum wire format version this build can read. Equal to
 /// `WIRE_FORMAT_VERSION`: floor == ceiling, no backward compat window.