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/data.rs

@@ -20,6 +20,12 @@ pub enum EngineMessage<Exe: Executor> {
     Nack(MessageIdData),
     UnackedRedelivery,
     GetConnection(oneshot::Sender<Arc<Connection<Exe>>>),
+    /// 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<MessageIdData>),
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]

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,
+    last_dequeued_message_id: Option<MessageIdData>,
 }
 
 impl<Exe: Executor> ConsumerEngine<Exe> {
@@ -91,6 +92,7 @@ impl<Exe: Executor> ConsumerEngine<Exe> {
             unacked_messages: HashMap::new(),
             dead_letter_policy,
             options,
+            last_dequeued_message_id: None,
         }
     }
 
@@ -337,6 +339,15 @@ impl<Exe: Executor> ConsumerEngine<Exe> {
                 });
                 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 +625,7 @@ impl<Exe: Executor> ConsumerEngine<Exe> {
                 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 +649,16 @@ 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 (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.
+        let options = resubscribe_options(&self.options, self.last_dequeued_message_id.as_ref());
+
         let messages = retry_subscribe_consumer(
             &self.client,
             &mut self.connection,
@@ -646,7 +668,7 @@ impl<Exe: Executor> ConsumerEngine<Exe> {
             self.sub_type,
             self.id,
             &self.name,
-            &self.options,
+            &options,
             self.batch_size,
         )
         .await?;
@@ -706,3 +728,72 @@ impl<Exe: Executor> std::ops::Drop for ConsumerEngine<Exe> {
         }));
     }
 }
+
+/// 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).
+#[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
+fn resubscribe_options(
+    options: &ConsumerOptions,
+    last_dequeued_message_id: Option<&MessageIdData>,
+) -> ConsumerOptions {
+    let mut options = options.clone();
+    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;
+    use crate::{consumer::ConsumerOptions, message::proto::MessageIdData};
+
+    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, 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, 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);
+    }
+}

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

src/consumer/topic.rs

@@ -244,8 +244,15 @@ impl<T: DeserializeMessage, Exe: Executor> TopicConsumer<T, Exe> {
         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))
+            .await
+            .map_err(|e| Error::Custom(format!("could not notify the engine of the seek: {e}")))?;
         Ok(())
     }