armdb 0.14.2

use std::{
    collections::HashMap,
    io::{self, Read},
    net::{SocketAddr, TcpListener, TcpStream, ToSocketAddrs},
    path::{Path, PathBuf},
    sync::{
        Arc,
        atomic::{AtomicBool, AtomicUsize, Ordering},
        mpsc::{Receiver, RecvTimeoutError, SyncSender, sync_channel},
    },
    thread::{self, JoinHandle},
    time::{Duration, Instant},
};

#[cfg(unix)]
use std::os::unix::fs::FileTypeExt;
#[cfg(unix)]
use std::os::unix::net::{UnixListener, UnixStream};

use armour_rpc::RpcError;
use armour_rpc::protocol::{
    OpCode, Request, RequestPayload, read_bound, read_bytes, read_u8, read_u32_be, read_u64_be,
    read_write_key,
};

use super::db::Db;
use super::handler::{RowWriter, RpcHandler};
use crate::shutdown::ShutdownSignal;
use crate::{DbError, DbResult};

// ─── Response errors (server-side only) ──────────────────────────────────────

#[derive(Debug)]
enum Response {
    Err { code: u16, message: String },
}

// ─── Codec ────────────────────────────────────────────────────────────────────

fn decode_request(bytes: &[u8]) -> Result<Request, RpcError> {
    let mut pos = 0;
    let op_byte = read_u8(bytes, &mut pos)?;
    let op = OpCode::from_repr(op_byte)
        .ok_or_else(|| RpcError::Protocol("unknown opcode".to_string()))?;
    let hashname = read_u64_be(bytes, &mut pos)?;

    let payload = match op {
        OpCode::Get | OpCode::Contains => RequestPayload::Key(read_bytes(bytes, &mut pos)?),
        OpCode::First | OpCode::Last | OpCode::ListCollections | OpCode::GetSchema => {
            RequestPayload::Empty
        }
        OpCode::Count => RequestPayload::Count {
            exact: read_u8(bytes, &mut pos)? != 0,
        },
        OpCode::EntryLen => RequestPayload::EntryLen {
            key: read_bytes(bytes, &mut pos)?,
        },
        OpCode::Range | OpCode::RangeKeys => RequestPayload::Range {
            start: read_bound(bytes, &mut pos)?,
            end: read_bound(bytes, &mut pos)?,
            limit: read_u32_be(bytes, &mut pos)?,
        },
        OpCode::Write => {
            let key = read_write_key(bytes, &mut pos)?;
            let flag = match read_u8(bytes, &mut pos)? {
                0 => None,
                1 => Some(true),
                2 => Some(false),
                _ => return Err(RpcError::Protocol("invalid write flag".to_string())),
            };
            RequestPayload::Write {
                key,
                flag,
                value: read_bytes(bytes, &mut pos)?,
            }
        }
        OpCode::Remove => RequestPayload::Remove {
            key: read_bytes(bytes, &mut pos)?,
            soft: read_u8(bytes, &mut pos)? != 0,
        },
        OpCode::Take => RequestPayload::Take {
            key: read_bytes(bytes, &mut pos)?,
            soft: read_u8(bytes, &mut pos)? != 0,
        },
        OpCode::ApplyBatch => {
            let count = read_u32_be(bytes, &mut pos)? as usize;
            const MAX_PREALLOC: usize = 64 * 1024;
            let mut items = Vec::with_capacity(count.min(MAX_PREALLOC));
            for _ in 0..count {
                let key = read_bytes(bytes, &mut pos)?;
                let value = match read_u8(bytes, &mut pos)? {
                    0 => None,
                    1 => Some(read_bytes(bytes, &mut pos)?),
                    _ => return Err(RpcError::Protocol("invalid batch value tag".to_string())),
                };
                items.push((key, value));
            }
            RequestPayload::Batch(items)
        }
    };

    Ok(Request {
        op,
        hashname,
        payload,
    })
}

/// Encode an error frame: `[0x01][code: u16 BE][len: u16 BE][msg bytes]`.
///
/// The length prefix is a `u16`, and the whole body must fit `max_frame`.
/// Truncate on a UTF-8 boundary so a client can decode the advertised bytes
/// without leaving a trailing fragment in the frame.
fn encode_err_frame(buf: &mut Vec<u8>, code: u16, message: &str, max_frame: usize) {
    const ERROR_HEADER_BYTES: usize = 5;
    buf.push(0x01);
    buf.extend_from_slice(&code.to_be_bytes());
    let max = message
        .len()
        .min(u16::MAX as usize)
        .min(max_frame.saturating_sub(ERROR_HEADER_BYTES));
    let end = message.floor_char_boundary(max);
    let msg = &message.as_bytes()[..end];
    buf.extend_from_slice(&(msg.len() as u16).to_be_bytes());
    buf.extend_from_slice(msg);
}

/// Build a wire error [`Response`] applying the **5xx-opaque policy**.
///
/// - **4xx** (client error): the message is forwarded verbatim — the client must
///   understand what it got wrong so it can fix the request (unknown collection,
///   invalid op, bad key, schema mismatch, …).
/// - **5xx** (server error): the real cause is logged server-side via `tracing`,
///   and the wire message becomes a generic `"internal error"` so internal
///   details (file paths, corruption state, codec internals) don't leak to RPC
///   clients. `original` is the verbatim cause to log.
///
/// **501 is exempt.** [`DbError::NotImplemented`] represents a non-sensitive capability
/// mismatch: either the requested mode is unsupported or the collection was
/// opened without the required capability. Its Display text is fixed and safe to
/// forward, and clients need the distinction from a transient internal failure.
/// It is therefore forwarded verbatim and logged at warn.
///
/// Note: `ShardMismatch` is a server-side routing bug (the engine mis-routed a
/// key), so it stays in the `_ => 500` catch-all of [`DbError::status_code`] and
/// is therefore opaque to the client.
fn err_response(code: u16, original: &str) -> Response {
    if code == 501 {
        tracing::warn!("RPC unsupported request ({code}): {original}");
        Response::Err {
            code,
            message: original.into(),
        }
    } else if code >= 500 {
        tracing::error!("RPC server error ({code}): {original}");
        Response::Err {
            code,
            message: "internal error".into(),
        }
    } else {
        Response::Err {
            code,
            message: original.into(),
        }
    }
}

#[cfg(test)]
#[path = "wire_golden.rs"]
pub(crate) mod wire_golden;

#[cfg(test)]
mod dispatch_into_tests {
    use std::collections::HashMap;
    use std::io::{self, Cursor, Read, Write};
    use std::ops::{Bound, ControlFlow};
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::sync::{Arc, LazyLock, Mutex};
    use std::thread;
    use std::time::{Duration, Instant};

    use armour_rpc::codec::encode_request;
    use armour_rpc::protocol::{OpCode, Request, RequestPayload, WriteKey};

    use crate::armour::handler::{KeyBytes, RowWriter, RpcHandler, ValueBytes};
    use crate::{DbError, DbResult};

    use super::{ListenerState, RpcConfig, TreeMap, dispatch_into, serve_connection, wire_golden};

    const STUB_HASH: u64 = 0xABCD;
    static UTF8_CLIENT_ERROR: LazyLock<String> = LazyLock::new(|| "é".repeat(40_000));

    struct StubHandler {
        large: bool,
        client_error: Option<&'static str>,
        reported_range_count: Option<u32>,
        found_without_row: bool,
        get_delay: Duration,
    }

    impl RpcHandler for StubHandler {
        fn name(&self) -> &str {
            "stub"
        }

        fn info(&self) -> (u64, u16) {
            (0, 1)
        }

        fn schema(&self) -> armour_rpc::SchemaResponse {
            armour_rpc::SchemaResponse {
                name: self.name().into(),
                version: 1,
                typ: <u8 as armour_core::GetType>::TYPE,
                key_scheme: armour_core::KeyScheme::Bytes,
            }
        }

        fn get_into(&self, key: &[u8], out: &mut Vec<u8>) -> DbResult<bool> {
            if !self.get_delay.is_zero() {
                thread::sleep(self.get_delay);
            }
            if let Some(message) = self.client_error {
                return Err(DbError::Client(message));
            }
            if key.is_empty() {
                return Ok(false);
            }
            out.extend_from_slice(&[1, 2]);
            Ok(true)
        }

        fn contains(&self, key: &[u8]) -> DbResult<bool> {
            Ok(!key.is_empty())
        }

        fn entry_len(&self, key: &[u8]) -> DbResult<Option<u32>> {
            Ok((!key.is_empty()).then_some(5))
        }

        fn first_into(&self, writer: &mut RowWriter<'_>) -> DbResult<bool> {
            if self.found_without_row {
                return Ok(true);
            }
            let _ = writer.row(&[1], |out| {
                out.push(2);
                Ok(())
            })?;
            Ok(true)
        }

        fn last_into(&self, _: &mut RowWriter<'_>) -> DbResult<bool> {
            Ok(false)
        }

        fn range_into(
            &self,
            start: Bound<KeyBytes>,
            _: Bound<KeyBytes>,
            writer: &mut RowWriter<'_>,
        ) -> DbResult<u32> {
            if matches!(start, Bound::Included(ref key) if key.is_empty()) {
                return Ok(0);
            }
            let value = if self.large { vec![9; 128] } else { vec![2] };
            if matches!(
                writer.row(&[1], |buf| {
                    buf.extend_from_slice(&value);
                    Ok(())
                })?,
                ControlFlow::Break(())
            ) {
                return Ok(writer.rows());
            }
            if !self.large {
                let _ = writer.row(&[3], |buf| {
                    buf.extend_from_slice(&[4, 5]);
                    Ok(())
                })?;
            }
            Ok(self.reported_range_count.unwrap_or_else(|| writer.rows()))
        }

        fn range_keys_into(
            &self,
            _: Bound<KeyBytes>,
            _: Bound<KeyBytes>,
            writer: &mut RowWriter<'_>,
        ) -> DbResult<u32> {
            let _ = writer.key_only(&[1])?;
            let _ = writer.key_only(&[2, 3])?;
            Ok(writer.rows())
        }

        fn write(&self, _: WriteKey, _: Option<bool>, _: ValueBytes) -> DbResult<ValueBytes> {
            Ok(vec![7, 7])
        }

        fn remove(&self, _: &[u8], _: bool) -> DbResult<()> {
            Ok(())
        }

        fn take_into(&self, key: &[u8], _: bool, out: &mut Vec<u8>) -> DbResult<bool> {
            if key.is_empty() {
                return Ok(false);
            }
            out.extend_from_slice(&[1, 2]);
            Ok(true)
        }

        fn count(&self, _: bool) -> DbResult<u64> {
            Ok(u64::MAX)
        }

        fn apply_batch(&self, _: Vec<(KeyBytes, Option<ValueBytes>)>) -> DbResult<()> {
            Ok(())
        }
    }

    fn stub_trees(large: bool) -> TreeMap {
        let mut handlers: HashMap<u64, Arc<dyn RpcHandler>> = HashMap::new();
        handlers.insert(
            STUB_HASH,
            Arc::new(StubHandler {
                large,
                client_error: None,
                reported_range_count: None,
                found_without_row: false,
                get_delay: Duration::ZERO,
            }),
        );
        Arc::new(handlers)
    }

    fn stub_trees_with_client_error(message: &'static str) -> TreeMap {
        let mut handlers: HashMap<u64, Arc<dyn RpcHandler>> = HashMap::new();
        handlers.insert(
            STUB_HASH,
            Arc::new(StubHandler {
                large: false,
                client_error: Some(message),
                reported_range_count: None,
                found_without_row: false,
                get_delay: Duration::ZERO,
            }),
        );
        Arc::new(handlers)
    }

    fn stub_trees_with_bad_row_contract(
        reported_range_count: Option<u32>,
        found_without_row: bool,
    ) -> TreeMap {
        let mut handlers: HashMap<u64, Arc<dyn RpcHandler>> = HashMap::new();
        handlers.insert(
            STUB_HASH,
            Arc::new(StubHandler {
                large: false,
                client_error: None,
                reported_range_count,
                found_without_row,
                get_delay: Duration::ZERO,
            }),
        );
        Arc::new(handlers)
    }

    fn stub_trees_with_get_delay(get_delay: Duration) -> TreeMap {
        let mut handlers: HashMap<u64, Arc<dyn RpcHandler>> = HashMap::new();
        handlers.insert(
            STUB_HASH,
            Arc::new(StubHandler {
                large: false,
                client_error: None,
                reported_range_count: None,
                found_without_row: false,
                get_delay,
            }),
        );
        Arc::new(handlers)
    }

    fn range_request() -> Vec<u8> {
        let mut request = Vec::new();
        encode_request(
            Request {
                op: OpCode::Range,
                hashname: STUB_HASH,
                payload: RequestPayload::Range {
                    start: Bound::Unbounded,
                    end: Bound::Unbounded,
                    limit: 0,
                },
            },
            &mut request,
        );
        request
    }

    fn get_request() -> Vec<u8> {
        let mut request = Vec::new();
        encode_request(
            Request {
                op: OpCode::Get,
                hashname: STUB_HASH,
                payload: RequestPayload::Key(vec![1]),
            },
            &mut request,
        );
        request
    }

    struct RecordingStream {
        input: Cursor<Vec<u8>>,
        output: Arc<Mutex<Vec<u8>>>,
        timeout: Arc<Mutex<Option<Duration>>>,
    }

    impl Read for RecordingStream {
        fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
            self.input.read(buf)
        }
    }

    impl Write for RecordingStream {
        fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
            self.output
                .lock()
                .expect("output lock")
                .extend_from_slice(buf);
            Ok(buf.len())
        }

        fn flush(&mut self) -> io::Result<()> {
            Ok(())
        }
    }

    impl armour_rpc::DeadlineWriter for RecordingStream {
        fn set_write_timeout(&self, timeout: Option<Duration>) -> io::Result<()> {
            *self.timeout.lock().expect("timeout lock") = timeout;
            Ok(())
        }
    }

    struct BlockingStream {
        input: Cursor<Vec<u8>>,
        timeout: Arc<Mutex<Option<Duration>>>,
        write_started: Arc<AtomicBool>,
    }

    impl Read for BlockingStream {
        fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
            self.input.read(buf)
        }
    }

    impl Write for BlockingStream {
        fn write(&mut self, _: &[u8]) -> io::Result<usize> {
            self.write_started.store(true, Ordering::Release);
            thread::sleep(
                self.timeout
                    .lock()
                    .expect("timeout lock")
                    .expect("write timeout"),
            );
            Err(io::Error::new(io::ErrorKind::TimedOut, "blocked peer"))
        }

        fn flush(&mut self) -> io::Result<()> {
            Ok(())
        }
    }

    impl armour_rpc::DeadlineWriter for BlockingStream {
        fn set_write_timeout(&self, timeout: Option<Duration>) -> io::Result<()> {
            *self.timeout.lock().expect("timeout lock") = timeout;
            Ok(())
        }
    }

    fn framed_request() -> Vec<u8> {
        let request = get_request();
        let mut framed = (request.len() as u32).to_be_bytes().to_vec();
        framed.extend_from_slice(&request);
        framed
    }

    #[test]
    fn slow_handler_does_not_consume_the_response_write_deadline() {
        let output = Arc::new(Mutex::new(Vec::new()));
        let state = Arc::new(ListenerState {
            stop: AtomicBool::new(false),
            live: std::sync::atomic::AtomicUsize::new(0),
            cfg: RpcConfig::builder()
                .write_deadline(Duration::from_millis(10))
                .read_timeout(Duration::from_millis(1))
                .build(),
            reaped: std::sync::atomic::AtomicUsize::new(0),
        });
        serve_connection(
            RecordingStream {
                input: Cursor::new(framed_request()),
                output: Arc::clone(&output),
                timeout: Arc::new(Mutex::new(None)),
            },
            state,
            stub_trees_with_get_delay(Duration::from_millis(30)),
        );

        assert!(
            !output.lock().expect("output lock").is_empty(),
            "a completed handler must receive a fresh write deadline"
        );
    }

    #[test]
    fn local_stop_interrupts_a_blocked_response_write_at_read_timeout() {
        let timeout = Arc::new(Mutex::new(None));
        let write_started = Arc::new(AtomicBool::new(false));
        let state = Arc::new(ListenerState {
            stop: AtomicBool::new(false),
            live: std::sync::atomic::AtomicUsize::new(0),
            cfg: RpcConfig::builder()
                .write_deadline(Duration::from_secs(2))
                .read_timeout(Duration::from_millis(10))
                .build(),
            reaped: std::sync::atomic::AtomicUsize::new(0),
        });
        let worker_state = Arc::clone(&state);
        let worker = thread::spawn({
            let timeout = Arc::clone(&timeout);
            let write_started = Arc::clone(&write_started);
            move || {
                serve_connection(
                    BlockingStream {
                        input: Cursor::new(framed_request()),
                        timeout,
                        write_started,
                    },
                    worker_state,
                    stub_trees(false),
                );
            }
        });

        let wait_started = Instant::now();
        while !write_started.load(Ordering::Acquire) {
            assert!(
                wait_started.elapsed() < Duration::from_secs(1),
                "write never started"
            );
            thread::yield_now();
        }
        let stopped = Instant::now();
        state.stop.store(true, Ordering::Release);
        worker.join().expect("connection worker");
        assert!(
            stopped.elapsed() < Duration::from_millis(250),
            "blocked write delayed stop for {:?}",
            stopped.elapsed()
        );
    }

    #[test]
    fn range_count_is_taken_from_rows_written_not_handler_report() {
        let mut out = Vec::new();
        dispatch_into(
            &stub_trees_with_bad_row_contract(Some(99), false),
            &range_request(),
            &mut out,
            1 << 20,
        );

        assert_eq!(
            u32::from_be_bytes(out[1..5].try_into().expect("range count")),
            2,
            "the frame count must match the rows actually written"
        );
    }

    #[test]
    fn optional_row_requires_one_row_when_handler_reports_found() {
        let mut request = Vec::new();
        encode_request(
            Request {
                op: OpCode::First,
                hashname: STUB_HASH,
                payload: RequestPayload::Empty,
            },
            &mut request,
        );
        let mut out = Vec::new();
        dispatch_into(
            &stub_trees_with_bad_row_contract(None, true),
            &request,
            &mut out,
            1 << 20,
        );

        assert!(
            matches!(
                armour_rpc::codec::decode_response(&out),
                Ok(armour_rpc::Response::Err { code: 500, .. })
            ),
            "a found row without bytes must be a server error"
        );
    }

    #[test]
    fn dispatch_reproduces_the_golden_response_bytes() {
        let mut out = vec![0, 0, 0, 0];
        dispatch_into(&stub_trees(false), &range_request(), &mut out, 1 << 20);

        let expected = wire_golden::RESPONSE_VECTORS
            .iter()
            .find(|(name, _)| *name == "kv_two")
            .map(|(_, bytes)| bytes)
            .expect("Task 1 kv_two golden response");
        assert_eq!(&out[4..], expected);
    }

    #[test]
    fn oversize_response_becomes_413_on_a_clean_buffer() {
        let mut out = vec![0, 0, 0, 0];
        dispatch_into(&stub_trees(true), &range_request(), &mut out, 64);

        assert_eq!(&out[..4], &[0, 0, 0, 0]);
        assert_eq!(out[4], 0x01, "expected an error frame");
        assert_eq!(u16::from_be_bytes([out[5], out[6]]), 413);
    }

    #[test]
    fn dispatch_counts_only_the_response_body_against_max_frame() {
        let expected = wire_golden::RESPONSE_VECTORS
            .iter()
            .find(|(name, _)| *name == "kv_two")
            .map(|(_, bytes)| bytes)
            .expect("Task 1 kv_two golden response");
        let mut out = vec![0, 0, 0, 0];

        dispatch_into(
            &stub_trees(false),
            &range_request(),
            &mut out,
            expected.len(),
        );

        assert_eq!(&out[..4], &[0, 0, 0, 0], "real transport prefix remains");
        assert_eq!(&out[4..], expected, "body exactly fills max_frame");
    }

    #[test]
    fn dispatch_writes_schema_as_a_raw_json_payload() {
        let mut body = Vec::new();
        encode_request(
            Request {
                op: OpCode::GetSchema,
                hashname: STUB_HASH,
                payload: RequestPayload::Empty,
            },
            &mut body,
        );
        let mut out = Vec::new();
        dispatch_into(&stub_trees(false), &body, &mut out, 1 << 20);

        let expected = wire_golden::RESPONSE_VECTORS
            .iter()
            .find(|(name, _)| *name == "schema_bytes")
            .map(|(_, bytes)| bytes)
            .expect("Task 1 schema response vector");
        assert_eq!(out, *expected);
    }

    #[test]
    fn dispatch_reproduces_frozen_client_error_frames() {
        let mut body = Vec::new();
        encode_request(
            Request {
                op: OpCode::Get,
                hashname: STUB_HASH,
                payload: RequestPayload::Key(vec![1]),
            },
            &mut body,
        );

        for (message, golden_name) in [
            ("bad", "err_400"),
            (UTF8_CLIENT_ERROR.as_str(), "err_utf8_truncated"),
        ] {
            let mut actual = vec![0, 0, 0, 0];
            dispatch_into(
                &stub_trees_with_client_error(message),
                &body,
                &mut actual,
                1 << 20,
            );
            let expected = wire_golden::RESPONSE_VECTORS
                .iter()
                .find(|(name, _)| *name == golden_name)
                .map(|(_, bytes)| bytes)
                .expect("Task 1 error response vector");
            assert_eq!(&actual[..4], &[0, 0, 0, 0]);
            assert_eq!(&actual[4..], expected, "{golden_name}");
        }
    }

    #[test]
    fn dispatch_reproduces_frozen_vectors_for_every_handler_payload_shape() {
        let cases = vec![
            (
                "empty",
                Request {
                    op: OpCode::Remove,
                    hashname: STUB_HASH,
                    payload: RequestPayload::Remove {
                        key: vec![],
                        soft: false,
                    },
                },
            ),
            (
                "bool_true",
                Request {
                    op: OpCode::Contains,
                    hashname: STUB_HASH,
                    payload: RequestPayload::Key(vec![1]),
                },
            ),
            (
                "bool_false",
                Request {
                    op: OpCode::Contains,
                    hashname: STUB_HASH,
                    payload: RequestPayload::Key(vec![]),
                },
            ),
            (
                "opt_data_none",
                Request {
                    op: OpCode::Get,
                    hashname: STUB_HASH,
                    payload: RequestPayload::Key(vec![]),
                },
            ),
            (
                "opt_data_some",
                Request {
                    op: OpCode::Get,
                    hashname: STUB_HASH,
                    payload: RequestPayload::Key(vec![1]),
                },
            ),
            (
                "opt_len_none",
                Request {
                    op: OpCode::EntryLen,
                    hashname: STUB_HASH,
                    payload: RequestPayload::EntryLen { key: vec![] },
                },
            ),
            (
                "opt_len_some",
                Request {
                    op: OpCode::EntryLen,
                    hashname: STUB_HASH,
                    payload: RequestPayload::EntryLen { key: vec![1] },
                },
            ),
            (
                "opt_kv_some",
                Request {
                    op: OpCode::First,
                    hashname: STUB_HASH,
                    payload: RequestPayload::Empty,
                },
            ),
            (
                "opt_kv_none",
                Request {
                    op: OpCode::Last,
                    hashname: STUB_HASH,
                    payload: RequestPayload::Empty,
                },
            ),
            (
                "kv_empty",
                Request {
                    op: OpCode::Range,
                    hashname: STUB_HASH,
                    payload: RequestPayload::Range {
                        start: Bound::Included(vec![]),
                        end: Bound::Unbounded,
                        limit: 0,
                    },
                },
            ),
            (
                "kv_two",
                Request {
                    op: OpCode::Range,
                    hashname: STUB_HASH,
                    payload: RequestPayload::Range {
                        start: Bound::Unbounded,
                        end: Bound::Unbounded,
                        limit: 0,
                    },
                },
            ),
            (
                "keys_two",
                Request {
                    op: OpCode::RangeKeys,
                    hashname: STUB_HASH,
                    payload: RequestPayload::Range {
                        start: Bound::Unbounded,
                        end: Bound::Unbounded,
                        limit: 0,
                    },
                },
            ),
            (
                "key",
                Request {
                    op: OpCode::Write,
                    hashname: STUB_HASH,
                    payload: RequestPayload::Write {
                        key: WriteKey::Provided(vec![1]),
                        flag: None,
                        value: vec![],
                    },
                },
            ),
            (
                "count",
                Request {
                    op: OpCode::Count,
                    hashname: STUB_HASH,
                    payload: RequestPayload::Count { exact: false },
                },
            ),
            (
                "opt_data_none",
                Request {
                    op: OpCode::Take,
                    hashname: STUB_HASH,
                    payload: RequestPayload::Take {
                        key: vec![],
                        soft: false,
                    },
                },
            ),
            (
                "opt_data_some",
                Request {
                    op: OpCode::Take,
                    hashname: STUB_HASH,
                    payload: RequestPayload::Take {
                        key: vec![1],
                        soft: false,
                    },
                },
            ),
            (
                "empty",
                Request {
                    op: OpCode::ApplyBatch,
                    hashname: STUB_HASH,
                    payload: RequestPayload::Batch(vec![]),
                },
            ),
        ];

        for (golden_name, request) in cases {
            let mut body = Vec::new();
            encode_request(request, &mut body);
            let mut out = Vec::new();
            dispatch_into(&stub_trees(false), &body, &mut out, 1 << 20);
            let expected = wire_golden::RESPONSE_VECTORS
                .iter()
                .find(|(name, _)| *name == golden_name)
                .map(|(_, bytes)| bytes)
                .expect("Task 1 response vector");
            assert_eq!(out, *expected, "{golden_name}");
        }
    }
}

// ─── Server ───────────────────────────────────────────────────────────────────

pub type TreeMap = Arc<HashMap<u64, Arc<dyn RpcHandler>>>;

const ACCEPT_BACKOFF_MIN: Duration = Duration::from_millis(5);
const ACCEPT_BACKOFF_MAX: Duration = Duration::from_millis(500);

struct AcceptBackoff {
    next: Duration,
}

impl AcceptBackoff {
    fn new() -> Self {
        Self {
            next: ACCEPT_BACKOFF_MIN,
        }
    }

    fn on_error(&mut self) -> Duration {
        let delay = self.next;
        self.next = self.next.saturating_mul(2).min(ACCEPT_BACKOFF_MAX);
        delay
    }

    fn on_accept(&mut self) {
        self.next = ACCEPT_BACKOFF_MIN;
    }
}

/// Blocking listener settings. All durations are non-zero because each one is
/// also a polling or deadline boundary in the thread-per-connection server.
#[derive(Clone, Debug, bon::Builder)]
pub struct RpcConfig {
    #[builder(default = 256)]
    pub max_connections: usize,
    #[builder(default = 64 * 1024 * 1024)]
    pub max_frame: usize,
    #[builder(default = 64 * 1024)]
    pub idle_capacity: usize,
    #[builder(default = Duration::from_secs(30))]
    pub write_deadline: Duration,
    #[builder(default = Duration::from_secs(30))]
    pub read_deadline: Duration,
    #[builder(default = Duration::from_millis(250))]
    pub read_timeout: Duration,
    #[builder(default = Duration::from_millis(50))]
    pub poll_interval: Duration,
}

impl Default for RpcConfig {
    fn default() -> Self {
        Self::builder().build()
    }
}

impl RpcConfig {
    /// Check bounds that cannot be expressed in the builder type itself.
    pub fn validate(&self) -> DbResult<()> {
        let bad = |message: &'static str| Err(DbError::Client(message));
        if self.max_connections == 0 {
            return bad("max_connections must be > 0");
        }
        if !(5..=u32::MAX as usize).contains(&self.max_frame) {
            return bad("max_frame must be in 5..=u32::MAX");
        }
        if self.idle_capacity > self.max_frame {
            return bad("idle_capacity exceeds max_frame");
        }
        for duration in [
            self.write_deadline,
            self.read_deadline,
            self.read_timeout,
            self.poll_interval,
        ] {
            if duration.is_zero() {
                return bad("durations must be non-zero");
            }
        }
        Ok(())
    }
}

/// Shared by a listener and every connection it accepted. A stopped handle
/// therefore reaches connection read loops as well as the accept loop.
struct ListenerState {
    stop: AtomicBool,
    live: AtomicUsize,
    cfg: RpcConfig,
    #[cfg(test)]
    reaped: AtomicUsize,
}

/// Keeps the admission count correct even if a connection thread panics.
struct ConnGuard(Arc<ListenerState>);

impl Drop for ConnGuard {
    fn drop(&mut self) {
        self.0.live.fetch_sub(1, Ordering::Release);
    }
}

enum Acceptor {
    Tcp(TcpListener),
    #[cfg(unix)]
    Uds {
        listener: UnixListener,
        path: PathBuf,
    },
}

fn admit(state: &ListenerState) -> bool {
    let mut live = state.live.load(Ordering::Acquire);
    loop {
        if live >= state.cfg.max_connections {
            return false;
        }
        match state
            .live
            .compare_exchange_weak(live, live + 1, Ordering::AcqRel, Ordering::Acquire)
        {
            Ok(_) => return true,
            Err(current) => live = current,
        }
    }
}

fn configure_tcp(stream: &TcpStream, cfg: &RpcConfig) -> io::Result<()> {
    // Accepted sockets can inherit the listener's non-blocking setting.
    stream.set_nonblocking(false)?;
    stream.set_nodelay(true)?;
    stream.set_read_timeout(Some(cfg.read_timeout))
}

#[cfg(unix)]
fn configure_uds(stream: &UnixStream, cfg: &RpcConfig) -> io::Result<()> {
    stream.set_nonblocking(false)?;
    stream.set_read_timeout(Some(cfg.read_timeout))
}

trait RejectorSocket: armour_rpc::DeadlineWriter {
    fn configure_rejector(&self, cfg: &RpcConfig) -> io::Result<()>;
    fn clone_for_rejection(&self) -> io::Result<Self>
    where
        Self: Sized;
}

impl RejectorSocket for TcpStream {
    fn configure_rejector(&self, cfg: &RpcConfig) -> io::Result<()> {
        self.set_nonblocking(false)?;
        self.set_write_timeout(Some(cfg.read_timeout))
    }

    fn clone_for_rejection(&self) -> io::Result<Self> {
        self.try_clone()
    }
}

#[cfg(unix)]
impl RejectorSocket for UnixStream {
    fn configure_rejector(&self, cfg: &RpcConfig) -> io::Result<()> {
        self.set_nonblocking(false)?;
        self.set_write_timeout(Some(cfg.read_timeout))
    }

    fn clone_for_rejection(&self) -> io::Result<Self> {
        self.try_clone()
    }
}

fn rejector_loop<S: RejectorSocket>(rx: Receiver<S>, state: Arc<ListenerState>) {
    let mut buf = Vec::new();
    loop {
        if state.stop.load(Ordering::Acquire) {
            return;
        }
        let mut stream = match rx.recv_timeout(state.cfg.read_timeout) {
            Ok(stream) => stream,
            Err(RecvTimeoutError::Timeout) => continue,
            Err(RecvTimeoutError::Disconnected) => return,
        };
        if let Err(error) = stream.configure_rejector(&state.cfg) {
            tracing::debug!("rpc: rejection socket setup failed: {error}");
            continue;
        }
        let deadline = Instant::now() + state.cfg.write_deadline;
        let _ = armour_rpc::write_frame_with_timeout(
            &mut stream,
            &mut buf,
            deadline,
            Some(state.cfg.read_timeout),
            || state.stop.load(Ordering::Acquire),
            |body| {
                encode_err_frame(body, 503, "too many connections", state.cfg.max_frame);
                Ok(())
            },
        );
    }
}

fn serve_connection<S: Read + armour_rpc::DeadlineWriter>(
    mut stream: S,
    state: Arc<ListenerState>,
    trees: TreeMap,
) {
    if state.stop.load(Ordering::Acquire) {
        return;
    }

    let mut reader = armour_rpc::FrameReader::new(state.cfg.max_frame);
    let mut out = Vec::with_capacity(state.cfg.idle_capacity);
    let mut frame_started = None;

    loop {
        if state.stop.load(Ordering::Acquire) {
            return;
        }
        match reader.poll_frame(&mut stream) {
            Ok(armour_rpc::FramePoll::Ready(body)) => {
                frame_started = None;
                out.clear();
                out.extend_from_slice(&[0; 4]);
                dispatch_into(&trees, body, &mut out, state.cfg.max_frame);
                let body_len = out.len() - 4;
                out[..4].copy_from_slice(
                    &u32::try_from(body_len)
                        .expect("response body must fit the wire length")
                        .to_be_bytes(),
                );
                // The transport deadline is absolute on Linux; XNU can re-arm
                // its socket timeout while a frame is trickled, so macOS is
                // necessarily best-effort (see the transport contract).
                let deadline = Instant::now() + state.cfg.write_deadline;
                if armour_rpc::write_prepared_frame_with_timeout(
                    &mut stream,
                    &out,
                    deadline,
                    Some(state.cfg.read_timeout),
                    || state.stop.load(Ordering::Acquire),
                )
                .is_err()
                {
                    return;
                }
                reader.shrink(state.cfg.idle_capacity);
                reset_response_buffer(&mut out, state.cfg.idle_capacity);
            }
            Ok(armour_rpc::FramePoll::Pending { in_frame }) => {
                if in_frame {
                    let started = *frame_started.get_or_insert_with(Instant::now);
                    if started.elapsed() >= state.cfg.read_deadline {
                        return;
                    }
                } else {
                    frame_started = None;
                }
            }
            Ok(armour_rpc::FramePoll::Closed) => return,
            Err(RpcError::FrameTooLarge { .. }) => {
                out.clear();
                out.extend_from_slice(&[0; 4]);
                encode_err_frame(
                    &mut out,
                    400,
                    "frame exceeds the server limit",
                    state.cfg.max_frame,
                );
                let body_len = out.len() - 4;
                out[..4].copy_from_slice(
                    &u32::try_from(body_len)
                        .expect("response body must fit the wire length")
                        .to_be_bytes(),
                );
                let deadline = Instant::now() + state.cfg.write_deadline;
                let _ = armour_rpc::write_prepared_frame_with_timeout(
                    &mut stream,
                    &out,
                    deadline,
                    Some(state.cfg.read_timeout),
                    || state.stop.load(Ordering::Acquire),
                );
                return;
            }
            Err(_) => return,
        }
    }
}

fn accept_loop<S, L, A, C>(
    listener: L,
    state: Arc<ListenerState>,
    shutdown: ShutdownSignal,
    trees: TreeMap,
    rejector: SyncSender<S>,
    accept: A,
    configure: C,
) where
    S: Read + RejectorSocket + Send + 'static,
    A: Fn(&L) -> io::Result<S>,
    C: Fn(&S, &RpcConfig) -> io::Result<()>,
{
    let mut handles: Vec<JoinHandle<()>> = Vec::new();
    let mut backoff = AcceptBackoff::new();
    loop {
        if state.stop.load(Ordering::Acquire) || shutdown.is_shutdown() {
            break;
        }
        reap_finished_connection_workers(&state, &mut handles);

        match accept(&listener) {
            Ok(stream) if !admit(&state) => {
                backoff.on_accept();
                let _ = rejector.try_send(stream);
            }
            Ok(stream) => {
                backoff.on_accept();
                let guard = ConnGuard(Arc::clone(&state));
                if let Err(error) = configure(&stream, &state.cfg) {
                    tracing::error!("rpc: socket setup failed: {error}");
                    drop(guard);
                    continue;
                }
                let rejected = match stream.clone_for_rejection() {
                    Ok(stream) => Some(stream),
                    Err(error) => {
                        tracing::debug!("rpc: could not retain a rejection socket: {error}");
                        None
                    }
                };
                let state = Arc::clone(&state);
                let trees = Arc::clone(&trees);
                match thread::Builder::new()
                    .name("armdb-rpc-conn".into())
                    .stack_size(256 * 1024)
                    .spawn(move || {
                        let _guard = guard;
                        if std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                            serve_connection(stream, state, trees);
                        }))
                        .is_err()
                        {
                            tracing::error!("rpc: connection worker panicked while serving a peer");
                        }
                    }) {
                    Ok(handle) => handles.push(handle),
                    Err(error) => {
                        tracing::error!("rpc: connection thread spawn failed: {error}");
                        if let Some(stream) = rejected
                            && rejector.try_send(stream).is_err()
                        {
                            tracing::debug!(
                                "rpc: rejection queue full after connection spawn failure"
                            );
                        }
                    }
                }
            }
            Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
                thread::sleep(state.cfg.poll_interval);
            }
            Err(error) => {
                tracing::error!("rpc: accept error: {error}");
                thread::sleep(backoff.on_error());
            }
        }
    }
    drop(rejector);
    for handle in handles {
        let _ = handle.join();
    }
}

fn reap_finished_connection_workers(state: &ListenerState, handles: &mut Vec<JoinHandle<()>>) {
    #[cfg(not(test))]
    let _ = state;
    #[cfg(test)]
    let mut reaped = 0;
    let mut running = Vec::with_capacity(handles.len());
    for handle in handles.drain(..) {
        if handle.is_finished() {
            let _ = handle.join();
            #[cfg(test)]
            {
                reaped += 1;
            }
        } else {
            running.push(handle);
        }
    }
    *handles = running;
    #[cfg(test)]
    state.reaped.fetch_add(reaped, Ordering::Relaxed);
}

/// Appends one response body to `buf`, which already contains the transport
/// frame's length prefix. Errors discard every partial response byte, but never
/// the caller-owned prefix.
#[cfg_attr(feature = "ctx-gates", doc(hidden))]
#[cfg_attr(
    feature = "ctx-gates",
    doc = "Allocation-gate entrypoint for direct response dispatch."
)]
#[cfg_attr(feature = "ctx-gates", allow(clippy::missing_panics_doc))]
#[cfg(feature = "ctx-gates")]
pub fn dispatch_into(trees: &TreeMap, body: &[u8], buf: &mut Vec<u8>, max_frame: usize) {
    let body_start = buf.len();
    if let Err(Response::Err { code, message }) =
        dispatch_ok(trees, body, buf, body_start, max_frame)
    {
        buf.truncate(body_start);
        encode_err_frame(buf, code, &message, max_frame);
    }
}

#[cfg(not(feature = "ctx-gates"))]
fn dispatch_into(trees: &TreeMap, body: &[u8], buf: &mut Vec<u8>, max_frame: usize) {
    let body_start = buf.len();
    if let Err(Response::Err { code, message }) =
        dispatch_ok(trees, body, buf, body_start, max_frame)
    {
        buf.truncate(body_start);
        encode_err_frame(buf, code, &message, max_frame);
    }
}

fn dispatch_ok(
    trees: &TreeMap,
    body: &[u8],
    buf: &mut Vec<u8>,
    body_start: usize,
    max_frame: usize,
) -> Result<(), Response> {
    let request = decode_request(body).map_err(|error| err_response(400, &error.to_string()))?;

    if request.op == OpCode::ListCollections {
        write_collections(trees, buf)?;
        return check_frame_size(buf, body_start, max_frame);
    }

    let handler = trees.get(&request.hashname).ok_or_else(|| Response::Err {
        code: 404,
        message: format!("tree not found: {:#X}", request.hashname),
    })?;

    if request.op == OpCode::GetSchema {
        let schema = serde_json::to_vec(&handler.schema())
            .map_err(|error| err_response(500, &error.to_string()))?;
        write_schema_bytes(buf, &schema);
        return check_frame_size(buf, body_start, max_frame);
    }

    match (request.op, request.payload) {
        (OpCode::Get, RequestPayload::Key(key)) => {
            write_optional_data(buf, body_start, max_frame, |out| {
                handler.get_into(&key, out)
            })
        }
        (OpCode::Contains, RequestPayload::Key(key)) => {
            buf.push(0x00);
            let found = handler
                .contains(&key)
                .map_err(|error| err_response(error.status_code(), &error.to_string()))?;
            buf.push(u8::from(found));
            check_frame_size(buf, body_start, max_frame)
        }
        (OpCode::EntryLen, RequestPayload::EntryLen { key }) => {
            buf.push(0x00);
            match handler
                .entry_len(&key)
                .map_err(|error| err_response(error.status_code(), &error.to_string()))?
            {
                Some(len) => {
                    buf.push(1);
                    buf.extend_from_slice(&len.to_be_bytes());
                }
                None => buf.push(0),
            }
            check_frame_size(buf, body_start, max_frame)
        }
        (OpCode::First, RequestPayload::Empty) => {
            write_optional_row(buf, body_start, max_frame, |writer| {
                handler.first_into(writer)
            })
        }
        (OpCode::Last, RequestPayload::Empty) => {
            write_optional_row(buf, body_start, max_frame, |writer| {
                handler.last_into(writer)
            })
        }
        (OpCode::Range, RequestPayload::Range { start, end, limit }) => {
            write_range(buf, body_start, max_frame, limit, |writer| {
                handler.range_into(start, end, writer)
            })
        }
        (OpCode::RangeKeys, RequestPayload::Range { start, end, limit }) => {
            write_range(buf, body_start, max_frame, limit, |writer| {
                handler.range_keys_into(start, end, writer)
            })
        }
        (OpCode::Count, RequestPayload::Count { exact }) => {
            buf.push(0x00);
            let count = handler
                .count(exact)
                .map_err(|error| err_response(error.status_code(), &error.to_string()))?;
            buf.extend_from_slice(&count.to_be_bytes());
            check_frame_size(buf, body_start, max_frame)
        }
        (OpCode::Write, RequestPayload::Write { key, flag, value }) => {
            buf.push(0x00);
            let key = handler
                .write(key, flag, value)
                .map_err(|error| err_response(error.status_code(), &error.to_string()))?;
            write_bytes(buf, &key)?;
            check_frame_size(buf, body_start, max_frame)
        }
        (OpCode::Remove, RequestPayload::Remove { key, soft }) => {
            handler
                .remove(&key, soft)
                .map_err(|error| err_response(error.status_code(), &error.to_string()))?;
            buf.push(0x00);
            check_frame_size(buf, body_start, max_frame)
        }
        (OpCode::Take, RequestPayload::Take { key, soft }) => {
            write_optional_data(buf, body_start, max_frame, |out| {
                handler.take_into(&key, soft, out)
            })
        }
        (OpCode::ApplyBatch, RequestPayload::Batch(items)) => {
            handler
                .apply_batch(items)
                .map_err(|error| err_response(error.status_code(), &error.to_string()))?;
            buf.push(0x00);
            check_frame_size(buf, body_start, max_frame)
        }
        _ => Err(Response::Err {
            code: 400,
            message: "invalid op/payload combination".into(),
        }),
    }
}

fn write_optional_data<F>(
    buf: &mut Vec<u8>,
    body_start: usize,
    max_frame: usize,
    write: F,
) -> Result<(), Response>
where
    F: FnOnce(&mut Vec<u8>) -> DbResult<bool>,
{
    buf.push(0x00);
    let presence_at = buf.len();
    buf.push(0);
    let len_at = buf.len();
    buf.extend_from_slice(&[0; 4]);
    let found =
        write(buf).map_err(|error| err_response(error.status_code(), &error.to_string()))?;
    if found {
        let len = u32::try_from(buf.len() - len_at - 4).map_err(|_| response_too_large())?;
        buf[presence_at] = 1;
        buf[len_at..len_at + 4].copy_from_slice(&len.to_be_bytes());
    } else {
        buf.truncate(presence_at + 1);
    }
    check_frame_size(buf, body_start, max_frame)
}

fn write_optional_row<F>(
    buf: &mut Vec<u8>,
    body_start: usize,
    max_frame: usize,
    write: F,
) -> Result<(), Response>
where
    F: FnOnce(&mut RowWriter<'_>) -> DbResult<bool>,
{
    buf.push(0x00);
    let presence_at = buf.len();
    buf.push(0);
    let (found, rows) = {
        let budget = max_frame.saturating_sub(buf.len() - body_start);
        let mut writer = RowWriter::new(buf, 1, budget, "rpc response");
        let found = write(&mut writer)
            .map_err(|error| err_response(error.status_code(), &error.to_string()))?;
        (found, writer.rows())
    };
    if found {
        if rows != 1 {
            return Err(Response::Err {
                code: 500,
                message: "internal error".into(),
            });
        }
        buf[presence_at] = 1;
    } else {
        buf.truncate(presence_at + 1);
    }
    check_frame_size(buf, body_start, max_frame)
}

fn write_range<F>(
    buf: &mut Vec<u8>,
    body_start: usize,
    max_frame: usize,
    limit: u32,
    write: F,
) -> Result<(), Response>
where
    F: FnOnce(&mut RowWriter<'_>) -> DbResult<u32>,
{
    buf.push(0x00);
    let count_at = buf.len();
    buf.extend_from_slice(&[0; 4]);
    let count = {
        let budget = max_frame.saturating_sub(buf.len() - body_start);
        let mut writer = RowWriter::new(buf, limit, budget, "rpc response");
        let _reported = write(&mut writer)
            .map_err(|error| err_response(error.status_code(), &error.to_string()))?;
        writer.rows()
    };
    buf[count_at..count_at + 4].copy_from_slice(&count.to_be_bytes());
    check_frame_size(buf, body_start, max_frame)
}

fn write_bytes(buf: &mut Vec<u8>, bytes: &[u8]) -> Result<(), Response> {
    let len = u32::try_from(bytes.len()).map_err(|_| response_too_large())?;
    buf.extend_from_slice(&len.to_be_bytes());
    buf.extend_from_slice(bytes);
    Ok(())
}

fn write_schema_bytes(buf: &mut Vec<u8>, schema: &[u8]) {
    buf.push(0x00);
    buf.extend_from_slice(schema);
}

fn write_collections(trees: &TreeMap, buf: &mut Vec<u8>) -> Result<(), Response> {
    buf.push(0x00);
    let count = u32::try_from(trees.len()).map_err(|_| response_too_large())?;
    buf.extend_from_slice(&count.to_be_bytes());
    let mut entries: Vec<_> = trees.iter().collect();
    entries.sort_unstable_by(|(_, left), (_, right)| left.name().cmp(right.name()));
    for (&hashname, handler) in entries {
        let name = handler.name().as_bytes();
        let (typ_hash, version) = handler.info();
        let count = handler.count(false).unwrap_or_else(|error| {
            tracing::warn!("count() failed for '{}': {error}", handler.name());
            0
        });
        // armdb exposes its collection name as the partition name as well.
        write_bytes(buf, name)?;
        write_bytes(buf, name)?;
        buf.extend_from_slice(&hashname.to_be_bytes());
        buf.extend_from_slice(&typ_hash.to_be_bytes());
        buf.extend_from_slice(&version.to_be_bytes());
        buf.extend_from_slice(&count.to_be_bytes());
        buf.push(u8::from(handler.scannable()));
    }
    Ok(())
}

fn check_frame_size(buf: &[u8], body_start: usize, max_frame: usize) -> Result<(), Response> {
    if buf.len() - body_start > max_frame {
        return Err(response_too_large());
    }
    Ok(())
}

fn response_too_large() -> Response {
    err_response(413, "response exceeds the frame limit")
}

fn reset_response_buffer(out: &mut Vec<u8>, idle_capacity: usize) {
    out.clear();
    if out.capacity() > idle_capacity.saturating_mul(2) {
        out.shrink_to(idle_capacity);
    }
}

// ─── RpcHandle ───────────────────────────────────────────────────────────────

/// Handle for a running RPC listener. Dropping it stops the acceptor and all
/// active connections before joining the listener-owned threads.
pub struct RpcHandle {
    acceptor: Option<JoinHandle<()>>,
    rejector: Option<JoinHandle<()>>,
    state: Arc<ListenerState>,
}

impl RpcHandle {
    pub fn stop(&mut self) {
        self.state.stop.store(true, Ordering::Release);
        if let Some(handle) = self.acceptor.take() {
            let _ = handle.join();
        }
        if let Some(handle) = self.rejector.take() {
            let _ = handle.join();
        }
    }
}

impl Drop for RpcHandle {
    fn drop(&mut self) {
        self.stop();
    }
}

// ─── Db listen methods ────────────────────────────────────────────────────

impl Db {
    /// Start a TCP RPC listener on `addr`.
    ///
    /// # Trust model
    ///
    /// The RPC server assumes a **trusted network**: peers are not authenticated,
    /// operations are not authorized, and heavy requests are unbudgeted — a `Range`
    /// with unbounded bounds and `limit = 0` (or `Count { exact: true }`) scans the
    /// whole collection, materializes the response in that connection's buffer,
    /// and occupies that connection's thread while it runs; other connection
    /// threads continue independently. `limit = 0` deliberately means
    /// "everything" (it is the wire spelling of "no limit", and an unset default
    /// field is indistinguishable from a deliberate full scan), so it is capped at
    /// [`UNLIMITED_RANGE_CAP`](super::handler::UNLIMITED_RANGE_CAP) rows rather
    /// than left truly unbounded; an explicit non-zero `limit` is honored as-is.
    /// Bind to loopback / a private interface (or prefer
    /// [`listen_uds`](Self::listen_uds)) and front untrusted networks with an mTLS
    /// tunnel — do not expose this listener directly. See the "Network trust model"
    /// section in `docs/design.md`.
    pub fn listen_tcp(&self, addr: impl ToSocketAddrs) -> DbResult<SocketAddr> {
        self.listen_tcp_with(addr, RpcConfig::default())
    }

    /// Bind TCP synchronously, then start its blocking acceptor.
    pub fn listen_tcp_with(
        &self,
        addr: impl ToSocketAddrs,
        cfg: RpcConfig,
    ) -> DbResult<SocketAddr> {
        cfg.validate()?;
        let listener = TcpListener::bind(addr)?;
        let local = listener.local_addr()?;
        listener.set_nonblocking(true)?;
        self.spawn_acceptor(Acceptor::Tcp(listener), cfg)?;
        Ok(local)
    }

    #[cfg(unix)]
    pub fn listen_uds(&self, path: impl AsRef<Path>) -> DbResult<()> {
        self.listen_uds_with(path, RpcConfig::default())
    }

    /// Bind a Unix socket synchronously, removing only a stale socket path.
    #[cfg(unix)]
    pub fn listen_uds_with(&self, path: impl AsRef<Path>, cfg: RpcConfig) -> DbResult<()> {
        cfg.validate()?;
        let path = path.as_ref().to_path_buf();
        if path.exists() {
            remove_stale_uds_socket(&path)?;
        }
        let listener = UnixListener::bind(&path)?;
        listener.set_nonblocking(true)?;
        self.spawn_acceptor(Acceptor::Uds { listener, path }, cfg)
    }

    fn spawn_acceptor(&self, acceptor: Acceptor, cfg: RpcConfig) -> DbResult<()> {
        let trees = self.build_tree_map();
        let state = Arc::new(ListenerState {
            stop: AtomicBool::new(self.shutdown.is_shutdown()),
            live: AtomicUsize::new(0),
            cfg: cfg.clone(),
            #[cfg(test)]
            reaped: AtomicUsize::new(0),
        });
        let stop = Arc::clone(&state);
        self.shutdown.listen(Arc::new(move || {
            stop.stop.store(true, Ordering::Release);
        }));

        let shutdown = self.shutdown.clone();
        match acceptor {
            Acceptor::Tcp(listener) => {
                let (reject_tx, reject_rx) = sync_channel(32);
                let reject_state = Arc::clone(&state);
                let rejector = thread::Builder::new()
                    .name("armdb-rpc-reject".into())
                    .spawn(move || rejector_loop(reject_rx, reject_state))?;
                let accept_state = Arc::clone(&state);
                let acceptor = thread::Builder::new()
                    .name("armdb-rpc-accept".into())
                    .spawn(move || {
                        accept_loop(
                            listener,
                            accept_state,
                            shutdown,
                            trees,
                            reject_tx,
                            |listener| listener.accept().map(|(stream, _)| stream),
                            configure_tcp,
                        );
                    });
                let acceptor = match acceptor {
                    Ok(handle) => handle,
                    Err(error) => {
                        state.stop.store(true, Ordering::Release);
                        let _ = rejector.join();
                        return Err(error.into());
                    }
                };
                crate::sync::lock(&self.rpc_handles).push(RpcHandle {
                    acceptor: Some(acceptor),
                    rejector: Some(rejector),
                    state,
                });
            }
            #[cfg(unix)]
            Acceptor::Uds { listener, path } => {
                let (reject_tx, reject_rx) = sync_channel(32);
                let reject_state = Arc::clone(&state);
                let rejector = thread::Builder::new()
                    .name("armdb-rpc-reject".into())
                    .spawn(move || rejector_loop(reject_rx, reject_state))?;
                let accept_state = Arc::clone(&state);
                let acceptor = thread::Builder::new()
                    .name("armdb-rpc-accept".into())
                    .spawn(move || {
                        accept_loop(
                            listener,
                            accept_state,
                            shutdown,
                            trees,
                            reject_tx,
                            |listener| listener.accept().map(|(stream, _)| stream),
                            configure_uds,
                        );
                        if let Err(error) = std::fs::remove_file(path)
                            && error.kind() != io::ErrorKind::NotFound
                        {
                            tracing::warn!("rpc: failed to unlink UDS socket: {error}");
                        }
                    });
                let acceptor = match acceptor {
                    Ok(handle) => handle,
                    Err(error) => {
                        state.stop.store(true, Ordering::Release);
                        let _ = rejector.join();
                        return Err(error.into());
                    }
                };
                crate::sync::lock(&self.rpc_handles).push(RpcHandle {
                    acceptor: Some(acceptor),
                    rejector: Some(rejector),
                    state,
                });
            }
        }
        Ok(())
    }
}

#[cfg(unix)]
fn remove_stale_uds_socket(path: &Path) -> io::Result<()> {
    match UnixStream::connect(path) {
        Ok(_) => Err(io::Error::new(
            io::ErrorKind::AddrInUse,
            "RPC UDS path is already served",
        )),
        Err(error)
            if error.kind() == io::ErrorKind::ConnectionRefused
                && std::fs::symlink_metadata(path)?.file_type().is_socket() =>
        {
            std::fs::remove_file(path)
        }
        Err(error) => Err(error),
    }
}

#[cfg(test)]
mod err_frame_tests {
    use super::encode_err_frame;

    #[test]
    fn err_frame_never_exceeds_the_configured_response_body_limit() {
        for (code, message, expected) in [
            (400, "aaa€", "aa"),
            (413, "response exceeds the frame limit", "re"),
            (503, "too many connections", "to"),
        ] {
            let mut buf = Vec::new();
            encode_err_frame(&mut buf, code, message, 7);

            assert_eq!(buf.len(), 7, "{code} must fit max_frame exactly");
            assert_eq!(u16::from_be_bytes([buf[1], buf[2]]), code);
            assert_eq!(u16::from_be_bytes([buf[3], buf[4]]), 2);
            assert_eq!(
                std::str::from_utf8(&buf[5..]).expect("utf-8 boundary"),
                expected
            );
        }
    }

    // AI-11: the length prefix is a `u16`; a message ≥ 64 KiB must be truncated so
    // the prefix equals the bytes actually written. Pre-fix the prefix wrapped mod
    // 65536 while all bytes were emitted, desyncing the client frame parser.
    #[test]
    fn err_frame_len_prefix_matches_written_bytes_when_over_u16() {
        let big = "x".repeat(70_000);

        // The pre-fix formula (`len as u16`) disagreed with the byte count.
        let buggy_prefix = big.len() as u16; // 70000 as u16 == 4464
        assert_ne!(
            buggy_prefix as usize,
            big.len(),
            "sanity: pre-fix u16 prefix disagreed with the byte count"
        );

        let mut buf = Vec::new();
        encode_err_frame(&mut buf, 500, &big, usize::MAX);

        // Frame layout: [tag:1][code:2][len:2][msg…]
        let len = u16::from_be_bytes([buf[3], buf[4]]) as usize;
        let msg_written = buf.len() - 5;
        assert_eq!(
            len, msg_written,
            "len prefix must equal bytes actually written"
        );
        assert_eq!(len, u16::MAX as usize, "message capped at u16::MAX");
    }

    #[test]
    fn err_frame_small_message_roundtrips() {
        let mut buf = Vec::new();
        encode_err_frame(&mut buf, 404, "not found", usize::MAX);
        assert_eq!(buf[0], 0x01);
        assert_eq!(u16::from_be_bytes([buf[1], buf[2]]), 404);
        let len = u16::from_be_bytes([buf[3], buf[4]]) as usize;
        assert_eq!(len, "not found".len());
        assert_eq!(&buf[5..], b"not found");
    }

    #[test]
    fn err_frame_truncates_at_utf8_boundary() {
        let message = format!("{}€", "a".repeat(u16::MAX as usize - 1));
        let mut buf = Vec::new();
        encode_err_frame(&mut buf, 500, &message, usize::MAX);
        let len = u16::from_be_bytes([buf[3], buf[4]]) as usize;
        let encoded = &buf[5..5 + len];
        assert_eq!(len, u16::MAX as usize - 1);
        assert_eq!(std::str::from_utf8(encoded).unwrap(), "a".repeat(len));
        assert_eq!(buf.len(), 5 + len);
    }
}

#[cfg(test)]
mod response_buffer_tests {
    use super::reset_response_buffer;

    #[test]
    fn response_buffer_releases_large_written_body_before_becoming_idle() {
        let mut response = Vec::with_capacity(256 * 1024);
        response.extend(std::iter::repeat_n(0_u8, 128 * 1024));

        reset_response_buffer(&mut response, 4096);

        assert!(
            response.is_empty(),
            "an idle buffer must not retain response bytes"
        );
        assert!(
            response.capacity() <= 8192,
            "large response capacity remained: {}",
            response.capacity()
        );
    }
}

#[cfg(test)]
mod config_tests {
    use super::{AcceptBackoff, Db, RpcConfig};
    #[cfg(unix)]
    use std::fs;
    use std::time::Duration;

    #[test]
    fn config_rejects_degenerate_values() {
        assert!(
            RpcConfig::builder()
                .max_connections(0)
                .build()
                .validate()
                .is_err()
        );
        assert!(
            RpcConfig::builder()
                .max_frame(0)
                .build()
                .validate()
                .is_err()
        );
        assert!(
            RpcConfig::builder()
                .max_frame(4)
                .build()
                .validate()
                .is_err()
        );
        assert!(
            RpcConfig::builder()
                .max_frame(u32::MAX as usize + 1)
                .build()
                .validate()
                .is_err()
        );
        assert!(
            RpcConfig::builder()
                .read_timeout(Duration::ZERO)
                .build()
                .validate()
                .is_err()
        );
        assert!(
            RpcConfig::builder()
                .poll_interval(Duration::ZERO)
                .build()
                .validate()
                .is_err()
        );
        assert!(RpcConfig::default().validate().is_ok());
    }

    #[test]
    fn listen_tcp_returns_the_bound_address_and_reports_bind_failure() {
        let dir = tempfile::tempdir().expect("temporary database directory");
        let db = Db::open_test(dir.path()).expect("open test database");
        let addr = db.listen_tcp("127.0.0.1:0").expect("bind");
        assert_ne!(addr.port(), 0);
        assert!(db.listen_tcp("127.0.0.1:1").is_err() || !cfg!(target_os = "linux"));
    }

    #[cfg(unix)]
    #[test]
    fn listen_uds_does_not_replace_a_regular_file() {
        let dir = tempfile::tempdir().expect("temporary database directory");
        let path = dir.path().join("not-a-socket");
        fs::write(&path, b"must remain untouched").expect("create regular file");
        let db = Db::open_test(dir.path()).expect("open test database");

        assert!(db.listen_uds(&path).is_err());
        assert_eq!(
            fs::read(path).expect("read regular file"),
            b"must remain untouched"
        );
    }

    #[test]
    fn accept_backoff_grows_on_errors_and_resets_after_success() {
        let mut backoff = AcceptBackoff::new();
        assert_eq!(backoff.on_error(), Duration::from_millis(5));
        assert_eq!(backoff.on_error(), Duration::from_millis(10));
        backoff.on_accept();
        assert_eq!(backoff.on_error(), Duration::from_millis(5));
    }
}

#[cfg(test)]
mod rpc_handle_tests {
    use super::Db;
    use std::{
        io::Read,
        net::TcpStream,
        sync::atomic::Ordering,
        thread,
        time::{Duration, Instant},
    };

    // AI-05: `RpcHandle::stop` must terminate its own listener's live
    // connection without setting Db-global shutdown. This must be a unit test:
    // integration tests cannot take the intentionally private handle.
    #[test]
    fn rpc_handle_stop_terminates_active_connection_without_global_shutdown() {
        let dir = tempfile::tempdir().expect("temporary database directory");
        let db = Db::open_test(dir.path()).expect("open test database");
        let addr = db.listen_tcp("127.0.0.1:0").expect("bind listener");
        let mut client = TcpStream::connect(addr).expect("connect client");
        client
            .set_read_timeout(Some(Duration::from_secs(1)))
            .expect("set client read timeout");

        // Take the private owner, then wait for its admission counter rather
        // than sleeping: reaching one proves the connection thread is live.
        let mut handle = crate::sync::lock(&db.rpc_handles)
            .pop()
            .expect("one rpc handle");
        let witness_started = Instant::now();
        while handle.state.live.load(Ordering::Acquire) != 1 {
            assert!(
                witness_started.elapsed() < Duration::from_secs(1),
                "connection was not admitted"
            );
            thread::yield_now();
        }

        let stopped_started = Instant::now();
        handle.stop();
        assert!(
            stopped_started.elapsed() < Duration::from_secs(1),
            "RpcHandle::stop did not terminate the listener"
        );
        assert!(
            !db.shutdown_signal().is_shutdown(),
            "local listener stop must not trigger Db-global shutdown"
        );
        assert_eq!(
            client.read(&mut [0; 1]).expect("connection close"),
            0,
            "local stop did not close the active connection"
        );
    }
}

#[cfg(test)]
mod connection_worker_tests {
    use std::collections::HashMap;
    use std::io::{self, Read, Write};
    use std::sync::{
        Arc,
        atomic::{AtomicBool, AtomicUsize, Ordering},
        mpsc::sync_channel,
    };
    use std::time::Duration;

    use super::{ListenerState, RejectorSocket, RpcConfig, TreeMap, accept_loop};
    use crate::shutdown::ShutdownSignal;

    struct PanickingStream;

    impl Read for PanickingStream {
        fn read(&mut self, _: &mut [u8]) -> io::Result<usize> {
            panic!("connection read panic");
        }
    }

    impl Write for PanickingStream {
        fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
            Ok(bytes.len())
        }

        fn flush(&mut self) -> io::Result<()> {
            Ok(())
        }
    }

    impl armour_rpc::DeadlineWriter for PanickingStream {
        fn set_write_timeout(&self, _: Option<Duration>) -> io::Result<()> {
            Ok(())
        }
    }

    impl RejectorSocket for PanickingStream {
        fn configure_rejector(&self, _: &RpcConfig) -> io::Result<()> {
            Ok(())
        }

        fn clone_for_rejection(&self) -> io::Result<Self> {
            Ok(Self)
        }
    }

    struct ClosedStream;

    impl Read for ClosedStream {
        fn read(&mut self, _: &mut [u8]) -> io::Result<usize> {
            Ok(0)
        }
    }

    impl Write for ClosedStream {
        fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
            Ok(bytes.len())
        }

        fn flush(&mut self) -> io::Result<()> {
            Ok(())
        }
    }

    impl armour_rpc::DeadlineWriter for ClosedStream {
        fn set_write_timeout(&self, _: Option<Duration>) -> io::Result<()> {
            Ok(())
        }
    }

    impl RejectorSocket for ClosedStream {
        fn configure_rejector(&self, _: &RpcConfig) -> io::Result<()> {
            Ok(())
        }

        fn clone_for_rejection(&self) -> io::Result<Self> {
            Ok(Self)
        }
    }

    #[test]
    fn reaps_finished_connection_workers_before_the_listener_stops() {
        let state = Arc::new(ListenerState {
            stop: AtomicBool::new(false),
            live: AtomicUsize::new(0),
            cfg: RpcConfig::builder()
                .poll_interval(Duration::from_millis(1))
                .build(),
            reaped: AtomicUsize::new(0),
        });
        let state_for_accept = Arc::clone(&state);
        let accepted_once = AtomicBool::new(false);
        let started = std::time::Instant::now();
        let (rejector, _reject_rx) = sync_channel(1);
        let trees: TreeMap = Arc::new(HashMap::new());

        accept_loop(
            (),
            Arc::clone(&state),
            ShutdownSignal::new(),
            trees,
            rejector,
            move |_| {
                if !accepted_once.swap(true, Ordering::AcqRel) {
                    return Ok(ClosedStream);
                }
                if state_for_accept.reaped.load(Ordering::Acquire) == 1
                    || started.elapsed() >= Duration::from_secs(1)
                {
                    state_for_accept.stop.store(true, Ordering::Release);
                }
                Err(io::Error::new(io::ErrorKind::WouldBlock, "service tick"))
            },
            |_, _| Ok(()),
        );

        assert_eq!(
            state.reaped.load(Ordering::Acquire),
            1,
            "the accept loop did not reap its completed connection worker"
        );
    }

    #[test]
    fn panicking_connection_worker_releases_its_admission_slot() {
        let state = Arc::new(ListenerState {
            stop: AtomicBool::new(false),
            live: AtomicUsize::new(0),
            cfg: RpcConfig::default(),
            reaped: AtomicUsize::new(0),
        });
        let stop_after_first_accept = Arc::clone(&state);
        let accepted_once = AtomicBool::new(false);
        let (rejector, _reject_rx) = sync_channel(1);
        let trees: TreeMap = Arc::new(HashMap::new());

        accept_loop(
            (),
            Arc::clone(&state),
            ShutdownSignal::new(),
            trees,
            rejector,
            move |_| {
                if !accepted_once.swap(true, Ordering::AcqRel) {
                    Ok(PanickingStream)
                } else {
                    stop_after_first_accept.stop.store(true, Ordering::Release);
                    Err(io::Error::new(io::ErrorKind::WouldBlock, "stop listener"))
                }
            },
            |_, _| Ok(()),
        );

        assert_eq!(
            state.live.load(Ordering::Acquire),
            0,
            "a panicking worker retained its admission slot"
        );
    }
}

#[cfg(test)]
mod rejector_tests {
    use std::io::{self, Write};
    use std::sync::{
        Arc,
        atomic::{AtomicBool, AtomicUsize, Ordering},
        mpsc::sync_channel,
    };
    use std::time::Duration;

    use super::{ListenerState, RejectorSocket, RpcConfig, rejector_loop};

    struct RejectedSocket(Arc<AtomicUsize>);

    impl Write for RejectedSocket {
        fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
            self.0.fetch_add(1, Ordering::Relaxed);
            Ok(bytes.len())
        }

        fn flush(&mut self) -> io::Result<()> {
            Ok(())
        }
    }

    impl armour_rpc::DeadlineWriter for RejectedSocket {
        fn set_write_timeout(&self, _: Option<Duration>) -> io::Result<()> {
            Ok(())
        }
    }

    impl RejectorSocket for RejectedSocket {
        fn configure_rejector(&self, _: &RpcConfig) -> io::Result<()> {
            Ok(())
        }

        fn clone_for_rejection(&self) -> io::Result<Self> {
            Ok(Self(Arc::clone(&self.0)))
        }
    }

    #[test]
    fn stopped_rejector_discards_queued_sockets_without_writing_them() {
        let writes = Arc::new(AtomicUsize::new(0));
        let (tx, rx) = sync_channel(32);
        for _ in 0..32 {
            tx.send(RejectedSocket(Arc::clone(&writes)))
                .expect("queue rejection");
        }
        drop(tx);
        let state = Arc::new(ListenerState {
            stop: AtomicBool::new(true),
            live: AtomicUsize::new(0),
            cfg: RpcConfig::default(),
            reaped: AtomicUsize::new(0),
        });

        rejector_loop(rx, state);

        assert_eq!(
            writes.load(Ordering::Relaxed),
            0,
            "stop must discard the queue"
        );
    }
}

#[cfg(test)]
mod dispatch_status_code_tests {
    use crate::error::{DbError, SchemaMismatchKind};

    #[test]
    fn key_not_found_maps_to_404() {
        assert_eq!(DbError::KeyNotFound.status_code(), 404);
    }
    #[test]
    fn key_exists_maps_to_409() {
        assert_eq!(DbError::KeyExists.status_code(), 409);
    }
    #[test]
    fn client_error_maps_to_400() {
        assert_eq!(DbError::Client("bad").status_code(), 400);
    }
    #[test]
    fn not_implemented_maps_to_501() {
        assert_eq!(DbError::NotImplemented.status_code(), 501);
    }
    #[test]
    fn schema_mismatch_maps_to_422() {
        let e = DbError::SchemaMismatch {
            name: "x".into(),
            kind: SchemaMismatchKind::MissingStep { from: 0 },
        };
        assert_eq!(e.status_code(), 422);
    }
    #[test]
    fn tx_conflict_maps_to_409() {
        // D4: TxConflict is a client-actionable "retry" signal → 409, not 500.
        assert_eq!(DbError::TxConflict.status_code(), 409);
    }
    #[test]
    fn shard_mismatch_stays_500() {
        // D4: wrong-shard routing is a server bug → stays opaque 500 (not 4xx).
        assert_eq!(DbError::ShardMismatch.status_code(), 500);
    }
}

#[cfg(test)]
mod err_response_policy_tests {
    use super::{Response, err_response};

    fn unpack(r: Response) -> (u16, String) {
        match r {
            Response::Err { code, message } => (code, message),
        }
    }

    #[test]
    fn client_error_4xx_forwards_message_verbatim() {
        // 4xx is the client's fault: it must see *what* it got wrong.
        let (code, msg) = unpack(err_response(404, "tree not found: 0xDEAD"));
        assert_eq!(code, 404);
        assert_eq!(msg, "tree not found: 0xDEAD");
    }

    #[test]
    fn server_error_5xx_sends_opaque_message() {
        // 5xx leaks nothing: the original (paths, corruption state) must NOT
        // appear on the wire — only a generic "internal error".
        let original = "corrupted entry at offset 12345 in /var/data/shard_007/000012.data";
        let (code, msg) = unpack(err_response(500, original));
        assert_eq!(code, 500);
        assert_eq!(msg, "internal error");
        assert!(
            !msg.contains("12345") && !msg.contains("/var/data"),
            "5xx response must not echo internal details"
        );
    }

    #[test]
    fn not_implemented_501_forwards_message() {
        // 501 is exempt from the opaque policy: capability mismatches and
        // unsupported modes use a fixed Display string that is safe to forward.
        let (code, msg) = unpack(err_response(501, "not implemented"));
        assert_eq!(code, 501);
        assert_eq!(msg, "not implemented");
    }
}

#[cfg(test)]
mod collection_metadata_tests {
    use std::collections::HashMap;
    use std::ops::Bound;
    use std::sync::Arc;

    use armour_rpc::codec::encode_request;
    use armour_rpc::protocol::{OpCode, Request, RequestPayload, WriteKey};

    use crate::armour::handler::{KeyBytes, RowWriter, RpcHandler, ValueBytes};
    use crate::error::{DbError, DbResult};

    use super::{TreeMap, dispatch_into, wire_golden, write_collections};

    struct StubHandler {
        name: &'static str,
        scannable: bool,
        typ_hash: u64,
        version: u16,
        count: u64,
    }

    impl StubHandler {
        fn basic(name: &'static str, scannable: bool) -> Arc<Self> {
            Arc::new(Self {
                name,
                scannable,
                typ_hash: 0,
                version: 1,
                count: 0,
            })
        }
    }

    impl RpcHandler for StubHandler {
        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.into(),
                version: 1,
                typ: <u8 as armour_core::GetType>::TYPE,
                key_scheme: armour_core::KeyScheme::Bytes,
            }
        }
        fn scannable(&self) -> bool {
            self.scannable
        }
        fn get_into(&self, _: &[u8], _: &mut Vec<u8>) -> DbResult<bool> {
            Ok(false)
        }
        fn contains(&self, _: &[u8]) -> DbResult<bool> {
            Ok(false)
        }
        fn entry_len(&self, _: &[u8]) -> DbResult<Option<u32>> {
            Ok(None)
        }
        fn first_into(&self, _: &mut RowWriter<'_>) -> DbResult<bool> {
            Err(DbError::NotImplemented)
        }
        fn last_into(&self, _: &mut RowWriter<'_>) -> DbResult<bool> {
            Ok(false)
        }
        fn range_into(
            &self,
            _: Bound<KeyBytes>,
            _: Bound<KeyBytes>,
            _: &mut RowWriter<'_>,
        ) -> DbResult<u32> {
            Ok(0)
        }
        fn range_keys_into(
            &self,
            _: Bound<KeyBytes>,
            _: Bound<KeyBytes>,
            _: &mut RowWriter<'_>,
        ) -> DbResult<u32> {
            Ok(0)
        }
        fn write(&self, _: WriteKey, _: Option<bool>, _: ValueBytes) -> DbResult<ValueBytes> {
            Ok(Vec::new())
        }
        fn remove(&self, _: &[u8], _: bool) -> DbResult<()> {
            Ok(())
        }
        fn take_into(&self, _: &[u8], _: bool, _: &mut Vec<u8>) -> DbResult<bool> {
            Ok(false)
        }
        fn count(&self, _: bool) -> DbResult<u64> {
            Ok(self.count)
        }
        fn apply_batch(&self, _: Vec<(KeyBytes, Option<ValueBytes>)>) -> DbResult<()> {
            Ok(())
        }
    }

    #[test]
    fn collections_encoder_places_scannable_byte_per_record() {
        let mut handlers: HashMap<u64, Arc<dyn RpcHandler>> = HashMap::new();
        handlers.insert(0x0102, StubHandler::basic("a", false));
        handlers.insert(0x0708, StubHandler::basic("bc", true));
        let trees: TreeMap = Arc::new(handlers);
        let mut bytes = Vec::new();
        write_collections(&trees, &mut bytes).expect("encode collections");

        assert_eq!(u32::from_be_bytes(bytes[1..5].try_into().unwrap()), 2);
        let records = decode_collection_flags(&bytes);
        assert_eq!(records["a"], false);
        assert_eq!(records["bc"], true);
    }

    #[test]
    fn collections_metadata_propagates_scannable_by_name() {
        let mut handlers: HashMap<u64, Arc<dyn RpcHandler>> = HashMap::new();
        handlers.insert(1, StubHandler::basic("alpha", true));
        handlers.insert(2, StubHandler::basic("beta", false));
        let trees: TreeMap = Arc::new(handlers);

        let mut request = Vec::new();
        encode_request(
            Request {
                op: OpCode::ListCollections,
                hashname: 0,
                payload: RequestPayload::Empty,
            },
            &mut request,
        );
        let mut response = Vec::new();
        dispatch_into(&trees, &request, &mut response, 1 << 20);

        assert_eq!(response[0], 0);
        assert_eq!(u32::from_be_bytes(response[1..5].try_into().unwrap()), 2);
        let records = decode_collection_flags(&response);
        assert_eq!(records["alpha"], true);
        assert_eq!(records["beta"], false);
    }

    #[test]
    fn not_implemented_scan_dispatches_as_501() {
        let mut handlers: HashMap<u64, Arc<dyn RpcHandler>> = HashMap::new();
        handlers.insert(7, StubHandler::basic("plain", false));
        let trees = Arc::new(handlers);
        let mut request = Vec::new();
        encode_request(
            Request {
                op: OpCode::First,
                hashname: 7,
                payload: RequestPayload::Empty,
            },
            &mut request,
        );
        let mut response = Vec::new();
        dispatch_into(&trees, &request, &mut response, 1 << 20);
        assert_eq!(response[0], 1);
        assert_eq!(u16::from_be_bytes(response[1..3].try_into().unwrap()), 501);
        let message_len = u16::from_be_bytes(response[3..5].try_into().unwrap()) as usize;
        assert_eq!(
            &response[5..5 + message_len],
            DbError::NotImplemented.to_string().as_bytes()
        );
    }

    #[test]
    fn dispatch_reproduces_the_frozen_collections_bytes() {
        let mut handlers: HashMap<u64, Arc<dyn RpcHandler>> = HashMap::new();
        handlers.insert(
            0x0102_0304_0506_0708,
            Arc::new(StubHandler {
                name: "a",
                scannable: false,
                typ_hash: 0x1112_1314_1516_1718,
                version: 0x191A,
                count: 0x1B1C_1D1E_1F20_2122,
            }),
        );
        handlers.insert(
            0x2324_2526_2728_292A,
            Arc::new(StubHandler {
                name: "bc",
                scannable: true,
                typ_hash: 0x2B2C_2D2E_2F30_3132,
                version: 0x3334,
                count: 0x3536_3738_393A_3B3C,
            }),
        );
        let trees: TreeMap = Arc::new(handlers);
        let mut request = Vec::new();
        encode_request(
            Request {
                op: OpCode::ListCollections,
                hashname: 0,
                payload: RequestPayload::Empty,
            },
            &mut request,
        );
        let mut actual = Vec::new();
        dispatch_into(&trees, &request, &mut actual, 1 << 20);
        let expected = wire_golden::RESPONSE_VECTORS
            .iter()
            .find(|(name, _)| *name == "collections_two")
            .map(|(_, bytes)| bytes)
            .expect("Task 1 collections response vector");
        assert_eq!(actual, *expected);
    }

    fn decode_collection_flags(bytes: &[u8]) -> HashMap<String, bool> {
        let mut pos = 1;
        let count = u32::from_be_bytes(bytes[pos..pos + 4].try_into().unwrap()) as usize;
        pos += 4;
        let mut flags = HashMap::new();
        for _ in 0..count {
            let name_len = u32::from_be_bytes(bytes[pos..pos + 4].try_into().unwrap()) as usize;
            pos += 4;
            let name = std::str::from_utf8(&bytes[pos..pos + name_len])
                .unwrap()
                .to_owned();
            pos += name_len;
            let partition_len =
                u32::from_be_bytes(bytes[pos..pos + 4].try_into().unwrap()) as usize;
            pos += 4 + partition_len + 8 + 8 + 2 + 8;
            flags.insert(name, bytes[pos] != 0);
            pos += 1;
        }
        assert_eq!(pos, bytes.len());
        flags
    }
}