liminal-rs 0.10.0

use super::{DurabilityError, DurableStore, MessageEnvelope, StoredEntry};

const READ_BATCH_SIZE: usize = 1_024;

/// Replays serialized durable channel envelopes from `offset` to the partition head.
///
/// # Errors
///
/// Propagates store read errors, envelope deserialization errors, and read-offset overflow.
pub async fn replay_from(
    store: &dyn DurableStore,
    partition_key: &str,
    offset: u64,
) -> Result<Vec<MessageEnvelope>, DurabilityError> {
    let mut stored_entries = Vec::new();
    let mut next_offset = offset;
    loop {
        let batch = store
            .read_from(partition_key, next_offset, READ_BATCH_SIZE)
            .await?;
        let batch_len = batch.len();
        if batch_len == 0 {
            break;
        }
        stored_entries.extend(batch);
        next_offset = next_offset
            .checked_add(len_to_u64(batch_len)?)
            .ok_or_else(|| {
                DurabilityError::ConfigError("replay read offset overflow".to_owned())
            })?;
    }

    deserialize_in_sequence_order(stored_entries)
}

/// Replays at most `limit` durable channel envelopes from `offset`, each paired
/// with the sequence it was stored at.
///
/// The bounded sibling of [`replay_from`], and the read A1's host-side
/// auto-catch-up uses (`docs/design/A1-DEFER-SEMANTICS.md` §4): a lagging
/// subscriber sizes each replay batch to its own free buffer band, so replay
/// can never blow the bound it exists to serve. `replay_from` reads to the head
/// in one call and so cannot serve a bounded consumer.
///
/// The sequences come back because the caller advances a per-subscriber replay
/// cursor with them; `replay_from` discards them, which is why this is a
/// sibling rather than a parameterisation of it.
///
/// An empty result means the read reached the partition head. Callers treating
/// that as "caught up" must guard against the head having moved — see §4's
/// loop-until-caught-up rule.
///
/// # Errors
///
/// Propagates store read errors and envelope deserialization errors.
pub async fn replay_range(
    store: &dyn DurableStore,
    partition_key: &str,
    offset: u64,
    limit: usize,
) -> Result<Vec<(u64, MessageEnvelope)>, DurabilityError> {
    let mut stored_entries = store.read_from(partition_key, offset, limit).await?;
    stored_entries.sort_by_key(|entry| entry.sequence);
    stored_entries.truncate(limit);
    let mut envelopes = Vec::with_capacity(stored_entries.len());
    for stored in stored_entries {
        envelopes.push((
            stored.sequence,
            MessageEnvelope::deserialize(&stored.payload)?,
        ));
    }
    Ok(envelopes)
}

fn deserialize_in_sequence_order(
    mut stored_entries: Vec<StoredEntry>,
) -> Result<Vec<MessageEnvelope>, DurabilityError> {
    stored_entries.sort_by_key(|entry| entry.sequence);

    let mut envelopes = Vec::with_capacity(stored_entries.len());
    for stored in stored_entries {
        envelopes.push(MessageEnvelope::deserialize(&stored.payload)?);
    }
    Ok(envelopes)
}

fn len_to_u64(len: usize) -> Result<u64, DurabilityError> {
    u64::try_from(len).map_err(|error| {
        DurabilityError::ConfigError(format!("entry count cannot fit u64: {error}"))
    })
}

#[cfg(test)]
mod tests;