feat(simulator): implement bare bone version of simulator

.github/workflows/_common.yml

@@ -107,6 +107,7 @@ jobs:
             metadata
             message_bus
             storage
+            simulator
             configs
 
   license-headers:

Cargo.toml

@@ -50,6 +50,7 @@ members = [
     "core/partitions",
     "core/sdk",
     "core/server",
+    "core/simulator",
     "core/tools",
     "examples/rust",
 ]

DEPENDENCIES.md

@@ -730,6 +730,7 @@ simd-adler32: 0.3.8, "MIT",
 simd-json: 0.17.0, "Apache-2.0 OR MIT",
 simdutf8: 0.1.5, "Apache-2.0 OR MIT",
 simple_asn1: 0.6.3, "ISC",
+simulator: 0.1.0, "N/A",
 siphasher: 1.0.1, "Apache-2.0 OR MIT",
 slab: 0.4.11, "MIT",
 smallvec: 1.15.1, "Apache-2.0 OR MIT",

core/common/src/sender/mod.rs

@@ -86,6 +86,7 @@ pub trait Sender {
 }
 
 #[allow(clippy::large_enum_variant)]
+#[derive(Debug)]
 pub enum SenderKind {
     Tcp(TcpSender),
     TcpTls(TcpTlsSender),

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

@@ -114,6 +114,8 @@ pub enum Operation {
     UpdatePermissions = 145,
     CreatePersonalAccessToken = 146,
     DeletePersonalAccessToken = 147,
+
+    Reserved = 200,
 }
 
 #[repr(C)]
@@ -172,6 +174,30 @@ pub struct RequestHeader {
     pub reserved: [u8; 95],
 }
 
+impl Default for RequestHeader {
+    fn default() -> Self {
+        Self {
+            reserved: [0; 95],
+            checksum: 0,
+            checksum_body: 0,
+            cluster: 0,
+            size: 0,
+            epoch: 0,
+            view: 0,
+            release: 0,
+            protocol: 0,
+            command: Default::default(),
+            replica: 0,
+            reserved_frame: [0; 12],
+            client: 0,
+            request_checksum: 0,
+            timestamp: 0,
+            request: 0,
+            operation: Default::default(),
+        }
+    }
+}
+
 unsafe impl Pod for RequestHeader {}
 unsafe impl Zeroable for RequestHeader {}
 
@@ -348,7 +374,7 @@ impl ConsensusHeader for CommitHeader {
 }
 
 #[repr(C)]
-#[derive(Debug, Clone, Copy)]
+#[derive(Default, Debug, Clone, Copy)]
 pub struct ReplyHeader {
     pub checksum: u128,
     pub checksum_body: u128,

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

@@ -15,8 +15,9 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use crate::types::consensus::header::{
-    self, CommitHeader, ConsensusHeader, GenericHeader, PrepareHeader, PrepareOkHeader, ReplyHeader,
+use crate::{
+    header::RequestHeader,
+    types::consensus::header::{self, ConsensusHeader, PrepareHeader, PrepareOkHeader},
 };
 use bytes::Bytes;
 use std::marker::PhantomData;
@@ -278,33 +279,36 @@ where
 #[derive(Debug)]
 #[allow(unused)]
 pub enum MessageBag {
-    Generic(Message<GenericHeader>),
+    Request(Message<RequestHeader>),
     Prepare(Message<PrepareHeader>),
     PrepareOk(Message<PrepareOkHeader>),
-    Commit(Message<CommitHeader>),
-    Reply(Message<ReplyHeader>),
 }
 
 impl MessageBag {
     #[allow(unused)]
     pub fn command(&self) -> header::Command2 {
         match self {
-            MessageBag::Generic(message) => message.header().command,
+            MessageBag::Request(message) => message.header().command,
             MessageBag::Prepare(message) => message.header().command,
             MessageBag::PrepareOk(message) => message.header().command,
-            MessageBag::Commit(message) => message.header().command,
-            MessageBag::Reply(message) => message.header().command,
         }
     }
 
     #[allow(unused)]
     pub fn size(&self) -> u32 {
         match self {
-            MessageBag::Generic(message) => message.header().size(),
+            MessageBag::Request(message) => message.header().size(),
             MessageBag::Prepare(message) => message.header().size(),
             MessageBag::PrepareOk(message) => message.header().size(),
-            MessageBag::Commit(message) => message.header().size(),
-            MessageBag::Reply(message) => message.header().size(),
+        }
+    }
+
+    #[allow(unused)]
+    pub fn operation(&self) -> header::Operation {
+        match self {
+            MessageBag::Request(message) => message.header().operation,
+            MessageBag::Prepare(message) => message.header().operation,
+            MessageBag::PrepareOk(message) => message.header().operation,
         }
     }
 }
@@ -326,17 +330,17 @@ where
                     unsafe { Message::<header::PrepareHeader>::from_buffer_unchecked(buffer) };
                 MessageBag::Prepare(msg)
             }
-            header::Command2::Commit => {
-                let msg = unsafe { Message::<header::CommitHeader>::from_buffer_unchecked(buffer) };
-                MessageBag::Commit(msg)
+            header::Command2::Request => {
+                let msg =
+                    unsafe { Message::<header::RequestHeader>::from_buffer_unchecked(buffer) };
+                MessageBag::Request(msg)
             }
-            header::Command2::Reply => {
-                let msg = unsafe { Message::<header::ReplyHeader>::from_buffer_unchecked(buffer) };
-                MessageBag::Reply(msg)
+            header::Command2::PrepareOk => {
+                let msg =
+                    unsafe { Message::<header::PrepareOkHeader>::from_buffer_unchecked(buffer) };
+                MessageBag::PrepareOk(msg)
             }
-            _ => unreachable!(
-                "For now we only support Prepare, Commit, and Reply. In the future we will support more commands. Command2: {command:?}"
-            ),
+            _ => unreachable!(),
         }
     }
 }
@@ -500,33 +504,6 @@ mod tests {
 
         assert_eq!(bag.command(), header::Command2::Prepare);
         assert!(matches!(bag, MessageBag::Prepare(_)));
-        assert!(!matches!(bag, MessageBag::Commit(_)));
-        assert!(!matches!(bag, MessageBag::Reply(_)));
-        assert!(!matches!(bag, MessageBag::Generic(_)));
-    }
-
-    #[test]
-    fn test_message_bag_from_commit() {
-        let commit = header::CommitHeader::create_test();
-        let bag = MessageBag::from(commit);
-
-        assert_eq!(bag.command(), header::Command2::Commit);
-        assert!(!matches!(bag, MessageBag::Prepare(_)));
-        assert!(matches!(bag, MessageBag::Commit(_)));
-        assert!(!matches!(bag, MessageBag::Reply(_)));
-        assert!(!matches!(bag, MessageBag::Generic(_)));
-    }
-
-    #[test]
-    fn test_message_bag_from_reply() {
-        let reply = header::ReplyHeader::create_test();
-        let bag = MessageBag::from(reply);
-
-        assert_eq!(bag.command(), header::Command2::Reply);
-        assert!(!matches!(bag, MessageBag::Prepare(_)));
-        assert!(!matches!(bag, MessageBag::Commit(_)));
-        assert!(matches!(bag, MessageBag::Reply(_)));
-        assert!(!matches!(bag, MessageBag::Generic(_)));
     }
 }