From 56f3d1ac1b0734e64a87ce0fe19577786a1531bb Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Mon, 26 Oct 2026 01:43:22 +0100 Subject: [PATCH] fix: preserve consumer config on update, warn not abort; adr accuracy --- crates/adapters/nats/src/consumer.rs | 33 +++++++++++++++------- crates/adapters/nats/src/tests/consumer.rs | 4 +++ docs/adr/0015-transactional-outbox.md | 22 +++++++++------ 3 files changed, 41 insertions(+), 18 deletions(-) diff --git a/crates/adapters/nats/src/consumer.rs b/crates/adapters/nats/src/consumer.rs index b98191e9..1879beed 100644 --- a/crates/adapters/nats/src/consumer.rs +++ b/crates/adapters/nats/src/consumer.rs @@ -141,22 +141,20 @@ fn stream_config_update(existing: &StreamConfig) -> Option { } /// `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 { +) -> 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(), + 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); diff --git a/crates/adapters/nats/src/tests/consumer.rs b/crates/adapters/nats/src/tests/consumer.rs index c8d8dffa..73899f58 100644 --- a/crates/adapters/nats/src/tests/consumer.rs +++ b/crates/adapters/nats/src/tests/consumer.rs @@ -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 diff --git a/docs/adr/0015-transactional-outbox.md b/docs/adr/0015-transactional-outbox.md index 19b50a09..07f6a535 100644 --- a/docs/adr/0015-transactional-outbox.md +++ b/docs/adr/0015-transactional-outbox.md @@ -10,8 +10,10 @@ event ran its handlers twice. **The outbox is `event_queue`.** The same SQLite table the DB-backed bus already used is now the outbox for every backend. Migration `0101` adds `event_queue.event_id` (UUID, unique when set) and -`claimed_at`, and creates `processed_events` (`event_id`, `handler`, `status`, `started_at`, owner -token). Legacy rows have a NULL `event_id` and are still delivered; they just get no dedup. +`claimed_at`, and creates `processed_events` (`event_id`, `handler`, `status`, `started_at`, +`processed_at`). The migration backfills every existing row with a random event id, so legacy rows get dedup too. +The column stays nullable, but the only insert path (`enqueue_events`) always sets it, and the +consumer still tolerates a NULL id (no dedup) for rows from foreign writers. **Writes are atomic with their events.** Repository and command-port methods that change state take the events they cause and write them in the same transaction as the data. `shared::enqueue_events` @@ -33,8 +35,9 @@ backends dead-letter after the same number of attempts. `claimed_at` is older than 600 s, so a crash never strands an event in `processing`. 3. *Processed-event ledger.* Before a handler runs, `(event_id, handler)` is claimed in `processed_events` (INSERT `in_progress`, or take over a claim older than the 1800 s lease). - Success marks it `done`; failure releases it; both are conditioned on an owner token so a - stale worker cannot clobber a newer claim. A delivery that sees a fresh claim held elsewhere is + Success marks it `done`; failure releases it; both are conditioned on the claim's millisecond + `started_at` value (`WHERE ... AND started_at = ?`), which acts as the owner token, so a stale + worker cannot clobber a newer claim. A delivery that sees a fresh claim held elsewhere is deferred (`AckHandle::defer`: rescheduled to the lease expiry without consuming an attempt; JetStream `nak` with delay, which does count toward `max_deliver`). A `done` row skips the handler. Ledger rows are kept 30 days. A ledger read error skips the handler and nacks. @@ -60,8 +63,8 @@ backends dead-letter after the same number of attempts. Fixing it needs transaction hooks in k-ap. - **k-ap 0.5.0 mints random ids for some activities.** Create, Announce and Like ids are deterministic (`broadcast.rs:22-35`, `:283-288`; backfill `backfill.rs:208-212`). Delete - (`broadcast.rs:157`), Add/Undo (`:210`), Update (`:321`), Move (`:377`) and the actor update - (`:339-344`) use `Uuid::new_v4()` via `url_scheme.rs:52-53` with no override; only + (`broadcast.rs:157`), Add/Undo (`:210`), Update (`:321`), Move (`:377`), the Undo wrappers for + Announce (`:81`) and Like (`:129`) and the actor update (`:339-344`) use `Uuid::new_v4()` via `url_scheme.rs:52-53` with no override; only `broadcast_raw_to_followers` (`:395`) takes a caller id. A duplicate of those after a crash is a fresh-id activity. They are semantically idempotent for remote servers (deleting twice, updating to the same state), but it is a new activity, not a replay. The ledger makes this rare; closing it @@ -77,8 +80,11 @@ backends dead-letter after the same number of attempts. - Existing deployments keep their JetStream stream. `get_or_create_stream` applies config only on creation, so the consumer now compares an existing stream and consumer with what this build needs and updates `duplicate_window` (1 h) and the consumer's `max_deliver`/`ack_wait` in place, with a - warning in the log. Both are mutable on a live stream/consumer, so no recreate is needed; - immutable consumer fields (name, filter, deliver policy) are never touched. + warning in the log. Both are mutable on a live stream/consumer, so no recreate is needed. Each update is a + clone of the existing config with only those fields changed, so operator-set fields and the + immutable ones (name, filter, deliver policy) are preserved. If an update is rejected the worker + logs a warning with the manual `nats stream edit` / `nats consumer edit` command and keeps + starting. - Server and worker both migrate at startup (docker compose starts them together). `sqlite::migrate` retries with a bounded backoff (100 ms to 1 s, 5 retries), so the loser of a migration race re-reads `_sqlx_migrations`, skips what the winner applied and succeeds; only a persistent error