liminal-rs 0.11.0

//! 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};

/// 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, settling an unknown outcome by flushing once more: appends and
/// swaps are committed as they are made, so a second flush only repeats the
/// barrier.
///
/// # Errors
///
/// The second flush's error.
pub async fn flush_settled(store: &dyn DurableStore) -> Result<(), DurabilityError> {
    match store.flush().await {
        Err(DurabilityError::OutcomeUnknown { .. }) => store.flush().await,
        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.
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}"
            ))
        })
}