ruststream-rdkafka 0.7.1

//! The subscription descriptors and the group settings they share.
//!
//! Kafka reads a topic in three shapes, and each is a type of its own because they differ in
//! what a deferred retry copy can be published to: [`KafkaTopic`] reads one topic and a publish
//! to that topic reaches it again, while [`KafkaTopics`] (several topics, or a regex) and
//! [`KafkaPartitions`] (named partitions of one topic) read a set no single publish addresses.

mod partitions;
mod topic;
mod topics;

pub use partitions::KafkaPartitions;
pub use topic::KafkaTopic;
pub use topics::KafkaTopics;

use rdkafka::error::KafkaError as RdKafkaError;

use crate::error::KafkaError;
use crate::properties::Properties;

/// Where a consumer group starts reading when it has no valid committed offset.
///
/// Kafka resumes from the group's committed position when a valid one exists; this choice (it
/// maps to librdkafka's `auto.offset.reset`) applies when there is none - the group has never
/// committed the partition, or the committed offset was deleted by retention / is out of
/// range. The second case is why it matters for long-idle groups: with the librdkafka default
/// (latest) an expired group skips to the end instead of reprocessing.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[non_exhaustive]
pub enum StartOffset {
    /// Leave the choice to librdkafka (its default resets to the latest offset).
    #[default]
    Committed,
    /// Start from the earliest retained offset.
    Earliest,
    /// Start from the latest offset: only messages published after the subscription opened.
    Latest,
}

/// The partition assignment strategy for the consumer group (librdkafka's
/// `partition.assignment.strategy`).
///
/// These are librdkafka's built-in strategies; the client offers no API for a custom group
/// assignor (the rebalance callback only observes assignments). Cooperative and eager
/// strategies cannot mix within one group - librdkafka rejects the join, and the error
/// surfaces on the subscriber stream.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum Assignment {
    /// Co-partitioned ranges per topic (the Kafka default family).
    Range,
    /// Round-robin across all subscribed topics.
    RoundRobin,
    /// Incremental cooperative rebalancing: unaffected partitions keep flowing during a
    /// rebalance instead of stopping the world.
    CooperativeSticky,
}

impl Assignment {
    pub(crate) fn as_config_value(self) -> &'static str {
        match self {
            Self::Range => "range",
            Self::RoundRobin => "roundrobin",
            Self::CooperativeSticky => "cooperative-sticky",
        }
    }
}

/// What drives keyed worker lanes (`workers(n, by_key)`) for this subscription.
///
/// The runtime lanes deliveries by [`IncomingMessage::partition_key`]
/// (deliveries sharing a lane key process in order on one lane); this choice picks what that
/// key is for Kafka.
///
/// [`IncomingMessage::partition_key`]: ruststream::IncomingMessage::partition_key
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[non_exhaustive]
pub enum LaneKey {
    /// The source partition (the default): lanes mirror Kafka's own ordering unit, so
    /// everything a partition delivers (keyless included) processes in order on one lane.
    #[default]
    Partition,
    /// The native record key: per-key ordering, finer than a partition, so messages of one
    /// partition may process concurrently when their keys differ. Keyless deliveries carry no
    /// lane key and rotate across lanes, losing their partition order.
    RecordKey,
}

/// How processed deliveries are committed back to the consumer group.
///
/// # Examples
///
/// A tracked subscription with one lane per record key, a cap on deliveries and a dead-letter
/// topic:
///
/// ```
/// # #[cfg(feature = "json")]
/// # mod demo {
/// use ruststream_rdkafka::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Deserialize)]
/// struct Payment {
///     id: u64,
/// }
///
/// /// One lane per record key, so payments of one account stay ordered while others run
/// /// concurrently.
/// #[subscriber(
///     KafkaTopic::new("payments")
///         .group("payments-svc")
///         .commit(Commit::Tracked)
///         .start(StartOffset::Earliest)
///         .lane_key(LaneKey::RecordKey),
///     workers(8, by_key)
/// )]
/// async fn charge(payment: &Payment) -> HandlerOutcome {
///     if payment.id == 0 {
///         return HandlerOutcome::drop();
///     }
///     HandlerOutcome::ack()
/// }
///
/// fn app() -> RustStream {
///     RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(
///         KafkaBroker::new(["localhost:9092"]),
///         |b| {
///             // Five deliveries of one payment, the first included; the sixth goes to the
///             // dead-letter topic instead, payload and headers as they arrived.
///             b.include(charge)
///                 .max_attempts(nonzero!(5u32))
///                 .dead_letter("payments.dlq");
///         },
///     )
/// }
/// # }
/// # fn main() {}
/// ```
#[derive(Debug, Clone, PartialEq, Eq, Default)]
#[non_exhaustive]
pub enum Commit {
    /// librdkafka auto-commit, the librdkafka default: positions are stored as messages are
    /// handed to the application and committed every `auto.commit.interval.ms`. `ack` and
    /// `nack(false)` are advisory no-ops and `nack(true)` reports
    /// [`AckError::Unsupported`](ruststream::AckError::Unsupported); a crash can lose the tail of
    /// processed-but-uncommitted work or skip unprocessed deliveries that were already stored.
    #[default]
    Auto,
    /// Per-message acknowledgement: `enable.auto.offset.store` is switched off and an `ack`
    /// advances the stored position to just below the lowest still-unsettled delivery (or to
    /// the highest delivered offset once none are outstanding). At-least-once stays precise
    /// with concurrent handler lanes, and offset gaps the consumer never receives (transaction
    /// markers, compacted-away records) cannot block the position. Auto-commit still flushes
    /// the stored position in the background and once more when the consumer closes.
    Tracked,
    /// Exactly-once: the consumer never commits its own offsets - the
    /// [`EosPipeline`](crate::EosPipeline) whose transactional id matches this name commits
    /// them through the producer transaction (`send_offsets_to_transaction`), so source
    /// positions move atomically with the records the handlers publish. `enable.auto.commit`
    /// and `enable.auto.offset.store` are switched off; `ack` advances the shared watermark
    /// exactly like [`Tracked`](Self::Tracked), and the pipeline picks the watermark up at its
    /// next window commit.
    Transactional(String),
}

/// The consumer-group settings every subscription form carries.
///
/// One struct rather than six repeated fields: the three descriptors differ in what they read,
/// not in how the group behaves around it.
#[derive(Debug, Clone, Default)]
pub(crate) struct GroupSettings {
    pub(crate) group: Option<String>,
    pub(crate) start: StartOffset,
    pub(crate) commit: Commit,
    pub(crate) assignment: Option<Assignment>,
    pub(crate) lane_key: LaneKey,
    pub(crate) config: Properties,
}

/// How the consumer takes its partitions: through the group protocol, or by naming them.
#[derive(Debug)]
pub(crate) enum Reader {
    /// librdkafka `subscribe()` to these names; an entry starting with `^` is a topic regex.
    Subscribed(Vec<String>),
    /// librdkafka `assign()` of exactly these partitions of one topic: no group membership,
    /// no rebalancing.
    Assigned { topic: String, partitions: Vec<i32> },
}

/// Whether Kafka accepts `name` as a topic: 1 to 249 characters from `[a-zA-Z0-9._-]`, and not
/// `.` or `..`.
pub(crate) fn legal_topic(name: &str) -> bool {
    !name.is_empty()
        && name.len() <= 249
        && name != "."
        && name != ".."
        && name
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-'))
}

/// Refuses a subscription to a topic name the cluster would refuse. librdkafka accepts such a
/// subscription and reports the name only later, on the consumer's queue, so the subscription
/// would open and deliver nothing; refusing it here fails the service at startup instead.
/// Patterns are regexes and are left to librdkafka.
pub(crate) fn refuse_illegal_topics(reader: &Reader) -> Result<(), KafkaError> {
    let illegal = match reader {
        Reader::Subscribed(names) => names
            .iter()
            .find(|name| !name.starts_with('^') && !legal_topic(name)),
        Reader::Assigned { topic, .. } => Some(topic).filter(|topic| !legal_topic(topic)),
    };
    illegal.map_or(Ok(()), |topic| {
        Err(KafkaError::subscribe(RdKafkaError::Subscription(format!(
            "{topic:?} is not a legal Kafka topic name"
        ))))
    })
}

/// A descriptor resolved into what opening a consumer needs, with nothing left to validate.
#[derive(Debug)]
pub(crate) struct SubscriptionPlan {
    /// The handler-metadata name, and the channel the generated document reports.
    pub(crate) name: String,
    pub(crate) reader: Reader,
    pub(crate) settings: GroupSettings,
}

/// Rejects a name that librdkafka would read as a topic regex.
///
/// The check is at startup because the name is a runtime string: a mount site may take it from
/// configuration, so no type can carry the anchor.
pub(crate) fn reject_pattern(name: &str, form: &str) -> Result<(), KafkaError> {
    if name.starts_with('^') {
        return Err(KafkaError::InvalidOptions(format!(
            "{form} reads one topic, and {name:?} is a librdkafka topic regex (the leading '^' \
             is its anchor); subscribe to a pattern with `KafkaTopics::pattern`"
        )));
    }
    Ok(())
}

/// The librdkafka properties a commit mode decides for itself.
///
/// [`Commit::Tracked`] and [`Commit::Transactional`] settle by a position this crate computes,
/// and both of these say who owns that position. A passthrough setting one does not tune the
/// mode, it undoes it: the acks keep running and decide nothing, which is a silent loss of the
/// guarantee the mode was chosen for.
const MODE_OWNED_PROPERTIES: [&str; 2] = ["enable.auto.offset.store", "enable.auto.commit"];

/// Rejects a passthrough that would take a commit mode's own property away from it.
///
/// The passthrough is applied last and wins over every typed option, which is what makes it an
/// escape hatch - and what would make this particular clash invisible. [`Commit::Auto`] leaves
/// the position to librdkafka and owns neither property, so there the passthrough keeps winning.
pub(crate) fn reject_commit_mode_clash(
    name: &str,
    commit: &Commit,
    config: &[(String, String)],
) -> Result<(), KafkaError> {
    let mode = match commit {
        Commit::Auto => return Ok(()),
        Commit::Tracked => "`Commit::Tracked`",
        Commit::Transactional(_) => "`Commit::Transactional`",
    };
    for (key, value) in config {
        let Some(owned) = MODE_OWNED_PROPERTIES
            .iter()
            .find(|owned| key.eq_ignore_ascii_case(owned))
        else {
            continue;
        };
        return Err(KafkaError::InvalidOptions(format!(
            "subscription to {name:?} sets {owned}={value:?} through its config passthrough, \
             and {mode} owns that property: the passthrough would leave the committed position \
             to librdkafka while every ack decided nothing. Drop it, or take `Commit::Auto`, \
             which owns neither"
        )));
    }
    Ok(())
}

/// Rejects an empty topic name, whichever form named it.
pub(crate) fn reject_empty(name: &str) -> Result<(), KafkaError> {
    if name.is_empty() {
        return Err(KafkaError::InvalidOptions(
            "topic name must not be empty; name the topic the handler consumes".to_owned(),
        ));
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn topic_names_follow_kafkas_rule() {
        assert!(legal_topic("orders.eu-1_v2"));
        for bad in ["", ".", "..", "orders eu", "orders/eu", &"x".repeat(250)] {
            assert!(!legal_topic(bad), "{bad:?} must be refused");
        }
    }
}