numinnex commented on code in PR #2389:
URL: https://github.com/apache/iggy/pull/2389#discussion_r2549102392
##########
core/common/src/types/consensus/header.rs:
##########
@@ -15,7 +15,310 @@
// specific language governing permissions and limitations
// under the License.
+use bytemuck::{Pod, Zeroable};
+
#[expect(unused)]
pub struct Header {}
-pub trait ConsensusHeader {}
+pub trait ConsensusHeader: Sized + Pod + Zeroable {
+ const COMMAND: Command;
+
+ fn size(&self) -> u32;
+ fn command(&self) -> Command;
+ fn checksum(&self) -> u128;
+ fn cluster(&self) -> u128;
+
+ fn validate(&self) -> Result<(), &'static str>;
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum Command {
+ Reserved = 0,
+
+ Ping = 1,
+ Pong = 2,
+ PingClient = 3,
+ PongClient = 4,
+
+ Request = 5,
+ Prepare = 6,
+ PrepareOk = 7,
+ Reply = 8,
+ Commit = 9,
+
+ StartViewChange = 10,
+ DoViewChange = 11,
+ StartView = 24,
+
+ RequestStartView = 12,
+ RequstHeads = 13,
+ RequestPrepare = 14,
+ RequestReply = 15,
+
+ Headers = 16,
+
+ Eviction = 17,
+
+ RequestBlocks = 18,
+ Block = 19,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum Operation {
+ Reserved = 0,
+ Root = 1,
+ Register = 2,
+ Reconfigure = 3,
+ Pulse = 4,
+ Upgrade = 5,
+
+ CreateStream = 128,
+ UpdateStream = 129,
+ DeleteStream = 130,
+ PurgeStream = 131,
+ CreateTopic = 132,
+ UpdateTopic = 133,
+ DeleteTopic = 134,
+ PurgeTopic = 135,
+ CreatePartitions = 136,
+ DeletePartitions = 137,
+ DeleteSegments = 138,
+ CreateConsumerGroup = 139,
+ DeleteConsumerGroup = 140,
+ CreateUser = 141,
+ UpdateUser = 142,
+ DeleteUser = 143,
+ ChangePassword = 144,
+ UpdatePermissions = 145,
+ CreatePersonalAccessToken = 146,
+ DeletePersonalAccessToken = 147,
+}
+
+#[repr(C)]
+#[derive(Debug, Clone, Copy)]
+pub struct GenericHeader {
+ pub checksum: u128,
+ pub checksum_padding: u128,
+ pub checksum_body: u128,
+ pub checksum_body_padding: u128,
+ pub nonce_reserved: u128,
+ pub cluster: u128,
+ pub size: u32,
+ pub epoch: u32,
+ pub view: u32,
+ pub release: u32,
+ pub protocol: u16,
+ pub command: Command,
+ pub replica: u8,
+ pub reserved_frame: [u8; 12],
+
+ pub reserved_command: [u8; 128],
+}
+
+unsafe impl Pod for GenericHeader {}
+unsafe impl Zeroable for GenericHeader {}
+
+impl ConsensusHeader for GenericHeader {
+ const COMMAND: Command = Command::Reserved;
+
+ fn size(&self) -> u32 {
+ self.size
+ }
+ fn command(&self) -> Command {
+ self.command
+ }
+ fn checksum(&self) -> u128 {
+ self.checksum
+ }
+ fn cluster(&self) -> u128 {
+ self.cluster
+ }
+
+ fn validate(&self) -> Result<(), &'static str> {
+ Ok(())
+ }
+}
+
+#[repr(C)]
+#[derive(Debug, Clone, Copy)]
+pub struct PrepareHeader {
+ pub checksum: u128,
+ pub checksum_padding: u128,
+ pub checksum_body: u128,
+ pub checksum_body_padding: u128,
+ pub nonce_reserved: u128,
+ pub cluster: u128,
+ pub size: u32,
+ pub epoch: u32,
+ pub view: u32,
+ pub release: u32,
+ pub protocol: u16,
+ pub command: Command,
+ pub replica: u8,
+ pub reserved_frame: [u8; 12],
+
+ pub parent: u128,
+ pub parent_padding: u128,
+ pub request_checksum: u128,
+ pub request_checksum_padding: u128,
+ pub checkpoint_id: u128,
+ pub client: u128,
+ pub op: u64,
+ pub commit: u64,
+ pub timestamp: u64,
+ pub request: u32,
+ pub operation: Operation,
+ pub reserved: [u8; 3],
+}
+
+unsafe impl Pod for PrepareHeader {}
+unsafe impl Zeroable for PrepareHeader {}
+
+impl ConsensusHeader for PrepareHeader {
+ const COMMAND: Command = Command::Prepare;
+
+ fn size(&self) -> u32 {
+ self.size
+ }
+ fn command(&self) -> Command {
+ self.command
+ }
+ fn checksum(&self) -> u128 {
+ self.checksum
+ }
+ fn cluster(&self) -> u128 {
+ self.cluster
+ }
+
+ fn validate(&self) -> Result<(), &'static str> {
+ if self.command != Command::Prepare {
+ return Err("command must be Prepare");
+ }
+ if self.parent_padding != 0 {
+ return Err("parent_padding must be 0");
+ }
+ if self.request_checksum_padding != 0 {
+ return Err("request_checksum_padding must be 0");
+ }
+ if self.operation == Operation::Reserved {
+ return Err("operation must be Reserved");
+ }
+ Ok(())
+ }
+}
+
+#[repr(C)]
+#[derive(Debug, Clone, Copy)]
+pub struct CommitHeader {
+ pub checksum: u128,
+ pub checksum_padding: u128,
+ pub checksum_body: u128,
+ pub checksum_body_padding: u128,
+ pub nonce_reserved: u128,
+ pub cluster: u128,
+ pub size: u32,
+ pub epoch: u32,
+ pub view: u32,
+ pub release: u32,
+ pub protocol: u16,
+ pub command: Command,
+ pub replica: u8,
+ pub reserved_frame: [u8; 12],
+
+ pub commit_checksum: u128,
+ pub timestamp_monotonic: u64,
+ pub checkpoint_id: u128,
+ pub commit: u64,
+ pub checkpoint_op: u64,
+ pub reserved: [u8; 80],
+}
+
+unsafe impl Pod for CommitHeader {}
+unsafe impl Zeroable for CommitHeader {}
+
+impl ConsensusHeader for CommitHeader {
+ const COMMAND: Command = Command::Commit;
+
+ fn size(&self) -> u32 {
+ self.size
+ }
+ fn command(&self) -> Command {
+ self.command
+ }
+ fn checksum(&self) -> u128 {
+ self.checksum
+ }
+ fn cluster(&self) -> u128 {
+ self.cluster
+ }
+
+ fn validate(&self) -> Result<(), &'static str> {
Review Comment:
Create an `ConsesusError` enum.
##########
core/common/src/types/consensus/header.rs:
##########
@@ -15,7 +15,310 @@
// specific language governing permissions and limitations
// under the License.
+use bytemuck::{Pod, Zeroable};
+
#[expect(unused)]
pub struct Header {}
-pub trait ConsensusHeader {}
+pub trait ConsensusHeader: Sized + Pod + Zeroable {
+ const COMMAND: Command;
+
+ fn size(&self) -> u32;
+ fn command(&self) -> Command;
+ fn checksum(&self) -> u128;
+ fn cluster(&self) -> u128;
+
+ fn validate(&self) -> Result<(), &'static str>;
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum Command {
+ Reserved = 0,
+
+ Ping = 1,
+ Pong = 2,
+ PingClient = 3,
+ PongClient = 4,
+
+ Request = 5,
+ Prepare = 6,
+ PrepareOk = 7,
+ Reply = 8,
+ Commit = 9,
+
+ StartViewChange = 10,
+ DoViewChange = 11,
+ StartView = 24,
+
+ RequestStartView = 12,
+ RequstHeads = 13,
+ RequestPrepare = 14,
+ RequestReply = 15,
+
+ Headers = 16,
+
+ Eviction = 17,
+
+ RequestBlocks = 18,
+ Block = 19,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum Operation {
+ Reserved = 0,
+ Root = 1,
+ Register = 2,
+ Reconfigure = 3,
+ Pulse = 4,
+ Upgrade = 5,
+
+ CreateStream = 128,
+ UpdateStream = 129,
+ DeleteStream = 130,
+ PurgeStream = 131,
+ CreateTopic = 132,
+ UpdateTopic = 133,
+ DeleteTopic = 134,
+ PurgeTopic = 135,
+ CreatePartitions = 136,
+ DeletePartitions = 137,
+ DeleteSegments = 138,
+ CreateConsumerGroup = 139,
+ DeleteConsumerGroup = 140,
+ CreateUser = 141,
+ UpdateUser = 142,
+ DeleteUser = 143,
+ ChangePassword = 144,
+ UpdatePermissions = 145,
+ CreatePersonalAccessToken = 146,
+ DeletePersonalAccessToken = 147,
+}
+
+#[repr(C)]
+#[derive(Debug, Clone, Copy)]
+pub struct GenericHeader {
+ pub checksum: u128,
+ pub checksum_padding: u128,
+ pub checksum_body: u128,
+ pub checksum_body_padding: u128,
+ pub nonce_reserved: u128,
+ pub cluster: u128,
+ pub size: u32,
+ pub epoch: u32,
+ pub view: u32,
+ pub release: u32,
+ pub protocol: u16,
+ pub command: Command,
+ pub replica: u8,
+ pub reserved_frame: [u8; 12],
+
+ pub reserved_command: [u8; 128],
+}
+
+unsafe impl Pod for GenericHeader {}
+unsafe impl Zeroable for GenericHeader {}
+
+impl ConsensusHeader for GenericHeader {
+ const COMMAND: Command = Command::Reserved;
+
+ fn size(&self) -> u32 {
+ self.size
+ }
+ fn command(&self) -> Command {
+ self.command
+ }
+ fn checksum(&self) -> u128 {
+ self.checksum
+ }
+ fn cluster(&self) -> u128 {
+ self.cluster
+ }
+
+ fn validate(&self) -> Result<(), &'static str> {
+ Ok(())
+ }
+}
+
+#[repr(C)]
+#[derive(Debug, Clone, Copy)]
+pub struct PrepareHeader {
+ pub checksum: u128,
+ pub checksum_padding: u128,
+ pub checksum_body: u128,
+ pub checksum_body_padding: u128,
+ pub nonce_reserved: u128,
+ pub cluster: u128,
+ pub size: u32,
+ pub epoch: u32,
+ pub view: u32,
+ pub release: u32,
+ pub protocol: u16,
+ pub command: Command,
+ pub replica: u8,
+ pub reserved_frame: [u8; 12],
+
+ pub parent: u128,
+ pub parent_padding: u128,
+ pub request_checksum: u128,
+ pub request_checksum_padding: u128,
+ pub checkpoint_id: u128,
+ pub client: u128,
+ pub op: u64,
+ pub commit: u64,
+ pub timestamp: u64,
+ pub request: u32,
+ pub operation: Operation,
+ pub reserved: [u8; 3],
+}
+
+unsafe impl Pod for PrepareHeader {}
+unsafe impl Zeroable for PrepareHeader {}
+
+impl ConsensusHeader for PrepareHeader {
+ const COMMAND: Command = Command::Prepare;
+
+ fn size(&self) -> u32 {
+ self.size
+ }
+ fn command(&self) -> Command {
+ self.command
+ }
+ fn checksum(&self) -> u128 {
+ self.checksum
+ }
+ fn cluster(&self) -> u128 {
+ self.cluster
+ }
+
+ fn validate(&self) -> Result<(), &'static str> {
+ if self.command != Command::Prepare {
+ return Err("command must be Prepare");
+ }
+ if self.parent_padding != 0 {
+ return Err("parent_padding must be 0");
+ }
+ if self.request_checksum_padding != 0 {
+ return Err("request_checksum_padding must be 0");
+ }
+ if self.operation == Operation::Reserved {
+ return Err("operation must be Reserved");
+ }
+ Ok(())
+ }
+}
+
+#[repr(C)]
+#[derive(Debug, Clone, Copy)]
+pub struct CommitHeader {
+ pub checksum: u128,
+ pub checksum_padding: u128,
+ pub checksum_body: u128,
+ pub checksum_body_padding: u128,
+ pub nonce_reserved: u128,
+ pub cluster: u128,
+ pub size: u32,
+ pub epoch: u32,
+ pub view: u32,
+ pub release: u32,
+ pub protocol: u16,
+ pub command: Command,
+ pub replica: u8,
+ pub reserved_frame: [u8; 12],
+
+ pub commit_checksum: u128,
+ pub timestamp_monotonic: u64,
+ pub checkpoint_id: u128,
+ pub commit: u64,
+ pub checkpoint_op: u64,
+ pub reserved: [u8; 80],
+}
+
+unsafe impl Pod for CommitHeader {}
+unsafe impl Zeroable for CommitHeader {}
+
+impl ConsensusHeader for CommitHeader {
+ const COMMAND: Command = Command::Commit;
+
+ fn size(&self) -> u32 {
+ self.size
+ }
+ fn command(&self) -> Command {
+ self.command
+ }
+ fn checksum(&self) -> u128 {
+ self.checksum
+ }
+ fn cluster(&self) -> u128 {
+ self.cluster
+ }
+
+ fn validate(&self) -> Result<(), &'static str> {
+ if self.command != Command::Commit {
+ return Err("command != Commit");
+ }
+ if self.size != 256 {
+ return Err("size != 256");
+ }
+ Ok(())
+ }
+}
+
+#[repr(C)]
+#[derive(Debug, Clone, Copy)]
+pub struct ReplyHeader {
+ pub checksum: u128,
+ pub checksum_padding: u128,
+ pub checksum_body: u128,
+ pub checksum_body_padding: u128,
+ pub nonce_reserved: u128,
+ pub cluster: u128,
+ pub size: u32,
+ pub epoch: u32,
+ pub view: u32,
+ pub release: u32,
+ pub protocol: u16,
+ pub command: Command,
+ pub replica: u8,
+ pub reserved_frame: [u8; 12],
+
+ pub request_checksum: u128,
+ pub request_checksum_padding: u128,
+ pub context: u128,
+ pub context_padding: u128,
+ pub client: u128,
+ pub op: u64,
+ pub commit: u64,
+ pub timestamp: u64,
+ pub request: u32,
+ pub operation: Operation,
+ pub reserved: [u8; 19],
+}
+
+unsafe impl Pod for ReplyHeader {}
+unsafe impl Zeroable for ReplyHeader {}
+
+impl ConsensusHeader for ReplyHeader {
+ const COMMAND: Command = Command::Reply;
+
+ fn size(&self) -> u32 {
+ self.size
+ }
+ fn command(&self) -> Command {
+ self.command
+ }
+ fn checksum(&self) -> u128 {
+ self.checksum
+ }
+ fn cluster(&self) -> u128 {
Review Comment:
I am wondering whether it is necessary to be part of the `ConsesusHeader`
trait tho
##########
core/common/src/types/consensus/header.rs:
##########
@@ -15,7 +15,310 @@
// specific language governing permissions and limitations
// under the License.
+use bytemuck::{Pod, Zeroable};
+
#[expect(unused)]
pub struct Header {}
-pub trait ConsensusHeader {}
+pub trait ConsensusHeader: Sized + Pod + Zeroable {
+ const COMMAND: Command;
+
+ fn size(&self) -> u32;
+ fn command(&self) -> Command;
+ fn checksum(&self) -> u128;
+ fn cluster(&self) -> u128;
+
+ fn validate(&self) -> Result<(), &'static str>;
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum Command {
+ Reserved = 0,
+
+ Ping = 1,
+ Pong = 2,
+ PingClient = 3,
+ PongClient = 4,
+
+ Request = 5,
+ Prepare = 6,
+ PrepareOk = 7,
+ Reply = 8,
+ Commit = 9,
+
+ StartViewChange = 10,
+ DoViewChange = 11,
+ StartView = 24,
+
+ RequestStartView = 12,
+ RequstHeads = 13,
+ RequestPrepare = 14,
+ RequestReply = 15,
+
+ Headers = 16,
+
+ Eviction = 17,
+
+ RequestBlocks = 18,
+ Block = 19,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum Operation {
+ Reserved = 0,
+ Root = 1,
+ Register = 2,
+ Reconfigure = 3,
+ Pulse = 4,
+ Upgrade = 5,
+
+ CreateStream = 128,
+ UpdateStream = 129,
+ DeleteStream = 130,
+ PurgeStream = 131,
+ CreateTopic = 132,
+ UpdateTopic = 133,
+ DeleteTopic = 134,
+ PurgeTopic = 135,
+ CreatePartitions = 136,
+ DeletePartitions = 137,
+ DeleteSegments = 138,
+ CreateConsumerGroup = 139,
+ DeleteConsumerGroup = 140,
+ CreateUser = 141,
+ UpdateUser = 142,
+ DeleteUser = 143,
+ ChangePassword = 144,
+ UpdatePermissions = 145,
+ CreatePersonalAccessToken = 146,
+ DeletePersonalAccessToken = 147,
+}
+
+#[repr(C)]
+#[derive(Debug, Clone, Copy)]
+pub struct GenericHeader {
+ pub checksum: u128,
+ pub checksum_padding: u128,
+ pub checksum_body: u128,
+ pub checksum_body_padding: u128,
+ pub nonce_reserved: u128,
+ pub cluster: u128,
Review Comment:
I wonder why they use u128 for cluster, do they store a hash of the cluster
rather than just plain id ?
##########
core/common/src/types/consensus/header.rs:
##########
@@ -15,7 +15,310 @@
// specific language governing permissions and limitations
// under the License.
+use bytemuck::{Pod, Zeroable};
+
#[expect(unused)]
pub struct Header {}
-pub trait ConsensusHeader {}
+pub trait ConsensusHeader: Sized + Pod + Zeroable {
+ const COMMAND: Command;
+
+ fn size(&self) -> u32;
+ fn command(&self) -> Command;
+ fn checksum(&self) -> u128;
+ fn cluster(&self) -> u128;
+
+ fn validate(&self) -> Result<(), &'static str>;
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum Command {
+ Reserved = 0,
+
+ Ping = 1,
+ Pong = 2,
+ PingClient = 3,
+ PongClient = 4,
+
+ Request = 5,
+ Prepare = 6,
+ PrepareOk = 7,
+ Reply = 8,
+ Commit = 9,
+
+ StartViewChange = 10,
+ DoViewChange = 11,
+ StartView = 24,
+
+ RequestStartView = 12,
+ RequstHeads = 13,
+ RequestPrepare = 14,
+ RequestReply = 15,
+
+ Headers = 16,
+
+ Eviction = 17,
+
+ RequestBlocks = 18,
+ Block = 19,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum Operation {
+ Reserved = 0,
+ Root = 1,
+ Register = 2,
+ Reconfigure = 3,
+ Pulse = 4,
+ Upgrade = 5,
Review Comment:
We don't need these.
##########
core/common/src/types/consensus/header.rs:
##########
@@ -15,7 +15,310 @@
// specific language governing permissions and limitations
// under the License.
+use bytemuck::{Pod, Zeroable};
+
#[expect(unused)]
pub struct Header {}
-pub trait ConsensusHeader {}
+pub trait ConsensusHeader: Sized + Pod + Zeroable {
+ const COMMAND: Command;
+
+ fn size(&self) -> u32;
+ fn command(&self) -> Command;
+ fn checksum(&self) -> u128;
+ fn cluster(&self) -> u128;
+
+ fn validate(&self) -> Result<(), &'static str>;
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum Command {
+ Reserved = 0,
+
+ Ping = 1,
+ Pong = 2,
+ PingClient = 3,
+ PongClient = 4,
+
+ Request = 5,
+ Prepare = 6,
+ PrepareOk = 7,
+ Reply = 8,
+ Commit = 9,
+
+ StartViewChange = 10,
+ DoViewChange = 11,
+ StartView = 24,
+
+ RequestStartView = 12,
+ RequstHeads = 13,
+ RequestPrepare = 14,
+ RequestReply = 15,
+
+ Headers = 16,
+
+ Eviction = 17,
+
+ RequestBlocks = 18,
+ Block = 19,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum Operation {
+ Reserved = 0,
+ Root = 1,
+ Register = 2,
+ Reconfigure = 3,
+ Pulse = 4,
+ Upgrade = 5,
+
+ CreateStream = 128,
+ UpdateStream = 129,
+ DeleteStream = 130,
+ PurgeStream = 131,
+ CreateTopic = 132,
+ UpdateTopic = 133,
+ DeleteTopic = 134,
+ PurgeTopic = 135,
+ CreatePartitions = 136,
+ DeletePartitions = 137,
+ DeleteSegments = 138,
+ CreateConsumerGroup = 139,
+ DeleteConsumerGroup = 140,
+ CreateUser = 141,
+ UpdateUser = 142,
+ DeleteUser = 143,
+ ChangePassword = 144,
+ UpdatePermissions = 145,
+ CreatePersonalAccessToken = 146,
+ DeletePersonalAccessToken = 147,
+}
+
+#[repr(C)]
+#[derive(Debug, Clone, Copy)]
+pub struct GenericHeader {
+ pub checksum: u128,
+ pub checksum_padding: u128,
Review Comment:
I think TB uses those because they want to use an u256 checksum ? We will
use u128 checksum so we don't need that padding.
##########
core/common/src/types/consensus/header.rs:
##########
@@ -15,7 +15,310 @@
// specific language governing permissions and limitations
// under the License.
+use bytemuck::{Pod, Zeroable};
+
#[expect(unused)]
pub struct Header {}
-pub trait ConsensusHeader {}
+pub trait ConsensusHeader: Sized + Pod + Zeroable {
+ const COMMAND: Command;
+
+ fn size(&self) -> u32;
+ fn command(&self) -> Command;
+ fn checksum(&self) -> u128;
+ fn cluster(&self) -> u128;
+
+ fn validate(&self) -> Result<(), &'static str>;
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum Command {
+ Reserved = 0,
+
+ Ping = 1,
+ Pong = 2,
+ PingClient = 3,
+ PongClient = 4,
+
+ Request = 5,
+ Prepare = 6,
+ PrepareOk = 7,
+ Reply = 8,
+ Commit = 9,
+
+ StartViewChange = 10,
+ DoViewChange = 11,
+ StartView = 24,
+
+ RequestStartView = 12,
Review Comment:
Don't include those, they are not needed for now.
##########
core/common/src/types/consensus/header.rs:
##########
@@ -15,7 +15,310 @@
// specific language governing permissions and limitations
// under the License.
+use bytemuck::{Pod, Zeroable};
+
#[expect(unused)]
pub struct Header {}
-pub trait ConsensusHeader {}
+pub trait ConsensusHeader: Sized + Pod + Zeroable {
+ const COMMAND: Command;
+
+ fn size(&self) -> u32;
+ fn command(&self) -> Command;
+ fn checksum(&self) -> u128;
+ fn cluster(&self) -> u128;
+
+ fn validate(&self) -> Result<(), &'static str>;
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum Command {
+ Reserved = 0,
+
+ Ping = 1,
+ Pong = 2,
+ PingClient = 3,
+ PongClient = 4,
+
+ Request = 5,
+ Prepare = 6,
+ PrepareOk = 7,
+ Reply = 8,
+ Commit = 9,
+
+ StartViewChange = 10,
+ DoViewChange = 11,
+ StartView = 24,
+
+ RequestStartView = 12,
+ RequstHeads = 13,
+ RequestPrepare = 14,
+ RequestReply = 15,
+
+ Headers = 16,
+
+ Eviction = 17,
+
+ RequestBlocks = 18,
+ Block = 19,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum Operation {
+ Reserved = 0,
+ Root = 1,
+ Register = 2,
+ Reconfigure = 3,
+ Pulse = 4,
+ Upgrade = 5,
+
+ CreateStream = 128,
+ UpdateStream = 129,
+ DeleteStream = 130,
+ PurgeStream = 131,
+ CreateTopic = 132,
+ UpdateTopic = 133,
+ DeleteTopic = 134,
+ PurgeTopic = 135,
+ CreatePartitions = 136,
+ DeletePartitions = 137,
+ DeleteSegments = 138,
+ CreateConsumerGroup = 139,
+ DeleteConsumerGroup = 140,
+ CreateUser = 141,
+ UpdateUser = 142,
+ DeleteUser = 143,
+ ChangePassword = 144,
+ UpdatePermissions = 145,
+ CreatePersonalAccessToken = 146,
+ DeletePersonalAccessToken = 147,
+}
+
+#[repr(C)]
+#[derive(Debug, Clone, Copy)]
+pub struct GenericHeader {
+ pub checksum: u128,
+ pub checksum_padding: u128,
+ pub checksum_body: u128,
+ pub checksum_body_padding: u128,
+ pub nonce_reserved: u128,
Review Comment:
We don't need the nonce, as we won't encrypt our log using that approach,
instead it will come from an config.
##########
core/common/src/types/consensus/header.rs:
##########
@@ -15,7 +15,310 @@
// specific language governing permissions and limitations
// under the License.
+use bytemuck::{Pod, Zeroable};
+
#[expect(unused)]
pub struct Header {}
-pub trait ConsensusHeader {}
+pub trait ConsensusHeader: Sized + Pod + Zeroable {
+ const COMMAND: Command;
+
+ fn size(&self) -> u32;
+ fn command(&self) -> Command;
+ fn checksum(&self) -> u128;
+ fn cluster(&self) -> u128;
+
+ fn validate(&self) -> Result<(), &'static str>;
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum Command {
+ Reserved = 0,
+
+ Ping = 1,
+ Pong = 2,
+ PingClient = 3,
+ PongClient = 4,
+
+ Request = 5,
+ Prepare = 6,
+ PrepareOk = 7,
+ Reply = 8,
+ Commit = 9,
+
+ StartViewChange = 10,
+ DoViewChange = 11,
+ StartView = 24,
+
+ RequestStartView = 12,
+ RequstHeads = 13,
+ RequestPrepare = 14,
+ RequestReply = 15,
+
+ Headers = 16,
+
+ Eviction = 17,
+
+ RequestBlocks = 18,
+ Block = 19,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum Operation {
+ Reserved = 0,
+ Root = 1,
+ Register = 2,
+ Reconfigure = 3,
+ Pulse = 4,
+ Upgrade = 5,
+
+ CreateStream = 128,
+ UpdateStream = 129,
+ DeleteStream = 130,
+ PurgeStream = 131,
+ CreateTopic = 132,
+ UpdateTopic = 133,
+ DeleteTopic = 134,
+ PurgeTopic = 135,
+ CreatePartitions = 136,
+ DeletePartitions = 137,
+ DeleteSegments = 138,
+ CreateConsumerGroup = 139,
+ DeleteConsumerGroup = 140,
+ CreateUser = 141,
+ UpdateUser = 142,
+ DeleteUser = 143,
+ ChangePassword = 144,
+ UpdatePermissions = 145,
+ CreatePersonalAccessToken = 146,
+ DeletePersonalAccessToken = 147,
+}
+
+#[repr(C)]
+#[derive(Debug, Clone, Copy)]
+pub struct GenericHeader {
+ pub checksum: u128,
+ pub checksum_padding: u128,
+ pub checksum_body: u128,
+ pub checksum_body_padding: u128,
+ pub nonce_reserved: u128,
+ pub cluster: u128,
+ pub size: u32,
+ pub epoch: u32,
+ pub view: u32,
+ pub release: u32,
+ pub protocol: u16,
+ pub command: Command,
+ pub replica: u8,
+ pub reserved_frame: [u8; 12],
+
+ pub reserved_command: [u8; 128],
+}
+
+unsafe impl Pod for GenericHeader {}
+unsafe impl Zeroable for GenericHeader {}
+
+impl ConsensusHeader for GenericHeader {
+ const COMMAND: Command = Command::Reserved;
+
+ fn size(&self) -> u32 {
+ self.size
+ }
+ fn command(&self) -> Command {
+ self.command
+ }
+ fn checksum(&self) -> u128 {
+ self.checksum
+ }
+ fn cluster(&self) -> u128 {
+ self.cluster
+ }
+
+ fn validate(&self) -> Result<(), &'static str> {
+ Ok(())
+ }
+}
+
+#[repr(C)]
+#[derive(Debug, Clone, Copy)]
+pub struct PrepareHeader {
Review Comment:
Same applies here as the comments from `GenericHeader`
##########
core/common/src/types/consensus/header.rs:
##########
@@ -15,7 +15,310 @@
// specific language governing permissions and limitations
// under the License.
+use bytemuck::{Pod, Zeroable};
+
#[expect(unused)]
pub struct Header {}
-pub trait ConsensusHeader {}
+pub trait ConsensusHeader: Sized + Pod + Zeroable {
+ const COMMAND: Command;
+
+ fn size(&self) -> u32;
+ fn command(&self) -> Command;
+ fn checksum(&self) -> u128;
+ fn cluster(&self) -> u128;
+
+ fn validate(&self) -> Result<(), &'static str>;
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum Command {
+ Reserved = 0,
+
+ Ping = 1,
+ Pong = 2,
+ PingClient = 3,
+ PongClient = 4,
+
+ Request = 5,
+ Prepare = 6,
+ PrepareOk = 7,
+ Reply = 8,
+ Commit = 9,
+
+ StartViewChange = 10,
+ DoViewChange = 11,
+ StartView = 24,
+
+ RequestStartView = 12,
+ RequstHeads = 13,
+ RequestPrepare = 14,
+ RequestReply = 15,
+
+ Headers = 16,
+
+ Eviction = 17,
+
+ RequestBlocks = 18,
+ Block = 19,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum Operation {
+ Reserved = 0,
+ Root = 1,
+ Register = 2,
+ Reconfigure = 3,
+ Pulse = 4,
+ Upgrade = 5,
+
+ CreateStream = 128,
+ UpdateStream = 129,
+ DeleteStream = 130,
+ PurgeStream = 131,
+ CreateTopic = 132,
+ UpdateTopic = 133,
+ DeleteTopic = 134,
+ PurgeTopic = 135,
+ CreatePartitions = 136,
+ DeletePartitions = 137,
+ DeleteSegments = 138,
+ CreateConsumerGroup = 139,
+ DeleteConsumerGroup = 140,
+ CreateUser = 141,
+ UpdateUser = 142,
+ DeleteUser = 143,
+ ChangePassword = 144,
+ UpdatePermissions = 145,
+ CreatePersonalAccessToken = 146,
+ DeletePersonalAccessToken = 147,
+}
+
+#[repr(C)]
+#[derive(Debug, Clone, Copy)]
+pub struct GenericHeader {
+ pub checksum: u128,
+ pub checksum_padding: u128,
+ pub checksum_body: u128,
+ pub checksum_body_padding: u128,
+ pub nonce_reserved: u128,
+ pub cluster: u128,
+ pub size: u32,
+ pub epoch: u32,
+ pub view: u32,
+ pub release: u32,
+ pub protocol: u16,
+ pub command: Command,
+ pub replica: u8,
+ pub reserved_frame: [u8; 12],
+
+ pub reserved_command: [u8; 128],
+}
+
+unsafe impl Pod for GenericHeader {}
+unsafe impl Zeroable for GenericHeader {}
+
+impl ConsensusHeader for GenericHeader {
+ const COMMAND: Command = Command::Reserved;
+
+ fn size(&self) -> u32 {
+ self.size
+ }
+ fn command(&self) -> Command {
+ self.command
+ }
+ fn checksum(&self) -> u128 {
+ self.checksum
+ }
+ fn cluster(&self) -> u128 {
+ self.cluster
+ }
+
+ fn validate(&self) -> Result<(), &'static str> {
+ Ok(())
+ }
+}
+
+#[repr(C)]
+#[derive(Debug, Clone, Copy)]
+pub struct PrepareHeader {
+ pub checksum: u128,
+ pub checksum_padding: u128,
+ pub checksum_body: u128,
+ pub checksum_body_padding: u128,
+ pub nonce_reserved: u128,
+ pub cluster: u128,
+ pub size: u32,
+ pub epoch: u32,
+ pub view: u32,
+ pub release: u32,
+ pub protocol: u16,
+ pub command: Command,
+ pub replica: u8,
+ pub reserved_frame: [u8; 12],
+
+ pub parent: u128,
+ pub parent_padding: u128,
+ pub request_checksum: u128,
+ pub request_checksum_padding: u128,
+ pub checkpoint_id: u128,
+ pub client: u128,
+ pub op: u64,
+ pub commit: u64,
+ pub timestamp: u64,
+ pub request: u32,
Review Comment:
Let's maybe make request an u64, request is the unique identifier attached
by the client. In the worst case scenario the client can send messages one by
one, which means that there is 1-to-1 mapping between request number and
message offset, since message offset is u64, let's make this u64 aswell.
##########
core/common/src/types/consensus/header.rs:
##########
@@ -15,7 +15,310 @@
// specific language governing permissions and limitations
// under the License.
+use bytemuck::{Pod, Zeroable};
+
#[expect(unused)]
pub struct Header {}
-pub trait ConsensusHeader {}
+pub trait ConsensusHeader: Sized + Pod + Zeroable {
+ const COMMAND: Command;
+
+ fn size(&self) -> u32;
+ fn command(&self) -> Command;
+ fn checksum(&self) -> u128;
+ fn cluster(&self) -> u128;
+
+ fn validate(&self) -> Result<(), &'static str>;
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum Command {
+ Reserved = 0,
+
+ Ping = 1,
+ Pong = 2,
+ PingClient = 3,
+ PongClient = 4,
+
+ Request = 5,
+ Prepare = 6,
+ PrepareOk = 7,
+ Reply = 8,
+ Commit = 9,
+
+ StartViewChange = 10,
+ DoViewChange = 11,
+ StartView = 24,
+
+ RequestStartView = 12,
+ RequstHeads = 13,
+ RequestPrepare = 14,
+ RequestReply = 15,
+
+ Headers = 16,
+
+ Eviction = 17,
+
+ RequestBlocks = 18,
+ Block = 19,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum Operation {
+ Reserved = 0,
+ Root = 1,
+ Register = 2,
+ Reconfigure = 3,
+ Pulse = 4,
+ Upgrade = 5,
+
+ CreateStream = 128,
+ UpdateStream = 129,
+ DeleteStream = 130,
+ PurgeStream = 131,
+ CreateTopic = 132,
+ UpdateTopic = 133,
+ DeleteTopic = 134,
+ PurgeTopic = 135,
+ CreatePartitions = 136,
+ DeletePartitions = 137,
+ DeleteSegments = 138,
+ CreateConsumerGroup = 139,
+ DeleteConsumerGroup = 140,
+ CreateUser = 141,
+ UpdateUser = 142,
+ DeleteUser = 143,
+ ChangePassword = 144,
+ UpdatePermissions = 145,
+ CreatePersonalAccessToken = 146,
+ DeletePersonalAccessToken = 147,
+}
+
+#[repr(C)]
+#[derive(Debug, Clone, Copy)]
+pub struct GenericHeader {
+ pub checksum: u128,
+ pub checksum_padding: u128,
+ pub checksum_body: u128,
+ pub checksum_body_padding: u128,
+ pub nonce_reserved: u128,
+ pub cluster: u128,
+ pub size: u32,
+ pub epoch: u32,
+ pub view: u32,
+ pub release: u32,
Review Comment:
Is u32 enough for that ? check our `SemanticVersion` and how many bytes it
requires.
##########
core/common/src/types/consensus/header.rs:
##########
@@ -15,7 +15,310 @@
// specific language governing permissions and limitations
// under the License.
+use bytemuck::{Pod, Zeroable};
+
#[expect(unused)]
pub struct Header {}
-pub trait ConsensusHeader {}
+pub trait ConsensusHeader: Sized + Pod + Zeroable {
+ const COMMAND: Command;
+
+ fn size(&self) -> u32;
+ fn command(&self) -> Command;
+ fn checksum(&self) -> u128;
+ fn cluster(&self) -> u128;
+
+ fn validate(&self) -> Result<(), &'static str>;
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum Command {
+ Reserved = 0,
+
+ Ping = 1,
+ Pong = 2,
+ PingClient = 3,
+ PongClient = 4,
+
+ Request = 5,
+ Prepare = 6,
+ PrepareOk = 7,
+ Reply = 8,
+ Commit = 9,
+
+ StartViewChange = 10,
+ DoViewChange = 11,
+ StartView = 24,
+
+ RequestStartView = 12,
+ RequstHeads = 13,
+ RequestPrepare = 14,
+ RequestReply = 15,
+
+ Headers = 16,
+
+ Eviction = 17,
+
+ RequestBlocks = 18,
+ Block = 19,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum Operation {
+ Reserved = 0,
+ Root = 1,
+ Register = 2,
+ Reconfigure = 3,
+ Pulse = 4,
+ Upgrade = 5,
+
+ CreateStream = 128,
+ UpdateStream = 129,
+ DeleteStream = 130,
+ PurgeStream = 131,
+ CreateTopic = 132,
+ UpdateTopic = 133,
+ DeleteTopic = 134,
+ PurgeTopic = 135,
+ CreatePartitions = 136,
+ DeletePartitions = 137,
+ DeleteSegments = 138,
+ CreateConsumerGroup = 139,
+ DeleteConsumerGroup = 140,
+ CreateUser = 141,
+ UpdateUser = 142,
+ DeleteUser = 143,
+ ChangePassword = 144,
+ UpdatePermissions = 145,
+ CreatePersonalAccessToken = 146,
+ DeletePersonalAccessToken = 147,
+}
+
+#[repr(C)]
+#[derive(Debug, Clone, Copy)]
+pub struct GenericHeader {
+ pub checksum: u128,
+ pub checksum_padding: u128,
+ pub checksum_body: u128,
+ pub checksum_body_padding: u128,
+ pub nonce_reserved: u128,
+ pub cluster: u128,
+ pub size: u32,
+ pub epoch: u32,
+ pub view: u32,
+ pub release: u32,
+ pub protocol: u16,
+ pub command: Command,
+ pub replica: u8,
+ pub reserved_frame: [u8; 12],
+
+ pub reserved_command: [u8; 128],
+}
+
+unsafe impl Pod for GenericHeader {}
+unsafe impl Zeroable for GenericHeader {}
+
+impl ConsensusHeader for GenericHeader {
+ const COMMAND: Command = Command::Reserved;
+
+ fn size(&self) -> u32 {
+ self.size
+ }
+ fn command(&self) -> Command {
+ self.command
+ }
+ fn checksum(&self) -> u128 {
+ self.checksum
+ }
+ fn cluster(&self) -> u128 {
+ self.cluster
+ }
+
+ fn validate(&self) -> Result<(), &'static str> {
+ Ok(())
+ }
+}
+
+#[repr(C)]
+#[derive(Debug, Clone, Copy)]
+pub struct PrepareHeader {
+ pub checksum: u128,
+ pub checksum_padding: u128,
+ pub checksum_body: u128,
+ pub checksum_body_padding: u128,
+ pub nonce_reserved: u128,
+ pub cluster: u128,
+ pub size: u32,
+ pub epoch: u32,
+ pub view: u32,
+ pub release: u32,
+ pub protocol: u16,
+ pub command: Command,
+ pub replica: u8,
+ pub reserved_frame: [u8; 12],
+
+ pub parent: u128,
+ pub parent_padding: u128,
+ pub request_checksum: u128,
+ pub request_checksum_padding: u128,
+ pub checkpoint_id: u128,
+ pub client: u128,
Review Comment:
We most likely don't need to store the `client` as we won't use the
`clients_table` part of VSR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
