feat: expose startMessageRollbackDurationInSec in ConsumerOptions

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
+        );
+    }
 }

src/consumer/engine.rs

@@ -51,6 +51,7 @@ pub struct ConsumerEngine<Exe: Executor> {
     unacked_messages: HashMap<MessageIdData, Instant>,
     dead_letter_policy: Option<DeadLetterPolicy>,
     options: ConsumerOptions,
+    dequeued_any_message: bool,
 }
 
 impl<Exe: Executor> ConsumerEngine<Exe> {
@@ -91,6 +92,7 @@ impl<Exe: Executor> ConsumerEngine<Exe> {
             unacked_messages: HashMap::new(),
             dead_letter_policy,
             options,
+            dequeued_any_message: false,
         }
     }
 
@@ -614,6 +616,7 @@ impl<Exe: Executor> ConsumerEngine<Exe> {
                 error!("tx returned {:?}", e);
                 Error::Custom("tx closed".to_string())
             })?;
+        self.dequeued_any_message = true;
         if let Some(duration) = self.unacked_message_redelivery_delay {
             self.unacked_messages.insert(message_id, now + duration);
         }
@@ -637,6 +640,14 @@ impl<Exe: Executor> ConsumerEngine<Exe> {
 
         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: the rollback is only sent while the consumer has made
+        // no progress (ConsumerImpl sends it only while startMessageId still
+        // equals the initial one).
+        let options = resubscribe_options(&self.options, self.dequeued_any_message);
+
         let messages = retry_subscribe_consumer(
             &self.client,
             &mut self.connection,
@@ -646,7 +657,7 @@ impl<Exe: Executor> ConsumerEngine<Exe> {
             self.sub_type,
             self.id,
             &self.name,
-            &self.options,
+            &options,
             self.batch_size,
         )
         .await?;
@@ -706,3 +717,45 @@ impl<Exe: Executor> std::ops::Drop for ConsumerEngine<Exe> {
         }));
     }
 }
+
+/// Options to use when resubscribing after a connection loss.
+///
+/// The start-message rollback must only be applied while the consumer has
+/// made no progress, matching the Java client (`ConsumerImpl` only sends
+/// `startMessageRollbackDurationInSec` while its `startMessageId` still
+/// equals the initial one). Re-sending it once messages were dequeued would
+/// rewind the cursor by the whole window on every reconnect.
+#[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
+fn resubscribe_options(options: &ConsumerOptions, dequeued_any_message: bool) -> ConsumerOptions {
+    let mut options = options.clone();
+    if dequeued_any_message {
+        options.start_message_rollback_duration_secs = None;
+    }
+    options
+}
+
+#[cfg(test)]
+mod tests {
+    use super::resubscribe_options;
+    use crate::consumer::ConsumerOptions;
+
+    #[test]
+    fn rollback_survives_resubscribe_without_progress() {
+        let options = ConsumerOptions::default().with_start_message_rollback_duration_secs(600);
+        assert_eq!(
+            resubscribe_options(&options, false).start_message_rollback_duration_secs,
+            Some(600)
+        );
+    }
+
+    #[test]
+    fn rollback_cleared_on_resubscribe_after_progress() {
+        let options = ConsumerOptions::default().with_start_message_rollback_duration_secs(600);
+        assert_eq!(
+            resubscribe_options(&options, true).start_message_rollback_duration_secs,
+            None
+        );
+        // everything else must be preserved
+        assert_eq!(resubscribe_options(&options, true).durable, options.durable);
+    }
+}

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<u64>,
 }
 
 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
+    }
 }