pub fn apply_batch(&self, batch: Batch) -> io::Result<()> {
        // NB: we rely on lexicographic lock acquisition
        // by iterating over the batch's BTreeMap to avoid
        // deadlocks during 2PL
        let mut acquired_locks: BTreeMap<
            InlineArray,
            (
                ArcRwLockWriteGuard<RawRwLock, CacheBox<LEAF_FANOUT>>,
                Object<LEAF_FANOUT>,
            ),
        > = BTreeMap::new();

        // Phase 1: lock acquisition
        let mut last: Option<(
            InlineArray,
            ArcRwLockWriteGuard<RawRwLock, CacheBox<LEAF_FANOUT>>,
            Object<LEAF_FANOUT>,
        )> = None;

        for key in batch.writes.keys() {
            if let Some((_lo, w, _id)) = &last {
                let leaf = w.leaf.as_ref().unwrap();
                assert!(&leaf.lo <= key);
                if let Some(hi) = &leaf.hi {
                    if hi <= key {
                        let (lo, w, n) = last.take().unwrap();
                        acquired_locks.insert(lo, (w, n));
                    }
                }
            }
            if last.is_none() {
                // TODO evaluate whether this is correct, as page_in performs
                // cache maintenance internally if it over/undershoots due to
                // concurrent modifications.
                last =
                    Some(self.page_in(key, self.cache.current_flush_epoch())?);
            }
        }

        if let Some((lo, w, id)) = last.take() {
            acquired_locks.insert(lo, (w, id));
        }

        // NB: add the flush epoch at the end of the lock acquisition
        // process when all locks have been acquired, to avoid situations
        // where a leaf is already dirty with an epoch "from the future".
        let flush_epoch_guard = self.cache.check_into_flush_epoch();
        let new_epoch = flush_epoch_guard.epoch();

        // Flush any leaves that are dirty from a previous flush epoch
        // before performing operations.
        for (write, node) in acquired_locks.values_mut() {
            let leaf = write.leaf.as_mut().unwrap();
            if let Some(old_flush_epoch) = leaf.dirty_flush_epoch {
                if old_flush_epoch == new_epoch {
                    // no need to cooperatively flush
                    continue;
                }

                assert!(old_flush_epoch < new_epoch);

                log::trace!(
                    "cooperatively flushing {:?} with dirty {:?} after checking into {:?}",
                    node.object_id,
                    old_flush_epoch,
                    new_epoch
                );

                self.cooperatively_serialize_leaf(node.object_id, &mut *leaf);
            }
        }

        let mut splits: Vec<(InlineArray, Object<LEAF_FANOUT>)> = vec![];
        let mut merges: BTreeMap<InlineArray, Object<LEAF_FANOUT>> =
            BTreeMap::new();

        // Insert and split when full
        for (key, value_opt) in batch.writes {
            let range = ..=&key;
            let (lo, (w, object)) = acquired_locks
                .range_mut::<InlineArray, _>(range)
                .next_back()
                .unwrap();
            let leaf = w.leaf.as_mut().unwrap();

            assert_eq!(lo, &leaf.lo);
            assert!(leaf.lo <= key);
            if let Some(hi) = &leaf.hi {
                assert!(hi > &key);
            }

            if let Some(value) = value_opt {
                leaf.insert(key, value);
                merges.remove(lo);

                merges.remove(&leaf.lo);

                if let Some((split_key, rhs_node)) = leaf.split_if_full(
                    new_epoch,
                    &self.cache,
                    self.collection_id,
                ) {
                    #[cfg(feature = "for-internal-testing-only")]
                    let _b1 = track_blocks();

                    let write = rhs_node.inner.write_arc();
                    assert!(write.leaf.is_some());

                    splits.push((split_key.clone(), rhs_node.clone()));
                    acquired_locks.insert(split_key, (write, rhs_node));
                }
            } else {
                leaf.remove(&key);

                if leaf.is_empty() {
                    assert_eq!(leaf.lo, lo);
                    merges.insert(leaf.lo.clone(), object.clone());
                }
            }
        }

        // Make splits globally visible
        for (split_key, rhs_node) in splits {
            self.cache
                .object_id_index
                .insert(rhs_node.object_id, rhs_node.clone());
            self.index.insert(split_key, rhs_node);
        }

        // Add all written leaves to dirty and prepare to mark cache accesses
        let mut cache_accesses = Vec::with_capacity(acquired_locks.len());
        for (low_key, (write, node)) in &mut acquired_locks {
            let leaf = write.leaf.as_mut().unwrap();
            leaf.set_dirty_epoch(new_epoch);
            leaf.mutation_count += 1;
            cache_accesses.push((node.object_id, leaf.in_memory_size));
            self.cache.install_dirty(
                new_epoch,
                node.object_id,
                Dirty::NotYetSerialized {
                    collection_id: self.collection_id,
                    node: node.clone(),
                    low_key: low_key.clone(),
                },
            );

            #[cfg(feature = "for-internal-testing-only")]
            {
                self.cache.event_verifier.mark(
                    node.object_id,
                    new_epoch,
                    event_verifier::State::Dirty,
                    concat!(file!(), ':', line!(), ":apply-batch"),
                );
            }
        }

        // Drop locks
        drop(acquired_locks);

        // Perform cache maintenance
        for (object_id, size) in cache_accesses {
            self.cache.mark_access_and_evict(object_id, size, new_epoch)?;
        }

        Ok(())
    }
Create a new batched update that is applied
atomically. Readers will atomically see all updates
at an atomic instant, and if the database crashes,
either 0% or 100% of the full batch will be recovered,
but never a partial batch. If a `flush` operation succeeds
after this, it is guaranteed that 100% of the batch will be
visible, unless later concurrent updates changed the values
before the flush.

# Examples

```
# fn main() -> Result<(), Box<dyn std::error::Error>> {
# let _ = std::fs::remove_dir_all("batch_doctest");
# let db: sled::Db<1024> = sled::open("batch_doctest")?;
db.insert("key_0", "val_0")?;

let mut batch = sled::Batch::default();
batch.insert("key_a", "val_a");
batch.insert("key_b", "val_b");
batch.insert("key_c", "val_c");
batch.remove("key_0");

db.apply_batch(batch)?;
// key_0 no longer exists, and key_a, key_b, and key_c
// now do exist.
# let _ = std::fs::remove_dir_all("batch_doctest");
# Ok(()) }
```
src/tree.rs