liminal-rs 0.11.0

//! A write answered `OutcomeUnknown` that lands only after its read-back.
//!
//! The store had started the write when its caller stopped waiting, so it may
//! land at any moment after: here, just after the settle's read found nothing.
//! The resend then meets the caller's own row, which is applied, never a
//! conflict, and the stream holds the row once.
//!
//! The settle's other paths are pinned after it, on a store that answers each
//! append from a script: a write that had not landed is resent once; another
//! writer's row is the conflict, with nothing resent; a write that never
//! started is answered by name and not resent; and a resend that itself ends
//! unknown is answered by name after two writes, so the settle never loops.

use std::error::Error;
use std::sync::{Arc, Mutex};

use super::bridge::block_on;
use super::settle::{append_all_settled, append_settled, cas_settled};
use super::tests_memory_store::MemoryStore;
use super::{DurabilityError, DurableStore, ScanCursor, ScanPage, StoredEntry};

/// A write held back by the store, and the key it lands on.
#[derive(Debug)]
enum Held {
    Append {
        key: String,
        payloads: Vec<Vec<u8>>,
        expected: u64,
    },
    Cas {
        key: String,
        old: u64,
        new: u64,
    },
}

impl Held {
    fn key(&self) -> &str {
        match self {
            Self::Append { key, .. } | Self::Cas { key, .. } => key,
        }
    }
}

/// Answers the first write `OutcomeUnknown` without applying it, then applies
/// it right after the next read of its key has answered.
#[derive(Debug)]
struct LateStore {
    inner: Arc<dyn DurableStore>,
    armed: Mutex<bool>,
    held: Mutex<Option<Held>>,
}

impl LateStore {
    fn new() -> Self {
        Self {
            inner: Arc::new(MemoryStore::default()),
            armed: Mutex::new(true),
            held: Mutex::new(None),
        }
    }

    /// Whether this write is held back, holding it.
    fn holds(&self, held: Held) -> Result<bool, DurabilityError> {
        let mut armed = self.armed.lock().map_err(|_| poisoned())?;
        if !*armed {
            return Ok(false);
        }
        *armed = false;
        drop(armed);
        *self.held.lock().map_err(|_| poisoned())? = Some(held);
        Ok(true)
    }

    /// Lands the held write if it is on `key`, after a read of `key`.
    async fn land_after_read(&self, key: &str) -> Result<(), DurabilityError> {
        let held = {
            let mut held = self.held.lock().map_err(|_| poisoned())?;
            if held.as_ref().is_some_and(|held| held.key() == key) {
                held.take()
            } else {
                None
            }
        };
        match held {
            None => Ok(()),
            Some(Held::Append {
                key,
                payloads,
                expected,
            }) => self
                .inner
                .append_all(&key, payloads, expected)
                .await
                .map(drop),
            Some(Held::Cas { key, old, new }) => self.inner.cas(&key, old, new).await,
        }
    }
}

fn poisoned() -> DurabilityError {
    DurabilityError::ConfigError("the late store's lock was poisoned".to_owned())
}

fn outcome_unknown() -> DurabilityError {
    haematite::ApiError::Storage(haematite::DatabaseError::OutcomeUnknown { pid: 1 }).into()
}

#[async_trait::async_trait]
impl DurableStore for LateStore {
    async fn append(
        &self,
        stream_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        let held = Held::Append {
            key: stream_key.to_owned(),
            payloads: vec![payload.clone()],
            expected: expected_seq,
        };
        if self.holds(held)? {
            return Err(outcome_unknown());
        }
        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> {
        let held = Held::Append {
            key: stream_key.to_owned(),
            payloads: payloads.clone(),
            expected: expected_seq,
        };
        if self.holds(held)? {
            return Err(outcome_unknown());
        }
        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 read = self.inner.read_from(stream_key, offset, limit).await;
        self.land_after_read(stream_key).await?;
        read
    }

    async fn read_at(
        &self,
        stream_key: &str,
        sequence: u64,
    ) -> Result<Option<StoredEntry>, DurabilityError> {
        let read = self.inner.read_at(stream_key, sequence).await;
        self.land_after_read(stream_key).await?;
        read
    }

    async fn cas(&self, key: &str, old_value: u64, new_value: u64) -> Result<(), DurabilityError> {
        let held = Held::Cas {
            key: key.to_owned(),
            old: old_value,
            new: new_value,
        };
        if self.holds(held)? {
            return Err(outcome_unknown());
        }
        self.inner.cas(key, old_value, new_value).await
    }

    async fn read_value(&self, key: &str) -> Result<Option<u64>, DurabilityError> {
        let read = self.inner.read_value(key).await;
        self.land_after_read(key).await?;
        read
    }

    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
    }
}

const STREAM: &str = "settle-stream";

/// The rows of `STREAM`, read past the store's hold.
fn rows(store: &LateStore) -> Result<Vec<Vec<u8>>, Box<dyn Error>> {
    Ok(block_on(store.inner.read_from(STREAM, 0, 16))??
        .into_iter()
        .map(|entry| entry.payload)
        .collect())
}

#[test]
fn an_append_that_lands_after_its_read_back_is_settled_as_applied() -> Result<(), Box<dyn Error>> {
    let store = LateStore::new();
    let settled = block_on(append_settled(&store, STREAM, b"own".to_vec(), 0))?;
    let settled = settled.map_err(|error| error.to_string());
    assert_eq!(
        (settled, rows(&store)?),
        (Ok(0), vec![b"own".to_vec()]),
        "an append that landed after its read-back was not settled as applied, once"
    );
    Ok(())
}

#[test]
fn a_batch_that_lands_after_its_read_back_is_settled_as_applied() -> Result<(), Box<dyn Error>> {
    let store = LateStore::new();
    let batch = vec![b"one".to_vec(), b"two".to_vec()];
    let settled = block_on(append_all_settled(&store, STREAM, batch.clone(), 0))?;
    let settled = settled.map_err(|error| error.to_string());
    assert_eq!(
        (settled, rows(&store)?),
        (Ok(2), batch),
        "a batch that landed after its read-back was not settled as applied, once"
    );
    Ok(())
}

#[test]
fn a_swap_that_lands_after_its_read_back_is_settled_as_applied() -> Result<(), Box<dyn Error>> {
    let store = LateStore::new();
    let settled = block_on(cas_settled(&store, "settle-key", 0, 7))?;
    let settled = settled.map_err(|error| error.to_string());
    let stored = block_on(store.inner.read_value("settle-key"))??;
    assert_eq!(
        (settled, stored),
        (Ok(()), Some(7)),
        "a swap that landed after its read-back was not settled as applied"
    );
    Ok(())
}

/// What the store does with one append.
#[derive(Clone, Copy, Debug)]
enum Answer {
    /// Applies nothing and answers `OutcomeUnknown`.
    UnknownUnapplied,
    /// Applies nothing and answers `NotStarted`.
    NotStarted,
    /// Applies the append and answers as the store does.
    Pass,
}

/// Answers each append by the next scripted [`Answer`] (then [`Answer::Pass`]),
/// counting every append it is sent.
#[derive(Debug)]
struct ScriptedStore {
    inner: MemoryStore,
    script: Mutex<Vec<Answer>>,
    appends: Mutex<u64>,
}

impl ScriptedStore {
    fn new(script: &[Answer]) -> Self {
        let mut script = script.to_vec();
        script.reverse();
        Self {
            inner: MemoryStore::default(),
            script: Mutex::new(script),
            appends: Mutex::new(0),
        }
    }

    fn next(&self) -> Result<Answer, DurabilityError> {
        *self.appends.lock().map_err(|_| poisoned())? += 1;
        Ok(self
            .script
            .lock()
            .map_err(|_| poisoned())?
            .pop()
            .unwrap_or(Answer::Pass))
    }

    fn appends(&self) -> Result<u64, DurabilityError> {
        Ok(*self.appends.lock().map_err(|_| poisoned())?)
    }
}

fn not_started() -> DurabilityError {
    DurabilityError::NotStarted { pid: 1 }
}

fn named_unknown() -> DurabilityError {
    DurabilityError::OutcomeUnknown { pid: 1 }
}

#[async_trait::async_trait]
impl DurableStore for ScriptedStore {
    async fn append(
        &self,
        stream_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        match self.next()? {
            Answer::UnknownUnapplied => Err(named_unknown()),
            Answer::NotStarted => Err(not_started()),
            Answer::Pass => 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> {
        match self.next()? {
            Answer::UnknownUnapplied => Err(named_unknown()),
            Answer::NotStarted => Err(not_started()),
            Answer::Pass => {
                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> {
        self.inner.read_from(stream_key, offset, limit).await
    }

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

    async fn cas(&self, key: &str, old_value: u64, new_value: u64) -> Result<(), DurabilityError> {
        self.inner.cas(key, old_value, new_value).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
    }
}

/// The rows of `STREAM` on a scripted store.
fn scripted_rows(store: &ScriptedStore) -> Result<Vec<Vec<u8>>, Box<dyn Error>> {
    Ok(block_on(store.inner.read_from(STREAM, 0, 16))??
        .into_iter()
        .map(|entry| entry.payload)
        .collect())
}

/// Empty: an unknown outcome that had not landed is resent once, and the
/// resend's row is the only row.
#[test]
fn an_append_that_had_not_landed_is_resent_once() -> Result<(), Box<dyn Error>> {
    let store = ScriptedStore::new(&[Answer::UnknownUnapplied]);
    let settled = block_on(append_settled(&store, STREAM, b"own".to_vec(), 0))?;
    let settled = settled.map_err(|error| error.to_string());
    assert_eq!(
        (settled, store.appends()?, scripted_rows(&store)?),
        (Ok(0), 2, vec![b"own".to_vec()]),
        "an append that had not landed was not resent exactly once"
    );
    Ok(())
}

/// Other: another writer's row where the append was sent is the conflict the
/// append would have met, and nothing is resent.
#[test]
fn another_writers_row_where_the_append_was_sent_is_a_conflict() -> Result<(), Box<dyn Error>> {
    let store = ScriptedStore::new(&[Answer::Pass, Answer::UnknownUnapplied]);
    block_on(store.append(STREAM, b"other".to_vec(), 0))??;
    let settled = block_on(append_settled(&store, STREAM, b"own".to_vec(), 0))?;
    assert!(
        matches!(
            settled,
            Err(DurabilityError::SequenceConflict {
                expected: 0,
                actual: 1
            })
        ),
        "an append whose position holds another writer's row was answered {settled:?}"
    );
    assert_eq!(
        (store.appends()?, scripted_rows(&store)?),
        (2, vec![b"other".to_vec()]),
        "an append whose position holds another writer's row was resent"
    );
    Ok(())
}

/// `NotStarted`: nothing was written, so the error is the caller's to resend;
/// the settle neither reads nor resends it.
#[test]
fn a_write_that_never_started_is_answered_by_name_and_not_resent() -> Result<(), Box<dyn Error>> {
    let store = ScriptedStore::new(&[Answer::NotStarted]);
    let settled = block_on(append_settled(&store, STREAM, b"own".to_vec(), 0))?;
    assert!(
        matches!(settled, Err(DurabilityError::NotStarted { .. })),
        "a write that never started was answered {settled:?}"
    );
    assert_eq!(
        (store.appends()?, scripted_rows(&store)?),
        (1, Vec::<Vec<u8>>::new()),
        "a write that never started was resent by the settle"
    );
    Ok(())
}

/// A resend that itself ends unknown is answered by name after two writes:
/// the settle never loops.
#[test]
fn a_resend_that_ends_unknown_is_answered_by_name_after_two_writes() -> Result<(), Box<dyn Error>> {
    let store = ScriptedStore::new(&[Answer::UnknownUnapplied, Answer::UnknownUnapplied]);
    let settled = block_on(append_settled(&store, STREAM, b"own".to_vec(), 0))?;
    assert!(
        matches!(settled, Err(DurabilityError::OutcomeUnknown { .. })),
        "a resend that ended unknown was answered {settled:?}"
    );
    assert_eq!(
        store.appends()?,
        2,
        "a resend that ended unknown was sent other than twice"
    );
    Ok(())
}

/// The same two paths for a batch: one resend when it had not landed, and a
/// named answer after two writes when the resend ends unknown.
#[test]
fn a_batch_is_resent_once_and_never_loops() -> Result<(), Box<dyn Error>> {
    let batch = vec![b"one".to_vec(), b"two".to_vec()];
    let resent = ScriptedStore::new(&[Answer::UnknownUnapplied]);
    let settled = block_on(append_all_settled(&resent, STREAM, batch.clone(), 0))?;
    let settled = settled.map_err(|error| error.to_string());
    assert_eq!(
        (settled, resent.appends()?, scripted_rows(&resent)?),
        (Ok(2), 2, batch.clone()),
        "a batch that had not landed was not resent exactly once"
    );
    let unknown = ScriptedStore::new(&[Answer::UnknownUnapplied, Answer::UnknownUnapplied]);
    let settled = block_on(append_all_settled(&unknown, STREAM, batch, 0))?;
    assert!(
        matches!(settled, Err(DurabilityError::OutcomeUnknown { .. })),
        "a batch whose resend ended unknown was answered {settled:?}"
    );
    assert_eq!(
        unknown.appends()?,
        2,
        "a batch whose resend ended unknown was sent other than twice"
    );
    Ok(())
}