liminal-rs 0.16.2

//! Writes that settle an unknown outcome by reading the position back.
//!
//! A store answers [`DurabilityError::OutcomeUnknown`] for a write it had
//! started when its caller stopped waiting: the write may have been applied.
//! It is never "not written". Each helper here performs one write and, on that
//! answer, reads the position the write targeted, as
//! [`super::channel`]'s pending append is resolved: the caller's own bytes
//! standing there mean the write landed and its result is returned; nothing
//! standing there means it had not landed yet, and the write is sent once
//! more; anything else is the conflict it would have met. The first write may
//! still land after that read, so a resend that meets a conflict reads the
//! position once more, and the caller's own bytes there mean applied, never a
//! conflict. When a read fails, the original
//! [`DurabilityError::OutcomeUnknown`] is returned, so the caller still meets
//! it by name.
//!
//! Every write the store offers is guarded by the position it expects (a
//! sequence, or a swap's old value), so a resend can never duplicate a row:
//! the guard refuses it once the first write stands.

use super::{DurabilityError, DurableStore, RoutedBatchEntry, RoutedBatchItem};

/// Appends `payload` at `expected_seq`, settling an unknown outcome.
///
/// # Errors
///
/// The append's error; [`DurabilityError::OutcomeUnknown`] when the position
/// cannot be read back; [`DurabilityError::SequenceConflict`] when another row
/// stands where the append was sent.
pub async fn append_settled(
    store: &dyn DurableStore,
    stream_key: &str,
    payload: Vec<u8>,
    expected_seq: u64,
) -> Result<u64, DurabilityError> {
    match store
        .append(stream_key, payload.clone(), expected_seq)
        .await
    {
        Err(unknown @ DurabilityError::OutcomeUnknown { .. }) => {
            match own_row(store, stream_key, expected_seq, &payload).await {
                Err(read) => Err(unread(unknown, stream_key, &read)),
                Ok(Standing::Own) => Ok(expected_seq),
                Ok(Standing::Other) => Err(conflict(expected_seq)),
                Ok(Standing::Empty) => {
                    match store
                        .append(stream_key, payload.clone(), expected_seq)
                        .await
                    {
                        Err(resent @ DurabilityError::SequenceConflict { .. }) => {
                            match own_row(store, stream_key, expected_seq, &payload).await {
                                Ok(Standing::Own) => Ok(expected_seq),
                                Ok(_) => Err(resent),
                                Err(read) => Err(unread(unknown, stream_key, &read)),
                            }
                        }
                        outcome => outcome,
                    }
                }
            }
        }
        outcome => outcome,
    }
}

/// Appends `payloads` from `expected_seq` in one write, settling an unknown
/// outcome. Answers the sequence after the last row.
///
/// # Errors
///
/// As [`append_settled`].
pub async fn append_all_settled(
    store: &dyn DurableStore,
    stream_key: &str,
    payloads: Vec<Vec<u8>>,
    expected_seq: u64,
) -> Result<u64, DurabilityError> {
    match store
        .append_all(stream_key, payloads.clone(), expected_seq)
        .await
    {
        Err(unknown @ DurabilityError::OutcomeUnknown { .. }) => {
            let after = after_batch(stream_key, expected_seq, payloads.len())?;
            match own_rows(store, stream_key, expected_seq, &payloads).await {
                Err(read) => Err(unread(unknown, stream_key, &read)),
                Ok(Standing::Own) => Ok(after),
                Ok(Standing::Other) => Err(conflict(expected_seq)),
                Ok(Standing::Empty) => {
                    match store
                        .append_all(stream_key, payloads.clone(), expected_seq)
                        .await
                    {
                        Err(resent @ DurabilityError::SequenceConflict { .. }) => {
                            match own_rows(store, stream_key, expected_seq, &payloads).await {
                                Ok(Standing::Own) => Ok(after),
                                Ok(_) => Err(resent),
                                Err(read) => Err(unread(unknown, stream_key, &read)),
                            }
                        }
                        outcome => outcome,
                    }
                }
            }
        }
        outcome => outcome,
    }
}

/// Swaps `key` from `old_value` to `new_value`, settling an unknown outcome.
///
/// `new_value` stored means the swap landed; `old_value` (or nothing, for an
/// `old_value` of 0) means it did not, and it runs once more.
///
/// # Errors
///
/// The swap's error; [`DurabilityError::OutcomeUnknown`] when the value
/// cannot be read back; [`DurabilityError::CursorRegression`] when another
/// value stands there.
pub async fn cas_settled(
    store: &dyn DurableStore,
    key: &str,
    old_value: u64,
    new_value: u64,
) -> Result<(), DurabilityError> {
    match store.cas(key, old_value, new_value).await {
        Err(unknown @ DurabilityError::OutcomeUnknown { .. }) => {
            let stored = match store.read_value(key).await {
                Ok(stored) => stored.unwrap_or(0),
                Err(read) => return Err(unread(unknown, key, &read)),
            };
            if stored == new_value {
                return Ok(());
            }
            if stored != old_value {
                return Err(DurabilityError::CursorRegression {
                    stored,
                    attempted: old_value,
                });
            }
            match store.cas(key, old_value, new_value).await {
                Err(resent @ DurabilityError::CursorRegression { .. }) => {
                    match store.read_value(key).await {
                        Ok(stored) if stored.unwrap_or(0) == new_value => Ok(()),
                        Ok(_) => Err(resent),
                        Err(read) => Err(unread(unknown, key, &read)),
                    }
                }
                outcome => outcome,
            }
        }
        outcome => outcome,
    }
}

/// Flushes once, preserving any failure as an unconfirmed durability barrier.
///
/// # Errors
///
/// Returns the store's original flush failure without retrying the barrier.
pub async fn flush_settled(store: &dyn DurableStore) -> Result<(), DurabilityError> {
    store.flush().await
}

/// Applies one routed batch and settles an unknown outcome by reading every
/// key the batch targeted.
///
/// The caller's own bytes at every key mean the batch landed; every key still
/// at the state the batch expected means it had not landed yet, and the batch
/// is sent once more; anything else is the conflict it would have met. A
/// resend that meets the batch's own hash guard reads once more, and the
/// caller's own bytes there mean applied.
///
/// # Errors
/// Returns the store's refusal unchanged, the original unknown outcome when
/// the read-back fails or the keys stand mixed, and the guard's refusal when a
/// resend meets it over foreign bytes.
pub async fn apply_routed_batch_settled(
    store: &dyn DurableStore,
    route_key: &str,
    items: Vec<RoutedBatchItem>,
) -> Result<(), DurabilityError> {
    // An empty batch writes nothing and has no outcome to settle, so it never
    // costs the store a call (the self-stream drainer's counts pin zero).
    if items.is_empty() {
        return Ok(());
    }
    match store.apply_routed_batch(route_key, items.clone()).await {
        Err(unknown @ DurabilityError::OutcomeUnknown { .. }) => {
            match own_keys(store, route_key, &items).await {
                Err(read) => Err(unread(unknown, route_key, &read)),
                Ok(Standing::Own) => Ok(()),
                Ok(Standing::Other) => Err(unread(
                    unknown,
                    route_key,
                    &DurabilityError::ConfigError(
                        "routed batch keys stand neither as written nor as expected".to_owned(),
                    ),
                )),
                Ok(Standing::Empty) => {
                    match store.apply_routed_batch(route_key, items.clone()).await {
                        Err(resent @ DurabilityError::ConfigError(_)) => {
                            match own_keys(store, route_key, &items).await {
                                Ok(Standing::Own) => Ok(()),
                                Ok(_) => Err(resent),
                                Err(read) => Err(unread(unknown, route_key, &read)),
                            }
                        }
                        outcome => outcome,
                    }
                }
            }
        }
        outcome => outcome,
    }
}

/// Applies one routed batch of puts and deletes and settles an unknown
/// outcome by reading every key the batch targeted (DUR-028 R3).
///
/// As [`apply_routed_batch_settled`]: the batch landed when every put reads
/// as the caller's own bytes and every deleted key reads absent; it had not
/// landed when every key still stands as the batch expected (a put's guard,
/// a delete's hash), and it is sent once more; anything else is the
/// conflict it would have met.
///
/// # Errors
/// Returns the store's refusal unchanged, the original unknown outcome when
/// the read-back fails or the keys stand mixed, and the guard's refusal when
/// a resend meets it over foreign bytes.
pub async fn apply_routed_entries_settled(
    store: &dyn DurableStore,
    route_key: &str,
    entries: Vec<RoutedBatchEntry>,
) -> Result<(), DurabilityError> {
    if entries.is_empty() {
        return Ok(());
    }
    match store.apply_routed_entries(route_key, entries.clone()).await {
        Err(unknown @ DurabilityError::OutcomeUnknown { .. }) => {
            match own_entries(store, route_key, &entries).await {
                Err(read) => Err(unread(unknown, route_key, &read)),
                Ok(Standing::Own) => Ok(()),
                Ok(Standing::Other) => Err(unread(
                    unknown,
                    route_key,
                    &DurabilityError::ConfigError(
                        "routed batch keys stand neither as written nor as expected".to_owned(),
                    ),
                )),
                Ok(Standing::Empty) => {
                    match store.apply_routed_entries(route_key, entries.clone()).await {
                        Err(
                            resent @ (DurabilityError::ConfigError(_)
                            | DurabilityError::StoreError(_)),
                        ) => match own_entries(store, route_key, &entries).await {
                            Ok(Standing::Own) => Ok(()),
                            Ok(_) => Err(resent),
                            Err(read) => Err(unread(unknown, route_key, &read)),
                        },
                        outcome => outcome,
                    }
                }
            }
        }
        outcome => outcome,
    }
}

/// The conflict an append sent to `expected_seq` meets when another row
/// stands there.
const fn conflict(expected_seq: u64) -> DurabilityError {
    DurabilityError::SequenceConflict {
        expected: expected_seq,
        actual: expected_seq.saturating_add(1),
    }
}

/// The unknown outcome a failed read-back leaves standing, with the read's
/// failure warned beside it.
fn unread(unknown: DurabilityError, key: &str, read: &DurabilityError) -> DurabilityError {
    tracing::warn!(
        key,
        %read,
        "a write's unknown outcome could not be read back; it stays unknown"
    );
    unknown
}

/// What stands where a write was sent.
enum Standing {
    /// The caller's own bytes: the write landed.
    Own,
    /// Nothing: the write has not landed.
    Empty,
    /// Another writer's row.
    Other,
}

/// What stands at `sequence` against `payload`.
async fn own_row(
    store: &dyn DurableStore,
    stream_key: &str,
    sequence: u64,
    payload: &[u8],
) -> Result<Standing, DurabilityError> {
    Ok(match store.read_at(stream_key, sequence).await? {
        None => Standing::Empty,
        Some(entry) if entry.payload == payload => Standing::Own,
        Some(_) => Standing::Other,
    })
}

/// What stands from `expected_seq` against `payloads`, as one batch: a batch
/// lands whole or not at all.
/// How every key of a routed batch stands against the bytes the batch wrote
/// there: all the caller's own bytes, all still at the state the batch
/// expected (absent, or hashing to the guard), or mixed.
async fn own_keys(
    store: &dyn DurableStore,
    route_key: &str,
    items: &[RoutedBatchItem],
) -> Result<Standing, DurabilityError> {
    let mut own = 0_usize;
    let mut expected = 0_usize;
    for item in items {
        let stored = store.get_routed(route_key, &item.key).await?;
        match stored {
            Some(bytes) if bytes == item.value => own += 1,
            None if item.expected.is_none() => expected += 1,
            Some(bytes) if item.expected == Some(haematite::tree::Hash::of(&bytes)) => {
                expected += 1;
            }
            _ => {}
        }
    }
    Ok(if own == items.len() {
        Standing::Own
    } else if expected == items.len() {
        Standing::Empty
    } else {
        Standing::Other
    })
}

/// How every key of a routed batch of entries stands against it: a put as
/// [`own_keys`] reads one, a delete as its own when the key is absent and
/// as expected when the value still hashes to the delete's guard.
async fn own_entries(
    store: &dyn DurableStore,
    route_key: &str,
    entries: &[RoutedBatchEntry],
) -> Result<Standing, DurabilityError> {
    let mut own = 0_usize;
    let mut expected = 0_usize;
    for entry in entries {
        let stored = store.get_routed(route_key, entry.key()).await?;
        match (entry, stored) {
            (RoutedBatchEntry::Put(item), Some(bytes)) if bytes == item.value => own += 1,
            (RoutedBatchEntry::Put(item), None) if item.expected.is_none() => expected += 1,
            (RoutedBatchEntry::Put(item), Some(bytes))
                if item.expected == Some(haematite::tree::Hash::of(&bytes)) =>
            {
                expected += 1;
            }
            (RoutedBatchEntry::Delete { .. }, None) => own += 1,
            (
                RoutedBatchEntry::Delete {
                    expected: guard, ..
                },
                Some(bytes),
            ) if *guard == haematite::tree::Hash::of(&bytes) => {
                expected += 1;
            }
            _ => {}
        }
    }
    Ok(if own == entries.len() {
        Standing::Own
    } else if expected == entries.len() {
        Standing::Empty
    } else {
        Standing::Other
    })
}

async fn own_rows(
    store: &dyn DurableStore,
    stream_key: &str,
    expected_seq: u64,
    payloads: &[Vec<u8>],
) -> Result<Standing, DurabilityError> {
    let standing = store
        .read_from(stream_key, expected_seq, payloads.len())
        .await?;
    if standing.is_empty() {
        return Ok(Standing::Empty);
    }
    let own =
        standing.len() == payloads.len()
            && standing.iter().zip(payloads).zip(expected_seq..).all(
                |((entry, payload), sequence)| {
                    entry.sequence == sequence && entry.payload == *payload
                },
            );
    Ok(if own { Standing::Own } else { Standing::Other })
}

/// The sequence after a batch of `rows` from `expected_seq`.
fn after_batch(stream_key: &str, expected_seq: u64, rows: usize) -> Result<u64, DurabilityError> {
    u64::try_from(rows)
        .ok()
        .and_then(|rows| expected_seq.checked_add(rows))
        .ok_or_else(|| {
            DurabilityError::ConfigError(format!(
                "{stream_key}: sequence overflow settling a batch at {expected_seq}"
            ))
        })
}