use std::collections::{BTreeMap, BTreeSet};
use std::fmt;
use std::time::Duration;
use serde::{Deserialize, Serialize};
use crate::grid::elasticity::RegionId;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PersistenceDurability {
Sync,
AsyncBounded {
max_lag: usize,
},
}
impl Default for PersistenceDurability {
fn default() -> Self {
Self::AsyncBounded { max_lag: 1024 }
}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PersistenceEviction {
#[default]
None,
Lru,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PersistenceInMemoryFormat {
Object,
#[default]
Binary,
}
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub enum RegionSelector {
#[default]
All,
Only(BTreeSet<RegionId>),
HomeRegionOnly,
}
impl RegionSelector {
pub fn only(regions: impl IntoIterator<Item = RegionId>) -> Self {
Self::Only(regions.into_iter().collect())
}
pub fn selects(&self, local_region: &RegionId, placement: &PersistenceRegionPlacement) -> bool {
match self {
Self::All => placement.contains(local_region),
Self::Only(regions) => regions.contains(local_region),
Self::HomeRegionOnly => placement.home() == local_region,
}
}
pub fn validate_placement(
&self,
pattern: &PersistenceMatcher,
placement: &PersistenceRegionPlacement,
) -> Result<(), PersistencePolicyError> {
let missing = match self {
Self::All | Self::HomeRegionOnly => Vec::new(),
Self::Only(regions) => regions
.iter()
.filter(|region| !placement.contains(region))
.cloned()
.collect::<Vec<_>>(),
};
if missing.is_empty() {
return Ok(());
}
let missing = missing
.iter()
.map(RegionId::as_str)
.collect::<Vec<_>>()
.join(", ");
Err(PersistencePolicyError::new(format!(
"persistence rule '{pattern}' selects region(s) outside placement: {missing}"
)))
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PersistenceRegionPlacement {
home: RegionId,
replicated_regions: BTreeSet<RegionId>,
}
impl PersistenceRegionPlacement {
pub fn new(
home: impl Into<RegionId>,
replicated_regions: impl IntoIterator<Item = RegionId>,
) -> Self {
let home = home.into();
let mut replicated_regions = replicated_regions.into_iter().collect::<BTreeSet<_>>();
replicated_regions.insert(home.clone());
Self {
home,
replicated_regions,
}
}
pub fn home_region_only(home: impl Into<RegionId>) -> Self {
Self::new(home, [])
}
pub fn active_active(
home: impl Into<RegionId>,
peers: impl IntoIterator<Item = RegionId>,
) -> Self {
Self::new(home, peers)
}
pub fn home(&self) -> &RegionId {
&self.home
}
pub fn contains(&self, region: &RegionId) -> bool {
self.replicated_regions.contains(region)
}
pub fn regions(&self) -> &BTreeSet<RegionId> {
&self.replicated_regions
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct NamespacePersistenceSettings {
pub persist: bool,
pub durability: PersistenceDurability,
pub snapshot_interval: Option<Duration>,
pub eviction: PersistenceEviction,
pub backup_count: usize,
pub in_memory_format: PersistenceInMemoryFormat,
pub persist_in_regions: RegionSelector,
}
impl NamespacePersistenceSettings {
pub fn ram_only() -> Self {
Self {
persist: false,
durability: PersistenceDurability::default(),
snapshot_interval: None,
eviction: PersistenceEviction::None,
backup_count: 0,
in_memory_format: PersistenceInMemoryFormat::Binary,
persist_in_regions: RegionSelector::All,
}
}
pub fn persistent() -> Self {
Self {
persist: true,
durability: PersistenceDurability::default(),
snapshot_interval: None,
eviction: PersistenceEviction::None,
backup_count: 0,
in_memory_format: PersistenceInMemoryFormat::Binary,
persist_in_regions: RegionSelector::All,
}
}
pub fn with_durability(mut self, durability: PersistenceDurability) -> Self {
self.durability = durability;
self
}
pub fn with_snapshot_interval(mut self, interval: Duration) -> Self {
self.snapshot_interval = Some(interval);
self
}
pub fn with_region_selector(mut self, selector: RegionSelector) -> Self {
self.persist_in_regions = selector;
self
}
}
impl Default for NamespacePersistenceSettings {
fn default() -> Self {
Self::ram_only()
}
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub enum PersistenceMatcher {
Exact(String),
Prefix(String),
Default,
}
impl PersistenceMatcher {
pub fn parse(pattern: impl Into<String>) -> Result<Self, PersistencePolicyError> {
let pattern = pattern.into();
if pattern == "default" {
return Ok(Self::Default);
}
if pattern.is_empty() {
return Err(PersistencePolicyError::new(
"persistence pattern must not be empty",
));
}
if let Some(prefix) = pattern.strip_suffix(".*") {
if prefix.is_empty() {
return Err(PersistencePolicyError::new(
"wildcard persistence pattern must have a prefix",
));
}
return Ok(Self::Prefix(format!("{prefix}.")));
}
if pattern.contains('*') {
return Err(PersistencePolicyError::new(format!(
"unsupported persistence wildcard pattern '{pattern}'"
)));
}
Ok(Self::Exact(pattern))
}
fn matches(&self, namespace: &str) -> bool {
match self {
Self::Exact(exact) => exact == namespace,
Self::Prefix(prefix) => namespace.starts_with(prefix),
Self::Default => true,
}
}
fn precedence_key(&self) -> (u8, usize) {
match self {
Self::Exact(exact) => (3, exact.len()),
Self::Prefix(prefix) => (2, prefix.len()),
Self::Default => (1, 0),
}
}
}
impl fmt::Display for PersistenceMatcher {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Exact(exact) => formatter.write_str(exact),
Self::Prefix(prefix) => write!(formatter, "{prefix}*"),
Self::Default => formatter.write_str("default"),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct NamespacePersistenceRule {
pub matcher: PersistenceMatcher,
pub settings: NamespacePersistenceSettings,
}
impl NamespacePersistenceRule {
pub fn new(
pattern: impl Into<String>,
settings: NamespacePersistenceSettings,
) -> Result<Self, PersistencePolicyError> {
Ok(Self {
matcher: PersistenceMatcher::parse(pattern)?,
settings,
})
}
pub fn persistent(pattern: impl Into<String>) -> Result<Self, PersistencePolicyError> {
Self::new(pattern, NamespacePersistenceSettings::persistent())
}
pub fn ram_only(pattern: impl Into<String>) -> Result<Self, PersistencePolicyError> {
Self::new(pattern, NamespacePersistenceSettings::ram_only())
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ResolvedPersistence {
pub namespace: String,
pub matched_by: Option<PersistenceMatcher>,
pub settings: NamespacePersistenceSettings,
}
impl ResolvedPersistence {
pub fn persists(&self) -> bool {
self.settings.persist
}
pub fn persists_in_region(
&self,
local_region: &RegionId,
placement: &PersistenceRegionPlacement,
) -> bool {
self.settings.persist
&& self
.settings
.persist_in_regions
.selects(local_region, placement)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PersistencePolicy {
rules: Vec<NamespacePersistenceRule>,
persistable_namespace_roster: BTreeSet<String>,
}
impl PersistencePolicy {
pub fn try_new(
rules: impl IntoIterator<Item = NamespacePersistenceRule>,
) -> Result<Self, PersistencePolicyError> {
let mut by_matcher = BTreeMap::<PersistenceMatcher, NamespacePersistenceSettings>::new();
let mut retained = Vec::new();
for rule in rules {
match by_matcher.get(&rule.matcher) {
Some(existing) if existing != &rule.settings => {
return Err(PersistencePolicyError::new(format!(
"conflicting persistence rules for pattern '{}'",
rule.matcher
)));
}
Some(_) => continue,
None => {
by_matcher.insert(rule.matcher.clone(), rule.settings.clone());
retained.push(rule);
}
}
}
Ok(Self {
persistable_namespace_roster: retained
.iter()
.filter_map(|rule| match (&rule.matcher, rule.settings.persist) {
(PersistenceMatcher::Exact(namespace), true) => Some(namespace.clone()),
_ => None,
})
.collect(),
rules: retained,
})
}
pub fn ram_only() -> Self {
Self {
rules: Vec::new(),
persistable_namespace_roster: BTreeSet::new(),
}
}
pub fn resolve(&self, namespace: &str) -> ResolvedPersistence {
let mut best: Option<&NamespacePersistenceRule> = None;
for rule in &self.rules {
if !rule.matcher.matches(namespace) {
continue;
}
let replace = best
.map(|current| rule.matcher.precedence_key() > current.matcher.precedence_key())
.unwrap_or(true);
if replace {
best = Some(rule);
}
}
match best {
Some(rule) => ResolvedPersistence {
namespace: namespace.to_owned(),
matched_by: Some(rule.matcher.clone()),
settings: rule.settings.clone(),
},
None => ResolvedPersistence {
namespace: namespace.to_owned(),
matched_by: None,
settings: NamespacePersistenceSettings::ram_only(),
},
}
}
pub fn resolve_for_region(
&self,
namespace: &str,
local_region: &RegionId,
placement: &PersistenceRegionPlacement,
) -> Result<ResolvedPersistence, PersistencePolicyError> {
self.validate_namespace_placement(namespace, placement)?;
let mut resolved = self.resolve(namespace);
if !resolved.persists_in_region(local_region, placement) {
resolved.settings.persist = false;
}
Ok(resolved)
}
pub fn validate_namespace_placement(
&self,
namespace: &str,
placement: &PersistenceRegionPlacement,
) -> Result<(), PersistencePolicyError> {
for rule in self
.rules
.iter()
.filter(|rule| rule.settings.persist && rule.matcher.matches(namespace))
{
rule.settings
.persist_in_regions
.validate_placement(&rule.matcher, placement)?;
}
Ok(())
}
pub fn persistable_namespace_roster(&self) -> &BTreeSet<String> {
&self.persistable_namespace_roster
}
pub fn validate_engine_available(
&self,
durable_engine_available: bool,
) -> Result<(), PersistencePolicyError> {
if durable_engine_available {
return Ok(());
}
let missing = self
.rules
.iter()
.filter(|rule| rule.settings.persist)
.map(|rule| rule.matcher.to_string())
.collect::<Vec<_>>();
if missing.is_empty() {
Ok(())
} else {
Err(PersistencePolicyError::new(format!(
"persistence requested for {} but no durable value store is available",
missing.join(", ")
)))
}
}
}
impl Default for PersistencePolicy {
fn default() -> Self {
Self::ram_only()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PersistencePolicyError {
message: String,
}
impl PersistencePolicyError {
pub fn new(message: impl Into<String>) -> Self {
Self {
message: message.into(),
}
}
}
impl fmt::Display for PersistencePolicyError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(&self.message)
}
}
impl std::error::Error for PersistencePolicyError {}