feat(shard): Implement shard router

Cargo.toml

@@ -127,6 +127,7 @@ configs_derive = { path = "core/configs_derive", version = "0.1.0" }
 consensus = { path = "core/consensus" }
 console-subscriber = "0.5.0"
 crossbeam = "0.8.4"
+crossfire = "3.1.4"
 ctor = "0.6.3"
 ctrlc = { version = "3.5", features = ["termination"] }
 cucumber = "0.22"

DEPENDENCIES.md

@@ -210,6 +210,7 @@ crossbeam-deque: 0.8.6, "Apache-2.0 OR MIT",
 crossbeam-epoch: 0.9.18, "Apache-2.0 OR MIT",
 crossbeam-queue: 0.3.12, "Apache-2.0 OR MIT",
 crossbeam-utils: 0.8.21, "Apache-2.0 OR MIT",
+crossfire: 3.1.4, "Apache-2.0",
 crossterm: 0.29.0, "MIT",
 crossterm_winapi: 0.9.1, "MIT",
 crunchy: 0.2.4, "MIT",

core/common/src/types/consensus/header.rs

@@ -124,6 +124,46 @@ pub enum Operation {
     Reserved = 200,
 }
 
+impl Operation {
+    /// Returns `true` for metadata / control-plane operations (streams, topics,
+    /// users, consumer groups, etc.) that are always handled by shard 0.
+    #[inline]
+    pub fn is_metadata(&self) -> bool {
+        matches!(
+            self,
+            Operation::CreateStream
+                | Operation::UpdateStream
+                | Operation::DeleteStream
+                | Operation::PurgeStream
+                | Operation::CreateTopic
+                | Operation::UpdateTopic
+                | Operation::DeleteTopic
+                | Operation::PurgeTopic
+                | Operation::CreatePartitions
+                | Operation::DeletePartitions
+                | Operation::CreateConsumerGroup
+                | Operation::DeleteConsumerGroup
+                | Operation::CreateUser
+                | Operation::UpdateUser
+                | Operation::DeleteUser
+                | Operation::ChangePassword
+                | Operation::UpdatePermissions
+                | Operation::CreatePersonalAccessToken
+                | Operation::DeletePersonalAccessToken
+        )
+    }
+
+    /// Returns `true` for data-plane operations that are routed to the shard
+    /// owning the partition identified by the message's namespace.
+    #[inline]
+    pub fn is_partition(&self) -> bool {
+        matches!(
+            self,
+            Operation::SendMessages | Operation::StoreConsumerOffset | Operation::DeleteSegments
+        )
+    }
+}
+
 #[repr(C)]
 #[derive(Debug, Clone, Copy)]
 pub struct GenericHeader {

core/metadata/src/impls/metadata.rs

@@ -25,8 +25,7 @@ use consensus::{
 };
 use iggy_common::{
     header::{
-        Command2, ConsensusHeader, GenericHeader, Operation, PrepareHeader, PrepareOkHeader,
-        RequestHeader,
+        Command2, ConsensusHeader, GenericHeader, PrepareHeader, PrepareOkHeader, RequestHeader,
     },
     message::Message,
 };
@@ -260,30 +259,7 @@ where
             message.header().command(),
             Command2::Request | Command2::Prepare | Command2::PrepareOk
         ));
-        let operation = message.header().operation();
-        // TODO: Use better selection, smth like greater or equal based on op number.
-        matches!(
-            operation,
-            Operation::CreateStream
-                | Operation::UpdateStream
-                | Operation::DeleteStream
-                | Operation::PurgeStream
-                | Operation::CreateTopic
-                | Operation::UpdateTopic
-                | Operation::DeleteTopic
-                | Operation::PurgeTopic
-                | Operation::CreatePartitions
-                | Operation::DeletePartitions
-                | Operation::CreateConsumerGroup
-                | Operation::DeleteConsumerGroup
-                | Operation::CreateUser
-                | Operation::UpdateUser
-                | Operation::DeleteUser
-                | Operation::ChangePassword
-                | Operation::UpdatePermissions
-                | Operation::CreatePersonalAccessToken
-                | Operation::DeletePersonalAccessToken
-        )
+        message.header().operation().is_metadata()
     }
 }
 

core/partitions/src/iggy_partitions.rs

@@ -514,12 +514,7 @@ where
             message.header().command(),
             Command2::Request | Command2::Prepare | Command2::PrepareOk
         ));
-        let operation = message.header().operation();
-        // TODO: Use better selection, smth like greater or equal based on op number.
-        matches!(
-            operation,
-            Operation::DeleteSegments | Operation::SendMessages | Operation::StoreConsumerOffset
-        )
+        message.header().operation().is_partition()
     }
 }
 

core/shard/Cargo.toml

@@ -22,8 +22,13 @@ edition = "2024"
 
 [dependencies]
 consensus = { path = "../consensus" }
+crossfire = { workspace = true }
+futures = { workspace = true }
+hash32 = { workspace = true }
 iggy_common = { path = "../common" }
 journal = { path = "../journal" }
 message_bus = { path = "../message_bus" }
 metadata = { path = "../metadata" }
+papaya = { workspace = true }
 partitions = { path = "../partitions" }
+tracing = { workspace = true }

core/shard/src/lib.rs

@@ -15,10 +15,10 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use consensus::{
-    MetadataHandle, MuxPlane, NamespacedPipeline, PartitionsHandle, Plane, PlaneIdentity,
-    VsrConsensus,
-};
+mod router;
+pub mod shards_table;
+
+use consensus::{MuxPlane, NamespacedPipeline, PartitionsHandle, Plane, VsrConsensus};
 use iggy_common::header::{GenericHeader, PrepareHeader, PrepareOkHeader, RequestHeader};
 use iggy_common::message::{Message, MessageBag};
 use iggy_common::sharding::IggyNamespace;
@@ -28,6 +28,7 @@ use message_bus::MessageBus;
 use metadata::IggyMetadata;
 use metadata::stm::StateMachine;
 use partitions::IggyPartitions;
+use shards_table::ShardsTable;
 
 pub type ShardPlane<B, J, S, M> = MuxPlane<
     variadic!(
@@ -36,30 +37,116 @@ pub type ShardPlane<B, J, S, M> = MuxPlane<
     ),
 >;
 
-pub struct IggyShard<B, J, S, M>
+/// Bounded mpsc channel sender (blocking send).
+pub type Sender<T> = crossfire::MTx<crossfire::mpsc::Array<T>>;
+
+/// Bounded mpsc channel receiver (async recv).
+pub type Receiver<T> = crossfire::AsyncRx<crossfire::mpsc::Array<T>>;
+
+/// Create a bounded mpsc channel with a blocking sender and async receiver.
+pub fn channel<T: Send + 'static>(capacity: usize) -> (Sender<T>, Receiver<T>) {
+    crossfire::mpsc::bounded_blocking_async(capacity)
+}
+
+/// Envelope for inter-shard channel messages.
+///
+/// Wraps a consensus [`Message`] together with an optional one-shot response
+/// channel.  Fire-and-forget dispatches leave `response_sender` as `None`;
+/// request-response dispatches provide a sender that the message pump will
+/// notify once the message has been processed.
+///
+/// The response type `R` is generic so that higher layers (e.g. HTTP handlers)
+/// can carry a response enum while the consensus layer can default to `()`.
+pub struct ShardFrame<R: Send + 'static = ()> {
+    pub message: Message<GenericHeader>,
+    pub response_sender: Option<Sender<R>>,
+}
+
+impl<R: Send + 'static> ShardFrame<R> {
+    /// Create a fire-and-forget frame (no caller waiting for completion).
+    pub fn fire_and_forget(message: Message<GenericHeader>) -> Self {
+        Self {
+            message,
+            response_sender: None,
+        }
+    }
+
+    /// Create a request-response frame.  Returns the frame and a receiver
+    /// that the caller can await for completion notification.
+    pub fn with_response(message: Message<GenericHeader>) -> (Self, Receiver<R>) {
+        let (tx, rx) = channel(1);
+        (
+            Self {
+                message,
+                response_sender: Some(tx),
+            },
+            rx,
+        )
+    }
+}
+
+pub struct IggyShard<B, J, S, M, T = (), R: Send + 'static = ()>
 where
     B: MessageBus,
 {
-    pub id: u8,
+    pub id: u16,
     pub name: String,
     pub plane: ShardPlane<B, J, S, M>,
+
+    /// Channel senders to every shard, indexed by shard id.
+    /// Includes a sender to self so that local routing goes through the
+    /// same channel path as remote routing.
+    senders: Vec<Sender<ShardFrame<R>>>,
+
+    /// Receiver end of this shard's inbox.  Peer shards (and self) send
+    /// messages here via the corresponding sender.
+    inbox: Receiver<ShardFrame<R>>,
+
+    /// Partition namespace -> owning shard lookup.
+    shards_table: T,
 }
 
-impl<B, J, S, M> IggyShard<B, J, S, M>
+impl<B, J, S, M, T, R: Send + 'static> IggyShard<B, J, S, M, T, R>
 where
     B: MessageBus,
+    T: ShardsTable,
 {
-    /// Create a new shard from pre-built metadata and partition planes.
+    /// Create a new shard with channel links and a shards table.
+    ///
+    /// * `senders` - one sender per shard in the cluster (indexed by shard id).
+    /// * `inbox` - the receiver that this shard drains in its message pump.
+    /// * `shards_table` - namespace -> shard routing table.
     pub fn new(
-        id: u8,
+        id: u16,
         name: String,
         metadata: IggyMetadata<VsrConsensus<B>, J, S, M>,
         partitions: IggyPartitions<VsrConsensus<B, NamespacedPipeline>>,
+        senders: Vec<Sender<ShardFrame<R>>>,
+        inbox: Receiver<ShardFrame<R>>,
+        shards_table: T,
     ) -> Self {
         let plane = MuxPlane::new(variadic!(metadata, partitions));
-        Self { id, name, plane }
+        Self {
+            id,
+            name,
+            plane,
+            senders,
+            inbox,
+            shards_table,
+        }
     }
 
+    pub fn shards_table(&self) -> &T {
+        &self.shards_table
+    }
+}
+
+/// Local message processing — these methods handle messages that have been
+/// routed to this shard via the message pump.
+impl<B, J, S, M, T, R: Send + 'static> IggyShard<B, J, S, M, T, R>
+where
+    B: MessageBus,
+{
     /// Dispatch an incoming network message to the appropriate consensus plane.
     ///
     /// Routes requests, replication messages, and acks to either the metadata
@@ -93,11 +180,7 @@ where
             >,
         M: StateMachine<Input = Message<PrepareHeader>>,
     {
-        if self.plane.metadata().is_applicable(&request) {
-            self.plane.metadata().on_request(request).await;
-        } else {
-            self.plane.partitions().on_request(request).await;
-        }
+        self.plane.on_request(request).await;
     }
 
     pub async fn on_replicate(&self, prepare: Message<PrepareHeader>)
@@ -111,11 +194,7 @@ where
             >,
         M: StateMachine<Input = Message<PrepareHeader>>,
     {
-        if self.plane.metadata().is_applicable(&prepare) {
-            self.plane.metadata().on_replicate(prepare).await;
-        } else {
-            self.plane.partitions().on_replicate(prepare).await;
-        }
+        self.plane.on_replicate(prepare).await;
     }
 
     pub async fn on_ack(&self, prepare_ok: Message<PrepareOkHeader>)
@@ -129,11 +208,7 @@ where
             >,
         M: StateMachine<Input = Message<PrepareHeader>>,
     {
-        if self.plane.metadata().is_applicable(&prepare_ok) {
-            self.plane.metadata().on_ack(prepare_ok).await;
-        } else {
-            self.plane.partitions().on_ack(prepare_ok).await;
-        }
+        self.plane.on_ack(prepare_ok).await;
     }
 
     /// Drain and dispatch loopback messages for each consensus plane.