fix: serialize unsafe backend structural mutations

src/index/arctic.rs

@@ -6,6 +6,7 @@ use std::ops::{Bound, RangeBounds};
 use std::sync::atomic::{AtomicUsize, Ordering};
 
 use arctic::{ConcurrentMap, Key, Order};
+use parking_lot::RwLock;
 
 use super::UniqueIndex;
 
@@ -71,6 +72,10 @@ impl_arctic_key!(u16, u32, u64, u128);
 /// satisfy WorkTable's double-ended query interface.
 pub struct ArcticIndex<K: ArcticKey, V> {
     inner: ConcurrentMap<K::Raw, Box<V>>,
+    // arctic-wt 0.1.4 can trip its raw-cursor structural assertion when
+    // mutations overlap and can expose a transient structural rewrite to a
+    // point read. Reads share access; structural mutations take it exclusively.
+    access: RwLock<()>,
     len: AtomicUsize,
 }
 
@@ -89,6 +94,7 @@ where
     fn default() -> Self {
         Self {
             inner: ConcurrentMap::default(),
+            access: RwLock::new(()),
             len: AtomicUsize::new(0),
         }
     }
@@ -115,6 +121,7 @@ where
         let len = inner.all().entries(Order::Ascend).count();
         Ok(Self {
             inner,
+            access: RwLock::new(()),
             len: AtomicUsize::new(len),
         })
     }
@@ -127,23 +134,27 @@ where
 {
     #[inline]
     fn get_value(&self, key: &K) -> Option<V> {
-        self.with_value(key, Clone::clone)
+        let _access = self.access.read();
+        let key = key.to_arctic();
+        self.inner.get(key.borrow()).map(|value| value.clone())
     }
 
     #[inline]
     fn with_value<R>(&self, key: &K, read: impl FnOnce(&V) -> R) -> Option<R> {
-        let key = key.to_arctic();
-        self.inner.get(key.borrow()).map(|value| read(&value))
+        let value = self.get_value(key)?;
+        Some(read(&value))
     }
 
     #[inline]
     fn contains_key(&self, key: &K) -> bool {
+        let _access = self.access.read();
         let key = key.to_arctic();
         self.inner.get(key.borrow()).is_some()
     }
 
     #[inline]
     fn insert_value(&self, key: K, value: V) -> Option<V> {
+        let _access = self.access.write();
         let key = key.to_arctic();
         let updated = self.inner.upsert(key.as_insert(), Box::new(value));
         let old = updated.old().cloned();
@@ -155,6 +166,7 @@ where
 
     #[inline]
     fn insert_value_checked(&self, key: K, value: V) -> Option<()> {
+        let _access = self.access.write();
         let key = key.to_arctic();
         match self.inner.insert(key.as_insert(), Box::new(value)) {
             Ok(_) => {
@@ -170,6 +182,7 @@ where
 
     #[inline]
     fn remove_value(&self, key: &K) -> Option<(K, V)> {
+        let _access = self.access.write();
         let raw_key = key.to_arctic();
         let old = self.inner.remove(raw_key.borrow())?;
         self.len.fetch_sub(1, Ordering::Relaxed);
@@ -182,6 +195,7 @@ where
     }
 
     fn iter_values(&self) -> impl DoubleEndedIterator<Item = (K, V)> + '_ {
+        let _access = self.access.read();
         let shard = self.inner.all();
         shard
             .entries(Order::Ascend)
@@ -198,6 +212,7 @@ where
     where
         R: RangeBounds<K> + 'a,
     {
+        let _access = self.access.read();
         let lower = match range.start_bound() {
             Bound::Included(key) => Some(key.to_arctic()),
             Bound::Excluded(key) => key.to_arctic().next(),

src/index/congee.rs

@@ -6,7 +6,7 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicUsize, Ordering};
 
 use congee::{CongeeRaw, DefaultAllocator};
-use parking_lot::Mutex;
+use parking_lot::RwLock;
 
 use super::UniqueIndex;
 
@@ -48,10 +48,10 @@ impl_congee_key!(u64);
 /// reclamation pattern used by Congee's own `CongeeArc` implementation.
 pub struct CongeeIndex<K, V> {
     inner: CongeeRaw<usize, usize>,
-    // congee-wt 0.4.1 can lose disjoint insert/remove mutations when their
-    // structural updates overlap. Keep point reads native and concurrent, but
-    // serialize mutations until the backend offers the required visibility.
-    mutation: Mutex<()>,
+    // congee-wt 0.4.1 can lose disjoint structural mutations and let point
+    // reads miss while another mutation rewrites the tree. Reads remain
+    // concurrent with reads; structural mutations take exclusive access.
+    access: RwLock<()>,
     len: AtomicUsize,
     marker: std::marker::PhantomData<(K, V)>,
 }
@@ -77,7 +77,7 @@ where
         };
         Self {
             inner: CongeeRaw::new_with_drainer(DefaultAllocator {}, drainer),
-            mutation: Mutex::new(()),
+            access: RwLock::new(()),
             len: AtomicUsize::new(0),
             marker: std::marker::PhantomData,
         }
@@ -116,6 +116,7 @@ where
     where
         R: RangeBounds<K>,
     {
+        let _access = self.access.read();
         let start = match range.start_bound() {
             Bound::Included(key) => key.into_congee(),
             Bound::Excluded(key) => match key.into_congee().checked_add(1) {
@@ -196,7 +197,7 @@ where
         let len = inner.keys().len();
         Ok(Self {
             inner,
-            mutation: Mutex::new(()),
+            access: RwLock::new(()),
             len: AtomicUsize::new(len),
             marker: std::marker::PhantomData,
         })
@@ -210,28 +211,30 @@ where
 {
     #[inline]
     fn get_value(&self, key: &K) -> Option<V> {
-        self.with_value(key, Clone::clone)
+        let _access = self.access.read();
+        let guard = self.inner.pin();
+        let pointer = self.inner.get(&key.into_congee(), &guard)?;
+        // SAFETY: the read lock prevents removal and the epoch guard retains
+        // the tree-owned allocation while its value is cloned.
+        Some(unsafe { &*std::ptr::with_exposed_provenance::<V>(pointer) }.clone())
     }
 
     #[inline]
     fn with_value<R>(&self, key: &K, read: impl FnOnce(&V) -> R) -> Option<R> {
-        let guard = self.inner.pin();
-        let pointer = self.inner.get(&key.into_congee(), &guard)?;
-        // SAFETY: the epoch guard keeps the tree-owned `Arc<V>` alive for the
-        // duration of `read`, and the pointer originated from `Arc::into_raw`.
-        let value = unsafe { &*std::ptr::with_exposed_provenance::<V>(pointer) };
-        Some(read(value))
+        let value = self.get_value(key)?;
+        Some(read(&value))
     }
 
     #[inline]
     fn contains_key(&self, key: &K) -> bool {
+        let _access = self.access.read();
         let guard = self.inner.pin();
         self.inner.get(&key.into_congee(), &guard).is_some()
     }
 
     #[inline]
     fn insert_value(&self, key: K, value: V) -> Option<V> {
-        let _mutation = self.mutation.lock();
+        let _access = self.access.write();
         let guard = self.inner.pin();
         let pointer = Arc::into_raw(Arc::new(value)).expose_provenance();
         match self.inner.insert(key.into_congee(), pointer, &guard) {
@@ -250,7 +253,7 @@ where
 
     #[inline]
     fn insert_value_checked(&self, key: K, value: V) -> Option<()> {
-        let _mutation = self.mutation.lock();
+        let _access = self.access.write();
         let guard = self.inner.pin();
         let pointer = Arc::into_raw(Arc::new(value)).expose_provenance();
         let result = self
@@ -278,7 +281,7 @@ where
 
     #[inline]
     fn remove_value(&self, key: &K) -> Option<(K, V)> {
-        let _mutation = self.mutation.lock();
+        let _access = self.access.write();
         let guard = self.inner.pin();
         let pointer = self.inner.remove(&key.into_congee(), &guard)?;
         self.len.fetch_sub(1, Ordering::Relaxed);

src/index/unique.rs

@@ -206,7 +206,7 @@ pub type UpstreamIndexPair<K, V> = VanillaPair<K, V>;
 
 #[cfg(test)]
 mod tests {
-    use std::sync::Arc;
+    use std::sync::{Arc, Barrier};
 
     use super::{UniqueIndex, UpstreamIndexMap};
     use crate::{ArcticIndex, CongeeIndex, IndexMap};
@@ -317,6 +317,32 @@ mod tests {
         assert!(index.is_empty());
     }
 
+    fn assert_separate_instances_do_not_interfere<I>()
+    where
+        I: UniqueIndex<u64, u64> + Send + Sync + 'static,
+    {
+        let barrier = Arc::new(Barrier::new(9));
+        let mut threads = Vec::new();
+        for worker in 0..8_u64 {
+            let barrier = Arc::clone(&barrier);
+            threads.push(std::thread::spawn(move || {
+                let index = I::default();
+                barrier.wait();
+                for sequence in 0..10_000_u64 {
+                    let key = worker * 10_000 + sequence;
+                    assert_eq!(index.insert_value_checked(key, key + 1), Some(()));
+                    assert_eq!(index.get_value(&key), Some(key + 1));
+                    assert_eq!(index.remove_value(&key), Some((key, key + 1)));
+                }
+                assert!(index.is_empty());
+            }));
+        }
+        barrier.wait();
+        for thread in threads {
+            thread.join().unwrap();
+        }
+    }
+
     #[test]
     fn worktables_index_implements_contract() {
         assert_unique_index_contract::<IndexMap<u64, u64>>();
@@ -335,9 +361,29 @@ mod tests {
         assert_disjoint_concurrent_insert_then_remove::<ArcticIndex<u64, u64>>();
     }
 
+    #[test]
+    fn worktables_index_preserves_disjoint_concurrent_mutations() {
+        assert_disjoint_concurrent_insert_then_remove::<IndexMap<u64, u64>>();
+    }
+
+    #[test]
+    fn upstream_indexset_preserves_disjoint_concurrent_mutations() {
+        assert_disjoint_concurrent_insert_then_remove::<UpstreamIndexMap<u64, u64>>();
+    }
+
     #[test]
     fn art_backends_make_disjoint_mutations_immediately_visible() {
         assert_immediate_disjoint_crud::<CongeeIndex<u64, u64>>();
         assert_immediate_disjoint_crud::<ArcticIndex<u64, u64>>();
     }
+
+    #[test]
+    fn congee_instances_do_not_share_mutation_state() {
+        assert_separate_instances_do_not_interfere::<CongeeIndex<u64, u64>>();
+    }
+
+    #[test]
+    fn arctic_instances_do_not_share_mutation_state() {
+        assert_separate_instances_do_not_interfere::<ArcticIndex<u64, u64>>();
+    }
 }