armdb 0.14.2

use std::mem::size_of;
use std::num::NonZeroU32;
use std::ops::{Bound, ControlFlow};
use std::sync::Arc;

#[cfg(feature = "rpc")]
use armour_rpc::protocol::WriteKey;

use armour_core::{KeyScheme, Typ};

#[cfg(feature = "rpc")]
use crate::access::HeldOps;
use crate::access::{Access, Wire};
#[cfg(feature = "rpc")]
use crate::armour::handler::{
    KeyBytes, RowWriter, ValueBytes, bound_to_key_bound, check_key_len, resolve_key,
};
use crate::armour::inspection::{
    InspectionCollection, InspectionEntry, InspectionError, InspectionHandler, InspectionOperation,
    InspectionPage, InspectionSchema, IterationDirection, NormalizedBounds, PageAccumulator,
    ScanCapability, SnapshotAccumulator,
};
use crate::armour::seq::SeqGen;
use crate::error::{DbError, DbResult};
use crate::key::Key;
use crate::map_index::inspection_batch;
use crate::tree_meta::TreeMeta;

fn iteration_direction(reversed: bool) -> IterationDirection {
    if reversed {
        IterationDirection::Descending
    } else {
        IterationDirection::Ascending
    }
}

/// Owned-bound counterpart to the RPC-side `bound_to_key_bound`: it converts
/// bounds the `Inspector`/`normalize_range` path has already validated, so
/// unlike `bound_to_key_bound` it performs no `len() == size_of::<K>()`
/// check. Its `DbResult` return has no failure path today — `K::from_bytes`
/// does not fail — and is kept only for call-site symmetry with the RPC path.
fn bound_to_key_bound_owned<K: Key>(bound: &Bound<Vec<u8>>) -> DbResult<Bound<K>> {
    match bound {
        Bound::Included(b) => Ok(Bound::Included(K::from_bytes(b))),
        Bound::Excluded(b) => Ok(Bound::Excluded(K::from_bytes(b))),
        Bound::Unbounded => Ok(Bound::Unbounded),
    }
}

/// One generic bridge from a collection's [`Access`] description to both
/// consumer surfaces: it implements `InspectionHandler` and, under the `rpc`
/// feature, `RpcHandler` — a single `Adapter<S, W>` backs every built-in
/// collection family on both surfaces.
pub(crate) struct Adapter<S, W> {
    pub(crate) name: String,
    /// Cached type-hash, reported by `RpcHandler::info` and `::schema`.
    #[cfg_attr(not(feature = "rpc"), allow(dead_code))]
    pub(crate) typ_hash: u64,
    pub(crate) ty: Typ,
    pub(crate) key_scheme: KeyScheme,
    pub(crate) group_bits: u32,
    pub(crate) version: u16,
    pub(crate) access: Arc<S>,
    pub(crate) wire: W,
    /// Sequence generator backing `WriteKey::Sequence` writes.
    #[cfg_attr(not(feature = "rpc"), allow(dead_code))]
    pub(crate) seq: Arc<SeqGen>,
}

impl<S, W> Adapter<S, W> {
    pub(crate) fn new(meta: &TreeMeta, access: Arc<S>, wire: W, seq: Arc<SeqGen>) -> Self {
        Self {
            name: meta.name.to_owned(),
            typ_hash: meta.ty.h(),
            ty: meta.ty,
            key_scheme: meta.key_scheme,
            group_bits: meta.group_bits,
            version: meta.version,
            access,
            wire,
            seq,
        }
    }
}

impl<S, W> InspectionHandler for Adapter<S, W>
where
    S: Access,
    W: Wire<S::Val>,
{
    fn key_group_bits(&self) -> u32 {
        self.group_bits
    }

    fn metadata(&self) -> Result<InspectionCollection, InspectionError> {
        let scan = self.access.scan_capability();
        Ok(InspectionCollection {
            name: self.name.clone(),
            schema: InspectionSchema {
                version: self.version,
                typ: self.ty,
                key_scheme: self.key_scheme,
                key_size: size_of::<S::Key>() as u32,
                // Zero derives this from the `Typ` and the value width at
                // runtime; every other family returns `C::INSPECTION_ENCODING`.
                encoding: self.access.encoding(&self.ty),
            },
            storage: S::STORAGE,
            count: Some(self.access.approx_len()),
            scan,
            // Live contract: a non-iterable Map reports `None`, everything
            // ordered reports its iteration direction.
            direction: (scan == ScanCapability::Ordered)
                .then(|| iteration_direction(self.access.reversed())),
        })
    }

    fn get_raw(&self, key: &[u8]) -> DbResult<Option<Vec<u8>>> {
        // No length check here: `Inspector::get` validates against `key_size`
        // before dispatch (inspection.rs), and hardening it would change the
        // error variant callers already see.
        let key = S::Key::from_bytes(key);
        // LSM: storage and wire share the codec, so the stored bytes are the
        // wire bytes. Decided at compile time — a raw miss is final and costs
        // one lookup, via the `Access::stored_bytes` + `Wire::validate_stored`
        // pair (the only family where `S::RAW_WIRE_BYTES && W::PASSTHROUGH`
        // both hold).
        if S::RAW_WIRE_BYTES && W::PASSTHROUGH {
            return match self.access.stored_bytes(&key)? {
                Some(raw) => {
                    self.wire.validate_stored(&raw)?;
                    Ok(Some(raw))
                }
                None => Ok(None),
            };
        }
        self.access
            .with_value(&key, |value| self.wire.encode(value))?
            .transpose()
    }

    fn exact_count(&self) -> DbResult<u64> {
        // Deliberately NOT `rpc_exact_count`: on `VarTypedTree` the Inspector
        // counts live keys strictly and the RPC surface counts decodable rows.
        self.access.inspection_exact_count()
    }

    fn ordered_page(
        &self,
        bounds: &NormalizedBounds,
        limit: NonZeroU32,
    ) -> DbResult<InspectionPage> {
        // A non-iterable Map answers `501` — checked before the empty-bounds
        // shortcut, exactly as the live Map handlers do, so a caller never gets
        // an empty page in place of the capability error.
        if self.access.scan_capability() != ScanCapability::Ordered {
            return Err(DbError::NotImplemented);
        }
        if bounds.empty {
            return Ok(InspectionPage {
                entries: vec![],
                next_key: None,
            });
        }
        let start = bound_to_key_bound_owned::<S::Key>(&bounds.start)?;
        let end = bound_to_key_bound_owned::<S::Key>(&bounds.end)?;
        let batch = inspection_batch(limit, self.access.inspection_shard_count());
        let wire = &self.wire;
        let mut page = PageAccumulator::new(limit);
        // Strict scan: a row that cannot be read or decoded fails the page
        // instead of shortening it.
        self.access
            .try_scan_range(start.as_ref(), end.as_ref(), batch, &mut |key, value| {
                page.push_with(|| {
                    Ok(InspectionEntry {
                        key: key.as_bytes().to_vec(),
                        value: wire.encode(value)?,
                    })
                })
            })?;
        Ok(page.finish())
    }

    fn unordered_snapshot(
        &self,
        max_entries: NonZeroU32,
    ) -> Result<Vec<InspectionEntry>, InspectionError> {
        // Trees and iterable Maps are `Ordered` and reject this operation.
        if self.access.scan_capability() != ScanCapability::UnorderedSnapshot {
            return Err(InspectionError::UnsupportedOperation {
                name: self.name.clone(),
                operation: InspectionOperation::SnapshotUnordered,
            });
        }
        // Non-iterable Maps keep an exact live count, so the cap is checked
        // before any row is read.
        let observed = self.access.approx_len();
        let cap = max_entries.get();
        if observed > u64::from(cap) {
            return Err(InspectionError::SnapshotTooLarge {
                limit: cap,
                observed: Some(observed),
            });
        }
        let mut acc = SnapshotAccumulator::new(max_entries);
        let mut failure = None;
        self.access.scan_all(&mut |row| {
            let (key, value) = match row {
                Ok(pair) => pair,
                // First storage/decode failure aborts the snapshot: the first
                // `Err` row from `Access::scan_all` breaks the walk here.
                Err(error) => {
                    failure = Some(error);
                    return ControlFlow::Break(());
                }
            };
            match self.wire.encode(value) {
                Ok(bytes) => acc.push(InspectionEntry {
                    key: key.as_bytes().to_vec(),
                    value: bytes,
                }),
                Err(error) => {
                    failure = Some(error);
                    ControlFlow::Break(())
                }
            }
        });
        if let Some(error) = failure {
            return Err(InspectionError::Database(error));
        }
        acc.finish()
    }
}

#[cfg(feature = "rpc")]
impl<S, W> crate::armour::handler::RpcHandler for Adapter<S, W>
where
    S: Access,
    W: Wire<S::Val>,
{
    fn name(&self) -> &str {
        &self.name
    }

    fn info(&self) -> (u64, u16) {
        (self.typ_hash, self.version)
    }

    fn schema(&self) -> armour_rpc::SchemaResponse {
        armour_rpc::SchemaResponse {
            name: self.name.clone(),
            version: self.version,
            typ: self.ty,
            key_scheme: self.key_scheme,
        }
    }

    fn scannable(&self) -> bool {
        self.access.scan_capability() == ScanCapability::Ordered
    }

    fn get_into(&self, key: &[u8], out: &mut Vec<u8>) -> DbResult<bool> {
        let key: S::Key = check_key_len(key)?;
        // LSM: stored bytes are wire bytes, decided before any I/O so a miss
        // costs one lookup.
        if S::RAW_WIRE_BYTES && W::PASSTHROUGH {
            return match self.access.stored_bytes(&key)? {
                Some(raw) => {
                    self.wire.validate_stored(&raw)?;
                    out.extend_from_slice(&raw);
                    Ok(true)
                }
                None => Ok(false),
            };
        }
        // `Access::with_value` is the strict read, kept deliberately over the
        // lenient facade `get`: on `VarTypedTree`/`VarTypedMap` an unreadable or
        // undecodable value now fails the request instead of surfacing as a
        // successful empty response. The client sees only an opaque `500
        // internal error` (`err_response` redacts it); the real cause reaches
        // `tracing::error!`, not the wire. Every other family fronts an
        // infallible read, so nothing else moves.
        self.access
            .with_value(&key, |value| self.wire.encode_append(value, out))?
            .transpose()
            .map(|value| value.is_some())
    }

    fn contains(&self, key: &[u8]) -> DbResult<bool> {
        let key: S::Key = check_key_len(key)?;
        self.access.contains(&key)
    }

    fn entry_len(&self, key: &[u8]) -> DbResult<Option<u32>> {
        let key: S::Key = check_key_len(key)?;
        self.access.entry_len(&key, &self.wire)
    }

    fn first_into(&self, writer: &mut RowWriter<'_>) -> DbResult<bool> {
        self.access
            .with_first(|key, value| {
                writer.row(key.as_bytes(), |out| self.wire.encode_append(value, out))
            })?
            .transpose()
            .map(|value| value.is_some())
    }

    fn last_into(&self, writer: &mut RowWriter<'_>) -> DbResult<bool> {
        self.access
            .with_last(|key, value| {
                writer.row(key.as_bytes(), |out| self.wire.encode_append(value, out))
            })?
            .transpose()
            .map(|value| value.is_some())
    }

    fn range_into(
        &self,
        start: Bound<KeyBytes>,
        end: Bound<KeyBytes>,
        writer: &mut RowWriter<'_>,
    ) -> DbResult<u32> {
        // Iterability first, before the bounds are even parsed — matching the
        // live Map handlers' precedence. A malformed bound against a
        // non-iterable Map must still answer `NotImplemented` (501), not a
        // bound-parse `Client` error (400).
        if self.access.scan_capability() != ScanCapability::Ordered {
            return Err(DbError::NotImplemented);
        }
        let start = bound_to_key_bound::<S::Key>(&start)?;
        let end = bound_to_key_bound::<S::Key>(&end)?;
        let wire = &self.wire;
        // Lenient scan — the RPC `range` contract. A row that will not decode
        // is skipped (logged) on the families whose iterators do that today.
        self.access
            .scan_range(start.as_ref(), end.as_ref(), &mut |key, value| {
                writer.row(key.as_bytes(), |out| wire.encode_append(value, out))
            })?;
        Ok(writer.rows())
    }

    fn range_keys_into(
        &self,
        start: Bound<KeyBytes>,
        end: Bound<KeyBytes>,
        writer: &mut RowWriter<'_>,
    ) -> DbResult<u32> {
        // Same precedence as `range` above: capability before bounds.
        if self.access.scan_capability() != ScanCapability::Ordered {
            return Err(DbError::NotImplemented);
        }
        let start = bound_to_key_bound::<S::Key>(&start)?;
        let end = bound_to_key_bound::<S::Key>(&end)?;
        // Yields keys only, but the value is not skipped everywhere: the three
        // Map families' `keys_range_bounds` never touches a value. The
        // Tree/VarTyped/LSM key walk shares the row iterator `range` uses and
        // reads (VarTyped: decodes) the value just to step past it, then
        // discards it without ever handing it to `f`.
        self.access
            .scan_keys(start.as_ref(), end.as_ref(), &mut |key| {
                writer.key_only(key.as_bytes())
            })?;
        Ok(writer.rows())
    }

    fn write(&self, key: WriteKey, flag: Option<bool>, value: ValueBytes) -> DbResult<ValueBytes> {
        let (typed_key, key_bytes) = resolve_key::<S::Key>(key, &self.seq, &self.name)?;
        let decoded = self.wire.decode(&value)?;
        match flag {
            None => {
                self.access.put(&typed_key, decoded)?;
                Ok(key_bytes)
            }
            // Existence-only precondition under one held shard lock. Do not
            // route through `upsert_via_held`/`with_value`: an undecodable but
            // present row must stay `KeyExists`/`KeyNotFound`, not corruption.
            // (Collection `upsert` still decodes; that is a different contract.)
            Some(false) => self.access.atomic_at(&typed_key, |shard| {
                if shard.try_contains(&typed_key)? {
                    return Err(DbError::KeyExists);
                }
                shard.put(&typed_key, decoded)?;
                Ok(key_bytes)
            }),
            Some(true) => self.access.atomic_at(&typed_key, |shard| {
                if !shard.try_contains(&typed_key)? {
                    return Err(DbError::KeyNotFound);
                }
                shard.put(&typed_key, decoded)?;
                Ok(key_bytes)
            }),
        }
    }

    fn remove(&self, key: &[u8], soft: bool) -> DbResult<()> {
        if soft {
            return Err(DbError::NotImplemented);
        }
        let key: S::Key = check_key_len(key)?;
        self.access.delete(&key)
    }

    fn take_into(&self, key: &[u8], soft: bool, out: &mut Vec<u8>) -> DbResult<bool> {
        if soft {
            return Err(DbError::NotImplemented);
        }
        let key: S::Key = check_key_len(key)?;
        let wire = &self.wire;
        self.access.atomic_at(&key, |shard| {
            let Some(encoded) = shard.with_value(&key, |value| wire.encode_append(value, out))?
            else {
                return Ok(false);
            };
            encoded?;
            shard.delete(&key)?;
            Ok(true)
        })
    }

    fn count(&self, exact: bool) -> DbResult<u64> {
        if exact {
            // NOT `inspection_exact_count` — on `VarTypedTree` the RPC surface
            // counts decodable rows while the Inspector counts live keys.
            self.access.rpc_exact_count()
        } else {
            Ok(self.access.approx_len())
        }
    }

    fn apply_batch(&self, items: Vec<(KeyBytes, Option<ValueBytes>)>) -> DbResult<()> {
        // Validate and decode the whole batch first: one bad item must abort
        // before any write lands.
        let decoded: Vec<(S::Key, Option<S::Val>)> = items
            .into_iter()
            .map(|(key, value)| {
                if key.len() != size_of::<S::Key>() {
                    return Err(DbError::Client("invalid key length in batch"));
                }
                let value = value
                    .map(|bytes| self.wire.decode_batch_item(&bytes))
                    .transpose()?;
                Ok((S::Key::from_bytes(&key), value))
            })
            .collect::<DbResult<_>>()?;
        for (key, value) in decoded {
            match value {
                Some(value) => self.access.put(&key, value)?,
                None => self.access.delete(&key)?,
            }
        }
        Ok(())
    }
}

#[cfg(test)]
#[cfg(feature = "typed-tree")]
#[cfg(feature = "rapira-codec")]
mod tests {
    use std::num::NonZeroU32;
    use std::ops::Bound;

    use armour_core::{CollectionShape, GetType, StorageBackend, StorageClass, ValueModel};
    use rapira::Rapira;

    use crate::access::{Adapter, CodecWire};
    use crate::armour::Db;
    use crate::armour::inspection::InspectionError;
    use crate::armour::inspection::{
        InspectionHandler, InspectionOperation, IterationDirection, ScanCapability, normalize_range,
    };
    use crate::tree_meta::TreeMeta;
    use crate::{CollectionMeta, Config, NoHook, RapiraCodec};

    #[derive(Clone, Debug, PartialEq, Rapira, GetType)]
    struct AdapterRow {
        n: u64,
    }
    impl CollectionMeta for AdapterRow {
        type SelfId = [u8; 8];
        const NAME: &'static str = "adapter_typed_tree";
        const VERSION: u16 = 7;
    }

    fn key(id: u64) -> [u8; 8] {
        id.to_be_bytes()
    }

    /// A byte-level regression guard for the streaming RPC surface: replacing
    /// `RowWriter` with a collected row representation would no longer emit
    /// the key/value length prefixes in the response frame.
    #[cfg(feature = "rpc")]
    #[test]
    fn range_into_writes_rows_in_wire_order() {
        use crate::armour::handler::RowWriter;

        let dir = tempfile::tempdir().expect("dir");
        let db = Db::open_test(dir.path()).expect("open");
        let tree = db
            .open_typed_tree::<AdapterRow, RapiraCodec, _>(
                Config::balanced()
                    .shard_count(2)
                    .hints(true)
                    .reversed(false)
                    .build(),
                NoHook,
                &[],
            )
            .expect("open tree");
        tree.put(&key(1), AdapterRow { n: 7 }).expect("put");
        tree.put(&key(2), AdapterRow { n: 8 }).expect("put");

        let handler = db.rpc_handler_for(AdapterRow::NAME).expect("handler");
        let mut buf = Vec::new();
        let mut writer = RowWriter::new(&mut buf, 0, 1 << 20, AdapterRow::NAME);
        let rows = handler
            .range_into(Bound::Unbounded, Bound::Unbounded, &mut writer)
            .expect("range");

        assert_eq!(rows, 2);
        assert_eq!(&buf[..4], &8u32.to_be_bytes(), "first key length");
        assert_eq!(&buf[4..12], &1u64.to_be_bytes(), "first key");
    }

    #[test]
    fn adapter_reports_metadata_reads_and_pages_for_a_typed_tree() {
        let dir = tempfile::tempdir().expect("tmp");
        let db = Db::open_test(dir.path()).expect("db");
        let cfg = Config::balanced()
            .shard_count(2)
            .hints(true)
            .reversed(false)
            .build();
        let tree = db
            .open_typed_tree::<AdapterRow, RapiraCodec, _>(cfg, NoHook, &[])
            .expect("tree");
        for id in 1..=5 {
            tree.put(&key(id), AdapterRow { n: id }).expect("seed");
        }

        let mut tree_meta = TreeMeta::of::<AdapterRow>();
        tree_meta.group_bits = 11;
        let adapter = Adapter::new(
            &tree_meta,
            tree.clone(),
            CodecWire(RapiraCodec),
            db.seq.clone(),
        );

        assert_eq!(adapter.key_group_bits(), 11);

        let meta = adapter.metadata().expect("metadata");
        assert_eq!(meta.name, AdapterRow::NAME);
        assert_eq!(meta.schema.version, 7);
        assert_eq!(meta.schema.key_size, 8);
        assert_eq!(
            meta.storage,
            StorageClass {
                model: ValueModel::Typed,
                shape: CollectionShape::Tree,
                backend: StorageBackend::Bitcask,
            }
        );
        assert_eq!(meta.count, Some(5));
        assert_eq!(meta.scan, ScanCapability::Ordered);
        assert_eq!(meta.direction, Some(IterationDirection::Ascending));

        let raw = adapter.get_raw(&key(3)).expect("get_raw").expect("present");
        assert_eq!(raw, rapira::serialize(&AdapterRow { n: 3 }));
        assert!(adapter.get_raw(&key(99)).expect("get_raw").is_none());
        assert_eq!(adapter.exact_count().expect("exact count"), 5);

        // Page of 2 out of 5: cursor is the last returned key.
        let bounds = normalize_range(
            8,
            IterationDirection::Ascending,
            Bound::Unbounded,
            Bound::Unbounded,
            None,
        )
        .expect("bounds");
        let page = adapter
            .ordered_page(&bounds, NonZeroU32::new(2).unwrap())
            .expect("page");
        assert_eq!(
            page.entries
                .iter()
                .map(|e| e.key.clone())
                .collect::<Vec<_>>(),
            vec![key(1).to_vec(), key(2).to_vec()]
        );
        assert_eq!(page.next_key, Some(key(2).to_vec()));

        // Last page: no cursor.
        let bounds = normalize_range(
            8,
            IterationDirection::Ascending,
            Bound::Unbounded,
            Bound::Unbounded,
            Some(&key(3)),
        )
        .expect("bounds");
        let page = adapter
            .ordered_page(&bounds, NonZeroU32::new(10).unwrap())
            .expect("page");
        assert_eq!(page.entries.len(), 2);
        assert!(page.next_key.is_none());

        // An ordered collection refuses the unordered snapshot.
        let err = adapter
            .unordered_snapshot(NonZeroU32::new(10).unwrap())
            .unwrap_err();
        assert!(matches!(
            err,
            InspectionError::UnsupportedOperation {
                operation: InspectionOperation::SnapshotUnordered,
                ..
            }
        ));
    }

    #[cfg(feature = "rpc")]
    #[test]
    fn adapter_rpc_read_write_count_and_batch_on_a_typed_tree() {
        use armour_rpc::protocol::WriteKey;
        use std::ops::Bound;

        use crate::armour::handler::{RowWriter, RpcHandler};

        let dir = tempfile::tempdir().expect("tmp");
        let db = Db::open_test(dir.path()).expect("db");
        let cfg = Config::balanced()
            .shard_count(2)
            .hints(true)
            .reversed(false)
            .build();
        let tree = db
            .open_typed_tree::<AdapterRow, RapiraCodec, _>(cfg, NoHook, &[])
            .expect("tree");
        for id in 1..=5 {
            tree.put(&key(id), AdapterRow { n: id }).expect("seed");
        }

        let adapter = Adapter::new(
            &TreeMeta::of::<AdapterRow>(),
            tree.clone(),
            CodecWire(RapiraCodec),
            db.seq.clone(),
        );

        assert_eq!(adapter.name(), AdapterRow::NAME);
        assert_eq!(adapter.info(), (AdapterRow::TYPE.h(), 7));
        assert!(adapter.scannable());

        let encoded = rapira::serialize(&AdapterRow { n: 3 });
        let mut get_bytes = Vec::new();
        assert!(adapter.get_into(&key(3), &mut get_bytes).expect("get"));
        assert_eq!(get_bytes, encoded);
        let mut missing_bytes = Vec::new();
        assert!(!adapter.get_into(&key(99), &mut missing_bytes).expect("get"));
        assert!(missing_bytes.is_empty());
        assert!(adapter.contains(&key(3)).expect("contains"));
        assert_eq!(
            adapter.entry_len(&key(3)).expect("entry_len"),
            Some(encoded.len() as u32)
        );

        // A wrong-length key is `KeyNotFound`, not a panic — `check_key_len`.
        let mut invalid_key_bytes = Vec::new();
        assert!(matches!(
            adapter.get_into(&[0u8; 4], &mut invalid_key_bytes),
            Err(crate::DbError::KeyNotFound)
        ));

        let mut first_bytes = Vec::new();
        let mut first = RowWriter::new(&mut first_bytes, 1, 1 << 20, AdapterRow::NAME);
        assert!(adapter.first_into(&mut first).expect("first"));
        assert_eq!(&first_bytes[..4], &8u32.to_be_bytes());
        assert_eq!(&first_bytes[4..12], &key(1));
        let mut last_bytes = Vec::new();
        let mut last = RowWriter::new(&mut last_bytes, 1, 1 << 20, AdapterRow::NAME);
        assert!(adapter.last_into(&mut last).expect("last"));
        assert_eq!(&last_bytes[..4], &8u32.to_be_bytes());
        assert_eq!(&last_bytes[4..12], &key(5));

        // limit == 0 means everything.
        let mut all_bytes = Vec::new();
        let mut all = RowWriter::new(&mut all_bytes, 0, 1 << 20, AdapterRow::NAME);
        assert_eq!(
            adapter
                .range_into(Bound::Unbounded, Bound::Unbounded, &mut all)
                .expect("range"),
            5
        );
        let mut two_bytes = Vec::new();
        let mut two = RowWriter::new(&mut two_bytes, 2, 1 << 20, AdapterRow::NAME);
        assert_eq!(
            adapter
                .range_into(Bound::Unbounded, Bound::Unbounded, &mut two)
                .expect("range"),
            2
        );
        assert_eq!(&two_bytes[..4], &8u32.to_be_bytes());
        assert_eq!(&two_bytes[4..12], &key(1));
        let second = 4 + 8 + 4 + rapira::serialize(&AdapterRow { n: 1 }).len();
        assert_eq!(&two_bytes[second..second + 4], &8u32.to_be_bytes());
        assert_eq!(&two_bytes[second + 4..second + 12], &key(2));
        let mut key_bytes = Vec::new();
        let mut keys = RowWriter::new(&mut key_bytes, 3, 1 << 20, AdapterRow::NAME);
        assert_eq!(
            adapter
                .range_keys_into(Bound::Unbounded, Bound::Unbounded, &mut keys)
                .expect("range_keys"),
            3
        );
        assert_eq!(key_bytes.len(), 3 * (4 + 8));

        // Conditional write: insert-if-absent on a present key must fail, and
        // update-only on an absent key must fail — both without writing.
        let fresh = rapira::serialize(&AdapterRow { n: 42 });
        assert!(matches!(
            adapter.write(
                WriteKey::Provided(key(3).to_vec()),
                Some(false),
                fresh.clone()
            ),
            Err(crate::DbError::KeyExists)
        ));
        assert!(matches!(
            adapter.write(
                WriteKey::Provided(key(90).to_vec()),
                Some(true),
                fresh.clone()
            ),
            Err(crate::DbError::KeyNotFound)
        ));
        assert!(!tree.contains(&key(90)));
        adapter
            .write(WriteKey::Provided(key(90).to_vec()), None, fresh)
            .expect("unconditional write");
        assert!(tree.contains(&key(90)));

        // `take` returns the value and removes it; an absent key returns None
        // without deleting anything.
        let mut taken = Vec::new();
        assert!(
            adapter
                .take_into(&key(90), false, &mut taken)
                .expect("take")
        );
        assert_eq!(taken, rapira::serialize(&AdapterRow { n: 42 }));
        assert!(!tree.contains(&key(90)));
        let mut absent = Vec::new();
        assert!(
            !adapter
                .take_into(&key(90), false, &mut absent)
                .expect("take absent")
        );

        // `soft` is unimplemented on every family.
        assert!(matches!(
            adapter.remove(&key(1), true),
            Err(crate::DbError::NotImplemented)
        ));
        let mut soft_taken = Vec::new();
        assert!(matches!(
            adapter.take_into(&key(1), true, &mut soft_taken),
            Err(crate::DbError::NotImplemented)
        ));

        assert_eq!(adapter.count(false).expect("cheap count"), 5);
        assert_eq!(adapter.count(true).expect("exact count"), 5);

        // A batch with one bad item writes nothing.
        let before = tree.len();
        let err = adapter
            .apply_batch(vec![
                (
                    key(7).to_vec(),
                    Some(rapira::serialize(&AdapterRow { n: 7 })),
                ),
                (key(8).to_vec(), Some(vec![0xff, 0xfe])),
            ])
            .unwrap_err();
        assert!(matches!(
            err,
            crate::DbError::Client("invalid value bytes in batch")
        ));
        assert_eq!(tree.len(), before, "a failed batch must write nothing");

        adapter
            .apply_batch(vec![
                (
                    key(7).to_vec(),
                    Some(rapira::serialize(&AdapterRow { n: 7 })),
                ),
                (key(1).to_vec(), None),
            ])
            .expect("good batch");
        assert!(tree.contains(&key(7)));
        assert!(!tree.contains(&key(1)));
    }

    #[cfg(feature = "rpc")]
    #[test]
    fn write_insert_only_uses_single_held_section() {
        use armour_rpc::protocol::WriteKey;

        use crate::armour::handler::RpcHandler;

        let dir = tempfile::tempdir().expect("tmp");
        let db = Db::open_test(dir.path()).expect("db");
        let cfg = Config::balanced()
            .shard_count(2)
            .hints(true)
            .reversed(false)
            .build();
        let tree = db
            .open_typed_tree::<AdapterRow, RapiraCodec, _>(cfg, NoHook, &[])
            .expect("tree");
        tree.put(&key(1), AdapterRow { n: 1 }).expect("seed");

        let adapter = Adapter::new(
            &TreeMeta::of::<AdapterRow>(),
            tree.clone(),
            CodecWire(RapiraCodec),
            db.seq.clone(),
        );

        let fresh = rapira::serialize(&AdapterRow { n: 99 });
        assert!(matches!(
            adapter.write(
                WriteKey::Provided(key(1).to_vec()),
                Some(false),
                fresh.clone()
            ),
            Err(crate::DbError::KeyExists)
        ));
        assert!(matches!(
            adapter.write(
                WriteKey::Provided(key(2).to_vec()),
                Some(true),
                fresh.clone()
            ),
            Err(crate::DbError::KeyNotFound)
        ));
        let key_bytes = adapter
            .write(WriteKey::Provided(key(2).to_vec()), Some(false), fresh)
            .expect("insert-only on absent key");
        assert_eq!(key_bytes, key(2).to_vec());
        let mut bytes = Vec::new();
        assert!(adapter.get_into(&key(2), &mut bytes).expect("get"));
        assert_eq!(bytes, rapira::serialize(&AdapterRow { n: 99 }));
    }
}

#[cfg(all(test, feature = "var-collections", feature = "rapira-codec"))]
mod var_corruption_tests {
    use std::num::NonZeroU32;
    use std::ops::Bound;

    use armour_core::GetType;
    use rapira::Rapira;

    use crate::armour::inspection::InspectionError;
    use crate::armour::{CountMode, Db};
    use crate::{CollectionMeta, Config, NoHook, RapiraCodec};

    #[derive(Clone, Debug, PartialEq, Rapira, GetType)]
    struct CorruptVarTree {
        n: u64,
    }
    impl CollectionMeta for CorruptVarTree {
        type SelfId = [u8; 8];
        const NAME: &'static str = "adapter_corrupt_var_tree";
    }

    #[derive(Clone, Debug, PartialEq, Rapira, GetType)]
    struct CorruptVarMap {
        n: u64,
    }
    impl CollectionMeta for CorruptVarMap {
        type SelfId = [u8; 8];
        const NAME: &'static str = "adapter_corrupt_var_map";
    }

    fn key(id: u64) -> [u8; 8] {
        id.to_be_bytes()
    }

    fn iterable_config() -> Config {
        Config::balanced()
            .shard_count(2)
            .hints(true)
            .iterable(true)
            .reversed(false)
            .build()
    }

    /// `ordered_page` must FAIL on an undecodable row, not silently return a
    /// shorter page. And `CountMode::Exact` on a `VarTypedTree` counts live
    /// keys — the bad row is included, unlike the RPC `count(true)`.
    #[test]
    fn inspection_var_typed_tree_page_fails_and_exact_count_includes_bad_row() {
        let dir = tempfile::tempdir().expect("tmp");
        let db = Db::open_test(dir.path()).expect("db");
        let tree = db
            .open_var_typed_tree::<CorruptVarTree, RapiraCodec, _>(iterable_config(), NoHook, &[])
            .expect("var tree");
        tree.put(&key(1), &CorruptVarTree { n: 1 }).expect("seed");
        tree.put(&key(3), &CorruptVarTree { n: 3 }).expect("seed");
        // Readable as a live key, undecodable as the value type.
        tree.as_inner()
            .put(&key(2), &[0xff, 0xfe, 0xfd])
            .expect("poison");

        let inspector = db.inspector();
        let err = inspector
            .range(
                CorruptVarTree::NAME,
                Bound::Unbounded,
                Bound::Unbounded,
                None,
                NonZeroU32::new(10).unwrap(),
            )
            .unwrap_err();
        assert!(
            matches!(err, InspectionError::Database(_)),
            "the strict Inspector scan must surface the corrupt row: {err:?}"
        );

        assert_eq!(
            inspector
                .count(CorruptVarTree::NAME, CountMode::Exact)
                .expect("exact count"),
            3,
            "CountMode::Exact counts live keys, including the undecodable one"
        );
    }

    /// Same contract on an iterable `VarTypedMap`, which reaches the strict
    /// merge iterator through a different path than the Tree.
    #[test]
    fn inspection_var_typed_map_page_fails_on_undecodable_row() {
        let dir = tempfile::tempdir().expect("tmp");
        let db = Db::open_test(dir.path()).expect("db");
        let map = db
            .open_var_typed_map::<CorruptVarMap, RapiraCodec, _>(iterable_config(), NoHook, &[])
            .expect("var map");
        for id in [1u64, 3, 4] {
            map.put(&key(id), &CorruptVarMap { n: id }).expect("seed");
        }
        map.as_inner()
            .put(&key(2), &[0xff, 0xfe, 0xfd])
            .expect("poison");

        let err = db
            .inspector()
            .range(
                CorruptVarMap::NAME,
                Bound::Unbounded,
                Bound::Unbounded,
                None,
                NonZeroU32::new(10).unwrap(),
            )
            .unwrap_err();
        assert!(
            matches!(err, InspectionError::Database(_)),
            "the strict Inspector scan must surface the corrupt row: {err:?}"
        );
    }

    /// The RPC conditional-write precondition is **existence-only**: it never
    /// decodes the stored value. So an undecodable-but-present row answers
    /// `KeyExists` for insert-only and is left in place, while update-only sees
    /// a present row and replaces it. Neither arm surfaces `CorruptedEntry`.
    /// This mirrors `lsm_handler_write_{insert,update}_only_*` on the other
    /// family whose reads can fail to decode.
    #[cfg(feature = "rpc")]
    #[test]
    fn write_precondition_on_var_typed_tree_is_existence_only() {
        use armour_rpc::protocol::WriteKey;

        use crate::access::{Adapter, CodecWire};
        use crate::armour::handler::RpcHandler;
        use crate::tree_meta::TreeMeta;

        let dir = tempfile::tempdir().expect("tmp");
        let db = Db::open_test(dir.path()).expect("db");
        let tree = db
            .open_var_typed_tree::<CorruptVarTree, RapiraCodec, _>(iterable_config(), NoHook, &[])
            .expect("var tree");
        tree.as_inner()
            .put(&key(1), &[0xff, 0xfe, 0xfd])
            .expect("poison");

        let adapter = Adapter::new(
            &TreeMeta::of::<CorruptVarTree>(),
            tree.clone(),
            CodecWire(RapiraCodec),
            db.seq.clone(),
        );
        let fresh = rapira::serialize(&CorruptVarTree { n: 99 });

        let err = adapter
            .write(
                WriteKey::Provided(key(1).to_vec()),
                Some(false),
                fresh.clone(),
            )
            .unwrap_err();
        assert!(matches!(err, crate::DbError::KeyExists), "{err:?}");
        assert_eq!(
            tree.as_inner().get(&key(1)).map(|b| b.to_vec()),
            Some(vec![0xff, 0xfe, 0xfd]),
            "insert-only must not have overwritten the corrupt row"
        );

        adapter
            .write(WriteKey::Provided(key(1).to_vec()), Some(true), fresh)
            .expect("update-only over a present (if undecodable) row must succeed");
        assert_eq!(
            tree.get(&key(1)),
            Some(CorruptVarTree { n: 99 }),
            "update-only must have replaced the corrupt row"
        );
    }
}

#[cfg(all(test, feature = "fjall", feature = "rapira-codec"))]
mod lsm_corruption_tests {
    use std::num::NonZeroU32;
    use std::ops::Bound;

    use armour_core::GetType;
    use rapira::Rapira;

    use crate::armour::Db;
    use crate::armour::inspection::InspectionError;
    use crate::lsm::{LsmConfig, LsmDbConfig};
    use crate::{CollectionMeta, Key, NoHook, RapiraCodec};

    #[derive(Clone, Debug, PartialEq, Rapira, GetType)]
    struct CorruptLsm {
        n: u64,
    }
    impl CollectionMeta for CorruptLsm {
        type SelfId = [u8; 8];
        const NAME: &'static str = "adapter_corrupt_lsm";
    }

    fn key(id: u64) -> [u8; 8] {
        id.to_be_bytes()
    }

    #[test]
    fn inspection_lsm_page_fails_on_undecodable_row() {
        let dir = tempfile::tempdir().expect("tmp");
        let db = Db::open_test(dir.path()).expect("db");
        let tree = db
            .open_lsm_tree::<CorruptLsm, RapiraCodec, _>(
                Some(LsmDbConfig::default()),
                LsmConfig {
                    reversed: false,
                    ..LsmConfig::default()
                },
                NoHook,
                &[],
            )
            .expect("lsm tree");
        tree.put(&key(1), &CorruptLsm { n: 1 }).expect("seed");
        tree.put(&key(3), &CorruptLsm { n: 3 }).expect("seed");
        tree.keyspace()
            .insert(key(2).as_bytes(), b"corrupt")
            .expect("poison");

        let err = db
            .inspector()
            .range(
                CorruptLsm::NAME,
                Bound::Unbounded,
                Bound::Unbounded,
                None,
                NonZeroU32::new(10).unwrap(),
            )
            .unwrap_err();
        assert!(
            matches!(err, InspectionError::Database(_)),
            "the strict Inspector scan must surface the corrupt row: {err:?}"
        );
    }
}