liminal-server 0.20.1

//! Deterministic participant-source and observer-Advance durability barriers.

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::routed_stream_codec::row_stream;

const OBSERVER_STREAM_KEY: &str = "liminal:participant-observer-recovery";

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum BarrierKind {
    Source,
    Advance,
}

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

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

impl ObserverBarrierStore {
    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 = BarrierKind>,
    ) -> Result<(), Box<dyn Error>> {
        let mut state = self.state.lock().map_err(|_| "barrier state poisoned")?;
        state.gates = gates.into_iter().collect();
        state.reached = None;
        state.released = false;
        drop(state);
        Ok(())
    }

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

    pub(super) fn release(&self, expected: BarrierKind) -> Result<(), Box<dyn Error>> {
        let mut state = self.state.lock().map_err(|_| "barrier state 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(())
    }

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

    fn cross_flush_barrier(&self) -> Result<bool, DurabilityError> {
        let mut state = self.state.lock().map_err(|_| barrier_fault())?;
        let pending = state.pending.take();
        let fail = if state.fail_next == pending && pending.is_some() {
            state.fail_next = None;
            true
        } else {
            if state.gates.front().copied() == pending && pending.is_some() {
                state.reached = pending;
                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();
            }
            false
        };
        drop(state);
        Ok(fail)
    }
}

#[async_trait::async_trait]
impl DurableStore for ObserverBarrierStore {
    /// One routed batch is the append and the flush of every row it holds:
    /// a participant-source row crosses the Source gate and an observer
    /// Advance row the Advance gate, in that order, before anything is
    /// applied, exactly where the native append's flush crossed them. 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 source = items.iter().any(|item| {
            row_stream(&item.key).is_some_and(|stream| stream.starts_with(STREAM_PREFIX))
        });
        let advance = items.iter().any(|item| {
            row_stream(&item.key) == Some(OBSERVER_STREAM_KEY)
                && item
                    .value
                    .windows(b"\"row\":\"advance\"".len())
                    .any(|window| window == b"\"row\":\"advance\"")
        });
        for kind in [
            source.then_some(BarrierKind::Source),
            advance.then_some(BarrierKind::Advance),
        ]
        .into_iter()
        .flatten()
        {
            let mut state = self.state.lock().map_err(|_| barrier_fault())?;
            state.pending = Some(kind);
            drop(state);
            if self.cross_flush_barrier()? {
                return Err(barrier_fault());
            }
        }
        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 kind = if stream_key.starts_with(STREAM_PREFIX) {
            Some(BarrierKind::Source)
        } else if stream_key == OBSERVER_STREAM_KEY
            && payload
                .windows(b"\"row\":\"advance\"".len())
                .any(|window| window == b"\"row\":\"advance\"")
        {
            Some(BarrierKind::Advance)
        } else {
            None
        };
        let assigned = self.inner.append(stream_key, payload, expected_seq).await?;
        let mut state = self.state.lock().map_err(|_| barrier_fault())?;
        state.pending = kind;
        drop(state);
        Ok(assigned)
    }

    async fn append_all(
        &self,
        stream_key: &str,
        payloads: Vec<Vec<u8>>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        let kind = if stream_key.starts_with(STREAM_PREFIX) {
            Some(BarrierKind::Source)
        } else if stream_key == OBSERVER_STREAM_KEY
            && payloads.iter().any(|payload| {
                payload
                    .windows(b"\"row\":\"advance\"".len())
                    .any(|window| window == b"\"row\":\"advance\"")
            })
        {
            Some(BarrierKind::Advance)
        } else {
            None
        };
        let next = self
            .inner
            .append_all(stream_key, payloads, expected_seq)
            .await?;
        let mut state = self.state.lock().map_err(|_| barrier_fault())?;
        state.pending = kind;
        drop(state);
        Ok(next)
    }

    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 self.cross_flush_barrier()? {
            return Err(barrier_fault());
        }
        self.inner.flush().await
    }
}

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