feat(metadata): introduce mux_state_machine with state impls (#2544)

Implement `users` , `streams`, `consumer_groups` states for
MuxStateMachine.

Author: numinnex

Cargo.lock

@@ -36,6 +36,7 @@ fast_async_lock = ["dep:fast-async-mutex"]
 aes-gcm = { workspace = true }
 ahash = { workspace = true }
 base64 = { workspace = true }
+blake3 = { workspace = true }
 bon = { workspace = true }
 byte-unit = { workspace = true }
 bytemuck = { workspace = true }
@@ -57,6 +58,7 @@ humantime = { workspace = true }
 nix = { workspace = true }
 once_cell = { workspace = true }
 rcgen = "0.14.6"
+ring = "0.17.14"
 rustls = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }

core/common/Cargo.toml

@@ -88,6 +88,7 @@ pub use types::message::*;
 pub use types::partition::*;
 pub use types::permissions::permissions_global::*;
 pub use types::permissions::personal_access_token::*;
+pub use types::personal_access_tokens::*;
 pub use types::snapshot::*;
 pub use types::stats::*;
 pub use types::stream::*;
@@ -100,6 +101,7 @@ pub use utils::checksum::*;
 pub use utils::crypto::*;
 pub use utils::duration::{IggyDuration, SEC_IN_MICRO};
 pub use utils::expiry::IggyExpiry;
+pub use utils::hash::*;
 pub use utils::personal_access_token_expiry::PersonalAccessTokenExpiry;
 pub use utils::text;
 pub use utils::timestamp::*;

core/common/src/lib.rs

@@ -89,24 +89,25 @@ where
         }
     }
 
-    /// Replace the header with a new one, using the provided function to generate it.
-    pub fn replace_header<T: ConsensusHeader>(self, f: impl FnOnce(&H) -> T) -> Message<T> {
+    /// Transmute the header to a different type, using the provided function to modify the new header.
+    pub fn transmute_header<T: ConsensusHeader>(self, f: impl FnOnce(H, &mut T)) -> Message<T> {
         assert_eq!(size_of::<H>(), size_of::<T>());
-        let prev = self.header();
-        let header = f(prev);
 
-        let header_bytes = bytemuck::bytes_of(&header);
-        let buffer = self.into_inner();
+        // Copy old header to stack to avoid UB.
+        let old_header = *self.header();
+
         // Safety: We ensured that size_of::<H>() == size_of::<T>()
         // On top of that, there is going to be only one reference to buffer during this function call
         // so no other references can observe the mutation.
         // In the future we can replace the `Bytes` buffer with something that does not allow sharing between different threads.
+        let buffer = self.into_inner();
         unsafe {
             let ptr = buffer.as_ptr() as *mut u8;
-            let slice = std::slice::from_raw_parts_mut(ptr, buffer.len());
-            slice[..size_of::<H>()].copy_from_slice(header_bytes);
+            let slice = std::slice::from_raw_parts_mut(ptr, size_of::<T>());
+            let new_header = bytemuck::from_bytes_mut(slice);
+            f(old_header, new_header);
         }
-        // TODO: Recalculate checksums
+
         Message {
             buffer,
             _marker: PhantomData,

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

@@ -29,6 +29,7 @@ pub(crate) mod identifier;
 pub(crate) mod message;
 pub(crate) mod partition;
 pub(crate) mod permissions;
+pub(crate) mod personal_access_tokens;
 pub(crate) mod snapshot;
 pub(crate) mod stats;
 pub(crate) mod stream;

core/common/src/types/mod.rs

@@ -15,11 +15,11 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-use crate::streaming::utils::hash;
-use iggy_common::IggyExpiry;
-use iggy_common::IggyTimestamp;
-use iggy_common::UserId;
-use iggy_common::text::as_base64;
+use crate::IggyExpiry;
+use crate::IggyTimestamp;
+use crate::UserId;
+use crate::text::as_base64;
+use crate::utils::hash;
 use ring::rand::SecureRandom;
 use std::sync::Arc;
 
@@ -96,8 +96,8 @@ impl PersonalAccessToken {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use iggy_common::IggyDuration;
-    use iggy_common::IggyTimestamp;
+    use crate::IggyDuration;
+    use crate::IggyTimestamp;
 
     #[test]
     fn personal_access_token_should_be_created_with_random_secure_value_and_hashed_successfully() {

…l_access_tokens/personal_access_token.rs …/src/types/personal_access_tokens/mod.rscore/server/src/streaming/personal_access_tokens/personal_access_token.rs renamed to core/common/src/types/personal_access_tokens/mod.rs core/server/src/streaming/personal_access_tokens/personal_access_token.rs renamed to core/common/src/types/personal_access_tokens/mod.rs

@@ -20,6 +20,7 @@ pub(crate) mod checksum;
 pub(crate) mod crypto;
 pub(crate) mod duration;
 pub(crate) mod expiry;
+pub(crate) mod hash;
 pub(crate) mod personal_access_token_expiry;
 pub mod text;
 pub(crate) mod timestamp;

core/server/src/streaming/utils/hash.rs core/common/src/utils/hash.rscore/server/src/streaming/utils/hash.rs renamed to core/common/src/utils/hash.rs

@@ -513,22 +513,22 @@ impl Project<Message<PrepareHeader>> for Message<RequestHeader> {
     fn project(self, consensus: &Self::Consensus) -> Message<PrepareHeader> {
         let op = consensus.sequencer.current_sequence() + 1;
 
-        self.replace_header(|prev| {
-            PrepareHeader {
+        self.transmute_header(|old, new| {
+            *new = PrepareHeader {
                 cluster: consensus.cluster,
-                size: prev.size,
+                size: old.size,
                 epoch: 0,
                 view: consensus.view.get(),
-                release: prev.release,
+                release: old.release,
                 command: Command2::Prepare,
                 replica: consensus.replica,
                 parent: 0, // TODO: Get parent checksum from the previous entry in the journal (figure out how to pass that ctx here)
-                request_checksum: prev.request_checksum,
-                request: prev.request,
+                request_checksum: old.request_checksum,
+                request: old.request,
                 commit: consensus.commit.get(),
                 op,
                 timestamp: 0, // 0 for now. Implement correct way to get timestamp later
-                operation: prev.operation,
+                operation: old.operation,
                 ..Default::default()
             }
         })
@@ -537,25 +537,26 @@ impl Project<Message<PrepareHeader>> for Message<RequestHeader> {
 
 impl Project<Message<PrepareOkHeader>> for Message<PrepareHeader> {
     type Consensus = VsrConsensus;
+
     fn project(self, consensus: &Self::Consensus) -> Message<PrepareOkHeader> {
-        self.replace_header(|prev| {
-            PrepareOkHeader {
+        self.transmute_header(|old, new| {
+            *new = PrepareOkHeader {
                 command: Command2::PrepareOk,
-                parent: prev.parent,
-                prepare_checksum: prev.checksum,
-                request: prev.request,
+                parent: old.parent,
+                prepare_checksum: old.checksum,
+                request: old.request,
                 cluster: consensus.cluster,
                 replica: consensus.replica,
                 epoch: 0, // TODO: consensus.epoch
                 // It's important to use the view of the replica, not the received prepare!
                 view: consensus.view.get(),
-                op: prev.op,
+                op: old.op,
                 commit: consensus.commit.get(),
-                timestamp: prev.timestamp,
-                operation: prev.operation,
+                timestamp: old.timestamp,
+                operation: old.operation,
                 // PrepareOks are only header no body
                 ..Default::default()
-            }
+            };
         })
     }
 }

core/common/src/utils/mod.rs

@@ -28,8 +28,11 @@ repository = "https://github.com/apache/iggy"
 readme = "../../../README.md"
 
 [dependencies]
+ahash = { workspace = true }
+bytes = { workspace = true }
 consensus = { path = "../consensus" }
 iggy_common = { path = "../common" }
 journal = { path = "../journal" }
 message_bus = { path = "../message_bus" }
+slab = "0.4.11"
 tracing = { workspace = true }