liminal-server 0.20.0

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

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

#[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,
        route_key: &str,
        items: Vec<liminal::durability::RoutedBatchItem>,
    ) -> Result<(), liminal::durability::DurabilityError> {
        self.inner.apply_routed_batch(route_key, items).await
    }

    async fn get_routed(
        &self,
        route_key: &str,
        key: &[u8],
    ) -> Result<Option<Vec<u8>>, liminal::durability::DurabilityError> {
        self.inner.get_routed(route_key, key).await
    }

    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 deferred = DeferredStore::new(Box::new(|| open_ephemeral(1)));
    let probe = SecondAppendFault {
        inner: &deferred,
        single_calls: AtomicUsize::new(0),
    };
    let result = block_on(probe.append_all(
        "DeferredStore",
        vec![b"first".to_vec(), b"second".to_vec()],
        0,
    ))?;
    let rows = block_on(probe.read_from("DeferredStore", 0, 3))??;
    let prefix = rows
        .into_iter()
        .map(|row| String::from_utf8(row.payload))
        .collect::<Result<Vec<_>, _>>()?;
    assert_eq!(
        (
            result.is_ok(),
            prefix,
            probe.single_calls.load(Ordering::Relaxed)
        ),
        (true, vec!["first".to_owned(), "second".to_owned()], 0),
        "DeferredStore exposed a prefix after its second single append failed"
    );
    Ok(())
}