diff --git a/core/sdk/src/clients/producer.rs b/core/sdk/src/clients/producer.rs index a6c3e57918..23b945c6a3 100644 --- a/core/sdk/src/clients/producer.rs +++ b/core/sdk/src/clients/producer.rs @@ -474,6 +474,321 @@ impl ProducerCoreBackend for ProducerCore { unsafe impl Send for IggyProducer {} unsafe impl Sync for IggyProducer {} +/// Appends messages to one topic of one stream. +/// +/// A topic is split into partitions, and a partition is an ordered log that producers append to. +/// `IggyProducer` lets you configure [where](#where-messages-land-and-ordering) and +/// [how](#how-messages-are-sent) messages are sent. +/// +/// # Creating a producer +/// +/// The easiest way to create a producer is through an [`IggyClient`] with a configured connection. +/// [`IggyClient::producer()`] returns an [`IggyProducerBuilder`] that uses that client's connection. +/// +/// Building never talks to the server. [`init()`](Self::init) must be awaited before the first send. +/// +/// # Examples +/// +/// A producer with the defaults, sending one batch and reading the confirmations: +/// +/// ```rust,no_run +/// use iggy::prelude::*; +/// use std::str::FromStr; +/// +/// # async fn example() -> Result<(), IggyError> { +/// let client = IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?; +/// client.connect().await?; +/// +/// let producer = client.producer("my-stream", "my-topic")?.build(); +/// producer.init().await?; +/// +/// let messages = vec![IggyMessage::from_str("hello")?, IggyMessage::from_str("world")?]; +/// let response = producer.send(messages).await?; +/// for confirmation in &response.confirmations { +/// println!( +/// "Partition: {}, base offset: {}", +/// confirmation.partition_id, confirmation.base_offset +/// ); +/// } +/// # Ok(()) +/// # } +/// ``` +/// +/// A `background` producer, which queues a batch and sends it later, and shuts down cleanly: +/// +/// ```rust,no_run +/// use iggy::prelude::*; +/// use std::str::FromStr; +/// +/// # async fn example() -> Result<(), IggyError> { +/// let client = IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?; +/// client.connect().await?; +/// +/// let producer = client +/// .producer("my-stream", "my-topic")? +/// .background( +/// BackgroundConfig::builder() +/// .linger_time(IggyDuration::new_from_secs(1)) +/// .batch_size(64 * 1024) +/// .build(), +/// ) +/// .build(); +/// producer.init().await?; +/// +/// // Returns once the batch is queued. The send itself happens on a background worker. +/// producer.send_one(IggyMessage::from_str("hello")?).await?; +/// +/// // Without graceful shutdown, buffered messages have no completion guarantee and may be lost. +/// producer.shutdown().await; +/// # Ok(()) +/// # } +/// ``` +/// +/// Keying messages to a partition and separating confirmed chunks from the unconfirmed tail: +/// +/// ```rust,no_run +/// use iggy::prelude::*; +/// use std::str::FromStr; +/// +/// # async fn example() -> Result<(), IggyError> { +/// let client = IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?; +/// client.connect().await?; +/// +/// let producer = client +/// .producer("my-stream", "my-topic")? +/// // Every message of this producer goes to the partition the server derives from the key. +/// .partitioning(Partitioning::messages_key_str("my-key")?) +/// .send_retries(Some(5), Some(NonZeroIggyDuration::ONE_SECOND)) +/// .build(); +/// producer.init().await?; +/// +/// let messages = vec![IggyMessage::from_str("hello")?]; +/// if let Err(IggyError::ProducerSendFailed { cause, failed, committed, .. }) = +/// producer.send(messages).await +/// { +/// // `committed` holds confirmations for earlier chunks, `failed` the unconfirmed tail. See +/// // "Retrying and what a failure means" before resending `failed`. +/// eprintln!("{} messages have no usable confirmation: {cause}", failed.len()); +/// eprintln!("{} chunk(s) were confirmed before the failure", committed.len()); +/// } +/// # Ok(()) +/// # } +/// ``` +/// +/// # How messages are sent +/// +/// There are two options: [`direct()`] and [`background()`]. You will pick one of the two modes, +/// and a producer stays in that mode for its lifetime. +/// +/// A **direct** producer (the default) sends from the calling task. [`send()`](Self::send) awaits +/// the server and returns its [confirmations](#confirmations). A batch longer than +/// [`DirectConfig::batch_length`] is split into that many messages per request, and the requests are +/// awaited one after another, so a failure in the middle leaves confirmations for the successful +/// prefix. [`DirectConfig::linger_time`] requests a minimum gap between sequential send calls. It +/// does not space out chunks within one call and does not serialize concurrent callers. +/// +/// A **background** producer hands the batch to a [`ProducerDispatcher`] and returns after queueing +/// it without waiting for the write. A worker may already have started or completed the write by +/// then, but [`send()`](Self::send) reports no write result and returns no confirmations. The +/// dispatcher runs [`BackgroundConfig::num_shards`] workers, each buffering the batches routed to it +/// and flushing them when one of three limits is hit: [`BackgroundConfig::batch_length`] queued +/// sends, [`BackgroundConfig::batch_size`] bytes, or [`BackgroundConfig::linger_time`] since the +/// first of them was buffered. Adjacent buffered sends that share a stream, topic, and partitioning +/// are merged into one request. +/// +/// The sharding strategy decides how background dispatch affects message order: +/// - [`BackgroundConfig::sharding`] routes a batch to a worker. The default [`OrderedSharding`] picks +/// it from the stream and the topic, so everything going to one topic stays on one worker and keeps +/// its order. [`BalancedSharding`] spreads batches round-robin and gives up ordering, which can +/// improve throughput when multiple shards are allowed to write concurrently. +/// - [`BackgroundConfig::max_in_flight`] bounds concurrent writes across workers. A worker itself +/// remains sequential for every value and awaits retries before starting its next request. Raising +/// this setting does not break the per-topic order provided by [`OrderedSharding`], but it allows +/// strategies that spread one topic across workers to write those shards concurrently. +/// +/// Since a background send is queued rather than written, the dispatcher charges queued and +/// in-flight sends against [`BackgroundConfig::max_buffer_size`] over all workers. The default is +/// bounded, while a value of zero disables the byte budget. +/// [`BackgroundConfig::failure_mode`] decides what a send does when that budget is exhausted. You can block +/// until it frees up (the default), block with a timeout and fail with +/// [`IggyError::BackgroundSendTimeout`], or fail right away with +/// [`IggyError::BackgroundSendBufferOverflow`]. A single send larger than the whole budget always +/// fails with [`IggyError::BackgroundSendBufferOverflow`], whatever the mode. +/// +/// In contrast to the direct mode, which awaits the confirmations from the server, a background send +/// has no caller left to return to. Write failures are reported to +/// [`BackgroundConfig::error_callback`] instead. It receives the cause, the unconfirmed tail, and +/// confirmations returned for earlier chunks, see +/// [Retrying and what a failure means](#retrying-and-what-a-failure-means). The default callback +/// logs the context and drops it. A custom [`ErrorCallback`] can retain or persist failed sends +/// according to the application's at-least-once policy. +/// +/// # Where messages land and ordering +/// +/// The partition is the unit of order in Iggy. Inside one partition, messages receive increasing +/// offsets in the server's append order. Between partitions there is no global order, so a consumer +/// reading several partitions can observe their messages interleaved. +/// +/// Two messages therefore stay in order only if both of these hold: +/// 1. they are appended to the **same partition**, and +/// 2. their requests reach the server **one after another**, rather than at the same time. +/// +/// Point 1 is what the partitioning strategy and point 2 is what the send mode and the client's own +/// concurrency decide. +/// +/// | Setting | Order | What happens | +/// | --- | --- | --- | +/// | [`Partitioning::balanced()`], the default | not guaranteed with multiple partitions | the server chooses a partition per request, so consecutive sends and chunks may land in different logs | +/// | [`Partitioning::partition_id()`], [`Partitioning::messages_key()`] with a stable key, [`partitioner()`] returning a stable id | same partition | every batch is routed to the same log, satisfying the first ordering requirement | +/// | one task, awaiting each [`send()`](Self::send) before the next | sequential | requests reach a partition in call order | +/// | several tasks sharing the producer, or overlapping sends | not guaranteed | requests race, so call order does not determine append order | +/// | `direct` producer | sequential within one call | the calling task writes that call's chunks one after another | +/// | `background` producer with [`OrderedSharding`], the default | sequential per stream/topic pair | one pair is bound to one worker, which writes its queue in order | +/// | `background` producer with [`BalancedSharding`] and multiple shards | not guaranteed | consecutive batches can reach different workers, so a later batch may be written first | +/// | [`BackgroundConfig::max_in_flight`] | depends on sharding | it controls concurrency between workers, while each worker remains sequential | +/// +/// In short, per-partition order survives if you name the partition and let one sequential writer +/// write to it, for example keyed or fixed partitioning from one task, or a background producer with +/// ordered sharding. [`Partitioning::messages_key()`] maps the same key consistently to a partition +/// for a fixed topic layout. Different keys can share a partition, so this does not create a +/// separate physical log per key. +/// +/// One thing this does not protect against is a message appearing twice. Delivery is at-least-once, +/// so a retried batch can be appended a second time at a higher offset, see +/// [Retrying and what a failure means](#retrying-and-what-a-failure-means). +/// +/// # Confirmations +/// +/// A successful direct send returns [`SendMessagesResponse`], normally holding one +/// [`SendMessagesConfirmationResponse`] per chunk. Each confirmation records a partition and the +/// `base_offset` assigned to the first message in that chunk. A legacy server may return no +/// confirmation payload, and a background producer always returns an empty confirmation list. +/// +/// A confirmation means the server committed the batch in memory. It does not mean the batch was +/// fsynced. After a crash and restart, a later batch can receive an offset that a client recorded +/// before the crash. Delivery is also at least once, so a retry can commit the same messages at +/// another offset. +/// +/// # Retrying and what a failure means +/// +/// A retry is the same request sent again unchanged, meaning a producer never rewrites, splits, or +/// reorders a batch to retry a send. [`send_retries()`] sets how many times it may try and how +/// later retries are paced. The default allows three retries and configures a one-second interval. +/// The first retry is immediate. Later retries wait for the next tick of an interval timer, so they +/// are at most one interval apart, and an attempt that outlasts the interval is followed by the next +/// one right away. Passing `None` as the interval retries back-to-back without any delay. The retry +/// policy applies to both direct and background producers. +/// +/// A request can fail after the server has already appended it, so the first request of an +/// unconfirmed tail may already be in the partition. Sending the tail again can therefore write the +/// same messages twice, leaving the batch in the partition at multiple offsets. A consumer that +/// cannot accept duplicates must recognize them itself. +/// +/// What is retried, and what is not: +/// +/// | Situation | Retried | What the caller ends up with | +/// | --- | --- | --- | +/// | the client is not signed in, so nothing may be sent yet | yes | the send goes ahead as soon as the client is signed in, or fails with [`IggyError::CannotSendMessagesDueToClientDisconnection`] once the retry budget is spent | +/// | the request failed without an indication that it committed | yes, the identical request is sent again | the confirmation of the attempt that finally succeeds, or the last error once the retry budget is spent | +/// | the batch committed, but its confirmation could not be read ([`IggyError::InvalidBytesResponse`] or [`IggyError::InvalidJsonResponse`], raised on the HTTP transport only) | no | the write did happen and retrying would duplicate it on purpose | +/// | encrypting or partitioning the batch failed | no | that cause, with the whole batch returned as unconfirmed because nothing was sent, although earlier messages may already have been encrypted in place | +/// | [`send_retries()`] passed `None` or `0` | no | the outcome of the single attempt, which skips the sign-in check above and fails with the transport error when the client is disconnected | +/// +/// To be precise: the budget is spent per request. Every chunk of a split is a request that gets its own retry budget. +/// Also, waiting for the client to sign in is counted separately from retrying the write itself. +/// +/// Where a failure surfaces is the main difference between the two modes. Either way it names the +/// same three pieces: the `cause`, the unconfirmed tail, and confirmations returned for earlier +/// chunks. +/// +/// - A **direct** send returns them to the caller as [`IggyError::ProducerSendFailed`]. `committed` +/// contains confirmations returned for earlier chunks, while `failed` is the unconfirmed tail. +/// Inspect `cause` before resending it. Encryption mutates messages before sending or +/// partitioning, so `failed` contains encrypted messages when an encryptor is configured. Passing +/// them back to the same producer would encrypt them again. +/// - A **background** send returned to its caller long before the write, so they go to +/// [`BackgroundConfig::error_callback`] as an +/// [`ErrorCtx`](crate::clients::producer_error_callback::ErrorCtx) instead, once no further +/// automatic retry will be attempted. The default callback logs the failure and drops the context. +/// Implement [`ErrorCallback`] when failed sends must be retained. +/// +/// # Options and defaults +/// +/// Everything is configured on the [`IggyProducerBuilder`] before [`build()`] and is fixed +/// afterwards. +/// +/// | Option | Default | Controls | +/// | --- | --- | --- | +/// | [`stream()`], [`topic()`] | the values passed to [`IggyClient::producer()`] | where messages are appended | +/// | [`direct()`] / [`background()`] | [`direct()`] with the [`DirectConfig`] defaults, 1000 messages per request and no linger time | whether a send waits for the write | +/// | [`partitioning()`] | [`Partitioning::balanced()`] | which partition a batch lands in | +/// | [`partitioner()`] | none | computing the partition on the client instead | +/// | [`send_retries()`] | three retries, one-second interval after the immediate first retry | retrying a failed request | +/// | [`create_stream_if_not_exists()`] | on | creating the stream during [`init()`](Self::init) | +/// | [`create_topic_if_not_exists()`] | on, one partition, server defaults for expiry and max size | creating the topic during [`init()`](Self::init) | +/// | [`encryptor()`] | inherited from the client | encrypting payloads and user headers | +/// +/// There are inverse setters as well, such as [`without_partitioning()`], +/// [`without_partitioner()`], [`without_encryptor()`], [`do_not_create_stream_if_not_exists()`] and +/// [`do_not_create_topic_if_not_exists()`]. +/// +/// # Encryption +/// +/// When the [`IggyClient`] was created with an encryptor, a producer built from that client inherits +/// it and encrypts payloads and user headers before a batch leaves the producer. A consumer must use +/// a matching key to decrypt them. Producers and consumers built from the same client inherit the +/// same encryptor unless either builder overrides or clears it. An encryption failure fails the +/// whole send before any request leaves the producer. Encryption runs before a custom +/// [`partitioner()`], so that partitioner observes the encrypted payload and user headers. +/// +/// # Concurrency +/// +/// `IggyProducer` is `Send` and `Sync` but not `Clone`, and every send method takes `&self`. An +/// `Arc` can therefore be shared across tasks without further wrapping. A background +/// dispatcher routes those calls into worker queues and preserves order only within each worker. +/// +/// Concurrent direct sends are independent requests, so nothing orders them against each other. The +/// chunk order described above holds within one call to [`send()`](Self::send) only. +/// +/// # Shutting down +/// +/// Call [`shutdown()`](Self::shutdown) when production is complete. It takes the producer by value, +/// drains a background producer's queues, flushes its remaining buffers, and waits for the worker +/// and error tasks. Dropping a background producer provides no completion guarantee and can lose +/// buffered messages. A direct producer has nothing buffered, so shutdown is a no-op. +/// +/// [`IggyClient`]: crate::clients::client::IggyClient +/// [`IggyClient::producer()`]: crate::clients::client::IggyClient::producer +/// [`IggyProducerBuilder`]: crate::clients::producer_builder::IggyProducerBuilder +/// [`BackgroundConfig`]: crate::clients::producer_config::BackgroundConfig +/// [`BackgroundConfig::batch_length`]: crate::clients::producer_config::BackgroundConfig::batch_length +/// [`BackgroundConfig::batch_size`]: crate::clients::producer_config::BackgroundConfig::batch_size +/// [`BackgroundConfig::error_callback`]: crate::clients::producer_config::BackgroundConfig::error_callback +/// [`BackgroundConfig::failure_mode`]: crate::clients::producer_config::BackgroundConfig::failure_mode +/// [`BackgroundConfig::linger_time`]: crate::clients::producer_config::BackgroundConfig::linger_time +/// [`BackgroundConfig::max_buffer_size`]: crate::clients::producer_config::BackgroundConfig::max_buffer_size +/// [`BackgroundConfig::max_in_flight`]: crate::clients::producer_config::BackgroundConfig::max_in_flight +/// [`BackgroundConfig::num_shards`]: crate::clients::producer_config::BackgroundConfig::num_shards +/// [`BackgroundConfig::sharding`]: crate::clients::producer_config::BackgroundConfig::sharding +/// [`BalancedSharding`]: crate::clients::producer_sharding::BalancedSharding +/// [`ErrorCallback`]: crate::clients::producer_error_callback::ErrorCallback +/// [`OrderedSharding`]: crate::clients::producer_sharding::OrderedSharding +/// [`background()`]: crate::clients::producer_builder::IggyProducerBuilder::background +/// [`build()`]: crate::clients::producer_builder::IggyProducerBuilder::build +/// [`create_stream_if_not_exists()`]: crate::clients::producer_builder::IggyProducerBuilder::create_stream_if_not_exists +/// [`create_topic_if_not_exists()`]: crate::clients::producer_builder::IggyProducerBuilder::create_topic_if_not_exists +/// [`direct()`]: crate::clients::producer_builder::IggyProducerBuilder::direct +/// [`do_not_create_stream_if_not_exists()`]: crate::clients::producer_builder::IggyProducerBuilder::do_not_create_stream_if_not_exists +/// [`do_not_create_topic_if_not_exists()`]: crate::clients::producer_builder::IggyProducerBuilder::do_not_create_topic_if_not_exists +/// [`encryptor()`]: crate::clients::producer_builder::IggyProducerBuilder::encryptor +/// [`partitioner()`]: crate::clients::producer_builder::IggyProducerBuilder::partitioner +/// [`partitioning()`]: crate::clients::producer_builder::IggyProducerBuilder::partitioning +/// [`send_retries()`]: crate::clients::producer_builder::IggyProducerBuilder::send_retries +/// [`stream()`]: crate::clients::producer_builder::IggyProducerBuilder::stream +/// [`topic()`]: crate::clients::producer_builder::IggyProducerBuilder::topic +/// [`without_encryptor()`]: crate::clients::producer_builder::IggyProducerBuilder::without_encryptor +/// [`without_partitioner()`]: crate::clients::producer_builder::IggyProducerBuilder::without_partitioner +/// [`without_partitioning()`]: crate::clients::producer_builder::IggyProducerBuilder::without_partitioning pub struct IggyProducer { core: Arc, dispatcher: Option, @@ -532,37 +847,100 @@ impl IggyProducer { Self { core, dispatcher } } + /// Returns the identifier of the stream this producer appends to. pub fn stream(&self) -> &Identifier { &self.core.stream_id } + /// Returns the identifier of the topic this producer appends to. pub fn topic(&self) -> &Identifier { &self.core.topic_id } - /// Initializes the producer by subscribing to diagnostic events, creating the stream and topic if they do not exist etc. + /// Initializes the producer and makes it ready to send messages. + /// + /// This must be called before the first send. Calling it again after successful initialization + /// does nothing and returns immediately. + /// + /// Initialization ensures that: + /// - The producer subscribes to client [`DiagnosticEvent`] values and tracks whether sending is + /// currently allowed. The gate starts open and follows the events observed after this call: + /// connect, disconnect, sign-out, and shutdown close it, and only a sign-in reopens it. It is + /// checked only when + /// [`send_retries()`](crate::clients::producer_builder::IggyProducerBuilder::send_retries) + /// allows at least one retry. + /// - the stream exists, creating it when `create_stream_if_not_exists` is set (the default). + /// - the topic exists, creating it when `create_topic_if_not_exists` is set (the default), with + /// the partitions count, message expiry and max size passed to + /// [`IggyProducerBuilder::create_topic_if_not_exists`](crate::clients::producer_builder::IggyProducerBuilder::create_topic_if_not_exists). + /// These are the only topic options controlled by producer initialization. All other settings + /// come from [`TopicCreateOptions::default`]. /// - /// Note: This method must be invoked before producing messages. + /// # Errors + /// + /// - [`IggyError::StreamNameNotFound`] or [`IggyError::TopicNameNotFound`] when the stream or + /// the topic does not exist and its auto creation is disabled. + /// - Any other error the server raised while looking up or creating the stream or the topic. pub async fn init(&self) -> Result<(), IggyError> { self.core.init().await } - /// Sends `messages` and returns the commit confirmations of every chunk the - /// send was split into, concatenated in chunk order. A retried chunk - /// contributes only the confirmation of the attempt that finally succeeded. + /// Sends `messages` to the stream and topic this producer was built for, with the partitioning it + /// was built with. + /// + /// What a returned `Ok` tells you depends on the send mode: + /// + /// | | `direct` producer | `background` producer | + /// | --- | --- | --- | + /// | the call returns | once the server has answered every request the batch was split into | once the batch is queued on a worker | + /// | `Ok` means | the server returned success for every request | the messages were accepted into a worker queue, nothing more | + /// | confirmations | normally one per request, in order, but legacy servers may return none | always empty | + /// | a write that fails | comes back as [`IggyError::ProducerSendFailed`] | goes to [`error_callback`] later | + /// + /// An empty `messages` vector is a no-op in both modes and returns an empty confirmation list. + /// + /// # Confirmations + /// + /// A [`SendMessagesConfirmationResponse`] names the partition a chunk of the batch landed in and + /// the `base_offset` its first message was given. + /// An offset is a position, not an identity. Delivery is at-least-once, so an earlier retry may + /// have committed the same messages at a lower offset, see + /// [Retrying and what a failure means](IggyProducer#retrying-and-what-a-failure-means). + /// A confirmation reports an in-memory commit, not an fsync. A crash and restart can therefore + /// lose an acknowledged batch and later reuse an offset the client already observed. + /// + /// # How long the call takes /// - /// Delivery is at-least-once. An earlier retry may already have committed - /// the same messages at a lower offset, so `base_offset` never implies - /// uniqueness. + /// Both modes can wait: /// - /// A batch is confirmed once it is committed in memory, not once it is - /// fsynced. A crash-restart can stamp a later batch with an offset a client - /// has already recorded. + /// - A `direct` send first waits out whatever is left of [`DirectConfig::linger_time`] since the + /// previous send, then awaits its requests one after another, so it returns no earlier than the + /// last one is written. + /// - A `background` send waits when the dispatcher has no room for the batch, which + /// [`failure_mode`] configures. The default [`BackpressureMode::Block`] waits for as long as it + /// takes. It can wait on the queue of the worker it was routed to as well, which holds 256 + /// queued sends. /// - /// The confirmation list is empty whenever the server sends no confirmation - /// payload, as the legacy server never sends one, and for a `background` - /// producer, which hands the messages to a dispatcher and returns before the - /// send happens. Branch on `confirmations.is_empty()` instead of indexing. + /// # Errors + /// + /// A `direct` producer wraps a send failure in [`IggyError::ProducerSendFailed`]. The cause can be + /// [`IggyError::CannotSendMessagesDueToClientDisconnection`] after the readiness retry budget is + /// exhausted, an encryption or partitioning failure raised before anything left the producer, + /// a server or transport error after write retries, or an unreadable confirmation for a request + /// that may already have committed. + /// + /// A `background` producer only reports what queueing the batch ran into: + /// [`IggyError::ProducerClosed`] after [`shutdown()`](Self::shutdown), + /// [`IggyError::BackgroundSendBufferOverflow`] or [`IggyError::BackgroundSendTimeout`] under + /// back pressure, and [`IggyError::BackgroundSendError`] when a worker is gone. A batch larger + /// than [`max_buffer_size`] always fails with [`IggyError::BackgroundSendBufferOverflow`], + /// however idle the producer is. Failures of the write itself reach [`error_callback`] instead, + /// which drops the messages unless it is implemented to keep them. + /// + /// [`BackpressureMode::Block`]: crate::clients::producer_config::BackpressureMode::Block + /// [`error_callback`]: crate::clients::producer_config::BackgroundConfig::error_callback + /// [`failure_mode`]: crate::clients::producer_config::BackgroundConfig::failure_mode + /// [`max_buffer_size`]: crate::clients::producer_config::BackgroundConfig::max_buffer_size pub async fn send( &self, messages: Vec, @@ -588,12 +966,20 @@ impl IggyProducer { } } - /// See [`IggyProducer::send`] for the confirmation semantics. + /// Sends one message. + /// + /// This has the same mode-dependent confirmation, backpressure, retry, and error semantics as + /// [`IggyProducer::send`]. pub async fn send_one(&self, message: IggyMessage) -> Result { self.send(vec![message]).await } - /// See [`IggyProducer::send`] for the confirmation semantics. + /// Sends `messages` to the partition `partitioning` selects, overriding the partitioning this + /// producer was built with for this call only. `None` falls back to that configured + /// partitioning. + /// + /// A [`partitioner()`](crate::clients::producer_builder::IggyProducerBuilder::partitioner) still + /// wins over the argument, since it computes the partition from the messages themselves. pub async fn send_with_partitioning( &self, messages: Vec, @@ -620,7 +1006,13 @@ impl IggyProducer { } } - /// See [`IggyProducer::send`] for the confirmation semantics. + /// Sends `messages` to any stream and topic, not only the pair this producer was built for. + /// + /// The target has to exist, since [`init()`](Self::init) only creates the producer's own stream + /// and topic. Everything else stays in force: encryption, partitioning, retries, and for a + /// `background` producer the routing to a worker [`Shard`]. + /// + /// [`Shard`]: crate::clients::producer_sharding::Shard pub async fn send_to( &self, stream: Arc, @@ -646,10 +1038,15 @@ impl IggyProducer { } } - /// Flushes buffered messages in `background` mode before returning. A - /// `direct`-mode producer has nothing to flush. Dropping the producer - /// instead of calling this silently discards unflushed `background` + /// Shuts the producer down. + /// + /// For a background producer, this drains the dispatcher queues, flushes the remaining shard + /// buffers, waits for writes and error callbacks to finish, and then returns. Stop every sender + /// first: a send racing the shutdown can be queued after the drain and is lost without an error. + /// Dropping a background producer instead provides no such guarantee and can lose buffered /// messages. + /// + /// A direct producer has nothing buffered, so calling `shutdown()` is a no-op. pub async fn shutdown(self) { if let Some(dispatcher) = self.dispatcher { dispatcher.shutdown().await; diff --git a/core/sdk/src/clients/producer_builder.rs b/core/sdk/src/clients/producer_builder.rs index 0d05adbcb5..97e575ab66 100644 --- a/core/sdk/src/clients/producer_builder.rs +++ b/core/sdk/src/clients/producer_builder.rs @@ -91,7 +91,7 @@ impl IggyProducerBuilder { Self { stream, ..self } } - /// Sets the stream name. + /// Sets the topic identifier. pub fn topic(self, topic: Identifier) -> Self { Self { topic, ..self } } diff --git a/core/sdk/src/clients/producer_config.rs b/core/sdk/src/clients/producer_config.rs index f2d5948527..6da8553c27 100644 --- a/core/sdk/src/clients/producer_config.rs +++ b/core/sdk/src/clients/producer_config.rs @@ -22,132 +22,244 @@ use bon::Builder; use iggy_common::{IggyByteSize, IggyDuration}; use std::sync::Arc; -/// Determines how the `send_messages` API should behave when problem is encountered +/// What a background send does when the dispatcher's byte budget, +/// [`BackgroundConfig::max_buffer_size`], is exhausted. +/// +/// Set it through [`BackgroundConfig::failure_mode`]. These modes govern waiting for buffer capacity, +/// not retries of a server write. A single batch larger than the whole budget fails with +/// [`IggyError::BackgroundSendBufferOverflow`] under all of them. +/// +/// [`IggyError::BackgroundSendBufferOverflow`]: iggy_common::IggyError::BackgroundSendBufferOverflow #[derive(Debug, Clone)] pub enum BackpressureMode { - /// Block until the send succeeds + /// Waits until enough byte-budget capacity is released (default). + /// + /// This wait has no timeout. It can last indefinitely if queued or in-flight writes do not + /// complete and release their permits. Block, - /// Block with a timeout, after which the send fails + /// Waits for the given duration, then fails the send with + /// [`IggyError::BackgroundSendTimeout`](iggy_common::IggyError::BackgroundSendTimeout). BlockWithTimeout(IggyDuration), - /// Fail immediately without retrying + /// Gives up at once with + /// [`IggyError::BackgroundSendBufferOverflow`](iggy_common::IggyError::BackgroundSendBufferOverflow), + /// leaving the batch unqueued. FailImmediately, } -// Configuration for the *background* (asynchronous) producer +/// Configuration for a producer that sends messages in the background. +/// +/// A background producer passes every non-empty send to a [`ProducerDispatcher`]. The dispatcher +/// returns once the batch is queued, and one of its worker [`Shard`]s writes the batch later. This type controls +/// how many workers exist, which worker receives a batch, when workers flush, how many bytes may be +/// queued or in flight, and where write failures are reported. +/// +/// # Defaults +/// +/// | Field | Default | Controls | +/// | --- | --- | --- | +/// | [`num_shards`](Self::num_shards) | 1 | how many worker queues exist | +/// | [`sharding`](Self::sharding) | [`OrderedSharding`] | which worker a batch is queued on | +/// | [`batch_size`](Self::batch_size) | 1 MiB | flush once a worker holds this many bytes | +/// | [`batch_length`](Self::batch_length) | 1000 | flush once a worker holds this many queued sends | +/// | [`linger_time`](Self::linger_time) | 1 ms | how long a worker holds a non-empty buffer before flushing it | +/// | [`max_buffer_size`](Self::max_buffer_size) | 32 MiB | bytes the whole producer may hold | +/// | [`failure_mode`](Self::failure_mode) | [`BackpressureMode::Block`] | what a send does when that budget is full | +/// | [`max_in_flight`](Self::max_in_flight) | 1 | requests being written at once | +/// | [`error_callback`](Self::error_callback) | [`LogErrorCallback`] | what happens to a failed write | +/// +/// The default [`OrderedSharding`] strategy preserves dispatch order for each stream/topic pair by +/// routing that pair to one worker. That worker awaits each request, including its retries, before +/// starting the next. [`BalancedSharding`] can route consecutive batches for one destination to +/// different workers and therefore gives up that ordering. [`max_in_flight`](Self::max_in_flight) +/// only controls how many workers may write concurrently. It does not make a single worker process +/// more than one request at a time. +/// +/// # Zero values +/// +/// `0` disables the [`batch_size`](Self::batch_size) and +/// [`batch_length`](Self::batch_length) flush thresholds. A zero +/// [`linger_time`](Self::linger_time) flushes as soon as the worker picks up a send. A zero +/// [`max_buffer_size`](Self::max_buffer_size) is treated as unbounded. A zero +/// [`max_in_flight`](Self::max_in_flight) uses `Semaphore::MAX_PERMITS`, and a zero +/// [`num_shards`](Self::num_shards) is read as one worker. +/// /// # Examples /// /// ``` +/// use iggy::clients::producer_config::BackpressureMode; /// use iggy::prelude::*; -/// use iggy_common::{IggyDuration, IggyByteSize}; -/// -/// // Use default config -/// let config = BackgroundConfig::builder() -/// .build(); +/// use std::time::Duration; /// -/// // Set custom batch size and disable length limit -/// let config = BackgroundConfig::builder() -/// .batch_size(256 * 1024) // 256 KiB -/// .batch_length(0) // unlimited -/// .build(); +/// // Ordered and bounded, as described above. +/// let ordered = BackgroundConfig::builder().build(); /// -/// // Configure low-latency flush -/// let config = BackgroundConfig::builder() -/// .linger_time(IggyDuration::from(200)) // 200ms +/// // Throughput: four workers share one topic, flushing at 4 MiB, with a 256 MiB byte budget. +/// // Up to 4 requests can be in flight at the same time, one per worker. +/// let fast = BackgroundConfig::builder() +/// .num_shards(4) +/// .sharding(Box::new(BalancedSharding::default())) +/// .batch_size(4 * 1024 * 1024) +/// .max_buffer_size(IggyByteSize::from(256 * 1024 * 1024)) +/// .max_in_flight(4) /// .build(); /// -/// // Disable all limits (not recommended for production) -/// let config = BackgroundConfig::builder() -/// .batch_size(0) -/// .batch_length(0) -/// .max_buffer_size(IggyByteSize::from(0)) -/// .max_in_flight(0) +/// // Latency: flush within 5 ms or after 50 queued sends. Do not wait for byte-budget capacity. +/// // The bounded per-worker channel can still make dispatch wait when its 256 slots are occupied. +/// let responsive = BackgroundConfig::builder() +/// .linger_time(IggyDuration::new(Duration::from_millis(5))) +/// .batch_length(50) +/// .failure_mode(BackpressureMode::FailImmediately) /// .build(); /// ``` +/// +/// [`BalancedSharding`]: crate::clients::producer_sharding::BalancedSharding +/// [`ProducerDispatcher`]: crate::clients::producer_dispatcher::ProducerDispatcher +/// [`Shard`]: crate::clients::producer_sharding::Shard #[derive(Debug, Builder)] pub struct BackgroundConfig { - /// Number of shard-workers that run in parallel. + /// Number of worker [`Shard`]s the dispatcher runs, each with a queue of its own. /// - /// With the default `OrderedSharding` strategy, messages to the same - /// stream/topic are always routed to the same shard, preserving ordering. - /// Increasing shards improves throughput only when sending to multiple streams/topics. + /// Every batch is routed to exactly one worker by [`sharding`](Self::sharding), and each worker + /// buffers and writes independently. /// - /// With `BalancedSharding`, messages are distributed round-robin across all shards - /// for maximum single-destination throughput, but ordering is **not** preserved. + /// `0` is read as one worker. + /// + /// [`Shard`]: crate::clients::producer_sharding::Shard #[builder(default = 1)] pub num_shards: usize, - /// How long a shard may wait before flushing an *incomplete* batch. + /// Upper bound on how long a worker holds a non-empty buffer before flushing it. + /// + /// The window starts when a send enters an empty buffer, so an idle worker does not wake up. + /// A worker flushes as soon as any of `linger_time`, [`batch_length`](Self::batch_length) or + /// [`batch_size`](Self::batch_size) is reached. Lowering it reduces the time the write is delayed + /// at the price of smaller writes. `0` flushes as soon as the worker picks up a send. /// - /// Combines with `batch_size` / `batch_length`: whichever limit fires - /// first triggers the flush. + /// Note that [`IggyDuration::from`] reads a plain number as **microseconds**, so the default of + /// `1000` is 1 ms. #[builder(default = IggyDuration::from(1000))] pub linger_time: IggyDuration, - /// User-supplied asynchronous callback that will be executed whenever - /// the producer encounters an error it cannot automatically recover from - /// (e.g. network failure). + /// Where a background write that failed ends up. + /// + /// The dispatcher runs one task that owns this callback. A worker invokes it with an [`ErrorCtx`] + /// when its backend returns [`IggyError::ProducerSendFailed`]. Other error variants from a custom + /// backend are logged by the worker without invoking this callback. The context contains the + /// cause, destination, unconfirmed tail, and confirmations returned for earlier chunks. + /// + /// The default [`LogErrorCallback`] logs the failure and drops the messages with the context. + /// Implement [`ErrorCallback`] with your own logic to keep them. + /// + /// [`ErrorCtx`]: crate::clients::producer_error_callback::ErrorCtx + /// [`IggyError::ProducerSendFailed`]: iggy_common::IggyError::ProducerSendFailed #[builder(default = Arc::new(Box::new(LogErrorCallback)))] pub error_callback: Arc>, - /// Strategy that maps a message to a shard. + /// Picks the worker a batch is queued on, out of [`num_shards`](Self::num_shards). /// - /// Default is `OrderedSharding` which routes all messages for the same - /// stream/topic to the same shard, preserving message ordering. + /// The default [`OrderedSharding`] hashes stream and topic, so every batch for one topic queues + /// on one worker and keeps the order it was dispatched in. [`BalancedSharding`] hands them out + /// round-robin, which lets a single topic occupy all workers but gives up that order. Implement + /// [`Sharding`] to build your own logic for picking a `Shard`. /// - /// Use `BalancedSharding` for maximum throughput when ordering doesn't matter. + /// [`BalancedSharding`]: crate::clients::producer_sharding::BalancedSharding #[builder(default = Box::new(OrderedSharding))] pub sharding: Box, - /// Maximum **total size in bytes** of a batch. - /// `0` ⇒ unlimited (size-based batching disabled). + /// Flush threshold in bytes buffered on one worker. + /// + /// Sends accumulate until their reported sizes reach or exceed this value. The threshold is + /// checked after a send is added, so it is a flush trigger rather than a hard size ceiling. + /// `0` disables the threshold + /// and leaves [`batch_length`](Self::batch_length) and [`linger_time`](Self::linger_time) to + /// trigger the flush. + /// + /// This is a per-worker flush trigger. It is separate from + /// [`max_buffer_size`](Self::max_buffer_size), which caps the producer as a whole. #[builder(default = MIB)] pub batch_size: usize, - /// Maximum **number of messages** per batch. - /// `0` ⇒ unlimited (length-based batching disabled). + /// Flush threshold in number of queued batches on one worker. + /// + /// Counts the queued sends a worker holds, not the individual messages inside them. A worker + /// flushes after this many dispatches have been routed to it. `0` disables the threshold. + /// #[builder(default = 1000)] pub batch_length: usize, - /// Action to apply when back-pressure limits are reached + /// What a send does once [`max_buffer_size`](Self::max_buffer_size) is exhausted. #[builder(default = BackpressureMode::Block)] pub failure_mode: BackpressureMode, /// Upper bound for the **bytes buffered or in flight** across *all* shards. /// Bytes remain charged until the corresponding write completes. - /// `IggyByteSize::from(0)` ⇒ unlimited. + /// `IggyByteSize::from(0)` means unlimited. A nonzero value greater than + /// `Semaphore::MAX_PERMITS` makes [`ProducerDispatcher::new`] panic. + /// + /// [`ProducerDispatcher::new`]: crate::clients::producer_dispatcher::ProducerDispatcher::new #[builder(default = IggyByteSize::from(32 * MIB as u64))] pub max_buffer_size: IggyByteSize, - /// Maximum number of **in-flight requests** (batches being sent). + /// Upper bound on the requests being written concurrently, shared by *all* workers. /// - /// **WARNING**: Using more than 1 may cause message reordering if retries occur. - /// With max_in_flight > 1, a failed batch could be retried after later batches succeed. + /// A worker takes one permit before each request and holds it until the write returns. This + /// bounds write concurrency across the producer rather than per worker, and it does not bound + /// queued bytes. /// - /// The default is `1` to preserve message ordering. - /// `0` ⇒ unlimited (no ordering guarantee). + /// Each worker still sends sequentially regardless of this value. Raising it only lets different + /// workers write concurrently. Per-destination order is therefore governed by + /// [`sharding`](Self::sharding): [`OrderedSharding`] keeps one destination on one sequential + /// worker, while strategies that spread a destination across workers may reorder it. + /// A nonzero value greater than `Semaphore::MAX_PERMITS` makes + /// [`ProducerDispatcher::new`] panic. + /// + /// [`ProducerDispatcher::new`]: crate::clients::producer_dispatcher::ProducerDispatcher::new #[builder(default = 1)] pub max_in_flight: usize, } -/// Configuration for the *synchronous* (blocking) producer. +/// Configuration for a direct producer. +/// +/// A direct producer writes from the calling task. [`send()`] splits a batch into requests of at +/// most [`batch_length`](Self::batch_length) messages, awaits them one after another and returns +/// their confirmations. Nothing is buffered between calls. Unlike background mode, there is no +/// queue to bound, no worker to route to, and nothing to flush on shutdown. +/// +/// A send that fails part way through returns [`IggyError::ProducerSendFailed`], where `committed` +/// holds the confirmations returned for earlier requests and `failed` holds the unconfirmed tail. +/// See [`send()`] for what resending that tail means. +/// /// # Examples /// /// ```rust /// use iggy::prelude::*; -/// use iggy_common::IggyDuration; +/// use std::time::Duration; /// -/// // Send messages one-by-one (max latency, min memory per request) -/// let cfg = DirectConfig::builder() +/// // One request per message, with no configured delay between sequential calls. +/// let low_latency = DirectConfig::builder() /// .batch_length(1) /// .linger_time(IggyDuration::from(0)) /// .build(); /// -/// // Send in chunks of up to 500 messages, -/// // with a delay of at least 200 ms between consecutive sends. -/// let cfg = DirectConfig::builder() +/// // Up to 500 messages per request, with a 200 ms minimum gap between sequential sends. +/// let paced = DirectConfig::builder() /// .batch_length(500) -/// .linger_time(IggyDuration::from(200)) +/// .linger_time(IggyDuration::new(Duration::from_millis(200))) /// .build(); /// ``` +/// +/// [`send()`]: crate::clients::producer::IggyProducer::send +/// [`IggyError::ProducerSendFailed`]: iggy_common::IggyError::ProducerSendFailed #[derive(Clone, Builder)] pub struct DirectConfig { - /// Maximum number of messages to pack into **one** synchronous request. - /// `0` ⇒ MAX_BATCH_LENGTH(). + /// Maximum number of messages in one request. + /// + /// A send carrying more than this is split into consecutive requests of this size, each awaited + /// before the next one starts. A batch of 2500 therefore becomes three requests at the default. + /// + /// `0` limits to 1,000,000 messages per request. #[builder(default = 1000)] pub batch_length: u32, - /// How long to wait for more messages before flushing the current set. + /// Requested minimum gap between sequential direct sends. + /// + /// A send waits out whatever is left of this interval since the previous request completed + /// successfully. + /// Concurrent callers can wait against the same timestamp and then proceed together, so this is + /// not a global rate limiter. + /// When one call is split by [`batch_length`](Self::batch_length), the linger interval is applied + /// before the call rather than between its chunks. The default of zero does not wait. #[builder(default = IggyDuration::from(0))] pub linger_time: IggyDuration, } diff --git a/core/sdk/src/clients/producer_dispatcher.rs b/core/sdk/src/clients/producer_dispatcher.rs index 447a1b86cd..09d2e29562 100644 --- a/core/sdk/src/clients/producer_dispatcher.rs +++ b/core/sdk/src/clients/producer_dispatcher.rs @@ -17,15 +17,139 @@ use crate::clients::producer::ProducerCoreBackend; use crate::clients::producer_config::{BackgroundConfig, BackpressureMode}; -use crate::clients::producer_error_callback::ErrorCtx; +use crate::clients::producer_error_callback::{ErrorCallback, ErrorCtx}; use crate::clients::producer_sharding::{Shard, ShardMessage, ShardMessageWithPermit}; use futures::FutureExt; use iggy_common::{Identifier, IggyByteSize, IggyError, IggyMessage, Partitioning, Sizeable}; +use std::any::Any; +use std::panic::AssertUnwindSafe; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use tokio::sync::{Semaphore, broadcast}; use tokio::task::JoinHandle; +/// The background machinery of an [`IggyProducer`](crate::clients::producer::IggyProducer), built from a +/// [`BackgroundConfig`] when the producer is configured with +/// [`IggyProducerBuilder::background`](crate::clients::producer_builder::IggyProducerBuilder::background). +/// +/// The dispatcher owns the background workers responsible for writing messages (the [`Shard`]s) +/// and coordinates two limits, both configured on [`BackgroundConfig`]: +/// [`max_buffer_size`](BackgroundConfig::max_buffer_size), the upper bound on the message bytes it +/// holds at once, and [`max_in_flight`](BackgroundConfig::max_in_flight), the upper bound on the +/// requests being written at once across all of its workers. +/// +/// A background send dispatches messages to a [`Shard`] through a channel. The batch is queued and +/// the write happens later on one of the shard workers owned by this type. Write failures therefore +/// surface on a shard after the queueing caller has returned. The shard forwards each failure to the +/// dispatcher's error task, which invokes [`BackgroundConfig::error_callback`]. The default +/// [`LogErrorCallback`] logs the failure, and applications can provide another [`ErrorCallback`] +/// implementation. +/// +/// Which worker a batch lands on, and what that means for ordering, is described in +/// [`IggyProducer`](crate::clients::producer::IggyProducer). +/// +/// # Examples +/// +/// Configuring background sending through the producer builder: +/// +/// ```rust,no_run +/// use iggy::prelude::*; +/// use std::str::FromStr; +/// +/// # async fn example() -> Result<(), IggyError> { +/// let client = IggyClient::from_connection_string("iggy://iggy:iggy@localhost:8090")?; +/// client.connect().await?; +/// +/// let producer = client +/// .producer("my-stream", "my-topic")? +/// .background(BackgroundConfig::builder().num_shards(4).build()) +/// .build(); +/// producer.init().await?; +/// +/// producer.send_one(IggyMessage::from_str("hello")?).await?; +/// producer.shutdown().await; +/// # Ok(()) +/// # } +/// ``` +/// +/// You can also implement a backend and wrap the dispatcher around it. This example prints instead +/// of sending anything to a server. +/// +/// ```rust,no_run +/// use iggy::clients::producer::ProducerCoreBackend; +/// use iggy::clients::producer_dispatcher::ProducerDispatcher; +/// use iggy::prelude::*; +/// use std::str::FromStr; +/// use std::sync::Arc; +/// +/// #[derive(Debug)] +/// struct CountingBackend; +/// +/// impl ProducerCoreBackend for CountingBackend { +/// async fn send_internal( +/// &self, +/// stream: &Identifier, +/// topic: &Identifier, +/// messages: Vec, +/// _partitioning: Option>, +/// ) -> Result { +/// println!("{} messages to {stream}/{topic}", messages.len()); +/// Ok(SendMessagesResponse { confirmations: Vec::new() }) +/// } +/// } +/// +/// # async fn example() -> Result<(), IggyError> { +/// let dispatcher = ProducerDispatcher::new( +/// Arc::new(CountingBackend), +/// BackgroundConfig::builder().num_shards(2).build(), +/// ); +/// +/// let stream = Arc::new(Identifier::named("my-stream")?); +/// let topic = Arc::new(Identifier::named("my-topic")?); +/// +/// // Returns once the batch is queued, not once it is written. +/// dispatcher +/// .dispatch(vec![IggyMessage::from_str("hello")?], stream, topic, None) +/// .await?; +/// +/// // Writes what is still queued. Dropping the dispatcher instead may lose buffered messages. +/// dispatcher.shutdown().await; +/// # Ok(()) +/// # } +/// ``` +/// +/// # Write constraints +/// +/// ## Memory budget +/// +/// [`dispatch()`](Self::dispatch) charges a batch against a budget of +/// [`BackgroundConfig::max_buffer_size`] bytes before queueing it. The charge is released when +/// [`ProducerCoreBackend::send_internal`] returns, so the budget covers queued sends and requests +/// whose result is still pending. What gets charged is the size [`ShardMessage`] reports, which +/// counts stream and topic identifiers alongside the messages rather than payloads alone. +/// +/// [`BackgroundConfig::failure_mode`] decides how the byte budget backpressures the caller. The +/// bounded channel feeding each shard is a separate source of backpressure and can also make a +/// dispatch wait. The second configured limit, [`BackgroundConfig::max_in_flight`], is shared by the +/// workers rather than the callers. A worker takes one of its permits for the batch it is about to +/// write, so it bounds concurrent writes across all workers, not queued bytes. +/// +/// ## In-flight limit +/// +/// [`BackgroundConfig::max_in_flight`] limits how many workers may write concurrently. The permit +/// pool is shared by every worker and defaults to one. A shard itself remains sequential for every +/// value: it awaits a request and all of its retries before starting its next request. Raising the +/// limit therefore affects concurrency between shards. It can expose reordering only when the +/// configured sharding strategy sends one ordered destination to different shards. +/// +/// # Shutdown +/// +/// [`shutdown()`](Self::shutdown) broadcasts a stop signal, drains every shard channel, flushes each +/// remaining buffer, and waits for the shard and error tasks. Dropping the dispatcher provides no +/// such completion guarantee and can lose batches that a shard has buffered but not written. +/// +/// [`ErrorCallback`]: crate::clients::producer_error_callback::ErrorCallback +/// [`LogErrorCallback`]: crate::clients::producer_error_callback::LogErrorCallback pub struct ProducerDispatcher { shards: Vec, config: Arc, @@ -36,6 +160,108 @@ pub struct ProducerDispatcher { } impl ProducerDispatcher { + /// Spawns the [`Shard`] workers that write messages to the server. + /// + /// The dispatcher owns the shards over which writes are distributed. + /// [`BackgroundConfig::sharding`] decides which of the + /// [`BackgroundConfig::num_shards`] workers receives each batch. + /// + /// The dispatcher also starts an error task. When a shard receives + /// [`IggyError::ProducerSendFailed`] from its backend, it sends the context to this task, which + /// invokes the [`ErrorCallback`] configured through [`BackgroundConfig::error_callback`]. A + /// custom backend error that is not `ProducerSendFailed` is logged by the shard and does not + /// invoke the callback. + /// + /// Two semaphores enforce [`BackgroundConfig::max_buffer_size`] and + /// [`BackgroundConfig::max_in_flight`]. Both are shared by every shard and therefore apply to + /// the entire producer rather than per worker. They + /// count different things at different points of a batch's life: `max_buffer_size` is charged in + /// bytes by [`dispatch()`](Self::dispatch) before the batch is queued and released once it has + /// completed, so it bounds queued and in-flight bytes together, while `max_in_flight` is taken + /// as one permit by a worker that is about to write and released when that request returns, so + /// it bounds concurrent requests and says nothing about their size. A batch therefore has to pass + /// the byte budget to enter a queue, and to take a request slot to leave it. + /// + /// # Panics + /// + /// Panics when a nonzero [`BackgroundConfig::max_buffer_size`] or + /// [`BackgroundConfig::max_in_flight`] exceeds `Semaphore::MAX_PERMITS`. + /// + /// # Examples + /// + /// Four workers allowed to write in parallel, fed round-robin, with a 64 MiB budget for queued + /// and in-flight bytes: + /// + /// ```rust,no_run + /// # use iggy::clients::producer::ProducerCoreBackend; + /// use iggy::clients::producer_dispatcher::ProducerDispatcher; + /// use iggy::prelude::*; + /// use std::sync::Arc; + /// + /// # #[derive(Debug)] + /// # struct Backend; + /// # impl ProducerCoreBackend for Backend { + /// # async fn send_internal( + /// # &self, + /// # _stream: &Identifier, + /// # _topic: &Identifier, + /// # _messages: Vec, + /// # _partitioning: Option>, + /// # ) -> Result { + /// # Ok(SendMessagesResponse { confirmations: Vec::new() }) + /// # } + /// # } + /// # fn example(backend: Arc) { + /// let dispatcher = ProducerDispatcher::new( + /// backend, + /// BackgroundConfig::builder() + /// .num_shards(4) + /// .max_in_flight(4) + /// // Ordering is given up for throughput: a batch can land on any of the four workers. + /// .sharding(Box::new(BalancedSharding::default())) + /// .max_buffer_size(IggyByteSize::from(64 * 1024 * 1024)) + /// .build(), + /// ); + /// # } + /// ``` + /// + /// `num_shards(0)` is read as one worker. A zero byte budget is treated as unbounded and a zero + /// in-flight limit uses the semaphore maximum. This dispatcher therefore never refuses a batch + /// for lack of byte-budget capacity, although its bounded shard channel can still make dispatch + /// wait: + /// + /// ```rust,no_run + /// # use iggy::clients::producer::ProducerCoreBackend; + /// use iggy::clients::producer_dispatcher::ProducerDispatcher; + /// use iggy::prelude::*; + /// use std::sync::Arc; + /// + /// # #[derive(Debug)] + /// # struct Backend; + /// # impl ProducerCoreBackend for Backend { + /// # async fn send_internal( + /// # &self, + /// # _stream: &Identifier, + /// # _topic: &Identifier, + /// # _messages: Vec, + /// # _partitioning: Option>, + /// # ) -> Result { + /// # Ok(SendMessagesResponse { confirmations: Vec::new() }) + /// # } + /// # } + /// # fn example(backend: Arc) { + /// let dispatcher = ProducerDispatcher::new( + /// backend, + /// BackgroundConfig::builder() + /// .num_shards(0) + /// .max_buffer_size(IggyByteSize::from(0)) + /// .max_in_flight(0) + /// .build(), + /// ); + /// # } + /// ``` + /// + /// [`ErrorCallback`]: crate::clients::producer_error_callback::ErrorCallback pub fn new(core: Arc, config: BackgroundConfig) -> Self { let num_shards = if config.num_shards == 0 { 1 @@ -51,10 +277,7 @@ impl ProducerDispatcher { let handle = tokio::spawn(async move { while let Ok(ctx) = err_rx.recv_async().await { - if let Err(panic) = std::panic::AssertUnwindSafe(err_callback.call(ctx)) - .catch_unwind() - .await - { + if let Err(panic) = call_error_callback(&**err_callback, ctx).await { tracing::error!("error_callback panicked: {:?}", panic); } } @@ -96,6 +319,133 @@ impl ProducerDispatcher { } } + /// Queues a batch on one of the worker [`Shard`]s and returns without waiting for it to be written. + /// + /// The batch is charged against the [`BackgroundConfig::max_buffer_size`] semaphore before it is + /// queued. Its permit travels with it, so those bytes stay charged until + /// [`ProducerCoreBackend::send_internal`] returns. A batch larger than the entire budget can + /// never be charged and fails with + /// [`IggyError::BackgroundSendBufferOverflow`]. + /// + /// When the budget is exhausted, [`BackgroundConfig::failure_mode`] decides what happens to the + /// caller. [`BackpressureMode::FailImmediately`] gives up with + /// [`IggyError::BackgroundSendBufferOverflow`], [`BackpressureMode::Block`] waits until enough + /// capacity is released, and [`BackpressureMode::BlockWithTimeout`] waits for its duration before + /// failing with [`IggyError::BackgroundSendTimeout`]. These modes do not control retries of the + /// server write. + /// + /// [`BackgroundConfig::sharding`] then picks the worker, and the batch is handed to that worker's + /// queue. The queue holds 256 entries, so a full queue can make the caller wait independently of + /// the byte-budget failure mode. + /// + /// # Errors + /// + /// [`IggyError::ProducerClosed`] once [`shutdown()`](Self::shutdown) has begun, and + /// [`IggyError::BackgroundSendError`] if the picked worker is already gone, which leaves the + /// batch unqueued and unsent in both cases. The budget can additionally fail the call with + /// [`IggyError::BackgroundSendBufferOverflow`] or [`IggyError::BackgroundSendTimeout`] as + /// described above. `BackgroundSendBufferOverflow` is also returned when the batch's reported + /// size does not fit the semaphore API's `u32` permit count, including when the configured byte + /// budget is unbounded. + /// + /// # Panics + /// + /// Panics if the configured [`Sharding`](crate::clients::producer_sharding::Sharding) + /// implementation returns an index outside the dispatcher's shard list. + /// + /// # Examples + /// + /// Dispatching with a strategy for partitioning. + /// + /// ```rust,no_run + /// # use iggy::clients::producer::ProducerCoreBackend; + /// use iggy::clients::producer_dispatcher::ProducerDispatcher; + /// use iggy::prelude::*; + /// use std::str::FromStr; + /// use std::sync::Arc; + /// + /// # #[derive(Debug)] + /// # struct Backend; + /// # impl ProducerCoreBackend for Backend { + /// # async fn send_internal( + /// # &self, + /// # _stream: &Identifier, + /// # _topic: &Identifier, + /// # _messages: Vec, + /// # _partitioning: Option>, + /// # ) -> Result { + /// # Ok(SendMessagesResponse { confirmations: Vec::new() }) + /// # } + /// # } + /// # async fn example(dispatcher: ProducerDispatcher) -> Result<(), IggyError> { + /// let stream = Arc::new(Identifier::named("orders")?); + /// let topic = Arc::new(Identifier::named("created")?); + /// let partitioning = Arc::new(Partitioning::messages_key_str("order-42")?); + /// + /// dispatcher + /// .dispatch( + /// vec![IggyMessage::from_str("order created")?], + /// stream.clone(), + /// topic.clone(), + /// Some(partitioning), + /// ) + /// .await?; + /// + /// // Dispatch only guarantees that the first batch was queued. A worker may already have + /// // started or completed its write. + /// dispatcher + /// .dispatch(vec![IggyMessage::from_str("order updated")?], stream, topic, None) + /// .await?; + /// # Ok(()) + /// # } + /// ``` + /// + /// Fail immediately if the `max_buffer_size` is exceeded. + /// + /// ```rust,no_run + /// # use iggy::clients::producer::ProducerCoreBackend; + /// use iggy::clients::producer_config::BackpressureMode; + /// use iggy::clients::producer_dispatcher::ProducerDispatcher; + /// use iggy::prelude::*; + /// use std::str::FromStr; + /// use std::sync::Arc; + /// + /// # #[derive(Debug)] + /// # struct Backend; + /// # impl ProducerCoreBackend for Backend { + /// # async fn send_internal( + /// # &self, + /// # _stream: &Identifier, + /// # _topic: &Identifier, + /// # _messages: Vec, + /// # _partitioning: Option>, + /// # ) -> Result { + /// # Ok(SendMessagesResponse { confirmations: Vec::new() }) + /// # } + /// # } + /// # async fn example(backend: Arc) -> Result<(), IggyError> { + /// let dispatcher = ProducerDispatcher::new( + /// backend, + /// BackgroundConfig::builder() + /// .max_buffer_size(IggyByteSize::from(1024 * 1024)) + /// .failure_mode(BackpressureMode::FailImmediately) + /// .build(), + /// ); + /// + /// let messages = vec![IggyMessage::from_str("hello")?]; + /// let stream = Arc::new(Identifier::named("orders")?); + /// let topic = Arc::new(Identifier::named("created")?); + /// + /// match dispatcher.dispatch(messages, stream, topic, None).await { + /// Ok(()) => println!("queued"), + /// // The workers are behind, or this single batch is larger than the whole budget. + /// Err(IggyError::BackgroundSendBufferOverflow) => println!("dropped, budget is full"), + /// Err(IggyError::ProducerClosed) => println!("dropped, dispatcher is shutting down"), + /// Err(error) => return Err(error), + /// } + /// # Ok(()) + /// # } + /// ``` pub async fn dispatch( &self, messages: Vec, @@ -162,7 +512,9 @@ impl ProducerDispatcher { &shard_message.stream, &shard_message.topic, ); + debug_assert!(shard_ix < self.shards.len()); + let shard = &self.shards[shard_ix]; shard @@ -199,6 +551,16 @@ impl ProducerDispatcher { } } +/// Catches a panic in `call()` itself as well as in the future it returns, so one misbehaving +/// callback cannot end the error task and silently drop every later failure. +async fn call_error_callback( + callback: &(dyn ErrorCallback + Send + Sync), + ctx: ErrorCtx, +) -> Result<(), Box> { + let future = std::panic::catch_unwind(AssertUnwindSafe(|| callback.call(ctx)))?; + AssertUnwindSafe(future).catch_unwind().await +} + #[cfg(test)] mod tests { use std::pin::Pin; @@ -209,7 +571,6 @@ mod tests { use tokio::time::sleep; use crate::clients::producer::{MockProducerCoreBackend, no_confirmations}; - use crate::clients::producer_error_callback::ErrorCallback; use crate::clients::producer_sharding::Sharding; use super::*; @@ -536,4 +897,65 @@ mod tests { assert_eq!(error_called.load(Ordering::SeqCst), 1); assert_eq!(last_batch_len.load(Ordering::SeqCst), 1); } + + /// Panics inside `call()` itself, before any future exists, on the first invocation only. + #[derive(Debug)] + struct PanicOnceErrorCallback { + called: Arc, + } + + impl ErrorCallback for PanicOnceErrorCallback { + fn call(&self, _ctx: ErrorCtx) -> Pin + Send + 'static>> { + if self.called.fetch_add(1, Ordering::SeqCst) == 0 { + panic!("first failure panics before returning a future"); + } + Box::pin(async {}) + } + } + + #[tokio::test] + async fn test_error_task_survives_panic_in_error_callback_call() { + let mut mock = MockProducerCoreBackend::new(); + mock.expect_send_internal().returning(|_, _, _, _| { + Box::pin(async { + Err(IggyError::ProducerSendFailed { + cause: Box::new(IggyError::Error), + failed: Arc::new(vec![dummy_message(10)]), + committed: Arc::new(Vec::new()), + stream_name: "1".to_string(), + topic_name: "1".to_string(), + }) + }) + }); + + let called = Arc::new(AtomicUsize::new(0)); + let config = BackgroundConfig::builder() + .num_shards(1) + .error_callback(Arc::new(Box::new(PanicOnceErrorCallback { + called: called.clone(), + }))) + .build(); + let dispatcher = ProducerDispatcher::new(Arc::new(mock), config); + + // Distinct topics keep the two sends from merging into one request, whichever branch of + // the worker ends up flushing them. + for topic_id in 1..=2 { + dispatcher + .dispatch( + vec![dummy_message(10)], + dummy_identifier(), + Arc::new(Identifier::numeric(topic_id).unwrap()), + None, + ) + .await + .unwrap(); + } + dispatcher.shutdown().await; + + assert_eq!( + called.load(Ordering::SeqCst), + 2, + "the failure after the panicking one must still reach the callback" + ); + } } diff --git a/core/sdk/src/clients/producer_error_callback.rs b/core/sdk/src/clients/producer_error_callback.rs index 2d2bc8514a..0c3a4758b4 100644 --- a/core/sdk/src/clients/producer_error_callback.rs +++ b/core/sdk/src/clients/producer_error_callback.rs @@ -23,32 +23,155 @@ use std::pin::Pin; use std::sync::Arc; use tracing::error; +/// Everything known about a background write that did not return a usable confirmation. +/// +/// A [`background()`] producer returns from [`send()`] once the batch is queued. The write happens +/// later on one of the dispatcher's worker [`Shard`]s, when there is no caller waiting for its +/// result. The worker therefore sends an `ErrorCtx` to the dedicated error task, which invokes the +/// [`ErrorCallback`] configured in [`BackgroundConfig::error_callback`]. +/// +/// # What to do with it +/// +/// [`messages`](Self::messages) is the unconfirmed tail of the send. No further automatic retry will +/// be attempted. Depending on [`cause`](Self::cause), the retry budget may be exhausted, the failure +/// may be deliberately non-retriable, or encryption or partitioning may have failed before a request +/// was sent. The tail is not proof that nothing committed. A request can commit before its response +/// is lost, and an HTTP confirmation decoding failure specifically occurs after a successful status. +/// Resending `messages` is therefore an at-least-once operation and can create duplicates. +/// Encryption mutates messages before the write, so this tail contains encrypted messages when the +/// producer uses an encryptor. Passing them back through the same producer would encrypt them again. +/// +/// [`stream`](Self::stream) and [`topic`](Self::topic) identify the destination. +/// [`partitioning`](Self::partitioning) contains only the per-send override passed to the dispatcher. +/// `None` means the producer's configured or default partitioning was used, not that the request had +/// no partitioning. A callback can retain the context, persist it in a dead-letter store, alert an +/// operator, or retry it when duplicate delivery is acceptable. The default [`LogErrorCallback`] +/// only logs the failure and then drops the context. +/// +/// [`background()`]: crate::clients::producer_builder::IggyProducerBuilder::background +/// [`send()`]: crate::clients::producer::IggyProducer::send +/// [`BackgroundConfig::error_callback`]: crate::clients::producer_config::BackgroundConfig::error_callback +/// [`Shard`]: crate::clients::producer_sharding::Shard #[derive(Debug)] pub struct ErrorCtx { + /// Error that ended the send. No further automatic retry will be attempted. pub cause: Box, + /// Stream identifier used by the failed request. pub stream: Arc, + /// Stream name configured when the producer was built. + /// + /// For a failure from + /// [`IggyProducer::send_to`](crate::clients::producer::IggyProducer::send_to), this may not name + /// [`Self::stream`]. pub stream_name: String, + /// Topic identifier used by the failed request. pub topic: Arc, + /// Topic name configured when the producer was built. + /// + /// For a failure from + /// [`IggyProducer::send_to`](crate::clients::producer::IggyProducer::send_to), this may not name + /// [`Self::topic`]. pub topic_name: String, + /// Per-send partitioning override, or `None` when the producer configuration was used. pub partitioning: Option>, + /// Unconfirmed tail of the send, see [`ErrorCtx`] for what resending it means. pub messages: Arc>, - /// Confirmations of the chunks that committed before the failure; `messages` - /// is the tail that did not. + /// Confirmations returned for chunks before the failure. pub committed: Arc>, } -/// A trait for handling background sending errors. +/// Handles a background write failure after the queueing caller has returned. +/// +/// A [`background()`](crate::clients::producer_builder::IggyProducerBuilder::background) producer +/// acknowledges a send once it is queued, so a later write failure cannot be returned by +/// [`IggyProducer::send`]. The dispatcher owns one implementation of this trait, set with +/// [`BackgroundConfig::error_callback`], and invokes it with an [`ErrorCtx`] whenever a worker's +/// backend returns [`IggyError::ProducerSendFailed`]. Other error variants from a custom backend are +/// logged by the worker without invoking this callback. +/// +/// # Implementing it +/// +/// - [`call()`](Self::call) returns a boxed future that the error task awaits, so the callback may do +/// asynchronous I/O. +/// - It runs on its own task rather than on a shard worker, so awaiting it does not stall batching. +/// Calls are serialized. One failure is handled at a time, and the unbounded error channel can +/// grow while a callback is slow. +/// - A panic inside it, whether in [`call()`](Self::call) itself or in the returned future, is +/// caught and logged, and the next failure is still delivered. +/// - `Send + Sync + Debug + 'static` is required because the dispatcher's task owns the callback for +/// the producer's lifetime and [`BackgroundConfig`] implements [`Debug`]. +/// +/// # Example /// -/// This is used when a message batch fails to send in an asynchronous background task. -/// Implementors can define custom logic such as logging, retrying, alerting, etc. +/// Forward each failed batch to a separate task instead of dropping it. The callback only enqueues +/// the context, so a slow store does not hold up later callbacks: +/// +/// ```no_run +/// use iggy::clients::producer_error_callback::{ErrorCallback, ErrorCtx}; +/// use iggy::prelude::*; +/// use std::pin::Pin; +/// use std::sync::Arc; +/// use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender}; +/// use tracing::warn; +/// +/// #[derive(Debug)] +/// struct FailedMessages { +/// failures: UnboundedSender, +/// } +/// +/// impl ErrorCallback for FailedMessages { +/// fn call(&self, ctx: ErrorCtx) -> Pin + Send + 'static>> { +/// let failures = self.failures.clone(); +/// Box::pin(async move { +/// let num_messages = ctx.messages.len(); +/// if failures.send(ctx).is_err() { +/// warn!(num_messages, "Failed messages task is gone, dropping messages"); +/// } +/// }) +/// } +/// } +/// +/// // Replace this warning with durable storage or another application-specific policy. +/// async fn drain(mut failures: UnboundedReceiver) { +/// while let Some(ctx) = failures.recv().await { +/// warn!( +/// cause = %ctx.cause, +/// stream_name = ctx.stream_name, +/// topic_name = ctx.topic_name, +/// num_messages = ctx.messages.len(), +/// "Received failed batch", +/// ); +/// } +/// } +/// +/// # async fn example() { +/// let (failures, receiver) = tokio::sync::mpsc::unbounded_channel(); +/// tokio::spawn(drain(receiver)); +/// +/// let config = BackgroundConfig::builder() +/// .error_callback(Arc::new(Box::new(FailedMessages { failures }))) +/// .build(); +/// # } +/// ``` +/// +/// [`BackgroundConfig`]: crate::clients::producer_config::BackgroundConfig +/// [`BackgroundConfig::error_callback`]: crate::clients::producer_config::BackgroundConfig::error_callback +/// [`IggyProducer::send`]: crate::clients::producer::IggyProducer::send pub trait ErrorCallback: Send + Sync + Debug + 'static { + /// Handles one failed request described by `ctx`. + /// + /// The dispatcher's error task calls this once per failed request and awaits the returned future + /// before taking the next failure from the queue. fn call(&self, ctx: ErrorCtx) -> Pin + Send + 'static>>; } -/// Default implementation of [`ErrorCallback`] that logs the error using `tracing::error!`. +/// Default [`ErrorCallback`] implementation that logs the error using `tracing::error!`. +/// +/// Logs include stream, topic, optional partitioning, number of messages, how many earlier chunks +/// returned confirmations, and the cause. /// -/// Logs include stream, topic, optional partitioning, number of messages, how -/// many chunks committed before the failure, and the cause. +/// The messages themselves are dropped with the context, so a background producer that keeps this +/// callback has no way to recover them. Implement [`ErrorCallback`] to hold on to them. #[derive(Debug, Default)] pub struct LogErrorCallback; diff --git a/core/sdk/src/clients/producer_sharding.rs b/core/sdk/src/clients/producer_sharding.rs index 619470b31d..26656f69a6 100644 --- a/core/sdk/src/clients/producer_sharding.rs +++ b/core/sdk/src/clients/producer_sharding.rs @@ -34,7 +34,12 @@ use crate::clients::producer_error_callback::ErrorCtx; /// Implementors of this trait define how to choose a shard for a given batch of messages. /// This allows customizing message routing based on message content, stream/topic identifiers, /// or round-robin load balancing. +/// +/// [`pick_shard`](Self::pick_shard) must return an index smaller than `num_shards`. The dispatcher +/// normalizes a configured shard count of zero to one, so `num_shards` passed here is never zero. +/// Returning an out-of-range index makes dispatch panic when it indexes the shard list. pub trait Sharding: Send + Sync + std::fmt::Debug + 'static { + /// Chooses the zero-based shard index for one batch. fn pick_shard( &self, num_shards: usize, @@ -90,11 +95,19 @@ impl Sharding for OrderedSharding { } } +/// One producer send after the dispatcher has selected its destination metadata. +/// +/// A sharding strategy sees the messages, stream, and topic before this value is queued. The +/// optional partitioning value overrides the producer's configured partitioning for this send. #[derive(Debug)] pub struct ShardMessage { + /// Target stream. pub stream: Arc, + /// Target topic. pub topic: Arc, + /// Messages carried by this send. pub messages: Vec, + /// Per-send partitioning override, or `None` to use the producer configuration. pub partitioning: Option>, } @@ -113,7 +126,13 @@ impl Sizeable for ShardMessage { } } +/// A [`ShardMessage`] together with its charge against the dispatcher's byte budget. +/// +/// The optional permit is acquired before the message enters a shard queue and remains owned by +/// this value until the worker finishes the write or drops the message. Keeping permits from merged +/// messages separate avoids the `u32` permit-count limit in Tokio's semaphore API. pub struct ShardMessageWithPermit { + /// Routed send and destination metadata. pub inner: ShardMessage, size_bytes: u64, bytes_permit: Option, @@ -121,6 +140,9 @@ pub struct ShardMessageWithPermit { } impl ShardMessageWithPermit { + /// Wraps `msg` with its previously acquired byte-budget permit. + /// + /// `bytes_permit` is `None` when the dispatcher's byte budget is unbounded. pub fn new(msg: ShardMessage, bytes_permit: Option) -> Self { let size_bytes = msg.get_size_bytes().as_bytes_u64(); Self { @@ -141,6 +163,36 @@ impl ShardMessageWithPermit { } } +/// Represents one background worker of a +/// [`ProducerDispatcher`](crate::clients::producer_dispatcher::ProducerDispatcher), +/// together with the channel (queue) that feeds it. +/// +/// Each shard owns a task that buffers sends routed to it, merges adjacent sends that share a +/// destination, and writes them through [`ProducerCoreBackend::send_internal`]. +/// +/// The dispatcher enqueues a batch by sending it on that channel, which returns once the batch is +/// queued. Shards are created and owned by the dispatcher, so they are rarely handled directly. +/// +/// # Worker loop +/// +/// The task selects over three sources: +/// +/// - **An enqueued send.** [`ShardMessageWithPermit`]s are appended to the buffer, which is flushed +/// once it reaches [`batch_length`](BackgroundConfig::batch_length) queued sends or +/// [`batch_size`](BackgroundConfig::batch_size) reported bytes. Either threshold is disabled when +/// configured as `0`. +/// - **The linger deadline.** Armed when a send enters an empty buffer and due +/// [`linger_time`](BackgroundConfig::linger_time) later, it flushes the buffer whether or not +/// either batching threshold was reached. An empty buffer arms nothing, so an idle shard does not +/// wake up. +/// - **The stop broadcast** sent by the dispatcher on shutdown. Marks the shard closed +/// so later sends fail with [`IggyError::ProducerClosed`], drains what is still +/// queued, flushes once, then ends the loop. A send that races the stop signal can be queued +/// after that drain and is lost without an error. +/// Dropping the [`ProducerDispatcher`] instead of shutting it down provides no completion +/// guarantee and can lose buffered messages. +/// +/// [`ProducerDispatcher`]: crate::clients::producer_dispatcher::ProducerDispatcher pub struct Shard { tx: flume::Sender, closed: Arc, @@ -148,6 +200,7 @@ pub struct Shard { } impl Shard { + /// Spawns the worker task described in the [`Shard`] type documentation. pub fn new( core: Arc, config: Arc, @@ -162,14 +215,18 @@ impl Shard { let handle = tokio::spawn(async move { let mut buffer = Vec::new(); let mut buffer_bytes = 0; - let mut last_flush = tokio::time::Instant::now(); + // Armed by the first send buffered after a flush and polled only while the buffer is + // non-empty, so an idle shard never wakes up and a zero linger cannot spin. + let mut linger_deadline = tokio::time::Instant::now(); loop { - let deadline = last_flush + config.linger_time.get_duration(); tokio::select! { maybe_msg = rx.recv_async() => { match maybe_msg { Ok(msg) => { + if buffer.is_empty() { + linger_deadline = tokio::time::Instant::now() + config.linger_time.get_duration(); + } buffer_bytes += msg.size_bytes as usize; buffer.push(msg); debug!( @@ -196,18 +253,13 @@ impl Shard { new_buffer_bytes = buffer_bytes, "Buffer flushed" ); - - last_flush = tokio::time::Instant::now(); } } Err(_) => break, } } - _ = tokio::time::sleep_until(deadline) => { - if !buffer.is_empty() { - Self::flush_buffer(&core, &slots_permit, &mut buffer, &mut buffer_bytes, &err_sender).await; - } - last_flush = tokio::time::Instant::now(); + _ = tokio::time::sleep_until(linger_deadline), if !buffer.is_empty() => { + Self::flush_buffer(&core, &slots_permit, &mut buffer, &mut buffer_bytes, &err_sender).await; } _ = stop_rx.recv() => { closed_clone.store(true, Ordering::Release); @@ -297,6 +349,10 @@ impl Shard { *buffer_bytes = 0; } + /// Queues one [`ShardMessageWithPermit`] on this worker. + /// + /// Returns [`IggyError::ProducerClosed`] after graceful shutdown has closed the shard, or + /// [`IggyError::BackgroundSendError`] if its worker channel is disconnected. pub(crate) async fn send(&self, message: ShardMessageWithPermit) -> Result<(), IggyError> { if self.closed.load(Ordering::Acquire) { return Err(IggyError::ProducerClosed); @@ -644,6 +700,63 @@ mod tests { sleep(Duration::from_millis(100)).await; } + #[tokio::test] + async fn test_shard_flushes_at_once_with_zero_linger() { + let sends = Arc::new(AtomicUsize::new(0)); + let sends_seen_by_mock = sends.clone(); + let mut mock = MockProducerCoreBackend::new(); + mock.expect_send_internal().returning(move |_, _, _, _| { + sends_seen_by_mock.fetch_add(1, Ordering::SeqCst); + Box::pin(async { Ok(no_confirmations()) }) + }); + + let config = Arc::new( + BackgroundConfig::builder() + .batch_length(10) + .batch_size(10_000) + .linger_time(IggyDuration::from(0)) + .build(), + ); + let permit_bytes = Arc::new(Semaphore::new(10_000)); + let slots_permit = Arc::new(Semaphore::new(100)); + + let (stop_tx, stop_rx) = broadcast::channel(1); + let shard = Shard::new( + Arc::new(mock), + config, + slots_permit, + flume::unbounded().0, + stop_rx, + ); + + let message = ShardMessage { + stream: dummy_identifier(), + topic: dummy_identifier(), + messages: vec![dummy_message(1)], + partitioning: None, + }; + let wrapped = ShardMessageWithPermit::new( + message, + Some(permit_bytes.clone().acquire_many_owned(1).await.unwrap()), + ); + shard.send(wrapped).await.unwrap(); + + sleep(Duration::from_millis(50)).await; + assert_eq!( + sends.load(Ordering::SeqCst), + 1, + "a zero linger must flush without waiting for a batching threshold" + ); + + stop_tx.send(()).unwrap(); + shard.handle.await.unwrap(); + assert_eq!( + sends.load(Ordering::SeqCst), + 1, + "the stop flush must find nothing left" + ); + } + #[tokio::test] async fn test_shard_forwards_error() { let mut mock = MockProducerCoreBackend::new();