liminal-server 0.25.4

//! A throwaway store opened by its first write, never at start.
//!
//! An incarnation's ephemeral dedup namespace lives in a store of its own when
//! the services' store outlives the process. Opening that store creates a
//! database directory and flushes it, which a server that never publishes a
//! keyed message to an ephemeral channel does not need, and which would sit on
//! the boot path before the listener binds. [`DeferredStore`] opens it when
//! something is first written to it. Until then it holds nothing, so a read
//! answers empty without opening it, and a flush has nothing to flush.
use std::fmt;
use std::sync::{Mutex, OnceLock, PoisonError};

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

/// Opens the throwaway store; called at most once, by the first write.
pub(super) type ThrowawayStoreOpener =
    Box<dyn Fn() -> Result<EphemeralHaematiteStore, DurabilityError> + Send + Sync>;

/// A store opened by its first write. See the module documentation.
pub(super) struct DeferredStore {
    open: ThrowawayStoreOpener,
    store: OnceLock<EphemeralHaematiteStore>,
    /// Held while the store is opened, so two first writes open it once.
    opening: Mutex<()>,
}

impl DeferredStore {
    pub(super) fn new(open: ThrowawayStoreOpener) -> Self {
        Self {
            open,
            store: OnceLock::new(),
            opening: Mutex::new(()),
        }
    }

    /// The store, once a write has opened it.
    fn opened(&self) -> Option<&EphemeralHaematiteStore> {
        self.store.get()
    }

    /// The store, opened now when no write has opened it yet.
    fn for_write(&self) -> Result<&EphemeralHaematiteStore, DurabilityError> {
        if let Some(store) = self.store.get() {
            return Ok(store);
        }
        let _opening = self.opening.lock().unwrap_or_else(PoisonError::into_inner);
        if let Some(store) = self.store.get() {
            return Ok(store);
        }
        let store = (self.open)()?;
        Ok(self.store.get_or_init(|| store))
    }
}

impl fmt::Debug for DeferredStore {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("DeferredStore")
            .field("store", &self.store.get())
            .finish_non_exhaustive()
    }
}

#[async_trait::async_trait]
impl DurableStore for DeferredStore {
    async fn apply_routed_batch(
        &self,
        route_key: &str,
        items: Vec<liminal::durability::RoutedBatchItem>,
    ) -> Result<(), DurabilityError> {
        if items.is_empty() {
            return Ok(());
        }
        self.for_write()?.apply_routed_batch(route_key, items).await
    }

    async fn apply_routed_entries(
        &self,
        route_key: &str,
        entries: Vec<RoutedBatchEntry>,
    ) -> Result<(), DurabilityError> {
        if entries.is_empty() {
            return Ok(());
        }
        self.for_write()?
            .apply_routed_entries(route_key, entries)
            .await
    }

    async fn get_routed(
        &self,
        route_key: &str,
        key: &[u8],
    ) -> Result<Option<Vec<u8>>, DurabilityError> {
        match self.opened() {
            Some(store) => store.get_routed(route_key, key).await,
            None => Ok(None),
        }
    }

    async fn append(
        &self,
        stream_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.for_write()?
            .append(stream_key, payload, expected_seq)
            .await
    }

    async fn append_all(
        &self,
        stream_key: &str,
        payloads: Vec<Vec<u8>>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.for_write()?
            .append_all(stream_key, payloads, expected_seq)
            .await
    }

    async fn read_from(
        &self,
        stream_key: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Vec<StoredEntry>, DurabilityError> {
        match self.opened() {
            Some(store) => store.read_from(stream_key, offset, limit).await,
            None => Ok(Vec::new()),
        }
    }

    async fn read_at(
        &self,
        stream_key: &str,
        sequence: u64,
    ) -> Result<Option<StoredEntry>, DurabilityError> {
        match self.opened() {
            Some(store) => store.read_at(stream_key, sequence).await,
            None => Ok(None),
        }
    }

    async fn cas(&self, key: &str, old_value: u64, new_value: u64) -> Result<(), DurabilityError> {
        self.for_write()?.cas(key, old_value, new_value).await
    }

    async fn read_stream_next_seq(&self, stream_key: &str) -> Result<Option<u64>, DurabilityError> {
        match self.opened() {
            Some(store) => store.read_stream_next_seq(stream_key).await,
            None => Ok(None),
        }
    }

    async fn read_value(&self, key: &str) -> Result<Option<u64>, DurabilityError> {
        match self.opened() {
            Some(store) => store.read_value(key).await,
            None => Ok(None),
        }
    }

    async fn scan_routed(
        &self,
        route_key: &str,
        from: &[u8],
        to: &[u8],
    ) -> Result<Vec<(Vec<u8>, Vec<u8>)>, DurabilityError> {
        match self.opened() {
            Some(store) => store.scan_routed(route_key, from, to).await,
            None => Ok(Vec::new()),
        }
    }

    async fn remove_routed_range(
        &self,
        route_key: &str,
        from: &[u8],
        to: &[u8],
    ) -> Result<u64, DurabilityError> {
        match self.opened() {
            Some(store) => store.remove_routed_range(route_key, from, to).await,
            None => Ok(0),
        }
    }

    async fn scan_page(
        &self,
        prefix: &str,
        cursor: Option<ScanCursor>,
        limit: usize,
    ) -> Result<ScanPage, DurabilityError> {
        match self.opened() {
            Some(store) => store.scan_page(prefix, cursor, limit).await,
            None => Ok(ScanPage {
                entries: Vec::new(),
                next: None,
            }),
        }
    }

    async fn flush(&self) -> Result<(), DurabilityError> {
        match self.opened() {
            Some(store) => store.flush().await,
            None => Ok(()),
        }
    }

    /// An unopened store holds no rows, so a cut deletes none: 0 is the
    /// count, not a stand-in for one. It is not opened to find that out.
    async fn truncate_before(
        &self,
        stream_key: &str,
        before_seq: u64,
    ) -> Result<u64, DurabilityError> {
        match self.opened() {
            Some(store) => store.truncate_before(stream_key, before_seq).await,
            None => Ok(0),
        }
    }
}

#[cfg(test)]
#[path = "deferred_store_batch_tests.rs"]
mod batch_tests;