use std::collections::HashMap;
use std::sync::Arc;
use crate::Key;
use crate::error::{DbError, DbResult};
use crate::hex::hex;
use crate::hook::{TypedWriteHook, WriteHook};
use crate::projection::{IndexKeyed, ProjectableIndexTx};
use crate::upsert::Upsert;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OnMissingTarget {
Error,
Skip,
Create,
}
#[derive(Debug)]
pub(crate) enum CounterApply {
Put(Vec<u8>),
Delete,
Skip,
}
pub trait CounterCollection: IndexKeyed {
fn decode_value(&self, bytes: &[u8]) -> DbResult<Self::V>;
fn contains_bytes(&self, key: &[u8]) -> DbResult<bool>;
fn read_bytes(&self, key: &[u8]) -> DbResult<Option<Vec<u8>>>;
#[allow(clippy::type_complexity)]
fn upsert_counter(
&self,
key: &[u8],
f: &mut dyn FnMut(Option<&Self::V>) -> Upsert<Self::V>,
) -> DbResult<()>;
}
type ReadFn<V> = Box<dyn Fn(&V) -> u64 + Send + Sync>;
type ApplyFn<V> = Box<dyn Fn(Option<&V>, u64) -> Upsert<V> + Send + Sync>;
pub struct TypedCounterTarget<C: CounterCollection> {
coll: Arc<C>,
field: &'static str,
on_missing: OnMissingTarget,
read: ReadFn<C::V>,
apply: ApplyFn<C::V>,
}
impl<C: CounterCollection> TypedCounterTarget<C> {
pub fn new(
coll: Arc<C>,
field: &'static str,
on_missing: OnMissingTarget,
read: impl Fn(&C::V) -> u64 + Send + Sync + 'static,
apply: impl Fn(Option<&C::V>, u64) -> Upsert<C::V> + Send + Sync + 'static,
) -> Self {
Self {
coll,
field,
on_missing,
read: Box::new(read),
apply: Box::new(apply),
}
}
pub(crate) fn resolve_typed(&self, cur: Option<&C::V>, delta: i128) -> DbResult<Upsert<C::V>> {
if cur.is_none() {
match self.on_missing {
OnMissingTarget::Error => {
return Err(DbError::CounterTargetMissing {
targets: self.coll.collection_name().to_string(),
key: String::from("<resolve>"),
});
}
OnMissingTarget::Skip => return Ok(Upsert::Keep),
OnMissingTarget::Create => {}
}
}
let base = i128::from(cur.map_or(0, |v| (self.read)(v)));
let next = base
.checked_add(delta)
.and_then(|n| u64::try_from(n).ok())
.ok_or_else(|| DbError::CounterOverflow {
collection: self.coll.collection_name(),
field: self.field,
key: String::from("<resolve>"),
})?;
match (self.apply)(cur, next) {
Upsert::Put(v) => Ok(Upsert::Put(v)),
Upsert::Keep => Ok(Upsert::Keep),
Upsert::Delete if next == 0 => Ok(Upsert::Delete),
Upsert::Delete => Err(DbError::Internal(
"counter apply returned Delete at a non-zero count",
)),
}
}
pub(crate) fn resolve(&self, cur: Option<&[u8]>, delta: i128) -> DbResult<CounterApply> {
let decoded = cur.map(|bytes| self.coll.decode_value(bytes)).transpose()?;
match self.resolve_typed(decoded.as_ref(), delta)? {
Upsert::Put(v) => Ok(CounterApply::Put(self.coll.encode_value(&v)?)),
Upsert::Keep => Ok(CounterApply::Skip),
Upsert::Delete => Ok(CounterApply::Delete),
}
}
}
impl DbError {
pub(crate) fn with_key(self, key: &[u8]) -> DbError {
match self {
DbError::CounterTargetMissing { targets, .. } => DbError::CounterTargetMissing {
targets,
key: hex(key),
},
DbError::CounterOverflow {
collection, field, ..
} => DbError::CounterOverflow {
collection,
field,
key: hex(key),
},
other => other,
}
}
}
pub(crate) trait ErasedCounterTarget: Send + Sync {
fn collection_name(&self) -> &'static str;
fn field(&self) -> &'static str;
fn on_missing(&self) -> OnMissingTarget;
fn resolve(&self, cur: Option<&[u8]>, delta: i128) -> DbResult<CounterApply>;
fn apply_delta(&self, key: &[u8], delta: i128) -> DbResult<bool>;
fn contains(&self, key: &[u8]) -> DbResult<bool>;
fn for_each_counter_row(&self, f: &mut dyn FnMut(&[u8], u64) -> DbResult<()>) -> DbResult<()>;
fn durable_flush(&self) -> DbResult<()>;
fn as_tx_participant(&self) -> Arc<dyn crate::txn::participant::TxParticipant>;
}
impl std::fmt::Debug for dyn ErasedCounterTarget {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ErasedCounterTarget")
.field("collection", &self.collection_name())
.field("field", &self.field())
.finish()
}
}
impl<C: CounterCollection + ProjectableIndexTx> ErasedCounterTarget for TypedCounterTarget<C> {
fn collection_name(&self) -> &'static str {
self.coll.collection_name()
}
fn field(&self) -> &'static str {
self.field
}
fn on_missing(&self) -> OnMissingTarget {
self.on_missing
}
fn resolve(&self, cur: Option<&[u8]>, delta: i128) -> DbResult<CounterApply> {
TypedCounterTarget::resolve(self, cur, delta)
}
fn apply_delta(&self, key: &[u8], delta: i128) -> DbResult<bool> {
counter_delta_fault()?;
let mut outcome: DbResult<()> = Ok(());
let mut wrote = false;
self.coll
.upsert_counter(key, &mut |cur| match self.resolve_typed(cur, delta) {
Ok(decision) => {
wrote = !matches!(decision, Upsert::Keep);
decision
}
Err(e) => {
outcome = Err(e.with_key(key));
Upsert::Keep
}
})?;
outcome.map(|()| wrote)
}
fn contains(&self, key: &[u8]) -> DbResult<bool> {
self.coll.contains_bytes(key)
}
fn for_each_counter_row(&self, f: &mut dyn FnMut(&[u8], u64) -> DbResult<()>) -> DbResult<()> {
self.coll.for_each_bytes(&mut |k, v| {
let val = self.coll.decode_value(v)?;
f(k, (self.read)(&val))
})
}
fn durable_flush(&self) -> DbResult<()> {
self.coll.durable_flush()
}
fn as_tx_participant(&self) -> Arc<dyn crate::txn::participant::TxParticipant> {
Arc::clone(&self.coll).as_tx_participant()
}
}
#[derive(Clone)]
pub(crate) struct ErasedGroup {
pub(crate) members: Vec<Arc<dyn ErasedCounterTarget>>,
pub(crate) one_of: bool,
}
#[derive(Clone, Default)]
pub struct CounterTargets {
groups: Vec<ErasedGroup>,
}
impl CounterTargets {
pub fn new() -> Self {
Self::default()
}
#[allow(private_bounds)]
pub fn with_counter<C: CounterCollection + ProjectableIndexTx>(
mut self,
coll: &Arc<C>,
field: &'static str,
on_missing: OnMissingTarget,
read: impl Fn(&C::V) -> u64 + Send + Sync + 'static,
apply: impl Fn(Option<&C::V>, u64) -> Upsert<C::V> + Send + Sync + 'static,
) -> Self {
let name = coll.collection_name();
assert!(
self.groups
.iter()
.flat_map(|g| g.members.iter())
.all(|m| m.collection_name() != name),
"with_counter: collection `{name}` already has a declared counter target in this CounterTargets"
);
let t = TypedCounterTarget::new(Arc::clone(coll), field, on_missing, read, apply);
self.groups.push(ErasedGroup {
members: vec![Arc::new(t)],
one_of: false,
});
self
}
pub fn one_of(self) -> OneOfBuilder {
OneOfBuilder {
targets: self,
members: Vec::new(),
}
}
pub(crate) fn groups(&self) -> &[ErasedGroup] {
&self.groups
}
pub(crate) fn all(&self) -> impl Iterator<Item = &Arc<dyn ErasedCounterTarget>> {
self.groups.iter().flat_map(|g| g.members.iter())
}
pub fn is_empty(&self) -> bool {
self.groups.is_empty()
}
pub fn len(&self) -> usize {
self.groups.len()
}
pub(crate) fn resolve_group(
&self,
gi: usize,
key: &[u8],
) -> DbResult<Option<&Arc<dyn ErasedCounterTarget>>> {
let g = self
.groups
.get(gi)
.ok_or(DbError::Internal("counter group index"))?;
if !g.one_of {
return Ok(Some(&g.members[0]));
}
let mut holder = None;
for m in &g.members {
if m.contains(key)? {
if holder.is_some() {
return Err(DbError::AmbiguousCounterTarget {
targets: group_names(g),
key: hex(key),
});
}
holder = Some(m);
}
}
holder
.map(Some)
.ok_or_else(|| DbError::CounterTargetMissing {
targets: group_names(g),
key: hex(key),
})
}
}
fn group_names(g: &ErasedGroup) -> String {
g.members
.iter()
.map(|m| m.collection_name())
.collect::<Vec<_>>()
.join(", ")
}
pub struct OneOfBuilder {
targets: CounterTargets,
members: Vec<Arc<dyn ErasedCounterTarget>>,
}
impl OneOfBuilder {
#[allow(private_bounds)]
pub fn with<C: CounterCollection + ProjectableIndexTx>(
mut self,
coll: &Arc<C>,
field: &'static str,
on_missing: OnMissingTarget,
read: impl Fn(&C::V) -> u64 + Send + Sync + 'static,
apply: impl Fn(Option<&C::V>, u64) -> Upsert<C::V> + Send + Sync + 'static,
) -> Self {
let name = coll.collection_name();
assert!(
self.targets
.groups
.iter()
.flat_map(|g| g.members.iter())
.chain(self.members.iter())
.all(|m| m.collection_name() != name),
"one_of: collection `{name}` already has a declared counter target in this CounterTargets"
);
let t = TypedCounterTarget::new(Arc::clone(coll), field, on_missing, read, apply);
self.members.push(Arc::new(t));
self
}
pub fn build(mut self) -> CounterTargets {
if self.members.is_empty() {
return self.targets;
}
debug_assert!(
self.members.len() >= 2,
"one_of() group must have at least two members"
);
let one_of = self.members.len() >= 2;
if one_of {
assert!(
!self.targets.groups.iter().any(|g| g.one_of),
"one_of: a CounterTargets may declare at most one one_of group"
);
}
self.targets.groups.push(ErasedGroup {
members: self.members,
one_of,
});
self.targets
}
}
pub struct CounterSink<'a> {
targets: &'a CounterTargets,
max_rows: usize,
max_bytes: usize,
emissions: usize,
bytes: usize,
acc: HashMap<(usize, Vec<u8>), u128>,
}
impl CounterSink<'_> {
pub(crate) fn new(
targets: &CounterTargets,
max_rows: usize,
max_bytes: usize,
) -> CounterSink<'_> {
CounterSink {
targets,
max_rows,
max_bytes,
emissions: 0,
bytes: 0,
acc: HashMap::new(),
}
}
#[cfg(test)]
pub(crate) fn contributions(&self) -> &HashMap<(usize, Vec<u8>), u128> {
&self.acc
}
pub(crate) fn into_acc(self) -> HashMap<(usize, Vec<u8>), u128> {
self.acc
}
pub fn add<C: CounterCollection>(
&mut self,
target: &Arc<C>,
key: C::K,
weight: u64,
) -> DbResult<()> {
let name = target.collection_name();
let gi = self
.targets
.groups()
.iter()
.position(|g| !g.one_of && g.members[0].collection_name() == name)
.ok_or(DbError::UndeclaredCounterTarget { target: name })?;
self.record(gi, C::encode_key(&key), weight)
}
pub fn add_one_of(&mut self, key: impl Key, weight: u64) -> DbResult<()> {
let gi = self
.targets
.groups()
.iter()
.position(|g| g.one_of)
.ok_or(DbError::UndeclaredCounterTarget { target: "<one_of>" })?;
self.record(gi, key.as_bytes().to_vec(), weight)
}
fn record(&mut self, gi: usize, key: Vec<u8>, weight: u64) -> DbResult<()> {
self.emissions += 1;
self.bytes += key.len() + 8;
if self.emissions > self.max_rows || self.bytes > self.max_bytes {
return Err(DbError::ProjectionBudgetExceeded {
rows: self.emissions,
bytes: self.bytes,
});
}
let slot = self.acc.entry((gi, key)).or_insert(0u128);
*slot = slot
.checked_add(u128::from(weight))
.ok_or(DbError::Internal("counter contribution sum overflow"))?;
Ok(())
}
}
#[derive(Debug, Clone)]
pub(crate) struct PendingDelta {
pub group: usize,
pub key: Vec<u8>,
pub delta: i128,
}
#[cfg(feature = "armour")]
pub(crate) fn into_tx_counter_deltas(
targets: &CounterTargets,
deltas: Vec<PendingDelta>,
) -> DbResult<Vec<crate::txn::participant::TxCounterDelta>> {
let mut out = Vec::with_capacity(deltas.len());
for d in deltas {
let Some(target) = targets.resolve_group(d.group, &d.key)? else {
return Err(DbError::Internal(
"counter group resolved to no target inside a transaction",
));
};
out.push(crate::txn::participant::TxCounterDelta {
participant: target.as_tx_participant(),
target: Arc::clone(target),
key: d.key,
delta: d.delta,
});
}
Ok(out)
}
fn counter_snapshot<K: Key, T, H: TypedWriteHook<K, T>>(
hook: &H,
targets: &CounterTargets,
max_rows: usize,
max_bytes: usize,
key: &K,
snapshot: Option<&T>,
) -> DbResult<HashMap<(usize, Vec<u8>), u128>> {
let mut sink = CounterSink::new(targets, max_rows, max_bytes);
if let Some(value) = snapshot {
hook.counter_project(key, value, &mut sink)?;
}
Ok(sink.into_acc())
}
fn counter_snapshot_bytes<K: Key, H: WriteHook<K>>(
hook: &H,
targets: &CounterTargets,
max_rows: usize,
max_bytes: usize,
key: &K,
snapshot: Option<&[u8]>,
) -> DbResult<HashMap<(usize, Vec<u8>), u128>> {
let mut sink = CounterSink::new(targets, max_rows, max_bytes);
if let Some(value) = snapshot {
hook.counter_project(key, None, Some(value), &mut sink)?;
}
Ok(sink.into_acc())
}
fn diff_counter_snapshots(
targets: &CounterTargets,
a: HashMap<(usize, Vec<u8>), u128>,
b: HashMap<(usize, Vec<u8>), u128>,
) -> DbResult<Vec<PendingDelta>> {
let overflow = |slot: &(usize, Vec<u8>)| -> DbError {
let (collection, field) = targets
.groups()
.get(slot.0)
.and_then(|g| g.members.first())
.map(|m| (m.collection_name(), m.field()))
.unwrap_or(("<unknown>", "<unknown>"));
DbError::CounterOverflow {
collection,
field,
key: hex(&slot.1),
}
};
let mut out = Vec::new();
for (slot, new_w) in &b {
let old_w = a.get(slot).copied().unwrap_or(0);
let delta = i128::try_from(*new_w).map_err(|_| overflow(slot))?
- i128::try_from(old_w).map_err(|_| overflow(slot))?;
if delta != 0 {
out.push(PendingDelta {
group: slot.0,
key: slot.1.clone(),
delta,
});
}
}
for (slot, old_w) in &a {
if !b.contains_key(slot) {
out.push(PendingDelta {
group: slot.0,
key: slot.1.clone(),
delta: -i128::try_from(*old_w).map_err(|_| overflow(slot))?,
});
}
}
out.sort_by(|x, y| (x.group, &x.key).cmp(&(y.group, &y.key)));
Ok(out)
}
pub(crate) fn diff_counters<K: Key, T, H: TypedWriteHook<K, T>>(
hook: &H,
targets: &CounterTargets,
max_rows: usize,
max_bytes: usize,
key: &K,
old: Option<&T>,
new: Option<&T>,
) -> DbResult<Vec<PendingDelta>> {
if !H::COUNTS {
return Ok(Vec::new());
}
let a = counter_snapshot(hook, targets, max_rows, max_bytes, key, old)?;
let b = counter_snapshot(hook, targets, max_rows, max_bytes, key, new)?;
diff_counter_snapshots(targets, a, b)
}
pub(crate) fn diff_counters_bytes<K: Key, H: WriteHook<K>>(
hook: &H,
targets: &CounterTargets,
max_rows: usize,
max_bytes: usize,
key: &K,
old: Option<&[u8]>,
new: Option<&[u8]>,
) -> DbResult<Vec<PendingDelta>> {
if !H::COUNTS {
return Ok(Vec::new());
}
let a = counter_snapshot_bytes(hook, targets, max_rows, max_bytes, key, old)?;
let b = counter_snapshot_bytes(hook, targets, max_rows, max_bytes, key, new)?;
diff_counter_snapshots(targets, a, b)
}
#[cfg(any(test, feature = "test-faults"))]
pub(crate) static FAIL_COUNTER_DELTA: std::sync::atomic::AtomicBool =
std::sync::atomic::AtomicBool::new(false);
#[inline(always)]
fn counter_delta_fault() -> DbResult<()> {
#[cfg(any(test, feature = "test-faults"))]
if FAIL_COUNTER_DELTA.load(std::sync::atomic::Ordering::SeqCst) {
return Err(DbError::Internal("injected counter delta failure"));
}
Ok(())
}
#[allow(clippy::type_complexity)]
pub(crate) fn recount<K: Key, V: ?Sized>(
project: impl Fn(&K, &V, &mut CounterSink<'_>) -> DbResult<()>,
targets: &CounterTargets,
limits: (usize, usize),
scan: &mut dyn FnMut(&mut dyn FnMut(&K, &V) -> DbResult<()>) -> DbResult<()>,
) -> DbResult<u64> {
let mut expected: HashMap<(usize, Vec<u8>), u128> = HashMap::new();
scan(&mut |k, v| {
let mut sink = CounterSink::new(targets, limits.0, limits.1);
project(k, v, &mut sink)?;
for (slot, w) in sink.into_acc() {
let e = expected.entry(slot).or_insert(0u128);
*e = e
.checked_add(w)
.ok_or(DbError::Internal("recount sum overflow"))?;
}
Ok(())
})?;
let mut repaired = 0u64;
for (gi, group) in targets.groups().iter().enumerate() {
let mut existing: HashMap<Vec<u8>, u64> = HashMap::new();
for member in &group.members {
member.for_each_counter_row(&mut |key, stored| {
existing.insert(key.to_vec(), stored);
Ok(())
})?;
}
let mut keys: std::collections::HashSet<Vec<u8>> = existing.keys().cloned().collect();
keys.extend(
expected
.keys()
.filter(|(g, _)| *g == gi)
.map(|(_, key)| key.clone()),
);
for key in keys {
let member = match targets.resolve_group(gi, &key) {
Ok(Some(m)) => m,
Ok(None) => continue,
Err(DbError::CounterTargetMissing { targets, key }) => {
tracing::warn!(
group = gi,
targets = %targets,
key = %key,
"recount: dangling contribution — no live target row holds this key, skipping (see schema::validate)"
);
continue;
}
Err(e) => return Err(e),
};
let want_raw = expected.get(&(gi, key.clone())).copied().unwrap_or(0);
let want = u64::try_from(want_raw).map_err(|_| DbError::CounterOverflow {
collection: member.collection_name(),
field: member.field(),
key: hex(&key),
})?;
let stored = existing.get(&key).copied().unwrap_or(0);
if want != stored {
let delta = i128::from(want) - i128::from(stored);
match member.apply_delta(&key, delta) {
Ok(true) => repaired += 1,
Ok(false) => {}
Err(DbError::CounterTargetMissing { targets, key }) => {
tracing::warn!(
collection = member.collection_name(),
field = member.field(),
targets = %targets,
key = %key,
"recount: dangling contribution — target row missing, skipping (see schema::validate)"
);
}
Err(e) => return Err(e),
}
}
}
}
for group in targets.groups() {
for member in &group.members {
member.durable_flush()?;
}
}
Ok(repaired)
}
#[cfg(test)]
pub(crate) mod tests {
use super::*;
use crate::projection::{ProjectableIndex, sealed};
use crate::txn::participant::{TxBackend, TxParticipant};
pub(crate) struct StubCounterColl {
name: &'static str,
state: Arc<std::sync::Mutex<Option<u64>>>,
}
impl StubCounterColl {
fn with(initial: Option<u64>) -> Self {
Self::named("stub_counter", initial)
}
pub(crate) fn named(name: &'static str, initial: Option<u64>) -> Self {
Self {
name,
state: Arc::new(std::sync::Mutex::new(initial)),
}
}
fn insert(&self, _key: &[u8; 8]) {
*self.state.lock().unwrap() = Some(0);
}
}
impl sealed::Sealed for StubCounterColl {}
impl ProjectableIndex for StubCounterColl {
fn collection_name(&self) -> &'static str {
self.name
}
fn put_bytes(&self, _key: &[u8], value: &[u8]) -> DbResult<()> {
let bytes: [u8; 8] = value
.try_into()
.map_err(|_| DbError::Internal("bad counter width"))?;
*self.state.lock().unwrap() = Some(u64::from_le_bytes(bytes));
Ok(())
}
fn delete_bytes(&self, _key: &[u8]) -> DbResult<()> {
*self.state.lock().unwrap() = None;
Ok(())
}
fn values_equal(&self, a: &[u8], b: &[u8]) -> DbResult<bool> {
Ok(a == b)
}
fn for_each_bytes(&self, f: &mut dyn FnMut(&[u8], &[u8]) -> DbResult<()>) -> DbResult<()> {
if let Some(n) = *self.state.lock().unwrap() {
f(&[0u8; 8], &n.to_le_bytes())?;
}
Ok(())
}
fn durable_flush(&self) -> DbResult<()> {
Ok(())
}
}
impl IndexKeyed for StubCounterColl {
type K = [u8; 8];
type V = u64;
fn encode_value(&self, v: &u64) -> DbResult<Vec<u8>> {
Ok(v.to_le_bytes().to_vec())
}
}
impl CounterCollection for StubCounterColl {
fn decode_value(&self, bytes: &[u8]) -> DbResult<u64> {
let arr: [u8; 8] = bytes
.try_into()
.map_err(|_| DbError::Internal("bad counter width"))?;
Ok(u64::from_le_bytes(arr))
}
fn contains_bytes(&self, _key: &[u8]) -> DbResult<bool> {
Ok(self.state.lock().unwrap().is_some())
}
fn read_bytes(&self, _key: &[u8]) -> DbResult<Option<Vec<u8>>> {
Ok(self.state.lock().unwrap().map(|n| n.to_le_bytes().to_vec()))
}
fn upsert_counter(
&self,
_key: &[u8],
f: &mut dyn FnMut(Option<&u64>) -> Upsert<u64>,
) -> DbResult<()> {
let mut guard = self.state.lock().unwrap();
match f(guard.as_ref()) {
Upsert::Put(v) => *guard = Some(v),
Upsert::Keep => {}
Upsert::Delete => *guard = None,
}
Ok(())
}
}
struct StubTxParticipant;
impl TxParticipant for StubTxParticipant {
fn name(&self) -> &'static str {
"stub_counter"
}
fn backend(&self) -> TxBackend {
TxBackend::Bitcask
}
fn schema_fingerprint(&self) -> u64 {
0
}
fn encrypted(&self) -> bool {
false
}
fn max_entry_bytes(&self) -> Option<usize> {
None
}
fn shard_of(&self, _key: &[u8]) -> DbResult<usize> {
Ok(0)
}
fn lock_shards<'a>(
&'a self,
_shards: &[usize],
) -> DbResult<Box<dyn crate::txn::participant::TxLocked + 'a>> {
Err(DbError::Internal("stub"))
}
}
impl ProjectableIndexTx for StubCounterColl {
fn as_tx_participant(self: Arc<Self>) -> Arc<dyn TxParticipant> {
Arc::new(StubTxParticipant)
}
}
fn stub_target(initial: Option<u64>) -> TypedCounterTarget<StubCounterColl> {
TypedCounterTarget::new(
Arc::new(StubCounterColl::with(initial)),
"count",
OnMissingTarget::Create,
|v: &u64| *v,
|_cur: Option<&u64>, n: u64| {
if n == 0 {
Upsert::Delete
} else {
Upsert::Put(n)
}
},
)
}
#[test]
fn counter_resolve_adds_and_subtracts() {
let t = stub_target(Some(5));
let CounterApply::Put(bytes) = t.resolve(Some(&5u64.to_le_bytes()), 3).unwrap() else {
panic!("expected Put");
};
assert_eq!(u64::from_le_bytes(bytes.try_into().unwrap()), 8);
}
#[test]
fn counter_resolve_zero_deletes() {
let t = stub_target(Some(1));
assert!(matches!(
t.resolve(Some(&1u64.to_le_bytes()), -1).unwrap(),
CounterApply::Delete
));
}
#[test]
fn counter_resolve_underflow_is_an_error_not_saturation() {
let t = stub_target(Some(1));
let err = t.resolve(Some(&1u64.to_le_bytes()), -2).unwrap_err();
assert!(matches!(err, DbError::CounterOverflow { .. }));
}
#[test]
fn counter_resolve_overflow_is_an_error() {
let t = stub_target(Some(u64::MAX));
let err = t.resolve(Some(&u64::MAX.to_le_bytes()), 1).unwrap_err();
assert!(matches!(err, DbError::CounterOverflow { .. }));
}
#[test]
fn counter_resolve_missing_row_follows_the_declared_policy() {
let t = TypedCounterTarget::new(
Arc::new(StubCounterColl::with(None)),
"count",
OnMissingTarget::Error,
|v: &u64| *v,
|_c, n| Upsert::Put(n),
);
assert!(matches!(
t.resolve(None, 1).unwrap_err(),
DbError::CounterTargetMissing { .. }
));
let t = TypedCounterTarget::new(
Arc::new(StubCounterColl::with(None)),
"count",
OnMissingTarget::Skip,
|v: &u64| *v,
|_c, n| Upsert::Put(n),
);
assert!(matches!(t.resolve(None, 1).unwrap(), CounterApply::Skip));
let t = stub_target(None);
assert!(matches!(t.resolve(None, 2).unwrap(), CounterApply::Put(_)));
}
#[test]
fn counter_resolve_rejects_delete_at_a_non_zero_count() {
let t = TypedCounterTarget::new(
Arc::new(StubCounterColl::with(Some(4))),
"count",
OnMissingTarget::Create,
|v: &u64| *v,
|_c, _n| Upsert::Delete, );
assert!(matches!(
t.resolve(Some(&4u64.to_le_bytes()), 1).unwrap_err(),
DbError::Internal(_)
));
}
#[test]
fn counter_resolve_keep_erases_to_skip() {
let t = TypedCounterTarget::new(
Arc::new(StubCounterColl::with(Some(5))),
"count",
OnMissingTarget::Create,
|v: &u64| *v,
|_cur, _n| Upsert::Keep, );
assert!(matches!(
t.resolve(Some(&5u64.to_le_bytes()), 3).unwrap(),
CounterApply::Skip
));
}
#[test]
fn counter_one_of_with_no_members_yields_empty_targets() {
let targets = CounterTargets::new().one_of().build();
assert!(targets.is_empty());
assert_eq!(targets.len(), 0);
}
pub(crate) fn single_stub_targets() -> (CounterTargets, Arc<StubCounterColl>) {
let coll = Arc::new(StubCounterColl::named("declared_counter", None));
let targets = CounterTargets::new().with_counter(
&coll,
"count",
OnMissingTarget::Create,
|v: &u64| *v,
|_cur: Option<&u64>, n: u64| Upsert::Put(n),
);
(targets, coll)
}
fn one_of_stub_targets() -> (CounterTargets, Arc<StubCounterColl>, Arc<StubCounterColl>) {
let a = Arc::new(StubCounterColl::named("stub_a", None));
let b = Arc::new(StubCounterColl::named("stub_b", None));
let targets = CounterTargets::new()
.one_of()
.with(
&a,
"count",
OnMissingTarget::Create,
|v: &u64| *v,
|_cur: Option<&u64>, n: u64| Upsert::Put(n),
)
.with(
&b,
"count",
OnMissingTarget::Create,
|v: &u64| *v,
|_cur: Option<&u64>, n: u64| Upsert::Put(n),
)
.build();
(targets, a, b)
}
#[test]
fn counter_sink_sums_duplicate_contributions() {
let (targets, coll) = single_stub_targets();
let mut sink = CounterSink::new(&targets, 16, 1024);
sink.add(&coll, [1u8; 8], 2).unwrap();
sink.add(&coll, [1u8; 8], 3).unwrap();
assert_eq!(*sink.contributions().values().next().unwrap(), 5u128);
}
#[test]
fn counter_sink_charges_budget_before_coalescing() {
let (targets, coll) = single_stub_targets();
let mut sink = CounterSink::new(&targets, 2, 1024); sink.add(&coll, [1u8; 8], 1).unwrap();
sink.add(&coll, [1u8; 8], 1).unwrap();
let err = sink.add(&coll, [1u8; 8], 1).unwrap_err(); assert!(matches!(err, DbError::ProjectionBudgetExceeded { .. }));
}
#[test]
fn counter_sink_rejects_an_undeclared_target() {
let (targets, _declared) = single_stub_targets();
let other = Arc::new(StubCounterColl::with(Some(0))); let mut sink = CounterSink::new(&targets, 16, 1024);
assert!(matches!(
sink.add(&other, [1u8; 8], 1).unwrap_err(),
DbError::UndeclaredCounterTarget { .. }
));
}
#[test]
fn counter_one_of_group_requires_exactly_one_holder() {
let (targets, a, b) = one_of_stub_targets(); let key = [9u8; 8];
assert!(matches!(
targets.resolve_group(0, &key).unwrap_err(),
DbError::CounterTargetMissing { .. }
));
a.insert(&key);
assert!(targets.resolve_group(0, &key).unwrap().is_some());
b.insert(&key);
assert!(matches!(
targets.resolve_group(0, &key).unwrap_err(),
DbError::AmbiguousCounterTarget { .. }
));
}
#[test]
#[should_panic(expected = "already has a declared counter target")]
fn counter_targets_rejects_a_second_target_on_the_same_collection() {
let coll = Arc::new(StubCounterColl::named("dup_counter", None));
let _ = CounterTargets::new()
.with_counter(
&coll,
"count",
OnMissingTarget::Create,
|v: &u64| *v,
|_cur: Option<&u64>, n: u64| Upsert::Put(n),
)
.with_counter(
&coll,
"other_field",
OnMissingTarget::Create,
|v: &u64| *v,
|_cur: Option<&u64>, n: u64| Upsert::Put(n),
);
}
#[test]
#[should_panic(expected = "at most one one_of group")]
fn counter_targets_rejects_a_second_one_of_group() {
let a = Arc::new(StubCounterColl::named("one_of_a", None));
let b = Arc::new(StubCounterColl::named("one_of_b", None));
let c = Arc::new(StubCounterColl::named("one_of_c", None));
let d = Arc::new(StubCounterColl::named("one_of_d", None));
let _ = CounterTargets::new()
.one_of()
.with(
&a,
"count",
OnMissingTarget::Create,
|v: &u64| *v,
|_cur: Option<&u64>, n: u64| Upsert::Put(n),
)
.with(
&b,
"count",
OnMissingTarget::Create,
|v: &u64| *v,
|_cur: Option<&u64>, n: u64| Upsert::Put(n),
)
.build()
.one_of()
.with(
&c,
"count",
OnMissingTarget::Create,
|v: &u64| *v,
|_cur: Option<&u64>, n: u64| Upsert::Put(n),
)
.with(
&d,
"count",
OnMissingTarget::Create,
|v: &u64| *v,
|_cur: Option<&u64>, n: u64| Upsert::Put(n),
)
.build();
}
}