use super::{Command, ConsensusError, Operation};
use bytemuck::{CheckedBitPattern, NoUninit};
use std::mem::offset_of;
pub const HEADER_SIZE: usize = 256;
pub const RESERVED_COMMAND_LEN: usize = 128;
pub const SIZE_FIELD_OFFSET: usize = 48;
#[inline]
#[must_use]
pub fn read_size_field(header: &[u8]) -> Option<u32> {
header
.get(SIZE_FIELD_OFFSET..SIZE_FIELD_OFFSET + 4)
.and_then(|s| s.try_into().ok())
.map(u32::from_le_bytes)
}
#[must_use]
pub fn frame_checksum_bytes(header: &[u8; HEADER_SIZE]) -> u128 {
u128::from(twox_hash::XxHash3_64::oneshot(&header[size_of::<u128>()..]))
}
pub trait ConsensusHeader: Sized + CheckedBitPattern + NoUninit {
const COMMAND: Command;
#[must_use]
fn accepts(command: Command) -> bool {
command == Self::COMMAND
}
const FRAME_SEALED: bool;
fn validate(&self) -> Result<(), ConsensusError>;
fn operation(&self) -> Operation;
fn command(&self) -> Command;
fn size(&self) -> u32;
fn checksum(&self) -> u128;
fn set_checksum(&mut self, checksum: u128);
#[must_use]
fn frame_checksum(&self) -> u128 {
let bytes: &[u8; HEADER_SIZE] = bytemuck::bytes_of(self)
.try_into()
.expect("every consensus header is HEADER_SIZE bytes");
frame_checksum_bytes(bytes)
}
fn seal(&mut self) {
debug_assert!(
Self::FRAME_SEALED,
"sealing a header whose checksum field means something else",
);
let checksum = self.frame_checksum();
self.set_checksum(checksum);
}
fn verify_frame(&self) -> Result<(), ConsensusError> {
if !Self::FRAME_SEALED {
return Ok(());
}
let expected = self.frame_checksum();
let found = self.checksum();
if found == expected {
Ok(())
} else {
Err(ConsensusError::FrameChecksumMismatch {
command: self.command(),
expected,
found,
})
}
}
}
#[repr(C)]
#[derive(Debug, Clone, Copy, CheckedBitPattern, NoUninit)]
pub struct GenericHeader {
pub checksum: u128,
pub checksum_body: u128,
pub cluster: u128,
pub size: u32,
pub view: u32,
pub release: u32,
pub command: Command,
pub replica: u8,
pub reserved_frame: [u8; 66],
pub reserved_command: [u8; RESERVED_COMMAND_LEN],
}
const _: () = {
assert!(size_of::<GenericHeader>() == HEADER_SIZE);
assert!(
offset_of!(GenericHeader, size) == SIZE_FIELD_OFFSET,
"GenericHeader.size offset drifted; transports decode wrong bytes",
);
assert!(
offset_of!(GenericHeader, reserved_command)
== offset_of!(GenericHeader, reserved_frame) + size_of::<[u8; 66]>()
);
assert!(
offset_of!(GenericHeader, reserved_command) + size_of::<[u8; RESERVED_COMMAND_LEN]>()
== HEADER_SIZE
);
};
impl ConsensusHeader for GenericHeader {
const COMMAND: Command = Command::Reserved;
const FRAME_SEALED: bool = false;
fn checksum(&self) -> u128 {
self.checksum
}
fn set_checksum(&mut self, checksum: u128) {
self.checksum = checksum;
}
fn operation(&self) -> Operation {
Operation::Reserved
}
fn command(&self) -> Command {
self.command
}
fn validate(&self) -> Result<(), ConsensusError> {
Ok(())
}
fn size(&self) -> u32 {
self.size
}
}
#[repr(C)]
#[derive(Debug, Clone, Copy, CheckedBitPattern, NoUninit)]
pub struct RequestHeader {
pub checksum: u128,
pub checksum_body: u128,
pub cluster: u128,
pub size: u32,
pub view: u32,
pub release: u32,
pub command: Command,
pub replica: u8,
pub reserved_frame: [u8; 66],
pub client: u128,
pub request_checksum: u128,
pub timestamp: u64,
pub request: u64,
pub operation: Operation,
pub operation_padding: [u8; 7],
pub session: u64,
pub user_id: u32,
pub reserved: [u8; 60],
}
const _: () = {
assert!(size_of::<RequestHeader>() == HEADER_SIZE);
assert!(
offset_of!(RequestHeader, client)
== offset_of!(RequestHeader, reserved_frame) + size_of::<[u8; 66]>()
);
assert!(
offset_of!(RequestHeader, user_id) == offset_of!(RequestHeader, session) + size_of::<u64>()
);
assert!(offset_of!(RequestHeader, reserved) + size_of::<[u8; 60]>() == HEADER_SIZE);
};
#[repr(C)]
#[derive(Debug, Clone, Copy, CheckedBitPattern, NoUninit)]
pub struct RoutedRequestHeader {
pub checksum: u128,
pub checksum_body: u128,
pub cluster: u128,
pub size: u32,
pub view: u32,
pub release: u32,
pub command: Command,
pub replica: u8,
pub reserved_frame: [u8; 66],
pub client: u128,
pub request_checksum: u128,
pub timestamp: u64,
pub request: u64,
pub operation: Operation,
pub operation_padding: [u8; 7],
pub session: u64,
pub user_id: u32,
pub reserved: [u8; 52],
pub group: u64,
}
const _: () = {
assert!(size_of::<RoutedRequestHeader>() == HEADER_SIZE);
assert!(offset_of!(RoutedRequestHeader, client) == offset_of!(RequestHeader, client));
assert!(offset_of!(RoutedRequestHeader, session) == offset_of!(RequestHeader, session));
assert!(offset_of!(RoutedRequestHeader, user_id) == offset_of!(RequestHeader, user_id));
assert!(offset_of!(RoutedRequestHeader, reserved) == offset_of!(RequestHeader, reserved));
assert!(offset_of!(RoutedRequestHeader, group) + size_of::<u64>() == HEADER_SIZE);
};
impl Default for RoutedRequestHeader {
fn default() -> Self {
Self {
checksum: 0,
checksum_body: 0,
cluster: 0,
size: 0,
view: 0,
release: 0,
command: Command::Reserved,
replica: 0,
reserved_frame: [0; 66],
client: 0,
request_checksum: 0,
timestamp: 0,
request: 0,
operation: Operation::Reserved,
operation_padding: [0; 7],
session: 0,
user_id: 0,
reserved: [0; 52],
group: 0,
}
}
}
impl Default for RequestHeader {
fn default() -> Self {
Self {
checksum: 0,
checksum_body: 0,
cluster: 0,
size: 0,
view: 0,
release: 0,
command: Command::Reserved,
replica: 0,
reserved_frame: [0; 66],
client: 0,
request_checksum: 0,
timestamp: 0,
request: 0,
operation: Operation::Reserved,
operation_padding: [0; 7],
session: 0,
user_id: 0,
reserved: [0; 60],
}
}
}
fn validate_request_fields(
client: u128,
operation: Operation,
session: u64,
request: u64,
) -> Result<(), ConsensusError> {
if client == 0 {
return Err(ConsensusError::InvalidField(
"request: client must be != 0".to_string(),
));
}
if operation == Operation::Reserved {
return Err(ConsensusError::InvalidField(
"operation must not be Reserved".to_string(),
));
}
if operation == Operation::Register {
if session != 0 {
return Err(ConsensusError::InvalidField(
"register: session must be 0".to_string(),
));
}
if request != 0 {
return Err(ConsensusError::InvalidField(
"register: request must be 0".to_string(),
));
}
} else if operation != Operation::NonReplicated {
if session == 0 {
return Err(ConsensusError::InvalidField(
"non-register: session must be > 0".to_string(),
));
}
if request == 0 {
return Err(ConsensusError::InvalidField(
"non-register: request must be > 0".to_string(),
));
}
}
Ok(())
}
impl ConsensusHeader for RoutedRequestHeader {
const COMMAND: Command = Command::Request;
const FRAME_SEALED: bool = false;
fn checksum(&self) -> u128 {
self.checksum
}
fn set_checksum(&mut self, checksum: u128) {
self.checksum = checksum;
}
fn operation(&self) -> Operation {
self.operation
}
fn command(&self) -> Command {
self.command
}
fn size(&self) -> u32 {
self.size
}
fn validate(&self) -> Result<(), ConsensusError> {
if self.command != Command::Request {
return Err(ConsensusError::InvalidCommand {
expected: Command::Request,
found: self.command,
});
}
validate_request_fields(self.client, self.operation, self.session, self.request)
}
}
impl ConsensusHeader for RequestHeader {
const COMMAND: Command = Command::Request;
const FRAME_SEALED: bool = false;
fn checksum(&self) -> u128 {
self.checksum
}
fn set_checksum(&mut self, checksum: u128) {
self.checksum = checksum;
}
fn operation(&self) -> Operation {
self.operation
}
fn command(&self) -> Command {
self.command
}
fn size(&self) -> u32 {
self.size
}
fn validate(&self) -> Result<(), ConsensusError> {
if self.command != Command::Request {
return Err(ConsensusError::InvalidCommand {
expected: Command::Request,
found: self.command,
});
}
validate_request_fields(self.client, self.operation, self.session, self.request)
}
}
#[repr(C)]
#[derive(Debug, Clone, Copy, CheckedBitPattern, NoUninit)]
pub struct ReplyHeader {
pub checksum: u128,
pub checksum_body: u128,
pub cluster: u128,
pub size: u32,
pub view: u32,
pub release: u32,
pub command: Command,
pub replica: u8,
pub reserved_frame: [u8; 66],
pub request_checksum: u128,
pub context: u128,
pub client: u128,
pub op: u64,
pub commit: u64,
pub timestamp: u64,
pub request: u64,
pub operation: Operation,
pub operation_padding: [u8; 7],
pub status: u32,
pub reserved: [u8; 36],
}
const _: () = {
assert!(size_of::<ReplyHeader>() == HEADER_SIZE);
assert!(
offset_of!(ReplyHeader, request_checksum)
== offset_of!(ReplyHeader, reserved_frame) + size_of::<[u8; 66]>()
);
assert!(offset_of!(ReplyHeader, reserved) + size_of::<[u8; 36]>() == HEADER_SIZE);
};
impl Default for ReplyHeader {
fn default() -> Self {
Self {
checksum: 0,
checksum_body: 0,
cluster: 0,
size: 0,
view: 0,
release: 0,
command: Command::Reserved,
replica: 0,
reserved_frame: [0; 66],
request_checksum: 0,
context: 0,
client: 0,
op: 0,
commit: 0,
timestamp: 0,
request: 0,
operation: Operation::Reserved,
operation_padding: [0; 7],
status: 0,
reserved: [0; 36],
}
}
}
impl ConsensusHeader for ReplyHeader {
const COMMAND: Command = Command::Reply;
const FRAME_SEALED: bool = false;
fn checksum(&self) -> u128 {
self.checksum
}
fn set_checksum(&mut self, checksum: u128) {
self.checksum = checksum;
}
fn operation(&self) -> Operation {
self.operation
}
fn command(&self) -> Command {
self.command
}
fn size(&self) -> u32 {
self.size
}
fn validate(&self) -> Result<(), ConsensusError> {
if self.command != Command::Reply {
return Err(ConsensusError::ReplyInvalidCommand);
}
Ok(())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, NoUninit, CheckedBitPattern)]
#[repr(u8)]
pub enum EvictionReason {
Reserved = 0,
NoSession = 1,
ClientReleaseTooLow = 2,
ClientReleaseTooHigh = 3,
InvalidRequestOperation = 4,
InvalidRequestBody = 5,
InvalidRequestBodySize = 6,
SessionTooLow = 7,
SessionReleaseMismatch = 8,
InvalidCredentials = 9,
InvalidToken = 10,
UserInactive = 11,
SessionError = 12,
StaleClient = 13,
IncompatibleProtocol = 14,
MalformedLogin = 15,
}
#[repr(C)]
#[derive(Debug, Clone, Copy, CheckedBitPattern, NoUninit)]
pub struct EvictionHeader {
pub checksum: u128,
pub checksum_body: u128,
pub cluster: u128,
pub size: u32,
pub view: u32,
pub release: u32,
pub command: Command,
pub replica: u8,
pub reserved_frame: [u8; 66],
pub client: u128,
pub server_protocol_version: u32,
pub server_protocol_version_min: u32,
pub reserved: [u8; 103],
pub reason: EvictionReason,
}
const _: () = {
assert!(size_of::<EvictionHeader>() == HEADER_SIZE);
assert!(
offset_of!(EvictionHeader, client)
== offset_of!(EvictionHeader, reserved_frame) + size_of::<[u8; 66]>()
);
assert!(offset_of!(EvictionHeader, server_protocol_version) == 144);
assert!(offset_of!(EvictionHeader, server_protocol_version_min) == 148);
assert!(offset_of!(EvictionHeader, reason) + size_of::<EvictionReason>() == HEADER_SIZE);
};
impl EvictionHeader {
#[must_use]
#[allow(clippy::cast_possible_truncation)]
pub const fn new(
cluster: u128,
view: u32,
replica: u8,
client: u128,
reason: EvictionReason,
) -> Self {
debug_assert!(
!matches!(
reason,
EvictionReason::ClientReleaseTooLow
| EvictionReason::ClientReleaseTooHigh
| EvictionReason::IncompatibleProtocol,
),
"EvictionHeader::new: ClientRelease*/IncompatibleProtocol need extra fields",
);
debug_assert!(
replica < 32,
"EvictionHeader::new: replica >= REPLICAS_MAX(32)",
);
Self {
checksum: 0,
checksum_body: 0,
cluster,
size: HEADER_SIZE as u32,
view,
release: 0,
command: Command::Eviction,
replica,
reserved_frame: [0; 66],
client,
server_protocol_version: 0,
server_protocol_version_min: 0,
reserved: [0; 103],
reason,
}
}
#[must_use]
pub const fn incompatible_protocol(
cluster: u128,
view: u32,
replica: u8,
client: u128,
server_protocol_version: u32,
server_protocol_version_min: u32,
) -> Self {
let mut header = Self::new(cluster, view, replica, client, EvictionReason::NoSession);
header.reason = EvictionReason::IncompatibleProtocol;
header.server_protocol_version = server_protocol_version;
header.server_protocol_version_min = server_protocol_version_min;
header
}
}
impl ConsensusHeader for EvictionHeader {
const COMMAND: Command = Command::Eviction;
const FRAME_SEALED: bool = false;
fn checksum(&self) -> u128 {
self.checksum
}
fn set_checksum(&mut self, checksum: u128) {
self.checksum = checksum;
}
fn operation(&self) -> Operation {
Operation::Reserved
}
fn command(&self) -> Command {
self.command
}
fn size(&self) -> u32 {
self.size
}
#[allow(clippy::cast_possible_truncation)]
fn validate(&self) -> Result<(), ConsensusError> {
if self.command != Command::Eviction {
return Err(ConsensusError::InvalidCommand {
expected: Command::Eviction,
found: self.command,
});
}
if self.size as usize != HEADER_SIZE {
return Err(ConsensusError::InvalidSize {
expected: HEADER_SIZE as u32,
found: self.size,
});
}
if self.client == 0 {
return Err(ConsensusError::InvalidField(
"eviction: client must be != 0".to_string(),
));
}
if self.reserved_frame.iter().any(|&b| b != 0) {
return Err(ConsensusError::InvalidField(
"eviction: reserved_frame bytes must be zero".to_string(),
));
}
if self.reserved.iter().any(|&b| b != 0) {
return Err(ConsensusError::InvalidField(
"eviction: reserved bytes must be zero".to_string(),
));
}
if self.reason == EvictionReason::Reserved {
return Err(ConsensusError::InvalidField(
"eviction: reason must not be Reserved".to_string(),
));
}
if self.reason == EvictionReason::IncompatibleProtocol {
if self.server_protocol_version_min == 0
|| self.server_protocol_version < self.server_protocol_version_min
{
return Err(ConsensusError::InvalidField(
"eviction: incompatible-protocol window must satisfy 1 <= min <= max"
.to_string(),
));
}
} else if self.server_protocol_version != 0 || self.server_protocol_version_min != 0 {
return Err(ConsensusError::InvalidField(
"eviction: protocol window bytes must be zero".to_string(),
));
}
Ok(())
}
}
#[repr(C)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
pub struct PrepareHeader {
pub checksum: u128,
pub checksum_body: u128,
pub cluster: u128,
pub size: u32,
pub view: u32,
pub release: u32,
pub command: Command,
pub replica: u8,
pub reserved_frame: [u8; 66],
pub client: u128,
pub parent: u128,
pub request_checksum: u128,
pub op: u64,
pub commit: u64,
pub timestamp: u64,
pub request: u64,
pub operation: Operation,
pub operation_padding: [u8; 7],
pub group: u64,
pub user_id: u32,
pub reserved: [u8; 28],
}
const _: () = {
assert!(size_of::<PrepareHeader>() == HEADER_SIZE);
assert!(
offset_of!(PrepareHeader, client)
== offset_of!(PrepareHeader, reserved_frame) + size_of::<[u8; 66]>()
);
assert!(
offset_of!(PrepareHeader, user_id) == offset_of!(PrepareHeader, group) + size_of::<u64>()
);
assert!(offset_of!(PrepareHeader, reserved) + size_of::<[u8; 28]>() == HEADER_SIZE);
};
impl Default for PrepareHeader {
fn default() -> Self {
Self {
checksum: 0,
checksum_body: 0,
cluster: 0,
size: 0,
view: 0,
release: 0,
command: Command::Reserved,
replica: 0,
reserved_frame: [0; 66],
client: 0,
parent: 0,
request_checksum: 0,
op: 0,
commit: 0,
timestamp: 0,
request: 0,
operation: Operation::Reserved,
operation_padding: [0; 7],
group: 0,
user_id: 0,
reserved: [0; 28],
}
}
}
impl ConsensusHeader for PrepareHeader {
const COMMAND: Command = Command::Prepare;
const FRAME_SEALED: bool = false;
fn checksum(&self) -> u128 {
self.checksum
}
fn set_checksum(&mut self, checksum: u128) {
self.checksum = checksum;
}
fn operation(&self) -> Operation {
self.operation
}
fn command(&self) -> Command {
self.command
}
fn size(&self) -> u32 {
self.size
}
fn validate(&self) -> Result<(), ConsensusError> {
if self.command != Command::Prepare {
return Err(ConsensusError::InvalidCommand {
expected: Command::Prepare,
found: self.command,
});
}
if self.reserved_frame.iter().any(|&byte| byte != 0) {
return Err(ConsensusError::InvalidField(
"prepare: reserved_frame bytes must be zero".to_string(),
));
}
if self.reserved.iter().any(|&byte| byte != 0) {
return Err(ConsensusError::InvalidField(
"prepare: reserved bytes must be zero".to_string(),
));
}
Ok(())
}
}
pub const CHECKSUM_UNSEALED: u128 = 0;
#[must_use]
pub fn frame_body(frame: &[u8], size: u32) -> &[u8] {
let end = size as usize;
if end <= HEADER_SIZE || end > frame.len() {
return &[];
}
&frame[HEADER_SIZE..end]
}
impl PrepareHeader {
#[must_use]
pub fn identity_checksum(&self) -> u128 {
let mut covered = *self;
covered.checksum = 0;
covered.view = 0;
u128::from(twox_hash::XxHash3_64::oneshot(bytemuck::bytes_of(&covered)))
}
}
#[repr(transparent)]
#[derive(Debug, Clone, Copy, CheckedBitPattern, NoUninit)]
pub struct RepairPrepareHeader(pub PrepareHeader);
impl ConsensusHeader for RepairPrepareHeader {
const COMMAND: Command = Command::RepairPrepare;
const FRAME_SEALED: bool = false;
fn checksum(&self) -> u128 {
self.0.checksum
}
fn set_checksum(&mut self, checksum: u128) {
self.0.checksum = checksum;
}
fn operation(&self) -> Operation {
self.0.operation
}
fn command(&self) -> Command {
self.0.command
}
fn size(&self) -> u32 {
self.0.size
}
fn validate(&self) -> Result<(), ConsensusError> {
if self.0.command != Command::RepairPrepare {
return Err(ConsensusError::InvalidCommand {
expected: Command::RepairPrepare,
found: self.0.command,
});
}
if self.0.reserved_frame.iter().any(|&byte| byte != 0) {
return Err(ConsensusError::InvalidField(
"repair_prepare: reserved_frame bytes must be zero".to_string(),
));
}
if self.0.reserved.iter().any(|&byte| byte != 0) {
return Err(ConsensusError::InvalidField(
"repair_prepare: reserved bytes must be zero".to_string(),
));
}
Ok(())
}
}
#[repr(C)]
#[derive(Debug, Clone, Copy, CheckedBitPattern, NoUninit)]
pub struct PrepareOkHeader {
pub checksum: u128,
pub checksum_body: u128,
pub cluster: u128,
pub size: u32,
pub view: u32,
pub release: u32,
pub command: Command,
pub replica: u8,
pub reserved_frame: [u8; 66],
pub parent: u128,
pub prepare_checksum: u128,
pub op: u64,
pub commit: u64,
pub timestamp: u64,
pub request: u64,
pub operation: Operation,
pub operation_padding: [u8; 7],
pub group: u64,
pub reserved: [u8; 48],
}
const _: () = {
assert!(size_of::<PrepareOkHeader>() == HEADER_SIZE);
assert!(
offset_of!(PrepareOkHeader, parent)
== offset_of!(PrepareOkHeader, reserved_frame) + size_of::<[u8; 66]>()
);
assert!(offset_of!(PrepareOkHeader, reserved) + size_of::<[u8; 48]>() == HEADER_SIZE);
};
impl Default for PrepareOkHeader {
fn default() -> Self {
Self {
checksum: 0,
checksum_body: 0,
cluster: 0,
size: 0,
view: 0,
release: 0,
command: Command::Reserved,
replica: 0,
reserved_frame: [0; 66],
parent: 0,
prepare_checksum: 0,
op: 0,
commit: 0,
timestamp: 0,
request: 0,
operation: Operation::Reserved,
operation_padding: [0; 7],
group: 0,
reserved: [0; 48],
}
}
}
impl ConsensusHeader for PrepareOkHeader {
const FRAME_SEALED: bool = true;
const COMMAND: Command = Command::PrepareOk;
fn checksum(&self) -> u128 {
self.checksum
}
fn set_checksum(&mut self, checksum: u128) {
self.checksum = checksum;
}
fn operation(&self) -> Operation {
self.operation
}
fn command(&self) -> Command {
self.command
}
fn size(&self) -> u32 {
self.size
}
fn validate(&self) -> Result<(), ConsensusError> {
if self.command != Command::PrepareOk {
return Err(ConsensusError::InvalidCommand {
expected: Command::PrepareOk,
found: self.command,
});
}
Ok(())
}
}
#[repr(C)]
#[derive(Debug, Clone, Copy, CheckedBitPattern, NoUninit)]
pub struct CommitHeader {
pub checksum: u128,
pub checksum_body: u128,
pub cluster: u128,
pub size: u32,
pub view: u32,
pub release: u32,
pub command: Command,
pub replica: u8,
pub reserved_frame: [u8; 66],
pub commit_checksum: u128,
pub timestamp_monotonic: u64,
pub commit: u64,
pub checkpoint_op: u64,
pub group: u64,
pub reserved: [u8; 80],
}
const _: () = {
assert!(size_of::<CommitHeader>() == HEADER_SIZE);
assert!(
offset_of!(CommitHeader, commit_checksum)
== offset_of!(CommitHeader, reserved_frame) + size_of::<[u8; 66]>()
);
assert!(offset_of!(CommitHeader, reserved) + size_of::<[u8; 80]>() == HEADER_SIZE);
};
impl ConsensusHeader for CommitHeader {
const FRAME_SEALED: bool = true;
const COMMAND: Command = Command::Commit;
fn checksum(&self) -> u128 {
self.checksum
}
fn set_checksum(&mut self, checksum: u128) {
self.checksum = checksum;
}
fn operation(&self) -> Operation {
Operation::Reserved
}
fn command(&self) -> Command {
self.command
}
fn size(&self) -> u32 {
self.size
}
fn validate(&self) -> Result<(), ConsensusError> {
if self.command != Command::Commit {
return Err(ConsensusError::CommitInvalidCommand);
}
if self.size != 256 {
return Err(ConsensusError::CommitInvalidSize(self.size));
}
Ok(())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
#[repr(C)]
pub struct StartViewChangeHeader {
pub checksum: u128,
pub checksum_body: u128,
pub cluster: u128,
pub size: u32,
pub view: u32,
pub release: u32,
pub command: Command,
pub replica: u8,
pub reserved_frame: [u8; 66],
pub group: u64,
pub reserved: [u8; 120],
}
const _: () = {
assert!(size_of::<StartViewChangeHeader>() == HEADER_SIZE);
assert!(
offset_of!(StartViewChangeHeader, group)
== offset_of!(StartViewChangeHeader, reserved_frame) + size_of::<[u8; 66]>()
);
assert!(offset_of!(StartViewChangeHeader, reserved) + size_of::<[u8; 120]>() == HEADER_SIZE);
};
impl ConsensusHeader for StartViewChangeHeader {
const FRAME_SEALED: bool = true;
const COMMAND: Command = Command::StartViewChange;
fn checksum(&self) -> u128 {
self.checksum
}
fn set_checksum(&mut self, checksum: u128) {
self.checksum = checksum;
}
fn operation(&self) -> Operation {
Operation::Reserved
}
fn command(&self) -> Command {
self.command
}
fn size(&self) -> u32 {
self.size
}
fn validate(&self) -> Result<(), ConsensusError> {
if self.command != Command::StartViewChange {
return Err(ConsensusError::InvalidCommand {
expected: Command::StartViewChange,
found: self.command,
});
}
if self.release != 0 {
return Err(ConsensusError::InvalidField("release != 0".to_string()));
}
Ok(())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
#[repr(C)]
pub struct DoViewChangeHeader {
pub checksum: u128,
pub checksum_body: u128,
pub cluster: u128,
pub size: u32,
pub view: u32,
pub release: u32,
pub command: Command,
pub replica: u8,
pub reserved_frame: [u8; 66],
pub op: u64,
pub commit: u64,
pub group: u64,
pub log_view: u32,
pub reserved: [u8; 68],
pub nack_bitset: u128,
pub present_bitset: u128,
}
const _: () = {
assert!(size_of::<DoViewChangeHeader>() == HEADER_SIZE);
assert!(
offset_of!(DoViewChangeHeader, op)
== offset_of!(DoViewChangeHeader, reserved_frame) + size_of::<[u8; 66]>()
);
assert!(offset_of!(DoViewChangeHeader, reserved) == 156);
assert!(offset_of!(DoViewChangeHeader, nack_bitset) % 16 == 0);
assert!(offset_of!(DoViewChangeHeader, present_bitset) % 16 == 0);
assert!(
offset_of!(DoViewChangeHeader, nack_bitset)
== offset_of!(DoViewChangeHeader, reserved) + size_of::<[u8; 68]>()
);
assert!(offset_of!(DoViewChangeHeader, present_bitset) + size_of::<u128>() == HEADER_SIZE);
};
pub const DVC_HEADERS_MAX: usize = 128;
impl ConsensusHeader for DoViewChangeHeader {
const FRAME_SEALED: bool = true;
const COMMAND: Command = Command::DoViewChange;
fn checksum(&self) -> u128 {
self.checksum
}
fn set_checksum(&mut self, checksum: u128) {
self.checksum = checksum;
}
fn operation(&self) -> Operation {
Operation::Reserved
}
fn command(&self) -> Command {
self.command
}
fn size(&self) -> u32 {
self.size
}
fn validate(&self) -> Result<(), ConsensusError> {
if self.command != Command::DoViewChange {
return Err(ConsensusError::InvalidCommand {
expected: Command::DoViewChange,
found: self.command,
});
}
if self.release != 0 {
return Err(ConsensusError::InvalidField(
"release must be 0".to_string(),
));
}
if self.log_view > self.view {
return Err(ConsensusError::InvalidField(
"log_view cannot exceed view".to_string(),
));
}
if self.commit > self.op {
return Err(ConsensusError::InvalidField(
"commit cannot exceed op".to_string(),
));
}
let suffix_len = self.suffix_len()?;
if suffix_len < DVC_HEADERS_MAX {
let beyond = !((1u128 << suffix_len) - 1);
if self.nack_bitset & beyond != 0 || self.present_bitset & beyond != 0 {
return Err(ConsensusError::InvalidField(format!(
"do_view_change: bitset bits set past the {suffix_len}-entry suffix"
)));
}
}
Ok(())
}
}
impl DoViewChangeHeader {
pub fn suffix_len(&self) -> Result<usize, ConsensusError> {
suffix_len_of("do_view_change", self.size)
}
}
fn suffix_len_of(frame: &str, size: u32) -> Result<usize, ConsensusError> {
let size = size as usize;
let Some(body_len) = size.checked_sub(HEADER_SIZE) else {
return Err(ConsensusError::InvalidField(format!(
"{frame}: size {size} is shorter than the {HEADER_SIZE}-byte header"
)));
};
if body_len % HEADER_SIZE != 0 {
return Err(ConsensusError::InvalidField(format!(
"{frame}: body of {body_len} bytes is not a whole number of headers"
)));
}
let suffix_len = body_len / HEADER_SIZE;
if suffix_len > DVC_HEADERS_MAX {
return Err(ConsensusError::InvalidField(format!(
"{frame}: {suffix_len} suffix entries exceeds the maximum {DVC_HEADERS_MAX}"
)));
}
Ok(suffix_len)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
#[repr(C)]
pub struct StartViewHeader {
pub checksum: u128,
pub checksum_body: u128,
pub cluster: u128,
pub size: u32,
pub view: u32,
pub release: u32,
pub command: Command,
pub replica: u8,
pub reserved_frame: [u8; 66],
pub op: u64,
pub commit: u64,
pub group: u64,
pub reserved: [u8; 88],
pub incarnation: u128,
}
const _: () = {
assert!(size_of::<StartViewHeader>() == HEADER_SIZE);
assert!(
offset_of!(StartViewHeader, op)
== offset_of!(StartViewHeader, reserved_frame) + size_of::<[u8; 66]>()
);
assert!(offset_of!(StartViewHeader, incarnation) + size_of::<u128>() == HEADER_SIZE);
assert!(offset_of!(StartViewHeader, incarnation) % 16 == 0);
};
impl ConsensusHeader for StartViewHeader {
const FRAME_SEALED: bool = true;
const COMMAND: Command = Command::StartView;
fn checksum(&self) -> u128 {
self.checksum
}
fn set_checksum(&mut self, checksum: u128) {
self.checksum = checksum;
}
fn operation(&self) -> Operation {
Operation::Reserved
}
fn command(&self) -> Command {
self.command
}
fn size(&self) -> u32 {
self.size
}
fn validate(&self) -> Result<(), ConsensusError> {
if self.command != Command::StartView {
return Err(ConsensusError::InvalidCommand {
expected: Command::StartView,
found: self.command,
});
}
if self.release != 0 {
return Err(ConsensusError::InvalidField(
"release must be 0".to_string(),
));
}
if self.commit > self.op {
return Err(ConsensusError::InvalidField(
"commit cannot exceed op".to_string(),
));
}
self.suffix_len()?;
Ok(())
}
}
impl StartViewHeader {
pub fn suffix_len(&self) -> Result<usize, ConsensusError> {
suffix_len_of("start_view", self.size)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
#[repr(C)]
pub struct RequestStartViewHeader {
pub checksum: u128,
pub checksum_body: u128,
pub cluster: u128,
pub size: u32,
pub view: u32,
pub release: u32,
pub command: Command,
pub replica: u8,
pub reserved_frame: [u8; 66],
pub group: u64,
pub reserved: [u8; 104],
pub incarnation: u128,
}
const _: () = {
assert!(size_of::<RequestStartViewHeader>() == HEADER_SIZE);
assert!(
offset_of!(RequestStartViewHeader, group)
== offset_of!(RequestStartViewHeader, reserved_frame) + size_of::<[u8; 66]>()
);
assert!(offset_of!(RequestStartViewHeader, incarnation) + size_of::<u128>() == HEADER_SIZE);
assert!(offset_of!(RequestStartViewHeader, incarnation) % 16 == 0);
};
impl ConsensusHeader for RequestStartViewHeader {
const FRAME_SEALED: bool = true;
const COMMAND: Command = Command::RequestStartView;
fn checksum(&self) -> u128 {
self.checksum
}
fn set_checksum(&mut self, checksum: u128) {
self.checksum = checksum;
}
fn operation(&self) -> Operation {
Operation::Reserved
}
fn command(&self) -> Command {
self.command
}
fn size(&self) -> u32 {
self.size
}
fn validate(&self) -> Result<(), ConsensusError> {
if self.command != Command::RequestStartView {
return Err(ConsensusError::InvalidCommand {
expected: Command::RequestStartView,
found: self.command,
});
}
if self.release != 0 {
return Err(ConsensusError::InvalidField(
"release must be 0".to_string(),
));
}
Ok(())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
#[repr(C)]
pub struct RequestPreparesHeader {
pub checksum: u128,
pub checksum_body: u128,
pub cluster: u128,
pub size: u32,
pub view: u32,
pub release: u32,
pub command: Command,
pub replica: u8,
pub reserved_frame: [u8; 66],
pub nonce: u128,
pub from_op: u64,
pub to_op: u64,
pub group: u64,
pub reserved: [u8; 88],
}
const _: () = {
assert!(size_of::<RequestPreparesHeader>() == HEADER_SIZE);
assert!(
offset_of!(RequestPreparesHeader, nonce)
== offset_of!(RequestPreparesHeader, reserved_frame) + size_of::<[u8; 66]>()
);
assert!(offset_of!(RequestPreparesHeader, reserved) + size_of::<[u8; 88]>() == HEADER_SIZE);
};
impl ConsensusHeader for RequestPreparesHeader {
const FRAME_SEALED: bool = true;
const COMMAND: Command = Command::RequestPrepares;
fn checksum(&self) -> u128 {
self.checksum
}
fn set_checksum(&mut self, checksum: u128) {
self.checksum = checksum;
}
fn operation(&self) -> Operation {
Operation::Reserved
}
fn command(&self) -> Command {
self.command
}
fn size(&self) -> u32 {
self.size
}
fn validate(&self) -> Result<(), ConsensusError> {
if self.command != Command::RequestPrepares {
return Err(ConsensusError::InvalidCommand {
expected: Command::RequestPrepares,
found: self.command,
});
}
if self.from_op == 0 || self.from_op > self.to_op {
return Err(ConsensusError::InvalidField(
"repair range must be non-empty and 1-based".to_string(),
));
}
Ok(())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
#[repr(C)]
pub struct RepairRangeReplyHeader {
pub checksum: u128,
pub checksum_body: u128,
pub cluster: u128,
pub size: u32,
pub view: u32,
pub release: u32,
pub command: Command,
pub replica: u8,
pub reserved_frame: [u8; 66],
pub nonce: u128,
pub op: u64,
pub group: u64,
pub reserved: [u8; 96],
}
const _: () = {
assert!(size_of::<RepairRangeReplyHeader>() == HEADER_SIZE);
assert!(
offset_of!(RepairRangeReplyHeader, nonce)
== offset_of!(RepairRangeReplyHeader, reserved_frame) + size_of::<[u8; 66]>()
);
assert!(offset_of!(RepairRangeReplyHeader, reserved) + size_of::<[u8; 96]>() == HEADER_SIZE);
};
impl ConsensusHeader for RepairRangeReplyHeader {
const FRAME_SEALED: bool = true;
const COMMAND: Command = Command::RepairDone;
fn checksum(&self) -> u128 {
self.checksum
}
fn set_checksum(&mut self, checksum: u128) {
self.checksum = checksum;
}
fn accepts(command: Command) -> bool {
command == Command::RepairDone || command == Command::RangeEvicted
}
fn operation(&self) -> Operation {
Operation::Reserved
}
fn command(&self) -> Command {
self.command
}
fn size(&self) -> u32 {
self.size
}
fn validate(&self) -> Result<(), ConsensusError> {
if self.command != Command::RepairDone && self.command != Command::RangeEvicted {
return Err(ConsensusError::InvalidCommand {
expected: Command::RepairDone,
found: self.command,
});
}
Ok(())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
#[repr(C)]
pub struct RequestStateTransferHeader {
pub checksum: u128,
pub checksum_body: u128,
pub cluster: u128,
pub size: u32,
pub view: u32,
pub release: u32,
pub command: Command,
pub replica: u8,
pub reserved_frame: [u8; 66],
pub nonce: u128,
pub group: u64,
pub reserved: [u8; 104],
}
const _: () = {
assert!(size_of::<RequestStateTransferHeader>() == HEADER_SIZE);
assert!(
offset_of!(RequestStateTransferHeader, nonce)
== offset_of!(RequestStateTransferHeader, reserved_frame) + size_of::<[u8; 66]>()
);
assert!(
offset_of!(RequestStateTransferHeader, reserved) + size_of::<[u8; 104]>() == HEADER_SIZE
);
};
impl ConsensusHeader for RequestStateTransferHeader {
const FRAME_SEALED: bool = true;
const COMMAND: Command = Command::RequestStateTransfer;
fn checksum(&self) -> u128 {
self.checksum
}
fn set_checksum(&mut self, checksum: u128) {
self.checksum = checksum;
}
fn operation(&self) -> Operation {
Operation::Reserved
}
fn command(&self) -> Command {
self.command
}
fn size(&self) -> u32 {
self.size
}
#[allow(clippy::cast_possible_truncation)]
fn validate(&self) -> Result<(), ConsensusError> {
if self.command != Command::RequestStateTransfer {
return Err(ConsensusError::InvalidCommand {
expected: Command::RequestStateTransfer,
found: self.command,
});
}
if self.size as usize != HEADER_SIZE {
return Err(ConsensusError::InvalidSize {
expected: HEADER_SIZE as u32,
found: self.size,
});
}
Ok(())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
#[repr(C)]
pub struct StateTransferTargetHeader {
pub checksum: u128,
pub checksum_body: u128,
pub cluster: u128,
pub size: u32,
pub view: u32,
pub release: u32,
pub command: Command,
pub replica: u8,
pub reserved_frame: [u8; 66],
pub nonce: u128,
pub commit_op: u64,
pub group: u64,
pub available: u8,
pub unavailable_transient: u8,
pub reserved_alignment: [u8; 6],
pub commit_max: u64,
pub reserved: [u8; 80],
}
const _: () = {
assert!(size_of::<StateTransferTargetHeader>() == HEADER_SIZE);
assert!(
offset_of!(StateTransferTargetHeader, nonce)
== offset_of!(StateTransferTargetHeader, reserved_frame) + size_of::<[u8; 66]>()
);
assert!(offset_of!(StateTransferTargetHeader, commit_op) == 144);
assert!(offset_of!(StateTransferTargetHeader, group) == 152);
assert!(offset_of!(StateTransferTargetHeader, available) == 160);
assert!(offset_of!(StateTransferTargetHeader, unavailable_transient) == 161);
assert!(offset_of!(StateTransferTargetHeader, commit_max) == 168);
assert!(offset_of!(StateTransferTargetHeader, reserved) + size_of::<[u8; 80]>() == HEADER_SIZE);
};
impl ConsensusHeader for StateTransferTargetHeader {
const FRAME_SEALED: bool = true;
const COMMAND: Command = Command::StateTransferTarget;
fn checksum(&self) -> u128 {
self.checksum
}
fn set_checksum(&mut self, checksum: u128) {
self.checksum = checksum;
}
fn operation(&self) -> Operation {
Operation::Reserved
}
fn command(&self) -> Command {
self.command
}
fn size(&self) -> u32 {
self.size
}
fn validate(&self) -> Result<(), ConsensusError> {
if self.command != Command::StateTransferTarget {
return Err(ConsensusError::InvalidCommand {
expected: Command::StateTransferTarget,
found: self.command,
});
}
if self.available > 1 {
return Err(ConsensusError::InvalidField(
"available must be 0 or 1".to_string(),
));
}
if self.unavailable_transient > 1 {
return Err(ConsensusError::InvalidField(
"unavailable_transient must be 0 or 1".to_string(),
));
}
if self.available == 1 && self.unavailable_transient == 1 {
return Err(ConsensusError::InvalidField(
"unavailable_transient must be 0 on an available offer".to_string(),
));
}
if self.available == 0 && self.size as usize != HEADER_SIZE {
return Err(ConsensusError::InvalidField(
"unavailable descriptor must be header-only".to_string(),
));
}
Ok(())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
#[repr(C)]
pub struct RequestStateChunkHeader {
pub checksum: u128,
pub checksum_body: u128,
pub cluster: u128,
pub size: u32,
pub view: u32,
pub release: u32,
pub command: Command,
pub replica: u8,
pub reserved_frame: [u8; 66],
pub nonce: u128,
pub offset: u64,
pub group: u64,
pub len: u32,
pub artifact: u32,
pub reserved: [u8; 88],
}
const _: () = {
assert!(size_of::<RequestStateChunkHeader>() == HEADER_SIZE);
assert!(
offset_of!(RequestStateChunkHeader, nonce)
== offset_of!(RequestStateChunkHeader, reserved_frame) + size_of::<[u8; 66]>()
);
assert!(offset_of!(RequestStateChunkHeader, reserved) + size_of::<[u8; 88]>() == HEADER_SIZE);
};
impl ConsensusHeader for RequestStateChunkHeader {
const FRAME_SEALED: bool = true;
const COMMAND: Command = Command::RequestStateChunk;
fn checksum(&self) -> u128 {
self.checksum
}
fn set_checksum(&mut self, checksum: u128) {
self.checksum = checksum;
}
fn operation(&self) -> Operation {
Operation::Reserved
}
fn command(&self) -> Command {
self.command
}
fn size(&self) -> u32 {
self.size
}
#[allow(clippy::cast_possible_truncation)]
fn validate(&self) -> Result<(), ConsensusError> {
if self.command != Command::RequestStateChunk {
return Err(ConsensusError::InvalidCommand {
expected: Command::RequestStateChunk,
found: self.command,
});
}
if self.len == 0 {
return Err(ConsensusError::InvalidField(
"chunk len must be non-zero".to_string(),
));
}
if self.size as usize != HEADER_SIZE {
return Err(ConsensusError::InvalidSize {
expected: HEADER_SIZE as u32,
found: self.size,
});
}
Ok(())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
#[repr(C)]
pub struct StateChunkHeader {
pub checksum: u128,
pub checksum_body: u128,
pub cluster: u128,
pub size: u32,
pub view: u32,
pub release: u32,
pub command: Command,
pub replica: u8,
pub reserved_frame: [u8; 66],
pub nonce: u128,
pub offset: u64,
pub group: u64,
pub artifact: u32,
pub reserved: [u8; 92],
}
const _: () = {
assert!(size_of::<StateChunkHeader>() == HEADER_SIZE);
assert!(
offset_of!(StateChunkHeader, nonce)
== offset_of!(StateChunkHeader, reserved_frame) + size_of::<[u8; 66]>()
);
assert!(offset_of!(StateChunkHeader, reserved) + size_of::<[u8; 92]>() == HEADER_SIZE);
};
impl ConsensusHeader for StateChunkHeader {
const FRAME_SEALED: bool = true;
const COMMAND: Command = Command::StateChunk;
fn checksum(&self) -> u128 {
self.checksum
}
fn set_checksum(&mut self, checksum: u128) {
self.checksum = checksum;
}
fn operation(&self) -> Operation {
Operation::Reserved
}
fn command(&self) -> Command {
self.command
}
fn size(&self) -> u32 {
self.size
}
fn validate(&self) -> Result<(), ConsensusError> {
if self.command != Command::StateChunk {
return Err(ConsensusError::InvalidCommand {
expected: Command::StateChunk,
found: self.command,
});
}
if (self.size as usize) < HEADER_SIZE {
return Err(ConsensusError::InvalidField(
"state chunk size below header size".to_string(),
));
}
Ok(())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
#[repr(C)]
pub struct ForwardRegisterHeader {
pub checksum: u128,
pub checksum_body: u128,
pub cluster: u128,
pub size: u32,
pub view: u32,
pub release: u32,
pub command: Command,
pub replica: u8,
pub reserved_frame: [u8; 66],
pub client: u128,
pub nonce: u128,
pub user_id: u32,
pub reserved: [u8; 92],
}
const _: () = {
assert!(size_of::<ForwardRegisterHeader>() == HEADER_SIZE);
assert!(
offset_of!(ForwardRegisterHeader, client)
== offset_of!(ForwardRegisterHeader, reserved_frame) + size_of::<[u8; 66]>()
);
assert!(offset_of!(ForwardRegisterHeader, nonce) == 144);
assert!(offset_of!(ForwardRegisterHeader, user_id) == 160);
assert!(offset_of!(ForwardRegisterHeader, reserved) + size_of::<[u8; 92]>() == HEADER_SIZE);
};
impl ConsensusHeader for ForwardRegisterHeader {
const FRAME_SEALED: bool = true;
const COMMAND: Command = Command::ForwardRegister;
fn checksum(&self) -> u128 {
self.checksum
}
fn set_checksum(&mut self, checksum: u128) {
self.checksum = checksum;
}
fn operation(&self) -> Operation {
Operation::Reserved
}
fn command(&self) -> Command {
self.command
}
fn size(&self) -> u32 {
self.size
}
fn validate(&self) -> Result<(), ConsensusError> {
if self.command != Command::ForwardRegister {
return Err(ConsensusError::InvalidCommand {
expected: Command::ForwardRegister,
found: self.command,
});
}
validate_forward_register_frame(self.size, self.client, self.nonce, &self.reserved)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
#[repr(C)]
pub struct ForwardRegisterResultHeader {
pub checksum: u128,
pub checksum_body: u128,
pub cluster: u128,
pub size: u32,
pub view: u32,
pub release: u32,
pub command: Command,
pub replica: u8,
pub reserved_frame: [u8; 66],
pub nonce: u128,
pub client: u128,
pub epoch: u64,
pub watermark: u64,
pub reserved: [u8; 79],
pub outcome: ForwardRegisterOutcome,
}
const _: () = {
assert!(size_of::<ForwardRegisterResultHeader>() == HEADER_SIZE);
assert!(
offset_of!(ForwardRegisterResultHeader, nonce)
== offset_of!(ForwardRegisterResultHeader, reserved_frame) + size_of::<[u8; 66]>()
);
assert!(offset_of!(ForwardRegisterResultHeader, client) == 144);
assert!(offset_of!(ForwardRegisterResultHeader, epoch) == 160);
assert!(offset_of!(ForwardRegisterResultHeader, watermark) == 168);
assert!(
offset_of!(ForwardRegisterResultHeader, outcome) + size_of::<ForwardRegisterOutcome>()
== HEADER_SIZE
);
};
#[derive(Debug, Clone, Copy, PartialEq, Eq, NoUninit, CheckedBitPattern)]
#[repr(u8)]
pub enum ForwardRegisterOutcome {
Ok = 0,
NotPrimary = 1,
NotCaughtUp = 2,
PipelineFull = 3,
InProgress = 4,
Canceled = 5,
ClientIdOwnedByAnotherUser = 6,
}
impl ConsensusHeader for ForwardRegisterResultHeader {
const FRAME_SEALED: bool = true;
const COMMAND: Command = Command::ForwardRegisterResult;
fn checksum(&self) -> u128 {
self.checksum
}
fn set_checksum(&mut self, checksum: u128) {
self.checksum = checksum;
}
fn operation(&self) -> Operation {
Operation::Reserved
}
fn command(&self) -> Command {
self.command
}
fn size(&self) -> u32 {
self.size
}
fn validate(&self) -> Result<(), ConsensusError> {
if self.command != Command::ForwardRegisterResult {
return Err(ConsensusError::InvalidCommand {
expected: Command::ForwardRegisterResult,
found: self.command,
});
}
validate_forward_register_frame(self.size, self.client, self.nonce, &self.reserved)?;
if self.outcome != ForwardRegisterOutcome::Ok && (self.epoch != 0 || self.watermark != 0) {
return Err(ConsensusError::InvalidField(
"forward register result bind must be zero on failure".to_string(),
));
}
Ok(())
}
}
fn validate_forward_register_frame(
size: u32,
client: u128,
nonce: u128,
reserved: &[u8],
) -> Result<(), ConsensusError> {
validate_forward_frame(size, client, nonce, reserved)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
#[repr(C)]
pub struct ForwardLogoutHeader {
pub checksum: u128,
pub checksum_body: u128,
pub cluster: u128,
pub size: u32,
pub view: u32,
pub release: u32,
pub command: Command,
pub replica: u8,
pub reserved_frame: [u8; 66],
pub client: u128,
pub nonce: u128,
pub session: u64,
pub request: u64,
pub reserved: [u8; 80],
}
const _: () = {
assert!(size_of::<ForwardLogoutHeader>() == HEADER_SIZE);
assert!(
offset_of!(ForwardLogoutHeader, client)
== offset_of!(ForwardLogoutHeader, reserved_frame) + size_of::<[u8; 66]>()
);
assert!(offset_of!(ForwardLogoutHeader, nonce) == 144);
assert!(offset_of!(ForwardLogoutHeader, session) == 160);
assert!(offset_of!(ForwardLogoutHeader, request) == 168);
assert!(offset_of!(ForwardLogoutHeader, reserved) + size_of::<[u8; 80]>() == HEADER_SIZE);
};
impl ConsensusHeader for ForwardLogoutHeader {
const FRAME_SEALED: bool = true;
const COMMAND: Command = Command::ForwardLogout;
fn checksum(&self) -> u128 {
self.checksum
}
fn set_checksum(&mut self, checksum: u128) {
self.checksum = checksum;
}
fn operation(&self) -> Operation {
Operation::Reserved
}
fn command(&self) -> Command {
self.command
}
fn size(&self) -> u32 {
self.size
}
fn validate(&self) -> Result<(), ConsensusError> {
if self.command != Command::ForwardLogout {
return Err(ConsensusError::InvalidCommand {
expected: Command::ForwardLogout,
found: self.command,
});
}
validate_forward_logout_frame(
self.size,
self.client,
self.nonce,
self.session,
self.request,
&self.reserved,
)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
#[repr(C)]
pub struct ForwardLogoutResultHeader {
pub checksum: u128,
pub checksum_body: u128,
pub cluster: u128,
pub size: u32,
pub view: u32,
pub release: u32,
pub command: Command,
pub replica: u8,
pub reserved_frame: [u8; 66],
pub nonce: u128,
pub client: u128,
pub commit: u64,
pub reserved: [u8; 87],
pub outcome: ForwardLogoutOutcome,
}
const _: () = {
assert!(size_of::<ForwardLogoutResultHeader>() == HEADER_SIZE);
assert!(
offset_of!(ForwardLogoutResultHeader, nonce)
== offset_of!(ForwardLogoutResultHeader, reserved_frame) + size_of::<[u8; 66]>()
);
assert!(offset_of!(ForwardLogoutResultHeader, client) == 144);
assert!(offset_of!(ForwardLogoutResultHeader, commit) == 160);
assert!(
offset_of!(ForwardLogoutResultHeader, outcome) + size_of::<ForwardLogoutOutcome>()
== HEADER_SIZE
);
};
#[derive(Debug, Clone, Copy, PartialEq, Eq, NoUninit, CheckedBitPattern)]
#[repr(u8)]
pub enum ForwardLogoutOutcome {
Ok = 0,
NotPrimary = 1,
PipelineFull = 2,
InProgress = 3,
Canceled = 4,
}
impl ConsensusHeader for ForwardLogoutResultHeader {
const FRAME_SEALED: bool = true;
const COMMAND: Command = Command::ForwardLogoutResult;
fn checksum(&self) -> u128 {
self.checksum
}
fn set_checksum(&mut self, checksum: u128) {
self.checksum = checksum;
}
fn operation(&self) -> Operation {
Operation::Reserved
}
fn command(&self) -> Command {
self.command
}
fn size(&self) -> u32 {
self.size
}
fn validate(&self) -> Result<(), ConsensusError> {
if self.command != Command::ForwardLogoutResult {
return Err(ConsensusError::InvalidCommand {
expected: Command::ForwardLogoutResult,
found: self.command,
});
}
validate_forward_frame(self.size, self.client, self.nonce, &self.reserved)?;
if self.outcome != ForwardLogoutOutcome::Ok && self.commit != 0 {
return Err(ConsensusError::InvalidField(
"forward logout result commit must be zero on failure".to_string(),
));
}
Ok(())
}
}
fn validate_forward_logout_frame(
size: u32,
client: u128,
nonce: u128,
session: u64,
request: u64,
reserved: &[u8],
) -> Result<(), ConsensusError> {
validate_forward_frame(size, client, nonce, reserved)?;
if session == 0 {
return Err(ConsensusError::InvalidField(
"forward logout session must be non-zero".to_string(),
));
}
if request == 0 {
return Err(ConsensusError::InvalidField(
"forward logout request must be non-zero".to_string(),
));
}
Ok(())
}
#[allow(clippy::cast_possible_truncation)]
fn validate_forward_frame(
size: u32,
client: u128,
nonce: u128,
reserved: &[u8],
) -> Result<(), ConsensusError> {
if size as usize != HEADER_SIZE {
return Err(ConsensusError::InvalidSize {
expected: HEADER_SIZE as u32,
found: size,
});
}
if client == 0 {
return Err(ConsensusError::InvalidField(
"forward client must be non-zero".to_string(),
));
}
if nonce == 0 {
return Err(ConsensusError::InvalidField(
"forward nonce must be non-zero".to_string(),
));
}
if reserved.iter().any(|&byte| byte != 0) {
return Err(ConsensusError::InvalidField(
"forward reserved bytes must be zero".to_string(),
));
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::{
Command, CommitHeader, ConsensusError, ConsensusHeader, DoViewChangeHeader, EvictionHeader,
EvictionReason, ForwardLogoutHeader, ForwardLogoutOutcome, ForwardLogoutResultHeader,
ForwardRegisterHeader, ForwardRegisterOutcome, ForwardRegisterResultHeader, GenericHeader,
HEADER_SIZE, Operation, PrepareHeader, PrepareOkHeader, RepairPrepareHeader,
RepairRangeReplyHeader, ReplyHeader, RequestHeader, RequestPreparesHeader,
RequestStartViewHeader, RequestStateChunkHeader, RequestStateTransferHeader,
RoutedRequestHeader, StartViewChangeHeader, StartViewHeader, StateChunkHeader,
StateTransferTargetHeader,
};
use aligned_vec::{AVec, ConstAlign};
fn aligned_zeroed(size: usize) -> AVec<u8, ConstAlign<16>> {
let mut v: AVec<u8, ConstAlign<16>> = AVec::new(16);
v.resize(size, 0);
v
}
#[repr(C, align(16))]
struct AlignedFrame([u8; HEADER_SIZE]);
fn control_header<H: ConsensusHeader>() -> H {
const COMMAND_OFF: usize = std::mem::offset_of!(GenericHeader, command);
const SIZE_OFF: usize = std::mem::offset_of!(GenericHeader, size);
let frame_len = u32::try_from(HEADER_SIZE).expect("HEADER_SIZE fits u32");
let mut frame = AlignedFrame([0u8; HEADER_SIZE]);
frame.0[COMMAND_OFF] = H::COMMAND as u8;
frame.0[SIZE_OFF..SIZE_OFF + 4].copy_from_slice(&frame_len.to_le_bytes());
*bytemuck::checked::try_from_bytes::<H>(&frame.0).expect("a zeroed frame is a valid header")
}
fn tamper<H: ConsensusHeader>(mut header: H, offset: usize) -> Result<(), ConsensusError> {
header.seal();
let mut frame = AlignedFrame([0u8; HEADER_SIZE]);
frame.0.copy_from_slice(bytemuck::bytes_of(&header));
frame.0[offset] ^= 0x01;
let tampered = bytemuck::checked::try_from_bytes::<H>(&frame.0)
.expect("a single flipped bit stays a valid bit pattern here");
tampered.verify_frame()
}
#[test]
fn given_a_sealed_control_header_when_verifying_should_accept() {
macro_rules! seals {
($($header:ty),+ $(,)?) => {$({
assert!(
<$header>::FRAME_SEALED,
"{} is a replica-to-replica control header and must seal",
stringify!($header),
);
let mut header = control_header::<$header>();
header.seal();
assert_eq!(
header.verify_frame(),
Ok(()),
"{} must accept its own seal",
stringify!($header),
);
})+};
}
seals!(
PrepareOkHeader,
CommitHeader,
StartViewChangeHeader,
DoViewChangeHeader,
StartViewHeader,
RequestStartViewHeader,
RequestPreparesHeader,
RepairRangeReplyHeader,
RequestStateTransferHeader,
StateTransferTargetHeader,
RequestStateChunkHeader,
StateChunkHeader,
ForwardRegisterHeader,
ForwardRegisterResultHeader,
ForwardLogoutHeader,
ForwardLogoutResultHeader,
);
}
#[test]
fn given_any_covered_byte_when_flipped_should_reject() {
for offset in size_of::<u128>()..HEADER_SIZE {
let header = control_header::<DoViewChangeHeader>();
if offset == std::mem::offset_of!(DoViewChangeHeader, command) {
continue;
}
assert!(
matches!(
tamper(header, offset),
Err(ConsensusError::FrameChecksumMismatch { .. })
),
"byte {offset} is inside the seal and must be covered"
);
}
}
#[test]
fn given_an_unsealed_control_header_when_verifying_should_reject() {
let header = control_header::<DoViewChangeHeader>();
assert_eq!(header.checksum, 0);
assert!(matches!(
header.verify_frame(),
Err(ConsensusError::FrameChecksumMismatch { found: 0, .. })
));
}
#[test]
fn given_an_identity_or_client_header_when_verifying_should_abstain() {
const {
assert!(!PrepareHeader::FRAME_SEALED);
assert!(!RepairPrepareHeader::FRAME_SEALED);
assert!(!RequestHeader::FRAME_SEALED);
assert!(!ReplyHeader::FRAME_SEALED);
assert!(!EvictionHeader::FRAME_SEALED);
assert!(!GenericHeader::FRAME_SEALED);
}
let prepare = PrepareHeader {
command: Command::Prepare,
checksum: 0xdead_beef,
..Default::default()
};
assert_eq!(prepare.verify_frame(), Ok(()));
}
#[test]
fn all_headers_are_256_bytes() {
assert_eq!(size_of::<GenericHeader>(), 256);
assert_eq!(size_of::<RequestHeader>(), 256);
assert_eq!(size_of::<ReplyHeader>(), 256);
assert_eq!(size_of::<PrepareHeader>(), 256);
assert_eq!(size_of::<PrepareOkHeader>(), 256);
assert_eq!(size_of::<CommitHeader>(), 256);
assert_eq!(size_of::<StartViewChangeHeader>(), 256);
assert_eq!(size_of::<DoViewChangeHeader>(), 256);
assert_eq!(size_of::<StartViewHeader>(), 256);
}
#[test]
fn generic_header_zero_copy() {
let buf = aligned_zeroed(256);
let header: &GenericHeader = bytemuck::checked::try_from_bytes(&buf).unwrap();
assert_eq!(header.command, Command::Reserved);
assert_eq!(header.size, 0);
}
#[test]
fn request_header_zero_copy() {
let mut buf = aligned_zeroed(256);
buf[60] = Command::Request as u8;
buf[128] = 1;
buf[std::mem::offset_of!(RequestHeader, operation)] = Operation::Register as u8;
let header: &RequestHeader = bytemuck::checked::try_from_bytes(&buf).unwrap();
assert_eq!(header.command, Command::Request);
assert!(header.validate().is_ok());
}
#[test]
fn request_header_wrong_command_fails_validation() {
let buf = aligned_zeroed(256);
let header: &RequestHeader = bytemuck::checked::try_from_bytes(&buf).unwrap();
assert!(header.validate().is_err());
}
#[test]
fn request_reserved_operation_rejected() {
let header = RequestHeader {
command: Command::Request,
client: 1,
operation: Operation::Reserved,
session: 1,
request: 1,
..RequestHeader::default()
};
assert!(header.validate().is_err());
}
#[test]
fn request_register_nonzero_session_rejected() {
let header = RequestHeader {
command: Command::Request,
operation: Operation::Register,
session: 5,
request: 0,
..RequestHeader::default()
};
assert!(header.validate().is_err());
}
#[test]
fn request_register_nonzero_request_rejected() {
let header = RequestHeader {
command: Command::Request,
operation: Operation::Register,
session: 0,
request: 1,
..RequestHeader::default()
};
assert!(header.validate().is_err());
}
#[test]
fn request_non_register_valid() {
let header = RequestHeader {
command: Command::Request,
operation: Operation::SendMessages,
client: 0xCAFE,
session: 10,
request: 1,
..RequestHeader::default()
};
assert!(header.validate().is_ok());
}
#[test]
fn request_non_register_zero_session_rejected() {
let header = RequestHeader {
command: Command::Request,
operation: Operation::SendMessages,
session: 0,
request: 1,
..RequestHeader::default()
};
assert!(header.validate().is_err());
}
#[test]
fn request_non_register_zero_request_rejected() {
let header = RequestHeader {
command: Command::Request,
operation: Operation::SendMessages,
session: 10,
request: 0,
..RequestHeader::default()
};
assert!(header.validate().is_err());
}
#[test]
fn reply_header_zero_copy() {
let mut buf = aligned_zeroed(256);
buf[60] = Command::Reply as u8;
let header: &ReplyHeader = bytemuck::checked::try_from_bytes(&buf).unwrap();
assert_eq!(header.command, Command::Reply);
assert!(header.validate().is_ok());
}
#[test]
fn reply_header_status_offset_and_size_pinned() {
use std::mem::offset_of;
assert_eq!(size_of::<ReplyHeader>(), HEADER_SIZE);
assert_eq!(offset_of!(ReplyHeader, status), 216);
assert_eq!(
offset_of!(ReplyHeader, reserved) + size_of::<[u8; 36]>(),
HEADER_SIZE
);
}
#[test]
fn routed_request_group_claims_reserved_tail() {
use std::mem::offset_of;
assert_eq!(
offset_of!(RoutedRequestHeader, reserved),
offset_of!(RequestHeader, reserved)
);
assert_eq!(
offset_of!(RoutedRequestHeader, group),
offset_of!(RequestHeader, reserved) + 52
);
assert_eq!(offset_of!(RoutedRequestHeader, group), 248);
assert_eq!(
offset_of!(RoutedRequestHeader, group) + size_of::<u64>(),
HEADER_SIZE
);
}
#[test]
fn routed_request_zero_client_rejected() {
let header = RoutedRequestHeader {
command: Command::Request,
operation: Operation::SendMessages,
session: 10,
request: 1,
..RoutedRequestHeader::default()
};
assert!(header.validate().is_err());
}
#[test]
fn routed_request_reserved_operation_rejected() {
let header = RoutedRequestHeader {
command: Command::Request,
client: 0xCAFE,
session: 10,
request: 1,
..RoutedRequestHeader::default()
};
assert!(header.validate().is_err());
}
#[test]
fn routed_request_default_fails_validate() {
assert!(RoutedRequestHeader::default().validate().is_err());
}
#[test]
fn routed_request_valid_fields_accepted() {
let header = RoutedRequestHeader {
command: Command::Request,
operation: Operation::SendMessages,
client: 0xCAFE,
session: 10,
request: 1,
group: 7,
..RoutedRequestHeader::default()
};
assert!(header.validate().is_ok());
}
#[test]
fn reply_header_nonzero_status_still_validates() {
let header = ReplyHeader {
command: Command::Reply,
status: 41,
..ReplyHeader::default()
};
assert!(header.validate().is_ok());
}
#[test]
fn eviction_reason_discriminants_pinned() {
assert_eq!(EvictionReason::Reserved as u8, 0);
assert_eq!(EvictionReason::NoSession as u8, 1);
assert_eq!(EvictionReason::ClientReleaseTooLow as u8, 2);
assert_eq!(EvictionReason::ClientReleaseTooHigh as u8, 3);
assert_eq!(EvictionReason::InvalidRequestOperation as u8, 4);
assert_eq!(EvictionReason::InvalidRequestBody as u8, 5);
assert_eq!(EvictionReason::InvalidRequestBodySize as u8, 6);
assert_eq!(EvictionReason::SessionTooLow as u8, 7);
assert_eq!(EvictionReason::SessionReleaseMismatch as u8, 8);
assert_eq!(EvictionReason::InvalidCredentials as u8, 9);
assert_eq!(EvictionReason::InvalidToken as u8, 10);
assert_eq!(EvictionReason::UserInactive as u8, 11);
assert_eq!(EvictionReason::SessionError as u8, 12);
assert_eq!(EvictionReason::StaleClient as u8, 13);
assert_eq!(EvictionReason::IncompatibleProtocol as u8, 14);
assert_eq!(EvictionReason::MalformedLogin as u8, 15);
}
#[test]
fn eviction_incompatible_protocol_accepts_valid_window() {
let header = EvictionHeader::incompatible_protocol(0, 0, 0, 0xCAFE, 2, 1);
assert!(header.validate().is_ok());
assert_eq!(header.reason, EvictionReason::IncompatibleProtocol);
assert_eq!(header.server_protocol_version, 2);
assert_eq!(header.server_protocol_version_min, 1);
}
#[test]
fn eviction_incompatible_protocol_rejects_inverted_window() {
let header = EvictionHeader::incompatible_protocol(0, 0, 0, 0xCAFE, 1, 2);
assert!(header.validate().is_err());
}
#[test]
fn eviction_incompatible_protocol_rejects_zero_min() {
let header = EvictionHeader::incompatible_protocol(0, 0, 0, 0xCAFE, 1, 0);
assert!(header.validate().is_err());
}
#[test]
fn eviction_validate_rejects_window_on_other_reason() {
let mut header = EvictionHeader::new(0, 0, 0, 0xCAFE, EvictionReason::NoSession);
header.server_protocol_version = 1;
assert!(header.validate().is_err());
}
#[test]
fn eviction_validate_rejects_zero_client() {
let header = EvictionHeader::new(0, 0, 0, 0, EvictionReason::NoSession);
assert!(header.validate().is_err());
}
#[test]
fn eviction_validate_rejects_reserved_reason() {
let header = EvictionHeader::new(0, 0, 0, 1, EvictionReason::Reserved);
assert!(header.validate().is_err());
}
#[test]
fn eviction_validate_rejects_nonzero_reserved() {
let mut header = EvictionHeader::new(0, 0, 0, 1, EvictionReason::NoSession);
header.reserved[0] = 1;
assert!(header.validate().is_err());
}
#[test]
fn eviction_validate_accepts_well_formed_frame() {
let header = EvictionHeader::new(0, 0, 0, 0xCAFE, EvictionReason::NoSession);
assert!(header.validate().is_ok());
}
#[test]
fn eviction_header_is_256_bytes() {
assert_eq!(size_of::<EvictionHeader>(), 256);
}
fn forward_register() -> ForwardRegisterHeader {
let mut header = ForwardRegisterHeader {
checksum: 0,
checksum_body: 0,
cluster: 7,
size: u32::try_from(HEADER_SIZE).expect("HEADER_SIZE fits u32"),
view: 3,
release: 0,
command: Command::ForwardRegister,
replica: 2,
reserved_frame: [0; 66],
client: 0xCAFE,
nonce: 0xF00D,
user_id: 41,
reserved: [0; 92],
};
header.seal();
header
}
fn forward_register_result(outcome: ForwardRegisterOutcome) -> ForwardRegisterResultHeader {
let mut header = ForwardRegisterResultHeader {
checksum: 0,
checksum_body: 0,
cluster: 7,
size: u32::try_from(HEADER_SIZE).expect("HEADER_SIZE fits u32"),
view: 3,
release: 0,
command: Command::ForwardRegisterResult,
replica: 0,
reserved_frame: [0; 66],
nonce: 0xF00D,
client: 0xCAFE,
epoch: u64::from(outcome == ForwardRegisterOutcome::Ok) * 91,
watermark: u64::from(outcome == ForwardRegisterOutcome::Ok) * 12,
reserved: [0; 79],
outcome,
};
header.seal();
header
}
#[test]
fn forward_register_round_trips_through_generic_bytes() {
for (command, bytes) in [
(
Command::ForwardRegister,
bytemuck::bytes_of(&forward_register()).to_vec(),
),
(
Command::ForwardRegisterResult,
bytemuck::bytes_of(&forward_register_result(ForwardRegisterOutcome::Ok)).to_vec(),
),
] {
let mut buf = aligned_zeroed(HEADER_SIZE);
buf.copy_from_slice(&bytes);
let generic = bytemuck::checked::try_from_bytes::<GenericHeader>(&buf)
.expect("a forward-register frame is a valid generic header");
assert_eq!(generic.command, command);
assert_eq!(generic.size as usize, HEADER_SIZE);
}
let buf = {
let mut buf = aligned_zeroed(HEADER_SIZE);
buf.copy_from_slice(bytemuck::bytes_of(&forward_register()));
buf
};
let typed = bytemuck::checked::try_from_bytes::<ForwardRegisterHeader>(&buf)
.expect("round-trips into its own type");
assert_eq!(typed.verify_frame(), Ok(()));
assert_eq!(typed.validate(), Ok(()));
assert_eq!(typed.user_id, 41);
assert_eq!(typed.nonce, 0xF00D);
let buf = {
let mut buf = aligned_zeroed(HEADER_SIZE);
buf.copy_from_slice(bytemuck::bytes_of(&forward_register_result(
ForwardRegisterOutcome::ClientIdOwnedByAnotherUser,
)));
buf
};
let typed = bytemuck::checked::try_from_bytes::<ForwardRegisterResultHeader>(&buf)
.expect("round-trips into its own type");
assert_eq!(typed.verify_frame(), Ok(()));
assert_eq!(typed.validate(), Ok(()));
assert_eq!(
typed.outcome,
ForwardRegisterOutcome::ClientIdOwnedByAnotherUser
);
}
#[test]
fn forward_register_tampered_user_id_fails_the_seal() {
const USER_ID_OFFSET: usize = std::mem::offset_of!(ForwardRegisterHeader, user_id);
assert!(matches!(
tamper(forward_register(), USER_ID_OFFSET),
Err(ConsensusError::FrameChecksumMismatch { .. })
));
}
#[test]
fn forward_register_validate_rejects_malformed_frames() {
let mut zero_client = forward_register();
zero_client.client = 0;
assert!(matches!(
zero_client.validate(),
Err(ConsensusError::InvalidField(_))
));
let mut zero_nonce = forward_register();
zero_nonce.nonce = 0;
assert!(matches!(
zero_nonce.validate(),
Err(ConsensusError::InvalidField(_))
));
let mut dirty_reserved = forward_register();
dirty_reserved.reserved[0] = 1;
assert!(matches!(
dirty_reserved.validate(),
Err(ConsensusError::InvalidField(_))
));
let mut wrong_size = forward_register();
wrong_size.size = 512;
assert!(matches!(
wrong_size.validate(),
Err(ConsensusError::InvalidSize { .. })
));
let mut result_zero_nonce = forward_register_result(ForwardRegisterOutcome::Ok);
result_zero_nonce.nonce = 0;
assert!(matches!(
result_zero_nonce.validate(),
Err(ConsensusError::InvalidField(_))
));
let mut failed_with_bind = forward_register_result(ForwardRegisterOutcome::NotPrimary);
failed_with_bind.epoch = 91;
failed_with_bind.watermark = 12;
assert!(matches!(
failed_with_bind.validate(),
Err(ConsensusError::InvalidField(_))
));
failed_with_bind.epoch = 0;
failed_with_bind.watermark = 0;
assert_eq!(failed_with_bind.validate(), Ok(()));
}
#[test]
fn forward_register_result_rejects_unknown_outcome() {
const OUTCOME_OFFSET: usize = std::mem::offset_of!(ForwardRegisterResultHeader, outcome);
let mut buf = aligned_zeroed(HEADER_SIZE);
buf.copy_from_slice(bytemuck::bytes_of(&forward_register_result(
ForwardRegisterOutcome::Ok,
)));
buf[OUTCOME_OFFSET] = 7;
assert!(bytemuck::checked::try_from_bytes::<ForwardRegisterResultHeader>(&buf).is_err());
}
#[test]
fn forward_register_outcome_discriminants_pinned() {
assert_eq!(ForwardRegisterOutcome::Ok as u8, 0);
assert_eq!(ForwardRegisterOutcome::NotPrimary as u8, 1);
assert_eq!(ForwardRegisterOutcome::NotCaughtUp as u8, 2);
assert_eq!(ForwardRegisterOutcome::PipelineFull as u8, 3);
assert_eq!(ForwardRegisterOutcome::InProgress as u8, 4);
assert_eq!(ForwardRegisterOutcome::Canceled as u8, 5);
assert_eq!(ForwardRegisterOutcome::ClientIdOwnedByAnotherUser as u8, 6);
}
fn forward_logout() -> ForwardLogoutHeader {
let mut header = ForwardLogoutHeader {
checksum: 0,
checksum_body: 0,
cluster: 7,
size: u32::try_from(HEADER_SIZE).expect("HEADER_SIZE fits u32"),
view: 3,
release: 0,
command: Command::ForwardLogout,
replica: 2,
reserved_frame: [0; 66],
client: 0xCAFE,
nonce: 0xF00D,
session: 91,
request: 12,
reserved: [0; 80],
};
header.seal();
header
}
fn forward_logout_result(outcome: ForwardLogoutOutcome) -> ForwardLogoutResultHeader {
let mut header = ForwardLogoutResultHeader {
checksum: 0,
checksum_body: 0,
cluster: 7,
size: u32::try_from(HEADER_SIZE).expect("HEADER_SIZE fits u32"),
view: 3,
release: 0,
command: Command::ForwardLogoutResult,
replica: 0,
reserved_frame: [0; 66],
nonce: 0xF00D,
client: 0xCAFE,
commit: if outcome == ForwardLogoutOutcome::Ok {
92
} else {
0
},
reserved: [0; 87],
outcome,
};
header.seal();
header
}
#[test]
fn forward_logout_headers_round_trip_and_validate() {
let forward = forward_logout();
assert_eq!(forward.verify_frame(), Ok(()));
assert_eq!(forward.validate(), Ok(()));
for outcome in [
ForwardLogoutOutcome::Ok,
ForwardLogoutOutcome::NotPrimary,
ForwardLogoutOutcome::PipelineFull,
ForwardLogoutOutcome::InProgress,
ForwardLogoutOutcome::Canceled,
] {
let result = forward_logout_result(outcome);
assert_eq!(result.verify_frame(), Ok(()));
assert_eq!(result.validate(), Ok(()));
let mut bytes = aligned_zeroed(HEADER_SIZE);
bytes.copy_from_slice(bytemuck::bytes_of(&result));
let generic = bytemuck::checked::try_from_bytes::<GenericHeader>(&bytes)
.expect("forward logout result is a valid generic header");
assert_eq!(generic.command, Command::ForwardLogoutResult);
}
}
#[test]
fn forward_logout_validate_rejects_malformed_frames() {
let mut header = forward_logout();
header.client = 0;
assert!(matches!(
header.validate(),
Err(ConsensusError::InvalidField(_))
));
header = forward_logout();
header.nonce = 0;
assert!(matches!(
header.validate(),
Err(ConsensusError::InvalidField(_))
));
header = forward_logout();
header.session = 0;
assert!(matches!(
header.validate(),
Err(ConsensusError::InvalidField(_))
));
header = forward_logout();
header.request = 0;
assert!(matches!(
header.validate(),
Err(ConsensusError::InvalidField(_))
));
header = forward_logout();
header.reserved[0] = 1;
assert!(matches!(
header.validate(),
Err(ConsensusError::InvalidField(_))
));
let mut result = forward_logout_result(ForwardLogoutOutcome::NotPrimary);
result.commit = 92;
assert!(matches!(
result.validate(),
Err(ConsensusError::InvalidField(_))
));
}
#[test]
fn forward_logout_result_rejects_unknown_outcome() {
const OUTCOME_OFFSET: usize = std::mem::offset_of!(ForwardLogoutResultHeader, outcome);
let mut bytes = aligned_zeroed(HEADER_SIZE);
bytes.copy_from_slice(bytemuck::bytes_of(&forward_logout_result(
ForwardLogoutOutcome::Ok,
)));
bytes[OUTCOME_OFFSET] = 5;
assert!(bytemuck::checked::try_from_bytes::<ForwardLogoutResultHeader>(&bytes).is_err());
}
#[test]
fn forward_logout_outcome_discriminants_pinned() {
assert_eq!(ForwardLogoutOutcome::Ok as u8, 0);
assert_eq!(ForwardLogoutOutcome::NotPrimary as u8, 1);
assert_eq!(ForwardLogoutOutcome::PipelineFull as u8, 2);
assert_eq!(ForwardLogoutOutcome::InProgress as u8, 3);
assert_eq!(ForwardLogoutOutcome::Canceled as u8, 4);
}
}