use std::sync::Arc;
use crate::byte_view::ByteView;
use crate::ctx::Family;
use crate::error::{DbError, DbResult};
use crate::hook::WriteHook;
use crate::key::Key;
use crate::txlog::record::{OpKind, VersionTag};
use crate::txn::held::{TxFamily, TxHeld, TxRaw};
use crate::txn::participant::{
ClaimOp, DeferredReplay, PlannedOp, TxBackend, TxLocked, TxParticipant,
};
use crate::var_map::{VarMap, VarMapShard};
use crate::var_tree::{VarShard, VarTree};
use crate::watch::WatchWakeBatch;
pub(crate) struct VarTreeDeferred<K: Key + 'static, H: WriteHook<K> + 'static> {
tree: Arc<VarTree<K, H>>,
events: Vec<(K, Option<ByteView>, Option<ByteView>)>,
}
impl<K: Key + 'static, H: WriteHook<K> + 'static> DeferredReplay for VarTreeDeferred<K, H> {
fn event_count(&self) -> usize {
self.events.len()
}
fn replay_one(&mut self, local: usize) {
if let Some(ev) = self.events.get(local) {
self.tree.fire_owned(ev);
}
}
}
pub(crate) struct VarMapDeferred<
K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
H: WriteHook<K> + 'static,
> {
map: Arc<VarMap<K, H>>,
events: Vec<(K, Option<ByteView>, Option<ByteView>)>,
}
impl<K: Key + Send + Sync + std::hash::Hash + Eq + 'static, H: WriteHook<K> + 'static>
DeferredReplay for VarMapDeferred<K, H>
{
fn event_count(&self) -> usize {
self.events.len()
}
fn replay_one(&mut self, local: usize) {
if let Some(ev) = self.events.get(local) {
self.map.fire_owned(ev);
}
}
}
impl<K: Key + 'static, H: WriteHook<K> + 'static> TxFamily for VarTree<K, H> {
type Shard<'a>
= VarShard<'a, K, H>
where
Self: 'a;
type Deferred<'a>
= VarTreeDeferred<K, H>
where
Self: 'a;
type ReleaseExtra = ();
fn tx_shard_count(&self) -> usize {
self.shard_count()
}
fn tx_shard_of(&self, key: &[u8]) -> usize {
self.shard_for(&K::from_bytes(key))
}
fn tx_make_shard(&self, sid: usize) -> Self::Shard<'_> {
self.tx_shard(sid)
}
fn tx_release(shard: Self::Shard<'_>) -> (Vec<<Self as Family>::Owned>, WatchWakeBatch, ()) {
let (events, wakes, guard) = Self::tx_release_shard(shard);
drop(guard);
(events, wakes, ())
}
fn tx_deferred(
&self,
events: Vec<<Self as Family>::Owned>,
(): (),
) -> DbResult<Box<dyn DeferredReplay>> {
Ok(Box::new(VarTreeDeferred {
tree: self.tx_arc()?,
events,
}))
}
}
impl<K: Key + 'static, H: WriteHook<K> + 'static> TxRaw for VarTree<K, H> {
fn raw_read(&self, shard: &mut Self::Shard<'_>, key: &[u8]) -> DbResult<Option<Vec<u8>>> {
Ok(shard.try_get(&K::from_bytes(key))?.map(|v| v.to_vec()))
}
fn raw_contains(&self, shard: &mut Self::Shard<'_>, key: &[u8]) -> DbResult<bool> {
shard.try_contains(&K::from_bytes(key))
}
fn raw_put(&self, shard: &mut Self::Shard<'_>, key: &[u8], value: &[u8]) -> DbResult<()> {
shard.put(&K::from_bytes(key), value).map(drop)
}
fn raw_delete(&self, shard: &mut Self::Shard<'_>, key: &[u8]) -> DbResult<()> {
shard.delete(&K::from_bytes(key)).map(drop)
}
fn reserve_in_shard(
&self,
_shard: &mut Self::Shard<'_>,
ops: &[PlannedOp<'_>],
) -> DbResult<Vec<VersionTag>> {
let base = self.tx_reserve_gsns(ops.len());
Ok((0..ops.len())
.map(|i| VersionTag::Bitcask {
gsn: base + i as u64,
})
.collect())
}
fn arm_tag(shard: &mut Self::Shard<'_>, tag: VersionTag) -> DbResult<()> {
let VersionTag::Bitcask { gsn } = tag else {
return Err(DbError::Internal(
"Fixed version tag on a Bitcask participant",
));
};
shard.tx_arm_gsn(gsn);
Ok(())
}
fn disarm_tag(shard: &mut Self::Shard<'_>) {
shard.tx_clear_armed_gsn();
}
fn release_reservation(_shard: &mut Self::Shard<'_>, _tag: VersionTag) {
}
fn fence_shard(shard: &mut Self::Shard<'_>, tag: VersionTag) -> Option<u64> {
match tag {
VersionTag::Bitcask { gsn } => Some(shard.tx_fence(gsn)),
VersionTag::Fixed { .. } => None,
}
}
fn buffered_events(shard: &Self::Shard<'_>) -> usize {
shard.tx_buffered_events()
}
fn needs_replay(
&self,
_shard: &mut Self::Shard<'_>,
key: &[u8],
_kind: OpKind,
tag: VersionTag,
) -> DbResult<bool> {
let VersionTag::Bitcask { gsn } = tag else {
return Err(DbError::Internal(
"Fixed version tag on a Bitcask participant",
));
};
let sid = <Self as TxFamily>::tx_shard_of(self, key);
Ok(gsn > self.tx_durable_recovered_gsn(sid))
}
}
impl<K: Key + 'static, H: WriteHook<K> + 'static> TxParticipant for VarTree<K, H> {
fn name(&self) -> &'static str {
self.name()
}
fn backend(&self) -> TxBackend {
TxBackend::Bitcask
}
fn schema_fingerprint(&self) -> u64 {
self.typ_hash()
}
fn encrypted(&self) -> bool {
self.tx_encrypted()
}
fn max_entry_bytes(&self) -> Option<usize> {
Some(self.config().write_buffer_size)
}
fn shard_of(&self, key: &[u8]) -> DbResult<usize> {
Ok(<Self as TxFamily>::tx_shard_of(self, key))
}
fn shard_count(&self) -> usize {
<Self as TxFamily>::tx_shard_count(self)
}
fn lock_shards<'a>(&'a self, shards: &[usize]) -> DbResult<Box<dyn TxLocked + 'a>> {
Ok(Box::new(TxHeld::lock(self, shards)))
}
fn tx_project(
&self,
key: &[u8],
old: Option<&[u8]>,
new: Option<&[u8]>,
) -> DbResult<Vec<crate::txn::participant::TxDerivedWrite>> {
self.tx_project_writes(key, old, new)
}
fn tx_stream_writes(
&self,
key: &[u8],
old: Option<&[u8]>,
new: Option<&[u8]>,
) -> DbResult<crate::txn::participant::TxStreamWrites> {
self.tx_stream_writes(key, old, new)
}
fn tx_counter_deltas(
&self,
key: &[u8],
old: Option<&[u8]>,
new: Option<&[u8]>,
) -> DbResult<Vec<crate::txn::participant::TxCounterDelta>> {
self.tx_counter_deltas_for(key, old, new)
}
fn tx_claims(
&self,
key: &[u8],
old: Option<&[u8]>,
new: Option<&[u8]>,
) -> DbResult<Vec<ClaimOp>> {
self.tx_claims_for(self.name(), key, old, new)
}
fn tx_preplan_projection_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
self.tx_preplan_projection_targets()
}
fn tx_preplan_stream_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
Ok(VarTree::tx_preplan_stream_targets(self))
}
fn tx_prepare_streams(&self) -> DbResult<Vec<crate::queue::retention::ProducerPermit>> {
VarTree::tx_prepare_streams(self)
}
}
impl<K: Key + Send + Sync + std::hash::Hash + Eq + 'static, H: WriteHook<K> + 'static> TxFamily
for VarMap<K, H>
{
type Shard<'a>
= VarMapShard<'a, K, H>
where
Self: 'a;
type Deferred<'a>
= VarMapDeferred<K, H>
where
Self: 'a;
type ReleaseExtra = ();
fn tx_shard_count(&self) -> usize {
self.shard_count()
}
fn tx_shard_of(&self, key: &[u8]) -> usize {
self.shard_for(&K::from_bytes(key))
}
fn tx_make_shard(&self, sid: usize) -> Self::Shard<'_> {
self.tx_shard(sid)
}
fn tx_release(shard: Self::Shard<'_>) -> (Vec<<Self as Family>::Owned>, WatchWakeBatch, ()) {
let (events, wakes) = Self::tx_release_shard(shard);
(events, wakes, ())
}
fn tx_deferred(
&self,
events: Vec<<Self as Family>::Owned>,
(): (),
) -> DbResult<Box<dyn DeferredReplay>> {
Ok(Box::new(VarMapDeferred {
map: self.tx_arc()?,
events,
}))
}
}
impl<K: Key + Send + Sync + std::hash::Hash + Eq + 'static, H: WriteHook<K> + 'static> TxRaw
for VarMap<K, H>
{
fn raw_read(&self, shard: &mut Self::Shard<'_>, key: &[u8]) -> DbResult<Option<Vec<u8>>> {
Ok(shard.try_get(&K::from_bytes(key))?.map(|v| v.to_vec()))
}
fn raw_contains(&self, shard: &mut Self::Shard<'_>, key: &[u8]) -> DbResult<bool> {
shard.try_contains(&K::from_bytes(key))
}
fn raw_put(&self, shard: &mut Self::Shard<'_>, key: &[u8], value: &[u8]) -> DbResult<()> {
shard.put(&K::from_bytes(key), value).map(drop)
}
fn raw_delete(&self, shard: &mut Self::Shard<'_>, key: &[u8]) -> DbResult<()> {
shard.delete(&K::from_bytes(key)).map(drop)
}
fn reserve_in_shard(
&self,
_shard: &mut Self::Shard<'_>,
ops: &[PlannedOp<'_>],
) -> DbResult<Vec<VersionTag>> {
let base = self.tx_reserve_gsns(ops.len());
Ok((0..ops.len())
.map(|i| VersionTag::Bitcask {
gsn: base + i as u64,
})
.collect())
}
fn arm_tag(shard: &mut Self::Shard<'_>, tag: VersionTag) -> DbResult<()> {
let VersionTag::Bitcask { gsn } = tag else {
return Err(DbError::Internal(
"Fixed version tag on a Bitcask participant",
));
};
shard.tx_arm_gsn(gsn);
Ok(())
}
fn disarm_tag(shard: &mut Self::Shard<'_>) {
shard.tx_clear_armed_gsn();
}
fn release_reservation(_shard: &mut Self::Shard<'_>, _tag: VersionTag) {
}
fn fence_shard(shard: &mut Self::Shard<'_>, tag: VersionTag) -> Option<u64> {
match tag {
VersionTag::Bitcask { gsn } => Some(shard.tx_fence(gsn)),
VersionTag::Fixed { .. } => None,
}
}
fn buffered_events(shard: &Self::Shard<'_>) -> usize {
shard.tx_buffered_events()
}
fn needs_replay(
&self,
_shard: &mut Self::Shard<'_>,
key: &[u8],
_kind: OpKind,
tag: VersionTag,
) -> DbResult<bool> {
let VersionTag::Bitcask { gsn } = tag else {
return Err(DbError::Internal(
"Fixed version tag on a Bitcask participant",
));
};
let sid = <Self as TxFamily>::tx_shard_of(self, key);
Ok(gsn > self.tx_durable_recovered_gsn(sid))
}
}
impl<K: Key + Send + Sync + std::hash::Hash + Eq + 'static, H: WriteHook<K> + 'static> TxParticipant
for VarMap<K, H>
{
fn name(&self) -> &'static str {
self.name()
}
fn backend(&self) -> TxBackend {
TxBackend::Bitcask
}
fn schema_fingerprint(&self) -> u64 {
self.typ_hash()
}
fn encrypted(&self) -> bool {
self.tx_encrypted()
}
fn max_entry_bytes(&self) -> Option<usize> {
Some(self.config().write_buffer_size)
}
fn shard_of(&self, key: &[u8]) -> DbResult<usize> {
Ok(<Self as TxFamily>::tx_shard_of(self, key))
}
fn lock_shards<'a>(&'a self, shards: &[usize]) -> DbResult<Box<dyn TxLocked + 'a>> {
Ok(Box::new(TxHeld::lock(self, shards)))
}
fn tx_project(
&self,
key: &[u8],
old: Option<&[u8]>,
new: Option<&[u8]>,
) -> DbResult<Vec<crate::txn::participant::TxDerivedWrite>> {
self.tx_project_writes(key, old, new)
}
fn tx_stream_writes(
&self,
key: &[u8],
old: Option<&[u8]>,
new: Option<&[u8]>,
) -> DbResult<crate::txn::participant::TxStreamWrites> {
self.tx_stream_writes(key, old, new)
}
fn tx_counter_deltas(
&self,
key: &[u8],
old: Option<&[u8]>,
new: Option<&[u8]>,
) -> DbResult<Vec<crate::txn::participant::TxCounterDelta>> {
self.tx_counter_deltas_for(key, old, new)
}
fn tx_claims(
&self,
key: &[u8],
old: Option<&[u8]>,
new: Option<&[u8]>,
) -> DbResult<Vec<ClaimOp>> {
self.tx_claims_for(self.name(), key, old, new)
}
fn tx_preplan_projection_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
self.tx_preplan_projection_targets()
}
fn tx_preplan_stream_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
Ok(VarMap::tx_preplan_stream_targets(self))
}
fn tx_prepare_streams(&self) -> DbResult<Vec<crate::queue::retention::ProducerPermit>> {
VarMap::tx_prepare_streams(self)
}
}
#[cfg(all(feature = "typed-tree", feature = "var-collections"))]
mod var_typed {
use super::*;
use crate::codec::Codec;
use crate::hook::TypedWriteHook;
use crate::txn::TxTarget;
use crate::txn::participant::TxCollectionName;
use crate::var_typed_map::VarTypedMap;
use crate::var_typed_tree::VarTypedTree;
impl<
K: Key + 'static,
T: armour_core::GetType + TxCollectionName + Send + Sync + 'static,
C: Codec<T> + Clone + Sync + 'static,
H: TypedWriteHook<K, T> + 'static,
> TxParticipant for VarTypedTree<K, T, C, H>
{
fn name(&self) -> &'static str {
T::tx_collection_name()
}
fn backend(&self) -> TxBackend {
TxBackend::Bitcask
}
fn schema_fingerprint(&self) -> u64 {
T::TYPE.h()
}
fn encrypted(&self) -> bool {
TxParticipant::encrypted(self.as_inner())
}
fn max_entry_bytes(&self) -> Option<usize> {
TxParticipant::max_entry_bytes(self.as_inner())
}
fn shard_of(&self, key: &[u8]) -> DbResult<usize> {
TxParticipant::shard_of(self.as_inner(), key)
}
fn shard_count(&self) -> usize {
TxParticipant::shard_count(self.as_inner())
}
fn lock_shards<'a>(&'a self, shards: &[usize]) -> DbResult<Box<dyn TxLocked + 'a>> {
TxParticipant::lock_shards(self.as_inner(), shards)
}
fn tx_project(
&self,
key: &[u8],
old: Option<&[u8]>,
new: Option<&[u8]>,
) -> DbResult<Vec<crate::txn::participant::TxDerivedWrite>> {
self.tx_project_writes(key, old, new)
}
fn tx_stream_writes(
&self,
key: &[u8],
old: Option<&[u8]>,
new: Option<&[u8]>,
) -> DbResult<crate::txn::participant::TxStreamWrites> {
TxParticipant::tx_stream_writes(self.as_inner(), key, old, new)
}
fn tx_counter_deltas(
&self,
key: &[u8],
old: Option<&[u8]>,
new: Option<&[u8]>,
) -> DbResult<Vec<crate::txn::participant::TxCounterDelta>> {
self.tx_counter_deltas_for(key, old, new)
}
fn tx_claims(
&self,
key: &[u8],
old: Option<&[u8]>,
new: Option<&[u8]>,
) -> DbResult<Vec<ClaimOp>> {
self.tx_claims_for(self.name(), key, old, new)
}
fn tx_preplan_projection_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
self.tx_preplan_projection_targets()
}
fn tx_preplan_stream_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
TxParticipant::tx_preplan_stream_targets(self.as_inner())
}
fn tx_prepare_streams(&self) -> DbResult<Vec<crate::queue::retention::ProducerPermit>> {
TxParticipant::tx_prepare_streams(self.as_inner())
}
}
impl<
K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
T: armour_core::GetType + TxCollectionName + Send + Sync + 'static,
C: Codec<T> + Clone + Sync + 'static,
H: TypedWriteHook<K, T> + 'static,
> TxParticipant for VarTypedMap<K, T, C, H>
{
fn name(&self) -> &'static str {
T::tx_collection_name()
}
fn backend(&self) -> TxBackend {
TxBackend::Bitcask
}
fn schema_fingerprint(&self) -> u64 {
T::TYPE.h()
}
fn encrypted(&self) -> bool {
TxParticipant::encrypted(self.as_inner())
}
fn max_entry_bytes(&self) -> Option<usize> {
TxParticipant::max_entry_bytes(self.as_inner())
}
fn shard_of(&self, key: &[u8]) -> DbResult<usize> {
TxParticipant::shard_of(self.as_inner(), key)
}
fn shard_count(&self) -> usize {
TxParticipant::shard_count(self.as_inner())
}
fn lock_shards<'a>(&'a self, shards: &[usize]) -> DbResult<Box<dyn TxLocked + 'a>> {
TxParticipant::lock_shards(self.as_inner(), shards)
}
fn tx_project(
&self,
key: &[u8],
old: Option<&[u8]>,
new: Option<&[u8]>,
) -> DbResult<Vec<crate::txn::participant::TxDerivedWrite>> {
self.tx_project_writes(key, old, new)
}
fn tx_stream_writes(
&self,
key: &[u8],
old: Option<&[u8]>,
new: Option<&[u8]>,
) -> DbResult<crate::txn::participant::TxStreamWrites> {
TxParticipant::tx_stream_writes(self.as_inner(), key, old, new)
}
fn tx_counter_deltas(
&self,
key: &[u8],
old: Option<&[u8]>,
new: Option<&[u8]>,
) -> DbResult<Vec<crate::txn::participant::TxCounterDelta>> {
self.tx_counter_deltas_for(key, old, new)
}
fn tx_claims(
&self,
key: &[u8],
old: Option<&[u8]>,
new: Option<&[u8]>,
) -> DbResult<Vec<ClaimOp>> {
self.tx_claims_for(self.name(), key, old, new)
}
fn tx_preplan_projection_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
self.tx_preplan_projection_targets()
}
fn tx_preplan_stream_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
TxParticipant::tx_preplan_stream_targets(self.as_inner())
}
fn tx_prepare_streams(&self) -> DbResult<Vec<crate::queue::retention::ProducerPermit>> {
TxParticipant::tx_prepare_streams(self.as_inner())
}
}
impl<
K: Key + 'static,
T: Send + Sync + 'static,
C: Codec<T> + Clone + Sync + 'static,
H: TypedWriteHook<K, T> + 'static,
> TxTarget for VarTypedTree<K, T, C, H>
{
type Key = K;
type Val = T;
fn tx_encode(&self, value: &T, out: &mut Vec<u8>) -> DbResult<()> {
self.codec().encode_to(value, out)
}
fn tx_decode(&self, bytes: &[u8]) -> DbResult<T> {
self.codec().decode_from(bytes)
}
}
impl<
K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
T: Send + Sync + 'static,
C: Codec<T> + Clone + Sync + 'static,
H: TypedWriteHook<K, T> + 'static,
> TxTarget for VarTypedMap<K, T, C, H>
{
type Key = K;
type Val = T;
fn tx_encode(&self, value: &T, out: &mut Vec<u8>) -> DbResult<()> {
self.codec().encode_to(value, out)
}
fn tx_decode(&self, bytes: &[u8]) -> DbResult<T> {
self.codec().decode_from(bytes)
}
}
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use crate::txn::participant::{PlannedOp, TxParticipant};
use crate::{Config, NoHook, VarTree};
fn open_tree(dir: &std::path::Path) -> Arc<VarTree<[u8; 8], NoHook>> {
let tree = Arc::new(VarTree::open(dir, Config::test()).unwrap());
VarTree::bind_tx_self(&tree);
tree
}
#[test]
fn tx_var_reads_and_writes_opaque_bytes() {
let dir = tempfile::tempdir().unwrap();
let tree = open_tree(dir.path());
tree.put(&[1u8; 8], b"hello").unwrap();
let sid = TxParticipant::shard_of(&*tree, &[1u8; 8]).unwrap();
let mut locked = TxParticipant::lock_shards(&*tree, &[sid]).unwrap();
assert_eq!(
locked.read(&[1u8; 8]).unwrap().as_deref(),
Some(&b"hello"[..])
);
let tags = locked
.reserve(&[
PlannedOp {
key: &[1u8; 8],
delete: false,
},
PlannedOp {
key: &[2u8; 8],
delete: true,
},
])
.unwrap();
locked.apply_put(&[1u8; 8], b"world!!", tags[0]).unwrap();
locked.apply_delete(&[2u8; 8], tags[1]).unwrap();
drop(
locked
.finish(
&mut Vec::new(),
&mut crate::watch::WatchWakeBatch::default(),
)
.unwrap(),
);
assert_eq!(tree.get(&[1u8; 8]).unwrap().as_ref(), &b"world!!"[..]);
}
#[test]
fn tx_var_spans_several_shards_of_one_collection() {
let dir = tempfile::tempdir().unwrap();
let tree: Arc<VarTree<[u8; 8], NoHook>> = Arc::new(
VarTree::open(
dir.path(),
Config {
shard_count: 4,
shard_prefix_bits: 2,
..Config::test()
},
)
.unwrap(),
);
VarTree::bind_tx_self(&tree);
let a = [0u8; 8];
let b = (0u8..=255)
.map(|i| [i, 0, 0, 0, 0, 0, 0, 0])
.find(|k| {
TxParticipant::shard_of(&*tree, k).unwrap()
!= TxParticipant::shard_of(&*tree, &a).unwrap()
})
.expect("two shards must be reachable");
let mut shards = vec![
TxParticipant::shard_of(&*tree, &a).unwrap(),
TxParticipant::shard_of(&*tree, &b).unwrap(),
];
shards.sort_unstable();
let mut locked = TxParticipant::lock_shards(&*tree, &shards).unwrap();
let tags = locked
.reserve(&[
PlannedOp {
key: &a,
delete: false,
},
PlannedOp {
key: &b,
delete: false,
},
])
.unwrap();
locked.apply_put(&a, b"A", tags[0]).unwrap();
locked.apply_put(&b, b"B", tags[1]).unwrap();
drop(
locked
.finish(
&mut Vec::new(),
&mut crate::watch::WatchWakeBatch::default(),
)
.unwrap(),
);
assert_eq!(tree.get(&a).unwrap().as_ref(), &b"A"[..]);
assert_eq!(tree.get(&b).unwrap().as_ref(), &b"B"[..]);
}
#[test]
fn tx_var_rejects_a_key_outside_the_locked_shards() {
let dir = tempfile::tempdir().unwrap();
let tree: Arc<VarTree<[u8; 8], NoHook>> = Arc::new(
VarTree::open(
dir.path(),
Config {
shard_count: 4,
shard_prefix_bits: 2,
..Config::test()
},
)
.unwrap(),
);
VarTree::bind_tx_self(&tree);
let a = [0u8; 8];
let sid = TxParticipant::shard_of(&*tree, &a).unwrap();
let other = (0u8..=255)
.map(|i| [i, 0, 0, 0, 0, 0, 0, 0])
.find(|k| TxParticipant::shard_of(&*tree, k).unwrap() != sid)
.expect("two shards must be reachable");
let mut locked = TxParticipant::lock_shards(&*tree, &[sid]).unwrap();
assert!(matches!(
locked.read(&other),
Err(crate::DbError::ShardMismatch)
));
locked.abort();
}
}