use super::*;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ClusterOwnershipDecision {
pub key: String,
pub owner: Option<ClusterMember>,
pub member_count: usize,
pub resolver: &'static str,
}
impl ClusterOwnershipDecision {
pub fn has_owner(&self) -> bool {
self.owner.is_some()
}
pub fn owner_node_id(&self) -> Option<&ClusterNodeId> {
self.owner.as_ref().map(|owner| &owner.node_id)
}
pub fn owner_generation(&self) -> Option<ClusterGeneration> {
self.owner.as_ref().map(|owner| owner.generation)
}
pub fn peer_fetch_request(&self) -> Option<ClusterPeerFetchRequest> {
self.owner.as_ref().map(|owner| {
ClusterPeerFetchRequest::new(owner.node_id.clone(), self.key.clone())
.generation(owner.generation)
})
}
}
pub trait ClusterOwnershipResolver: Send + Sync {
fn name(&self) -> &'static str;
fn resolve_owner(&self, key: &str, participants: &[ClusterMember]) -> ClusterOwnershipDecision;
}
#[derive(Debug, Clone, Copy, Default)]
pub struct RendezvousClusterOwnership;
impl ClusterOwnershipResolver for RendezvousClusterOwnership {
fn name(&self) -> &'static str {
"rendezvous"
}
fn resolve_owner(&self, key: &str, participants: &[ClusterMember]) -> ClusterOwnershipDecision {
let mut member_count = 0_usize;
let mut best: Option<(u64, ClusterMember)> = None;
for participant in participants
.iter()
.filter(|candidate| candidate.is_member())
{
member_count = member_count.saturating_add(1);
let score = rendezvous_score(key, &participant.node_id);
let replace = best
.as_ref()
.map(|(best_score, best_member)| {
score > *best_score
|| (score == *best_score && participant.node_id > best_member.node_id)
})
.unwrap_or(true);
if replace {
best = Some((score, participant.clone()));
}
}
ClusterOwnershipDecision {
key: key.to_owned(),
owner: best.map(|(_, member)| member),
member_count,
resolver: self.name(),
}
}
}
fn rendezvous_score(key: &str, node_id: &ClusterNodeId) -> u64 {
const FNV_OFFSET: u64 = 0xcbf29ce484222325;
const FNV_PRIME: u64 = 0x100000001b3;
let mut hash = FNV_OFFSET;
for byte in key.bytes().chain([0xff]).chain(node_id.as_str().bytes()) {
hash ^= u64::from(byte);
hash = hash.wrapping_mul(FNV_PRIME);
}
hash
}
#[derive(
Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize,
)]
pub struct PartitionId(u32);
impl PartitionId {
pub const fn new(value: u32) -> Self {
Self(value)
}
pub const fn value(self) -> u32 {
self.0
}
}
pub fn partition_for_key(key: impl AsRef<str>, partition_count: u32) -> PartitionId {
let partition_count = partition_count.max(1);
PartitionId((stable_key_hash(key.as_ref()) % u64::from(partition_count)) as u32)
}
fn stable_key_hash(key: &str) -> u64 {
const FNV_OFFSET: u64 = 0xcbf29ce484222325;
const FNV_PRIME: u64 = 0x100000001b3;
let mut hash = FNV_OFFSET;
for byte in key.bytes() {
hash ^= u64::from(byte);
hash = hash.wrapping_mul(FNV_PRIME);
}
hash
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ClusterReplicaConfigError {
MinReplicaZero,
QuorumZero,
QuorumExceedsReplication,
}
impl fmt::Display for ClusterReplicaConfigError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
let message = match self {
Self::MinReplicaZero => "min_replica must be at least 1",
Self::QuorumZero => "quorum must be at least 1",
Self::QuorumExceedsReplication => "quorum cannot exceed replication_factor",
};
formatter.write_str(message)
}
}
impl std::error::Error for ClusterReplicaConfigError {}
pub fn validate_replica_config(
min_replica: usize,
replication_factor: usize,
quorum: usize,
) -> std::result::Result<(), ClusterReplicaConfigError> {
if min_replica < 1 {
return Err(ClusterReplicaConfigError::MinReplicaZero);
}
if quorum == 0 {
return Err(ClusterReplicaConfigError::QuorumZero);
}
if quorum > replication_factor {
return Err(ClusterReplicaConfigError::QuorumExceedsReplication);
}
Ok(())
}