liminal-rs 0.16.3

use std::error::Error;
use std::sync::atomic::{AtomicUsize, Ordering};

use super::{DurabilityError, DurableStore, ScanCursor, ScanPage, StoredEntry, open_ephemeral};
use crate::durability::CountingStore;
use crate::durability::bridge::block_on;

#[derive(Debug)]
struct SecondAppendFault<'a> {
    inner: &'a dyn DurableStore,
    single_calls: AtomicUsize,
}

#[async_trait::async_trait]
impl DurableStore for SecondAppendFault<'_> {
    async fn apply_routed_batch(
        &self,
        _: &str,
        _: Vec<crate::durability::RoutedBatchItem>,
    ) -> Result<(), crate::durability::DurabilityError> {
        Err(crate::durability::DurabilityError::BatchUnsupported {
            store: std::any::type_name::<Self>(),
        })
    }

    async fn get_routed(
        &self,
        _: &str,
        _: &[u8],
    ) -> Result<Option<Vec<u8>>, crate::durability::DurabilityError> {
        Err(crate::durability::DurabilityError::BatchUnsupported {
            store: std::any::type_name::<Self>(),
        })
    }

    async fn append(
        &self,
        stream: &str,
        payload: Vec<u8>,
        expected: u64,
    ) -> Result<u64, DurabilityError> {
        if self.single_calls.fetch_add(1, Ordering::Relaxed) == 1 {
            return Err(DurabilityError::ConfigError(
                "second single append refused".into(),
            ));
        }
        self.inner.append(stream, payload, expected).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: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Vec<StoredEntry>, DurabilityError> {
        self.inner.read_from(stream, offset, limit).await
    }

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

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

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

    async fn scan_page(
        &self,
        prefix: &str,
        cursor: Option<ScanCursor>,
        limit: usize,
    ) -> Result<ScanPage, DurabilityError> {
        self.inner.scan_page(prefix, cursor, limit).await
    }

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

#[test]
fn no_admitted_batch_adapter_exposes_a_prefix() -> Result<(), Box<dyn Error>> {
    let ephemeral = open_ephemeral(1)?;
    let haematite = ephemeral.store()?.clone();
    let counted = CountingStore::new(haematite.clone());
    let payloads = vec![b"first".to_vec(), b"second".to_vec()];
    let mut observed = Vec::new();
    let mut wanted = Vec::new();
    for (name, store) in [
        ("HaematiteStore", &haematite as &dyn DurableStore),
        ("EphemeralHaematiteStore", &ephemeral as &dyn DurableStore),
        ("CountingStore", &counted as &dyn DurableStore),
    ] {
        let probe = SecondAppendFault {
            inner: store,
            single_calls: AtomicUsize::new(0),
        };
        let result = block_on(probe.append_all(name, payloads.clone(), 0))?;
        let rows = block_on(probe.read_from(name, 0, 3))??;
        let prefix = rows
            .into_iter()
            .map(|row| String::from_utf8(row.payload))
            .collect::<Result<Vec<_>, _>>()?;
        observed.push((
            name,
            result.is_ok(),
            prefix,
            probe.single_calls.load(Ordering::Relaxed),
        ));
        wanted.push((name, true, vec!["first".to_owned(), "second".to_owned()], 0));
    }
    assert_eq!(
        observed, wanted,
        "an admitted batch adapter exposed a prefix after its second single append failed"
    );
    Ok(())
}