You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2025/11/18 16:10:54 UTC
(iggy) branch prepare_consesus_header_setup created (now c93912d38)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a change to branch prepare_consesus_header_setup
in repository https://gitbox.apache.org/repos/asf/iggy.git
at c93912d38 feat(cluster): prepare consensus header setup and add clock trait
This branch includes the following new commits:
new c93912d38 feat(cluster): prepare consensus header setup and add clock trait
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 01/01: feat(cluster): prepare consensus header setup and add clock trait
Posted by gk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch prepare_consesus_header_setup
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit c93912d3823016b1b2818f19869f417371e9a045
Author: numminex <g....@gmail.com>
AuthorDate: Tue Nov 18 17:10:43 2025 +0100
feat(cluster): prepare consensus header setup and add clock trait
---
Cargo.lock | 12 ++++++
Cargo.toml | 1 +
core/clock/Cargo.toml | 7 ++++
core/clock/src/impls.rs | 12 ++++++
core/{consensus => clock}/src/lib.rs | 10 +++--
core/common/src/types/consensus/header.rs | 4 ++
core/common/src/types/consensus/mod.rs | 70 ++++++++++++++++++++++++++-----
core/consensus/Cargo.toml | 3 ++
core/consensus/src/impls.rs | 1 +
core/consensus/src/lib.rs | 22 ++++++++--
core/metadata/Cargo.toml | 2 +
core/metadata/src/impls/metadata.rs | 32 ++++++++++----
12 files changed, 151 insertions(+), 25 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 4e90c1b92..90a1e5def 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1413,6 +1413,13 @@ version = "0.7.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a1d728cc89cf3aee9ff92b05e62b19ee65a02b5702cff7d5a377e32c6ae29d8d"
+[[package]]
+name = "clock"
+version = "0.1.0"
+dependencies = [
+ "iggy_common",
+]
+
[[package]]
name = "cmake"
version = "0.1.54"
@@ -1696,6 +1703,9 @@ dependencies = [
[[package]]
name = "consensus"
version = "0.1.0"
+dependencies = [
+ "clock",
+]
[[package]]
name = "console"
@@ -5221,7 +5231,9 @@ checksum = "f52b00d39961fc5b2736ea853c9cc86238e165017a493d1d5c8eac6bdc4cc273"
name = "metadata"
version = "0.1.0"
dependencies = [
+ "clock",
"consensus",
+ "tracing",
]
[[package]]
diff --git a/Cargo.toml b/Cargo.toml
index 53e3ae956..811ec468d 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -27,6 +27,7 @@ members = [
"core/bench/runner",
"core/binary_protocol",
"core/cli",
+ "core/clock",
"core/common",
"core/connectors/runtime",
"core/connectors/sdk",
diff --git a/core/clock/Cargo.toml b/core/clock/Cargo.toml
new file mode 100644
index 000000000..0d2727986
--- /dev/null
+++ b/core/clock/Cargo.toml
@@ -0,0 +1,7 @@
+[package]
+name = "clock"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+iggy_common = { workspace = true }
diff --git a/core/clock/src/impls.rs b/core/clock/src/impls.rs
new file mode 100644
index 000000000..225a0e364
--- /dev/null
+++ b/core/clock/src/impls.rs
@@ -0,0 +1,12 @@
+use crate::Clock;
+use iggy_common::IggyTimestamp;
+
+pub struct IggySystemClock;
+
+impl Clock for IggySystemClock {
+ type Realtime = IggyTimestamp;
+
+ fn realtime(&self) -> Self::Realtime {
+ IggyTimestamp::now()
+ }
+}
diff --git a/core/consensus/src/lib.rs b/core/clock/src/lib.rs
similarity index 86%
copy from core/consensus/src/lib.rs
copy to core/clock/src/lib.rs
index c3241d3c7..2223894ed 100644
--- a/core/consensus/src/lib.rs
+++ b/core/clock/src/lib.rs
@@ -15,8 +15,10 @@
// specific language governing permissions and limitations
// under the License.
-pub trait Consensus {
- type RequestMessage;
- type ReplicateMessage;
- type AckMessage;
+pub trait Clock {
+ type Realtime;
+ fn realtime(&self) -> Self::Realtime;
}
+
+mod impls;
+pub use impls::IggySystemClock;
diff --git a/core/common/src/types/consensus/header.rs b/core/common/src/types/consensus/header.rs
new file mode 100644
index 000000000..6c2e97d45
--- /dev/null
+++ b/core/common/src/types/consensus/header.rs
@@ -0,0 +1,4 @@
+#[expect(unused)]
+pub struct Header {}
+
+pub trait ConsensusHeader {}
diff --git a/core/common/src/types/consensus/mod.rs b/core/common/src/types/consensus/mod.rs
index 6ecac7726..795695d12 100644
--- a/core/common/src/types/consensus/mod.rs
+++ b/core/common/src/types/consensus/mod.rs
@@ -17,6 +17,10 @@
use std::marker::PhantomData;
+use bytes::Bytes;
+
+use crate::types::consensus::header::ConsensusHeader;
+
// TODO: Header generic
// TODO: We will have to impl something like this (NOT ONE TO ONE JUST A SKETCH) as we will use `bytemuck`:
/*
@@ -40,14 +44,20 @@ pub trait Header: Sized {
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum Command {
- Reserved = 0,
- Ping = 1,
- Pong = 2,
- Request = 3,
- Prepare = 4,
- PrepareOk = 5,
- Reply = 6,
- Commit = 7,
+ reserved = 0,
+
+ ping = 1,
+ pong = 2,
+
+ request = 5,
+ prepare = 6,
+ prepare_ok = 7,
+ reply = 8,
+ commit = 9,
+
+ start_view_change = 10,
+ do_view_change = 11,
+ start_view = 24,
// ... etc
}
@@ -60,6 +70,9 @@ pub struct GenericHeader {
// ... other fields
}
+// first 128 bytes GenericHeader (checksum, command, size, etc....)
+// second 128 bytes specific header (PrepareHeader, CommitHeader, etc....)
+
impl Header for GenericHeader {
const COMMAND: Command = Command::Reserved;
@@ -86,6 +99,7 @@ impl Header for PrepareHeader {
fn size(&self) -> u32 { self.size }
fn command(&self) -> Command { self.command }
fn checksum(&self) -> u128 { self.checksum }
+}
*/
// And then for Message impl
@@ -152,11 +166,47 @@ pub enum Operation {
Consume = 101,
Fetch = 102,
}
+
+// Specific header types
+#[repr(C)]
+pub struct PrepareHeader {
+ checksum: u128,
+ command: Command,
+ size: u32,
+ // ... prepare-specific fields
+ view: u32,
+ op: u64,
+ commit: u64,
+
+ // second 128 bytes
+ operation: Operation,
+}
*/
// Which will have an method that returns discriminator between Metadata and Partition requests
+// TODO: Fill this enum
#[expect(unused)]
-pub struct Message<H> {
- _phantom_header: PhantomData<H>,
+pub enum MessageBag {
+ Void,
}
+
+#[expect(unused)]
+pub struct Message<H>
+where
+ H: ConsensusHeader,
+{
+ buffer: Bytes,
+ _h: PhantomData<H>,
+}
+
+impl<H> From<Message<H>> for MessageBag
+where
+ H: ConsensusHeader,
+{
+ fn from(_value: Message<H>) -> Self {
+ MessageBag::Void
+ }
+}
+
+pub(crate) mod header;
diff --git a/core/consensus/Cargo.toml b/core/consensus/Cargo.toml
index 9bf2b9ad8..2240a10e2 100644
--- a/core/consensus/Cargo.toml
+++ b/core/consensus/Cargo.toml
@@ -29,3 +29,6 @@ repository = "https://github.com/apache/iggy"
readme = "../../../README.md"
[dependencies]
+clock = { path = "../clock" }
+
+
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
new file mode 100644
index 000000000..0cfe6bbeb
--- /dev/null
+++ b/core/consensus/src/impls.rs
@@ -0,0 +1 @@
+pub struct VsrConsensus;
diff --git a/core/consensus/src/lib.rs b/core/consensus/src/lib.rs
index c3241d3c7..0a02bc089 100644
--- a/core/consensus/src/lib.rs
+++ b/core/consensus/src/lib.rs
@@ -14,9 +14,25 @@
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
+use clock::Clock;
-pub trait Consensus {
- type RequestMessage;
- type ReplicateMessage;
+pub trait Project<U, T, C>
+where
+ C: Consensus<T>,
+ T: Clock,
+{
+ fn project(self, consensus: &C) -> U;
+}
+
+pub trait Consensus<C>
+where
+ Self: Sized,
+ C: Clock,
+{
+ type RequestMessage: Project<Self::ReplicateMessage, C, Self>;
+ type ReplicateMessage: Project<Self::AckMessage, C, Self>;
type AckMessage;
}
+
+mod impls;
+pub use impls::*;
diff --git a/core/metadata/Cargo.toml b/core/metadata/Cargo.toml
index 87e578250..f026f763e 100644
--- a/core/metadata/Cargo.toml
+++ b/core/metadata/Cargo.toml
@@ -29,5 +29,7 @@ repository = "https://github.com/apache/iggy"
readme = "../../../README.md"
[dependencies]
+clock = { path = "../clock" }
consensus = { path = "../consensus" }
+tracing = { workspace = true }
diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs
index 659112f1e..f802d53f1 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -14,13 +14,17 @@
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
-use consensus::Consensus;
+use clock::Clock;
+use consensus::{Consensus, Project};
+use std::marker::PhantomData;
+use tracing::debug;
// TODO: Define a trait (probably in some external crate)
#[expect(unused)]
-trait Metadata<C>
+trait Metadata<C, T>
where
- C: Consensus,
+ C: Consensus<T>,
+ T: Clock,
{
fn on_request(&self, message: C::RequestMessage);
fn on_replicate(&self, message: C::ReplicateMessage);
@@ -28,19 +32,27 @@ where
}
#[expect(unused)]
-struct IggyMetadata<C, M, J, S> {
+struct IggyMetadata<C, M, J, S, T>
+where
+ C: Consensus<T>,
+ T: Clock,
+{
consensus: C,
mux_stm: M,
journal: J,
snapshot: S,
+
+ _t: PhantomData<T>,
}
-impl<C, M, J, S> Metadata<C> for IggyMetadata<C, M, J, S>
+impl<C, M, J, S, T> Metadata<C, T> for IggyMetadata<C, M, J, S, T>
where
- C: Consensus,
+ C: Consensus<T>,
+ T: Clock,
{
- fn on_request(&self, _message: C::RequestMessage) {
- todo!()
+ fn on_request(&self, message: C::RequestMessage) {
+ debug!("handling metadata request");
+ let _message = message.project(&self.consensus);
}
fn on_replicate(&self, _message: C::ReplicateMessage) {
@@ -51,3 +63,7 @@ where
todo!()
}
}
+
+// TODO: Hide with associated types all of those generics, so they are not leaking to the upper layer
+#[expect(unused)]
+pub trait MetadataHandle {}