use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, PoisonError, RwLock};
use chrono::{DateTime, Utc};
use serde::ser::SerializeStruct;
use serde::{Deserialize, Serialize, Serializer};
use crate::attribute::{Attribute, AttributeView};
use crate::db_log::log_attribute;
pub const ALL_NAMESPACE: &str = "_all";
pub const SHADOW_PREFIX: &str = "_shadow/";
pub const CONFIG_PREFIX: &str = "_config/";
pub const APIKEYS_NAMESPACE: &str = "_config/acl/apikeys/";
pub const DEFAULT_APIKEY: &str = "changeme";
pub const SNAPSHOT_VERSION: u32 = 1;
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct NotFound {
pub error: &'static str,
pub namespace: String,
pub value: String,
}
impl NotFound {
pub fn namespace(namespace: &str, value: &str) -> Self {
Self {
error: "Path not found",
namespace: namespace.to_string(),
value: value.to_string(),
}
}
pub fn value(namespace: &str, value: &str) -> Self {
Self {
error: "Value not found",
namespace: namespace.to_string(),
value: value.to_string(),
}
}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct DatabasePolicy {
pub stats_retention: usize,
pub shadow_ttl: u64,
}
#[derive(Debug, Clone, Copy, Default)]
pub struct WriteOpts {
pub consensus: bool,
pub ttl: Option<u64>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct Page<T> {
pub items: Vec<T>,
pub total: usize,
pub offset: usize,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct SweepReport {
pub values_removed: usize,
pub namespaces_removed: usize,
}
impl SweepReport {
pub fn is_empty(&self) -> bool {
self.values_removed == 0 && self.namespaces_removed == 0
}
}
#[derive(Default)]
struct Namespace {
values: RwLock<HashMap<String, Mutex<Attribute>>>,
has_ttl: AtomicBool,
}
impl Namespace {
fn from_values(values: HashMap<String, Attribute>) -> Self {
let has_ttl = values.values().any(|attr| attr.ttl > 0);
Self {
values: RwLock::new(
values
.into_iter()
.map(|(value, attr)| (value, Mutex::new(attr)))
.collect(),
),
has_ttl: AtomicBool::new(has_ttl),
}
}
fn record(
&self,
value: &str,
when: DateTime<Utc>,
ttl: Option<u64>,
retention: usize,
) -> (u64, bool, AttributeView) {
if ttl.is_some_and(|ttl| ttl > 0) {
self.has_ttl.store(true, Ordering::Relaxed);
}
{
let values = self.values.read().unwrap_or_else(PoisonError::into_inner);
if let Some(cell) = values.get(value) {
let mut attr = cell.lock().unwrap_or_else(PoisonError::into_inner);
if let Some(ttl) = ttl {
attr.set_ttl(ttl);
}
attr.increment(when, retention);
return (attr.count(), false, attr.view(0, false));
}
}
let mut values = self.values.write().unwrap_or_else(PoisonError::into_inner);
let is_new = !values.contains_key(value);
let cell = values
.entry(value.to_string())
.or_insert_with(|| Mutex::new(Attribute::new(value)));
let attr = cell.get_mut().unwrap_or_else(PoisonError::into_inner);
if let Some(ttl) = ttl {
attr.set_ttl(ttl);
}
attr.increment(when, retention);
(attr.count(), is_new, attr.view(0, false))
}
fn view(
&self,
value: &str,
consensus: u64,
with_stats: bool,
now: DateTime<Utc>,
) -> Option<AttributeView> {
let values = self.values.read().unwrap_or_else(PoisonError::into_inner);
let cell = values.get(value)?;
let attr = cell.lock().unwrap_or_else(PoisonError::into_inner);
(!attr.is_expired(now)).then(|| attr.view(consensus, with_stats))
}
fn count(&self, value: &str, now: DateTime<Utc>) -> u64 {
let values = self.values.read().unwrap_or_else(PoisonError::into_inner);
values.get(value).map_or(0, |cell| {
let attr = cell.lock().unwrap_or_else(PoisonError::into_inner);
if attr.is_expired(now) {
0
} else {
attr.count()
}
})
}
fn all_views(&self, with_stats: bool, now: DateTime<Utc>) -> Vec<AttributeView> {
let values = self.values.read().unwrap_or_else(PoisonError::into_inner);
values
.values()
.filter_map(|cell| {
let attr = cell.lock().unwrap_or_else(PoisonError::into_inner);
(!attr.is_expired(now)).then(|| attr.view(0, with_stats))
})
.collect()
}
fn remove_expired(&self, now: DateTime<Utc>) -> Vec<String> {
if !self.has_ttl.load(Ordering::Relaxed) {
return Vec::new();
}
{
let values = self.values.read().unwrap_or_else(PoisonError::into_inner);
let any_expired = values.values().any(|cell| {
cell.lock()
.unwrap_or_else(PoisonError::into_inner)
.is_expired(now)
});
if !any_expired {
return Vec::new();
}
}
let mut values = self.values.write().unwrap_or_else(PoisonError::into_inner);
let mut removed = Vec::new();
values.retain(|value, cell| {
let expired = cell
.get_mut()
.unwrap_or_else(PoisonError::into_inner)
.is_expired(now);
if expired {
removed.push(value.clone());
}
!expired
});
removed
}
fn release(&self, value: &str) {
let mut values = self.values.write().unwrap_or_else(PoisonError::into_inner);
let Some(cell) = values.get_mut(value) else {
return;
};
let remaining = cell
.get_mut()
.unwrap_or_else(PoisonError::into_inner)
.decrement();
if remaining == 0 {
values.remove(value);
}
}
fn value_names(&self) -> Vec<String> {
self.values
.read()
.unwrap_or_else(PoisonError::into_inner)
.keys()
.cloned()
.collect()
}
fn is_empty(&self) -> bool {
self.values
.read()
.unwrap_or_else(PoisonError::into_inner)
.is_empty()
}
}
#[derive(Default)]
pub struct Database {
namespaces: RwLock<HashMap<String, Arc<Namespace>>>,
policy: DatabasePolicy,
}
impl Database {
#[cfg(test)]
pub fn new() -> Database {
Database::with_policy(DatabasePolicy::default())
}
pub fn with_policy(policy: DatabasePolicy) -> Database {
Database {
namespaces: RwLock::new(HashMap::new()),
policy,
}
}
pub fn from_snapshot(data: SnapshotData, policy: DatabasePolicy) -> Database {
let namespaces = data
.namespaces
.into_iter()
.map(|(name, values)| (name, Arc::new(Namespace::from_values(values))))
.collect();
Database {
namespaces: RwLock::new(namespaces),
policy,
}
}
pub fn legacy_apikeys(&self) -> Vec<String> {
self.namespaces
.read()
.unwrap_or_else(PoisonError::into_inner)
.keys()
.filter_map(|name| name.strip_prefix(APIKEYS_NAMESPACE))
.filter(|key| !key.is_empty())
.map(String::from)
.collect()
}
pub fn write(&self, path: &str, value: &str, when: DateTime<Utc>, opts: WriteOpts) -> u64 {
let ttl = match opts.ttl {
Some(ttl) => Some(ttl),
None if path.starts_with(SHADOW_PREFIX) && self.policy.shadow_ttl > 0 => {
Some(self.policy.shadow_ttl)
}
None => None,
};
let namespace = self.namespace_or_create(path);
let (count, is_new, mut view) =
namespace.record(value, when, ttl, self.policy.stats_retention);
if opts.consensus && is_new {
self.write(ALL_NAMESPACE, value, when, WriteOpts::default());
}
view.consensus = self.count(ALL_NAMESPACE, value);
log_attribute(path, &view);
count
}
pub fn view(
&self,
path: &str,
value: &str,
consensus: u64,
with_stats: bool,
) -> Option<AttributeView> {
self.namespace(path)?
.view(value, consensus, with_stats, Utc::now())
}
pub fn count(&self, path: &str, value: &str) -> u64 {
let now = Utc::now();
self.namespace(path)
.map_or(0, |namespace| namespace.count(value, now))
}
pub fn namespace_exists(&self, namespace: &str) -> bool {
self.namespaces
.read()
.unwrap_or_else(PoisonError::into_inner)
.contains_key(namespace)
}
pub fn namespace_views(&self, namespace: &str) -> Option<Vec<AttributeView>> {
let mut views = self.namespace(namespace)?.all_views(false, Utc::now());
for view in &mut views {
view.consensus = self.count(ALL_NAMESPACE, &view.value);
}
Some(views)
}
pub fn delete(&self, name: &str) -> bool {
let Some(namespace) = self.namespace(name) else {
return false;
};
let values = namespace.value_names();
drop(namespace);
let removed = self
.namespaces
.write()
.unwrap_or_else(PoisonError::into_inner)
.remove(name)
.is_some();
if removed && counts_towards_consensus(name) {
for value in values {
self.release_consensus(&value);
}
}
removed
}
pub fn sweep(&self, now: DateTime<Utc>) -> SweepReport {
let entries: Vec<(String, Arc<Namespace>)> = {
let map = self
.namespaces
.read()
.unwrap_or_else(PoisonError::into_inner);
map.iter()
.map(|(name, namespace)| (name.clone(), Arc::clone(namespace)))
.collect()
};
let mut report = SweepReport::default();
for (name, namespace) in &entries {
if name.starts_with(CONFIG_PREFIX) {
continue;
}
let expired = namespace.remove_expired(now);
report.values_removed += expired.len();
if counts_towards_consensus(name) {
for value in expired {
self.release_consensus(&value);
}
}
}
drop(entries);
report.namespaces_removed = self.prune_empty();
report
}
pub fn snapshot(&self) -> Snapshot<'_> {
Snapshot(self)
}
pub fn namespace_count(&self) -> usize {
self.namespaces
.read()
.unwrap_or_else(PoisonError::into_inner)
.len()
}
pub fn namespace_page(
&self,
filter: &str,
offset: usize,
limit: usize,
allowed: impl Fn(&str) -> bool,
) -> Page<String> {
let filter = filter.to_ascii_lowercase();
let map = self
.namespaces
.read()
.unwrap_or_else(PoisonError::into_inner);
let mut names: Vec<&String> = map
.keys()
.filter(|name| !name.starts_with(CONFIG_PREFIX) && *name != ALL_NAMESPACE)
.filter(|name| filter.is_empty() || name.to_ascii_lowercase().contains(&filter))
.filter(|name| allowed(name))
.collect();
names.sort_unstable();
let total = names.len();
let items = names
.into_iter()
.skip(offset)
.take(limit)
.cloned()
.collect();
Page {
items,
total,
offset,
}
}
pub fn value_page(
&self,
namespace: &str,
filter: &str,
offset: usize,
limit: usize,
with_stats: bool,
) -> Option<Page<AttributeView>> {
let now = Utc::now();
let filter = filter.to_ascii_lowercase();
let ns = self.namespace(namespace)?;
let values = ns.values.read().unwrap_or_else(PoisonError::into_inner);
let mut matching: Vec<&String> = values
.iter()
.filter(|(value, _)| filter.is_empty() || value.to_ascii_lowercase().contains(&filter))
.filter(|(_, cell)| {
!cell
.lock()
.unwrap_or_else(PoisonError::into_inner)
.is_expired(now)
})
.map(|(value, _)| value)
.collect();
matching.sort_unstable();
let total = matching.len();
let items: Vec<AttributeView> = matching
.into_iter()
.skip(offset)
.take(limit)
.filter_map(|value| {
let attr = values
.get(value)?
.lock()
.unwrap_or_else(PoisonError::into_inner);
Some(attr.view(0, with_stats))
})
.collect();
drop(values);
let items = items
.into_iter()
.map(|mut view| {
view.consensus = self.count(ALL_NAMESPACE, &view.value);
view
})
.collect();
Some(Page {
items,
total,
offset,
})
}
fn release_consensus(&self, value: &str) {
if let Some(all) = self.namespace(ALL_NAMESPACE) {
all.release(value);
}
}
fn prune_empty(&self) -> usize {
let mut map = self
.namespaces
.write()
.unwrap_or_else(PoisonError::into_inner);
let before = map.len();
map.retain(|name, namespace| {
if name.starts_with(CONFIG_PREFIX) {
return true;
}
Arc::strong_count(namespace) > 1 || !namespace.is_empty()
});
before - map.len()
}
fn namespace(&self, name: &str) -> Option<Arc<Namespace>> {
self.namespaces
.read()
.unwrap_or_else(PoisonError::into_inner)
.get(name)
.cloned()
}
fn namespace_or_create(&self, name: &str) -> Arc<Namespace> {
if let Some(namespace) = self.namespace(name) {
return namespace;
}
self.namespaces
.write()
.unwrap_or_else(PoisonError::into_inner)
.entry(name.to_string())
.or_default()
.clone()
}
}
fn counts_towards_consensus(name: &str) -> bool {
name != ALL_NAMESPACE && !name.starts_with(SHADOW_PREFIX) && !name.starts_with(CONFIG_PREFIX)
}
#[derive(Debug, Deserialize)]
pub struct SnapshotData {
pub version: u32,
pub namespaces: HashMap<String, HashMap<String, Attribute>>,
}
pub struct Snapshot<'a>(&'a Database);
impl Serialize for Snapshot<'_> {
fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
let mut out = serializer.serialize_struct("Snapshot", 2)?;
out.serialize_field("version", &SNAPSHOT_VERSION)?;
out.serialize_field("namespaces", &NamespacesRef(self.0))?;
out.end()
}
}
struct NamespacesRef<'a>(&'a Database);
impl Serialize for NamespacesRef<'_> {
fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
use serde::ser::SerializeMap;
let entries: Vec<(String, Arc<Namespace>)> = {
let map = self
.0
.namespaces
.read()
.unwrap_or_else(PoisonError::into_inner);
map.iter()
.map(|(name, namespace)| (name.clone(), Arc::clone(namespace)))
.collect()
};
let mut out = serializer.serialize_map(Some(entries.len()))?;
for (name, namespace) in &entries {
out.serialize_entry(name, &NamespaceRef(namespace))?;
}
out.end()
}
}
struct NamespaceRef<'a>(&'a Namespace);
impl Serialize for NamespaceRef<'_> {
fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
use serde::ser::SerializeMap;
let values = self.0.values.read().unwrap_or_else(PoisonError::into_inner);
let mut out = serializer.serialize_map(Some(values.len()))?;
for (value, cell) in values.iter() {
let attr = cell.lock().unwrap_or_else(PoisonError::into_inner);
out.serialize_entry(value, &*attr)?;
}
out.end()
}
}
#[cfg(test)]
mod tests {
use super::*;
fn at(secs: i64) -> DateTime<Utc> {
DateTime::from_timestamp(secs, 0).expect("timestamp in range")
}
fn consensus() -> WriteOpts {
WriteOpts {
consensus: true,
ttl: None,
}
}
fn with_ttl(ttl: u64) -> WriteOpts {
WriteOpts {
consensus: true,
ttl: Some(ttl),
}
}
#[test]
fn write_returns_the_running_count() {
let db = Database::default();
assert_eq!(db.write("ns", "1.2.3.4", at(100), consensus()), 1);
assert_eq!(db.write("ns", "1.2.3.4", at(200), consensus()), 2);
assert_eq!(db.count("ns", "1.2.3.4"), 2);
}
#[test]
fn consensus_counts_namespaces_not_writes() {
let db = Database::default();
db.write("my/namespace", "127.0.0.1", at(100), consensus());
db.write("another/namespace", "127.0.0.1", at(200), consensus());
db.write("another/namespace", "127.0.0.1", at(300), consensus());
assert_eq!(db.count(ALL_NAMESPACE, "127.0.0.1"), 2);
}
#[test]
fn a_new_value_in_an_existing_namespace_still_counts_for_consensus() {
let db = Database::default();
db.write("ns", "a", at(100), consensus());
db.write("ns", "b", at(100), consensus());
assert_eq!(db.count(ALL_NAMESPACE, "b"), 1);
}
#[test]
fn writes_without_consensus_leave_all_alone() {
let db = Database::default();
db.write("ns", "a", at(100), WriteOpts::default());
assert_eq!(db.count(ALL_NAMESPACE, "a"), 0);
}
#[test]
fn missing_lookups_are_zero_and_none() {
let db = Database::default();
assert_eq!(db.count("nope", "nope"), 0);
assert!(db.view("nope", "nope", 0, false).is_none());
assert!(!db.namespace_exists("nope"));
assert!(db.namespace_views("nope").is_none());
}
#[test]
fn legacy_apikeys_are_recovered_from_old_snapshots() {
let db = Database::default();
assert!(db.legacy_apikeys().is_empty());
db.write(
&format!("{APIKEYS_NAMESPACE}{DEFAULT_APIKEY}"),
"",
at(100),
WriteOpts::default(),
);
db.write(
&format!("{APIKEYS_NAMESPACE}secret"),
"",
at(100),
WriteOpts::default(),
);
let mut keys = db.legacy_apikeys();
keys.sort();
assert_eq!(keys, [DEFAULT_APIKEY, "secret"]);
}
#[test]
fn a_fresh_database_stores_no_keys() {
let db = Database::new();
assert!(db.legacy_apikeys().is_empty());
}
#[test]
fn delete_removes_the_namespace_once() {
let db = Database::default();
db.write("ns", "a", at(100), consensus());
assert!(db.delete("ns"));
assert!(!db.delete("ns"));
assert!(!db.namespace_exists("ns"));
}
#[test]
fn delete_gives_back_the_consensus_it_was_holding() {
let db = Database::default();
db.write("a/ns", "v", at(100), consensus());
db.write("b/ns", "v", at(100), consensus());
assert_eq!(db.count(ALL_NAMESPACE, "v"), 2);
db.delete("a/ns");
assert_eq!(db.count(ALL_NAMESPACE, "v"), 1);
db.delete("b/ns");
assert_eq!(db.count(ALL_NAMESPACE, "v"), 0);
}
#[test]
fn an_expired_attribute_is_invisible_before_it_is_swept() {
let db = Database::default();
db.write("ns", "v", at(1000), with_ttl(60));
assert!(db.view("ns", "v", 0, false).is_none());
assert_eq!(db.count("ns", "v"), 0);
assert_eq!(db.namespace_views("ns").unwrap().len(), 0);
}
#[test]
fn a_live_attribute_reports_its_ttl() {
let db = Database::default();
db.write("ns", "v", Utc::now(), with_ttl(3600));
let view = db.view("ns", "v", 0, false).unwrap();
assert_eq!(view.ttl, 3600);
}
#[test]
fn writing_again_without_a_ttl_keeps_the_existing_one() {
let db = Database::default();
db.write("ns", "v", Utc::now(), with_ttl(3600));
db.write("ns", "v", Utc::now(), consensus());
assert_eq!(db.view("ns", "v", 0, false).unwrap().ttl, 3600);
}
#[test]
fn sweeping_reclaims_expired_values_and_their_consensus() {
let db = Database::default();
db.write("a/ns", "v", at(1000), with_ttl(60));
db.write("b/ns", "v", at(1000), consensus());
assert_eq!(db.count(ALL_NAMESPACE, "v"), 2);
let report = db.sweep(Utc::now());
assert_eq!(report.values_removed, 1);
assert_eq!(report.namespaces_removed, 1); assert!(!db.namespace_exists("a/ns"));
assert!(db.namespace_exists("b/ns"));
assert_eq!(db.count(ALL_NAMESPACE, "v"), 1);
}
#[test]
fn sweeping_leaves_live_data_alone() {
let db = Database::default();
db.write("ns", "forever", at(1000), consensus());
db.write("ns", "later", Utc::now(), with_ttl(3600));
assert_eq!(db.sweep(Utc::now()), SweepReport::default());
assert_eq!(db.namespace_views("ns").unwrap().len(), 2);
}
#[test]
fn sweeping_never_touches_api_keys() {
let db = Database::default();
let namespace = format!("{APIKEYS_NAMESPACE}{DEFAULT_APIKEY}");
db.write(&namespace, "", at(100), WriteOpts::default());
db.sweep(Utc::now());
assert!(db.namespace_exists(&namespace));
assert_eq!(db.legacy_apikeys(), [DEFAULT_APIKEY]);
}
#[test]
fn shadow_sightings_inherit_the_policy_ttl() {
let db = Database::with_policy(DatabasePolicy {
stats_retention: 0,
shadow_ttl: 60,
});
db.write("_shadow/ns", "v", at(1000), WriteOpts::default());
assert_eq!(db.count("_shadow/ns", "v"), 0);
assert_eq!(db.sweep(Utc::now()).values_removed, 1);
}
#[test]
fn the_policy_ttl_does_not_leak_into_ordinary_namespaces() {
let db = Database::with_policy(DatabasePolicy {
stats_retention: 0,
shadow_ttl: 60,
});
db.write("ns", "v", at(1000), consensus());
assert_eq!(db.count("ns", "v"), 1);
}
#[test]
fn stats_retention_is_applied_on_write() {
let db = Database::with_policy(DatabasePolicy {
stats_retention: 2,
shadow_ttl: 0,
});
for hour in 0..5 {
db.write("ns", "v", at(hour * 3600), consensus());
}
let view = db.view("ns", "v", 0, true).unwrap();
assert_eq!(view.stats.unwrap().len(), 2);
assert_eq!(view.count, 5);
}
#[test]
fn a_snapshot_round_trips() {
let db = Database::new();
db.write("my/ns", "1.2.3.4", at(1_600_000_000), consensus());
db.write("my/ns", "1.2.3.4", at(1_600_003_600), consensus());
db.write("other/ns", "1.2.3.4", at(1_600_000_000), with_ttl(99));
let json = serde_json::to_string(&db.snapshot()).unwrap();
let data: SnapshotData = serde_json::from_str(&json).unwrap();
assert_eq!(data.version, SNAPSHOT_VERSION);
let restored = Database::from_snapshot(data, DatabasePolicy::default());
assert_eq!(restored.count("my/ns", "1.2.3.4"), 2);
assert_eq!(restored.count(ALL_NAMESPACE, "1.2.3.4"), 2);
let view = restored.view("my/ns", "1.2.3.4", 0, true).unwrap();
assert_eq!(view.first_seen, 1_600_000_000);
assert_eq!(view.last_seen, 1_600_003_600);
assert_eq!(view.stats.unwrap().len(), 2);
}
#[test]
fn a_restored_database_still_knows_about_ttls() {
let db = Database::new();
db.write("ns", "v", at(1000), with_ttl(60));
let json = serde_json::to_string(&db.snapshot()).unwrap();
let restored = Database::from_snapshot(
serde_json::from_str(&json).unwrap(),
DatabasePolicy::default(),
);
assert_eq!(restored.sweep(Utc::now()).values_removed, 1);
}
#[test]
fn an_empty_database_snapshots_cleanly() {
let db = Database::default();
let json = serde_json::to_string(&db.snapshot()).unwrap();
assert_eq!(json, r#"{"version":1,"namespaces":{}}"#);
}
#[test]
fn namespaces_page_in_sorted_order() {
let db = Database::default();
for name in ["c/ns", "a/ns", "b/ns"] {
db.write(name, "v", at(100), consensus());
}
let first = db.namespace_page("", 0, 2, |_| true);
assert_eq!(first.items, ["a/ns", "b/ns"]);
assert_eq!(first.total, 3);
assert_eq!(first.offset, 0);
let second = db.namespace_page("", 2, 2, |_| true);
assert_eq!(second.items, ["c/ns"]);
}
#[test]
fn namespaces_can_be_filtered() {
let db = Database::default();
db.write("feeds/misp", "v", at(100), consensus());
db.write("feeds/otx", "v", at(100), consensus());
db.write("internal/notes", "v", at(100), consensus());
let page = db.namespace_page("feeds", 0, 10, |_| true);
assert_eq!(page.items, ["feeds/misp", "feeds/otx"]);
assert_eq!(page.total, 2);
}
#[test]
fn the_config_tree_is_not_listed() {
let db = Database::default();
db.write(
"_config/acl/apikeys/changeme",
"",
at(100),
WriteOpts::default(),
);
db.write("ns", "v", at(100), consensus());
let page = db.namespace_page("", 0, 100, |_| true);
assert!(!page.items.iter().any(|n| n.starts_with("_config")));
assert!(!page.items.iter().any(|n| n == ALL_NAMESPACE));
assert_eq!(page.items, ["ns"]);
}
#[test]
fn values_page_in_sorted_order_with_a_total() {
let db = Database::default();
for value in ["ccc", "aaa", "bbb", "ddd"] {
db.write("ns", value, at(100), consensus());
}
let page = db.value_page("ns", "", 1, 2, false).unwrap();
let values: Vec<&str> = page.items.iter().map(|v| v.value.as_str()).collect();
assert_eq!(values, ["bbb", "ccc"]);
assert_eq!(page.total, 4);
assert_eq!(page.offset, 1);
}
#[test]
fn values_can_be_filtered_and_carry_consensus() {
let db = Database::default();
db.write("a/ns", "1.2.3.4", at(100), consensus());
db.write("b/ns", "1.2.3.4", at(100), consensus());
db.write("a/ns", "9.9.9.9", at(100), consensus());
let page = db.value_page("a/ns", "1.2", 0, 10, false).unwrap();
assert_eq!(page.total, 1);
assert_eq!(page.items[0].value, "1.2.3.4");
assert_eq!(page.items[0].consensus, 2);
}
#[test]
fn stats_are_included_only_when_asked_for() {
let db = Database::default();
db.write("ns", "v", at(3600), consensus());
assert!(
db.value_page("ns", "", 0, 10, false).unwrap().items[0]
.stats
.is_none()
);
let with = db.value_page("ns", "", 0, 10, true).unwrap();
assert_eq!(with.items[0].stats.as_ref().unwrap().get(&3600), Some(&1));
}
#[test]
fn expired_values_do_not_appear_in_a_page() {
let db = Database::default();
db.write("ns", "live", Utc::now(), consensus());
db.write("ns", "dead", at(1000), with_ttl(60));
let page = db.value_page("ns", "", 0, 10, false).unwrap();
assert_eq!(page.total, 1);
assert_eq!(page.items[0].value, "live");
}
#[test]
fn paging_a_missing_namespace_is_none() {
assert!(
Database::default()
.value_page("nope", "", 0, 10, false)
.is_none()
);
}
#[test]
fn an_offset_past_the_end_is_an_empty_page_not_an_error() {
let db = Database::default();
db.write("ns", "v", at(100), consensus());
let page = db.value_page("ns", "", 500, 10, false).unwrap();
assert!(page.items.is_empty());
assert_eq!(page.total, 1);
}
#[test]
fn the_listing_hides_namespaces_the_caller_cannot_read() {
let db = Database::default();
db.write("feeds/misp", "v", at(100), consensus());
db.write("secrets/hr", "v", at(100), consensus());
let page = db.namespace_page("", 0, 100, |name| name.starts_with("feeds"));
assert_eq!(page.items, ["feeds/misp"]);
assert_eq!(page.total, 1);
}
#[test]
fn concurrent_writes_to_one_value_are_all_counted() {
const THREADS: usize = 8;
const PER_THREAD: usize = 500;
let db = Arc::new(Database::default());
std::thread::scope(|scope| {
for _ in 0..THREADS {
let db = Arc::clone(&db);
scope.spawn(move || {
for i in 0..PER_THREAD {
db.write("ns", "shared", at(i as i64), consensus());
}
});
}
});
assert_eq!(db.count("ns", "shared"), (THREADS * PER_THREAD) as u64);
assert_eq!(db.count(ALL_NAMESPACE, "shared"), 1);
}
#[test]
fn concurrent_writes_across_namespaces_agree_on_consensus() {
const THREADS: usize = 8;
let db = Arc::new(Database::default());
std::thread::scope(|scope| {
for t in 0..THREADS {
let db = Arc::clone(&db);
scope.spawn(move || {
for i in 0..200 {
db.write(&format!("ns/{t}"), "shared", at(i), consensus());
}
});
}
});
assert_eq!(db.count(ALL_NAMESPACE, "shared"), THREADS as u64);
for t in 0..THREADS {
assert_eq!(db.count(&format!("ns/{t}"), "shared"), 200);
}
}
#[test]
fn readers_and_writers_do_not_deadlock() {
let db = Arc::new(Database::default());
std::thread::scope(|scope| {
for t in 0..8 {
let db = Arc::clone(&db);
scope.spawn(move || {
for i in 0..500 {
let value = format!("v{}", i % 20);
db.write(&format!("ns/{}", i % 3), &value, at(i), consensus());
db.count(ALL_NAMESPACE, &value);
db.view(&format!("ns/{}", t % 3), &value, 0, true);
db.namespace_views(&format!("ns/{}", i % 3));
}
});
}
});
for v in 0..20 {
assert_eq!(db.count(ALL_NAMESPACE, &format!("v{v}")), 3);
}
}
#[test]
fn sweeping_concurrently_with_writers_loses_nothing() {
let db = Arc::new(Database::default());
let stop = Arc::new(AtomicBool::new(false));
std::thread::scope(|scope| {
let sweeper_db = Arc::clone(&db);
let sweeper_stop = Arc::clone(&stop);
scope.spawn(move || {
while !sweeper_stop.load(Ordering::Relaxed) {
sweeper_db.sweep(Utc::now());
}
});
for t in 0..4 {
let db = Arc::clone(&db);
scope.spawn(move || {
for _ in 0..500 {
db.write(&format!("ns/{t}"), "v", Utc::now(), consensus());
}
});
}
scope.spawn({
let stop = Arc::clone(&stop);
move || {
std::thread::sleep(std::time::Duration::from_millis(300));
stop.store(true, Ordering::Relaxed);
}
});
});
for t in 0..4 {
assert_eq!(db.count(&format!("ns/{t}"), "v"), 500, "namespace ns/{t}");
}
}
}