Add Nak backoff

Cargo.toml

@@ -50,13 +50,15 @@ rustls = "0.19.1"
 rustls-native-certs = "0.5.0"
 rustls-pemfile = "0.2.1"
 webpki = "0.21.0"
-serde = { version = "1.0.126", features = ["derive"] }
+serde = { version = "1.0.136", features = ["derive"] }
 serde_json = "1.0.64"
-serde_nanos = "0.1.1"
+# serde_nanos = { path = "../serde_nanos" }
+serde_nanos = "0.1"
 serde_repr = "0.1.7"
 memchr = "2.4.0"
 url = "2.2.2"
 time = { version = "0.3.6", features = ["parsing", "formatting", "serde", "serde-well-known"] }
+format-bytes = "0.1"
 
 [target.'cfg(unix)'.dependencies]
 libc = "0.2.98"

src/jetstream/types.rs

@@ -215,6 +215,9 @@ pub struct ConsumerConfig {
     /// Maximum number of times a specific message will be delivered. Use this to avoid poison pill messages that repeatedly crash your consumer processes forever.
     #[serde(default, skip_serializing_if = "is_default")]
     pub max_deliver: i64,
+    // Array of durations representing backoff directive that is used for delivery retries
+    #[serde(default, skip_serializing_if = "is_default")]
+    pub backoff: Vec<Duration>,
     /// When consuming from a Stream with many subjects, or wildcards, this selects only specific incoming subjects. Supports wildcards.
     #[serde(default, skip_serializing_if = "is_default")]
     pub filter_subject: String,
@@ -649,7 +652,7 @@ pub enum AckKind {
     /// Signals that the message will not be processed now
     /// and processing can move onto the next message, NAK'd
     /// message will be retried.
-    Nak,
+    Nak(Option<Duration>),
     /// When sent before the AckWait period indicates that
     /// work is ongoing and the period should be extended by
     /// another equal to AckWait.
@@ -663,15 +666,16 @@ pub enum AckKind {
     Term,
 }
 
-impl AsRef<[u8]> for AckKind {
-    fn as_ref(&self) -> &[u8] {
+impl AckKind {
+    pub fn to_bytes(&self) -> Vec<u8> {
         use AckKind::*;
         match self {
-            Ack => b"+ACK",
-            Nak => b"-NAK",
-            Progress => b"+WPI",
-            Next => b"+NXT",
-            Term => b"+TERM",
+            Ack => b"+ACK".to_vec(),
+            Nak(None) => b"-NAK".to_vec(),
+            Nak(Some(delay)) => format!("-NAK {}", delay.as_nanos()).into_bytes(),
+            Progress => b"+WPI".to_vec(),
+            Next => b"+NXT".to_vec(),
+            Term => b"+TERM".to_vec(),
         }
     }
 }

src/message.rs

@@ -199,7 +199,7 @@ impl Message {
     ///
     /// Does not check whether this message has already been double-acked.
     pub fn ack_kind(&self, ack_kind: crate::jetstream::AckKind) -> io::Result<()> {
-        self.respond(ack_kind)
+        self.respond(ack_kind.to_bytes())
     }
 
     /// Acknowledge a `JetStream` message and wait for acknowledgement from the server
@@ -241,7 +241,8 @@ impl Message {
             let sub =
                 crate::Subscription::new(sid, ack_reply.to_string(), receiver, client.clone());
 
-            let pub_ret = client.publish(original_reply, Some(&ack_reply), None, ack_kind.as_ref());
+            let pub_ret =
+                client.publish(original_reply, Some(&ack_reply), None, &ack_kind.to_bytes());
             if pub_ret.is_err() {
                 std::thread::sleep(std::time::Duration::from_millis(100));
                 continue;

tests/jetstream.rs

@@ -11,6 +11,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::time::Instant;
 use std::{io, time::Duration};
 
 mod util;
@@ -789,3 +790,72 @@ fn jetstream_pull_subscribe_bad_stream() {
     js.pull_subscribe("WRONG")
         .expect_err("expected not found stream for a given subject");
 }
+
+#[test]
+fn jetstream_nak_backoff() {
+    let s = util::run_server("tests/configs/jetstream.conf");
+    let con = nats::connect(s.client_url()).unwrap();
+
+    let js = nats::jetstream::new(con);
+
+    js.add_stream(&StreamConfig {
+        name: "TEST".to_string(),
+        subjects: vec!["foo".to_string()],
+        ..Default::default()
+    })
+    .unwrap();
+
+    js.stream_info("TEST").unwrap();
+    js.add_consumer(
+        "TEST",
+        ConsumerConfig {
+            durable_name: Some("CONSUMER".to_string()),
+            backoff: DurationVec {
+                inner: vec![
+                    Duration::from_secs(1),
+                    Duration::from_secs(2),
+                    Duration::from_secs(3),
+                ],
+            },
+            ..Default::default()
+        },
+    )
+    .unwrap();
+
+    // for _ in 0..10 {
+    js.publish("foo", b"data").unwrap();
+    // }
+
+    let consumer = js
+        .subscribe_with_options(
+            "foo",
+            &SubscribeOptions::bind("TEST".to_string(), "CONSUMER".to_string()),
+        )
+        .unwrap();
+
+    let message = consumer.next();
+    let mut now = Instant::now();
+    if let Some(message) = message {
+        // Nak the first message.
+        message.ack_kind(AckKind::Nak(None)).unwrap();
+    }
+    if let Some(message) = consumer.next() {
+        // check if we had to wait defined amount of time before getting redelivery.
+        assert!(now.elapsed().ge(&Duration::from_secs(1)));
+        message.ack_kind(AckKind::Nak(None)).unwrap();
+        now = Instant::now();
+    }
+    if let Some(message) = consumer.next() {
+        // check if we had to wait defined amount of time before getting another redelivery.
+        assert!(now.elapsed().ge(&Duration::from_secs(2)));
+        message
+            .ack_kind(AckKind::Nak(Some(Duration::from_millis(100))))
+            .unwrap();
+        now = Instant::now();
+    }
+    if let Some(message) = consumer.next() {
+        // check if we had to wait defined amount of time before getting another redelivery.
+        assert!(now.elapsed().le(&Duration::from_secs(200)));
+        message.ack().unwrap();
+    }
+}