Skip to main content

indexset/concurrent/
set.rs

1use ::core::borrow::Borrow;
2use ::core::fmt::Debug;
3use ::core::iter::FusedIterator;
4use ::core::marker::PhantomData;
5use ::core::ops::{Bound, RangeBounds};
6use ::core::sync::atomic::{AtomicPtr, AtomicU64, Ordering};
7use alloc::boxed::Box;
8use alloc::collections::BTreeMap;
9use alloc::sync::Arc;
10use alloc::vec;
11use alloc::vec::Vec;
12use parking_lot::{
13    ArcRwLockReadGuard, ArcRwLockWriteGuard, Mutex, MutexGuard, RawRwLock, RwLock, RwLockReadGuard, RwLockWriteGuard,
14};
15
16use crate::cdc::change::ChangeEvent;
17use crate::concurrent::operation::*;
18use crate::core::constants::DEFAULT_INNER_SIZE;
19use crate::core::node::*;
20
21use super::r#ref::Ref;
22
23/// Give the scheduler the core, when there is a scheduler to give it to.
24///
25/// `yield_now` is a `std` call, and a `no_std` build has no thread to yield.
26/// Spinning is the honest fallback there: it is what the caller was already
27/// doing on the fast path, without the syscall that would make it wait longer.
28#[inline]
29fn yield_now() {
30    #[cfg(feature = "std")]
31    std::thread::yield_now();
32    #[cfg(not(feature = "std"))]
33    ::core::hint::spin_loop();
34}
35
36const ROOT_PUBLICATION_SPIN_LIMIT: usize = 16;
37const STABLE_READ_BLOCKING_FALLBACK_AFTER: usize = 2;
38const PUBLICATION_BACKLOG_DRAIN_THRESHOLD: usize = 64;
39
40type NodeIndex<T, Node> = BTreeMap<T, Arc<RwLock<Node>>>;
41
42// Point-read routes are kept in immutable, cache-friendly chunks. Publishing
43// clones only the short vector of chunk Arcs and the one chunk containing the
44// changed boundary; it never copies the full node index. At WorkTable's
45// default 1,024 rows per node, one 128-route chunk covers roughly 131k rows.
46const PUBLISHED_ROUTES_PER_CHUNK: usize = 128;
47// Leave rebuilt chunks room for subsequent inserts, and merge only below the
48// split threshold so alternating insert/remove cannot thrash one boundary.
49const PUBLISHED_REBUILD_ROUTES_PER_CHUNK: usize = PUBLISHED_ROUTES_PER_CHUNK * 2 / 3;
50const PUBLISHED_ROUTE_MERGE_THRESHOLD: usize = PUBLISHED_ROUTES_PER_CHUNK * 3 / 4;
51
52struct PublishedChunk<T, Node> {
53    entries: Vec<(T, Arc<RwLock<Node>>)>,
54}
55
56impl<T: Clone, Node> Clone for PublishedChunk<T, Node> {
57    fn clone(&self) -> Self {
58        Self {
59            entries: self.entries.clone(),
60        }
61    }
62}
63
64struct PublishedNodeIndex<T, Node> {
65    chunks: Vec<Arc<PublishedChunk<T, Node>>>,
66    len: usize,
67}
68
69impl<T, Node> Clone for PublishedNodeIndex<T, Node> {
70    fn clone(&self) -> Self {
71        Self {
72            chunks: self.chunks.clone(),
73            len: self.len,
74        }
75    }
76}
77
78impl<T, Node> PublishedNodeIndex<T, Node>
79where
80    T: Ord + Clone,
81{
82    fn from_canonical(index: &NodeIndex<T, Node>) -> Self {
83        let mut chunks = Vec::with_capacity(index.len().div_ceil(PUBLISHED_REBUILD_ROUTES_PER_CHUNK));
84        let mut entries = Vec::with_capacity(PUBLISHED_REBUILD_ROUTES_PER_CHUNK);
85
86        for (key, node) in index {
87            entries.push((key.clone(), node.clone()));
88            if entries.len() == PUBLISHED_REBUILD_ROUTES_PER_CHUNK {
89                chunks.push(Arc::new(PublishedChunk { entries }));
90                entries = Vec::with_capacity(PUBLISHED_REBUILD_ROUTES_PER_CHUNK);
91            }
92        }
93        if !entries.is_empty() {
94            chunks.push(Arc::new(PublishedChunk { entries }));
95        }
96
97        Self {
98            chunks,
99            len: index.len(),
100        }
101    }
102
103    fn iter(&self) -> impl Iterator<Item = (&T, &Arc<RwLock<Node>>)> {
104        self.chunks
105            .iter()
106            .flat_map(|chunk| chunk.entries.iter().map(|(key, node)| (key, node)))
107    }
108
109    fn first_key_value(&self) -> Option<(&T, &Arc<RwLock<Node>>)> {
110        self.chunks.first()?.entries.first().map(|(key, node)| (key, node))
111    }
112
113    fn last_key_value(&self) -> Option<(&T, &Arc<RwLock<Node>>)> {
114        self.chunks.last()?.entries.last().map(|(key, node)| (key, node))
115    }
116
117    fn chunk_for<Q>(&self, key: &Q) -> usize
118    where
119        T: Borrow<Q>,
120        Q: Ord + ?Sized,
121    {
122        self.chunks.partition_point(|chunk| {
123            let max = &chunk.entries.last().expect("published chunks are non-empty").0;
124            <T as Borrow<Q>>::borrow(max) < key
125        })
126    }
127
128    fn first_for_bound<Q>(&self, bound: Bound<&Q>) -> Option<(&T, &Arc<RwLock<Node>>)>
129    where
130        T: Borrow<Q>,
131        Q: Ord + ?Sized,
132    {
133        let key = match bound {
134            Bound::Included(key) | Bound::Excluded(key) => key,
135            Bound::Unbounded => return self.first_key_value(),
136        };
137        let mut chunk_index = self.chunk_for(key);
138        while let Some(chunk) = self.chunks.get(chunk_index) {
139            let entry_index = chunk.entries.partition_point(|(candidate, _)| match bound {
140                Bound::Included(_) => <T as Borrow<Q>>::borrow(candidate) < key,
141                Bound::Excluded(_) => <T as Borrow<Q>>::borrow(candidate) <= key,
142                Bound::Unbounded => false,
143            });
144            if let Some((found, node)) = chunk.entries.get(entry_index) {
145                return Some((found, node));
146            }
147            chunk_index += 1;
148        }
149        None
150    }
151
152    fn insert(&mut self, key: T, node: Arc<RwLock<Node>>) -> Option<Arc<RwLock<Node>>> {
153        if self.chunks.is_empty() {
154            self.chunks.push(Arc::new(PublishedChunk {
155                entries: vec![(key, node)],
156            }));
157            self.len = 1;
158            return None;
159        }
160
161        let mut chunk_index = self.chunk_for(&key);
162        if chunk_index == self.chunks.len() {
163            chunk_index -= 1;
164        }
165        let chunk = Arc::make_mut(&mut self.chunks[chunk_index]);
166        match chunk.entries.binary_search_by(|(candidate, _)| candidate.cmp(&key)) {
167            Ok(index) => Some(::core::mem::replace(&mut chunk.entries[index].1, node)),
168            Err(index) => {
169                chunk.entries.insert(index, (key, node));
170                self.len += 1;
171                if chunk.entries.len() > PUBLISHED_ROUTES_PER_CHUNK {
172                    let right = chunk.entries.split_off(chunk.entries.len() / 2);
173                    self.chunks
174                        .insert(chunk_index + 1, Arc::new(PublishedChunk { entries: right }));
175                }
176                None
177            }
178        }
179    }
180
181    fn remove<Q>(&mut self, key: &Q) -> Option<Arc<RwLock<Node>>>
182    where
183        T: Borrow<Q>,
184        Q: Ord + ?Sized,
185    {
186        let chunk_index = self.chunk_for(key);
187        let entry_index = self
188            .chunks
189            .get(chunk_index)?
190            .entries
191            .binary_search_by(|(candidate, _)| <T as Borrow<Q>>::borrow(candidate).cmp(key))
192            .ok()?;
193        let chunk = Arc::make_mut(&mut self.chunks[chunk_index]);
194        let (_, removed) = chunk.entries.remove(entry_index);
195        self.len -= 1;
196
197        if chunk.entries.is_empty() {
198            self.chunks.remove(chunk_index);
199        } else if chunk_index > 0
200            && self.chunks[chunk_index - 1].entries.len() + self.chunks[chunk_index].entries.len()
201                <= PUBLISHED_ROUTE_MERGE_THRESHOLD
202        {
203            let right = self.chunks.remove(chunk_index);
204            Arc::make_mut(&mut self.chunks[chunk_index - 1])
205                .entries
206                .extend(right.entries.iter().cloned());
207        } else if chunk_index + 1 < self.chunks.len()
208            && self.chunks[chunk_index].entries.len() + self.chunks[chunk_index + 1].entries.len()
209                <= PUBLISHED_ROUTE_MERGE_THRESHOLD
210        {
211            let right = self.chunks.remove(chunk_index + 1);
212            Arc::make_mut(&mut self.chunks[chunk_index])
213                .entries
214                .extend(right.entries.iter().cloned());
215        }
216
217        Some(removed)
218    }
219}
220
221#[inline]
222fn node_identity<Node>(node: &Arc<RwLock<Node>>) -> usize {
223    // Identity token only: it is never converted back into or dereferenced as
224    // a pointer. The Arc stays live while the token is present, preventing
225    // allocator reuse from aliasing two published nodes.
226    Arc::as_ptr(node) as usize
227}
228
229struct RetiredIndex<T, Node>(*mut PublishedNodeIndex<T, Node>);
230
231// SAFETY: the pointer is uniquely owned after it has been swapped out of the
232// publication slot, and this wrapper exposes no access to the map. Its only
233// operation is destruction after the grace period. Dropping a shared route
234// chunk only decrements its Arc; dropping the final route path can move/drop T
235// and Node on the reclaiming thread, hence Send. The wrapper never dereferences
236// the index, and its private field prevents callers from adding such access
237// without revisiting this proof.
238unsafe impl<T: Send, Node: Send> Send for RetiredIndex<T, Node> {}
239
240impl<T, Node> Drop for RetiredIndex<T, Node> {
241    fn drop(&mut self) {
242        // SAFETY: this wrapper is created exactly once for a pointer returned
243        // by `Box::into_raw`, after that pointer has been atomically unlinked.
244        unsafe { drop(Box::from_raw(self.0)) }
245    }
246}
247
248struct PublishedIndex<T, Node> {
249    current: AtomicPtr<PublishedNodeIndex<T, Node>>,
250    domain: ps_reclaim::Domain,
251}
252
253impl<T, Node> PublishedIndex<T, Node> {
254    fn new() -> Self {
255        Self {
256            current: AtomicPtr::new(Box::into_raw(Box::new(PublishedNodeIndex {
257                chunks: Vec::new(),
258                len: 0,
259            }))),
260            domain: ps_reclaim::Domain::new(),
261        }
262    }
263}
264
265impl<T, Node> PublishedIndex<T, Node>
266where
267    T: Ord + Clone + Send + 'static,
268    Node: Send + 'static,
269{
270    /// Checks the final node in the currently published route root.
271    ///
272    /// The caller holds the structural writer lock, so another writer cannot
273    /// unlink `current` while the pointer is inspected. Readers never replace
274    /// the root.
275    fn last_is(&self, node: &Arc<RwLock<Node>>) -> bool {
276        let current = self.current.load(Ordering::Acquire);
277        // SAFETY: the structural writer lock keeps `current` linked for this
278        // call, and the returned references do not escape it.
279        unsafe {
280            (&*current)
281                .last_key_value()
282                .is_some_and(|(_, published)| Arc::ptr_eq(published, node))
283        }
284    }
285
286    fn snapshot(&self) -> PublishedNodeIndex<T, Node> {
287        let current = self.current.load(Ordering::Acquire);
288        // SAFETY: callers hold the only structural writer lock. `current`
289        // cannot be unlinked until that writer publishes its replacement.
290        unsafe { (&*current).clone() }
291    }
292
293    fn replace(&self, replacement: PublishedNodeIndex<T, Node>) -> RetiredIndex<T, Node> {
294        // The route index is structurally shared: publishing moves one root,
295        // and the writer copied only its chunk-Arc vector plus touched chunks.
296        let replacement = Box::into_raw(Box::new(replacement));
297        let retired = self.current.swap(replacement, Ordering::AcqRel);
298        RetiredIndex(retired)
299    }
300
301    fn retire(&self, retired: RetiredIndex<T, Node>) {
302        // Keep the pointer's provenance intact while transferring its unique
303        // ownership to the retirement callback.
304        self.domain.retire(move || drop(retired));
305    }
306
307    fn advance(&self) {
308        // A reader delayed on a node writer never holds a pin (see the point
309        // read paths below). Do not sweep ps-reclaim's 256-slot registry on
310        // every split: that turns the registry into the same reader/writer
311        // cache-line fight this publication path removes. A small bounded
312        // backlog amortizes the sweep while `advance` drains every route root
313        // whose grace period has elapsed.
314        if self.domain.pending() >= PUBLICATION_BACKLOG_DRAIN_THRESHOLD {
315            self.domain.advance();
316        }
317    }
318}
319
320impl<T, Node> Drop for PublishedIndex<T, Node> {
321    fn drop(&mut self) {
322        let current = *self.current.get_mut();
323        // SAFETY: exclusive access proves no reader can load `current`, and it
324        // is the one still-linked allocation created by `Box::into_raw`.
325        unsafe { drop(Box::from_raw(current)) }
326    }
327}
328
329// Publication invariant: every canonical node appears once and in the same
330// order in the published route index. At most one route key may differ from
331// its canonical key, and only for the canonical last node. That exception is
332// safe because point lookup falls back to the published last node above all
333// routes, while a stale route below the current maximum still selects that
334// same final node. A last-node shrink cannot reorder it before the preceding
335// node because node ranges are non-overlapping. Attachment repairs the route
336// before it can cease to be the last node.
337pub(crate) struct Topology<T, Node> {
338    index: RwLock<NodeIndex<T, Node>>,
339    // Writer-only reverse lookup from node identity to its current published
340    // route key. This differs from the canonical key only for the last node,
341    // whose stale route remains a valid final point-read fallback.
342    published_keys: Mutex<BTreeMap<usize, T>>,
343    published: PublishedIndex<T, Node>,
344    // Even values are stable publications; odd values mean a writer may have
345    // changed node contents or routing but has not published the new route.
346    generation: AtomicU64,
347}
348
349impl<T, Node> Debug for Topology<T, Node> {
350    fn fmt(&self, formatter: &mut ::core::fmt::Formatter<'_>) -> ::core::fmt::Result {
351        formatter
352            .debug_struct("Topology")
353            .field("nodes", &self.index.read().len())
354            .field("generation", &self.generation.load(Ordering::Relaxed))
355            .finish()
356    }
357}
358
359impl<T, Node> Topology<T, Node> {
360    fn new() -> Self {
361        Self {
362            index: RwLock::new(BTreeMap::new()),
363            published_keys: Mutex::new(BTreeMap::new()),
364            published: PublishedIndex::new(),
365            generation: AtomicU64::new(0),
366        }
367    }
368
369    #[inline]
370    pub(crate) fn read(&self) -> RwLockReadGuard<'_, NodeIndex<T, Node>> {
371        self.index.read()
372    }
373}
374
375impl<T, Node> Topology<T, Node>
376where
377    T: Ord + Clone + Send + 'static,
378    Node: Send + 'static,
379{
380    #[inline]
381    fn write(&self) -> TopologyWriteGuard<'_, T, Node> {
382        let index = self.index.write();
383        self.generation.fetch_add(1, Ordering::AcqRel);
384        TopologyWriteGuard {
385            topology: self,
386            index: Some(index),
387            published: None,
388            published_keys: None,
389            publish: true,
390            dirty: false,
391        }
392    }
393
394    /// Re-keys a node without replacing the point-read snapshot.
395    ///
396    /// This guard starts without a replacement snapshot. The commit may retain
397    /// that fast path only when re-keying the last node: point reads already
398    /// fall back to the last route beyond its published maximum. Re-keying any
399    /// earlier node must call `enable_publication`, because a subsequent insert
400    /// can fill a gap left by a shrinking maximum in the following node.
401    #[inline]
402    fn write_rekey(&self) -> TopologyWriteGuard<'_, T, Node> {
403        let index = self.index.write();
404        TopologyWriteGuard {
405            topology: self,
406            index: Some(index),
407            published: None,
408            published_keys: None,
409            publish: false,
410            dirty: false,
411        }
412    }
413
414    #[inline]
415    fn try_write(&self) -> Option<TopologyWriteGuard<'_, T, Node>> {
416        let index = self.index.try_write()?;
417        self.generation.fetch_add(1, Ordering::AcqRel);
418        Some(TopologyWriteGuard {
419            topology: self,
420            index: Some(index),
421            published: None,
422            published_keys: None,
423            publish: true,
424            dirty: false,
425        })
426    }
427}
428
429pub(crate) struct TopologyWriteGuard<'a, T, Node>
430where
431    T: Ord + Clone + Send + 'static,
432    Node: Send + 'static,
433{
434    topology: &'a Topology<T, Node>,
435    // Option lets Drop release the structural lock before advancing the
436    // reclamation domain. Range readers should not wait for a registry scan.
437    index: Option<RwLockWriteGuard<'a, NodeIndex<T, Node>>>,
438    published: Option<PublishedNodeIndex<T, Node>>,
439    published_keys: Option<MutexGuard<'a, BTreeMap<usize, T>>>,
440    publish: bool,
441    dirty: bool,
442}
443
444impl<T, Node> ::core::ops::Deref for TopologyWriteGuard<'_, T, Node>
445where
446    T: Ord + Clone + Send + 'static,
447    Node: Send + 'static,
448{
449    type Target = NodeIndex<T, Node>;
450
451    fn deref(&self) -> &Self::Target {
452        self.index.as_deref().expect("topology guard already released")
453    }
454}
455
456impl<'a, T, Node> TopologyWriteGuard<'a, T, Node>
457where
458    T: Ord + Clone + Send + 'static,
459    Node: Send + 'static,
460{
461    pub(crate) fn enable_publication(&mut self) {
462        if self.publish {
463            return;
464        }
465        debug_assert!(
466            !self.dirty,
467            "publication must be enabled before mutating an opt-out topology guard"
468        );
469        // `write_rekey` deliberately leaves the stable generation untouched
470        // for a route-safe last-node rekey. If commit discovers that the
471        // route really must change, enter the odd writer generation before
472        // constructing or publishing its replacement.
473        self.topology.generation.fetch_add(1, Ordering::AcqRel);
474        self.publish = true;
475    }
476
477    fn ensure_publication_snapshot(&mut self) {
478        if self.published.is_none() {
479            self.published_keys = Some(self.topology.published_keys.lock());
480            self.published = Some(self.topology.published.snapshot());
481        }
482    }
483
484    /// Restores all derived publication state from the canonical topology.
485    ///
486    /// Ordinary mutations update one route in O(log N). This bounded O(N)
487    /// recovery is reserved for an internal identity mismatch or impossible
488    /// key collision; it prevents a bookkeeping defect from panicking or
489    /// entering an unbounded repair loop while the structural lock is held.
490    fn rebuild_publication(&mut self) {
491        if self.published_keys.is_none() {
492            self.published_keys = Some(self.topology.published_keys.lock());
493        }
494        let canonical = self.index.as_deref().expect("topology guard already released");
495        let rebuilt = PublishedNodeIndex::from_canonical(canonical);
496        let rebuilt_keys = canonical
497            .iter()
498            .map(|(key, node)| (node_identity(node), key.clone()))
499            .collect();
500        **self
501            .published_keys
502            .as_mut()
503            .expect("publication identity lock was initialized") = rebuilt_keys;
504        self.published = Some(rebuilt);
505        self.dirty = true;
506    }
507
508    pub(crate) fn is_last_node(&self, node: &Arc<RwLock<Node>>) -> bool {
509        self.index
510            .as_deref()
511            .and_then(BTreeMap::last_key_value)
512            .is_some_and(|(_, candidate)| Arc::ptr_eq(candidate, node))
513    }
514
515    /// A last-node boundary may stay unpublished only while the node remains
516    /// last in both route orders. An emptied node can be refilled with a value
517    /// below its predecessor; treating that as an ordinary last-node shrink
518    /// leaves the canonical and published nodes in different orders.
519    pub(crate) fn can_rekey_last_node_without_publication(&self, node: &Arc<RwLock<Node>>, new_key: &T) -> bool {
520        let mut canonical = self
521            .index
522            .as_deref()
523            .expect("topology guard already released")
524            .iter()
525            .rev();
526        let Some((_, last_node)) = canonical.next() else {
527            return false;
528        };
529        if !Arc::ptr_eq(last_node, node) || !self.topology.published.last_is(node) {
530            return false;
531        }
532        match canonical.next() {
533            Some((predecessor, _)) => predecessor < new_key,
534            None => true,
535        }
536    }
537
538    /// Re-keys the one route that may safely remain stale for point reads.
539    ///
540    /// This is deliberately separate from `insert`/`remove`: those general
541    /// mutation methods require publication to be enabled. The caller has
542    /// already verified that `old_key` identifies `node` and that it is the
543    /// canonical last node. Its published route remains a valid final
544    /// fallback whether the maximum grows or shrinks.
545    pub(crate) fn rekey_last_node(&mut self, old_key: &T, new_key: T, node: Arc<RwLock<Node>>) {
546        debug_assert!(!self.publish, "last-node rekey must use the opt-out guard");
547        debug_assert!(
548            !self.dirty,
549            "an opt-out guard may perform only one explicit last-node rekey"
550        );
551        debug_assert!(
552            self.is_last_node(&node),
553            "only the canonical last node may skip publication"
554        );
555
556        let index = self.index.as_deref_mut().expect("topology guard already released");
557        let removed = index.remove(old_key);
558        debug_assert!(
559            removed.as_ref().is_some_and(|removed| Arc::ptr_eq(removed, &node)),
560            "last-node rekey must remove its expected canonical route"
561        );
562        let replaced = index.insert(new_key, node);
563        debug_assert!(
564            replaced.is_none(),
565            "last-node rekey must not collide with another canonical route"
566        );
567        self.dirty = true;
568    }
569
570    /// Makes the current canonical last-node route exact before another route
571    /// can be inserted after it. Attachment calls this once per batch.
572    fn repair_last_route_before_insert(&mut self) {
573        debug_assert!(self.publish, "route insertion repair requires publication");
574        let Some((canonical_key, last_node)) = self
575            .index
576            .as_deref()
577            .expect("topology guard already released")
578            .last_key_value()
579            .map(|(key, node)| (key.clone(), node.clone()))
580        else {
581            return;
582        };
583
584        self.ensure_publication_snapshot();
585        let published_key = self
586            .published_keys
587            .as_ref()
588            .expect("publication identity map initialized")
589            .get(&node_identity(&last_node))
590            .cloned();
591        let Some(published_key) = published_key else {
592            self.rebuild_publication();
593            return;
594        };
595        if published_key == canonical_key {
596            return;
597        }
598
599        let repaired_consistently = {
600            let published = self.published.as_mut().expect("publication snapshot initialized");
601            let published_keys = self
602                .published_keys
603                .as_mut()
604                .expect("publication identity map initialized");
605            let removed = published.remove(&published_key);
606            let displaced = published.insert(canonical_key.clone(), last_node.clone());
607            published_keys.insert(node_identity(&last_node), canonical_key);
608            removed.is_some_and(|old_node| Arc::ptr_eq(&old_node, &last_node)) && displaced.is_none()
609        };
610        if !repaired_consistently {
611            self.rebuild_publication();
612        } else {
613            self.dirty = true;
614        }
615    }
616
617    pub(crate) fn insert(&mut self, key: T, node: Arc<RwLock<Node>>) -> Option<Arc<RwLock<Node>>> {
618        self.repair_last_route_before_insert();
619        self.insert_after_last_route_repair(key, node)
620    }
621
622    fn insert_after_last_route_repair(&mut self, key: T, node: Arc<RwLock<Node>>) -> Option<Arc<RwLock<Node>>> {
623        debug_assert!(self.publish, "generic topology insertion requires publication");
624        let replaced = self
625            .index
626            .as_deref_mut()
627            .expect("topology guard already released")
628            .insert(key.clone(), node.clone());
629        self.dirty = true;
630
631        if self.publish {
632            self.ensure_publication_snapshot();
633            if let Some(replaced) = &replaced {
634                let removed_consistently = {
635                    let published = self.published.as_mut().expect("publication snapshot initialized");
636                    let published_keys = self
637                        .published_keys
638                        .as_mut()
639                        .expect("publication identity map initialized");
640                    published_keys
641                        .remove(&node_identity(replaced))
642                        .and_then(|old_key| published.remove(&old_key))
643                        .is_some_and(|old_node| Arc::ptr_eq(&old_node, replaced))
644                };
645                if !removed_consistently {
646                    self.rebuild_publication();
647                    return Some(replaced.clone());
648                }
649            }
650
651            let displaced = self
652                .published
653                .as_mut()
654                .expect("publication snapshot initialized")
655                .insert(key.clone(), node.clone());
656            self.published_keys
657                .as_mut()
658                .expect("publication identity map initialized")
659                .insert(node_identity(&node), key);
660            if displaced.is_some_and(|old_node| !Arc::ptr_eq(&old_node, &node)) {
661                // Canonical keys are unique, so a different node cannot
662                // lawfully occupy this route. Recover once from canonical
663                // state instead of scanning and chaining while the generation
664                // remains odd.
665                self.rebuild_publication();
666            }
667        }
668
669        replaced
670    }
671
672    pub(crate) fn remove<Q>(&mut self, key: &Q) -> Option<Arc<RwLock<Node>>>
673    where
674        T: Borrow<Q>,
675        Q: Ord + ?Sized,
676    {
677        debug_assert!(self.publish, "generic topology removal requires publication");
678        let removed = self
679            .index
680            .as_deref_mut()
681            .expect("topology guard already released")
682            .remove(key)?;
683        self.dirty = true;
684
685        if self.publish {
686            self.ensure_publication_snapshot();
687            let removed_consistently = {
688                let published = self.published.as_mut().expect("publication snapshot initialized");
689                let published_keys = self
690                    .published_keys
691                    .as_mut()
692                    .expect("publication identity map initialized");
693                published_keys
694                    .remove(&node_identity(&removed))
695                    .and_then(|route_key| published.remove::<T>(&route_key))
696                    .is_some_and(|route_node| Arc::ptr_eq(&route_node, &removed))
697            };
698            if !removed_consistently {
699                self.rebuild_publication();
700            }
701        }
702
703        Some(removed)
704    }
705}
706
707impl<T, Node> Drop for TopologyWriteGuard<'_, T, Node>
708where
709    T: Ord + Clone + Send + 'static,
710    Node: Send + 'static,
711{
712    fn drop(&mut self) {
713        #[cfg(debug_assertions)]
714        if self.publish && self.dirty {
715            let canonical = self.index.as_deref().expect("topology guard exists while validating");
716            let published = self.published.as_ref().expect("publishing guard carries snapshot");
717            let published_keys = self
718                .published_keys
719                .as_deref()
720                .expect("publishing guard carries identity map");
721            debug_assert_eq!(
722                canonical.len(),
723                published.len,
724                "canonical and published node counts diverged"
725            );
726            debug_assert_eq!(
727                canonical.len(),
728                published_keys.len(),
729                "canonical and published identity counts diverged"
730            );
731            for ((_, canonical_node), (route_key, route_node)) in canonical.iter().zip(published.iter()) {
732                debug_assert!(
733                    Arc::ptr_eq(route_node, canonical_node),
734                    "canonical and published node order diverged"
735                );
736                debug_assert!(
737                    published_keys.get(&node_identity(canonical_node)) == Some(route_key),
738                    "published identity key does not match the route index"
739                );
740            }
741        }
742
743        if !self.publish {
744            drop(self.published_keys.take());
745            drop(self.index.take());
746            return;
747        }
748        if !self.dirty {
749            self.topology.generation.fetch_add(1, Ordering::Release);
750            drop(self.published_keys.take());
751            drop(self.index.take());
752            return;
753        }
754        let retired = self.topology.published.replace(
755            self.published
756                .take()
757                .expect("publishing guard must carry a read snapshot"),
758        );
759        // Readers may proceed as soon as the O(1) root publication completes.
760        // Garbage bookkeeping and the registry sweep are deliberately outside
761        // that odd-generation window.
762        self.topology.generation.fetch_add(1, Ordering::Release);
763        self.topology.published.retire(retired);
764        drop(self.published_keys.take());
765        drop(self.index.take());
766        self.topology.published.advance();
767    }
768}
769
770// Default identity-adoption hook for replace-on-equality: plain sets and maps
771// have no hidden ordering state to carry over. See
772// `MultiPairLike::adopt_stored_identity`.
773pub(crate) fn no_identity_adoption<T>(_stored: &T, _incoming: &mut T) {}
774
775// `BTreeMap::range::<Q>` requires the borrowed ordering to be identical to
776// the stored-key ordering. That is true for ordinary borrowed keys such as
777// `String`/`str`, but deliberately false for multimap entries: many distinct
778// `(key, value)` entries borrow as the same `key`. Route those lookups by the
779// borrowed view explicitly so node maxima sharing one logical key remain
780// reachable.
781fn first_for_borrowed_bound<'a, T, Q, V>(
782    index: &'a BTreeMap<T, V>,
783    bound: Bound<&Q>,
784    borrow_order_matches: bool,
785) -> Option<(&'a T, &'a V)>
786where
787    T: Ord + Borrow<Q>,
788    Q: Ord + ?Sized,
789{
790    if borrow_order_matches {
791        return index.range::<Q, _>((bound, Bound::Unbounded)).next();
792    }
793
794    index.iter().find(|(key, _)| match bound {
795        Bound::Included(value) => <T as Borrow<Q>>::borrow(key) >= value,
796        Bound::Excluded(value) => <T as Borrow<Q>>::borrow(key) > value,
797        Bound::Unbounded => true,
798    })
799}
800
801fn first_published_for_borrowed_bound<'a, T, Q, V>(
802    index: &'a PublishedNodeIndex<T, V>,
803    bound: Bound<&Q>,
804    borrow_order_matches: bool,
805) -> Option<(&'a T, &'a Arc<RwLock<V>>)>
806where
807    T: Ord + Clone + Borrow<Q>,
808    Q: Ord + ?Sized,
809{
810    if borrow_order_matches {
811        return index.first_for_bound(bound);
812    }
813
814    index.iter().find(|(key, _)| match bound {
815        Bound::Included(value) => <T as Borrow<Q>>::borrow(key) >= value,
816        Bound::Excluded(value) => <T as Borrow<Q>>::borrow(key) > value,
817        Bound::Unbounded => true,
818    })
819}
820
821fn node_for_borrowed_end<'a, T, Q, V>(
822    index: &'a BTreeMap<T, V>,
823    end: &Q,
824    borrow_order_matches: bool,
825) -> Option<(&'a T, &'a V)>
826where
827    T: Ord + Borrow<Q>,
828    Q: Ord + ?Sized,
829{
830    if borrow_order_matches {
831        return index
832            .range::<Q, _>((Bound::Included(end), Bound::Unbounded))
833            .next()
834            .or_else(|| index.last_key_value());
835    }
836
837    let mut last_equal = None;
838    for entry @ (key, _) in index {
839        match <T as Borrow<Q>>::borrow(key).cmp(end) {
840            ::core::cmp::Ordering::Less => {}
841            ::core::cmp::Ordering::Equal => last_equal = Some(entry),
842            // The first node whose maximum is above the end may still start
843            // with values inside the range. `Range::new` ranks within that
844            // node to obtain the first out-of-range sentinel.
845            ::core::cmp::Ordering::Greater => return Some(entry),
846        }
847    }
848    last_equal.or_else(|| index.last_key_value())
849}
850
851/// A **persistent** concurrent ordered set based on a B-Tree.
852///
853/// See [`BTreeMap`]'s documentation for a detailed discussion of this collection's performance
854/// benefits and drawbacks.
855///
856/// It is a logic error for an item to be modified in such a way that the item's ordering relative
857/// to any other item, as determined by the [`Ord`] trait, changes while it is in the set. This is
858/// normally only possible through [`Cell`], [`RefCell`], global state, I/O, or unsafe code.
859/// The behavior resulting from such a logic error is not specified, but will be encapsulated to the
860/// `BTreeSet` that observed the logic error and not result in undefined behavior. This could
861/// include panics, incorrect results, aborts, memory leaks, and non-termination.
862///
863/// Iterators returned by [`crate::BTreeSet::iter`] produce their items in order, and take worst-case
864/// logarithmic and amortized constant time per item returned.
865///
866/// [`Cell`]: cratestd::cell::Cell
867/// [`RefCell`]: cratestd::cell::RefCell
868///
869/// # Examples
870///
871/// ```
872/// use indexset::concurrent::set::BTreeSet;
873///
874/// // Type inference lets us omit an explicit type signature (which
875/// // would be `BTreeSet<&str>` in this example).
876/// let mut books = BTreeSet::<&str>::new();
877///
878/// // Add some books.
879/// books.insert("A Dance With Dragons");
880/// books.insert("To Kill a Mockingbird");
881/// books.insert("The Odyssey");
882/// books.insert("The Great Gatsby");
883///
884/// // Check for a specific one.
885/// if !books.contains("The Winds of Winter") {
886///     println!("We have {} books, but The Winds of Winter ain't one.",
887///              books.len());
888/// }
889///
890/// // Remove a book.
891/// books.remove("The Odyssey");
892///
893/// // Iterate over everything.
894/// for book in &books {
895///     println!("{book}");
896/// }
897/// ```
898///
899/// A `BTreeSet` with a known list of items can be initialized from an array:
900///
901/// ```
902/// use indexset::concurrent::set::BTreeSet;
903///
904/// let set = BTreeSet::from_iter([1, 2, 3]);
905/// ```
906#[derive(Debug)]
907pub struct BTreeSet<T, Node = Vec<T>>
908where
909    T: Ord + Clone + 'static,
910    Node: NodeLike<T>,
911{
912    // Writers maintain the canonical ordered topology under one structural
913    // lock and publish immutable snapshots for point reads. The read path is
914    // therefore free of a shared reader-count cache line. Node contents use
915    // independent read/write locks, so readers routed to one node may proceed
916    // concurrently while mutations retain exclusive node access.
917    pub(crate) index: Topology<T, Node>,
918    node_capacity: usize,
919    // Ordinary set/map keys satisfy Borrow's ordering contract and retain a
920    // logarithmic BTreeMap route. Multimap entries intentionally borrow only
921    // their leading key, so equal borrowed-key groups need an explicit scan.
922    borrow_order_matches: bool,
923    #[cfg(feature = "cdc")]
924    // The counter provides unique sequence numbers only. Node/global locks
925    // order conflicting mutations, and the persistence queue publishes event
926    // payloads, so the counter itself does not carry memory visibility.
927    event_id: AtomicU64,
928}
929impl<T: Ord + Clone + 'static, Node: NodeLike<T>> Default for BTreeSet<T, Node> {
930    fn default() -> Self {
931        Self {
932            index: Topology::new(),
933            node_capacity: DEFAULT_INNER_SIZE,
934            borrow_order_matches: true,
935            #[cfg(feature = "cdc")]
936            event_id: AtomicU64::new(0),
937        }
938    }
939}
940
941impl<T, Node> BTreeSet<T, Node>
942where
943    T: Debug + Ord + Clone + Send,
944    Node: NodeLike<T> + Send + 'static,
945{
946    pub fn new() -> Self {
947        Self::default()
948    }
949    /// Makes a new, empty `BTreeSet` with the given maximum node size. Allocates one vec with
950    /// the capacity set to be the specified node size.
951    ///
952    /// # Examples
953    ///
954    /// ```
955    /// use indexset::concurrent::set::BTreeSet;
956    ///
957    /// let set: BTreeSet<i32> = BTreeSet::with_maximum_node_size(128);
958    pub fn with_maximum_node_size(node_capacity: usize) -> Self {
959        Self {
960            index: Topology::new(),
961            node_capacity,
962            borrow_order_matches: true,
963            #[cfg(feature = "cdc")]
964            event_id: AtomicU64::new(0),
965        }
966    }
967    pub(crate) fn with_grouped_borrow_routing(mut self) -> Self {
968        self.borrow_order_matches = false;
969        self
970    }
971    pub fn attach_node(&self, node: Node) {
972        self.attach_nodes(::core::iter::once(node));
973    }
974
975    /// Attaches a persisted topology in one structural publication.
976    ///
977    /// Nodes must be non-empty and internally sorted. Their values, together
978    /// with any nodes already attached to this set, must form mutually ordered
979    /// non-overlapping ranges. The same preconditions as [`Self::attach_node`]
980    /// apply to every item.
981    pub fn attach_nodes(&self, nodes: impl IntoIterator<Item = Node>) {
982        let mut nodes = nodes.into_iter().peekable();
983        if nodes.peek().is_none() {
984            return;
985        }
986
987        let mut index = self.index.write();
988        index.repair_last_route_before_insert();
989        for node in nodes {
990            let node_id = node
991                .max()
992                .cloned()
993                .expect("node should contain at least one value to be correct node");
994            index.insert_after_last_route_repair(node_id, Arc::new(RwLock::new(node)));
995        }
996    }
997
998    #[cfg(feature = "cdc")]
999    pub(crate) fn export_topology(&self) -> (usize, Vec<Vec<T>>) {
1000        let index = self.index.read();
1001        let nodes = index
1002            .values()
1003            .map(|node| node.read().iter().cloned().collect())
1004            .collect();
1005        (self.node_capacity, nodes)
1006    }
1007
1008    #[allow(clippy::type_complexity)]
1009    // Const specialization keeps ordinary writes free of CDC event construction
1010    // even when the crate is compiled with the `cdc` feature.
1011    fn put_checked_inner<const EMIT_CDC: bool>(
1012        &self,
1013        value: T,
1014        adopt: fn(&T, &mut T),
1015    ) -> Result<(Option<T>, Vec<ChangeEvent<T>>), (ArcRwLockWriteGuard<RawRwLock, Node>, usize, T)> {
1016        loop {
1017            let mut cdc = vec![];
1018            let index = self.index.read();
1019            let target_node_entry = match index.range(value.clone()..).next() {
1020                Some(entry) => entry,
1021                None => {
1022                    if let Some(last) = index.last_key_value() {
1023                        last
1024                    } else {
1025                        drop(index);
1026                        let mut spins = 0;
1027                        let mut index = loop {
1028                            if let Some(guard) = self.index.try_write() {
1029                                break guard;
1030                            }
1031                            if spins >= ROOT_PUBLICATION_SPIN_LIMIT {
1032                                // A bounded block gives root publication a
1033                                // deterministic progress path under reader
1034                                // contention instead of livelocking.
1035                                break self.index.write();
1036                            }
1037                            spins += 1;
1038                            ::core::hint::spin_loop();
1039                        };
1040                        // Another first writer may have published while this
1041                        // caller was acquiring the exclusive structural guard.
1042                        if !index.is_empty() {
1043                            continue;
1044                        }
1045
1046                        let mut first_node = Node::with_capacity(self.node_capacity);
1047                        first_node.insert(value.clone());
1048
1049                        #[cfg(feature = "cdc")]
1050                        if EMIT_CDC {
1051                            let node_insertion = ChangeEvent::CreateNode {
1052                                // is correct as index is locked and current thread is the only that can
1053                                // fetch event_id.
1054                                event_id: self.event_id.fetch_add(1, Ordering::Relaxed).into(),
1055                                max_value: value.clone(),
1056                            };
1057                            cdc.push(node_insertion);
1058                        }
1059
1060                        index.insert(value, Arc::new(RwLock::new(first_node)));
1061
1062                        return Ok((None, cdc));
1063                    }
1064                }
1065            };
1066
1067            let mut node_guard = target_node_entry.1.clone().write_arc();
1068
1069            #[allow(unused_assignments)]
1070            let mut operation = None;
1071            if !node_guard.need_to_split(self.node_capacity, &value) {
1072                let old_max = node_guard.max().cloned();
1073                let (inserted, idx) = NodeLike::insert(&mut *node_guard, value.clone());
1074                if inserted {
1075                    #[cfg(feature = "cdc")]
1076                    if EMIT_CDC {
1077                        let node_element_insertion = ChangeEvent::InsertAt {
1078                            // is correct as node is locked and current thread is the only that can
1079                            // fetch event_id, so events for this node will have monotonic id's.
1080                            event_id: self.event_id.fetch_add(1, Ordering::Relaxed).into(),
1081                            max_value: old_max.clone().unwrap_or(value.clone()),
1082                            value: value.clone(),
1083                            index: idx,
1084                        };
1085                        cdc.push(node_element_insertion);
1086                    }
1087
1088                    if node_guard.max().cloned() == old_max {
1089                        return Ok((None, cdc));
1090                    }
1091
1092                    // The node's maximum changed, so its index entry must be
1093                    // re-keyed. Address the repair by the entry's CURRENT key,
1094                    // not the observed maximum: during a stale-key window (a
1095                    // concurrent writer changed the maximum but its own repair
1096                    // has not committed, or a remove emptied the node before
1097                    // this insert refilled it, leaving `old_max` as `None`)
1098                    // the two differ, and a repair addressed by the maximum
1099                    // misses the entry at commit time and is silently dropped,
1100                    // leaving the entry permanently stale (unreachable values,
1101                    // and a pending `MakeUnreachable` could unlink the node
1102                    // containing this acknowledged insert).
1103                    operation = Some(Operation::UpdateMax(
1104                        target_node_entry.1.clone(),
1105                        target_node_entry.0.clone(),
1106                    ));
1107                } else {
1108                    return Err((node_guard, idx, old_max.unwrap()));
1109                }
1110            } else {
1111                operation = Some(Operation::Split(
1112                    target_node_entry.1.clone(),
1113                    target_node_entry.0.clone(),
1114                    value.clone(),
1115                ));
1116            }
1117
1118            drop(node_guard);
1119            drop(index);
1120
1121            let op = operation.unwrap();
1122            let mut index = match &op {
1123                Operation::UpdateMax(_, _) => self.index.write_rekey(),
1124                Operation::Split(_, _, _) | Operation::MakeUnreachable(_, _) => self.index.write(),
1125            };
1126            match &op {
1127                Operation::Split(_, _, _) => {
1128                    if let Ok((value, value_cdc)) = op.commit::<EMIT_CDC>(&mut index, adopt) {
1129                        #[cfg(feature = "cdc")]
1130                        if EMIT_CDC {
1131                            for unassigned_event in value_cdc {
1132                                let event_id = self.event_id.fetch_add(1, Ordering::Relaxed).into();
1133                                cdc.push(unassigned_event.assign_id(event_id));
1134                            }
1135                        }
1136                        return Ok((value, cdc));
1137                    } else {
1138                        continue;
1139                    }
1140                }
1141                Operation::UpdateMax(_, _) => {
1142                    return if let Ok((value, value_cdc)) = op.commit::<EMIT_CDC>(&mut index, adopt) {
1143                        #[cfg(feature = "cdc")]
1144                        if EMIT_CDC {
1145                            for unassigned_event in value_cdc {
1146                                let event_id = self.event_id.fetch_add(1, Ordering::Relaxed).into();
1147                                cdc.push(unassigned_event.assign_id(event_id));
1148                            }
1149                        }
1150                        Ok((value, cdc))
1151                    } else {
1152                        Ok((None, cdc))
1153                    }
1154                }
1155                Operation::MakeUnreachable(_, _) => unreachable!(),
1156            }
1157        }
1158    }
1159    fn put_inner<const EMIT_CDC: bool>(&self, value: T, adopt: fn(&T, &mut T)) -> (Option<T>, Vec<ChangeEvent<T>>) {
1160        match self.put_checked_inner::<EMIT_CDC>(value.clone(), adopt) {
1161            Ok(res) => res,
1162            Err((mut node_guard, idx, max)) => {
1163                // Replace-on-logical-equality: let the incoming value adopt
1164                // the stored value's hidden ordering state before it takes
1165                // the stored position (see MultiPairLike::adopt_stored_identity).
1166                let mut value = value;
1167                if let Some(stored) = node_guard.get_ith(idx) {
1168                    adopt(stored, &mut value);
1169                }
1170                let mut cdc = vec![];
1171                #[cfg(feature = "cdc")]
1172                if EMIT_CDC {
1173                    if node_guard.len() == 1 {
1174                        let node_removal = ChangeEvent::RemoveNode {
1175                            // is correct as node is locked and current thread is the only that can
1176                            // fetch event_id, so events for this node will have monotonic id's.
1177                            event_id: self.event_id.fetch_add(1, Ordering::Relaxed).into(),
1178                            max_value: max.clone(),
1179                        };
1180                        let node_insertion = ChangeEvent::CreateNode {
1181                            // same as for previous.
1182                            event_id: self.event_id.fetch_add(1, Ordering::Relaxed).into(),
1183                            max_value: value.clone(),
1184                        };
1185                        cdc.push(node_removal);
1186                        cdc.push(node_insertion);
1187                    } else if idx == node_guard.len() - 1 {
1188                        let new_max = if node_guard.len() <= 1 {
1189                            None
1190                        } else {
1191                            node_guard.get_ith(node_guard.len() - 2)
1192                        };
1193                        let node_element_removal = ChangeEvent::RemoveAt {
1194                            // is correct as node is locked and current thread is the only that can
1195                            // fetch event_id, so events for this node will have monotonic id's.
1196                            event_id: self.event_id.fetch_add(1, Ordering::Relaxed).into(),
1197                            max_value: max.clone(),
1198                            value: value.clone(),
1199                            index: idx,
1200                        };
1201                        let node_element_insertion = ChangeEvent::InsertAt {
1202                            // same as for previous.
1203                            event_id: self.event_id.fetch_add(1, Ordering::Relaxed).into(),
1204                            max_value: new_max.expect("length was checked so should be ok").clone(),
1205                            value: value.clone(),
1206                            index: idx,
1207                        };
1208                        cdc.push(node_element_removal);
1209                        cdc.push(node_element_insertion);
1210                    } else {
1211                        let node_element_removal = ChangeEvent::RemoveAt {
1212                            // is correct as node is locked and current thread is the only that can
1213                            // fetch event_id, so events for this node will have monotonic id's.
1214                            event_id: self.event_id.fetch_add(1, Ordering::Relaxed).into(),
1215                            max_value: max.clone(),
1216                            value: value.clone(),
1217                            index: idx,
1218                        };
1219                        let node_element_insertion = ChangeEvent::InsertAt {
1220                            // same as for previous.
1221                            event_id: self.event_id.fetch_add(1, Ordering::Relaxed).into(),
1222                            max_value: max.clone(),
1223                            value: value.clone(),
1224                            index: idx,
1225                        };
1226                        cdc.push(node_element_removal);
1227                        cdc.push(node_element_insertion);
1228                    }
1229                }
1230
1231                (NodeLike::replace(&mut *node_guard, idx, value.clone()), cdc)
1232            }
1233        }
1234    }
1235
1236    pub(crate) fn put(&self, value: T) -> Option<T> {
1237        self.put_inner::<false>(value, no_identity_adoption).0
1238    }
1239
1240    #[cfg(feature = "multimap")]
1241    pub(crate) fn put_with(&self, value: T, adopt: fn(&T, &mut T)) -> Option<T> {
1242        self.put_inner::<false>(value, adopt).0
1243    }
1244
1245    #[allow(clippy::type_complexity)]
1246    pub(crate) fn put_checked(
1247        &self,
1248        value: T,
1249    ) -> Result<(Option<T>, Vec<ChangeEvent<T>>), (ArcRwLockWriteGuard<RawRwLock, Node>, usize, T)> {
1250        self.put_checked_inner::<false>(value, no_identity_adoption)
1251    }
1252
1253    pub(crate) fn put_cdc(&self, value: T) -> (Option<T>, Vec<ChangeEvent<T>>) {
1254        self.put_inner::<true>(value, no_identity_adoption)
1255    }
1256
1257    #[cfg(all(feature = "multimap", feature = "cdc"))]
1258    pub(crate) fn put_cdc_with(&self, value: T, adopt: fn(&T, &mut T)) -> (Option<T>, Vec<ChangeEvent<T>>) {
1259        self.put_inner::<true>(value, adopt)
1260    }
1261
1262    #[allow(clippy::type_complexity)]
1263    pub(crate) fn put_cdc_checked(
1264        &self,
1265        value: T,
1266    ) -> Result<(Option<T>, Vec<ChangeEvent<T>>), (ArcRwLockWriteGuard<RawRwLock, Node>, usize, T)> {
1267        self.put_checked_inner::<true>(value, no_identity_adoption)
1268    }
1269
1270    /// Adds a value to the set.
1271    ///
1272    /// Returns whether the value was newly inserted. That is:
1273    ///
1274    /// - If the set did not previously contain an equal value, `true` is
1275    ///   returned.
1276    /// - If the set already contained an equal value, `false` is returned, and
1277    ///   the entry is not updated.
1278    ///
1279    /// # Examples
1280    ///
1281    /// ```
1282    /// use indexset::concurrent::set::BTreeSet;
1283    ///
1284    /// let mut set = BTreeSet::<usize>::new();
1285    ///
1286    /// assert_eq!(set.insert(2), true);
1287    /// assert_eq!(set.insert(2), false);
1288    /// assert_eq!(set.len(), 1);
1289    /// ```
1290    pub fn insert(&self, value: T) -> bool {
1291        self.put(value).is_none()
1292    }
1293    // See `put_checked_inner`: this is const-specialized to avoid paying for
1294    // discarded events in the ordinary `remove` path.
1295    fn remove_inner<const EMIT_CDC: bool, Q>(&self, value: &Q) -> (Option<T>, Vec<ChangeEvent<T>>)
1296    where
1297        T: Borrow<Q>,
1298        Q: Ord + ?Sized,
1299    {
1300        let mut cdc = vec![];
1301        let index = self.index.read();
1302        // Fall back to the last node when the value sorts above every index
1303        // key, exactly like `put` and `lock_node_for_value`: during a stale-key
1304        // window (a node whose maximum grew before its UpdateMax repair
1305        // committed) the value lives in the last node even though no index key
1306        // covers it. Without the fallback such a value is un-removable while
1307        // `contains` still finds it.
1308        if let Some(target_node_entry) =
1309            first_for_borrowed_bound(&index, Bound::Included(value), self.borrow_order_matches)
1310                .or_else(|| index.last_key_value())
1311        {
1312            let mut node_guard = target_node_entry.1.clone().write_arc();
1313            let old_max = node_guard.max().cloned();
1314            let deleted = NodeLike::delete(&mut *node_guard, value);
1315            if deleted.is_none() {
1316                return (None, cdc);
1317            }
1318            let (deleted, idx) = deleted.expect("should be ok as checked before");
1319
1320            let operation = if node_guard.len() > 0 {
1321                #[cfg(feature = "cdc")]
1322                if EMIT_CDC {
1323                    let node_element_removal = ChangeEvent::RemoveAt {
1324                        // is correct as node is locked and current thread is the only that can
1325                        // fetch event_id, so events for this node will have monotonic id's.
1326                        event_id: self.event_id.fetch_add(1, Ordering::Relaxed).into(),
1327                        max_value: old_max.clone().expect("Max value should exist as Node is not empty"),
1328                        value: deleted.clone(),
1329                        index: idx,
1330                    };
1331                    cdc.push(node_element_removal);
1332                }
1333
1334                if old_max.as_ref() == node_guard.max() {
1335                    return (Some(deleted), cdc);
1336                }
1337
1338                // Address the repair by the entry's current key, not by the
1339                // observed old maximum: see `put_checked_inner`. In a
1340                // stale-key window they differ, and a repair addressed by the
1341                // maximum is dropped at commit time, leaving the entry stale.
1342                Some(Operation::UpdateMax(
1343                    target_node_entry.1.clone(),
1344                    target_node_entry.0.clone(),
1345                ))
1346            } else {
1347                Some(Operation::MakeUnreachable(
1348                    target_node_entry.1.clone(),
1349                    target_node_entry.0.clone(),
1350                ))
1351            };
1352
1353            drop(node_guard);
1354            drop(index);
1355
1356            let operation = operation.unwrap();
1357            let mut index = match &operation {
1358                Operation::UpdateMax(_, _) => self.index.write_rekey(),
1359                Operation::Split(_, _, _) | Operation::MakeUnreachable(_, _) => self.index.write(),
1360            };
1361
1362            return if let Ok((_, value_cdc)) = operation.commit::<EMIT_CDC>(&mut index, no_identity_adoption) {
1363                #[cfg(feature = "cdc")]
1364                if EMIT_CDC {
1365                    for unassigned_event in value_cdc {
1366                        let event_id = self.event_id.fetch_add(1, Ordering::Relaxed).into();
1367                        cdc.push(unassigned_event.assign_id(event_id));
1368                    }
1369                }
1370                (Some(deleted), cdc)
1371            } else {
1372                (Some(deleted), cdc)
1373            };
1374        }
1375
1376        (None, vec![])
1377    }
1378
1379    pub fn remove_cdc<Q>(&self, value: &Q) -> (Option<T>, Vec<ChangeEvent<T>>)
1380    where
1381        T: Borrow<Q>,
1382        Q: Ord + ?Sized,
1383    {
1384        self.remove_inner::<true, Q>(value)
1385    }
1386    /// If the set contains an element equal to the value, removes it from the
1387    /// set and drops it. Returns whether such an element was present.
1388    ///
1389    /// The value may be any borrowed form of the set's element type,
1390    /// but the ordering on the borrowed form *must* match the
1391    /// ordering on the element type.
1392    ///
1393    /// # Examples
1394    ///
1395    /// ```
1396    /// use indexset::concurrent::set::BTreeSet;
1397    ///
1398    /// let mut set = BTreeSet::<usize>::new();
1399    ///
1400    /// set.insert(2);
1401    /// assert_eq!(set.remove(&2).is_some(), true);
1402    /// assert_eq!(set.remove(&2).is_some(), false);
1403    /// ```
1404    pub fn remove<Q>(&self, value: &Q) -> Option<T>
1405    where
1406        T: Borrow<Q>,
1407        Q: Ord + ?Sized,
1408    {
1409        self.remove_inner::<false, Q>(value).0
1410    }
1411
1412    // Slow-path recovery for multimap exact removal. Holding the structural
1413    // write guard makes predicate lookup, deletion, and node reindexing one
1414    // critical section after the ordinary point-removal path has missed. Only
1415    // the multimap paths use this, and it relies on NodeLike::delete_at (also
1416    // multimap-gated), so gate the whole family to avoid an unconditional break.
1417    #[inline(always)]
1418    fn lock_node_for_value_optimistic<Q>(&self, value: &Q) -> Option<ArcRwLockReadGuard<RawRwLock, Node>>
1419    where
1420        T: Borrow<Q>,
1421        Q: Ord + ?Sized,
1422    {
1423        let node = {
1424            let index = self.index.read();
1425            match first_for_borrowed_bound(&index, Bound::Included(value), self.borrow_order_matches) {
1426                Some((_, node)) => Some(node.clone()),
1427                None => index
1428                    .last_key_value()
1429                    .map(|(_, node)| node.clone())
1430                    .or_else(|| index.first_key_value().map(|(_, node)| node.clone())),
1431            }
1432        }?;
1433        Some(node.read_arc())
1434    }
1435
1436    /// Locates and locks the node whose structural range owns `value`.
1437    ///
1438    /// Readers route through an immutable published topology and validate its
1439    /// generation after locking the node. A concurrent structural change makes
1440    /// the read retry, so hits and misses remain definitive without updating a
1441    /// shared reader-count cache line.
1442    #[inline(always)]
1443    fn lock_node_for_value<Q>(&self, value: &Q) -> Option<ArcRwLockReadGuard<RawRwLock, Node>>
1444    where
1445        T: Borrow<Q>,
1446        Q: Ord + ?Sized,
1447    {
1448        let mut retries = 0;
1449        let mut writer_spins = 0;
1450
1451        loop {
1452            let generation = self.index.generation.load(Ordering::Acquire);
1453            if !generation.is_multiple_of(2) {
1454                if writer_spins < ROOT_PUBLICATION_SPIN_LIMIT {
1455                    writer_spins += 1;
1456                    ::core::hint::spin_loop();
1457                } else {
1458                    yield_now();
1459                }
1460                continue;
1461            }
1462            writer_spins = 0;
1463
1464            if retries >= STABLE_READ_BLOCKING_FALLBACK_AFTER {
1465                // Bounded progress fallback: hold the canonical topology read
1466                // guard while acquiring the node. Structural writers follow
1467                // the same topology-before-node order.
1468                let index = self.index.read();
1469                let node = match first_for_borrowed_bound(&index, Bound::Included(value), self.borrow_order_matches) {
1470                    Some((_, node)) => Some(node.clone()),
1471                    None => index
1472                        .last_key_value()
1473                        .map(|(_, node)| node.clone())
1474                        .or_else(|| index.first_key_value().map(|(_, node)| node.clone())),
1475                }?;
1476                let node_guard = node.read_arc();
1477                drop(index);
1478                return Some(node_guard);
1479            }
1480
1481            let pin = self.index.published.domain.pin();
1482            let snapshot = self.index.published.current.load(Ordering::Acquire);
1483            // SAFETY: `snapshot` was loaded after `pin`, and the publication
1484            // domain cannot reclaim it until `pin` is dropped.
1485            let index = unsafe { &*snapshot };
1486            let node =
1487                match first_published_for_borrowed_bound(index, Bound::Included(value), self.borrow_order_matches) {
1488                    Some((_, node)) => Some(node.clone()),
1489                    None => index
1490                        .last_key_value()
1491                        .map(|(_, node)| node.clone())
1492                        .or_else(|| index.first_key_value().map(|(_, node)| node.clone())),
1493                };
1494            let Some(node) = node else {
1495                if self.index.generation.load(Ordering::Acquire) == generation {
1496                    return None;
1497                }
1498                retries += 1;
1499                continue;
1500            };
1501
1502            // The snapshot pin protects the Arc only until it is cloned. Drop
1503            // it before the potentially blocking node acquisition so a slow
1504            // node writer cannot stall topology reclamation.
1505            drop(pin);
1506            let node_guard = node.read_arc();
1507            if self.index.generation.load(Ordering::Acquire) == generation {
1508                return Some(node_guard);
1509            }
1510            retries += 1;
1511        }
1512    }
1513
1514    #[inline(always)]
1515    fn get_with_guard<Q, R>(
1516        node_guard: ArcRwLockReadGuard<RawRwLock, Node>,
1517        value: &Q,
1518        read: impl FnOnce(&T) -> R,
1519    ) -> Option<R>
1520    where
1521        T: Borrow<Q>,
1522        Q: Ord + ?Sized,
1523    {
1524        let position = node_guard.try_select(value)?;
1525        node_guard.get_ith(position).map(read)
1526    }
1527
1528    #[inline(always)]
1529    pub(crate) fn get_with<Q, R>(&self, value: &Q, read: impl FnOnce(&T) -> R) -> Option<R>
1530    where
1531        T: Borrow<Q>,
1532        Q: Ord + ?Sized,
1533    {
1534        let mut retries = 0;
1535        let mut writer_spins = 0;
1536        let mut read = Some(read);
1537
1538        loop {
1539            let generation = self.index.generation.load(Ordering::Acquire);
1540            if !generation.is_multiple_of(2) {
1541                if writer_spins < ROOT_PUBLICATION_SPIN_LIMIT {
1542                    writer_spins += 1;
1543                    ::core::hint::spin_loop();
1544                } else {
1545                    yield_now();
1546                }
1547                continue;
1548            }
1549            writer_spins = 0;
1550
1551            if retries >= STABLE_READ_BLOCKING_FALLBACK_AFTER {
1552                let index = self.index.read();
1553                let node = first_for_borrowed_bound(&index, Bound::Included(value), self.borrow_order_matches)
1554                    .or_else(|| index.last_key_value())
1555                    .or_else(|| index.first_key_value())
1556                    .map(|(_, node)| node.clone())?;
1557                let node_guard = node.read_arc();
1558                drop(index);
1559                let position = node_guard.try_select(value)?;
1560                return node_guard
1561                    .get_ith(position)
1562                    .map(read.take().expect("read closure is consumed only on return"));
1563            }
1564
1565            let pin = self.index.published.domain.pin();
1566            let snapshot = self.index.published.current.load(Ordering::Acquire);
1567            // SAFETY: `snapshot` was loaded after `pin`, and the publication
1568            // domain cannot reclaim it until `pin` is dropped.
1569            let index = unsafe { &*snapshot };
1570            let node = first_published_for_borrowed_bound(index, Bound::Included(value), self.borrow_order_matches)
1571                .or_else(|| index.last_key_value())
1572                .or_else(|| index.first_key_value())
1573                .map(|(_, node)| node);
1574            let Some(node) = node else {
1575                if self.index.generation.load(Ordering::Acquire) == generation {
1576                    return None;
1577                }
1578                retries += 1;
1579                continue;
1580            };
1581
1582            // Borrow the Arc from the protected snapshot: unlike
1583            // `lock_node_for_value`, this owned-result path does not need an
1584            // Arc clone or its shared refcount update when the node is free.
1585            // On contention, clone it and release the reclamation pin before
1586            // blocking so a node writer cannot retain old topology paths.
1587            if let Some(node_guard) = node.try_read() {
1588                if self.index.generation.load(Ordering::Acquire) != generation {
1589                    retries += 1;
1590                    continue;
1591                }
1592                let position = node_guard.try_select(value)?;
1593                return node_guard
1594                    .get_ith(position)
1595                    .map(read.take().expect("read closure is consumed only on return"));
1596            }
1597
1598            let node = node.clone();
1599            drop(pin);
1600            let node_guard = node.read_arc();
1601            if self.index.generation.load(Ordering::Acquire) != generation {
1602                retries += 1;
1603                continue;
1604            }
1605            let position = node_guard.try_select(value)?;
1606            return node_guard
1607                .get_ith(position)
1608                .map(read.take().expect("read closure is consumed only on return"));
1609        }
1610    }
1611
1612    #[inline(always)]
1613    pub(crate) fn get_with_optimistic<Q, R>(&self, value: &Q, read: impl FnOnce(&T) -> R) -> Option<R>
1614    where
1615        T: Borrow<Q>,
1616        Q: Ord + ?Sized,
1617    {
1618        Self::get_with_guard(self.lock_node_for_value_optimistic(value)?, value, read)
1619    }
1620
1621    /// Returns `true` if the set contains an element equal to the value.
1622    ///
1623    /// The value may be any borrowed form of the set's element type,
1624    /// but the ordering on the borrowed form *must* match the
1625    /// ordering on the element type.
1626    ///
1627    /// # Examples
1628    ///
1629    /// ```
1630    /// use indexset::concurrent::set::BTreeSet;
1631    ///
1632    /// let set = BTreeSet::from_iter([1, 2, 3]);
1633    /// assert_eq!(set.contains(&1), true);
1634    /// assert_eq!(set.contains(&4), false);
1635    /// ```
1636    pub fn contains<Q>(&self, value: &Q) -> bool
1637    where
1638        T: Borrow<Q>,
1639        Q: Ord + ?Sized,
1640    {
1641        self.get_with(value, |_| ()).is_some()
1642    }
1643    pub fn get<'a, Q>(&'a self, value: &'a Q) -> Option<Ref<T, Node>>
1644    where
1645        T: Borrow<Q>,
1646        Q: Ord + ?Sized,
1647    {
1648        if let Some(node_guard) = self.lock_node_for_value(value) {
1649            let potential_position = node_guard.try_select(value);
1650
1651            if let Some(position) = potential_position {
1652                return Some(Ref {
1653                    node_guard,
1654                    position,
1655                    phantom_data: PhantomData,
1656                });
1657            }
1658        }
1659
1660        None
1661    }
1662
1663    pub fn len(&self) -> usize {
1664        self.index.read().values().map(|node| node.read().len()).sum()
1665    }
1666    pub fn is_empty(&self) -> bool {
1667        self.index.read().values().all(|node| node.read().is_empty())
1668    }
1669    pub fn capacity(&self) -> usize {
1670        self.index
1671            .read()
1672            .values()
1673            .map(|node| {
1674                let guard = node.read();
1675                guard.capacity()
1676            })
1677            .sum()
1678    }
1679    pub fn node_count(&self) -> usize {
1680        self.index.read().len()
1681    }
1682}
1683
1684impl<T> FromIterator<T> for BTreeSet<T>
1685where
1686    T: Debug + Ord + Clone + Send,
1687{
1688    fn from_iter<K: IntoIterator<Item = T>>(iter: K) -> Self {
1689        let btree = BTreeSet::new();
1690        iter.into_iter().for_each(|item| {
1691            btree.insert(item);
1692        });
1693
1694        btree
1695    }
1696}
1697
1698impl<T, const N: usize> From<[T; N]> for BTreeSet<T>
1699where
1700    T: Debug + Ord + Clone + Send,
1701{
1702    fn from(value: [T; N]) -> Self {
1703        let btree: BTreeSet<T> = Default::default();
1704
1705        value.into_iter().for_each(|item| {
1706            btree.insert(item);
1707        });
1708
1709        btree
1710    }
1711}
1712
1713/// An owned-yield iterator over a concurrent `BTreeSet`.
1714///
1715/// The iterator clones one node's remaining elements into an owned batch
1716/// while holding that node's mutex, releases the mutex, and then yields the
1717/// clones. No node lock and no structural lock is ever held between calls to
1718/// `next`/`next_back`, and every yielded `T` is an independent clone: items
1719/// collected from this iterator stay valid under arbitrary concurrent
1720/// mutation of the set.
1721///
1722/// The scan is weakly consistent, exactly like iterating any concurrent
1723/// collection: elements inserted or removed while the scan is in flight may
1724/// or may not be observed, but elements present for the whole scan are
1725/// yielded exactly once, in order.
1726pub struct Iter<'a, T, Node>
1727where
1728    T: Debug + Ord + Clone + Send + 'static,
1729    Node: NodeLike<T> + Send + 'static,
1730{
1731    tree: &'a BTreeSet<T, Node>,
1732    current_front_batch: Option<alloc::vec::IntoIter<T>>,
1733    current_back_batch: Option<alloc::vec::IntoIter<T>>,
1734    // Identity of the node the last batch in each direction was cloned from,
1735    // so the next install can step past it when the cursor lookup lands on
1736    // it again (its entry key can sit past every element it still holds).
1737    exhausted_front_node: Option<Arc<RwLock<Node>>>,
1738    exhausted_back_node: Option<Arc<RwLock<Node>>>,
1739    // The node a direction is partway through, and how many of its elements it
1740    // has taken. A batch that stops short of a node's end must resume inside
1741    // that node, and must never take less than it already has: the rank-based
1742    // skip alone cannot guarantee that, because a repositioned node can leave
1743    // the cursor ranking below elements already yielded. Recording the count
1744    // makes forward progress structural rather than incidental.
1745    front_partial: Option<(Arc<RwLock<Node>>, usize)>,
1746    back_partial: Option<(Arc<RwLock<Node>>, usize)>,
1747    // How many elements the next batch may clone, doubling per install.
1748    front_batch_limit: usize,
1749    back_batch_limit: usize,
1750    current_front_value: Option<T>,
1751    current_back_value: Option<T>,
1752    met: bool,
1753}
1754
1755/// Elements the first batch of a scan clones.
1756///
1757/// A one-element range is the common case and it used to clone a whole node to
1758/// yield one value. Starting small makes that cost proportional to what is
1759/// asked for; doubling means a real scan reaches whole-node batches after a
1760/// handful of installs and pays the same total clone count it always did.
1761const INITIAL_BATCH: usize = 4;
1762
1763/// Ceiling on the growth. `available` bounds a batch to what the node holds, so
1764/// this only stops the doubling running away on a very long scan.
1765const MAX_BATCH: usize = 4096;
1766
1767impl<'a, T, Node> Iter<'a, T, Node>
1768where
1769    T: Debug + Ord + Clone + Send + 'static,
1770    Node: NodeLike<T> + Send + 'static,
1771{
1772    pub fn new(btree: &'a BTreeSet<T, Node>) -> Self {
1773        // No node is chosen here: each direction positions itself from its
1774        // cursor when it installs a batch, atomically with reading the node.
1775        // Choosing a node ahead of time and locking it later reintroduces
1776        // the split-migration window closed in `install_front_batch`.
1777        Self {
1778            tree: btree,
1779            current_front_batch: None,
1780            current_back_batch: None,
1781            exhausted_front_node: None,
1782            exhausted_back_node: None,
1783            front_partial: None,
1784            back_partial: None,
1785            front_batch_limit: INITIAL_BATCH,
1786            back_batch_limit: INITIAL_BATCH,
1787            current_front_value: None,
1788            current_back_value: None,
1789            met: false,
1790        }
1791    }
1792
1793    // Select the node covering the forward cursor and clone its remaining
1794    // elements (strictly above the cursor) into an owned batch. Returns
1795    // false when no node remains and the scan is complete.
1796    //
1797    // Selection and the content read are ONE atomic step: the node is
1798    // locked while the structural read guard is still held. Both split
1799    // commits and re-keys take the structural write lock and the node lock,
1800    // so the chosen node cannot change contents or move between being
1801    // chosen and being read. Choosing under one guard and locking later
1802    // allowed a split to migrate not-yet-yielded elements into a new node
1803    // past the resume point, silently truncating the scan (deterministically
1804    // reproduced by `backward_scan_does_not_skip_values_split_away_after_positioning`).
1805    //
1806    // Lock order is topology then node, the same order every writer
1807    // uses, so this cannot deadlock ABBA against committers; holding the
1808    // node mutex alone pins its contents, so the structural guard is
1809    // released before the clone. No lock is held once this returns.
1810    //
1811    // One batched clone per node replaces the old per-item guard-holding
1812    // scheme: the previous design kept the node mutex alive inside the
1813    // iterator and transmuted the guard's slice iterator to the iterator's
1814    // lifetime, which let callers hold `&T` into node storage after the
1815    // guard was released (a use-after-free under concurrent mutation).
1816    // Owned batches make that impossible by construction.
1817    fn install_front_batch(&mut self) -> bool {
1818        let index = self.tree.index.read();
1819        let candidate = match self.current_front_value.as_ref() {
1820            Some(last_yielded) => index.range((Bound::Excluded(last_yielded), Bound::Unbounded)).next(),
1821            None => index.first_key_value(),
1822        };
1823        // Advance by the scan cursor with one logarithmic lookup: it lands
1824        // on whichever node now covers the cursor even after re-keys or
1825        // removals, and the yield-path filters skip anything already
1826        // yielded. Step past the just-exhausted node by identity so the
1827        // scan always makes progress.
1828        let entry = match (candidate, self.exhausted_front_node.as_ref()) {
1829            (Some((key, node)), Some(exhausted)) if Arc::ptr_eq(node, exhausted) => {
1830                index.range((Bound::Excluded(key), Bound::Unbounded)).next()
1831            }
1832            (candidate, _) => candidate,
1833        };
1834        let Some((_, entry)) = entry else {
1835            return false;
1836        };
1837        let node = entry.clone();
1838        let guard = node.read_arc();
1839        drop(index);
1840
1841        let rank_skip = self
1842            .current_front_value
1843            .as_ref()
1844            .and_then(|value| guard.rank(Bound::Excluded(value), true))
1845            .map_or(0, |rank| rank + 1);
1846        // Resuming a node this scan is partway through. `front_partial` counts
1847        // *positions*, and a position is not a stable cursor: deleting an
1848        // element this scan already yielded shifts the unyielded tail left
1849        // while the count stays put. Letting it win a `max` against the value
1850        // rank steps over an element that was present for the whole scan,
1851        // which is the one thing this iterator promises not to do. See
1852        // `deleting_a_yielded_element_does_not_skip_a_live_one`.
1853        //
1854        // The value rank is authoritative wherever there is one: it counts the
1855        // elements at or below the last yielded value, so the batch resumes
1856        // strictly above the cursor. No duplicates, and progress every time.
1857        // That is also what makes dropping the `max` safe -- the
1858        // non-termination it guarded against was a batch that came back all
1859        // duplicates and advanced nothing, and a value-ranked resume cannot
1860        // produce one.
1861        //
1862        // The position still matters before anything has been yielded, where
1863        // there is no value to rank against and it is the only record that this
1864        // node was already drawn from.
1865        let partial_skip = match self.front_partial.as_ref() {
1866            Some((partial, taken)) if Arc::ptr_eq(partial, &node) => *taken,
1867            _ => 0,
1868        };
1869        let skip = if self.current_front_value.is_some() {
1870            rank_skip
1871        } else {
1872            partial_skip
1873        };
1874
1875        // Clone what was asked for rather than the rest of the node. A range
1876        // that yields one value used to clone every remaining element of the
1877        // node it landed in, which is where the 2.6x cost of this path came
1878        // from; the owned batch is what makes the iterator sound, but nothing
1879        // about that soundness required cloning eagerly.
1880        let available = guard.len().saturating_sub(skip);
1881        let take = available.min(self.front_batch_limit);
1882        let batch = guard.iter().skip(skip).take(take).cloned().collect::<Vec<_>>();
1883        drop(guard);
1884
1885        if take == available {
1886            // The node is finished, so the next install must step past it.
1887            self.exhausted_front_node = Some(node);
1888            self.front_partial = None;
1889        } else {
1890            // More of this node remains: resume inside it rather than stepping
1891            // past, and remember how far in.
1892            self.exhausted_front_node = None;
1893            self.front_partial = Some((node, skip + take));
1894        }
1895        self.front_batch_limit = self.front_batch_limit.saturating_mul(2).min(MAX_BATCH);
1896        self.current_front_batch = Some(batch.into_iter());
1897        true
1898    }
1899
1900    // Mirror of `install_front_batch` for the backward cursor: select the
1901    // node covering the cursor (the last node when every entry key sits
1902    // below it) and clone the elements strictly below the cursor.
1903    fn install_back_batch(&mut self) -> bool {
1904        let index = self.tree.index.read();
1905        let candidate = match self.current_back_value.as_ref() {
1906            Some(last_yielded) => index
1907                .range((Bound::Included(last_yielded), Bound::Unbounded))
1908                .next()
1909                .or_else(|| index.last_key_value()),
1910            None => index.last_key_value(),
1911        };
1912        let entry = match (candidate, self.exhausted_back_node.as_ref()) {
1913            (Some((key, node)), Some(exhausted)) if Arc::ptr_eq(node, exhausted) => index.range(..key).next_back(),
1914            (candidate, _) => candidate,
1915        };
1916        let Some((_, entry)) = entry else {
1917            return false;
1918        };
1919        let node = entry.clone();
1920        let guard = node.read_arc();
1921        drop(index);
1922
1923        let truncate = self
1924            .current_back_value
1925            .as_ref()
1926            .and_then(|value| guard.rank(Bound::Excluded(value), false))
1927            .map_or(0, |rank| rank + 1);
1928        // Mirror of the forward cursor's partial resume, defect included:
1929        // walking backwards the count already taken is trimmed from the end
1930        // rather than skipped at the start, and removing an already-yielded
1931        // high element shifts the unyielded head right while the count stays
1932        // put. The value rank wins here for the same reason it wins there. See
1933        // `deleting_a_yielded_element_backwards_does_not_skip_a_live_one`.
1934        let partial_truncate = match self.back_partial.as_ref() {
1935            Some((partial, taken)) if Arc::ptr_eq(partial, &node) => *taken,
1936            _ => 0,
1937        };
1938        let truncate = if self.current_back_value.is_some() {
1939            truncate
1940        } else {
1941            partial_truncate
1942        };
1943        let available = guard.len().saturating_sub(truncate);
1944        let take = available.min(self.back_batch_limit);
1945        // The backward batch is the last `take` of what remains, so the skip is
1946        // whatever sits below it.
1947        let skip = available - take;
1948        let batch = guard.iter().skip(skip).take(take).cloned().collect::<Vec<_>>();
1949        drop(guard);
1950
1951        if take == available {
1952            self.exhausted_back_node = Some(node);
1953            self.back_partial = None;
1954        } else {
1955            self.exhausted_back_node = None;
1956            self.back_partial = Some((node, truncate + take));
1957        }
1958        self.back_batch_limit = self.back_batch_limit.saturating_mul(2).min(MAX_BATCH);
1959        self.current_back_batch = Some(batch.into_iter());
1960        true
1961    }
1962}
1963
1964impl<'a, T, Node> Iterator for Iter<'a, T, Node>
1965where
1966    T: Debug + Ord + Clone + Send + 'static,
1967    Node: NodeLike<T> + Send + 'static,
1968{
1969    type Item = T;
1970
1971    fn next(&mut self) -> Option<Self::Item> {
1972        loop {
1973            if self.met {
1974                return None;
1975            }
1976
1977            if self.current_front_batch.is_none() && !self.install_front_batch() {
1978                return None;
1979            }
1980
1981            let batch = self.current_front_batch.as_mut().expect("installed above");
1982            if let Some(value) = batch.next() {
1983                // A batch installed after repositioning can re-expose
1984                // elements at or below the last yielded value (a split
1985                // re-distributes the just-finished node, a repositioned node
1986                // covers part of the scanned range). Skip them instead of
1987                // yielding duplicates.
1988                if let Some(current_front_value) = self.current_front_value.as_ref() {
1989                    if value.le(current_front_value) {
1990                        continue;
1991                    }
1992                }
1993                if let Some(current_back_value) = self.current_back_value.as_ref() {
1994                    if value.ge(current_back_value) {
1995                        self.met = true;
1996                        return None;
1997                    }
1998                }
1999                self.current_front_value = Some(value.clone());
2000                return Some(value);
2001            } else {
2002                self.current_front_batch = None;
2003            }
2004        }
2005    }
2006}
2007
2008impl<'a, T, Node> DoubleEndedIterator for Iter<'a, T, Node>
2009where
2010    T: Debug + Ord + Clone + Send + 'static,
2011    Node: NodeLike<T> + Send + 'static,
2012{
2013    fn next_back(&mut self) -> Option<Self::Item> {
2014        loop {
2015            if self.met {
2016                return None;
2017            }
2018
2019            if self.current_back_batch.is_none() && !self.install_back_batch() {
2020                return None;
2021            }
2022
2023            let batch = self.current_back_batch.as_mut().expect("installed above");
2024            if let Some(value) = batch.next_back() {
2025                // Mirror of the forward path: skip elements at or above the
2026                // last value yielded from the back, which a freshly
2027                // installed batch can re-expose after churn.
2028                if let Some(current_back_value) = self.current_back_value.as_ref() {
2029                    if value.ge(current_back_value) {
2030                        continue;
2031                    }
2032                }
2033                if let Some(current_front_value) = self.current_front_value.as_ref() {
2034                    if value.le(current_front_value) {
2035                        self.met = true;
2036                        return None;
2037                    }
2038                }
2039                self.current_back_value = Some(value.clone());
2040                return Some(value);
2041            } else {
2042                self.current_back_batch = None;
2043            }
2044        }
2045    }
2046}
2047
2048impl<'a, T: Debug + Ord + Clone + Send, Node: NodeLike<T> + Send + 'static> FusedIterator for Iter<'a, T, Node> {}
2049
2050impl<'a, T, Node> IntoIterator for &'a BTreeSet<T, Node>
2051where
2052    T: Debug + Ord + Send + Clone,
2053    Node: NodeLike<T> + Send + 'static,
2054{
2055    type Item = T;
2056
2057    type IntoIter = Iter<'a, T, Node>;
2058
2059    fn into_iter(self) -> Self::IntoIter {
2060        Iter::new(self)
2061    }
2062}
2063
2064/// An owned-yield double-ended range iterator; see [`Iter`] for the
2065/// consistency and cloning semantics.
2066pub struct Range<'a, T, Node>
2067where
2068    T: Debug + Ord + Clone + Send + 'static,
2069    Node: NodeLike<T> + Send + 'static,
2070{
2071    iter: Iter<'a, T, Node>,
2072}
2073
2074impl<'a, T, Node> Range<'a, T, Node>
2075where
2076    T: Debug + Ord + Clone + Send + 'static,
2077    Node: NodeLike<T> + Send + 'static,
2078{
2079    pub fn new<Q, R>(btree: &'a BTreeSet<T, Node>, range: R) -> Self
2080    where
2081        T: Borrow<Q>,
2082        Q: Ord + ?Sized,
2083        R: RangeBounds<Q>,
2084    {
2085        let index = btree.index.read();
2086
2087        let start_bound = range.start_bound();
2088        let end_bound = range.end_bound();
2089        let mut met = match (start_bound, end_bound) {
2090            (Bound::Included(start), Bound::Included(end)) => start > end,
2091            (Bound::Included(start), Bound::Excluded(end))
2092            | (Bound::Excluded(start), Bound::Included(end))
2093            | (Bound::Excluded(start), Bound::Excluded(end)) => start >= end,
2094            _ => false,
2095        };
2096
2097        let current_front_entry = first_for_borrowed_bound(&index, start_bound, btree.borrow_order_matches);
2098
2099        let front_value = if let Some((front_key, front_node)) = current_front_entry {
2100            let front_guard = front_node.clone().read_arc();
2101            let rank = match start_bound {
2102                Bound::Included(v) => front_guard.rank(Bound::Included(v), true),
2103                Bound::Excluded(v) => front_guard.rank(Bound::Excluded(v), true),
2104                Bound::Unbounded => None,
2105            };
2106            if let Some(rank) = rank {
2107                let value = front_guard.iter().nth(rank).cloned();
2108                drop(front_guard);
2109
2110                value
2111            } else {
2112                // Release the current node before locking its neighbor: this
2113                // branch used to hold front then prev (descending) while the
2114                // back branch below held back then next (ascending), an
2115                // ABBA deadlock between two concurrent Range constructions.
2116                // Never hold two node locks here.
2117                drop(front_guard);
2118                if let Some((_, pre_front_node)) = index.range::<T, _>(..front_key).next_back() {
2119                    let pre_front_guard = pre_front_node.clone().read_arc();
2120                    pre_front_guard.iter().last().cloned()
2121                } else {
2122                    None
2123                }
2124            }
2125        } else {
2126            None
2127        };
2128
2129        let current_back_entry = match end_bound {
2130            Bound::Included(end) | Bound::Excluded(end) => {
2131                node_for_borrowed_end(&index, end, btree.borrow_order_matches)
2132            }
2133            Bound::Unbounded => index.last_key_value(),
2134        };
2135
2136        let back_value = if let Some((back_key, back_node)) = current_back_entry {
2137            let back_guard = back_node.clone().read_arc();
2138            let rank = match end_bound {
2139                Bound::Included(v) => back_guard.rank(Bound::Included(v), false),
2140                Bound::Excluded(v) => back_guard.rank(Bound::Excluded(v), false),
2141                Bound::Unbounded => None,
2142            };
2143            if let Some(rank) = rank {
2144                let value = back_guard.iter().nth_back(rank).cloned();
2145                drop(back_guard);
2146
2147                value
2148            } else {
2149                // See the front branch: release before locking the neighbor.
2150                drop(back_guard);
2151                if let Some((_, next_back_node)) = index
2152                    .range::<T, _>((Bound::Excluded(back_key), Bound::Unbounded))
2153                    .next()
2154                {
2155                    let next_back_guard = next_back_node.clone().read_arc();
2156                    next_back_guard.iter().next().cloned()
2157                } else {
2158                    None
2159                }
2160            }
2161        } else {
2162            None
2163        };
2164
2165        if front_value.is_none() && back_value.is_none() {
2166            // in this case we iter full or no iter at all
2167            if start_bound != Bound::Unbounded || end_bound != Bound::Unbounded {
2168                if let Some(max) = index
2169                    .last_key_value()
2170                    .and_then(|(_, node)| node.clone().read_arc().max().cloned())
2171                {
2172                    if let Bound::Included(v) = start_bound {
2173                        if v > max.borrow() {
2174                            met = true;
2175                        }
2176                    } else if let Bound::Excluded(v) = start_bound {
2177                        if v >= max.borrow() {
2178                            met = true;
2179                        }
2180                    }
2181                }
2182
2183                if let Some(min) = index
2184                    .first_key_value()
2185                    .and_then(|(_, node)| node.clone().read_arc().min().cloned())
2186                {
2187                    if let Bound::Included(v) = end_bound {
2188                        if v < min.borrow() {
2189                            met = true;
2190                        }
2191                    } else if let Bound::Excluded(v) = end_bound {
2192                        if v <= min.borrow() {
2193                            met = true;
2194                        }
2195                    }
2196                }
2197            }
2198        }
2199
2200        // Only the cursor sentinels position the iterator: each direction
2201        // selects and reads its node atomically at install time. Prewiring
2202        // the entries' node Arcs here would reintroduce the choose-then-lock
2203        // split-migration window (see `Iter::install_front_batch`).
2204        Self {
2205            iter: Iter {
2206                tree: btree,
2207                current_front_batch: None,
2208                current_back_batch: None,
2209                exhausted_front_node: None,
2210                exhausted_back_node: None,
2211                front_partial: None,
2212                back_partial: None,
2213                front_batch_limit: INITIAL_BATCH,
2214                back_batch_limit: INITIAL_BATCH,
2215                current_front_value: front_value,
2216                current_back_value: back_value,
2217                met,
2218            },
2219        }
2220    }
2221}
2222
2223impl<'a, T, Node> Iterator for Range<'a, T, Node>
2224where
2225    T: Debug + Ord + Clone + Send + 'static,
2226    Node: NodeLike<T> + Send + 'static,
2227{
2228    type Item = T;
2229
2230    fn next(&mut self) -> Option<Self::Item> {
2231        self.iter.next()
2232    }
2233}
2234
2235impl<'a, T, Node> DoubleEndedIterator for Range<'a, T, Node>
2236where
2237    T: Debug + Ord + Clone + Send + 'static,
2238    Node: NodeLike<T> + Send + 'static,
2239{
2240    fn next_back(&mut self) -> Option<Self::Item> {
2241        self.iter.next_back()
2242    }
2243}
2244
2245impl<'a, T, Node> FusedIterator for Range<'a, T, Node>
2246where
2247    T: Debug + Ord + Clone + Send + 'static,
2248    Node: NodeLike<T> + Send + 'static,
2249{
2250}
2251
2252impl<'a, T, Node> BTreeSet<T, Node>
2253where
2254    T: Debug + Ord + Clone + Send + 'static,
2255    Node: NodeLike<T> + Send + 'static,
2256{
2257    /// Gets an iterator that visits the elements in the `BTreeSet` in ascending
2258    /// order.
2259    ///
2260    /// The iterator yields owned clones of the stored elements (see [`Iter`]):
2261    /// collected values remain valid under arbitrary concurrent mutation of
2262    /// the set.
2263    ///
2264    /// # Examples
2265    ///
2266    /// ```
2267    /// use indexset::concurrent::set::BTreeSet;
2268    ///
2269    /// let set = BTreeSet::from_iter([1, 2, 3]);
2270    /// let mut set_iter = set.iter();
2271    /// assert_eq!(set_iter.next(), Some(1));
2272    /// assert_eq!(set_iter.next(), Some(2));
2273    /// assert_eq!(set_iter.next(), Some(3));
2274    /// assert_eq!(set_iter.next(), None);
2275    /// ```
2276    ///
2277    /// Values returned by the iterator are returned in ascending order:
2278    ///
2279    /// ```
2280    /// use indexset::concurrent::set::BTreeSet;
2281    ///
2282    /// let set = BTreeSet::from_iter([3, 1, 2]);
2283    /// let mut set_iter = set.iter();
2284    /// assert_eq!(set_iter.next(), Some(1));
2285    /// assert_eq!(set_iter.next(), Some(2));
2286    /// assert_eq!(set_iter.next(), Some(3));
2287    /// assert_eq!(set_iter.next(), None);
2288    /// ```
2289    pub fn iter(&'a self) -> Iter<'a, T, Node> {
2290        Iter::new(self)
2291    }
2292    /// Constructs a double-ended iterator over a sub-range of elements in the set.
2293    /// The simplest way is to use the range syntax `min..max`, thus `range(min..max)` will
2294    /// yield elements from min (inclusive) to max (exclusive).
2295    /// The range may also be entered as `(Bound<T>, Bound<T>)`, so for example
2296    /// `range((Excluded(4), Included(10)))` will yield a left-exclusive, right-inclusive
2297    /// range from 4 to 10.
2298    ///
2299    /// # Panics
2300    ///
2301    /// Panics if range `start > end`.
2302    /// Panics if range `start == end` and both bounds are `Excluded`.
2303    ///
2304    /// # Examples
2305    ///
2306    /// ```
2307    /// use indexset::concurrent::set::BTreeSet;
2308    /// use std::ops::Bound::Included;
2309    ///
2310    /// let mut set = BTreeSet::<usize>::new();
2311    /// set.insert(3);
2312    /// set.insert(5);
2313    /// set.insert(8);
2314    /// for elem in set.range((Included(&4), Included(&8))) {
2315    ///     println!("{elem}");
2316    /// }
2317    /// assert_eq!(Some(5), set.range(4..).next());
2318    /// ```
2319    pub fn range<Q, R>(&'a self, range: R) -> Range<'a, T, Node>
2320    where
2321        T: Borrow<Q>,
2322        Q: Ord + ?Sized,
2323        R: RangeBounds<Q>,
2324    {
2325        Range::new(self, range)
2326    }
2327}
2328
2329impl<T> BTreeSet<T>
2330where
2331    T: Debug + Ord + Clone + Send + 'static,
2332{
2333    pub fn remove_range<R, Q>(&self, range: R)
2334    where
2335        Q: Ord + ?Sized,
2336        T: Borrow<Q>,
2337        R: RangeBounds<Q>,
2338    {
2339        // Declare detached storage before the structural guard so element
2340        // destructors run only after that guard is released.
2341        let mut detached_nodes = Vec::new();
2342        let mut index = self.index.write();
2343
2344        let start_bound = range.start_bound();
2345        let end_bound = range.end_bound();
2346
2347        // First node that can contain an element within the start bound. If
2348        // no index key reaches the start bound, nothing qualifies.
2349        let Some((front_key, front_node)) = first_for_borrowed_bound(&index, start_bound, self.borrow_order_matches)
2350        else {
2351            return;
2352        };
2353        let front_key = front_key.clone();
2354        let front_node = front_node.clone();
2355
2356        // Last node that can contain an element within the end bound. Both an
2357        // inclusive and an exclusive end resolve to the first node whose key
2358        // is >= the bound value: a node keyed exactly at an exclusive bound
2359        // still holds elements below it. Past the last key, the last node is
2360        // the only candidate.
2361        let back_entry = match end_bound {
2362            Bound::Included(end) | Bound::Excluded(end) => {
2363                node_for_borrowed_end(&index, end, self.borrow_order_matches)
2364            }
2365            Bound::Unbounded => index.last_key_value(),
2366        };
2367        let Some((back_key, back_node)) = back_entry else {
2368            return;
2369        };
2370        let back_key = back_key.clone();
2371        let back_node = back_node.clone();
2372        if back_key < front_key {
2373            // The end bound resolves before the start bound: empty range.
2374            return;
2375        }
2376
2377        // Number of leading elements of the back node that fall within the
2378        // end bound (inclusive end: elements <= bound; exclusive: < bound).
2379        let removed_prefix_len = |guard: &Vec<T>| -> usize {
2380            match end_bound {
2381                Bound::Included(end) => guard.rank(Bound::Excluded(end), true).map_or(0, |last| last + 1),
2382                Bound::Excluded(end) => guard.rank(Bound::Included(end), true).map_or(0, |last| last + 1),
2383                Bound::Unbounded => guard.len(),
2384            }
2385        };
2386
2387        if Arc::ptr_eq(&front_node, &back_node) {
2388            // The whole range lives in one node.
2389            let mut guard = front_node.clone().write_arc();
2390            let front_position = guard.rank(start_bound, true).map_or(0, |last| last + 1);
2391            let back_position = removed_prefix_len(&guard);
2392            if back_position <= front_position {
2393                return;
2394            }
2395
2396            let original_len = guard.len();
2397            guard.drain(front_position..back_position);
2398            if back_position == original_len {
2399                // The node's maximum was removed: re-key the entry, or drop
2400                // it when the node was fully drained.
2401                index.remove::<T>(&front_key);
2402                if let Some(new_max) = guard.last().cloned() {
2403                    index.insert(new_max, front_node);
2404                }
2405            }
2406            return;
2407        }
2408
2409        let mut front_guard = front_node.clone().write_arc();
2410        let mut back_guard = back_node.clone().write_arc();
2411        let front_position = front_guard.rank(start_bound, true).map_or(0, |last| last + 1);
2412        let back_position = removed_prefix_len(&back_guard);
2413
2414        // Remove every node strictly between the front and the back one.
2415        let middle_keys = index
2416            .range::<T, _>((Bound::Excluded(&front_key), Bound::Excluded(&back_key)))
2417            .map(|(key, _)| key.clone())
2418            .collect::<Vec<_>>();
2419        for key in middle_keys {
2420            let node = index
2421                .remove::<T>(&key)
2422                .expect("middle key was collected under the write lock");
2423            let mut removed_node = node.write_arc();
2424            detached_nodes.push(::core::mem::take(&mut *removed_node));
2425        }
2426
2427        // Trim the front node from the start position: its maximum goes away,
2428        // so its entry must be re-keyed (or dropped when the node empties).
2429        index.remove::<T>(&front_key);
2430        front_guard.drain(front_position..);
2431        if !front_guard.is_empty() {
2432            let new_front_max = front_guard.last().unwrap().clone();
2433            index.insert(new_front_max, front_node);
2434        }
2435
2436        // Trim the back node's prefix: its maximum survives unless the whole
2437        // node drains, so the entry only changes when the node empties.
2438        if back_position >= back_guard.len() {
2439            index.remove::<T>(&back_key);
2440            back_guard.drain(..);
2441        } else if back_position > 0 {
2442            back_guard.drain(..back_position);
2443        }
2444    }
2445}
2446
2447#[cfg(test)]
2448mod tests {
2449    use crate::concurrent::operation::Operation;
2450    use crate::concurrent::set::{BTreeSet, Iter, DEFAULT_INNER_SIZE, INITIAL_BATCH};
2451    use crate::core::node::NodeLike;
2452    use parking_lot::RwLock;
2453    use rand::Rng;
2454    use std::collections::HashSet;
2455    use std::ops::Bound::{self, Included};
2456    use std::sync::mpsc;
2457    use std::sync::{Arc, Barrier, Mutex};
2458    use std::thread;
2459    use std::time::Duration;
2460
2461    // Regression for https://github.com/lucidarium-systems/indexset/issues/57.
2462    #[test]
2463    fn test_node_size_two_preserves_all_u64_values() {
2464        let set = BTreeSet::<u64>::with_maximum_node_size(2);
2465
2466        for value in 0..10_u64 {
2467            set.insert(value);
2468        }
2469
2470        assert_eq!(set.iter().collect::<Vec<_>>(), (0..10).collect::<Vec<_>>());
2471    }
2472
2473    // Regression for https://github.com/lucidarium-systems/indexset/issues/57.
2474    #[test]
2475    fn test_node_size_three_preserves_all_u8_values() {
2476        let set = BTreeSet::<u8>::with_maximum_node_size(3);
2477
2478        for value in 0..20_u8 {
2479            set.insert(value);
2480        }
2481
2482        assert_eq!(set.iter().collect::<Vec<_>>(), (0..20).collect::<Vec<_>>());
2483    }
2484
2485    #[test]
2486    fn concurrent_first_writers_preserve_disjoint_ranges() {
2487        const WRITERS: u64 = 8;
2488        const VALUES_PER_WRITER: u64 = 1_000;
2489
2490        let set = Arc::new(BTreeSet::<u64>::new());
2491        let start = Arc::new(Barrier::new(WRITERS as usize));
2492        let handles = (0..WRITERS)
2493            .map(|writer| {
2494                let set = Arc::clone(&set);
2495                let start = Arc::clone(&start);
2496                thread::spawn(move || {
2497                    start.wait();
2498                    let first = writer * VALUES_PER_WRITER;
2499                    for value in first..first + VALUES_PER_WRITER {
2500                        assert!(set.insert(value));
2501                    }
2502                })
2503            })
2504            .collect::<Vec<_>>();
2505
2506        for handle in handles {
2507            handle.join().unwrap();
2508        }
2509
2510        let expected = (0..WRITERS * VALUES_PER_WRITER).collect::<Vec<_>>();
2511        assert_eq!(set.len(), expected.len());
2512        assert_eq!(set.iter().collect::<Vec<_>>(), expected);
2513    }
2514
2515    #[test]
2516    fn published_point_reads_remain_definitive_across_splits() {
2517        use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
2518
2519        const STABLE_KEYS: usize = 256;
2520        const FINAL_KEYS: usize = 4_096;
2521        const READERS: usize = 4;
2522
2523        let set = Arc::new(BTreeSet::<usize>::with_maximum_node_size(8));
2524        for key in 0..STABLE_KEYS {
2525            set.insert(key);
2526        }
2527
2528        let start = Arc::new(Barrier::new(READERS + 1));
2529        let done = Arc::new(AtomicBool::new(false));
2530        let published_up_to = Arc::new(AtomicUsize::new(STABLE_KEYS - 1));
2531        let readers = (0..READERS)
2532            .map(|reader| {
2533                let set = Arc::clone(&set);
2534                let start = Arc::clone(&start);
2535                let done = Arc::clone(&done);
2536                let published_up_to = Arc::clone(&published_up_to);
2537                thread::spawn(move || {
2538                    start.wait();
2539                    let mut probe = reader;
2540                    while !done.load(Ordering::Acquire) {
2541                        let key = probe % STABLE_KEYS;
2542                        assert_eq!(set.get_with(&key, |value| *value), Some(key));
2543                        let newest_acknowledged = published_up_to.load(Ordering::Acquire);
2544                        assert_eq!(
2545                            set.get_with(&newest_acknowledged, |value| *value),
2546                            Some(newest_acknowledged),
2547                            "an acknowledged insert disappeared from the published route"
2548                        );
2549                        probe += READERS;
2550                    }
2551                })
2552            })
2553            .collect::<Vec<_>>();
2554
2555        start.wait();
2556        for key in STABLE_KEYS..FINAL_KEYS {
2557            set.insert(key);
2558            published_up_to.store(key, Ordering::Release);
2559        }
2560        done.store(true, Ordering::Release);
2561
2562        for reader in readers {
2563            reader.join().unwrap();
2564        }
2565        assert_eq!(set.len(), FINAL_KEYS);
2566        for key in 0..FINAL_KEYS {
2567            assert_eq!(set.get_with(&key, |value| *value), Some(key));
2568        }
2569    }
2570
2571    #[test]
2572    fn published_pointer_survives_reclamation_interleavings() {
2573        use std::sync::atomic::{AtomicBool, Ordering};
2574
2575        let set = Arc::new(BTreeSet::<usize>::with_maximum_node_size(2));
2576        for key in 0..8 {
2577            set.insert(key);
2578        }
2579        let start = Arc::new(Barrier::new(2));
2580        let done = Arc::new(AtomicBool::new(false));
2581
2582        let reader_set = Arc::clone(&set);
2583        let reader_start = Arc::clone(&start);
2584        let reader_done = Arc::clone(&done);
2585        let reader = thread::spawn(move || {
2586            reader_start.wait();
2587            let mut probe = 0;
2588            while !reader_done.load(Ordering::Acquire) {
2589                let key = probe % 8;
2590                assert_eq!(reader_set.get_with(&key, |value| *value), Some(key));
2591                probe += 1;
2592            }
2593        });
2594
2595        start.wait();
2596        for key in 8..80 {
2597            set.insert(key);
2598        }
2599        for key in 8..80 {
2600            assert_eq!(set.remove(&key), Some(key));
2601        }
2602        done.store(true, Ordering::Release);
2603        reader.join().unwrap();
2604
2605        for key in 0..8 {
2606            assert_eq!(set.get_with(&key, |value| *value), Some(key));
2607        }
2608    }
2609
2610    #[test]
2611    fn published_route_chunks_split_and_merge_without_losing_keys() {
2612        let set = BTreeSet::<usize>::with_maximum_node_size(2);
2613        for key in 0..600 {
2614            assert!(set.insert(key));
2615        }
2616        for key in (0..600).step_by(2) {
2617            assert_eq!(set.remove(&key), Some(key));
2618        }
2619        for key in 0..600 {
2620            assert_eq!(set.contains(&key), key % 2 == 1, "probe {key}");
2621        }
2622        for key in (1..600).step_by(2) {
2623            assert_eq!(set.remove(&key), Some(key));
2624        }
2625        assert!(set.is_empty());
2626    }
2627
2628    #[test]
2629    fn mixed_structural_and_node_lock_paths_complete_without_deadlock() {
2630        const THREADS: usize = 8;
2631        const OPERATIONS: usize = 1_000;
2632
2633        let set = Arc::new(BTreeSet::<usize>::with_maximum_node_size(8));
2634        for value in 0..256 {
2635            set.insert(value);
2636        }
2637
2638        let start = Arc::new(Barrier::new(THREADS));
2639        let (done_tx, done_rx) = mpsc::channel();
2640        let handles = (0..THREADS)
2641            .map(|worker| {
2642                let set = Arc::clone(&set);
2643                let start = Arc::clone(&start);
2644                let done_tx = done_tx.clone();
2645                thread::spawn(move || {
2646                    start.wait();
2647                    for operation in 0..OPERATIONS {
2648                        let value = (operation * 17 + worker * 31) % 512;
2649                        match (operation + worker) % 5 {
2650                            0 => {
2651                                set.insert(value);
2652                            }
2653                            1 => {
2654                                set.remove(&value);
2655                            }
2656                            2 => {
2657                                let _ = set.contains(&value);
2658                            }
2659                            3 => {
2660                                let _ = set.get_with(&value, Clone::clone);
2661                            }
2662                            _ => {
2663                                set.remove_range(value..=value);
2664                                set.insert(value);
2665                            }
2666                        }
2667                    }
2668                    done_tx.send(()).unwrap();
2669                })
2670            })
2671            .collect::<Vec<_>>();
2672        drop(done_tx);
2673
2674        for _ in 0..THREADS {
2675            done_rx
2676                .recv_timeout(Duration::from_secs(10))
2677                .expect("mixed structural/node-lock workload did not complete");
2678        }
2679        for handle in handles {
2680            handle.join().unwrap();
2681        }
2682    }
2683
2684    #[test]
2685    fn test_concurrent_insert() {
2686        let set = Arc::new(BTreeSet::<i32>::new());
2687        let num_threads = 128;
2688        let operations_per_thread = 10000;
2689        let mut handles = vec![];
2690
2691        let test_data: Vec<Vec<(i32, i32)>> = (0..num_threads)
2692            .map(|_| {
2693                let mut rng = rand::rng();
2694                (0..operations_per_thread)
2695                    .map(|_| {
2696                        let value = rng.random_range(0..100000);
2697                        let operation = rng.random_range(0..2);
2698                        (operation, value)
2699                    })
2700                    .collect()
2701            })
2702            .collect();
2703
2704        let expected_values = Arc::new(Mutex::new(HashSet::new()));
2705
2706        for thread_idx in 0..num_threads {
2707            let set_clone = Arc::clone(&set);
2708            let expected_values = Arc::clone(&expected_values);
2709            let thread_data = test_data[thread_idx].clone();
2710
2711            let handle = thread::spawn(move || {
2712                for (operation, value) in thread_data {
2713                    if operation == 0 {
2714                        let _a = set_clone.insert(value);
2715                        expected_values.lock().unwrap().insert(value);
2716                    }
2717                }
2718            });
2719            handles.push(handle);
2720        }
2721
2722        for handle in handles {
2723            handle.join().unwrap();
2724        }
2725
2726        let expected_values = expected_values.lock().unwrap();
2727        assert_eq!(set.len(), expected_values.len());
2728
2729        for value in expected_values.iter() {
2730            assert!(set.contains(value));
2731        }
2732    }
2733
2734    #[test]
2735    fn test_insert_desc() {
2736        let set = Arc::new(BTreeSet::<i32>::new());
2737
2738        assert!(set.insert(2));
2739        assert!(set.insert(1));
2740    }
2741
2742    #[test]
2743    fn test_insert_st() {
2744        let set = Arc::new(BTreeSet::<i32>::new());
2745        let mut rng = rand::rng();
2746
2747        let n = 2048 * 100;
2748        let range = 0..n;
2749        let mut inserted_values = HashSet::new();
2750        for _ in range {
2751            let value = rng.random_range(0..n);
2752            if inserted_values.insert(value) {
2753                set.insert(value);
2754            }
2755        }
2756
2757        assert_eq!(
2758            set.len(),
2759            inserted_values.len(),
2760            "Length did not match, missing: {:?}",
2761            set.index
2762                .read()
2763                .values()
2764                .flat_map(|node| node.read().iter().cloned().collect::<Vec<_>>())
2765                .collect::<HashSet<_>>()
2766                .symmetric_difference(&inserted_values)
2767                .collect::<Vec<_>>()
2768        );
2769        for i in inserted_values {
2770            assert!(
2771                set.contains(&i),
2772                "Did not find: {} with index: {:?}",
2773                i,
2774                set.index.read().keys().cloned().collect::<Vec<_>>(),
2775            );
2776        }
2777    }
2778
2779    #[test]
2780    fn test_single_element() {
2781        let set = BTreeSet::<i32>::new();
2782        set.insert(1);
2783        let mut iter = set.into_iter();
2784        assert_eq!(iter.next(), Some(1));
2785        assert_eq!(iter.next(), None);
2786        assert_eq!(iter.next_back(), None);
2787    }
2788
2789    #[test]
2790    fn test_multiple_elements() {
2791        let set = BTreeSet::<i32>::new();
2792        set.insert(1);
2793        set.insert(2);
2794        set.insert(3);
2795        let mut iter = set.into_iter();
2796        assert_eq!(iter.next(), Some(1));
2797        assert_eq!(iter.next_back(), Some(3));
2798        assert_eq!(iter.next(), Some(2));
2799        assert_eq!(iter.next(), None);
2800        assert_eq!(iter.next_back(), None);
2801    }
2802
2803    #[test]
2804    fn test_bidirectional_iteration() {
2805        let set = BTreeSet::<i32>::with_maximum_node_size(3);
2806        for i in 1..=20 {
2807            set.insert(i);
2808        }
2809        let mut iter = set.into_iter();
2810        for i in 0..10 {
2811            // (1, 20), (2, 19), (3, 18), (4, 17), (5, 16), (6, 15), (7, 14), (8, 13), (9, 12), (10, 11)
2812            let tree = set.index.read().keys().cloned().collect::<Vec<_>>();
2813
2814            let expected_next = i + 1;
2815            let actual_next = iter.next();
2816            assert_eq!(actual_next, Some(expected_next), "Tree: {:?}", tree);
2817
2818            let expected_next_back = 20 - i;
2819            let actual_next_back = iter.next_back();
2820            assert_eq!(actual_next_back, Some(expected_next_back), "Tree: {:?}", tree);
2821        }
2822        assert_eq!(iter.next(), None);
2823        assert_eq!(iter.next_back(), None);
2824    }
2825
2826    #[test]
2827    fn test_fused_iterator() {
2828        let set = BTreeSet::<i32>::new();
2829        set.insert(1);
2830        let mut iter = set.into_iter();
2831        assert_eq!(iter.next(), Some(1));
2832        assert_eq!(iter.next(), None);
2833        assert_eq!(iter.next(), None);
2834    }
2835
2836    #[test]
2837    fn test_fused_iterator_back() {
2838        let set = BTreeSet::<i32>::new();
2839        set.insert(1);
2840        let mut iter = set.into_iter();
2841        assert_eq!(iter.next_back(), Some(1));
2842        assert_eq!(iter.next_back(), None);
2843        assert_eq!(iter.next_back(), None);
2844    }
2845
2846    #[test]
2847    fn test_out_of_bounds_range() {
2848        let btree: BTreeSet<usize> = BTreeSet::from_iter(0..10);
2849        assert_eq!(btree.range((Included(5), Included(10))).count(), 5);
2850        assert_eq!(btree.range((Included(5), Included(11))).count(), 5);
2851        assert_eq!(btree.range((Included(5), Included(10 + DEFAULT_INNER_SIZE))).count(), 5);
2852        assert_eq!(btree.range((Included(0), Included(11))).count(), 10);
2853    }
2854
2855    #[test]
2856    fn test_iterating_over_blocks() {
2857        let btree = BTreeSet::from_iter((0..(DEFAULT_INNER_SIZE + 10)).into_iter());
2858        assert_eq!(btree.iter().count(), (0..(DEFAULT_INNER_SIZE + 10)).count());
2859        let start = btree.range(0..DEFAULT_INNER_SIZE).into_iter().collect::<Vec<_>>();
2860
2861        assert_eq!(start, (0..DEFAULT_INNER_SIZE).collect::<Vec<_>>());
2862        assert_eq!(
2863            btree.range(0..=DEFAULT_INNER_SIZE).into_iter().collect::<Vec<_>>(),
2864            (0..=DEFAULT_INNER_SIZE).collect::<Vec<_>>()
2865        );
2866        assert_eq!(
2867            btree.range(0..=DEFAULT_INNER_SIZE + 1).count(),
2868            (0..=DEFAULT_INNER_SIZE + 1).count()
2869        );
2870        assert_eq!(btree.iter().rev().count(), (0..(DEFAULT_INNER_SIZE + 10)).count());
2871        assert_eq!(
2872            btree.range(0..DEFAULT_INNER_SIZE).rev().count(),
2873            (0..DEFAULT_INNER_SIZE).count()
2874        );
2875        assert_eq!(
2876            btree.range(0..=DEFAULT_INNER_SIZE).rev().count(),
2877            (0..=DEFAULT_INNER_SIZE).count()
2878        );
2879        assert_eq!(
2880            btree.range(0..=DEFAULT_INNER_SIZE + 1).rev().count(),
2881            (0..=DEFAULT_INNER_SIZE + 1).count()
2882        );
2883    }
2884
2885    #[test]
2886    fn test_empty_set() {
2887        let btree: BTreeSet<usize> = BTreeSet::new();
2888        assert_eq!(btree.iter().count(), 0);
2889        assert_eq!(btree.range(0..0).count(), 0);
2890        assert_eq!(btree.range(0..).count(), 0);
2891        assert_eq!(btree.range(..0).count(), 0);
2892        assert_eq!(btree.range(..).count(), 0);
2893        assert_eq!(btree.range(0..=0).count(), 0);
2894        assert_eq!(btree.range(..1).count(), 0);
2895
2896        assert_eq!(btree.iter().rev().count(), 0);
2897        assert_eq!(btree.range(0..0).rev().count(), 0);
2898        assert_eq!(btree.range(..).rev().count(), 0);
2899        assert_eq!(btree.range(..1).rev().count(), 0);
2900
2901        assert_eq!(btree.range(..DEFAULT_INNER_SIZE).count(), 0);
2902        assert_eq!(btree.range(DEFAULT_INNER_SIZE..DEFAULT_INNER_SIZE * 2).count(), 0);
2903    }
2904
2905    #[test]
2906    fn test_remove_range() {
2907        // We have DEFAULT_INNER_SIZE * 2 elements
2908        let btree = BTreeSet::from_iter(0..(DEFAULT_INNER_SIZE * 2));
2909        let expected_len = DEFAULT_INNER_SIZE * 2;
2910        let actual_len = btree.len();
2911        assert_eq!(expected_len, actual_len);
2912
2913        // We remove 10 elements from the beginning, 5 included up to 15 excluded.
2914        btree.remove_range(5..15);
2915        let expected_len = expected_len - 10;
2916        let actual_len = btree.len();
2917        assert_eq!(expected_len, actual_len);
2918
2919        // Then take more 10 from the middle
2920        btree.remove_range(DEFAULT_INNER_SIZE - 5..DEFAULT_INNER_SIZE + 5);
2921        let expected_len = expected_len - 10;
2922        let actual_len = btree.len();
2923        assert_eq!(expected_len, actual_len);
2924
2925        // And then remove 512
2926        btree.remove_range(..DEFAULT_INNER_SIZE / 2);
2927        // We add +10 here because we are removing everything up to 512, but we already removed 5..15.
2928        let expected_len = expected_len - (DEFAULT_INNER_SIZE / 2) + 10;
2929        let actual_len = btree.len();
2930        assert_eq!(expected_len, actual_len);
2931
2932        // And then everything from (512 * 3) / 2 to the end, which is
2933        // exactly the upper 512 values.
2934        let from = (DEFAULT_INNER_SIZE * 3) / 2;
2935        btree.remove_range(from..);
2936        let expected_len = expected_len - DEFAULT_INNER_SIZE / 2;
2937        let actual_len = btree.len();
2938        assert_eq!(expected_len, actual_len);
2939
2940        // We now clear the tree
2941        btree.remove_range(..);
2942        assert_eq!(btree.len(), 0);
2943
2944        // Re-insert everything
2945        for i in 0..(DEFAULT_INNER_SIZE * 2) {
2946            btree.insert(i);
2947        }
2948        let expected_len = DEFAULT_INNER_SIZE * 2;
2949        let actual_len = btree.len();
2950        assert_eq!(expected_len, actual_len);
2951
2952        btree.remove_range((std::ops::Bound::Excluded(5), std::ops::Bound::Excluded(15)));
2953        let expected_len = expected_len - 9;
2954        let actual_len = btree.len();
2955        assert_eq!(expected_len, actual_len);
2956
2957        btree.remove_range((
2958            std::ops::Bound::Included(DEFAULT_INNER_SIZE),
2959            std::ops::Bound::Excluded(DEFAULT_INNER_SIZE + 10),
2960        ));
2961        let expected_len = expected_len - 10;
2962        let actual_len = btree.len();
2963        assert_eq!(expected_len, actual_len);
2964
2965        // This range exceeds the size of the tree
2966        btree.remove_range(DEFAULT_INNER_SIZE * 3..DEFAULT_INNER_SIZE * 4);
2967        let expected_len = expected_len;
2968        let actual_len = btree.len();
2969        assert_eq!(expected_len, actual_len);
2970
2971        // This range starts at the very end of the tree, and exceeds it
2972        btree.remove_range(DEFAULT_INNER_SIZE * 2 - 5..DEFAULT_INNER_SIZE * 3);
2973        let expected_len = expected_len - 5;
2974        let actual_len = btree.len();
2975        assert_eq!(expected_len, actual_len);
2976    }
2977
2978    #[test]
2979    fn remove_range_end_bound_regressions() {
2980        // `x..` must remove only the suffix, not also drain the first node.
2981        let set = BTreeSet::<u64>::with_maximum_node_size(4);
2982        for value in 0..10 {
2983            set.insert(value);
2984        }
2985        set.remove_range(7..);
2986        assert_eq!(set.iter().collect::<Vec<_>>(), (0..7).collect::<Vec<_>>());
2987
2988        // `..` must clear every node, not only the first one.
2989        let set = BTreeSet::<u64>::with_maximum_node_size(4);
2990        for value in 0..10 {
2991            set.insert(value);
2992        }
2993        set.remove_range(..);
2994        assert_eq!(set.len(), 0);
2995        assert!(set.is_empty());
2996
2997        // An inclusive end must remove every element up to and including it.
2998        let set = BTreeSet::<u64>::with_maximum_node_size(4);
2999        for value in 0..10 {
3000            set.insert(value);
3001        }
3002        set.remove_range(3..=5);
3003        assert_eq!(set.iter().collect::<Vec<_>>(), vec![0, 1, 2, 6, 7, 8, 9]);
3004
3005        // `x..=x` must remove exactly x, not drain to the node end.
3006        let set = BTreeSet::<u64>::with_maximum_node_size(4);
3007        for value in 0..10 {
3008            set.insert(value);
3009        }
3010        set.remove_range(2..=2);
3011        assert_eq!(set.iter().collect::<Vec<_>>(), vec![0, 1, 3, 4, 5, 6, 7, 8, 9]);
3012
3013        // An exclusive end equal to a node maximum must not drain the
3014        // following node.
3015        let set = BTreeSet::<u64>::with_maximum_node_size(3);
3016        for value in 0..9 {
3017            set.insert(value);
3018        }
3019        let boundary = *set.index.read().first_key_value().expect("node must exist").0;
3020        set.remove_range(0..boundary);
3021        let expected = (0..9).filter(|value| *value >= boundary).collect::<Vec<_>>();
3022        assert_eq!(set.iter().collect::<Vec<_>>(), expected);
3023    }
3024
3025    #[test]
3026    fn remove_range_matches_btreeset_oracle() {
3027        use std::ops::Bound;
3028
3029        fn oracle_case(node_size: usize, values: &[u64], start: Bound<u64>, end: Bound<u64>) {
3030            let set = BTreeSet::<u64>::with_maximum_node_size(node_size);
3031            for &value in values {
3032                set.insert(value);
3033            }
3034            let mut oracle = values.iter().copied().collect::<std::collections::BTreeSet<_>>();
3035
3036            let range = (start, end);
3037            oracle.retain(|value| !std::ops::RangeBounds::contains(&range, value));
3038            set.remove_range(range);
3039
3040            assert_eq!(
3041                set.iter().collect::<Vec<_>>(),
3042                oracle.iter().copied().collect::<Vec<_>>(),
3043                "node_size={node_size}, start={start:?}, end={end:?}"
3044            );
3045            assert_eq!(
3046                set.len(),
3047                oracle.len(),
3048                "node_size={node_size}, start={start:?}, end={end:?}"
3049            );
3050        }
3051
3052        // Even values only, so probes hit present values, absent values, and
3053        // both sides of every node boundary.
3054        let values = (0..15u64).map(|value| value * 2).collect::<Vec<_>>();
3055        let mut bounds = vec![Bound::Unbounded];
3056        for probe in 0..=30u64 {
3057            bounds.push(Bound::Included(probe));
3058            bounds.push(Bound::Excluded(probe));
3059        }
3060
3061        // Single-node and multi-node geometries, on and off node boundaries.
3062        for node_size in [4usize, 7, 64] {
3063            for &start in &bounds {
3064                for &end in &bounds {
3065                    oracle_case(node_size, &values, start, end);
3066                }
3067            }
3068        }
3069    }
3070
3071    #[test]
3072    fn remove_range_clears_detached_nodes() {
3073        // White-box geometry fixture: a failure after split/merge tuning may
3074        // mean node boundaries changed rather than detached-node clearing
3075        // regressed. WorkTable's persisted-index fixtures have the same
3076        // geometry coupling.
3077        let set = BTreeSet::<u64>::with_maximum_node_size(4);
3078        for value in 0..32 {
3079            set.insert(value);
3080        }
3081
3082        let detached = set
3083            .index
3084            .read()
3085            .range((Included(&2), Bound::Unbounded))
3086            .nth(1)
3087            .unwrap()
3088            .1
3089            .clone();
3090        let detached_values = detached.read().iter().copied().collect::<Vec<_>>();
3091        assert!(detached_values.iter().all(|value| (2..30).contains(value)));
3092
3093        set.remove_range(2..30);
3094
3095        assert!(detached.read().is_empty());
3096        assert!(detached_values.iter().all(|value| !set.contains(value)));
3097    }
3098
3099    #[test]
3100    fn remove_reaches_value_above_every_index_key() {
3101        let set = BTreeSet::<u64>::new();
3102        for value in [1u64, 2, 3] {
3103            set.insert(value);
3104        }
3105
3106        // Simulate a stale-key window: the last node's maximum grows past its
3107        // index key before the UpdateMax repair commits. `contains` already
3108        // reaches such a value through the back-node fallback; `remove` must
3109        // reach it the same way.
3110        {
3111            let node = set.index.read().last_key_value().expect("node must exist").1.clone();
3112            let mut guard = node.write();
3113            NodeLike::insert(&mut *guard, 5u64);
3114        }
3115
3116        assert!(set.contains(&5));
3117        assert_eq!(set.remove(&5), Some(5), "value above every index key must be removable");
3118        assert!(!set.contains(&5));
3119        assert_eq!(set.iter().collect::<Vec<_>>(), vec![1, 2, 3]);
3120    }
3121
3122    // Simulates the first phase of a remove that empties a node: the elements
3123    // are deleted under the node lock, leaving the index entry with a stale
3124    // key, and the caller receives the not-yet-committed MakeUnreachable.
3125    fn drain_node_with_pending_unlink(set: &BTreeSet<u64>, values: &[u64], stale_key: u64) -> Operation<u64, Vec<u64>> {
3126        let node = set.index.read().last_key_value().expect("node must exist").1.clone();
3127        {
3128            let mut guard = node.write();
3129            for value in values {
3130                NodeLike::delete(&mut *guard, value).expect("seeded value must be present");
3131            }
3132        }
3133        Operation::MakeUnreachable(node, stale_key)
3134    }
3135
3136    #[test]
3137    fn split_commit_against_drained_node_fails_instead_of_dropping_insert() {
3138        let set = BTreeSet::<u64>::new();
3139        for seeded in [10u64, 20, 30] {
3140            set.insert(seeded);
3141        }
3142        let node = set.index.read().last_key_value().expect("node must exist").1.clone();
3143        // A split is scheduled with a pending insert riding on it...
3144        let pending_split = Operation::Split(node.clone(), 30u64, 15u64);
3145        // ...then a concurrent remove drains the node before the commit.
3146        {
3147            let mut guard = node.write();
3148            for seeded in [10u64, 20, 30] {
3149                NodeLike::delete(&mut *guard, &seeded).expect("seeded value must be present");
3150            }
3151        }
3152
3153        // The commit must fail so the insert retries; it must neither drop
3154        // the pending value silently nor unlink the still-indexed node.
3155        assert!(pending_split
3156            .commit::<false>(&mut set.index.write(), super::no_identity_adoption)
3157            .is_err());
3158        assert!(
3159            set.index.read().get(&30).is_some(),
3160            "drained node must stay linked for the retry"
3161        );
3162
3163        // The retried insert lands and repairs the index.
3164        assert!(set.insert(15));
3165        assert!(set.contains(&15));
3166        assert_eq!(set.remove(&15), Some(15));
3167        assert!(set.is_empty());
3168    }
3169
3170    #[cfg(feature = "cdc")]
3171    #[test]
3172    fn split_commit_against_drained_node_does_not_panic_in_cdc_build() {
3173        let set = BTreeSet::<u64>::new();
3174        for seeded in [10u64, 20, 30] {
3175            set.insert(seeded);
3176        }
3177        let node = set.index.read().last_key_value().expect("node must exist").1.clone();
3178        let pending_split = Operation::Split(node.clone(), 30u64, 15u64);
3179        {
3180            let mut guard = node.write();
3181            for seeded in [10u64, 20, 30] {
3182                NodeLike::delete(&mut *guard, &seeded).expect("seeded value must be present");
3183            }
3184        }
3185
3186        // The cdc-emitting commit used to panic reading the drained node's
3187        // maximum while holding the structural write lock.
3188        assert!(pending_split
3189            .commit::<true>(&mut set.index.write(), super::no_identity_adoption)
3190            .is_err());
3191        assert!(set.index.read().get(&30).is_some());
3192
3193        let (old, _events) = set.put_cdc(15);
3194        assert!(old.is_none());
3195        assert!(set.contains(&15));
3196    }
3197
3198    #[test]
3199    fn insert_into_emptied_node_survives_stale_make_unreachable() {
3200        // One value below and one above the stale index key.
3201        for value in [5u64, 40u64] {
3202            let set = BTreeSet::<u64>::new();
3203            for seeded in [10u64, 20, 30] {
3204                set.insert(seeded);
3205            }
3206            let pending_unlink = drain_node_with_pending_unlink(&set, &[10, 20, 30], 30);
3207
3208            // The insert lands in the emptied node and must repair the stale
3209            // index key immediately.
3210            assert!(set.insert(value));
3211
3212            // The stale unlink then commits: it must not remove the node that
3213            // now contains the acknowledged insert.
3214            let _ = pending_unlink.commit::<false>(&mut set.index.write(), super::no_identity_adoption);
3215
3216            assert!(set.contains(&value), "value {value} lost after stale unlink");
3217            assert_eq!(set.iter().collect::<Vec<_>>(), vec![value]);
3218            assert_eq!(set.remove(&value), Some(value));
3219            assert!(!set.contains(&value));
3220            assert_eq!(set.len(), 0);
3221        }
3222    }
3223
3224    #[test]
3225    fn stale_make_unreachable_rekeys_refilled_node_instead_of_unlinking() {
3226        for value in [5u64, 40u64] {
3227            let set = BTreeSet::<u64>::new();
3228            for seeded in [10u64, 20, 30] {
3229                set.insert(seeded);
3230            }
3231            let node = set.index.read().last_key_value().expect("node must exist").1.clone();
3232            let pending_unlink = drain_node_with_pending_unlink(&set, &[10, 20, 30], 30);
3233
3234            // First phase of a concurrent insert: the value lands in the
3235            // routed (empty) node under the node lock; the UpdateMax repair
3236            // has not committed yet.
3237            {
3238                let mut guard = node.write();
3239                NodeLike::insert(&mut *guard, value);
3240            }
3241            let pending_repair = Operation::UpdateMax(node.clone(), 30u64);
3242
3243            // The remove's stale unlink commits first: it must observe the
3244            // refilled node and re-key it rather than unlink it.
3245            assert!(pending_unlink
3246                .commit::<false>(&mut set.index.write(), super::no_identity_adoption)
3247                .is_ok());
3248            // The insert's repair then finds the entry already re-keyed.
3249            let _ = pending_repair.commit::<false>(&mut set.index.write(), super::no_identity_adoption);
3250
3251            assert!(set.contains(&value), "value {value} lost to stale unlink");
3252            assert_eq!(set.remove(&value), Some(value));
3253            assert!(set.is_empty());
3254        }
3255    }
3256
3257    #[test]
3258    fn published_route_updates_when_a_non_last_boundary_shrinks() {
3259        let set = BTreeSet::<u64>::with_maximum_node_size(8);
3260        set.attach_nodes([vec![1, 10], vec![20, 30]]);
3261
3262        // Removing the first node's maximum changes its canonical route from
3263        // 10 to 1. If the published route remains at 10, the later insertion
3264        // of 5 correctly lands in the second node but a point read for 5 is
3265        // misrouted to the first node and reports a false miss.
3266        assert_eq!(set.remove(&10), Some(10));
3267        assert!(set.insert(5));
3268        assert!(set.contains(&5));
3269        assert_eq!(set.get(&5).map(|value| *value.get()), Some(5));
3270    }
3271
3272    #[test]
3273    fn attach_repairs_a_stale_last_node_boundary() {
3274        let set = BTreeSet::<u64>::with_maximum_node_size(8);
3275        set.attach_node(vec![1, 10]);
3276
3277        // A last-node maximum may remain conservatively published at its old
3278        // high boundary. Incremental restoration above that node must move
3279        // the old route down before installing a new node whose values occupy
3280        // the gap.
3281        assert_eq!(set.remove(&10), Some(10));
3282        set.attach_node(vec![5, 20]);
3283
3284        assert_eq!(set.get(&5).map(|value| *value.get()), Some(5));
3285        assert!(set.contains(&5));
3286    }
3287
3288    #[test]
3289    fn attach_repairs_a_stale_low_last_node_boundary() {
3290        let set = BTreeSet::<u64>::with_maximum_node_size(8);
3291        set.attach_node(vec![1, 10]);
3292
3293        // Growing the last node can leave its published boundary below its
3294        // canonical maximum. Once another node is attached, that old route is
3295        // no longer the final fallback and must be repaired as well.
3296        assert!(set.insert(20));
3297        set.attach_node(vec![25, 30]);
3298
3299        assert_eq!(set.get(&20).map(|value| *value.get()), Some(20));
3300        assert_eq!(set.get(&25).map(|value| *value.get()), Some(25));
3301    }
3302
3303    #[test]
3304    fn attach_recovers_from_a_missing_published_identity() {
3305        let set = BTreeSet::<u64>::with_maximum_node_size(8);
3306        set.attach_node(vec![1, 10]);
3307        let last_identity = {
3308            let index = set.index.read();
3309            super::node_identity(index.last_key_value().unwrap().1)
3310        };
3311        assert!(set.index.published_keys.lock().remove(&last_identity).is_some());
3312
3313        set.attach_node(vec![20, 30]);
3314
3315        for value in [1, 10, 20, 30] {
3316            assert_eq!(set.get(&value).map(|found| *found.get()), Some(value));
3317        }
3318    }
3319
3320    #[test]
3321    fn insert_recovers_from_a_missing_replaced_node_identity() {
3322        let set = BTreeSet::<u64>::with_maximum_node_size(8);
3323        set.attach_node(vec![1, 10]);
3324        let old_node = set.index.read().last_key_value().unwrap().1.clone();
3325        assert!(set
3326            .index
3327            .published_keys
3328            .lock()
3329            .remove(&super::node_identity(&old_node))
3330            .is_some());
3331
3332        {
3333            let mut index = set.index.write();
3334            let replaced = index.insert(10, Arc::new(parking_lot::RwLock::new(vec![5, 10])));
3335            assert!(replaced.is_some_and(|node| Arc::ptr_eq(&node, &old_node)));
3336        }
3337
3338        assert!(!set.contains(&1));
3339        assert!(set.contains(&5));
3340        assert!(set.contains(&10));
3341    }
3342
3343    #[test]
3344    fn remove_recovers_from_a_missing_node_identity() {
3345        let set = BTreeSet::<u64>::with_maximum_node_size(8);
3346        set.attach_node(vec![1, 10]);
3347        let old_node = set.index.read().last_key_value().unwrap().1.clone();
3348        assert!(set
3349            .index
3350            .published_keys
3351            .lock()
3352            .remove(&super::node_identity(&old_node))
3353            .is_some());
3354
3355        {
3356            let mut index = set.index.write();
3357            let removed = index.remove(&10).expect("canonical route exists");
3358            assert!(Arc::ptr_eq(&removed, &old_node));
3359        }
3360
3361        assert!(set.is_empty());
3362        assert!(!set.contains(&1));
3363    }
3364
3365    #[test]
3366    fn missing_published_remove_does_not_clone_a_shared_chunk() {
3367        let mut published = super::PublishedNodeIndex::<u64, Vec<u64>> {
3368            chunks: Vec::new(),
3369            len: 0,
3370        };
3371        for key in 0..16 {
3372            published.insert(key, Arc::new(parking_lot::RwLock::new(vec![key])));
3373        }
3374        let snapshot = published.clone();
3375        assert!(Arc::ptr_eq(&published.chunks[0], &snapshot.chunks[0]));
3376
3377        assert!(published.remove(&100).is_none());
3378
3379        assert!(Arc::ptr_eq(&published.chunks[0], &snapshot.chunks[0]));
3380    }
3381
3382    #[test]
3383    fn published_chunks_have_split_merge_hysteresis() {
3384        let mut published = super::PublishedNodeIndex::<u64, Vec<u64>> {
3385            chunks: Vec::new(),
3386            len: 0,
3387        };
3388        for key in 0..=128 {
3389            published.insert(key, Arc::new(parking_lot::RwLock::new(vec![key])));
3390        }
3391        assert_eq!(published.chunks.len(), 2);
3392
3393        for key in 0..33 {
3394            assert!(published.remove(&key).is_some());
3395        }
3396        assert_eq!(published.chunks.len(), 1);
3397
3398        published.insert(0, Arc::new(parking_lot::RwLock::new(vec![0])));
3399        assert_eq!(
3400            published.chunks.len(),
3401            1,
3402            "one insert after a merge must not split again"
3403        );
3404        assert!(published.remove(&0).is_some());
3405        assert_eq!(
3406            published.chunks.len(),
3407            1,
3408            "one remove after a merge must not change chunking"
3409        );
3410    }
3411
3412    #[test]
3413    fn attach_boundary_repair_is_a_complete_publication_by_itself() {
3414        let set = BTreeSet::<u64>::with_maximum_node_size(8);
3415        set.attach_node(vec![1, 10]);
3416        assert_eq!(set.remove(&10), Some(10));
3417
3418        // Model attachment stopping after its preflight repair (for example,
3419        // because user-provided Clone/Ord code panics while reading the first
3420        // incoming node). The repaired identity map and route snapshot must
3421        // still commit together when the guard drops.
3422        {
3423            let mut index = set.index.write();
3424            index.repair_last_route_before_insert();
3425        }
3426        set.attach_node(vec![5, 20]);
3427
3428        assert_eq!(set.get(&5).map(|found| *found.get()), Some(5));
3429    }
3430
3431    #[cfg(debug_assertions)]
3432    #[test]
3433    #[should_panic(expected = "generic topology removal requires publication")]
3434    fn publication_must_be_enabled_before_an_opt_out_guard_mutates() {
3435        let set = BTreeSet::<u64>::with_maximum_node_size(8);
3436        set.attach_node(vec![1, 10]);
3437        let mut index = set.index.write_rekey();
3438        index.remove(&10);
3439    }
3440
3441    #[test]
3442    fn published_point_routes_match_a_sequential_oracle_under_churn() {
3443        let set = BTreeSet::<u64>::with_maximum_node_size(4);
3444        let mut oracle = std::collections::BTreeSet::new();
3445        let mut state = 0x8f4d_2a71_c390_6be5u64;
3446
3447        for step in 0..2_000 {
3448            // Fixed xorshift stream: deterministic inserts/removes repeatedly
3449            // grow, shrink, empty, and split tiny nodes.
3450            state ^= state << 13;
3451            state ^= state >> 7;
3452            state ^= state << 17;
3453            let key = state % 64;
3454            if state & 1 == 0 {
3455                assert_eq!(set.insert(key), oracle.insert(key), "insert step {step}, key {key}");
3456            } else {
3457                assert_eq!(
3458                    set.remove(&key).is_some(),
3459                    oracle.remove(&key),
3460                    "remove step {step}, key {key}"
3461                );
3462            }
3463
3464            for probe in 0..64 {
3465                assert_eq!(
3466                    set.contains(&probe),
3467                    oracle.contains(&probe),
3468                    "point route diverged at step {step}, probe {probe}"
3469                );
3470            }
3471            assert_eq!(
3472                set.iter().collect::<Vec<_>>(),
3473                oracle.iter().copied().collect::<Vec<_>>()
3474            );
3475        }
3476    }
3477
3478    #[test]
3479    fn concurrent_remove_reinsert_over_emptying_nodes_preserves_all_keys() {
3480        const THREADS: u64 = 4;
3481        const ITERATIONS: u64 = 1_000;
3482
3483        // Tiny nodes over adjacent keys: removes empty nodes constantly, so
3484        // inserts keep racing pending MakeUnreachable repairs.
3485        let set = Arc::new(BTreeSet::<u64>::with_maximum_node_size(2));
3486        for key in 0..THREADS {
3487            set.insert(key);
3488        }
3489
3490        let start = Arc::new(Barrier::new(THREADS as usize));
3491        let (done_tx, done_rx) = mpsc::channel();
3492        let handles = (0..THREADS)
3493            .map(|key| {
3494                let set = Arc::clone(&set);
3495                let start = Arc::clone(&start);
3496                let done_tx = done_tx.clone();
3497                thread::spawn(move || {
3498                    start.wait();
3499                    for _ in 0..ITERATIONS {
3500                        // A point remove may transiently miss while another
3501                        // writer's index repair is still in flight, but the
3502                        // acknowledged insert must never be LOST: under a
3503                        // stable snapshot the key must still be somewhere, and
3504                        // the self-healing repairs must make it removable
3505                        // again promptly.
3506                        let mut attempts = 0;
3507                        while set.remove(&key).is_none() {
3508                            let index = set.index.read();
3509                            let present = index.values().any(|node| node.read().contains(&key));
3510                            drop(index);
3511                            assert!(present, "acknowledged insert of {key} was lost");
3512                            attempts += 1;
3513                            assert!(attempts < 10_000, "key {key} present but never became removable");
3514                            std::hint::spin_loop();
3515                        }
3516                        assert!(set.insert(key), "{key} still present after acknowledged remove");
3517                    }
3518                    done_tx.send(()).unwrap();
3519                })
3520            })
3521            .collect::<Vec<_>>();
3522        drop(done_tx);
3523
3524        for _ in 0..THREADS {
3525            done_rx
3526                .recv_timeout(Duration::from_secs(30))
3527                .expect("remove/reinsert workload did not complete in time");
3528        }
3529        for handle in handles {
3530            handle.join().unwrap();
3531        }
3532
3533        for key in 0..THREADS {
3534            assert!(set.contains(&key), "key {key} lost after churn");
3535            assert_eq!(set.remove(&key), Some(key));
3536        }
3537        assert!(set.is_empty());
3538    }
3539
3540    #[test]
3541    fn last_node_rekey_that_crosses_predecessor_is_published() {
3542        let set = BTreeSet::<u64>::with_maximum_node_size(2);
3543        set.attach_nodes([vec![0], vec![2], vec![3]]);
3544        let node = set.index.read().last_key_value().expect("last node").1.clone();
3545
3546        // Model the window where the final node is emptied and refilled below
3547        // its predecessor before the delayed boundary repair acquires the
3548        // structural lock.
3549        {
3550            let mut values = node.write();
3551            NodeLike::delete(&mut *values, &3).expect("seeded value");
3552            NodeLike::insert(&mut *values, 1);
3553        }
3554        Operation::UpdateMax(node, 3)
3555            .commit::<false>(&mut set.index.write_rekey(), super::no_identity_adoption)
3556            .expect("boundary repair");
3557
3558        assert_eq!(set.iter().collect::<Vec<_>>(), vec![0, 1, 2]);
3559        for value in 0..=2 {
3560            assert_eq!(set.get(&value).map(|found| *found.get()), Some(value));
3561        }
3562    }
3563
3564    #[test]
3565    fn structural_insert_repairs_a_stale_last_route() {
3566        let set = BTreeSet::<u64>::with_maximum_node_size(2);
3567        set.attach_nodes([vec![0], vec![3]]);
3568        let node = set.index.read().last_key_value().expect("last node").1.clone();
3569
3570        // This ordinary shrink can retain the old final fallback route.
3571        {
3572            let mut values = node.write();
3573            NodeLike::delete(&mut *values, &3).expect("seeded value");
3574            NodeLike::insert(&mut *values, 1);
3575        }
3576        Operation::UpdateMax(node, 3)
3577            .commit::<false>(&mut set.index.write_rekey(), super::no_identity_adoption)
3578            .expect("boundary repair");
3579
3580        // Installing a route after that node must first make its published
3581        // boundary exact, or the new node sorts before it only in the reader
3582        // snapshot.
3583        set.index.write().insert(2, Arc::new(RwLock::new(vec![2])));
3584
3585        assert_eq!(set.iter().collect::<Vec<_>>(), vec![0, 1, 2]);
3586        for value in 0..=2 {
3587            assert_eq!(set.get(&value).map(|found| *found.get()), Some(value));
3588        }
3589    }
3590
3591    #[test]
3592    fn test_remove_single_element() {
3593        let set = BTreeSet::<i32>::new();
3594        set.insert(5);
3595        assert!(set.contains(&5));
3596        assert!(set.remove(&5).is_some());
3597        assert!(!set.contains(&5));
3598        assert!(!set.remove(&5).is_some());
3599    }
3600
3601    #[test]
3602    fn test_remove_multiple_elements() {
3603        let set = BTreeSet::<i32>::new();
3604        for i in 0..2048 {
3605            set.insert(i);
3606        }
3607        for i in 0..2048 {
3608            assert!(set.remove(&i).is_some());
3609            assert!(!set.contains(&i));
3610        }
3611        assert_eq!(set.len(), 0);
3612    }
3613
3614    #[test]
3615    fn test_remove_non_existent() {
3616        let set = BTreeSet::<i32>::new();
3617        set.insert(5);
3618        assert!(!set.remove(&10).is_some());
3619        assert!(set.contains(&5));
3620    }
3621
3622    #[test]
3623    fn test_remove_stress() {
3624        let set = Arc::new(BTreeSet::<i32>::new());
3625        const NUM_ELEMENTS: i32 = 10000;
3626
3627        for i in 0..NUM_ELEMENTS {
3628            set.insert(i);
3629        }
3630        assert_eq!(set.len(), NUM_ELEMENTS as usize, "Incorrect size after insertion");
3631
3632        let num_threads = 8;
3633        let elements_per_thread = NUM_ELEMENTS / num_threads;
3634        let handles: Vec<_> = (0..num_threads)
3635            .map(|t| {
3636                let set = Arc::clone(&set);
3637                thread::spawn(move || {
3638                    for i in (t * elements_per_thread)..((t + 1) * elements_per_thread) {
3639                        if i % 2 == 1 {
3640                            assert!(set.remove(&i).is_some(), "Failed to remove {}", i);
3641                        }
3642                    }
3643                })
3644            })
3645            .collect();
3646
3647        for handle in handles {
3648            handle.join().unwrap();
3649        }
3650
3651        assert_eq!(set.len(), NUM_ELEMENTS as usize / 2, "Incorrect size after removal");
3652
3653        for i in 0..NUM_ELEMENTS {
3654            if i % 2 == 0 {
3655                assert!(set.contains(&i), "Even number {} should be in the set", i);
3656            } else {
3657                assert!(!set.contains(&i), "Odd number {} should not be in the set", i);
3658            }
3659        }
3660    }
3661
3662    #[test]
3663    fn test_remove_all_elements() {
3664        let set = BTreeSet::<i32>::new();
3665        let n = 2048;
3666
3667        for i in 0..n {
3668            set.insert(i);
3669        }
3670
3671        for i in 0..n {
3672            assert!(set.remove(&i).is_some(), "Failed to remove {}", i);
3673        }
3674
3675        assert_eq!(set.len(), 0, "Set should be empty");
3676
3677        for i in 0..n {
3678            assert!(!set.contains(&i), "Element {} should not be in the set", i);
3679        }
3680    }
3681
3682    #[test]
3683    fn test_range_edge_cases() {
3684        let set = BTreeSet::<i32>::with_maximum_node_size(10);
3685        for i in 0..20 {
3686            set.insert(i);
3687        }
3688        // Nodes are:
3689        // [0, 1, 2, 3, 4]
3690        // [5, 6, 7, 8, 9]
3691        // [10, 11, 12, 13, 14, 15, 16, 17, 18, 19]
3692
3693        // First value of the node only
3694        assert_eq!(set.range(0..=0).collect::<Vec<_>>(), vec![0]);
3695        assert_eq!(set.range(0..1).collect::<Vec<_>>(), vec![0]);
3696
3697        assert_eq!(set.range(5..=5).collect::<Vec<_>>(), vec![5]);
3698        assert_eq!(set.range(5..6).collect::<Vec<_>>(), vec![5]);
3699
3700        assert_eq!(set.range(10..=10).collect::<Vec<_>>(), vec![10]);
3701        assert_eq!(set.range(10..11).collect::<Vec<_>>(), vec![10]);
3702
3703        // From first value to middle
3704        assert_eq!(set.range(0..=3).collect::<Vec<_>>(), vec![0, 1, 2, 3]);
3705        assert_eq!(set.range(0..3).collect::<Vec<_>>(), vec![0, 1, 2]);
3706
3707        assert_eq!(set.range(5..=8).collect::<Vec<_>>(), vec![5, 6, 7, 8]);
3708        assert_eq!(set.range(5..8).collect::<Vec<_>>(), vec![5, 6, 7]);
3709
3710        assert_eq!(set.range(10..=13).collect::<Vec<_>>(), vec![10, 11, 12, 13]);
3711        assert_eq!(set.range(10..13).collect::<Vec<_>>(), vec![10, 11, 12]);
3712
3713        // Last value of the node
3714        assert_eq!(set.range(4..=4).collect::<Vec<_>>(), vec![4]);
3715        assert_eq!(set.range(4..5).collect::<Vec<_>>(), vec![4]);
3716
3717        assert_eq!(set.range(9..=9).collect::<Vec<_>>(), vec![9]);
3718        assert_eq!(set.range(9..10).collect::<Vec<_>>(), vec![9]);
3719
3720        assert_eq!(set.range(19..=19).collect::<Vec<_>>(), vec![19]);
3721        assert_eq!(set.range(19..20).collect::<Vec<_>>(), vec![19]);
3722
3723        // From middle to last value of the node
3724        assert_eq!(set.range(17..=19).collect::<Vec<_>>(), vec![17, 18, 19]);
3725        assert_eq!(set.range(17..20).collect::<Vec<_>>(), vec![17, 18, 19]);
3726
3727        assert_eq!(set.range(7..=9).collect::<Vec<_>>(), vec![7, 8, 9]);
3728        assert_eq!(set.range(7..10).collect::<Vec<_>>(), vec![7, 8, 9]);
3729
3730        assert_eq!(set.range(2..=4).collect::<Vec<_>>(), vec![2, 3, 4]);
3731        assert_eq!(set.range(2..5).collect::<Vec<_>>(), vec![2, 3, 4]);
3732
3733        // Full node
3734        assert_eq!(set.range(0..=4).collect::<Vec<_>>(), vec![0, 1, 2, 3, 4]);
3735        assert_eq!(set.range(0..5).collect::<Vec<_>>(), vec![0, 1, 2, 3, 4]);
3736
3737        assert_eq!(set.range(5..=9).collect::<Vec<_>>(), vec![5, 6, 7, 8, 9]);
3738        assert_eq!(set.range(5..10).collect::<Vec<_>>(), vec![5, 6, 7, 8, 9]);
3739
3740        assert_eq!(
3741            set.range(10..=19).collect::<Vec<_>>(),
3742            vec![10, 11, 12, 13, 14, 15, 16, 17, 18, 19]
3743        );
3744        assert_eq!(
3745            set.range(10..20).collect::<Vec<_>>(),
3746            vec![10, 11, 12, 13, 14, 15, 16, 17, 18, 19]
3747        );
3748
3749        // Node intersection
3750        assert_eq!(set.range(3..=6).collect::<Vec<_>>(), vec![3, 4, 5, 6]);
3751        assert_eq!(set.range(3..7).collect::<Vec<_>>(), vec![3, 4, 5, 6]);
3752
3753        assert_eq!(set.range(8..=11).collect::<Vec<_>>(), vec![8, 9, 10, 11]);
3754        assert_eq!(set.range(8..12).collect::<Vec<_>>(), vec![8, 9, 10, 11]);
3755
3756        // REVERSED
3757
3758        // First value of the node only
3759        assert_eq!(set.range(0..=0).rev().collect::<Vec<_>>(), vec![0]);
3760        assert_eq!(set.range(0..1).rev().collect::<Vec<_>>(), vec![0]);
3761
3762        assert_eq!(set.range(5..=5).rev().collect::<Vec<_>>(), vec![5]);
3763        assert_eq!(set.range(5..6).rev().collect::<Vec<_>>(), vec![5]);
3764
3765        assert_eq!(set.range(10..=10).rev().collect::<Vec<_>>(), vec![10]);
3766        assert_eq!(set.range(10..11).rev().collect::<Vec<_>>(), vec![10]);
3767
3768        // From first value to middle
3769        assert_eq!(set.range(0..=3).rev().collect::<Vec<_>>(), vec![3, 2, 1, 0]);
3770        assert_eq!(set.range(0..3).rev().collect::<Vec<_>>(), vec![2, 1, 0]);
3771
3772        assert_eq!(set.range(5..=8).rev().collect::<Vec<_>>(), vec![8, 7, 6, 5]);
3773        assert_eq!(set.range(5..8).rev().collect::<Vec<_>>(), vec![7, 6, 5]);
3774
3775        assert_eq!(set.range(10..=13).rev().collect::<Vec<_>>(), vec![13, 12, 11, 10]);
3776        assert_eq!(set.range(10..13).rev().collect::<Vec<_>>(), vec![12, 11, 10]);
3777
3778        // Last value of the node
3779        assert_eq!(set.range(4..=4).rev().collect::<Vec<_>>(), vec![4]);
3780        assert_eq!(set.range(4..5).rev().collect::<Vec<_>>(), vec![4]);
3781
3782        assert_eq!(set.range(9..=9).rev().collect::<Vec<_>>(), vec![9]);
3783        assert_eq!(set.range(9..10).rev().collect::<Vec<_>>(), vec![9]);
3784
3785        assert_eq!(set.range(19..=19).rev().collect::<Vec<_>>(), vec![19]);
3786        assert_eq!(set.range(19..20).rev().collect::<Vec<_>>(), vec![19]);
3787
3788        // From middle to last value of the node
3789        assert_eq!(set.range(17..=19).rev().collect::<Vec<_>>(), vec![19, 18, 17]);
3790        assert_eq!(set.range(17..20).rev().collect::<Vec<_>>(), vec![19, 18, 17]);
3791
3792        assert_eq!(set.range(7..=9).rev().collect::<Vec<_>>(), vec![9, 8, 7]);
3793        assert_eq!(set.range(7..10).rev().collect::<Vec<_>>(), vec![9, 8, 7]);
3794
3795        assert_eq!(set.range(2..=4).rev().collect::<Vec<_>>(), vec![4, 3, 2]);
3796        assert_eq!(set.range(2..5).rev().collect::<Vec<_>>(), vec![4, 3, 2]);
3797
3798        // Full node
3799        assert_eq!(set.range(0..=4).rev().collect::<Vec<_>>(), vec![4, 3, 2, 1, 0]);
3800        assert_eq!(set.range(0..5).rev().collect::<Vec<_>>(), vec![4, 3, 2, 1, 0]);
3801
3802        assert_eq!(set.range(5..=9).rev().collect::<Vec<_>>(), vec![9, 8, 7, 6, 5]);
3803        assert_eq!(set.range(5..10).rev().collect::<Vec<_>>(), vec![9, 8, 7, 6, 5]);
3804
3805        assert_eq!(
3806            set.range(10..=19).rev().collect::<Vec<_>>(),
3807            vec![19, 18, 17, 16, 15, 14, 13, 12, 11, 10]
3808        );
3809        assert_eq!(
3810            set.range(10..20).rev().collect::<Vec<_>>(),
3811            vec![19, 18, 17, 16, 15, 14, 13, 12, 11, 10]
3812        );
3813
3814        // Node intersection
3815        assert_eq!(set.range(3..=6).rev().collect::<Vec<_>>(), vec![6, 5, 4, 3]);
3816        assert_eq!(set.range(3..7).rev().collect::<Vec<_>>(), vec![6, 5, 4, 3]);
3817
3818        assert_eq!(set.range(8..=11).rev().collect::<Vec<_>>(), vec![11, 10, 9, 8]);
3819        assert_eq!(set.range(8..12).rev().collect::<Vec<_>>(), vec![11, 10, 9, 8]);
3820
3821        // Non-existent range
3822        assert!(set.range(20..).collect::<Vec<_>>().is_empty());
3823        assert!(set.range(..0).collect::<Vec<_>>().is_empty());
3824        assert!(set.range(20..).rev().collect::<Vec<_>>().is_empty());
3825        assert!(set.range(..0).rev().collect::<Vec<_>>().is_empty());
3826    }
3827
3828    #[test]
3829    fn concurrent_range_constructions_at_node_boundaries_do_not_deadlock() {
3830        const THREAD_ITERATIONS: usize = 20_000;
3831
3832        let set = Arc::new(BTreeSet::<u64>::with_maximum_node_size(4));
3833        for value in 0..64 {
3834            set.insert(value);
3835        }
3836
3837        // One thread constructs ranges whose start sits at node minima
3838        // (locking a node, then its predecessor); the other constructs
3839        // ranges whose end sits at node maxima (locking a node, then its
3840        // successor). Pre-fix these acquisitions ran in opposite orders
3841        // while both locks were held, an ABBA deadlock.
3842        let (done_tx, done_rx) = mpsc::channel();
3843        let forward = {
3844            let set = Arc::clone(&set);
3845            let done_tx = done_tx.clone();
3846            thread::spawn(move || {
3847                for iteration in 0..THREAD_ITERATIONS {
3848                    let start = (iteration % 64) as u64;
3849                    assert_eq!(set.range(start..).next(), Some(start));
3850                }
3851                done_tx.send(()).unwrap();
3852            })
3853        };
3854        let backward = {
3855            let set = Arc::clone(&set);
3856            let done_tx = done_tx.clone();
3857            thread::spawn(move || {
3858                for iteration in 0..THREAD_ITERATIONS {
3859                    let end = (iteration % 64) as u64;
3860                    assert_eq!(set.range(..=end).next_back(), Some(end));
3861                }
3862                done_tx.send(()).unwrap();
3863            })
3864        };
3865        drop(done_tx);
3866
3867        for _ in 0..2 {
3868            done_rx
3869                .recv_timeout(Duration::from_secs(30))
3870                .expect("concurrent range constructions deadlocked");
3871        }
3872        forward.join().unwrap();
3873        backward.join().unwrap();
3874    }
3875
3876    // Builds nodes [0, 10] (key 10), [20, 30] (key 30), [40, 50, 60] (key 60).
3877    fn three_node_set() -> BTreeSet<u64> {
3878        let set = BTreeSet::<u64>::with_maximum_node_size(4);
3879        for value in [0u64, 10, 20, 30, 40, 50, 60] {
3880            set.insert(value);
3881        }
3882        assert_eq!(
3883            set.index.read().keys().copied().collect::<Vec<_>>(),
3884            vec![10, 30, 60],
3885            "fixture geometry changed"
3886        );
3887        set
3888    }
3889
3890    #[test]
3891    fn forward_scan_repositions_when_current_node_vanishes() {
3892        let set = three_node_set();
3893
3894        let mut iter = set.iter();
3895        assert_eq!(iter.next(), Some(0));
3896        assert_eq!(iter.next(), Some(10));
3897        assert_eq!(iter.next(), Some(20));
3898
3899        // The node the iterator is parked in vanishes from the index, as
3900        // UpdateMax's remove-then-insert re-key does on every monotonic
3901        // insert. The scan must reposition, not end.
3902        set.index.write().remove(&30).expect("fixture entry");
3903
3904        assert_eq!(iter.next(), Some(30));
3905        assert_eq!(iter.next(), Some(40));
3906        assert_eq!(iter.next(), Some(50));
3907        assert_eq!(iter.next(), Some(60));
3908        assert_eq!(iter.next(), None);
3909    }
3910
3911    #[test]
3912    fn backward_scan_repositions_when_current_node_vanishes() {
3913        let set = three_node_set();
3914
3915        let mut iter = set.iter();
3916        assert_eq!(iter.next_back(), Some(60));
3917        assert_eq!(iter.next_back(), Some(50));
3918
3919        set.index.write().remove(&60).expect("fixture entry");
3920
3921        assert_eq!(iter.next_back(), Some(40));
3922        assert_eq!(iter.next_back(), Some(30));
3923        assert_eq!(iter.next_back(), Some(20));
3924        assert_eq!(iter.next_back(), Some(10));
3925        assert_eq!(iter.next_back(), Some(0));
3926        assert_eq!(iter.next_back(), None);
3927    }
3928
3929    #[test]
3930    fn forward_scan_does_not_re_yield_after_split_of_finished_node() {
3931        // Nodes [0, 10] (key 10) and [20, 30, 40] (key 40), as left behind by
3932        // a split of [0, 10, 20, 30].
3933        let set = BTreeSet::<u64>::with_maximum_node_size(4);
3934        for value in [0u64, 10, 20, 30, 40] {
3935            set.insert(value);
3936        }
3937
3938        // An iterator that had already yielded through 20 from the pre-split
3939        // node and just exhausted the lower half: advancing into the upper
3940        // half must not re-yield 20.
3941        let iter = Iter {
3942            tree: &set,
3943            current_front_batch: None,
3944            current_back_batch: None,
3945            exhausted_front_node: Some(set.index.read().first_key_value().expect("fixture node").1.clone()),
3946            exhausted_back_node: None,
3947            front_partial: None,
3948            back_partial: None,
3949            front_batch_limit: INITIAL_BATCH,
3950            back_batch_limit: INITIAL_BATCH,
3951            current_front_value: Some(20),
3952            current_back_value: None,
3953            met: false,
3954        };
3955
3956        assert_eq!(iter.collect::<Vec<_>>(), vec![30, 40]);
3957    }
3958
3959    #[test]
3960    fn backward_scan_does_not_re_yield_values_from_scanned_range() {
3961        // A stale-key window: the front node's maximum (35) grew past its
3962        // index key (10) while the backward scan had already advanced below
3963        // 30. Advancing into that node must not yield 35 again.
3964        let set = BTreeSet::<u64>::new();
3965        set.attach_node(vec![0u64, 10]);
3966        set.attach_node(vec![30u64, 40]);
3967        {
3968            let node = set.index.read().first_key_value().expect("fixture node").1.clone();
3969            let mut guard = node.write();
3970            NodeLike::insert(&mut *guard, 35u64);
3971        }
3972
3973        let mut iter = Iter {
3974            tree: &set,
3975            current_front_batch: None,
3976            current_back_batch: None,
3977            exhausted_front_node: None,
3978            exhausted_back_node: Some(set.index.read().last_key_value().expect("fixture node").1.clone()),
3979            front_partial: None,
3980            back_partial: None,
3981            front_batch_limit: INITIAL_BATCH,
3982            back_batch_limit: INITIAL_BATCH,
3983            current_front_value: None,
3984            current_back_value: Some(30),
3985            met: false,
3986        };
3987
3988        let mut collected = vec![];
3989        while let Some(value) = iter.next_back() {
3990            collected.push(value);
3991        }
3992        assert_eq!(collected, vec![10, 0]);
3993    }
3994
3995    #[test]
3996    fn backward_scan_does_not_skip_values_split_away_after_positioning() {
3997        // The heavy-tier churn failure in deterministic form. A backward
3998        // scan chooses its node (at construction or when advancing) and only
3999        // later locks it to read. A split committed in that window keeps the
4000        // node's LOWER half in the chosen Arc and moves the upper half to a
4001        // new node: values the scan has not yielded yet migrate above its
4002        // resume point and are silently skipped. Node selection and the
4003        // content read must be one atomic step under the structural guard.
4004        let set = BTreeSet::<u64>::with_maximum_node_size(4);
4005        for value in [0u64, 10, 20, 30] {
4006            set.insert(value);
4007        }
4008
4009        // Position the scan on the (single) node...
4010        let mut iter = set.iter();
4011        // ...then let a writer split it before the scan reads anything:
4012        // [0, 10] stays in the original Arc, [20, 30, 40] moves to a new
4013        // node above it.
4014        set.insert(40);
4015        assert!(set.node_count() > 1, "fixture must split");
4016
4017        let mut collected = vec![];
4018        while let Some(value) = iter.next_back() {
4019            collected.push(value);
4020        }
4021
4022        // 40 was inserted mid-scan, so a weakly consistent scan may or may
4023        // not observe it; every baseline value must be yielded. (Linear
4024        // scan on purpose: with NodeLike in scope, Vec::contains resolves
4025        // to NodeLike's binary search, which is wrong on this descending
4026        // vector.)
4027        for baseline in [30u64, 20, 10, 0] {
4028            assert!(
4029                collected.iter().any(|value| *value == baseline),
4030                "baseline value {baseline} skipped by backward scan (yielded: {collected:?})"
4031            );
4032        }
4033        assert!(
4034            collected.windows(2).all(|pair| pair[0] > pair[1]),
4035            "backward scan not strictly decreasing: {collected:?}"
4036        );
4037    }
4038
4039    #[test]
4040    fn bidirectional_meet_into_opposite_held_node_does_not_self_deadlock() {
4041        // Nodes [0, 10] (key 10) and [20, 30, 40] (key 40).
4042        let set = Arc::new(BTreeSet::<u64>::with_maximum_node_size(4));
4043        for value in [0u64, 10, 20, 30, 40] {
4044            set.insert(value);
4045        }
4046
4047        let (done_tx, done_rx) = mpsc::channel();
4048        let handle = {
4049            let set = Arc::clone(&set);
4050            thread::spawn(move || {
4051                // The back cursor enters the final node, then the forward
4052                // end exhausts the first node and must enter the node the
4053                // back cursor is positioned in to yield the middle. Under
4054                // the guard-holding design this double-locked the
4055                // non-reentrant node mutex; owned batches must keep this
4056                // lock-free.
4057                let mut finished = set.iter();
4058                assert_eq!(finished.next_back(), Some(40));
4059                assert_eq!(finished.next(), Some(0));
4060                assert_eq!(finished.next(), Some(10));
4061                assert_eq!(finished.next(), Some(20));
4062                assert_eq!(finished.next(), Some(30));
4063                assert_eq!(finished.next(), None);
4064                assert_eq!(finished.next_back(), None);
4065
4066                // `finished` met in the middle and stays alive: a finished
4067                // iterator must hold no node locks, or the next lock of its
4068                // final node (here by a second iterator on the same thread)
4069                // self-deadlocks.
4070                let mut iter = set.iter();
4071                assert_eq!(iter.next(), Some(0));
4072                assert_eq!(iter.next_back(), Some(40));
4073                assert_eq!(iter.next_back(), Some(30));
4074                assert_eq!(iter.next_back(), Some(20));
4075                assert_eq!(iter.next_back(), Some(10));
4076                assert_eq!(iter.next_back(), None);
4077                assert_eq!(iter.next(), None);
4078                drop(finished);
4079
4080                done_tx.send(()).unwrap();
4081            })
4082        };
4083
4084        done_rx
4085            .recv_timeout(Duration::from_secs(10))
4086            .expect("bidirectional meet-in-the-middle deadlocked");
4087        handle.join().unwrap();
4088    }
4089
4090    #[test]
4091    fn structural_commits_complete_against_paused_scan() {
4092        // Several tiny nodes; the scan will pin the first one.
4093        let set = Arc::new(BTreeSet::<u64>::with_maximum_node_size(2));
4094        for value in 0..8 {
4095            set.insert(value);
4096        }
4097
4098        let scan_holds_guard = Arc::new(Barrier::new(3));
4099        let (done_tx, done_rx) = mpsc::channel();
4100
4101        let scanner = {
4102            let set = Arc::clone(&set);
4103            let scan_holds_guard = Arc::clone(&scan_holds_guard);
4104            let done_tx = done_tx.clone();
4105            thread::spawn(move || {
4106                let mut iter = set.iter();
4107                // A paused scan must hold no node lock between calls;
4108                // under the guard-holding design the first node's mutex
4109                // stayed pinned here.
4110                assert_eq!(iter.next(), Some(0));
4111                scan_holds_guard.wait();
4112                // Give the writer time to take the structural write lock
4113                // and the first node's mutex while the scan is parked.
4114                thread::sleep(Duration::from_millis(100));
4115                // Resuming in the opposite direction acquires the
4116                // structural read guard; if the scan still held a node
4117                // mutex here it would deadlock ABBA against the writer
4118                // (writer: topology -> node).
4119                let mut collected = vec![];
4120                while let Some(value) = iter.next_back() {
4121                    collected.push(value);
4122                }
4123                assert_eq!(collected, vec![7, 6, 5, 4, 3, 2, 1]);
4124                done_tx.send(()).unwrap();
4125            })
4126        };
4127
4128        let writer = {
4129            let set = Arc::clone(&set);
4130            let scan_holds_guard = Arc::clone(&scan_holds_guard);
4131            let done_tx = done_tx.clone();
4132            thread::spawn(move || {
4133                scan_holds_guard.wait();
4134                // remove_range acquires the structural write lock and then
4135                // locks the node pinned by the scanner.
4136                set.remove_range(0..=0);
4137                done_tx.send(()).unwrap();
4138            })
4139        };
4140        drop(done_tx);
4141        scan_holds_guard.wait();
4142
4143        for _ in 0..2 {
4144            done_rx
4145                .recv_timeout(Duration::from_secs(10))
4146                .expect("scan or structural commit deadlocked");
4147        }
4148        scanner.join().unwrap();
4149        writer.join().unwrap();
4150        assert!(!set.contains(&0));
4151    }
4152
4153    #[test]
4154    fn full_scans_do_not_degrade_quadratically_with_node_count() {
4155        use std::time::Instant;
4156
4157        // Many tiny nodes: the node-advance cost dominates the scan.
4158        const VALUES: u64 = 30_000;
4159        let set = BTreeSet::<u64>::with_maximum_node_size(2);
4160        for value in 0..VALUES {
4161            set.insert(value);
4162        }
4163        assert!(
4164            set.node_count() >= (VALUES / 4) as usize,
4165            "fixture must be a many-node tree, got {} nodes",
4166            set.node_count()
4167        );
4168
4169        let started = Instant::now();
4170        assert_eq!(set.iter().count(), VALUES as usize);
4171        let forward = started.elapsed();
4172
4173        let started = Instant::now();
4174        assert_eq!(set.iter().rev().count(), VALUES as usize);
4175        let backward = started.elapsed();
4176
4177        // Advancing between nodes costs one logarithmic index lookup, so both
4178        // scans finish in milliseconds even in a debug build. The removed
4179        // linear identity relocation made each advance walk the index from
4180        // the front (~N^2/2 entry visits per scan, well over a minute at this
4181        // node count), so the generous budget still fails it decisively.
4182        let budget = Duration::from_secs(10);
4183        assert!(
4184            forward < budget,
4185            "forward scan took {forward:?}, node advance is not logarithmic"
4186        );
4187        assert!(
4188            backward < budget,
4189            "backward scan took {backward:?}, node advance is not logarithmic"
4190        );
4191    }
4192
4193    #[test]
4194    fn scans_stay_sorted_and_complete_under_monotonic_insert_churn() {
4195        use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
4196
4197        const BASELINE: u64 = 400;
4198        const EXTRA: u64 = 2_000;
4199        const SCAN_BOUND: usize = 10_000;
4200
4201        // Small nodes: every monotonic insert re-keys the last node and
4202        // regularly splits it, exercising the reposition paths constantly.
4203        let set = Arc::new(BTreeSet::<u64>::with_maximum_node_size(8));
4204        for value in 0..BASELINE {
4205            set.insert(value);
4206        }
4207
4208        let done = Arc::new(AtomicBool::new(false));
4209        let writer = {
4210            let set = Arc::clone(&set);
4211            let done = Arc::clone(&done);
4212            thread::spawn(move || {
4213                for value in BASELINE..BASELINE + EXTRA {
4214                    assert!(set.insert(value));
4215                }
4216                done.store(true, AtomicOrdering::Release);
4217            })
4218        };
4219
4220        let mut scans = 0usize;
4221        loop {
4222            let forward = set.iter().collect::<Vec<_>>();
4223            assert!(
4224                forward.windows(2).all(|pair| pair[0] < pair[1]),
4225                "forward scan not strictly increasing (duplicate or unordered yield)"
4226            );
4227            assert_eq!(
4228                forward.iter().filter(|value| **value < BASELINE).count() as u64,
4229                BASELINE,
4230                "forward scan truncated: baseline keys missing"
4231            );
4232
4233            let backward = set.iter().rev().collect::<Vec<_>>();
4234            assert!(
4235                backward.windows(2).all(|pair| pair[0] > pair[1]),
4236                "backward scan not strictly decreasing (duplicate or unordered yield)"
4237            );
4238            assert_eq!(
4239                backward.iter().filter(|value| **value < BASELINE).count() as u64,
4240                BASELINE,
4241                "backward scan truncated: baseline keys missing"
4242            );
4243
4244            scans += 1;
4245            if done.load(AtomicOrdering::Acquire) || scans >= SCAN_BOUND {
4246                break;
4247            }
4248        }
4249
4250        writer.join().unwrap();
4251
4252        let expected = (0..BASELINE + EXTRA).collect::<Vec<_>>();
4253        assert_eq!(set.iter().collect::<Vec<_>>(), expected);
4254        assert_eq!(set.iter().rev().collect::<Vec<_>>(), {
4255            let mut reversed = expected;
4256            reversed.reverse();
4257            reversed
4258        });
4259    }
4260
4261    #[test]
4262    fn collected_owned_values_survive_arbitrary_concurrent_mutation() {
4263        use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
4264
4265        const BASELINE: u64 = 512;
4266        const CHURN: u64 = 4_000;
4267
4268        // Small nodes so the churn constantly splits, re-keys, and unlinks
4269        // the nodes the scans are walking.
4270        let set = Arc::new(BTreeSet::<u64>::with_maximum_node_size(8));
4271        for value in 0..BASELINE {
4272            set.insert(value);
4273        }
4274
4275        let done = Arc::new(AtomicBool::new(false));
4276        let writer = {
4277            let set = Arc::clone(&set);
4278            let done = Arc::clone(&done);
4279            thread::spawn(move || {
4280                for value in BASELINE..BASELINE + CHURN {
4281                    assert!(set.insert(value));
4282                    assert_eq!(set.remove(&value), Some(value));
4283                }
4284                done.store(true, AtomicOrdering::Release);
4285            })
4286        };
4287
4288        // The type annotation is the point: collect() yields owned values,
4289        // not references tied to node storage. Under the previous borrowed
4290        // design this collected Vec<&u64> whose referents were unlocked node
4291        // slots, a use-after-free under exactly this churn.
4292        let mut snapshots: Vec<Vec<u64>> = Vec::new();
4293        loop {
4294            let snapshot: Vec<u64> = set.iter().collect();
4295            snapshots.push(snapshot);
4296            if done.load(AtomicOrdering::Acquire) {
4297                break;
4298            }
4299        }
4300        writer.join().unwrap();
4301
4302        // Mutate the set arbitrarily after the snapshots were taken; the
4303        // snapshots must remain fully usable because they own their values.
4304        set.remove_range(..);
4305        assert!(set.is_empty());
4306
4307        for snapshot in snapshots {
4308            assert!(
4309                snapshot.windows(2).all(|pair| pair[0] < pair[1]),
4310                "snapshot not strictly increasing"
4311            );
4312            assert_eq!(
4313                snapshot.iter().filter(|value| **value < BASELINE).count() as u64,
4314                BASELINE,
4315                "snapshot lost baseline keys"
4316            );
4317        }
4318    }
4319
4320    #[test]
4321    fn parallel_iter_and_mut() {
4322        let set = Arc::new(BTreeSet::<i32>::new());
4323        for i in 0..10_000 {
4324            set.insert(i);
4325        }
4326
4327        let set_clone = Arc::clone(&set);
4328        let handle = thread::spawn(move || {
4329            for _ in 0..1000 {
4330                let mut _sum = 0;
4331                for value in set_clone.iter() {
4332                    _sum += value;
4333                }
4334            }
4335        });
4336
4337        for i in 10_000..20_000 {
4338            set.insert(i);
4339        }
4340        handle.join().unwrap();
4341    }
4342
4343    /// A scan that spans several batch installs inside one node yields every
4344    /// element, once, in order.
4345    ///
4346    /// The batch is bounded and grows, so a node larger than `INITIAL_BATCH` is
4347    /// consumed over several installs rather than one. Each install re-selects
4348    /// the node by cursor and skips what has already been taken, which is where
4349    /// a partial batch can silently drop or repeat elements. A whole-node batch
4350    /// could not get this wrong because it never resumed inside a node.
4351    #[test]
4352    fn a_scan_across_several_batch_installs_is_complete_and_ordered() {
4353        let set: BTreeSet<u64> = BTreeSet::new();
4354        // Comfortably more than INITIAL_BATCH, and more than the first few
4355        // doublings, so the scan resumes inside a node repeatedly.
4356        let count = (INITIAL_BATCH * 20) as u64;
4357        for i in 0..count {
4358            set.insert(i);
4359        }
4360
4361        let seen: Vec<u64> = set.iter().collect();
4362        let expected: Vec<u64> = (0..count).collect();
4363        assert_eq!(seen, expected, "a partial-batch scan lost or repeated elements");
4364    }
4365
4366    /// The same property backwards.
4367    #[test]
4368    fn a_backward_scan_across_several_installs_is_complete_and_ordered() {
4369        let set: BTreeSet<u64> = BTreeSet::new();
4370        let count = (INITIAL_BATCH * 20) as u64;
4371        for i in 0..count {
4372            set.insert(i);
4373        }
4374
4375        let seen: Vec<u64> = set.iter().rev().collect();
4376        let expected: Vec<u64> = (0..count).rev().collect();
4377        assert_eq!(
4378            seen, expected,
4379            "a partial-batch backward scan lost or repeated elements"
4380        );
4381    }
4382
4383    /// A scan over a node big enough to hold everything, so every install after
4384    /// the first resumes inside the same node.
4385    #[test]
4386    fn a_scan_within_a_single_node_resumes_correctly() {
4387        let set: BTreeSet<u64> = BTreeSet::with_maximum_node_size(DEFAULT_INNER_SIZE);
4388        let count = 200u64;
4389        for i in 0..count {
4390            set.insert(i);
4391        }
4392        assert_eq!(set.node_count(), 1, "fixture wants one node");
4393
4394        let seen: Vec<u64> = set.iter().collect();
4395        assert_eq!(seen, (0..count).collect::<Vec<_>>());
4396    }
4397
4398    /// A one-element range yields exactly that element.
4399    ///
4400    /// The case the bounded batch exists for: this used to clone every
4401    /// remaining element of the node it landed in to produce one value.
4402    #[test]
4403    fn a_single_element_range_yields_one_element() {
4404        let set: BTreeSet<u64> = BTreeSet::new();
4405        for i in 0..1_000u64 {
4406            set.insert(i);
4407        }
4408
4409        for probe in [0u64, 1, 499, 998, 999] {
4410            let got: Vec<u64> = set.range(probe..=probe).collect();
4411            assert_eq!(got, vec![probe], "range({probe}..={probe})");
4412        }
4413        assert!(set.range(1_000..=1_000).next().is_none(), "absent key");
4414    }
4415
4416    /// Ranges of every width across a batch boundary.
4417    ///
4418    /// Widths either side of `INITIAL_BATCH` and its first doublings are where
4419    /// an off-by-one in the resume arithmetic shows up, and nowhere else.
4420    #[test]
4421    fn ranges_spanning_batch_boundaries_are_exact() {
4422        let set: BTreeSet<u64> = BTreeSet::new();
4423        for i in 0..500u64 {
4424            set.insert(i);
4425        }
4426
4427        for width in 1..=(INITIAL_BATCH * 8) as u64 {
4428            let start = 100u64;
4429            let got: Vec<u64> = set.range(start..start + width).collect();
4430            let expected: Vec<u64> = (start..start + width).collect();
4431            assert_eq!(got, expected, "range width {width}");
4432        }
4433    }
4434
4435    /// Meeting in the middle still terminates and yields each element once.
4436    ///
4437    /// Both cursors now resume inside nodes, so the point at which they meet is
4438    /// reached through a different sequence of installs than before.
4439    #[test]
4440    fn a_double_ended_scan_meets_without_repeating() {
4441        let set: BTreeSet<u64> = BTreeSet::new();
4442        let count = (INITIAL_BATCH * 10) as u64;
4443        for i in 0..count {
4444            set.insert(i);
4445        }
4446
4447        let mut iter = set.iter();
4448        let mut front = Vec::new();
4449        let mut back = Vec::new();
4450        loop {
4451            match iter.next() {
4452                Some(v) => front.push(v),
4453                None => break,
4454            }
4455            match iter.next_back() {
4456                Some(v) => back.push(v),
4457                None => break,
4458            }
4459        }
4460        back.reverse();
4461        front.extend(back);
4462        front.sort_unstable();
4463        assert_eq!(
4464            front,
4465            (0..count).collect::<Vec<_>>(),
4466            "double-ended scan is not a partition"
4467        );
4468    }
4469
4470    /// A scan under concurrent mutation terminates and does not stream
4471    /// duplicates forever.
4472    ///
4473    /// This is the case the partial-skip guard exists for, and it cannot be
4474    /// reached from one thread. A batch that stops short of a node's end
4475    /// resumes inside that node by cursor rank; a concurrent split or re-key
4476    /// can leave that rank *below* elements already yielded, the yield path
4477    /// then drops the whole batch as duplicates, and the next install computes
4478    /// the same skip again. Without the recorded take count that is a scan
4479    /// which never advances.
4480    ///
4481    /// A stall is asserted as a bound rather than by waiting: the scan is
4482    /// capped, and a run that reaches the cap is one that was not making
4483    /// progress. Removing the guard makes this fail rather than hang, which is
4484    /// the difference between a test and a timeout.
4485    #[test]
4486    fn a_scan_under_concurrent_mutation_terminates() {
4487        use std::sync::atomic::{AtomicBool, Ordering};
4488        use std::sync::Arc;
4489
4490        const SIZE: u64 = 4_000;
4491        // Generous: any honest scan yields at most SIZE plus whatever is
4492        // inserted while it runs. Reaching this many means it is looping.
4493        const CAP: usize = (SIZE * 20) as usize;
4494
4495        for _ in 0..8 {
4496            let set: Arc<BTreeSet<u64>> = Arc::new(BTreeSet::new());
4497            for i in 0..SIZE {
4498                set.insert(i);
4499            }
4500            let stop = Arc::new(AtomicBool::new(false));
4501
4502            // Churn that forces splits and re-keys under the scan.
4503            let writers: Vec<_> = (0..3)
4504                .map(|w| {
4505                    let (set, stop) = (Arc::clone(&set), Arc::clone(&stop));
4506                    std::thread::spawn(move || {
4507                        let mut i = SIZE + w * 100_000;
4508                        while !stop.load(Ordering::Relaxed) {
4509                            set.insert(i);
4510                            set.remove(&i);
4511                            i += 1;
4512                        }
4513                    })
4514                })
4515                .collect();
4516
4517            let mut yielded = 0usize;
4518            for _ in set.iter() {
4519                yielded += 1;
4520                if yielded >= CAP {
4521                    break;
4522                }
4523            }
4524
4525            stop.store(true, Ordering::Relaxed);
4526            for w in writers {
4527                w.join().expect("writer did not panic");
4528            }
4529
4530            assert!(
4531                yielded < CAP,
4532                "scan did not make progress under concurrent mutation: {yielded} yields"
4533            );
4534        }
4535    }
4536
4537    /// `Vec::contains` is not usable in this module: `NodeLike` is in scope and
4538    /// its `contains` for `Vec<T>` is a *binary search*, which silently answers
4539    /// nonsense for any sequence that is not sorted ascending. A scan's output
4540    /// is exactly such a sequence when it runs backwards.
4541    // Clippy suggests `seen.contains(&value)` here. Taking that suggestion
4542    // reintroduces the exact bug this helper exists to avoid, which is why the
4543    // lint is silenced rather than followed.
4544    #[allow(clippy::manual_contains)]
4545    fn yielded(seen: &[u64], value: u64) -> bool {
4546        seen.iter().any(|item| *item == value)
4547    }
4548
4549    /// WTI-1: `front_partial` counts *positions*, and a position is not a
4550    /// stable cursor under deletion.
4551    ///
4552    /// Deleting an element the scan already yielded shifts the unyielded tail
4553    /// left while the recorded count stays put, so the stale position wins the
4554    /// `max` and steps over a live element. Key `0` is removed after it has
4555    /// been yielded; every key above it was present for the whole scan and must
4556    /// still appear, which is what the iterator promises.
4557    ///
4558    /// The prefix is swept because the defect only bites when the deletion
4559    /// lands while the scan is partway through a node, and where that boundary
4560    /// falls depends on the doubling batch limit.
4561    #[test]
4562    fn deleting_a_yielded_element_does_not_skip_a_live_one() {
4563        for prefix in 1..12usize {
4564            let set: BTreeSet<u64> = BTreeSet::new();
4565            for i in 0..256u64 {
4566                set.insert(i);
4567            }
4568
4569            let mut seen = Vec::new();
4570            for value in set.iter() {
4571                seen.push(value);
4572                if seen.len() == prefix {
4573                    set.remove(&0);
4574                }
4575            }
4576
4577            for expected in 1..256u64 {
4578                assert!(
4579                    yielded(&seen, expected),
4580                    "prefix {prefix}: {expected} was present for the whole scan but was never yielded"
4581                );
4582            }
4583        }
4584    }
4585
4586    /// The backward mirror. `back_partial` trims from the end rather than
4587    /// skipping from the start, so the same staleness would drop an element off
4588    /// the low end of the scan.
4589    #[test]
4590    fn deleting_a_yielded_element_backwards_does_not_skip_a_live_one() {
4591        for prefix in 1..12usize {
4592            let set: BTreeSet<u64> = BTreeSet::new();
4593            for i in 0..256u64 {
4594                set.insert(i);
4595            }
4596
4597            let mut seen = Vec::new();
4598            for value in set.iter().rev() {
4599                seen.push(value);
4600                if seen.len() == prefix {
4601                    set.remove(&255);
4602                }
4603            }
4604
4605            for expected in 0..255u64 {
4606                assert!(
4607                    yielded(&seen, expected),
4608                    "prefix {prefix}: {expected} was present for the whole scan but was never yielded"
4609                );
4610            }
4611        }
4612    }
4613}