diff --git a/crates/adapters/nats/src/consumer.rs b/crates/adapters/nats/src/consumer.rs index d26b2156..b98191e9 100644 --- a/crates/adapters/nats/src/consumer.rs +++ b/crates/adapters/nats/src/consumer.rs @@ -126,6 +126,40 @@ fn consume_from_rx(rx: EnvelopeRx) -> BoxStream<'static, Result Option { + (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 { + 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); diff --git a/crates/adapters/nats/src/tests/consumer.rs b/crates/adapters/nats/src/tests/consumer.rs index d80c0d0e..c8d8dffa 100644 --- a/crates/adapters/nats/src/tests/consumer.rs +++ b/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()); +}