security: harden database-scoped event processing

Author: farhan-syah

nodedb-cluster-tests/tests/descriptor_lease_cross_node.rs

@@ -21,7 +21,7 @@ const WAIT_BUDGET: Duration = Duration::from_secs(2);
 const POLL: Duration = Duration::from_millis(20);
 
 fn coll_id(name: &str) -> DescriptorId {
-    DescriptorId::new(TENANT, DescriptorKind::Collection, name.to_string())
+    DescriptorId::new(0, TENANT, DescriptorKind::Collection, name.to_string())
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 6)]

nodedb-cluster-tests/tests/descriptor_lease_drain.rs

@@ -27,7 +27,7 @@ const WAIT_BUDGET: Duration = Duration::from_secs(3);
 const POLL: Duration = Duration::from_millis(20);
 
 fn coll_id(name: &str) -> DescriptorId {
-    DescriptorId::new(TENANT, DescriptorKind::Collection, name.to_string())
+    DescriptorId::new(0, TENANT, DescriptorKind::Collection, name.to_string())
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 6)]
@@ -260,28 +260,20 @@ async fn ddl_waits_for_existing_lease_to_release() {
     );
 
     wait_for(
-        "collection stamped v2 on every node",
+        "collection stamped v2 and drain cleared on every node",
         WAIT_BUDGET,
         POLL,
         || {
-            cluster
-                .nodes
-                .iter()
-                .all(|n| n.collection_descriptor(TENANT, "drainable").map(|s| s.0) == Some(2))
+            cluster.nodes.iter().all(|node| {
+                node.collection_descriptor(TENANT, "drainable")
+                    .map(|stored| stored.0)
+                    == Some(2)
+                    && !node.has_drain_for(&coll_id("drainable"), 1)
+            })
         },
     )
     .await;
 
-    // Drain should have cleared on every node (implicit clear on
-    // successful Put* apply).
-    for node in &cluster.nodes {
-        assert!(
-            !node.has_drain_for(&coll_id("drainable"), 1),
-            "node {} still has drain entry after DDL committed",
-            node.node_id
-        );
-    }
-
     let _ = shared; // Keep the earlier binding alive to silence warnings.
     cluster.shutdown().await;
 }

nodedb-cluster-tests/tests/descriptor_lease_planner_integration.rs

@@ -21,7 +21,7 @@ const WAIT_BUDGET: Duration = Duration::from_secs(10);
 const POLL: Duration = Duration::from_millis(20);
 
 fn coll_id(name: &str) -> DescriptorId {
-    DescriptorId::new(TENANT, DescriptorKind::Collection, name.to_string())
+    DescriptorId::new(0, TENANT, DescriptorKind::Collection, name.to_string())
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 6)]

nodedb-cluster-tests/tests/sql_cluster_cross_node_dml_tests/ddl_objects.rs

@@ -5,6 +5,7 @@
 use std::time::Duration;
 
 use crate::common::cluster_harness::{TestCluster, wait_for};
+use nodedb_types::DatabaseId;
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 6)]
 async fn create_on_any_node_is_visible_on_every_node() {
@@ -289,7 +290,7 @@ async fn change_stream_create_visible_on_every_node() {
             cluster
                 .nodes
                 .iter()
-                .all(|n| n.has_change_stream(1, "event_feed"))
+                .all(|n| n.has_change_stream(DatabaseId::DEFAULT, 1, "event_feed"))
         },
     )
     .await;
@@ -307,7 +308,7 @@ async fn change_stream_create_visible_on_every_node() {
             cluster
                 .nodes
                 .iter()
-                .all(|n| !n.has_change_stream(1, "event_feed"))
+                .all(|n| !n.has_change_stream(DatabaseId::DEFAULT, 1, "event_feed"))
         },
     )
     .await;

nodedb-cluster/src/metadata_group/descriptors/common.rs

@@ -7,10 +7,11 @@ use serde::{Deserialize, Serialize};
 
 use crate::metadata_group::state::DescriptorState;
 
-/// Globally unique, tenant-scoped identifier for a schema descriptor.
+/// Globally unique, database- and tenant-scoped identifier for a schema descriptor.
 ///
-/// `kind` disambiguates object types sharing the same `(tenant_id, name)`
-/// (e.g. a collection and an index can both be named `orders`).
+/// `kind` disambiguates object types sharing the same
+/// `(database_id, tenant_id, name)` (e.g. a collection and an index can both
+/// be named `orders`).
 #[derive(
     Debug,
     Clone,
@@ -23,21 +24,51 @@ use crate::metadata_group::state::DescriptorState;
     zerompk::FromMessagePack,
 )]
 pub struct DescriptorId {
+    pub database_id: u64,
     pub tenant_id: u64,
     pub kind: DescriptorKind,
     pub name: String,
 }
 
 impl DescriptorId {
-    pub fn new(tenant_id: u64, kind: DescriptorKind, name: impl Into<String>) -> Self {
+    pub fn new(
+        database_id: u64,
+        tenant_id: u64,
+        kind: DescriptorKind,
+        name: impl Into<String>,
+    ) -> Self {
         Self {
+            database_id,
             tenant_id,
             kind,
             name: name.into(),
         }
     }
 }
 
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::collections::HashSet;
+
+    #[test]
+    fn database_id_participates_in_equality_hash_and_wire_format() {
+        let alpha = DescriptorId::new(7, 1, DescriptorKind::Collection, "orders");
+        let beta = DescriptorId::new(8, 1, DescriptorKind::Collection, "orders");
+
+        assert_ne!(alpha, beta);
+        let mut ids = HashSet::new();
+        ids.insert(alpha.clone());
+        ids.insert(beta.clone());
+        assert_eq!(ids.len(), 2);
+
+        let encoded = zerompk::to_msgpack_vec(&alpha).expect("encode descriptor id");
+        let decoded: DescriptorId = zerompk::from_msgpack(&encoded).expect("decode descriptor id");
+        assert_eq!(decoded, alpha);
+        assert_eq!(decoded.database_id, 7);
+    }
+}
+
 /// Discriminant for [`DescriptorId`] — one variant per schema object type.
 #[derive(
     Debug,

nodedb-physical/src/physical_plan/cluster_event.rs

@@ -5,6 +5,14 @@
 //! These plans are cluster-RPC envelopes only. They must never cross the
 //! Control Plane → Data Plane bridge.
 
+use nodedb_types::DatabaseId;
+
+/// Hard cap for committed CDC cursors supplied in one cluster consume request.
+///
+/// This bounds a caller-controlled wire vector while still allowing one cursor
+/// for every partition in a reasonably sized routed stream.
+pub const MAX_REMOTE_CDC_COMMITTED_OFFSETS: usize = 4_096;
+
 /// Cluster-routed Event-Plane operation.
 #[derive(
     Debug,
@@ -16,29 +24,42 @@
     zerompk::FromMessagePack,
 )]
 pub enum ClusterEventOp {
-    /// Consume one CDC partition from the leader node's local event buffer.
+    /// Consume CDC events from the leader node's local event buffer.
+    ///
+    /// `committed_offsets` belongs to the caller node: each tuple is
+    /// `(partition_id, lsn, sequence)`. The receiver must use these cursors
+    /// rather than its own consumer-group offset store. Missing partitions
+    /// start at the initial `(0, 0)` position.
     ConsumeStream {
+        database_id: DatabaseId,
         stream_name: String,
         group_name: String,
-        partition: u32,
+        partition: Option<u32>,
         limit: u64,
+        committed_offsets: Vec<(u32, u64, u64)>,
     },
     /// Publish one durable-topic message on the topic's home node.
-    PublishTopic { topic_name: String, payload: String },
+    PublishTopic {
+        database_id: DatabaseId,
+        topic_name: String,
+        payload: String,
+    },
 }
 
 #[cfg(test)]
 mod tests {
-    use super::ClusterEventOp;
+    use super::{ClusterEventOp, DatabaseId};
     use crate::physical_plan::{PhysicalPlan, wire};
 
     #[test]
     fn cluster_event_plan_roundtrips_over_cluster_wire() {
         let plan = PhysicalPlan::ClusterEvent(ClusterEventOp::ConsumeStream {
+            database_id: DatabaseId::new(7),
             stream_name: "orders; no SQL".into(),
             group_name: "Analytics".into(),
-            partition: 7,
+            partition: Some(7),
             limit: 128,
+            committed_offsets: vec![(7, 42, 3)],
         });
         let encoded = wire::encode(&plan).expect("encode typed cluster event");
         assert_eq!(

nodedb-physical/src/physical_plan/mod.rs

@@ -27,7 +27,7 @@ pub mod wire;
 
 pub use array::{ArrayBinaryOp, ArrayOp, ArrayReducer};
 pub use cluster_array::ClusterArrayOp;
-pub use cluster_event::ClusterEventOp;
+pub use cluster_event::{ClusterEventOp, MAX_REMOTE_CDC_COMMITTED_OFFSETS};
 pub use columnar::{ColumnarInsertIntent, ColumnarOp};
 pub use crdt::CrdtOp;
 pub use document::{

nodedb-test-support/src/cluster_harness/node/inspect/catalog.rs

@@ -85,8 +85,16 @@ impl TestClusterNode {
 
     /// Check whether a change stream with the given name exists in
     /// this node's in-memory `stream_registry`.
-    pub fn has_change_stream(&self, tenant_id: u64, name: &str) -> bool {
-        self.shared.stream_registry.get(tenant_id, name).is_some()
+    pub fn has_change_stream(
+        &self,
+        database_id: nodedb_types::DatabaseId,
+        tenant_id: u64,
+        name: &str,
+    ) -> bool {
+        self.shared
+            .stream_registry
+            .get(database_id, tenant_id, name)
+            .is_some()
     }
 
     /// Check whether a user exists and is active in this node's

nodedb-test-support/src/cluster_harness/node/inspect/lease.rs

@@ -59,7 +59,12 @@ impl TestClusterNode {
         min_version: u64,
     ) -> bool {
         let now = self.shared.hlc_clock.peek();
-        let id = nodedb_cluster::DescriptorId::new(tenant_id, kind, name);
+        let id = nodedb_cluster::DescriptorId::new(
+            nodedb_types::DatabaseId::DEFAULT.as_u64(),
+            tenant_id,
+            kind,
+            name,
+        );
         let cache = self
             .shared
             .metadata_cache
@@ -79,7 +84,12 @@ impl TestClusterNode {
         tenant_id: u64,
         name: &str,
     ) -> Vec<nodedb_cluster::DescriptorLease> {
-        let id = nodedb_cluster::DescriptorId::new(tenant_id, kind, name);
+        let id = nodedb_cluster::DescriptorId::new(
+            nodedb_types::DatabaseId::DEFAULT.as_u64(),
+            tenant_id,
+            kind,
+            name,
+        );
         let cache = self
             .shared
             .metadata_cache
@@ -107,7 +117,12 @@ impl TestClusterNode {
         version: u64,
         duration: std::time::Duration,
     ) -> Result<nodedb_cluster::DescriptorLease, String> {
-        let id = nodedb_cluster::DescriptorId::new(tenant_id, kind, name.to_string());
+        let id = nodedb_cluster::DescriptorId::new(
+            nodedb_types::DatabaseId::DEFAULT.as_u64(),
+            tenant_id,
+            kind,
+            name.to_string(),
+        );
         self.shared
             .acquire_descriptor_lease(id, version, duration)
             .map_err(|e| format!("acquire failed: {e}"))

nodedb-test-support/src/cluster_harness/node/lifecycle/spawn_full.rs

@@ -16,7 +16,7 @@ use nodedb::config::auth::AuthMode;
 use nodedb::config::server::ClusterSettings;
 use nodedb::control::server::pgwire::listener::PgListener;
 use nodedb::control::state::SharedState;
-use nodedb::event::{EventPlane, create_event_bus};
+use nodedb::event::{EventPlane, EventPlaneConfig, create_event_bus};
 use nodedb::wal::WalManager;
 
 use crate::cluster_harness::cluster::ClusterSpawnConfig;
@@ -270,15 +270,18 @@ impl TestClusterNode {
         let trigger_dlq = Arc::new(std::sync::Mutex::new(
             nodedb::event::trigger::TriggerDlq::open(&data_dir_path)?,
         ));
-        let event_plane = EventPlane::spawn(
-            event_consumers,
-            Arc::clone(&wal),
+        let (pg_shutdown_bus, _) =
+            nodedb::control::shutdown::ShutdownBus::new(Arc::clone(&shared.shutdown));
+        let event_plane = EventPlane::spawn(EventPlaneConfig {
+            consumers_rx: event_consumers,
+            wal: Arc::clone(&wal),
             watermark_store,
-            Arc::clone(&shared),
+            shared_state: Arc::clone(&shared),
             trigger_dlq,
-            Arc::clone(&shared.cdc_router),
-            Arc::clone(&shared.shutdown),
-        );
+            cdc_router: Arc::clone(&shared.cdc_router),
+            shutdown: Arc::clone(&shared.shutdown),
+            shutdown_bus: pg_shutdown_bus.clone(),
+        });
 
         // Start Raft + install MetadataCommitApplier.
         let (cluster_shutdown_tx, cluster_shutdown_rx) = tokio::sync::watch::channel(false);
@@ -339,8 +342,6 @@ impl TestClusterNode {
         // accepts immediately without a startup-phase delay.
         let pg_listener = PgListener::bind("127.0.0.1:0".parse()?).await?;
         let pg_addr = pg_listener.local_addr();
-        let (pg_shutdown_bus, _) =
-            nodedb::control::shutdown::ShutdownBus::new(Arc::clone(&shared.shutdown));
         let shared_pg = Arc::clone(&shared);
         let test_startup_gate = Arc::clone(&shared.startup);
         let bus_pg = pg_shutdown_bus.clone();