liminal-rs 0.14.0

//! The recorded head of a long dedup key stream (`dedup/head.rs`): written
//! forward only with lost races counted, verified before use with refusals
//! counted, a refused head repaired by an equality swap that a racing forward
//! writer wins, and never needed for a lookup to answer on a store that
//! refuses the head's write.
//!
//! Every counter here is read through [`dedup_head_counts`], the accessor a
//! running process reads.

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};

use super::codec::DedupRecord;
use super::{DedupCache, DedupDecision, DedupEntry};
use crate::durability::bridge::block_on;
use crate::durability::head_checkpoint::READ_BATCH_SIZE;
use crate::durability::{
    CountingStore, DurabilityError, DurableStore, ProcessingReceipt, ScanCursor, ScanPage,
    StoredEntry, dedup_head_counts, head_checkpoint_key, open_ephemeral,
};

type TestResult = Result<(), Box<dyn std::error::Error>>;

const NAMESPACE: &str = "a12-head";

/// Claim-and-release rounds in each long key: more rows than one page.
const ROUNDS: u64 = 600;

/// A row another writer appends just before this store's next `cas`: its
/// stream, its payload and the sequence it lands at.
type InterloperRow = (String, Vec<u8>, u64);

/// A store passing every call to `inner`, except that its `cas` can first let
/// another writer append a row and record a head, or refuse outright as a
/// store opened without write access would.
#[derive(Debug)]
struct InterposedStore {
    inner: Arc<dyn DurableStore>,
    /// A head another writer records just before this store's next `cas`,
    /// expecting the same value that `cas` expects; zero for none.
    interloper: AtomicU64,
    /// A row another writer appends just before this store's next `cas`.
    interloper_row: Mutex<Option<InterloperRow>>,
    refuse_cas: bool,
    cas_calls: AtomicU64,
}

impl InterposedStore {
    fn new(inner: Arc<dyn DurableStore>, refuse_cas: bool) -> Self {
        Self {
            inner,
            interloper: AtomicU64::new(0),
            interloper_row: Mutex::new(None),
            refuse_cas,
            cas_calls: AtomicU64::new(0),
        }
    }
}

#[async_trait::async_trait]
impl DurableStore for InterposedStore {
    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_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.inner.append(stream_key, payload, 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 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_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.cas_calls.fetch_add(1, Ordering::Relaxed);
        if self.refuse_cas {
            return Err(DurabilityError::ConfigError(
                "this store was opened without write access".to_owned(),
            ));
        }
        let row = self
            .interloper_row
            .lock()
            .map_err(|_| DurabilityError::ConfigError("interloper row lock poisoned".to_owned()))?
            .take();
        if let Some((stream_key, payload, sequence)) = row {
            self.inner.append(&stream_key, payload, sequence).await?;
        }
        let interloper = self.interloper.swap(0, Ordering::Relaxed);
        if interloper > 0 {
            self.inner.cas(key, old_value, interloper).await?;
        }
        self.inner.cas(key, old_value, new_value).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
    }
}

/// Writes [`ROUNDS`] claim-and-release rounds of `key`, then `last`, to
/// `key`'s stream in one append, answering the stream's row count.
fn write_long_history(
    store: &dyn DurableStore,
    cache: &DedupCache,
    key: &str,
    last: &DedupRecord,
) -> Result<u64, Box<dyn std::error::Error>> {
    let claim = DedupEntry::new(key, None, 1).serialize()?;
    let release = DedupRecord::tombstone(key.to_owned(), 1).serialize()?;
    let mut rows = Vec::new();
    for _ in 0..ROUNDS {
        rows.push(claim.clone());
        rows.push(release.clone());
    }
    rows.push(last.serialize()?);
    Ok(block_on(store.append_all(
        &cache.stream_key_for(key),
        rows,
        0,
    ))??)
}

fn open_claim(key: &str) -> DedupRecord {
    DedupRecord::Active(DedupEntry::new(key, None, 1))
}

/// Two readers record the head of the same long key. The second writer's
/// record lands just before the first one's, at the same head: the first
/// finds it at its own and counts a lost race. Both lookups answer the same,
/// and the head stays where it was recorded.
#[test]
fn a_lost_head_race_is_counted_and_the_head_never_moves_back() -> TestResult {
    let inner: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let writer = DedupCache::new(Arc::clone(&inner), NAMESPACE);
    let rows = write_long_history(inner.as_ref(), &writer, "key", &open_claim("key"))?;
    let head = head_checkpoint_key(&writer.stream_key_for("key"));
    let racing = Arc::new(InterposedStore::new(Arc::clone(&inner), false));
    racing.interloper.store(rows, Ordering::Relaxed);

    let lost = dedup_head_counts().record_lost_races;
    let first = DedupCache::new(Arc::clone(&racing) as Arc<dyn DurableStore>, NAMESPACE);
    let first_found = block_on(first.lookup("key"))??;
    assert_eq!(
        dedup_head_counts().record_lost_races,
        lost + 1,
        "the lost race is counted"
    );
    assert_eq!(block_on(inner.read_value(&head))??, Some(rows));

    let second = DedupCache::new(Arc::clone(&inner), NAMESPACE);
    let second_found = block_on(second.lookup("key"))??;
    assert_eq!(first_found, Some(DedupDecision::InFlight));
    assert_eq!(second_found, first_found);
    assert_eq!(block_on(inner.read_value(&head))??, Some(rows));
    assert_eq!(dedup_head_counts().record_lost_races, lost + 1);
    Ok(())
}

/// DUR-008 A12 ruling 1, pin (a). A head planted past the stream's end is
/// refused, counted through the production accessor, and repaired: the first
/// lookup walks the stream and replaces the head with the walk's result. The
/// second lookup reads its first page and the head's row, and walks nothing:
/// its reads drop from the whole stream to one page.
#[test]
fn a_head_past_the_end_is_repaired_and_the_next_lookup_does_not_walk() -> TestResult {
    let counting = Arc::new(CountingStore::new(open_ephemeral(1)?));
    let counts = counting.handle();
    let store: Arc<dyn DurableStore> = counting;
    let cache = DedupCache::new(Arc::clone(&store), NAMESPACE);
    let rows = write_long_history(store.as_ref(), &cache, "key", &open_claim("key"))?;
    let head = head_checkpoint_key(&cache.stream_key_for("key"));
    let planted = rows + 1_000;
    block_on(store.cas(&head, 0, planted))??;

    let heads = dedup_head_counts();
    let before = counts.snapshot();
    assert_eq!(
        block_on(cache.lookup("key"))??,
        Some(DedupDecision::InFlight)
    );
    let first = counts.snapshot().since(&before);
    let repaired = dedup_head_counts();
    assert_eq!(
        repaired.refusals,
        heads.refusals + 1,
        "the refusal is counted"
    );
    assert_eq!(repaired.repairs, heads.repairs + 1, "the repair is counted");
    assert_eq!(
        first.read_from_rows, rows,
        "the first lookup walks: {first}"
    );
    assert_eq!(first.cas_calls, 1, "{first}");
    assert_eq!(
        block_on(store.read_value(&head))??,
        Some(rows),
        "the refused head is replaced by the walk's result"
    );

    let before = counts.snapshot();
    assert_eq!(
        block_on(cache.lookup("key"))??,
        Some(DedupDecision::InFlight)
    );
    let second = counts.snapshot().since(&before);
    assert_eq!(
        second.read_from_rows,
        u64::try_from(READ_BATCH_SIZE)?,
        "the second lookup reads its first page and nothing past the head: {second}"
    );
    assert_eq!(second.read_from_calls, 2, "{second}");
    assert_eq!(second.read_at_rows, 1, "the head's row verifies: {second}");
    assert_eq!(second.cas_calls, 0, "{second}");
    assert_eq!(dedup_head_counts(), repaired, "nothing refused or repaired");
    Ok(())
}

/// DUR-008 A12 ruling 1, pin (b). A repair racing a forward writer leaves the
/// forward value. Just before the repair's swap, another writer appends a row
/// of the key and records the larger, valid head past it, from the same
/// refused head. The repair's equality swap fails, is counted as a lost race
/// and not as a repair, and the stored head is the other writer's.
#[test]
fn a_repair_racing_a_forward_writer_leaves_the_forward_head() -> TestResult {
    let inner: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let writer = DedupCache::new(Arc::clone(&inner), NAMESPACE);
    let rows = write_long_history(inner.as_ref(), &writer, "key", &open_claim("key"))?;
    let stream = writer.stream_key_for("key");
    let head = head_checkpoint_key(&stream);
    block_on(inner.cas(&head, 0, rows + 1_000))??;
    let forward = rows + 1;
    let racing = Arc::new(InterposedStore::new(Arc::clone(&inner), false));
    *racing
        .interloper_row
        .lock()
        .map_err(|_| "interloper row lock poisoned")? =
        Some((stream, open_claim("key").serialize()?, rows));
    racing.interloper.store(forward, Ordering::Relaxed);

    let heads = dedup_head_counts();
    let cache = DedupCache::new(Arc::clone(&racing) as Arc<dyn DurableStore>, NAMESPACE);
    assert_eq!(
        block_on(cache.lookup("key"))??,
        Some(DedupDecision::InFlight)
    );
    let raced = dedup_head_counts();
    assert_eq!(raced.refusals, heads.refusals + 1);
    assert_eq!(
        raced.record_lost_races,
        heads.record_lost_races + 1,
        "the failed repair is counted as a lost race"
    );
    assert_eq!(raced.repairs, heads.repairs, "no repair landed");
    assert_eq!(racing.cas_calls.load(Ordering::Relaxed), 1, "not retried");
    assert_eq!(
        block_on(inner.read_value(&head))??,
        Some(forward),
        "the forward writer's head stands"
    );

    let reader = DedupCache::new(Arc::clone(&inner), NAMESPACE);
    assert_eq!(
        block_on(reader.lookup("key"))??,
        Some(DedupDecision::InFlight)
    );
    assert_eq!(dedup_head_counts(), raced, "the forward head verifies");
    Ok(())
}

/// A head whose row holds another key's record is refused and counted. The
/// walk that follows finds the other key's record in the stream and answers
/// the collision, as a full read of the stream does.
#[test]
fn a_head_naming_another_keys_record_is_refused() -> TestResult {
    let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let cache = DedupCache::new(Arc::clone(&store), NAMESPACE);
    let rows = write_long_history(store.as_ref(), &cache, "key", &open_claim("other"))?;
    let head = head_checkpoint_key(&cache.stream_key_for("key"));
    block_on(store.cas(&head, 0, rows))??;

    let before = dedup_head_counts();
    let found = block_on(cache.lookup("key"))?;
    assert!(
        matches!(found, Err(DurabilityError::DedupCollision { ref key }) if key == "key"),
        "{found:?}"
    );
    let after = dedup_head_counts();
    assert_eq!(after.refusals, before.refusals + 1);
    assert_eq!(after.repairs, before.repairs, "a collision repairs nothing");
    assert_eq!(block_on(store.read_value(&head))??, Some(rows));
    Ok(())
}

/// No dedup cache is opened over a read-only store today; a store whose `cas`
/// refuses stands in for one. A long key's lookup over it records no head and
/// still answers, and claims over it still complete.
#[test]
fn a_store_that_refuses_the_head_record_still_answers() -> TestResult {
    let inner: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let writer = DedupCache::new(Arc::clone(&inner), NAMESPACE);
    write_long_history(inner.as_ref(), &writer, "key", &open_claim("key"))?;
    let head = head_checkpoint_key(&writer.stream_key_for("key"));
    let refusing = Arc::new(InterposedStore::new(Arc::clone(&inner), true));
    let cache = DedupCache::new(Arc::clone(&refusing) as Arc<dyn DurableStore>, NAMESPACE);

    assert_eq!(
        block_on(cache.lookup("key"))??,
        Some(DedupDecision::InFlight)
    );
    assert_eq!(block_on(inner.read_value(&head))??, None, "no head landed");
    assert_eq!(refusing.cas_calls.load(Ordering::Relaxed), 1);

    block_on(cache.complete_receipt("key", ProcessingReceipt::for_position(1)))??;
    assert!(matches!(
        block_on(cache.lookup("key"))??,
        Some(DedupDecision::Completed(_))
    ));
    assert_eq!(block_on(inner.read_value(&head))??, None);
    Ok(())
}