liminal/channel/

durable_publish.rs

//! Durable admission persists a publish and enqueues its fan-out under one append
//! lock, and fans out a row whose uncertain append resolved as committed ahead of
//! the publish that resolved it.
use std::collections::VecDeque;
use std::sync::{Arc, Mutex};

use super::actor::{ChannelActorCore, PendingPublish};
use super::record::{PublishRecord, recorded_now};
use super::types::{ChannelDelivery, ChannelHandle, now_millis};
use crate::causal::CausalContext;
use crate::durability::bridge::block_on;
use crate::durability::{AppendResolution, DurableChannel, MessageEnvelope, UnresolvedAppend};
use crate::envelope::PublisherId;
use crate::error::LiminalError;

/// A durable channel's storage state, with the runtime metadata of the row it
/// still owes the live subscribers kept beside the channel's pending record.
pub(super) struct RuntimeDurable {
    pub(super) channel: DurableChannel,
    fanout: Option<PendingFanout>,
    batch_fanout: VecDeque<PendingFanout>,
}

/// The fan-out of a row whose publish returned before the row reached the
/// actor: the payload, publisher and causal context of the publish that
/// produced it, and the row's position once the store has established it.
struct PendingFanout {
    delivery_id: Option<crate::protocol::DeliveredId>,
    payload: Vec<u8>,
    publisher: PublisherId,
    causal: Option<CausalContext>,
    position: Option<u64>,
}

/// Why the row owed from an earlier publish could not be settled.
enum RecoveryFailure {
    /// The pending append could not be resolved; its record is unchanged.
    Unresolved(String),
    /// The pending append resolved as committed and its fan-out could not be
    /// enqueued; the fan-out stays owed.
    Fanout(LiminalError),
}

/// A row the store committed, carried past the append lock so a live-delivery
/// failure still names the row instead of reporting a failure that would
/// license a second append.
pub(super) struct CommittedRow {
    append: UnresolvedAppend,
}

impl CommittedRow {
    /// The committed row as an unresolved publish: stored, with its live
    /// delivery not confirmed.
    pub(super) fn unresolved(self, detail: impl std::fmt::Display) -> LiminalError {
        let mut append = self.append;
        append.set_cause(format!("row committed; live delivery failed: {detail}"));
        LiminalError::PublishUnresolved { append }
    }
}

impl std::fmt::Debug for RuntimeDurable {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RuntimeDurable")
            .field("channel", &self.channel)
            .field("pending_fanout", &self.fanout.is_some())
            .field("pending_batch_fanout", &self.batch_fanout.len())
            .finish()
    }
}

impl RuntimeDurable {
    pub(super) const fn new(channel: DurableChannel) -> Self {
        Self {
            channel,
            fanout: None,
            batch_fanout: VecDeque::new(),
        }
    }

    /// Resolves the channel's pending append and enqueues the fan-out a
    /// committed row still owes, returning its rendezvous. `None` means
    /// nothing is owed and the next append may proceed.
    fn recover_fanout(
        &mut self,
        core: &ChannelActorCore,
    ) -> Result<Option<PendingPublish>, RecoveryFailure> {
        if self.channel.pending_batch_evidence().is_some() {
            match block_on(self.channel.resolve_pending_batch()) {
                Ok(Ok(Some(AppendResolution::Absent))) => self.batch_fanout.clear(),
                Ok(Ok(_)) => {}
                Ok(Err(error)) => return Err(RecoveryFailure::Unresolved(error.to_string())),
                Err(error) => return Err(RecoveryFailure::Unresolved(error.to_string())),
            }
        }
        if let Some(fanout) = self.batch_fanout.front() {
            let pending = core
                .enqueue_identified(
                    fanout.payload.clone(),
                    fanout.publisher.clone(),
                    fanout.causal.clone(),
                    fanout.position,
                    None,
                    fanout.delivery_id,
                )
                .map_err(RecoveryFailure::Fanout)?;
            self.batch_fanout.pop_front();
            return Ok(Some(pending));
        }
        if self.channel.pending_append().is_some() {
            let resolution = match block_on(self.channel.resolve_pending()) {
                Ok(Ok(resolution)) => resolution,
                Ok(Err(error)) => return Err(RecoveryFailure::Unresolved(error.to_string())),
                Err(error) => return Err(RecoveryFailure::Unresolved(error.to_string())),
            };
            match resolution {
                Some(AppendResolution::Committed { position }) => {
                    if let Some(fanout) = self.fanout.as_mut() {
                        fanout.position = Some(position);
                    }
                }
                Some(AppendResolution::Absent) | None => self.fanout = None,
            }
        }
        let Some(fanout) = self.fanout.as_ref() else {
            return Ok(None);
        };
        let Some(position) = fanout.position else {
            // No pending record and no position: the row this metadata waited
            // for can no longer be established, so nothing is owed for it.
            self.fanout = None;
            return Ok(None);
        };
        let pending = core
            .enqueue_identified(
                fanout.payload.clone(),
                fanout.publisher.clone(),
                fanout.causal.clone(),
                Some(position),
                None,
                fanout.delivery_id,
            )
            .map_err(RecoveryFailure::Fanout)?;
        self.fanout = None;
        Ok(Some(pending))
    }

    /// Maps a refused durable publish. An append the channel still holds as
    /// pending is unresolved, and its fan-out metadata is kept beside the
    /// record; every other refusal is definite.
    fn refused(
        &mut self,
        message: String,
        payload: &[u8],
        publisher: PublisherId,
        causal: Option<CausalContext>,
        delivery_id: Option<crate::protocol::DeliveredId>,
    ) -> LiminalError {
        let Some(append) = self.channel.pending_append() else {
            return LiminalError::PublishFailed { message };
        };
        self.fanout = Some(PendingFanout {
            delivery_id,
            payload: payload.to_vec(),
            publisher,
            causal,
            position: None,
        });
        LiminalError::PublishUnresolved { append }
    }
}

/// One publish on its way to the durable store and the fan-out.
pub(super) struct Outgoing<'a> {
    pub(super) delivery_id: Option<crate::protocol::DeliveredId>,
    pub(super) payload: &'a [u8],
    pub(super) publisher_id: PublisherId,
    pub(super) causal_context: Option<CausalContext>,
    pub(super) idempotency_key: Option<&'a str>,
    /// The caller's record, for the fan-out to fill. Its recipient buffer is
    /// lent to the actor only as the publish is enqueued, so a failure before
    /// then leaves it with the caller.
    pub(super) record: Option<&'a mut PublishRecord>,
}

impl ChannelHandle {
    pub(super) fn persist_batch_and_enqueue(
        &self,
        core: &Arc<ChannelActorCore>,
        durable: &Arc<Mutex<RuntimeDurable>>,
        payloads: &[Vec<u8>],
        publisher: &PublisherId,
        causal: Option<&CausalContext>,
    ) -> Result<Vec<(PendingPublish, UnresolvedAppend)>, LiminalError> {
        loop {
            let mut state = durable
                .lock()
                .map_err(|error| LiminalError::PublishFailed {
                    message: format!("durable batch state unavailable: {error}"),
                })?;
            match state.recover_fanout(core) {
                Ok(Some(pending)) => {
                    drop(state);
                    self.finish_recovered(core, pending)?;
                    continue;
                }
                Ok(None) => {}
                Err(RecoveryFailure::Unresolved(message)) => {
                    return Err(LiminalError::PublishFailed { message });
                }
                Err(RecoveryFailure::Fanout(error)) => return Err(error),
            }
            return state.append_batch(core, payloads, publisher, causal);
        }
    }

    /// Publishes as [`ChannelHandle::publish_with_delivery`] does, and stores
    /// the producer's idempotency key in the persisted row so the row can be
    /// found by its key.
    ///
    /// # Errors
    ///
    /// Returns the errors of [`ChannelHandle::publish_with_delivery`]. A
    /// durable append whose outcome is not known is
    /// [`LiminalError::PublishUnresolved`], never a definite failure.
    pub fn publish_with_idempotency_key<Payload: AsRef<[u8]>>(
        &self,
        payload: Payload,
        publisher_id: PublisherId,
        causal_context: Option<CausalContext>,
        idempotency_key: &str,
    ) -> Result<ChannelDelivery, LiminalError> {
        self.publish_with_key(
            payload,
            publisher_id,
            causal_context,
            Some(idempotency_key),
            None,
        )
    }

    /// Persists one publish to the durable store and hands its fan-out to the
    /// actor's command queue **without releasing the append lock in between**
    /// (A1 §4; round-2 finding). Returns the assigned append sequence — the
    /// durable position G1 says must not be discarded — the pending
    /// rendezvous for the fan-out, and the committed row.
    ///
    /// **Why the two must not be separated.** `DurableChannel::publish` assigns
    /// sequences under this mutex, so durable positions are totally ordered.
    /// The actor's command queue is a `VecDeque` drained in push order, so the
    /// fan-out order is the enqueue order. Take the position under the lock and
    /// enqueue after releasing it, and the two orders are independent: two
    /// publishers can invert, the higher position's live push reaching an inbox
    /// first. That is not a cosmetic reordering. §4's replay cursor is one
    /// `u64` meaning "the contiguous prefix this subscriber has been offered",
    /// and an inversion makes that sentence false — the cursor steps past a
    /// message that has not been offered, and backsteps onto one that has.
    /// Enqueuing here, inside the same critical section, is what makes the two
    /// orders the same order.
    ///
    /// **A row owed from an earlier publish goes first.** An append whose
    /// outcome the store could not report is resolved by one read under the
    /// same lock before this publish appends. A committed row is enqueued at
    /// its position ahead of this one and awaited with the lock released, so
    /// the live order stays the position order.
    ///
    /// **What it costs.** The section grows by one `VecDeque` push under the
    /// command mutex plus one non-blocking scheduler wake — both O(1), neither
    /// blocking. The fan-out ITSELF is not held here (that is
    /// [`ChannelActorCore::await_publish`], called with this lock released), so
    /// the next publisher's append still overlaps this publisher's fan-out
    /// exactly as it did before: the pipeline is unchanged and no subscriber's
    /// notifier can ever stall an append.
    pub(super) fn persist_and_enqueue(
        &self,
        core: &Arc<ChannelActorCore>,
        durable: &Arc<Mutex<RuntimeDurable>>,
        outgoing: Outgoing<'_>,
    ) -> Result<(u64, PendingPublish, CommittedRow), LiminalError> {
        let Outgoing {
            delivery_id,
            payload,
            publisher_id,
            causal_context,
            idempotency_key,
            record,
        } = outgoing;
        loop {
            let mut state = durable
                .lock()
                .map_err(|error| LiminalError::PublishFailed {
                    message: format!("durable channel state unavailable: {error}"),
                })?;
            match state.recover_fanout(core) {
                Ok(Some(recovered)) => {
                    drop(state);
                    self.finish_recovered(core, recovered)?;
                    continue;
                }
                Ok(None) => {}
                Err(RecoveryFailure::Unresolved(detail)) => {
                    return Err(LiminalError::PublishFailed {
                        message: format!(
                            "durable publish to channel '{}' failed: {detail}",
                            self.config.name
                        ),
                    });
                }
                Err(RecoveryFailure::Fanout(error)) => return Err(error),
            }
            let envelope = MessageEnvelope {
                payload: payload.to_vec(),
                causal_context: causal_context.as_ref().map(|context| {
                    crate::durability::CausalContext {
                        parent_id: context.parent.map(|parent| parent.as_uuid().to_string()),
                        vector_clock_entry: None,
                    }
                }),
                timestamp: now_millis(),
                publisher_id: publisher_id.as_str().to_owned(),
                idempotency_key: idempotency_key.map(str::to_owned),
            };
            let append = match block_on(state.channel.publish_recorded(&envelope)) {
                Ok(Ok(append)) => append,
                Ok(Err(error)) => {
                    let message = format!(
                        "durable publish to channel '{}' failed: {error}",
                        self.config.name
                    );
                    return Err(state.refused(
                        message,
                        payload,
                        publisher_id,
                        causal_context,
                        delivery_id,
                    ));
                }
                Err(error) => {
                    let message = format!(
                        "durable publish bridge for channel '{}' failed: {error}",
                        self.config.name
                    );
                    return Err(state.refused(
                        message,
                        payload,
                        publisher_id,
                        causal_context,
                        delivery_id,
                    ));
                }
            };
            let position = append.sequence();
            let row = CommittedRow { append };
            let pending = match core.enqueue_identified(
                payload.to_vec(),
                publisher_id.clone(),
                causal_context.clone(),
                Some(position),
                // The append has completed; the fan-out comes after it.
                record.map(|record| {
                    record.appended_at = Some(recorded_now());
                    &mut record.recipients
                }),
                delivery_id,
            ) {
                Ok(pending) => pending,
                Err(error) => {
                    // The row is stored and owes its fan-out; the next admission
                    // enqueues it at its position.
                    state.fanout = Some(PendingFanout {
                        delivery_id,
                        payload: payload.to_vec(),
                        publisher: publisher_id,
                        causal: causal_context,
                        position: Some(position),
                    });
                    return Err(row.unresolved(error));
                }
            };
            drop(state);
            return Ok((position, pending, row));
        }
    }

    /// Where each partition of the durable channel stands: the sequence its
    /// next row takes. A claim records these before its publish appends, so
    /// a later lookup of its key starts here.
    ///
    /// # Errors
    ///
    /// Returns [`LiminalError::PublishFailed`] when the channel is not durable
    /// or its state is unavailable.
    pub fn durable_positions(&self) -> Result<Vec<u64>, LiminalError> {
        let durable = self
            .durable
            .as_ref()
            .ok_or_else(|| LiminalError::PublishFailed {
                message: format!("channel '{}' has no durable stream", self.config.name),
            })?;
        let state = durable
            .lock()
            .map_err(|error| LiminalError::PublishFailed {
                message: format!("durable channel state unavailable: {error}"),
            })?;
        Ok(state.channel.next_sequences().to_vec())
    }

    /// Resolves the channel's pending append, fanning out a committed row,
    /// then looks the row carrying `idempotency_key` up in the channel's
    /// stream from the positions in `from`, page by page, with the append
    /// lock released. Writes no row.
    ///
    /// # Errors
    ///
    /// Returns [`LiminalError::PublishUnresolved`] when the pending append
    /// cannot be read back, and [`LiminalError::ResolutionUnreadable`] naming
    /// the channel and the key when the stream cannot be read.
    pub fn resolve_idempotency_key(
        &self,
        idempotency_key: &str,
        from: &[u64],
    ) -> Result<AppendResolution, LiminalError> {
        let unreadable = |message: String| LiminalError::ResolutionUnreadable {
            channel: self.config.name.clone(),
            idempotency_key: idempotency_key.to_owned(),
            message,
        };
        let durable = self
            .durable
            .as_ref()
            .ok_or_else(|| unreadable("channel has no durable stream".to_owned()))?;
        let core = self.core()?;
        loop {
            let mut state = durable
                .lock()
                .map_err(|error| LiminalError::PublishFailed {
                    message: format!("durable channel state unavailable: {error}"),
                })?;
            match state.recover_fanout(&core) {
                Ok(Some(recovered)) => {
                    drop(state);
                    self.finish_recovered(&core, recovered)?;
                    continue;
                }
                Ok(None) => {}
                Err(RecoveryFailure::Unresolved(detail)) => {
                    return Err(state.channel.pending_append().map_or_else(
                        || LiminalError::PublishFailed {
                            message: format!(
                                "durable channel '{}' could not resolve its pending append: {detail}",
                                self.config.name
                            ),
                        },
                        |append| LiminalError::PublishUnresolved { append },
                    ));
                }
                Err(RecoveryFailure::Fanout(error)) => return Err(error),
            }
            // The pending append is settled and a held key is never appended
            // again, so the lookup reads with the append lock released.
            let lookup = state.channel.key_lookup();
            drop(state);
            return block_on(lookup.find(idempotency_key, from))
                .map_err(|error| unreadable(error.to_string()))?
                .map_err(|error| unreadable(error.to_string()));
        }
    }
}

impl RuntimeDurable {
    fn append_batch(
        &mut self,
        core: &ChannelActorCore,
        payloads: &[Vec<u8>],
        publisher: &PublisherId,
        causal: Option<&CausalContext>,
    ) -> Result<Vec<(PendingPublish, UnresolvedAppend)>, LiminalError> {
        let start =
            self.channel
                .next_expected_sequence(0)
                .ok_or_else(|| LiminalError::PublishFailed {
                    message: "durable batch has no partition head".to_owned(),
                })?;
        let envelopes: Vec<_> = payloads
            .iter()
            .map(|payload| MessageEnvelope {
                payload: payload.clone(),
                causal_context: causal.map(|context| crate::durability::CausalContext {
                    parent_id: context.parent.map(|parent| parent.as_uuid().to_string()),
                    vector_clock_entry: None,
                }),
                timestamp: now_millis(),
                publisher_id: publisher.as_str().to_owned(),
                idempotency_key: None,
            })
            .collect();
        let proofs = envelopes
            .iter()
            .enumerate()
            .map(|(index, envelope)| {
                let offset = u64::try_from(index).map_err(|error| LiminalError::PublishFailed {
                    message: format!("durable batch index overflow: {error}"),
                })?;
                let position =
                    start
                        .checked_add(offset)
                        .ok_or_else(|| LiminalError::PublishFailed {
                            message: "durable batch position overflow".to_owned(),
                        })?;
                let encoded =
                    envelope
                        .serialize()
                        .map_err(|error| LiminalError::PublishFailed {
                            message: format!("durable batch envelope encoding failed: {error}"),
                        })?;
                Ok(UnresolvedAppend::new(
                    self.channel.channel_id().to_owned(),
                    0,
                    position,
                    encoded,
                    None,
                ))
            })
            .collect::<Result<Vec<_>, LiminalError>>()?;
        self.batch_fanout = payloads
            .iter()
            .zip(&proofs)
            .map(|(payload, proof)| PendingFanout {
                delivery_id: None,
                payload: payload.clone(),
                publisher: publisher.clone(),
                causal: causal.cloned(),
                position: Some(proof.sequence()),
            })
            .collect();
        match block_on(self.channel.publish_batch(&envelopes)) {
            Ok(Ok(_)) => {}
            outcome => {
                if let Some(append) = self.channel.pending_batch_evidence() {
                    return Err(LiminalError::PublishUnresolved { append });
                }
                self.batch_fanout.clear();
                return Err(LiminalError::PublishFailed {
                    message: format!("durable batch failed: {outcome:?}"),
                });
            }
        }
        let mut queued = Vec::with_capacity(proofs.len());
        for mut proof in proofs {
            let fanout = self
                .batch_fanout
                .front()
                .ok_or_else(|| LiminalError::PublishFailed {
                    message: "committed durable batch lost its fanout".to_owned(),
                })?;
            match core.enqueue_publish(
                fanout.payload.clone(),
                fanout.publisher.clone(),
                fanout.causal.clone(),
                fanout.position,
                None,
            ) {
                Ok(pending) => {
                    self.batch_fanout.pop_front();
                    queued.push((pending, proof));
                }
                Err(error) => {
                    proof.set_cause(format!("batch committed; live enqueue failed: {error}"));
                    return Err(LiminalError::PublishUnresolved { append: proof });
                }
            }
        }
        Ok(queued)
    }
}