liminal-rs 0.14.1

//! Exact append evidence survives cancellation until an authoritative read settles its outcome.
use std::fmt;

use super::DurableChannel;
use crate::durability::DurabilityError;

/// A publish sent to storage whose outcome has not yet been established.
#[derive(Clone)]
pub struct UnresolvedAppend {
    channel: String,
    partition: usize,
    sequence: u64,
    payload: Vec<u8>,
    idempotency_key: Option<String>,
    cause: String,
}

impl UnresolvedAppend {
    pub(crate) fn new(
        channel: String,
        partition: usize,
        sequence: u64,
        payload: Vec<u8>,
        idempotency_key: Option<String>,
    ) -> Self {
        Self {
            channel,
            partition,
            sequence,
            payload,
            idempotency_key,
            cause: "append future did not return an outcome".to_owned(),
        }
    }

    /// The channel that owns this append.
    #[must_use]
    pub fn channel(&self) -> &str {
        &self.channel
    }

    /// Partition whose next slot is reserved.
    #[must_use]
    pub const fn partition(&self) -> usize {
        self.partition
    }

    /// Reserved durable sequence.
    #[must_use]
    pub const fn sequence(&self) -> u64 {
        self.sequence
    }

    /// Original append error, or the fact that it returned no outcome.
    #[must_use]
    pub fn cause(&self) -> &str {
        &self.cause
    }

    /// Producer's original idempotency key, if one was supplied.
    #[must_use]
    pub fn idempotency_key(&self) -> Option<&str> {
        self.idempotency_key.as_deref()
    }

    pub(crate) fn set_cause(&mut self, cause: String) {
        self.cause = cause;
    }

    pub(crate) fn payload(&self) -> &[u8] {
        &self.payload
    }
}

impl fmt::Display for UnresolvedAppend {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "channel {} partition {} sequence {}: {}",
            self.channel, self.partition, self.sequence, self.cause
        )
    }
}

impl fmt::Debug for UnresolvedAppend {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("UnresolvedAppend")
            .field("channel", &self.channel)
            .field("partition", &self.partition)
            .field("sequence", &self.sequence)
            .field("byte_length", &self.payload.len())
            .finish_non_exhaustive()
    }
}

/// Authoritative outcome of a pending append or idempotency-key lookup.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum AppendResolution {
    /// The row exists at its original durable position.
    Committed { position: u64 },
    /// The authoritative read established that the row is absent.
    Absent,
}

impl DurableChannel {
    /// Evidence retained for this channel's unresolved append.
    #[must_use]
    pub fn pending_append(&self) -> Option<UnresolvedAppend> {
        self.pending.clone()
    }

    /// Resolve the exact reserved slot without writing another row.
    ///
    /// A row with the pending bytes at the reserved sequence is the append,
    /// committed. No row, or another writer's row there, means the append is
    /// absent: the store refuses an append at an occupied sequence. Another
    /// writer's row also moves the partition head past it, as a sequence
    /// conflict does.
    ///
    /// # Errors
    /// An unreadable slot retains its original evidence and refuses by name.
    pub async fn resolve_pending(&mut self) -> Result<Option<AppendResolution>, DurabilityError> {
        let Some(pending) = self.pending.clone() else {
            return Ok(None);
        };
        let stream_key = self.stream_key_for(pending.partition);
        let row = self
            .store
            .read_at(&stream_key, pending.sequence)
            .await
            .map_err(|error| DurabilityError::PendingAppendUnreadable {
                stream_key: stream_key.clone(),
                sequence: pending.sequence,
                append_error: pending.cause.clone(),
                read_error: error.to_string(),
            })?;
        let Some(row) = row else {
            self.pending = None;
            return Ok(Some(AppendResolution::Absent));
        };
        let next = row.sequence.checked_add(1).ok_or_else(|| {
            DurabilityError::ConfigError(format!(
                "channel {stream_key} sequence overflow during recovery"
            ))
        })?;
        if row.sequence != pending.sequence || row.payload != pending.payload {
            self.pending = None;
            self.refresh_sequence(pending.partition, &stream_key, pending.sequence, next)
                .await?;
            return Ok(Some(AppendResolution::Absent));
        }
        self.pending = None;
        self.advance_partition(pending.partition, &stream_key, next)
            .await?;
        Ok(Some(AppendResolution::Committed {
            position: row.sequence,
        }))
    }
}