liminal-server 0.20.0

//! A keyed publish to a durable channel claims its idempotency key with its
//! evidence in one write before the publish: the channel, and where each of its
//! partitions stood. It holds the claim across an append whose outcome is not
//! known, and releases or completes a held claim only after looking the row up
//! by its key from those positions: at the channel's next keyed admission, on
//! a same-key retry, and when the services start.
//!
//! The held keys are a cache of the durable in-flight claims, rebuilt from them
//! on start (`unresolved_start.rs`). Nothing here runs on a timer: a held key is
//! resolved only when a publish or a start gives it the chance.
//!
//! A held key is never appended again, so once a lookup has answered for it the
//! answer stands: a settlement whose write fails keeps the answer and retries
//! only the write. A key that keeps failing to settle is tried by the channel's
//! admissions less and less often (after 1, 3, 7, ... admissions, up to
//! [`MAX_ADMISSION_BACKOFF`]); a same-key retry always tries it.
//!
//! A key stays held, marked as settling, until its settlement write has landed.
//! A retry that meets it then, or that finds the key settled while it read the
//! claim, is answered as unresolved so the producer retries, never as accepted.
use liminal::channel::PublishRecord;
use std::collections::BTreeMap;
use std::sync::{Mutex, MutexGuard, PoisonError};

use liminal::durability::bridge::block_on;
use liminal::durability::{AppendResolution, ClaimEvidence, DedupDecision, ProcessingReceipt};
use liminal::error::LiminalError;
use liminal::protocol::MessageEnvelope;

use super::{
    ConfiguredChannel, LiminalConnectionServices, PublishOutcome, dedup_timestamp_millis,
    publisher_for_origin,
};
use crate::ServerError;
use crate::server::self_streams::events::PublishOrigin;

/// The most admissions a held key that keeps failing to settle is passed over.
const MAX_ADMISSION_BACKOFF: u32 = 1 << 10;

/// What the services know about an in-flight claim they hold.
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) enum HeldKey {
    /// The claim names the durable channel whose stream answers for it, and
    /// where each partition stood when the claim was made.
    Checkable { channel: String, from: Vec<u64> },
    /// The claim carries no evidence, so nothing can say whether its row exists.
    Uncheckable,
}

/// One held key and how far its settlement has got.
#[derive(Debug)]
struct Held {
    key: HeldKey,
    /// The lookup's answer, once one has been read.
    found: Option<AppendResolution>,
    /// Whether a resolver is writing the claim's settlement.
    settling: bool,
    /// Settlements that failed in a row.
    failures: u32,
    /// Admissions still to pass over before the next attempt.
    skip: u32,
}

impl Held {
    fn begin(&mut self) -> Turn {
        if self.settling {
            return Turn::Settling;
        }
        self.settling = true;
        Turn::Settle {
            key: self.key.clone(),
            found: self.found,
        }
    }

    const fn new(key: HeldKey) -> Self {
        Self {
            key,
            found: None,
            settling: false,
            failures: 0,
            skip: 0,
        }
    }
}

#[derive(Debug, Default)]
struct HeldState {
    entries: BTreeMap<String, Held>,
    /// How many held keys have stopped being held, so a publish can tell a
    /// claim it read was settled under it.
    released: u64,
}

impl HeldState {
    fn release(&mut self, idempotency_key: &str) {
        if self.entries.remove(idempotency_key).is_some() {
            self.released = self.released.wrapping_add(1);
        }
    }
}

/// What a resolver may do with a held key.
enum Turn {
    /// The key is not held.
    NotHeld,
    /// Another resolver is settling it.
    Settling,
    /// This resolver settles it.
    Settle {
        key: HeldKey,
        found: Option<AppendResolution>,
    },
}

/// The index of held keys, by idempotency key.
#[derive(Debug, Default)]
pub(super) struct HeldKeys(Mutex<HeldState>);

impl HeldKeys {
    fn state(&self) -> MutexGuard<'_, HeldState> {
        // The map is a cache of durable claims; a panic elsewhere cannot leave
        // it inconsistent with them, so a poisoned lock is still usable.
        self.0.lock().unwrap_or_else(PoisonError::into_inner)
    }

    pub(super) fn hold(&self, idempotency_key: &str, held: HeldKey) {
        let mut state = self.state();
        match state.entries.get_mut(idempotency_key) {
            Some(entry) => {
                entry.key = held;
                entry.found = None;
            }
            None => {
                state
                    .entries
                    .insert(idempotency_key.to_owned(), Held::new(held));
            }
        }
    }

    fn forget(&self, idempotency_key: &str) {
        self.state().release(idempotency_key);
    }

    fn get(&self, idempotency_key: &str) -> Option<HeldKey> {
        self.state()
            .entries
            .get(idempotency_key)
            .map(|held| held.key.clone())
    }

    fn released(&self) -> u64 {
        self.state().released
    }

    /// The checkable keys of `channel` an admission should try now, passing
    /// over once each key still backing off.
    fn due_for(&self, channel: &str) -> Vec<String> {
        self.state()
            .entries
            .iter_mut()
            .filter(|(_, held)| {
                !held.settling
                    && matches!(&held.key, HeldKey::Checkable { channel: named, .. } if named == channel)
            })
            .filter_map(|(key, held)| {
                if held.skip == 0 {
                    Some(key.clone())
                } else {
                    held.skip -= 1;
                    None
                }
            })
            .collect()
    }

    /// Marks `idempotency_key` as settling when no resolver has it.
    fn begin(&self, idempotency_key: &str) -> Turn {
        self.state()
            .entries
            .get_mut(idempotency_key)
            .map_or(Turn::NotHeld, Held::begin)
    }

    /// Keeps a lookup's answer for the settlement write.
    fn found(&self, idempotency_key: &str, resolution: AppendResolution) {
        if let Some(held) = self.state().entries.get_mut(idempotency_key) {
            held.found = Some(resolution);
        }
    }

    /// The settlement did not land: the key stays held and backs off.
    fn failed(&self, idempotency_key: &str) {
        if let Some(held) = self.state().entries.get_mut(idempotency_key) {
            held.settling = false;
            held.failures = held.failures.saturating_add(1);
            held.skip = 1_u32
                .checked_shl(held.failures)
                .map_or(MAX_ADMISSION_BACKOFF, |backoff| {
                    backoff.min(MAX_ADMISSION_BACKOFF)
                })
                - 1;
        }
    }

    /// The settlement landed: the key stops being held.
    fn settled(&self, idempotency_key: &str) {
        self.state().release(idempotency_key);
    }
}

/// How a looked-up claim was settled, for the log line that names it.
#[derive(Clone, Copy)]
enum Settled {
    Committed,
    Absent,
}

impl Settled {
    const fn outcome(self) -> &'static str {
        match self {
            Self::Committed => "committed",
            Self::Absent => "absent",
        }
    }
}

fn unresolved(
    channel: &str,
    idempotency_key: &str,
    sequence: Option<u64>,
    message: String,
) -> ServerError {
    ServerError::PublishUnresolved {
        channel: channel.to_owned(),
        idempotency_key: Some(idempotency_key.to_owned()),
        sequence,
        message,
    }
}

impl LiminalConnectionServices {
    /// A publish carrying `idempotency_key` to the durable channel `channel`.
    pub(super) fn publish_durable_keyed(
        &self,
        configured: &ConfiguredChannel,
        channel: &str,
        publication: (&MessageEnvelope, PublishOrigin<'_>),
        idempotency_key: &str,
        recipients: Option<&mut PublishRecord>,
    ) -> Result<PublishOutcome, ServerError> {
        for held in self.held_keys.due_for(channel) {
            // A key whose claim cannot be settled stays held, with its claim,
            // for a later admission or retry; this publish does not wait on it.
            if let Err(error) = self.settle_held(configured, channel, &held) {
                tracing::debug!(channel, idempotency_key = %held, %error, "held key still unresolved");
            }
        }
        let released = self.held_keys.released();
        let (decision, from) = self.claim_with_evidence(configured, channel, idempotency_key)?;
        match decision {
            DedupDecision::Claimed => self.publish_claimed(
                configured,
                channel,
                publication,
                idempotency_key,
                from,
                recipients,
            ),
            DedupDecision::Completed(receipt) => Ok(self.suppressed(receipt.durable_position())),
            DedupDecision::InFlight => match self.held_keys.get(idempotency_key) {
                // A held key settled while this publish read its claim: the
                // claim it read may be gone with no row behind it.
                None if self.held_keys.released() != released => Err(unresolved(
                    channel,
                    idempotency_key,
                    None,
                    "the claim was settled while this publish read it".to_owned(),
                )),
                None => Ok(self.suppressed(None)),
                Some(HeldKey::Uncheckable) => {
                    tracing::error!(idempotency_key, "in-flight claim cannot be checked");
                    Err(unresolved(
                        channel,
                        idempotency_key,
                        None,
                        "the claim carries no evidence of where its row would be".to_owned(),
                    ))
                }
                Some(HeldKey::Checkable {
                    channel: evidence, ..
                }) => {
                    let evidence_channel = if evidence == channel {
                        configured.handle.clone()
                    } else {
                        self.admit_channel(&evidence)
                            .map(|entry| entry.handle.clone())
                            .map_err(|error| {
                                unresolved(&evidence, idempotency_key, None, error.to_string())
                            })?
                    };
                    match self.settle_held_with(&evidence_channel, &evidence, idempotency_key)? {
                        Some(position) => Ok(self.suppressed(Some(position))),
                        None => {
                            match self.claim_with_evidence(configured, channel, idempotency_key)? {
                                (DedupDecision::Claimed, from) => self.publish_claimed(
                                    configured,
                                    channel,
                                    publication,
                                    idempotency_key,
                                    from,
                                    recipients,
                                ),
                                (DedupDecision::Completed(receipt), _) => {
                                    Ok(self.suppressed(receipt.durable_position()))
                                }
                                (DedupDecision::InFlight, _) => Ok(self.suppressed(None)),
                            }
                        }
                    }
                }
            },
        }
    }

    /// Claims `idempotency_key` with `channel` and where its partitions stand
    /// as its evidence, in one write made before the publish, answering the
    /// decision and those positions. A failed write may still have landed, so
    /// the key is held: a same-key retry then looks its row up instead of
    /// being suppressed by a claim nobody would ever resolve.
    fn claim_with_evidence(
        &self,
        configured: &ConfiguredChannel,
        channel: &str,
        idempotency_key: &str,
    ) -> Result<(DedupDecision, Vec<u64>), ServerError> {
        let from =
            configured
                .handle
                .durable_positions()
                .map_err(|error| ServerError::ListenerAccept {
                    message: format!("durable channel '{channel}' positions unavailable: {error}"),
                })?;
        let claimed = block_on(self.dedup.claim_with_evidence(
            idempotency_key,
            ClaimEvidence::new(channel, from.clone()),
            dedup_timestamp_millis(),
        ))
        .map_err(|error| format!("dedup bridge failed for key '{idempotency_key}': {error}"))
        .and_then(|claim| {
            claim
                .map_err(|error| format!("dedup claim failed for key '{idempotency_key}': {error}"))
        });
        match claimed {
            Ok(decision) => Ok((decision, from)),
            Err(message) => {
                self.held_keys.hold(
                    idempotency_key,
                    HeldKey::Checkable {
                        channel: channel.to_owned(),
                        from,
                    },
                );
                Err(ServerError::ListenerAccept { message })
            }
        }
    }

    /// Publishes a freshly claimed key and settles its claim by the outcome.
    fn publish_claimed(
        &self,
        configured: &ConfiguredChannel,
        channel: &str,
        publication: (&MessageEnvelope, PublishOrigin<'_>),
        idempotency_key: &str,
        from: Vec<u64>,
        recipients: Option<&mut PublishRecord>,
    ) -> Result<PublishOutcome, ServerError> {
        let (envelope, origin) = publication;
        let delivery = recipients.map_or_else(
            || {
                configured.handle.publish_with_idempotency_key(
                    &envelope.payload,
                    publisher_for_origin(origin),
                    None,
                    idempotency_key,
                )
            },
            |recipients| {
                configured.handle.publish_recording(
                    &envelope.payload,
                    publisher_for_origin(origin),
                    None,
                    Some(idempotency_key),
                    recipients,
                )
            },
        );
        let delivery = match delivery {
            Ok(delivery) => delivery,
            Err(LiminalError::PublishUnresolved { append }) => {
                // The row may be in the stream: the claim stays, and a later
                // admission or retry looks the row up by its key.
                self.held_keys.hold(
                    idempotency_key,
                    HeldKey::Checkable {
                        channel: channel.to_owned(),
                        from,
                    },
                );
                return Err(unresolved(
                    channel,
                    idempotency_key,
                    Some(append.sequence()),
                    append.to_string(),
                ));
            }
            Err(error) => {
                Self::release_claim(&self.dedup, idempotency_key);
                self.held_keys.forget(idempotency_key);
                return Err(ServerError::ListenerAccept {
                    message: format!("liminal publish failed for channel '{channel}': {error}"),
                });
            }
        };
        // A1 §5, as for every keyed publish: an admission completes the claim
        // with the receipt written today, a Reject releases it.
        if delivery.is_admitted() {
            let Some(position) = delivery.durable_position() else {
                self.held_keys.hold(
                    idempotency_key,
                    HeldKey::Checkable {
                        channel: channel.to_owned(),
                        from,
                    },
                );
                return Err(unresolved(
                    channel,
                    idempotency_key,
                    None,
                    "admitted durable publication has no store position".to_owned(),
                ));
            };
            self.complete_claim(idempotency_key, ProcessingReceipt::for_position(position))?;
        } else {
            Self::release_claim(&self.dedup, idempotency_key);
        }
        self.held_keys.forget(idempotency_key);
        Ok(self.delivered(&delivery))
    }

    /// Looks a held key's row up on `channel` and settles the claim.
    fn settle_held(
        &self,
        configured: &ConfiguredChannel,
        channel: &str,
        idempotency_key: &str,
    ) -> Result<Option<u64>, ServerError> {
        self.settle_held_with(&configured.handle, channel, idempotency_key)
    }

    /// Looks `idempotency_key`'s row up on `handle`, the channel its claim
    /// names, from the positions the claim recorded. A row completes the claim
    /// with its position, which is answered; no row releases the claim,
    /// answered as `None`. The key stays held, marked as settling, until that
    /// write lands, and then stops being held. An unreadable stream or a
    /// failed write keeps the claim and the key held.
    pub(super) fn settle_held_with(
        &self,
        handle: &liminal::channel::ChannelHandle,
        channel: &str,
        idempotency_key: &str,
    ) -> Result<Option<u64>, ServerError> {
        // Exactly one resolver settles a key: whoever marks it settling. A
        // publish that meets it settling, or finds it already settled, cannot
        // say what its claim became and is answered as unresolved.
        let (from, found) = match self.held_keys.begin(idempotency_key) {
            Turn::NotHeld => {
                return Err(unresolved(
                    channel,
                    idempotency_key,
                    None,
                    "the claim was settled by another publish".to_owned(),
                ));
            }
            Turn::Settling => {
                return Err(unresolved(
                    channel,
                    idempotency_key,
                    None,
                    "the claim is being settled by another publish".to_owned(),
                ));
            }
            Turn::Settle {
                key: HeldKey::Checkable { from, .. },
                found,
            } => (from, found),
            Turn::Settle {
                key: HeldKey::Uncheckable,
                ..
            } => {
                self.held_keys.failed(idempotency_key);
                return Err(unresolved(
                    channel,
                    idempotency_key,
                    None,
                    "the claim carries no evidence of where its row would be".to_owned(),
                ));
            }
        };
        let resolution = match found {
            Some(resolution) => resolution,
            None => match handle.resolve_idempotency_key(idempotency_key, &from) {
                Ok(resolution) => {
                    self.held_keys.found(idempotency_key, resolution);
                    resolution
                }
                Err(error) => {
                    self.held_keys.failed(idempotency_key);
                    return Err(unresolved(
                        channel,
                        idempotency_key,
                        None,
                        error.to_string(),
                    ));
                }
            },
        };
        let position = match resolution {
            AppendResolution::Committed { position } => Some(position),
            AppendResolution::Absent => None,
        };
        let (settled, written) = position.map_or_else(
            || (Settled::Absent, self.release_held_claim(idempotency_key)),
            |position| {
                let receipt = ProcessingReceipt::for_position(position);
                (
                    Settled::Committed,
                    self.complete_claim(idempotency_key, receipt),
                )
            },
        );
        if let Err(error) = written {
            self.held_keys.failed(idempotency_key);
            return Err(error);
        }
        self.held_keys.settled(idempotency_key);
        tracing::info!(
            channel,
            idempotency_key,
            outcome = settled.outcome(),
            "unresolved publish resolved"
        );
        Ok(position)
    }

    /// Releases a held claim whose row the stream does not hold.
    fn release_held_claim(&self, idempotency_key: &str) -> Result<(), ServerError> {
        block_on(self.dedup.release_claim(idempotency_key))
            .map_err(|error| error.to_string())
            .and_then(|release| release.map_err(|error| error.to_string()))
            .map_err(|message| ServerError::ListenerAccept {
                message: format!(
                    "dedup release failed for unresolved key '{idempotency_key}': {message}"
                ),
            })
    }

    fn complete_claim(
        &self,
        idempotency_key: &str,
        receipt: ProcessingReceipt,
    ) -> Result<(), ServerError> {
        block_on(self.dedup.complete_receipt(idempotency_key, receipt))
            .map_err(|error| ServerError::ListenerAccept {
                message: format!(
                    "dedup receipt bridge failed for key '{idempotency_key}': {error}"
                ),
            })?
            .map_err(|error| ServerError::ListenerAccept {
                message: format!("dedup receipt write failed for key '{idempotency_key}': {error}"),
            })
    }

    /// The outcome of a same-key publish answered from its claim: accepted,
    /// delivered to no subscriber, carrying the original row's position when
    /// the claim knows it.
    fn suppressed(&self, durable_position: Option<u64>) -> PublishOutcome {
        crate::metrics::publish_accepted();
        PublishOutcome {
            message_id: self.next_message_id(),
            delivered: false,
            durable_position,
            reach: super::Reach::Suppressed,
        }
    }
}