use crate::Key;
use crate::error::{DbError, DbResult};
#[cfg(feature = "typed-tree")]
use crate::projection::ProjectionSink;
use std::mem::size_of;
pub trait WriteHook<K: Key>: Send + Sync {
const WATCHES: bool = false;
fn watch_channel(&self) -> Option<&std::sync::Arc<crate::watch::WatchChannel<K>>> {
None
}
fn watch_old(&self) -> bool {
false
}
const NEEDS_OLD_VALUE: bool = true;
const NEEDS_INIT: bool = false;
const NEEDS_WRITE: bool = true;
const EXPIRES: bool = false;
const TTL_REVISION: u32 = 0;
fn deadline(&self, _key: &K, _value: &[u8]) -> Option<u64> {
None
}
fn ttl_decode_failed(&self, _value: &[u8]) -> bool {
false
}
fn on_write(&self, key: &K, old: Option<&[u8]>, new: Option<&[u8]>);
fn on_init(&self, _key: &K, _value: &[u8]) {}
#[cfg(feature = "typed-tree")]
const PROJECTS: bool = false;
#[cfg(feature = "typed-tree")]
fn project(
&self,
_key: &K,
_old: Option<&[u8]>,
_new: Option<&[u8]>,
_sink: &mut ProjectionSink<'_>,
) -> DbResult<()> {
Ok(())
}
#[cfg(feature = "typed-tree")]
fn projection_targets(&self) -> Option<&crate::projection::IndexTargets> {
None
}
#[cfg(feature = "typed-tree")]
fn projection_limits(&self) -> (usize, usize) {
(
crate::projection::DEFAULT_MAX_PROJECTED_ROWS,
crate::projection::DEFAULT_MAX_PROJECTED_BYTES,
)
}
#[cfg(feature = "typed-tree")]
const COUNTS: bool = false;
#[cfg(feature = "typed-tree")]
fn counter_project(
&self,
_key: &K,
_old: Option<&[u8]>,
_new: Option<&[u8]>,
_sink: &mut crate::counter::CounterSink<'_>,
) -> DbResult<()> {
Ok(())
}
#[cfg(feature = "typed-tree")]
fn counter_targets(&self) -> Option<&crate::counter::CounterTargets> {
None
}
#[cfg(feature = "typed-tree")]
const CLAIMS: bool = false;
#[cfg(feature = "typed-tree")]
fn claim_project(
&self,
_key: &K,
_old: Option<&[u8]>,
_new: Option<&[u8]>,
_sink: &mut crate::unique::ClaimSink<'_>,
) -> DbResult<()> {
Ok(())
}
#[cfg(feature = "typed-tree")]
fn claim_targets(&self) -> Option<&crate::unique::UniqueTargets> {
None
}
#[cfg(all(feature = "typed-tree", feature = "armour"))]
const EMITS: bool = false;
#[cfg(all(feature = "typed-tree", feature = "armour"))]
fn stream_project(
&self,
_key: &K,
_old: Option<&[u8]>,
_new: Option<&[u8]>,
_sink: &mut crate::EmitSink<'_>,
) -> DbResult<()> {
Ok(())
}
#[cfg(all(feature = "typed-tree", feature = "armour"))]
fn stream_targets(&self) -> Option<&crate::QueueTargets> {
None
}
}
pub struct NoHook;
impl<K: Key> WriteHook<K> for NoHook {
const NEEDS_OLD_VALUE: bool = false;
const NEEDS_INIT: bool = false;
const NEEDS_WRITE: bool = false;
#[inline(always)]
fn on_write(&self, _key: &K, _old: Option<&[u8]>, _new: Option<&[u8]>) {}
#[inline(always)]
fn on_init(&self, _key: &K, _value: &[u8]) {}
#[cfg(feature = "typed-tree")]
const PROJECTS: bool = false;
#[cfg(feature = "typed-tree")]
fn project(
&self,
_key: &K,
_old: Option<&[u8]>,
_new: Option<&[u8]>,
_sink: &mut ProjectionSink<'_>,
) -> DbResult<()> {
Ok(())
}
#[cfg(feature = "typed-tree")]
const COUNTS: bool = false;
#[cfg(feature = "typed-tree")]
const CLAIMS: bool = false;
#[cfg(all(feature = "typed-tree", feature = "armour"))]
const EMITS: bool = false;
}
#[cfg(feature = "typed-tree")]
pub trait TypedWriteHook<K: Key, T>: Send + Sync {
const WATCHES: bool = false;
fn watch_channel(&self) -> Option<&std::sync::Arc<crate::watch::WatchChannel<K>>> {
None
}
fn watch_old(&self) -> bool {
false
}
const NEEDS_OLD_VALUE: bool = true;
const NEEDS_INIT: bool = false;
const NEEDS_WRITE: bool = true;
const EXPIRES: bool = false;
const TTL_REVISION: u32 = 0;
fn deadline(&self, _key: &K, _value: &T) -> Option<u64> {
None
}
fn on_write(&self, key: &K, old: Option<&T>, new: Option<&T>);
fn on_init(&self, _key: &K, _value: &T) {}
const PROJECTS: bool = false;
fn project(&self, _key: &K, _value: &T, _sink: &mut ProjectionSink<'_>) -> DbResult<()> {
Ok(())
}
fn projection_targets(&self) -> Option<&crate::projection::IndexTargets> {
None
}
fn projection_limits(&self) -> (usize, usize) {
(
crate::projection::DEFAULT_MAX_PROJECTED_ROWS,
crate::projection::DEFAULT_MAX_PROJECTED_BYTES,
)
}
const COUNTS: bool = false;
fn counter_project(
&self,
_key: &K,
_value: &T,
_sink: &mut crate::counter::CounterSink<'_>,
) -> DbResult<()> {
Ok(())
}
fn counter_targets(&self) -> Option<&crate::counter::CounterTargets> {
None
}
const CLAIMS: bool = false;
fn claim_project(
&self,
_key: &K,
_value: &T,
_sink: &mut crate::unique::ClaimSink<'_>,
) -> DbResult<()> {
Ok(())
}
fn claim_targets(&self) -> Option<&crate::unique::UniqueTargets> {
None
}
#[cfg(feature = "armour")]
const EMITS: bool = false;
#[cfg(feature = "armour")]
fn stream_project(
&self,
_key: &K,
_old: Option<&T>,
_new: Option<&T>,
_sink: &mut crate::EmitSink<'_>,
) -> DbResult<()> {
Ok(())
}
#[cfg(feature = "armour")]
fn stream_targets(&self) -> Option<&crate::QueueTargets> {
None
}
}
#[cfg(feature = "typed-tree")]
impl<K: Key, T> TypedWriteHook<K, T> for NoHook {
const NEEDS_OLD_VALUE: bool = false;
const NEEDS_INIT: bool = false;
const NEEDS_WRITE: bool = false;
#[inline(always)]
fn on_write(&self, _key: &K, _old: Option<&T>, _new: Option<&T>) {}
#[inline(always)]
fn on_init(&self, _key: &K, _value: &T) {}
const PROJECTS: bool = false;
fn project(&self, _key: &K, _value: &T, _sink: &mut ProjectionSink<'_>) -> DbResult<()> {
Ok(())
}
const COUNTS: bool = false;
const CLAIMS: bool = false;
}
#[cfg(feature = "typed-tree")]
pub struct ZeroHookAdapter<K, T, H> {
pub(crate) inner: H,
pub(crate) _marker: std::marker::PhantomData<fn() -> (K, T)>,
}
#[cfg(feature = "typed-tree")]
impl<K, T, H> ZeroHookAdapter<K, T, H> {
pub fn new(inner: H) -> Self {
Self {
inner,
_marker: std::marker::PhantomData,
}
}
}
#[cfg(feature = "typed-tree")]
unsafe impl<K, T, H: Send> Send for ZeroHookAdapter<K, T, H> {}
#[cfg(feature = "typed-tree")]
unsafe impl<K, T, H: Sync> Sync for ZeroHookAdapter<K, T, H> {}
#[cfg(feature = "typed-tree")]
impl<K: Key, T: Copy + zerocopy::FromBytes, H: TypedWriteHook<K, T>> WriteHook<K>
for ZeroHookAdapter<K, T, H>
{
const WATCHES: bool = H::WATCHES;
const NEEDS_OLD_VALUE: bool = H::NEEDS_OLD_VALUE;
const NEEDS_INIT: bool = H::NEEDS_INIT;
const NEEDS_WRITE: bool = H::NEEDS_WRITE;
const EXPIRES: bool = H::EXPIRES;
const TTL_REVISION: u32 = H::TTL_REVISION;
fn watch_channel(&self) -> Option<&std::sync::Arc<crate::watch::WatchChannel<K>>> {
self.inner.watch_channel()
}
fn watch_old(&self) -> bool {
self.inner.watch_old()
}
const PROJECTS: bool = H::PROJECTS;
const COUNTS: bool = H::COUNTS;
const CLAIMS: bool = H::CLAIMS;
#[cfg(feature = "armour")]
const EMITS: bool = H::EMITS;
fn deadline(&self, key: &K, value: &[u8]) -> Option<u64> {
debug_assert_eq!(value.len(), size_of::<T>());
let val: T = zerocopy::FromBytes::read_from_bytes(value).ok()?;
self.inner.deadline(key, &val)
}
#[inline]
fn on_write(&self, key: &K, old: Option<&[u8]>, new: Option<&[u8]>) {
if !H::NEEDS_WRITE {
return;
}
let read = |b: &[u8]| -> T {
debug_assert_eq!(b.len(), size_of::<T>());
zerocopy::FromBytes::read_from_bytes(b).expect("slice len == size_of::<T>()")
};
let old_val = old.map(read);
let new_val = new.map(read);
self.inner.on_write(key, old_val.as_ref(), new_val.as_ref());
}
#[inline]
fn on_init(&self, key: &K, value: &[u8]) {
debug_assert_eq!(value.len(), size_of::<T>());
let val: T =
zerocopy::FromBytes::read_from_bytes(value).expect("slice len == size_of::<T>()");
self.inner.on_init(key, &val);
}
fn project(
&self,
key: &K,
_old: Option<&[u8]>,
new: Option<&[u8]>,
sink: &mut ProjectionSink<'_>,
) -> DbResult<()> {
if !H::PROJECTS {
return Ok(());
}
if let Some(b) = new {
debug_assert_eq!(b.len(), size_of::<T>());
let val: T = zerocopy::FromBytes::read_from_bytes(b).map_err(|_| {
DbError::FormatMismatch("zero hook adapter project decode failed".into())
})?;
self.inner.project(key, &val, sink)?;
}
Ok(())
}
fn projection_targets(&self) -> Option<&crate::projection::IndexTargets> {
self.inner.projection_targets()
}
fn projection_limits(&self) -> (usize, usize) {
self.inner.projection_limits()
}
fn counter_project(
&self,
key: &K,
_old: Option<&[u8]>,
new: Option<&[u8]>,
sink: &mut crate::counter::CounterSink<'_>,
) -> DbResult<()> {
if !H::COUNTS {
return Ok(());
}
if let Some(b) = new {
debug_assert_eq!(b.len(), size_of::<T>());
let val: T = zerocopy::FromBytes::read_from_bytes(b).map_err(|_| {
DbError::FormatMismatch("zero hook adapter counter_project decode failed".into())
})?;
self.inner.counter_project(key, &val, sink)?;
}
Ok(())
}
fn counter_targets(&self) -> Option<&crate::counter::CounterTargets> {
self.inner.counter_targets()
}
fn claim_project(
&self,
key: &K,
_old: Option<&[u8]>,
new: Option<&[u8]>,
sink: &mut crate::unique::ClaimSink<'_>,
) -> DbResult<()> {
if !H::CLAIMS {
return Ok(());
}
if let Some(b) = new {
debug_assert_eq!(b.len(), size_of::<T>());
let val: T = zerocopy::FromBytes::read_from_bytes(b).map_err(|_| {
DbError::FormatMismatch("zero hook adapter claim_project decode failed".into())
})?;
self.inner.claim_project(key, &val, sink)?;
}
Ok(())
}
fn claim_targets(&self) -> Option<&crate::unique::UniqueTargets> {
self.inner.claim_targets()
}
#[cfg(feature = "armour")]
fn stream_project(
&self,
key: &K,
old: Option<&[u8]>,
new: Option<&[u8]>,
sink: &mut crate::EmitSink<'_>,
) -> DbResult<()> {
if !H::EMITS {
return Ok(());
}
let read = |b: &[u8]| -> Result<T, DbError> {
zerocopy::FromBytes::read_from_bytes(b).map_err(|_| {
DbError::FormatMismatch("zero hook adapter stream decode failed".into())
})
};
let old = old.map(read).transpose()?;
let new = new.map(read).transpose()?;
self.inner
.stream_project(key, old.as_ref(), new.as_ref(), sink)
}
#[cfg(feature = "armour")]
fn stream_targets(&self) -> Option<&crate::QueueTargets> {
self.inner.stream_targets()
}
}
#[cfg(all(feature = "typed-tree", feature = "var-collections"))]
pub struct VarTypedHookAdapter<K, T, C, H> {
pub(crate) inner: H,
pub(crate) codec: C,
pub(crate) _marker: std::marker::PhantomData<fn() -> (K, T)>,
}
#[cfg(all(feature = "typed-tree", feature = "var-collections"))]
impl<K, T, C, H> VarTypedHookAdapter<K, T, C, H> {
pub fn new(inner: H, codec: C) -> Self {
Self {
inner,
codec,
_marker: std::marker::PhantomData,
}
}
}
#[cfg(all(feature = "typed-tree", feature = "var-collections"))]
impl<K, T, C, H> VarTypedHookAdapter<K, T, C, H>
where
K: Key,
T: Send + Sync,
H: TypedWriteHook<K, T>,
{
pub(crate) fn typed_deadline(&self, key: &K, value: &T) -> Option<u64> {
if H::EXPIRES {
self.inner.deadline(key, value)
} else {
None
}
}
}
#[cfg(all(feature = "typed-tree", feature = "var-collections"))]
unsafe impl<K, T, C: Send, H: Send> Send for VarTypedHookAdapter<K, T, C, H> {}
#[cfg(all(feature = "typed-tree", feature = "var-collections"))]
unsafe impl<K, T, C: Sync, H: Sync> Sync for VarTypedHookAdapter<K, T, C, H> {}
#[cfg(all(feature = "typed-tree", feature = "var-collections"))]
impl<K, T, C, H> WriteHook<K> for VarTypedHookAdapter<K, T, C, H>
where
K: Key,
T: Send + Sync,
C: crate::codec::Codec<T>,
H: TypedWriteHook<K, T>,
{
const WATCHES: bool = H::WATCHES;
const NEEDS_OLD_VALUE: bool = H::NEEDS_OLD_VALUE;
const NEEDS_INIT: bool = H::NEEDS_INIT;
const NEEDS_WRITE: bool = H::NEEDS_WRITE;
const EXPIRES: bool = H::EXPIRES;
const TTL_REVISION: u32 = H::TTL_REVISION;
fn watch_channel(&self) -> Option<&std::sync::Arc<crate::watch::WatchChannel<K>>> {
self.inner.watch_channel()
}
fn watch_old(&self) -> bool {
self.inner.watch_old()
}
const PROJECTS: bool = H::PROJECTS;
const COUNTS: bool = H::COUNTS;
const CLAIMS: bool = H::CLAIMS;
#[cfg(feature = "armour")]
const EMITS: bool = H::EMITS;
fn deadline(&self, key: &K, value: &[u8]) -> Option<u64> {
self.codec
.decode_from(value)
.ok()
.and_then(|val| self.inner.deadline(key, &val))
}
fn ttl_decode_failed(&self, value: &[u8]) -> bool {
self.codec.decode_from(value).is_err()
}
#[inline]
fn on_write(&self, key: &K, old: Option<&[u8]>, new: Option<&[u8]>) {
if !H::NEEDS_WRITE {
return;
}
let old_t = old.and_then(|b| {
self.codec
.decode_from(b)
.map_err(|_| {
tracing::warn!(
side = "old",
value_len = b.len(),
"var_typed hook adapter: on_write decode failed"
);
})
.ok()
});
let new_t = new.and_then(|b| {
self.codec
.decode_from(b)
.map_err(|_| {
tracing::warn!(
side = "new",
value_len = b.len(),
"var_typed hook adapter: on_write decode failed"
);
})
.ok()
});
self.inner.on_write(key, old_t.as_ref(), new_t.as_ref());
}
#[inline]
fn on_init(&self, key: &K, value: &[u8]) {
match self.codec.decode_from(value) {
Ok(val) => self.inner.on_init(key, &val),
Err(_) => {
tracing::warn!(
key_len = std::mem::size_of::<K>(),
value_len = value.len(),
"var_typed hook adapter: on_init decode failed, entry skipped"
);
}
}
}
fn project(
&self,
key: &K,
_old: Option<&[u8]>,
new: Option<&[u8]>,
sink: &mut ProjectionSink<'_>,
) -> DbResult<()> {
if !H::PROJECTS {
return Ok(());
}
if let Some(b) = new {
let val = self.codec.decode_from(b)?;
self.inner.project(key, &val, sink)?;
}
Ok(())
}
fn projection_targets(&self) -> Option<&crate::projection::IndexTargets> {
self.inner.projection_targets()
}
fn projection_limits(&self) -> (usize, usize) {
self.inner.projection_limits()
}
fn counter_project(
&self,
key: &K,
_old: Option<&[u8]>,
new: Option<&[u8]>,
sink: &mut crate::counter::CounterSink<'_>,
) -> DbResult<()> {
if !H::COUNTS {
return Ok(());
}
if let Some(b) = new {
let val = self.codec.decode_from(b)?;
self.inner.counter_project(key, &val, sink)?;
}
Ok(())
}
fn counter_targets(&self) -> Option<&crate::counter::CounterTargets> {
self.inner.counter_targets()
}
fn claim_project(
&self,
key: &K,
_old: Option<&[u8]>,
new: Option<&[u8]>,
sink: &mut crate::unique::ClaimSink<'_>,
) -> DbResult<()> {
if !H::CLAIMS {
return Ok(());
}
if let Some(b) = new {
let val = self.codec.decode_from(b)?;
self.inner.claim_project(key, &val, sink)?;
}
Ok(())
}
fn claim_targets(&self) -> Option<&crate::unique::UniqueTargets> {
self.inner.claim_targets()
}
#[cfg(feature = "armour")]
fn stream_project(
&self,
key: &K,
old: Option<&[u8]>,
new: Option<&[u8]>,
sink: &mut crate::EmitSink<'_>,
) -> DbResult<()> {
if !H::EMITS {
return Ok(());
}
let old = old.map(|bytes| self.codec.decode_from(bytes)).transpose()?;
let new = new.map(|bytes| self.codec.decode_from(bytes)).transpose()?;
self.inner
.stream_project(key, old.as_ref(), new.as_ref(), sink)
}
#[cfg(feature = "armour")]
fn stream_targets(&self) -> Option<&crate::QueueTargets> {
self.inner.stream_targets()
}
}
#[cfg(all(test, feature = "typed-tree", feature = "armour"))]
mod tests {
use super::*;
use crate::projection::Projected;
use std::sync::Arc;
#[test]
fn zero_hook_adapter_forwards_counts() {
type Inner = Projected<[u8; 8], u64>;
assert!(<ZeroHookAdapter<[u8; 8], u64, Inner> as WriteHook<[u8; 8]>>::COUNTS);
}
#[cfg(all(feature = "var-collections", feature = "rapira-codec"))]
#[test]
fn var_typed_hook_adapter_forwards_counts() {
type Inner = Projected<[u8; 8], u64>;
assert!(
<VarTypedHookAdapter<[u8; 8], u64, crate::codec::RapiraCodec, Inner> as WriteHook<
[u8; 8],
>>::COUNTS
);
}
#[test]
fn zero_hook_adapter_forwards_claims() {
type Inner = Projected<[u8; 8], u64>;
assert!(<ZeroHookAdapter<[u8; 8], u64, Inner> as WriteHook<[u8; 8]>>::CLAIMS);
}
#[cfg(all(feature = "var-collections", feature = "rapira-codec"))]
#[test]
fn var_typed_hook_adapter_forwards_claims() {
type Inner = Projected<[u8; 8], u64>;
assert!(
<VarTypedHookAdapter<[u8; 8], u64, crate::codec::RapiraCodec, Inner> as WriteHook<
[u8; 8],
>>::CLAIMS
);
}
#[derive(Clone)]
struct SnapshotHook(Arc<std::sync::Mutex<Vec<(Option<u64>, Option<u64>)>>>);
impl TypedWriteHook<[u8; 8], u64> for SnapshotHook {
const EMITS: bool = true;
fn on_write(&self, _: &[u8; 8], _: Option<&u64>, _: Option<&u64>) {}
fn stream_project(
&self,
_: &[u8; 8],
old: Option<&u64>,
new: Option<&u64>,
_: &mut crate::EmitSink<'_>,
) -> DbResult<()> {
self.0.lock().unwrap().push((old.copied(), new.copied()));
Ok(())
}
}
#[test]
fn zero_hook_adapter_decodes_and_forwards_each_stream_snapshot_once() {
let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
let hook = ZeroHookAdapter::new(SnapshotHook(Arc::clone(&seen)));
let targets = crate::QueueTargets::new();
let mut sink = crate::EmitSink::new(&targets, 1, 1);
let old = 7u64.to_ne_bytes();
let new = 9u64.to_ne_bytes();
hook.stream_project(&[0; 8], Some(&old), Some(&new), &mut sink)
.unwrap();
assert_eq!(*seen.lock().unwrap(), vec![(Some(7), Some(9))]);
}
#[cfg(all(feature = "var-collections", feature = "rapira-codec"))]
#[test]
fn var_hook_adapter_decodes_and_forwards_each_stream_snapshot_once() {
let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
let hook =
VarTypedHookAdapter::new(SnapshotHook(Arc::clone(&seen)), crate::codec::RapiraCodec);
let targets = crate::QueueTargets::new();
let mut sink = crate::EmitSink::new(&targets, 1, 1);
let old = rapira::serialize(&7u64);
let new = rapira::serialize(&9u64);
hook.stream_project(&[0; 8], Some(&old), Some(&new), &mut sink)
.unwrap();
assert_eq!(*seen.lock().unwrap(), vec![(Some(7), Some(9))]);
}
#[cfg(feature = "armour")]
mod claim_forwarding {
use std::sync::Arc;
use crate::projection::IndexTargets;
use crate::unique::{ClaimSink, UniqueTargets};
use crate::unique_tests::{ClaimFwdKey, ClaimFwdRow, open_claim_ns};
use super::*;
#[test]
fn zero_hook_adapter_forwards_a_claim_end_to_end() {
let (_dir, _db, ns) = open_claim_ns();
let targets = UniqueTargets::new().with(&ns);
let ns_for_project = Arc::clone(&ns);
let inner =
Projected::<ClaimFwdKey, ClaimFwdRow>::new(IndexTargets::new(), |_, _, _| Ok(()))
.and_unique(targets.clone(), move |_k, _v: &ClaimFwdRow, sink| {
sink.claim(&ns_for_project, ClaimFwdKey::new(7))
});
let adapter = ZeroHookAdapter::new(inner);
let bytes = zerocopy::IntoBytes::as_bytes(&ClaimFwdRow(1)).to_vec();
let mut sink = ClaimSink::new(&targets, 16, 1024);
adapter
.claim_project(&ClaimFwdKey::new(0), None, Some(&bytes), &mut sink)
.expect("claim_project");
assert_eq!(
sink.claims().len(),
1,
"the inner Projected's claim must reach the adapter's caller"
);
}
#[cfg(all(feature = "var-collections", feature = "rapira-codec"))]
#[test]
fn var_typed_hook_adapter_forwards_a_claim_end_to_end() {
let (_dir, _db, ns) = open_claim_ns();
let targets = UniqueTargets::new().with(&ns);
let ns_for_project = Arc::clone(&ns);
let inner =
Projected::<ClaimFwdKey, ClaimFwdRow>::new(IndexTargets::new(), |_, _, _| Ok(()))
.and_unique(targets.clone(), move |_k, _v: &ClaimFwdRow, sink| {
sink.claim(&ns_for_project, ClaimFwdKey::new(7))
});
let adapter = VarTypedHookAdapter::new(inner, crate::RapiraCodec::default());
let bytes = rapira::serialize(&ClaimFwdRow(1));
let mut sink = ClaimSink::new(&targets, 16, 1024);
adapter
.claim_project(&ClaimFwdKey::new(0), None, Some(&bytes), &mut sink)
.expect("claim_project");
assert_eq!(
sink.claims().len(),
1,
"the inner Projected's claim must reach the adapter's caller"
);
}
}
}