fix: apply dedup window + max_deliver to existing jetstream stream/consumer

crates/adapters/nats/src/consumer.rs

@@ -126,6 +126,40 @@ fn consume_from_rx(rx: EnvelopeRx) -> BoxStream<'static, Result<EventEnvelope, D
 /// silently follow the server default.
 const DEDUP_WINDOW: std::time::Duration = std::time::Duration::from_secs(60 * 60);
 
+const ACK_WAIT: std::time::Duration = std::time::Duration::from_secs(600);
+
+/// `Some(updated)` when an existing stream's config differs from what this
+/// build needs. `get_or_create_stream` only applies config on creation, so a
+/// stream created by an older build would otherwise keep the server default
+/// duplicate window (2 minutes). Only the fields this code owns are changed;
+/// everything else the operator set is preserved.
+fn stream_config_update(existing: &StreamConfig) -> Option<StreamConfig> {
+    (existing.duplicate_window != DEDUP_WINDOW).then(|| StreamConfig {
+        duplicate_window: DEDUP_WINDOW,
+        ..existing.clone()
+    })
+}
+
+/// `Some(updated)` when an existing durable consumer disagrees on `max_deliver`
+/// or `ack_wait`. Both are mutable on a live consumer, so the update is safe
+/// and keeps the consumer's delivery position. Immutable fields (durable name,
+/// filter subject, deliver policy) are never touched.
+fn consumer_config_update(
+    existing: &jetstream::consumer::Config,
+    desired_max_deliver: i64,
+) -> Option<pull::Config> {
+    if existing.max_deliver == desired_max_deliver && existing.ack_wait == ACK_WAIT {
+        return None;
+    }
+    Some(pull::Config {
+        durable_name: existing.durable_name.clone(),
+        filter_subject: existing.filter_subject.clone(),
+        ack_wait: ACK_WAIT,
+        max_deliver: desired_max_deliver,
+        ..Default::default()
+    })
+}
+
 // ── JetStream consumer ────────────────────────────────────────────────────────
 
 pub struct NatsJetStreamConsumer {
@@ -145,6 +179,14 @@ impl NatsJetStreamConsumer {
                 ..Default::default()
             })
             .await?;
+        if let Some(updated) = stream_config_update(&stream.cached_info().config) {
+            tracing::warn!(
+                stream = %cfg.stream_name,
+                "existing JetStream stream has a different duplicate window; updating to {:?}",
+                DEDUP_WINDOW
+            );
+            js.update_stream(updated).await?;
+        }
 
         let subject_filter = consumer_subject_filter(&cfg.subject_prefix);
         let consumer = stream
@@ -153,12 +195,22 @@ impl NatsJetStreamConsumer {
                 pull::Config {
                     durable_name: Some(cfg.consumer_name.clone()),
                     filter_subject: subject_filter,
-                    ack_wait: std::time::Duration::from_secs(600),
+                    ack_wait: ACK_WAIT,
                     max_deliver: cfg.max_deliver,
                     ..Default::default()
                 },
             )
             .await?;
+        if let Some(updated) =
+            consumer_config_update(&consumer.cached_info().config, cfg.max_deliver)
+        {
+            tracing::warn!(
+                consumer = %cfg.consumer_name,
+                "existing JetStream consumer has different max_deliver/ack_wait; updating to max_deliver={}",
+                cfg.max_deliver
+            );
+            stream.update_consumer(updated).await?;
+        }
 
         let (tx, rx) = mpsc::channel(128);
 

crates/adapters/nats/src/tests/consumer.rs

@@ -44,3 +44,46 @@ fn decode_message_rejects_a_malformed_id() {
 fn decode_message_rejects_garbage() {
     assert!(decode_message(b"{not json").is_err());
 }
+
+#[test]
+fn stream_with_the_default_window_is_updated_and_keeps_other_fields() {
+    let existing = StreamConfig {
+        name: "S".into(),
+        max_messages: 42,
+        duplicate_window: std::time::Duration::from_secs(120),
+        ..Default::default()
+    };
+    let updated = stream_config_update(&existing).unwrap();
+    assert_eq!(updated.duplicate_window, DEDUP_WINDOW);
+    assert_eq!(updated.max_messages, 42);
+    assert_eq!(updated.name, "S");
+}
+
+#[test]
+fn stream_with_the_right_window_is_left_alone() {
+    let existing = StreamConfig {
+        duplicate_window: DEDUP_WINDOW,
+        ..Default::default()
+    };
+    assert!(stream_config_update(&existing).is_none());
+}
+
+#[test]
+fn consumer_max_deliver_drift_is_updated() {
+    let existing = jetstream::consumer::Config {
+        durable_name: Some("c".into()),
+        filter_subject: "a.>".into(),
+        max_deliver: -1,
+        ack_wait: ACK_WAIT,
+        ..Default::default()
+    };
+    let updated = consumer_config_update(&existing, 5).unwrap();
+    assert_eq!(updated.max_deliver, 5);
+    assert_eq!(updated.durable_name.as_deref(), Some("c"));
+    assert_eq!(updated.filter_subject, "a.>");
+    let same = jetstream::consumer::Config {
+        max_deliver: 5,
+        ..existing
+    };
+    assert!(consumer_config_update(&same, 5).is_none());
+}