fix: preserve consumer config on update, warn not abort; adr accuracy

This commit is contained in:
2026-10-26 01:43:22 +01:00
parent 0908b7046a
commit 56f3d1ac1b
3 changed files with 41 additions and 18 deletions

View File

@@ -141,22 +141,20 @@ fn stream_config_update(existing: &StreamConfig) -> Option<StreamConfig> {
}
/// `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.
/// or `ack_wait`. The update is a clone of the existing config with only those
/// two fields changed, so operator-set fields survive and immutable ones
/// (durable name, filter subject, deliver policy) cannot mismatch.
fn consumer_config_update(
existing: &jetstream::consumer::Config,
desired_max_deliver: i64,
) -> Option<pull::Config> {
) -> Option<jetstream::consumer::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(),
Some(jetstream::consumer::Config {
ack_wait: ACK_WAIT,
max_deliver: desired_max_deliver,
..Default::default()
..existing.clone()
})
}
@@ -185,7 +183,14 @@ impl NatsJetStreamConsumer {
"existing JetStream stream has a different duplicate window; updating to {:?}",
DEDUP_WINDOW
);
js.update_stream(updated).await?;
if let Err(e) = js.update_stream(updated).await {
tracing::warn!(
"could not update stream {}: {e}; continuing. Fix manually: nats stream edit {} --dupe-window {}s",
cfg.stream_name,
cfg.stream_name,
DEDUP_WINDOW.as_secs()
);
}
}
let subject_filter = consumer_subject_filter(&cfg.subject_prefix);
@@ -209,7 +214,15 @@ impl NatsJetStreamConsumer {
"existing JetStream consumer has different max_deliver/ack_wait; updating to max_deliver={}",
cfg.max_deliver
);
stream.update_consumer(updated).await?;
if let Err(e) = stream.update_consumer(updated).await {
tracing::warn!(
"could not update consumer {}: {e}; continuing. Fix manually: nats consumer edit {} {} --max-deliver {}",
cfg.consumer_name,
cfg.stream_name,
cfg.consumer_name,
cfg.max_deliver
);
}
}
let (tx, rx) = mpsc::channel(128);

View File

@@ -75,12 +75,16 @@ fn consumer_max_deliver_drift_is_updated() {
filter_subject: "a.>".into(),
max_deliver: -1,
ack_wait: ACK_WAIT,
max_ack_pending: 77,
description: Some("operator".into()),
..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.>");
assert_eq!(updated.max_ack_pending, 77);
assert_eq!(updated.description.as_deref(), Some("operator"));
let same = jetstream::consumer::Config {
max_deliver: 5,
..existing