liminal-server 0.24.9

//! Deterministic independent append/flush gates for both Unit 1 and Unit 2 streams.

use std::collections::VecDeque;
use std::error::Error;
use std::sync::{Arc, Condvar, Mutex};

use liminal::durability::{DurabilityError, DurableStore, ScanCursor, ScanPage, StoredEntry};

use super::log::STREAM_PREFIX;
use super::outbox_log::OUTBOX_STREAM_PREFIX;
use super::routed_stream_codec::row_stream;

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum OutboxBarrierKind {
    OperationAppend,
    OperationFlush,
    OutboxAppend,
    OutboxFlush,
}

#[derive(Debug, Default)]
struct BarrierState {
    gates: VecDeque<OutboxBarrierKind>,
    fail_next: Option<OutboxBarrierKind>,
    reached: Option<OutboxBarrierKind>,
    released: bool,
    pending_flush: Option<OutboxBarrierKind>,
}

#[derive(Debug)]
pub(super) struct OutboxBarrierStore {
    inner: Arc<dyn DurableStore>,
    state: Mutex<BarrierState>,
    changed: Condvar,
}

impl OutboxBarrierStore {
    pub(super) fn new(inner: Arc<dyn DurableStore>) -> Self {
        Self {
            inner,
            state: Mutex::new(BarrierState::default()),
            changed: Condvar::new(),
        }
    }

    pub(super) fn arm(
        &self,
        gates: impl IntoIterator<Item = OutboxBarrierKind>,
    ) -> Result<(), Box<dyn Error>> {
        let mut state = self.state.lock().map_err(|_| "outbox barrier poisoned")?;
        state.gates = gates.into_iter().collect();
        state.reached = None;
        state.released = false;
        state.pending_flush = None;
        drop(state);
        Ok(())
    }

    pub(super) fn fail_next(&self, kind: OutboxBarrierKind) -> Result<(), Box<dyn Error>> {
        let mut state = self.state.lock().map_err(|_| "outbox barrier poisoned")?;
        state.fail_next = Some(kind);
        drop(state);
        Ok(())
    }

    /// Whether the fault armed by [`Self::fail_next`] has struck: a fault
    /// left armed refused nothing, so the pin that armed it measured nothing.
    pub(super) fn fail_next_struck(&self) -> Result<bool, Box<dyn Error>> {
        Ok(self
            .state
            .lock()
            .map_err(|_| "outbox barrier poisoned")?
            .fail_next
            .is_none())
    }

    pub(super) fn wait_for(&self, expected: OutboxBarrierKind) -> Result<(), Box<dyn Error>> {
        let mut state = self.state.lock().map_err(|_| "outbox barrier poisoned")?;
        while state.reached != Some(expected) {
            state = self
                .changed
                .wait(state)
                .map_err(|_| "outbox barrier poisoned while waiting")?;
        }
        drop(state);
        Ok(())
    }

    pub(super) fn release(&self, expected: OutboxBarrierKind) -> Result<(), Box<dyn Error>> {
        let mut state = self.state.lock().map_err(|_| "outbox barrier poisoned")?;
        if state.reached != Some(expected) {
            return Err(format!(
                "attempted to release {expected:?} while {:?} was reached",
                state.reached
            )
            .into());
        }
        state.released = true;
        drop(state);
        self.changed.notify_all();
        Ok(())
    }

    fn cross(&self, kind: OutboxBarrierKind) -> Result<(), DurabilityError> {
        let mut state = self.state.lock().map_err(|_| barrier_fault())?;
        if state.fail_next == Some(kind) {
            state.fail_next = None;
            return Err(barrier_fault());
        }
        if state.gates.front().copied() == Some(kind) {
            state.reached = Some(kind);
            state.released = false;
            self.changed.notify_all();
            while !state.released {
                state = self.changed.wait(state).map_err(|_| barrier_fault())?;
            }
            state.reached = None;
            state.released = false;
            state.gates.pop_front();
            self.changed.notify_all();
        }
        drop(state);
        Ok(())
    }

    fn set_pending_flush(&self, kind: OutboxBarrierKind) -> Result<(), DurabilityError> {
        let mut state = self.state.lock().map_err(|_| barrier_fault())?;
        state.pending_flush = Some(kind);
        drop(state);
        Ok(())
    }

    fn take_pending_flush(&self) -> Result<Option<OutboxBarrierKind>, DurabilityError> {
        let mut state = self.state.lock().map_err(|_| barrier_fault())?;
        let pending = state.pending_flush.take();
        drop(state);
        Ok(pending)
    }
}

#[async_trait::async_trait]
impl DurableStore for OutboxBarrierStore {
    /// One routed batch is the append and the flush of every row it holds:
    /// an operation row crosses `OperationAppend` then `OperationFlush`, a
    /// Unit 2 row `OutboxAppend` then `OutboxFlush`, in the order the native appends
    /// crossed them, before anything is applied. A gate armed to fail refuses
    /// the whole batch, so nothing of it is durable.
    async fn apply_routed_batch(
        &self,
        route_key: &str,
        items: Vec<liminal::durability::RoutedBatchItem>,
    ) -> Result<(), liminal::durability::DurabilityError> {
        let holds = |prefix: &str| {
            items
                .iter()
                .any(|item| row_stream(&item.key).is_some_and(|stream| stream.starts_with(prefix)))
        };
        if holds(STREAM_PREFIX) {
            self.cross(OutboxBarrierKind::OperationAppend)?;
            self.cross(OutboxBarrierKind::OperationFlush)?;
        }
        if holds(OUTBOX_STREAM_PREFIX) {
            self.cross(OutboxBarrierKind::OutboxAppend)?;
            self.cross(OutboxBarrierKind::OutboxFlush)?;
        }
        self.inner.apply_routed_batch(route_key, items).await
    }

    async fn get_routed(
        &self,
        route_key: &str,
        key: &[u8],
    ) -> Result<Option<Vec<u8>>, liminal::durability::DurabilityError> {
        self.inner.get_routed(route_key, key).await
    }

    async fn append(
        &self,
        stream_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        let (append_gate, flush_gate) = if stream_key.starts_with(STREAM_PREFIX) {
            (
                Some(OutboxBarrierKind::OperationAppend),
                Some(OutboxBarrierKind::OperationFlush),
            )
        } else if stream_key.starts_with(OUTBOX_STREAM_PREFIX) {
            (
                Some(OutboxBarrierKind::OutboxAppend),
                Some(OutboxBarrierKind::OutboxFlush),
            )
        } else {
            (None, None)
        };
        if let Some(gate) = append_gate {
            self.cross(gate)?;
        }
        let assigned = self.inner.append(stream_key, payload, expected_seq).await?;
        if let Some(gate) = flush_gate {
            self.set_pending_flush(gate)?;
        }
        Ok(assigned)
    }

    async fn append_all(
        &self,
        stream_key: &str,
        payloads: Vec<Vec<u8>>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        let (append_gate, flush_gate) = if stream_key.starts_with(STREAM_PREFIX) {
            (
                Some(OutboxBarrierKind::OperationAppend),
                Some(OutboxBarrierKind::OperationFlush),
            )
        } else if stream_key.starts_with(OUTBOX_STREAM_PREFIX) {
            (
                Some(OutboxBarrierKind::OutboxAppend),
                Some(OutboxBarrierKind::OutboxFlush),
            )
        } else {
            (None, None)
        };
        if let Some(gate) = append_gate {
            self.cross(gate)?;
        }
        let assigned = self
            .inner
            .append_all(stream_key, payloads, expected_seq)
            .await?;
        if let Some(gate) = flush_gate {
            self.set_pending_flush(gate)?;
        }
        Ok(assigned)
    }

    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 cas(&self, key: &str, old_value: u64, new_value: u64) -> Result<(), DurabilityError> {
        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> {
        if let Some(gate) = self.take_pending_flush()? {
            self.cross(gate)?;
        }
        self.inner.flush().await
    }
}

const fn barrier_fault() -> DurabilityError {
    DurabilityError::SequenceConflict {
        expected: 0,
        actual: u64::MAX,
    }
}