liminal-server 0.25.4

//! Self-stream readers hold one durable position and never enter local publish admission.

use super::{ServerError, failure, lock};
use crate::server::connection::services::ConnectionSubscription;
use crate::server::self_streams::cost::ReaderWork;
use liminal::durability::bridge::block_on;
use liminal::durability::{ConsumerCursor, DurableStore, MessageEnvelope};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};

type PositionedPayload = Option<(Vec<u8>, Option<u64>)>;

#[derive(Clone, Debug)]
pub(super) enum Reader {
    Live(Arc<Mutex<ConnectionSubscription>>),
    Stored(Arc<Mutex<StoredSubscription>>),
}

impl Reader {
    pub(super) fn next(&self, subject: &str) -> Result<PositionedPayload, ServerError> {
        match self {
            Self::Live(subscription) => {
                let mut subscription = lock(subscription, "pending subscription")?;
                if subscription.is_overflowed() {
                    return Err(failure(format!(
                        "export subscription overflowed on '{subject}'"
                    )));
                }
                Ok(subscription
                    .try_next_positioned()?
                    .map(|(envelope, position)| (envelope.payload, position)))
            }
            Self::Stored(subscription) => lock(subscription, "stored delivery")?.next(),
        }
    }

    pub(super) fn acknowledge(&self, position: u64) -> Result<(), ServerError> {
        match self {
            Self::Live(subscription) => {
                lock(subscription, "subscription acknowledgement")?.acknowledge_position(position)
            }
            Self::Stored(subscription) => {
                lock(subscription, "stored acknowledgement")?.acknowledge(position)
            }
        }
    }

    pub(super) fn has_pending(&self) -> Result<bool, ServerError> {
        match self {
            Self::Live(subscription) => {
                Ok(lock(subscription, "attached subscription")?.has_pending())
            }
            Self::Stored(subscription) => Ok(lock(subscription, "stored readiness")?.ready()),
        }
    }
}

#[derive(Debug)]
pub(super) struct StoredSubscription {
    cursor: ConsumerCursor,
    pending: Option<u64>,
    store: Arc<dyn DurableStore>,
    head: Arc<AtomicU64>,
}

impl StoredSubscription {
    pub(super) fn resume(
        store: Arc<dyn DurableStore>,
        subject: &str,
        consumer: &str,
        head: Arc<AtomicU64>,
    ) -> Result<Self, ServerError> {
        let work = ReaderWork::enter();
        let result = block_on(ConsumerCursor::resume(
            consumer,
            format!("{subject}:0"),
            store.as_ref(),
        ))
        .map_err(|error| failure(error.to_string()))?
        .map_err(|error| failure(error.to_string()));
        drop(work);
        Ok(Self {
            cursor: result?,
            pending: None,
            store,
            head,
        })
    }

    pub(super) const fn offset(&self) -> u64 {
        self.cursor.current_offset()
    }

    fn ready(&self) -> bool {
        self.offset() < self.head.load(Ordering::Acquire)
    }

    fn next(&mut self) -> Result<PositionedPayload, ServerError> {
        if self.pending.is_some() {
            return Err(failure(
                "stored self-stream crossing still awaits acknowledgement",
            ));
        }
        if !self.ready() {
            return Ok(None);
        }
        let work = ReaderWork::enter();
        let rows = block_on(
            self.store
                .read_from(self.cursor.partition_key(), self.offset(), 1),
        )
        .map_err(|error| failure(error.to_string()))?
        .map_err(|error| failure(error.to_string()))?;
        drop(work);
        let row = rows.into_iter().next().ok_or_else(|| {
            failure(format!(
                "exported self-stream row absent: {} at {}",
                self.cursor.partition_key(),
                self.offset(),
            ))
        })?;
        if row.sequence != self.offset() {
            return Err(failure(format!(
                "exported self-stream rows no longer retained: {} expected {} found {}",
                self.cursor.partition_key(),
                self.offset(),
                row.sequence,
            )));
        }
        let stored = MessageEnvelope::deserialize(&row.payload)
            .map_err(|error| failure(error.to_string()))?;
        self.pending = Some(row.sequence);
        Ok(Some((stored.payload, Some(row.sequence))))
    }

    fn acknowledge(&mut self, position: u64) -> Result<(), ServerError> {
        if self.pending != Some(position) {
            return Err(failure(
                "stored acknowledgement does not name its delivered position",
            ));
        }
        let next = position
            .checked_add(1)
            .ok_or_else(|| failure("stored self-stream position overflow"))?;
        let work = ReaderWork::enter();
        block_on(self.cursor.checkpoint(self.store.as_ref(), next))
            .map_err(|error| failure(error.to_string()))?
            .map_err(|error| failure(error.to_string()))?;
        block_on(self.store.flush())
            .map_err(|error| failure(error.to_string()))?
            .map_err(|error| failure(error.to_string()))?;
        drop(work);
        self.pending = None;
        Ok(())
    }
}