liminal-server 0.25.4

//! Shared cold-restore fixtures and pins at the durability boundary.
//!
//! The large image extends a closed copy of the small image. Fixture creation
//! is serialized once per nextest run; each stage restores an independent copy.
//! Counts include returned payload bytes and numeric values, not engine I/O.
//! Scan answers lack per-entry stream keys, so their query prefix is explicit.

use std::cell::Cell;
use std::collections::BTreeMap;
use std::error::Error;
use std::fs::{self, File};
use std::ops::Range;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Instant;

use haematite::{Database, DatabaseConfig, EventStore, NodeCacheBudget, Retention};
use liminal::durability::bridge::block_on;
use liminal::durability::{
    DurabilityError, DurableStore, HaematiteStore, RoutedBatchEntry, RoutedBatchItem, ScanCursor,
    ScanPage, StoredEntry,
};
use rustix::fs::{FlockOperation, flock};

pub type TestResult<T> = Result<T, Box<dyn Error>>;

pub const HISTORY_ROWS: [u64; 2] = [1_000, 100_000];

/// Read work returned through every read method of the store seam.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct ReadCounts {
    pub(crate) rows: u64,
    pub(crate) bytes: u64,
}

/// The boundary distinguishes exact streams from numeric keys and scan queries.
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub enum ReadSource {
    Stream(String),
    StreamHead(String),
    RoutedValue { route: String, key: Vec<u8> },
    Value(String),
    ScanPrefix(String),
}

/// Stage name, then read source, then read counts.
pub type StageReads = BTreeMap<String, BTreeMap<ReadSource, ReadCounts>>;

thread_local! {
    static BOOT_STAGE: Cell<&'static str> = const { Cell::new("unassigned") };
}

struct StageGuard {
    previous: &'static str,
}

impl Drop for StageGuard {
    fn drop(&mut self) {
        BOOT_STAGE.with(|stage| stage.set(self.previous));
    }
}

/// Names synchronous boot work, restoring the outer stage even on a refusal.
pub fn during_stage<T>(stage: &'static str, work: impl FnOnce() -> T) -> T {
    let previous = BOOT_STAGE.with(|current| current.replace(stage));
    let guard = StageGuard { previous };
    let answer = work();
    drop(guard);
    answer
}

/// Boots the production handler through its ordinary constructor.
///
/// # Errors
/// Propagates the production constructor's validation or restore refusal.
pub fn boot_participant(
    store: Arc<dyn DurableStore>,
    config: crate::config::types::ParticipantConfig,
) -> Result<
    crate::server::participant::ProductionParticipantHandler,
    crate::server::participant::ParticipantSemanticError,
> {
    crate::server::participant::ProductionParticipantHandler::new(store, config)
}

#[derive(Debug)]
struct BootReadStore {
    inner: Arc<dyn DurableStore>,
    reads: Mutex<StageReads>,
}

impl BootReadStore {
    fn new(inner: Arc<dyn DurableStore>) -> Self {
        Self {
            inner,
            reads: Mutex::new(StageReads::new()),
        }
    }

    fn counts(&self) -> TestResult<StageReads> {
        Ok(self
            .reads
            .lock()
            .map_err(|_| "boot read counter lock poisoned")?
            .clone())
    }

    fn note(&self, source: ReadSource, rows: &[StoredEntry]) -> Result<(), DurabilityError> {
        let count = u64::try_from(rows.len())
            .map_err(|error| DurabilityError::ConfigError(format!("boot row count: {error}")))?;
        let bytes = rows.iter().try_fold(0_u64, |total, row| {
            let bytes = u64::try_from(row.payload.len()).map_err(|error| {
                DurabilityError::ConfigError(format!("boot payload size: {error}"))
            })?;
            total.checked_add(bytes).ok_or_else(|| {
                DurabilityError::ConfigError("boot payload byte count overflow".to_owned())
            })
        })?;
        self.note_counts(source, count, bytes)
    }

    fn note_counts(
        &self,
        source: ReadSource,
        rows: u64,
        bytes: u64,
    ) -> Result<(), DurabilityError> {
        let mut reads = self.reads.lock().map_err(|_| {
            DurabilityError::ConfigError("boot read counter lock poisoned".to_owned())
        })?;
        let stage = BOOT_STAGE.with(Cell::get);
        let previous = reads
            .get(stage)
            .and_then(|streams| streams.get(&source))
            .copied()
            .unwrap_or_default();
        let next = ReadCounts {
            rows: previous.rows.checked_add(rows).ok_or_else(|| {
                DurabilityError::ConfigError("boot row count overflow".to_owned())
            })?,
            bytes: previous.bytes.checked_add(bytes).ok_or_else(|| {
                DurabilityError::ConfigError("boot byte count overflow".to_owned())
            })?,
        };
        reads
            .entry(stage.to_owned())
            .or_default()
            .insert(source, next);
        drop(reads);
        Ok(())
    }
}

#[async_trait::async_trait]
impl DurableStore for BootReadStore {
    async fn apply_routed_batch(
        &self,
        route: &str,
        items: Vec<RoutedBatchItem>,
    ) -> Result<(), DurabilityError> {
        self.inner.apply_routed_batch(route, items).await
    }

    async fn apply_routed_entries(
        &self,
        route: &str,
        entries: Vec<RoutedBatchEntry>,
    ) -> Result<(), DurabilityError> {
        self.inner.apply_routed_entries(route, entries).await
    }

    async fn get_routed(
        &self,
        route: &str,
        key: &[u8],
    ) -> Result<Option<Vec<u8>>, DurabilityError> {
        let value = self.inner.get_routed(route, key).await?;
        let bytes = value.as_ref().map_or(Ok(0), |value| {
            u64::try_from(value.len()).map_err(|error| {
                DurabilityError::ConfigError(format!("boot routed value size: {error}"))
            })
        })?;
        self.note_counts(
            ReadSource::RoutedValue {
                route: route.to_owned(),
                key: key.to_vec(),
            },
            u64::from(value.is_some()),
            bytes,
        )?;
        Ok(value)
    }

    async fn read_stream_next_seq(&self, stream_key: &str) -> Result<Option<u64>, DurabilityError> {
        let head = self.inner.read_stream_next_seq(stream_key).await?;
        self.note_counts(
            ReadSource::StreamHead(stream_key.to_owned()),
            u64::from(head.is_some()),
            if head.is_some() { 8 } else { 0 },
        )?;
        Ok(head)
    }

    async fn append(
        &self,
        stream_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.inner.append(stream_key, payload, expected_seq).await
    }

    async fn append_all(
        &self,
        stream_key: &str,
        payloads: Vec<Vec<u8>>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.inner
            .append_all(stream_key, payloads, expected_seq)
            .await
    }

    async fn read_from(
        &self,
        stream_key: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Vec<StoredEntry>, DurabilityError> {
        let rows = self.inner.read_from(stream_key, offset, limit).await?;
        self.note(ReadSource::Stream(stream_key.to_owned()), &rows)?;
        Ok(rows)
    }

    async fn read_at(
        &self,
        stream_key: &str,
        sequence: u64,
    ) -> Result<Option<StoredEntry>, DurabilityError> {
        let row = self.inner.read_at(stream_key, sequence).await?;
        if let Some(row) = &row {
            self.note(
                ReadSource::Stream(stream_key.to_owned()),
                std::slice::from_ref(row),
            )?;
        }
        Ok(row)
    }

    async fn read_value(&self, key: &str) -> Result<Option<u64>, DurabilityError> {
        let value = self.inner.read_value(key).await?;
        if value.is_some() {
            self.note_counts(ReadSource::Value(key.to_owned()), 1, 8)?;
        }
        Ok(value)
    }

    async fn scan_page(
        &self,
        prefix: &str,
        cursor: Option<ScanCursor>,
        limit: usize,
    ) -> Result<ScanPage, DurabilityError> {
        let page = self.inner.scan_page(prefix, cursor, limit).await?;
        self.note(ReadSource::ScanPrefix(prefix.to_owned()), &page.entries)?;
        Ok(page)
    }

    async fn cas(&self, key: &str, old: u64, new: u64) -> Result<(), DurabilityError> {
        self.inner.cas(key, old, new).await
    }

    async fn flush(&self) -> Result<(), DurabilityError> {
        self.inner.flush().await
    }

    async fn truncate_before(
        &self,
        stream_key: &str,
        sequence: u64,
    ) -> Result<u64, DurabilityError> {
        self.inner.truncate_before(stream_key, sequence).await
    }

    async fn scan_routed(
        &self,
        route_key: &str,
        from: &[u8],
        to: &[u8],
    ) -> Result<Vec<(Vec<u8>, Vec<u8>)>, DurabilityError> {
        self.inner.scan_routed(route_key, from, to).await
    }

    async fn remove_routed_range(
        &self,
        route_key: &str,
        from: &[u8],
        to: &[u8],
    ) -> Result<u64, DurabilityError> {
        self.inner.remove_routed_range(route_key, from, to).await
    }

    fn store_counts(&self) -> Option<liminal::durability::StoreCountsHandle> {
        self.inner.store_counts()
    }
}

/// Two immutable images of the same store at different history lengths.
pub struct HistoryPair {
    images: std::path::PathBuf,
}

fn open(path: &Path) -> TestResult<Arc<dyn DurableStore>> {
    let store = HaematiteStore::new(Arc::new(EventStore::new(Database::open(
        path,
        haematite::RuntimeOptions {
            long_wait_interval: std::time::Duration::from_secs(5),
            reclaim_start: haematite::ReclaimStart::AtOpen,
        },
    )?)));
    Ok(Arc::new(store))
}

fn copy_directory(source: &Path, destination: &Path) -> TestResult<()> {
    fs::create_dir(destination)?;
    for entry in fs::read_dir(source)? {
        let entry = entry?;
        let target = destination.join(entry.file_name());
        let kind = entry.file_type()?;
        if kind.is_dir() {
            copy_directory(&entry.path(), &target)?;
        } else if kind.is_file() {
            fs::copy(entry.path(), &target)?;
            if fs::read(entry.path())? != fs::read(&target)? {
                return Err(format!("boot fixture copy differs: {}", target.display()).into());
            }
        } else {
            return Err("boot fixture contains a link or non-regular file".into());
        }
    }
    Ok(())
}

fn finish<T>(answer: TestResult<T>, cleanup: std::io::Result<()>) -> TestResult<T> {
    match (answer, cleanup) {
        (Ok(value), Ok(())) => Ok(value),
        (Err(error), Ok(())) => Err(error),
        (Ok(_), Err(error)) => Err(error.into()),
        (Err(error), Err(cleanup)) => {
            Err(format!("boot fixture work failed: {error}; cleanup failed: {cleanup}").into())
        }
    }
}

/// Builds one pair per run under an exclusive fixture lock.
///
/// The callback appends the supplied history range and maintains the fixture's
/// authoritative checkpoint and tail. It must preserve the live state being
/// tested while extending history. A different fixture shape needs a new name.
///
/// # Errors
/// Refuses an invalid name, absent run identity, wrong manifest, store or copy failure,
/// callback refusal, publication failure or cleanup failure.
pub fn history_pair(
    name: &str,
    append_history: impl Fn(&dyn DurableStore, Range<u64>) -> TestResult<()>,
) -> TestResult<HistoryPair> {
    if name.is_empty()
        || !name
            .bytes()
            .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'-')
    {
        return Err(
            "boot fixture name must contain only lowercase letters, digits and hyphens".into(),
        );
    }
    let run = std::env::var("NEXTEST_RUN_ID")?;
    if run.len() != 36
        || !run
            .bytes()
            .all(|byte| byte.is_ascii_hexdigit() || byte == b'-')
    {
        return Err("boot fixture run identity is not a nextest UUID".into());
    }
    let executable = std::env::current_exe()?;
    let root: PathBuf = executable
        .parent()
        .ok_or("boot fixture executable has no parent")?
        .join(format!("boot-history-fixtures-{run}"));
    fs::create_dir_all(&root)?;
    let images = root.join(format!("boot-history-{name}"));
    let lock = File::options()
        .create(true)
        .truncate(false)
        .read(true)
        .write(true)
        .open(root.join(format!("boot-history-{name}.lock")))?;
    let waiting = Instant::now();
    flock(&lock, FlockOperation::LockExclusive)?;
    println!("boot fixture {name}: lock wait {:?}", waiting.elapsed());
    if images.exists() {
        let manifest: [u64; 2] =
            serde_json::from_slice(&fs::read(images.join("history-rows.json"))?)?;
        if manifest != HISTORY_ROWS {
            return Err(
                "boot history fixture manifest differs from required history lengths".into(),
            );
        }
    } else {
        let building_started = Instant::now();
        let temporary = tempfile::Builder::new()
            .prefix("boot-history-build-")
            .tempdir_in(&root)?;
        let answer = (|| -> TestResult<()> {
            let building = temporary.path().join("images");
            fs::create_dir(&building)?;
            let small = building.join(HISTORY_ROWS[0].to_string());
            let database = Database::create(
                DatabaseConfig {
                    data_dir: small.clone(),
                    shard_count: 1,
                    distributed: None,
                    executor_threads: None,
                    node_cache_budget: Some(NodeCacheBudget::Unlimited),
                    retention: Retention::reclaim(0),
                },
                haematite::RuntimeOptions {
                    long_wait_interval: std::time::Duration::from_secs(5),
                    reclaim_start: haematite::ReclaimStart::AtOpen,
                },
            )?;
            let store = HaematiteStore::new(Arc::new(EventStore::new(database)));
            append_history(&store, 0..HISTORY_ROWS[0])?;
            block_on(store.flush())??;
            drop(store);
            let large = building.join(HISTORY_ROWS[1].to_string());
            copy_directory(&small, &large)?;
            let store = open(&large)?;
            append_history(store.as_ref(), HISTORY_ROWS[0]..HISTORY_ROWS[1])?;
            block_on(store.flush())??;
            drop(store);
            fs::write(
                building.join("history-rows.json"),
                serde_json::to_vec(&HISTORY_ROWS)?,
            )?;
            fs::rename(&building, &images)?;
            Ok(())
        })();
        finish(answer, temporary.close())?;
        println!(
            "boot fixture {name}: build {:?}",
            building_started.elapsed()
        );
    }
    drop(lock);
    Ok(HistoryPair { images })
}

impl HistoryPair {
    /// Restores each image once and pins both rows and bytes for this stage.
    ///
    /// The restore callback retains its state assertions. It cannot read the
    /// underlying store except through the counting wrapper it receives.
    ///
    /// # Errors
    /// Refuses an empty stage, unreadable fixture, failed restore or state
    /// assertion, cleanup failure, unequal history counts or a changed pin.
    pub(crate) fn assert_stage(
        &self,
        stage: &str,
        expected: &StageReads,
        restore: impl Fn(Arc<dyn DurableStore>) -> TestResult<()>,
    ) -> TestResult<[StageReads; 2]> {
        if stage.trim().is_empty() {
            return Err("boot stage name is empty".into());
        }
        let mut measured = [StageReads::new(), StageReads::new()];
        for (index, history) in HISTORY_ROWS.into_iter().enumerate() {
            let temporary = tempfile::Builder::new()
                .prefix("boot-history-restore-")
                .tempdir_in(self.images.parent().ok_or("boot fixture has no parent")?)?;
            let answer = (|| -> TestResult<StageReads> {
                let copying = Instant::now();
                let path = temporary.path().join("store");
                copy_directory(&self.images.join(history.to_string()), &path)?;
                let store = Arc::new(BootReadStore::new(open(&path)?));
                println!(
                    "boot stage {stage}, history {history}: copy/open {:?}",
                    copying.elapsed()
                );
                let restoring = Instant::now();
                let restored = during_stage("unassigned", || {
                    restore(Arc::clone(&store) as Arc<dyn DurableStore>)
                });
                println!(
                    "boot stage {stage}, history {history}: restore body {:?}",
                    restoring.elapsed()
                );
                let counts = store.counts();
                drop(store);
                restored?;
                counts
            })();
            let cleaning = Instant::now();
            measured[index] = finish(answer, temporary.close())?;
            println!(
                "boot stage {stage}, history {history}: cleanup {:?}; {:?}",
                cleaning.elapsed(),
                measured[index]
            );
        }
        if measured[0] != measured[1] || &measured[0] != expected {
            return Err(format!(
                "boot stage {stage}: expected {expected:?}; history {} read {:?}; history {} read {:?}",
                HISTORY_ROWS[0], measured[0], HISTORY_ROWS[1], measured[1]
            ).into());
        }
        Ok(measured)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn boot_history_production_constructor_ignores_unrelated_history() -> TestResult<()> {
        let pair = history_pair("empty-participant-v1", |store, range| {
            let expected = range.end;
            let start = range.start;
            let payloads = range.map(|row| row.to_be_bytes().to_vec()).collect();
            let next = block_on(store.append_all("boot-history:unrelated", payloads, start))??;
            assert_eq!(next, expected);
            Ok(())
        })?;
        // A store with no registry: the snapshot and the routed head are
        // absent routed values, the native head is absent, and the walk of an
        // install with no snapshot reads one empty page (DUR-019 R4). The
        // capacity snapshot is one more absent routed value (DUR-020 R3), and
        // no conversation is read: construction restores none (DUR-020 R1).
        let registry = "liminal:participant-conversation-registry";
        // The routed head key, framed as routed_stream_codec::head_key frames
        // it: prefix, the stream's length as a big-endian u32, the stream, 0.
        let mut head_key = b"liminal:routed-stream:v1:".to_vec();
        head_key.extend_from_slice(&u32::try_from(registry.len())?.to_be_bytes());
        head_key.extend_from_slice(registry.as_bytes());
        head_key.push(0);
        let routed = |key: &[u8]| {
            (
                ReadSource::RoutedValue {
                    route: registry.to_owned(),
                    key: key.to_vec(),
                },
                ReadCounts::default(),
            )
        };
        let expected = StageReads::from([(
            "registry".to_owned(),
            BTreeMap::from([
                routed(b"registry/snapshot/v1"),
                // DUR-020 R3: the clean-stop capacity snapshot, absent here.
                routed(b"registry/capacity/v1"),
                routed(&head_key),
                (
                    ReadSource::StreamHead(registry.to_owned()),
                    ReadCounts::default(),
                ),
                (
                    ReadSource::Stream(registry.to_owned()),
                    ReadCounts::default(),
                ),
            ]),
        )]);
        pair.assert_stage("participant", &expected, |store| {
            drop(boot_participant(
                store,
                crate::server::participant::test_participant_config(),
            )?);
            Ok(())
        })?;
        Ok(())
    }

    #[test]
    fn boot_history_reads_keep_streams_stages_and_scan_queries_distinct() -> TestResult<()> {
        let previous = BOOT_STAGE.with(Cell::get);
        let answer: Result<(), &str> = during_stage("registry", || {
            assert_eq!(BOOT_STAGE.with(Cell::get), "registry");
            during_stage("authority", || {
                assert_eq!(BOOT_STAGE.with(Cell::get), "authority");
                Err("named refusal")
            })
        });
        assert_eq!(answer, Err("named refusal"));
        assert_eq!(BOOT_STAGE.with(Cell::get), previous);

        let pair = history_pair("read-shape-v1", |store, range| {
            if range.start == 0 {
                block_on(store.append_all("boot-count:alpha", vec![vec![1, 2, 3]], 0))??;
                block_on(store.append_all("boot-count:beta", vec![vec![4, 5]], 0))??;
                block_on(store.cas("boot-count:head", 0, 7))??;
            }
            let expected = range.end;
            let start = range.start;
            let payloads = range.map(|row| row.to_be_bytes().to_vec()).collect();
            assert_eq!(
                block_on(store.append_all("boot-history:unrelated", payloads, start))??,
                expected,
            );
            Ok(())
        })?;
        let expected = StageReads::from([
            (
                "registry".to_owned(),
                BTreeMap::from([(
                    ReadSource::Stream("boot-count:alpha".to_owned()),
                    ReadCounts { rows: 1, bytes: 3 },
                )]),
            ),
            (
                "authority".to_owned(),
                BTreeMap::from([
                    (
                        ReadSource::Stream("boot-count:beta".to_owned()),
                        ReadCounts { rows: 1, bytes: 2 },
                    ),
                    (
                        ReadSource::Value("boot-count:head".to_owned()),
                        ReadCounts { rows: 1, bytes: 8 },
                    ),
                ]),
            ),
            (
                "scan".to_owned(),
                BTreeMap::from([(
                    ReadSource::ScanPrefix("boot-count:".to_owned()),
                    ReadCounts { rows: 2, bytes: 5 },
                )]),
            ),
        ]);
        pair.assert_stage("read-shape", &expected, |store| {
            during_stage("registry", || -> TestResult<()> {
                let rows = block_on(store.read_from("boot-count:alpha", 0, 1))??;
                assert_eq!(rows.len(), 1);
                assert_eq!(rows[0].payload, vec![1, 2, 3]);
                Ok(())
            })?;
            during_stage("authority", || -> TestResult<()> {
                let row = block_on(store.read_at("boot-count:beta", 0))??
                    .ok_or("point read did not return its fixture row")?;
                assert_eq!(row.payload, vec![4, 5]);
                assert_eq!(block_on(store.read_value("boot-count:head"))??, Some(7));
                Ok(())
            })?;
            during_stage("scan", || -> TestResult<()> {
                let page = block_on(store.scan_page("boot-count:", None, 64))??;
                assert_eq!(page.entries.len(), 2);
                assert!(page.next.is_none());
                Ok(())
            })?;
            Ok(())
        })?;
        Ok(())
    }

    #[test]
    fn boot_history_counts_head_and_raw_routed_keys_without_reading_rows() -> TestResult<()> {
        let inner = Arc::new(liminal::durability::open_ephemeral(1)?);
        let store = BootReadStore::new(inner);
        block_on(store.append_all("boot-count:head", vec![vec![1, 2, 3]], 0))??;
        let key = vec![0xff, 0];
        block_on(store.apply_routed_batch(
            "boot-count:route",
            vec![RoutedBatchItem {
                key: key.clone(),
                expected: None,
                value: vec![4, 5, 6],
                ttl: None,
            }],
        ))??;
        let other_key = vec![0xff, 1];
        block_on(store.apply_routed_batch(
            "boot-count:other",
            vec![RoutedBatchItem {
                key: other_key.clone(),
                expected: None,
                value: vec![7, 8],
                ttl: None,
            }],
        ))??;
        assert!(store.counts()?.is_empty());
        during_stage("head", || -> TestResult<()> {
            assert_eq!(
                block_on(store.read_stream_next_seq("boot-count:head"))??,
                Some(1)
            );
            assert_eq!(
                block_on(store.read_stream_next_seq("boot-count:head"))??,
                Some(1)
            );
            assert_eq!(
                block_on(store.read_stream_next_seq("boot-count:absent"))??,
                None
            );
            Ok(())
        })?;
        during_stage("checkpoint", || -> TestResult<()> {
            assert_eq!(
                block_on(store.get_routed("boot-count:route", &key))??,
                Some(vec![4, 5, 6])
            );
            assert_eq!(
                block_on(store.get_routed("boot-count:other", &other_key))??,
                Some(vec![7, 8])
            );
            assert_eq!(
                block_on(store.get_routed("boot-count:route", &[0xfe, 0]))??,
                None
            );
            Ok(())
        })?;
        assert_eq!(
            store.counts()?,
            StageReads::from([
                (
                    "head".to_owned(),
                    BTreeMap::from([
                        (
                            ReadSource::StreamHead("boot-count:head".to_owned()),
                            ReadCounts { rows: 2, bytes: 16 }
                        ),
                        (
                            ReadSource::StreamHead("boot-count:absent".to_owned()),
                            ReadCounts::default()
                        ),
                    ])
                ),
                (
                    "checkpoint".to_owned(),
                    BTreeMap::from([
                        (
                            ReadSource::RoutedValue {
                                route: "boot-count:route".to_owned(),
                                key
                            },
                            ReadCounts { rows: 1, bytes: 3 }
                        ),
                        (
                            ReadSource::RoutedValue {
                                route: "boot-count:other".to_owned(),
                                key: other_key
                            },
                            ReadCounts { rows: 1, bytes: 2 }
                        ),
                        (
                            ReadSource::RoutedValue {
                                route: "boot-count:route".to_owned(),
                                key: vec![0xfe, 0]
                            },
                            ReadCounts::default()
                        ),
                    ])
                ),
            ])
        );
        Ok(())
    }
}