diff --git a/src/connection.rs b/src/connection.rs index 12525f25..daff339a 100644 --- a/src/connection.rs +++ b/src/connection.rs @@ -1726,6 +1726,11 @@ pub(crate) mod messages { initial_position: Some(options.initial_position.into()), schema: options.schema, start_message_id: options.start_message_id, + // proto default is 0 = no rollback; only send a + // meaningful value + start_message_rollback_duration_sec: options + .start_message_rollback_duration_secs + .filter(|secs| *secs > 0), ..Default::default() }), ..Default::default() @@ -2091,4 +2096,56 @@ mod tests { _ => panic!("Unexpected message"), }; } + + #[test] + fn subscribe_carries_start_message_rollback_duration() { + use crate::{consumer::ConsumerOptions, message::proto}; + + let msg = super::messages::subscribe( + "topic".to_string(), + "sub".to_string(), + crate::message::proto::command_subscribe::SubType::Exclusive, + 1, + 2, + None, + ConsumerOptions::default().with_start_message_rollback_duration_secs(600), + ); + let subscribe = msg.command.subscribe.as_ref().unwrap(); + assert_eq!(subscribe.start_message_rollback_duration_sec, Some(600)); + + // Zero and unset must not be sent (proto default 0 = no rollback). + let msg = super::messages::subscribe( + "topic".to_string(), + "sub".to_string(), + proto::command_subscribe::SubType::Exclusive, + 1, + 2, + None, + ConsumerOptions::default().with_start_message_rollback_duration_secs(0), + ); + assert_eq!( + msg.command + .subscribe + .unwrap() + .start_message_rollback_duration_sec, + None + ); + + let msg = super::messages::subscribe( + "topic".to_string(), + "sub".to_string(), + proto::command_subscribe::SubType::Exclusive, + 1, + 2, + None, + ConsumerOptions::default(), + ); + assert_eq!( + msg.command + .subscribe + .unwrap() + .start_message_rollback_duration_sec, + None + ); + } } diff --git a/src/consumer/data.rs b/src/consumer/data.rs index 38200c00..1c79a470 100644 --- a/src/consumer/data.rs +++ b/src/consumer/data.rs @@ -20,6 +20,12 @@ pub enum EngineMessage { Nack(MessageIdData), UnackedRedelivery, GetConnection(oneshot::Sender>>), + /// The consumer seeked to a new position: the engine must resubscribe + /// from there on reconnect instead of any previously recorded position. + /// `None` for timestamp seeks, where no message id is available — a + /// reconnect then falls back to the subscription's default initial + /// position rather than a stale pre-seek one. + SeekPosition(Option), } #[derive(Clone, Debug, PartialEq, Eq, Hash)] diff --git a/src/consumer/engine.rs b/src/consumer/engine.rs index 46144205..3e45be6c 100644 --- a/src/consumer/engine.rs +++ b/src/consumer/engine.rs @@ -51,6 +51,12 @@ pub struct ConsumerEngine { unacked_messages: HashMap, dead_letter_policy: Option, options: ConsumerOptions, + last_dequeued_message_id: Option, + /// Whether reconnects resubscribe from the last dequeued message id. + /// Captured once at construction (see [`tracks_resume_position`]) and + /// not reset when a later seek clears the rollback from `options`: the + /// cursor stays client-driven for the lifetime of this engine. + track_resume_position: bool, } impl ConsumerEngine { @@ -72,6 +78,7 @@ impl ConsumerEngine { options: ConsumerOptions, ) -> ConsumerEngine { let (event_tx, event_rx) = mpsc::unbounded(); + let track_resume_position = tracks_resume_position(&options); ConsumerEngine { client, connection, @@ -91,6 +98,8 @@ impl ConsumerEngine { unacked_messages: HashMap::new(), dead_letter_policy, options, + last_dequeued_message_id: None, + track_resume_position, } } @@ -337,6 +346,15 @@ impl ConsumerEngine { }); true } + Some(EngineMessage::SeekPosition(message_id)) => { + // Everything recorded before the seek is stale: rebase the + // subscribe options on the seek target so a reconnect resumes + // there, and drop the rollback so it cannot rewind past it. + self.options.start_message_id = message_id; + self.options.start_message_rollback_duration_secs = None; + self.last_dequeued_message_id = None; + true + } } } @@ -614,6 +632,7 @@ impl ConsumerEngine { error!("tx returned {:?}", e); Error::Custom("tx closed".to_string()) })?; + self.last_dequeued_message_id = Some(message_id.clone()); if let Some(duration) = self.unacked_message_redelivery_delay { self.unacked_messages.insert(message_id, now + duration); } @@ -637,6 +656,20 @@ impl ConsumerEngine { let broker_address = self.client.lookup_topic(&self.topic).await?; + // Re-applying the start-message rollback on every resubscribe would + // reset the cursor back by the window again after each broker/network + // blip, replaying messages this consumer already processed. Mirror the + // Java client (ConsumerImpl.clearReceiverQueue): once the consumer has + // made progress, resubscribe from the last dequeued message id instead + // of rolling back — clearing the rollback alone would resubscribe at + // the default position (latest) and skip whatever was left of the + // window. Gated to rollback consumers: see track_resume_position. + let options = resubscribe_options( + &self.options, + self.track_resume_position, + self.last_dequeued_message_id.as_ref(), + ); + let messages = retry_subscribe_consumer( &self.client, &mut self.connection, @@ -646,7 +679,7 @@ impl ConsumerEngine { self.sub_type, self.id, &self.name, - &self.options, + &options, self.batch_size, ) .await?; @@ -706,3 +739,165 @@ impl std::ops::Drop for ConsumerEngine { })); } } + +/// Options to use when resubscribing after a connection loss. +/// +/// Mirrors the Java client (`ConsumerImpl.clearReceiverQueue`): while the +/// consumer has made no progress the original options — including the +/// start-message rollback — are re-sent unchanged; once a message has been +/// dequeued, the resubscribe starts from that message id (the broker resumes +/// delivery after it) and the rollback is dropped. Sending the rollback again +/// would rewind the cursor by the whole window on every reconnect, while +/// dropping it without a resume position would jump to the default position +/// (latest) and skip the undrained remainder of the window. +/// +/// For batched messages the resume id carries the batch index of the last +/// dequeued message; the broker redelivers the containing entry, so a few +/// messages of that batch may be seen again (at-least-once, same as the +/// Java client's coarser entry-level resume). +/// +/// `track_resume_position` gates the whole rewrite to consumers whose +/// cursor position is client-driven (see [`tracks_resume_position`]). For +/// every other consumer — durable subscriptions in particular — the options +/// are returned unchanged: there the broker-side cursor governs, and +/// sending a resume id would move it past in-flight unacked messages, +/// losing them instead of redelivering. +/// A zero rollback means "disabled", matching the wire encoding (the +/// `CommandSubscribe` builder filters zero out as "no rollback") — it must +/// not opt the consumer into client-side resume tracking either. +fn rollback_enabled(options: &ConsumerOptions) -> bool { + options + .start_message_rollback_duration_secs + .is_some_and(|secs| secs > 0) +} + +/// Resume tracking applies to consumers whose cursor position is +/// client-driven: non-durable subscriptions (readers included — the broker +/// drops their cursor on disconnect, so without a resume id a reconnect +/// pins back to the configured start position or jumps to the default) and +/// consumers created with a start-message rollback. This is the Java +/// client's behavior: `ConsumerImpl.clearReceiverQueue` advances +/// `startMessageId` to the last dequeued message for non-durable +/// subscriptions. Durable subscriptions without a rollback are untouched: +/// their broker-side cursor governs the resume position, and sending a +/// resume id could move it past in-flight unacked messages. +fn tracks_resume_position(options: &ConsumerOptions) -> bool { + rollback_enabled(options) || options.durable == Some(false) +} + +#[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))] +fn resubscribe_options( + options: &ConsumerOptions, + track_resume_position: bool, + last_dequeued_message_id: Option<&MessageIdData>, +) -> ConsumerOptions { + let mut options = options.clone(); + if !track_resume_position { + return options; + } + if let Some(resume_from) = last_dequeued_message_id { + options.start_message_id = Some(resume_from.clone()); + options.start_message_rollback_duration_secs = None; + } + options +} + +#[cfg(test)] +mod tests { + use super::{resubscribe_options, rollback_enabled, tracks_resume_position}; + use crate::{consumer::ConsumerOptions, message::proto::MessageIdData}; + + #[test] + fn zero_rollback_is_disabled() { + // Matches the wire encoding: CommandSubscribe omits a zero rollback, + // so it must not opt the consumer into resume tracking either. + assert!(!rollback_enabled(&ConsumerOptions::default())); + assert!(!rollback_enabled( + &ConsumerOptions::default().with_start_message_rollback_duration_secs(0) + )); + assert!(rollback_enabled( + &ConsumerOptions::default().with_start_message_rollback_duration_secs(600) + )); + } + + #[test] + fn resume_tracking_covers_client_driven_cursors_only() { + // Durable subscription (the default): broker cursor governs. + assert!(!tracks_resume_position(&ConsumerOptions::default())); + assert!(!tracks_resume_position( + &ConsumerOptions::default().durable(true) + )); + // Rollback consumers and non-durable consumers (readers) are + // client-driven: a reconnect without a resume id would rewind to + // the configured start or jump to the default position. + assert!(tracks_resume_position( + &ConsumerOptions::default().with_start_message_rollback_duration_secs(600) + )); + assert!(tracks_resume_position( + &ConsumerOptions::default().durable(false) + )); + } + + fn message_id(ledger_id: u64, entry_id: u64) -> MessageIdData { + MessageIdData { + ledger_id, + entry_id, + ..Default::default() + } + } + + #[test] + fn resubscribe_without_progress_keeps_rollback_and_start_position() { + let options = ConsumerOptions::default() + .with_start_message_rollback_duration_secs(600) + .starting_on_message(message_id(1, 1)); + let resubscribe = resubscribe_options(&options, true, None); + assert_eq!(resubscribe.start_message_rollback_duration_secs, Some(600)); + assert_eq!(resubscribe.start_message_id, Some(message_id(1, 1))); + } + + #[test] + fn resubscribe_after_progress_resumes_from_last_dequeued() { + let options = ConsumerOptions::default() + .with_start_message_rollback_duration_secs(600) + .starting_on_message(message_id(1, 1)); + let last = message_id(7, 42); + let resubscribe = resubscribe_options(&options, true, Some(&last)); + // the rollback must not rewind the cursor again... + assert_eq!(resubscribe.start_message_rollback_duration_secs, None); + // ...and the resume position replaces the original start id, so the + // broker continues after the last delivered message instead of + // jumping to the default position (latest). + assert_eq!(resubscribe.start_message_id, Some(last)); + // everything else must be preserved + assert_eq!(resubscribe.durable, options.durable); + } + + #[test] + fn reader_resumes_from_last_dequeued_not_original_start() { + // The reader path of the gate: non-durable, no rollback. After + // progress, a reconnect resumes from the last dequeued message + // instead of replaying from the configured start position. + let options = ConsumerOptions::default() + .durable(false) + .starting_on_message(message_id(1, 1)); + let last = message_id(7, 42); + let resubscribe = resubscribe_options(&options, true, Some(&last)); + assert_eq!(resubscribe.start_message_id, Some(last)); + assert_eq!(resubscribe.durable, Some(false)); + } + + #[test] + fn resubscribe_without_rollback_never_rewrites_the_start_position() { + // A consumer NOT created with the rollback option must reconnect + // with its original options even after progress: for a durable + // subscription the broker cursor governs the resume position, and + // sending the last dequeued id would move it past in-flight + // unacked messages, losing them instead of redelivering. + let options = ConsumerOptions::default().starting_on_message(message_id(1, 1)); + let last = message_id(7, 42); + let resubscribe = resubscribe_options(&options, false, Some(&last)); + assert_eq!(resubscribe.start_message_id, Some(message_id(1, 1))); + assert_eq!(resubscribe.start_message_rollback_duration_secs, None); + } +} diff --git a/src/consumer/mod.rs b/src/consumer/mod.rs index 3111c8df..98b0870c 100644 --- a/src/consumer/mod.rs +++ b/src/consumer/mod.rs @@ -200,6 +200,10 @@ impl Consumer { InnerConsumer::Multi(c) => { c.seek(consumer_ids, message_id, timestamp).await?; let topics = c.topics(); + // The central config stays untouched: it only seeds topics + // this consumer has never subscribed to (update_topics() + // discoveries), and for those a configured rollback + // legitimately applies — it is their first subscribe. let config = c.config().clone(); //currently, pulsar only supports seek for non partitioned topics @@ -210,7 +214,21 @@ impl Consumer { let topic_addr_pair = c.topics.iter().cloned().zip(addrs.iter().cloned()); let consumers = try_join_all(topic_addr_pair.map(|(topic, addr)| { - TopicConsumer::new(client.clone(), topic, addr, config.clone()) + // Rebuild each topic from its own consumer's config, as + // is: TopicConsumer::seek already rebased the seeked + // topics' configs (seek target in, one-shot rollback + // out), and the topics that were NOT seeked keep their + // original options — including a configured rollback, + // whose bounded window replay is at-least-once for + // non-durable consumers, whereas dropping it would + // resubscribe them at the default position and skip + // messages. The shared config is only the fallback. + let topic_config = c + .consumers + .get(&topic) + .map(|old| old.config().clone()) + .unwrap_or_else(|| config.clone()); + TopicConsumer::new(client.clone(), topic, addr, topic_config) })) .await?; @@ -223,7 +241,6 @@ impl Consumer { let topic_refresh = Duration::from_secs(30); let refresh = Box::pin(client.executor.interval(topic_refresh).map(drop)); let namespace = c.namespace.clone(); - let config = c.config().clone(); let topic_regex = c.topic_regex.clone(); InnerConsumer::Multi(MultiTopicConsumer { namespace, diff --git a/src/consumer/options.rs b/src/consumer/options.rs index 2a504947..4b4d34af 100644 --- a/src/consumer/options.rs +++ b/src/consumer/options.rs @@ -33,6 +33,19 @@ pub struct ConsumerOptions { /// } /// ``` pub initial_position: InitialPosition, + /// If specified (and non-zero), the subscription will reset its cursor + /// back by this many seconds at subscribe time and replay messages from + /// that point (the broker's `startMessageRollbackDurationInSec`). + /// + /// This is the protocol-native way to "start N seconds back" for + /// readers and non-durable consumers, without a separate `seek()` after + /// creation (which forces a broker-side consumer close + reconnect). + /// + /// The rollback is applied at subscribe time and on resubscribes only + /// while the consumer has made no progress; once messages have been + /// dequeued, reconnects no longer send it (matching the Java client), + /// so a network blip does not rewind the cursor by the whole window. + pub start_message_rollback_duration_secs: Option, } impl ConsumerOptions { @@ -85,4 +98,13 @@ impl ConsumerOptions { self.receiver_queue_size = Some(if size == 0 { 1000 } else { size }); self } + + /// within options, rolls the subscription cursor back by the given + /// number of seconds at subscribe time (the broker replays messages + /// published within that window) + #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))] + pub fn with_start_message_rollback_duration_secs(mut self, seconds: u64) -> Self { + self.start_message_rollback_duration_secs = Some(seconds); + self + } } diff --git a/src/consumer/topic.rs b/src/consumer/topic.rs index e9e9d917..3f1e3b1e 100644 --- a/src/consumer/topic.rs +++ b/src/consumer/topic.rs @@ -244,8 +244,21 @@ impl TopicConsumer { self.connection() .await? .sender() - .seek(consumer_id, message_id, timestamp) + .seek(consumer_id, message_id.clone(), timestamp) .await?; + // Rebase the engine's resume position, otherwise a later reconnect + // would resubscribe from a stale pre-seek position and silently undo + // the seek. + self.engine_tx + .send(EngineMessage::SeekPosition(message_id.clone())) + .await + .map_err(|e| Error::Custom(format!("could not notify the engine of the seek: {e}")))?; + // Consumer::seek rebuilds a replacement TopicConsumer from this + // config; rebase it the same way or the replacement would subscribe + // with the stale pre-seek start position and re-apply the rollback, + // moving the cursor away from the seek target. + self.config.options.start_message_id = message_id; + self.config.options.start_message_rollback_duration_secs = None; Ok(()) }