Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! The JetStream implementation of the bus.
3 : //!
4 : //! One `ANTARES_CHANGES` stream (Interest retention, subjects `changes.>`),
5 : //! one `ANTARES_REGISTRY` stream for registration deltas, one KV bucket
6 : //! (`antares_subscriptions`) for the compiled-subscription mirror.
7 : //!
8 : //! Consumer discipline, asserted not assumed (a lesson from Scorpio):
9 : //! *balanced* work (matcher, temporal recorder) = a shared DURABLE pull
10 : //! consumer — instances joining the same durable load-balance; *broadcast*
11 : //! work (per-instance mirrors) = an EPHEMERAL pull consumer — every instance
12 : //! sees every message. The distinction is explicit in the method you call,
13 : //! and `assert_topology` verifies the server agrees at startup.
14 : //!
15 : //! Delivery is at-least-once engineered to idempotent: publish-side dedup via
16 : //! `Nats-Msg-Id` inside the stream's duplicate window, explicit ack AFTER
17 : //! processing (never Scorpio's PRE_PROCESSING commit-before-work), bounded
18 : //! prefetch.
19 :
20 : use crate::{subjects, ChangeEvent};
21 : /// The consumer handle the balanced and broadcast constructors hand out.
22 : pub use async_nats::jetstream::consumer::PullConsumer;
23 : /// The handle [`NatsBus::subs_kv`] answers.
24 : pub use async_nats::jetstream::kv::Store as KvStore;
25 : use async_nats::jetstream::{self, consumer, stream};
26 : use futures_util::StreamExt;
27 :
28 : /// Bus failures. String-typed on purpose: callers either retry (drain loop)
29 : /// or die loudly (startup) — nobody branches on the variant.
30 : #[derive(Debug)]
31 : pub struct BusError(pub String);
32 :
33 : impl std::fmt::Display for BusError {
34 0 : fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
35 0 : write!(f, "bus error: {}", self.0)
36 0 : }
37 : }
38 : impl std::error::Error for BusError {}
39 :
40 0 : fn err<E: std::fmt::Display>(e: E) -> BusError {
41 0 : BusError(e.to_string())
42 0 : }
43 :
44 : pub const CHANGES_STREAM: &str = "ANTARES_CHANGES";
45 : pub const REGISTRY_STREAM: &str = "ANTARES_REGISTRY";
46 : pub const SUBS_BUCKET: &str = "antares_subscriptions";
47 :
48 : /// Bounded prefetch: how many unacked messages one consumer may
49 : /// hold. Any unbounded queue is a 3am page.
50 : pub const MAX_ACK_PENDING: i64 = 256;
51 :
52 : /// The balanced durables on `ANTARES_CHANGES`. Interest retention keeps a
53 : /// message only while a consumer is interested in it, so these exist from
54 : /// the first connect of ANY role: an api pod that publishes before a matcher
55 : /// pod has ever run would otherwise publish into a stream that discards
56 : /// every change, and its outbox drain frees the row on the publish ack.
57 : pub const BALANCED_DURABLES: &[&str] = &["matcher"];
58 :
59 : pub struct NatsBus {
60 : js: jetstream::Context,
61 : client: async_nats::Client,
62 : /// `Event::Connected` occurrences (the initial connect included; the
63 : /// getter subtracts it — surfaced on /q/health as `reconnects`).
64 : reconnects: std::sync::Arc<std::sync::atomic::AtomicU64>,
65 : }
66 :
67 : impl NatsBus {
68 : /// Connect and ensure the streams + KV bucket exist (idempotent).
69 0 : pub async fn connect(url: &str) -> Result<Self, BusError> {
70 0 : let reconnects = std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0));
71 0 : let counter = reconnects.clone();
72 0 : let client = async_nats::ConnectOptions::new()
73 0 : .event_callback(move |ev| {
74 0 : let counter = counter.clone();
75 0 : async move {
76 0 : match ev {
77 : // fires per successful (re)connect after the initial
78 : // ConnectOptions::connect returned
79 : async_nats::Event::Connected => {
80 0 : counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
81 0 : tracing::info!("bus reconnected to NATS");
82 : }
83 : async_nats::Event::Disconnected => {
84 0 : tracing::warn!("bus lost the NATS connection — reconnecting");
85 : }
86 0 : other => tracing::debug!("bus event: {other}"),
87 : }
88 0 : }
89 0 : })
90 0 : .connect(url)
91 0 : .await
92 0 : .map_err(err)?;
93 : // The change stream carries every tenant's ChangeEvent bodies
94 : // and MUST stay internal. If the server requires no auth, anything that
95 : // reaches it can read or forge all-tenant events — warn loudly so an
96 : // unauthenticated JetStream cluster is not shipped by accident.
97 0 : if !client.server_info().auth_required {
98 0 : tracing::warn!(
99 : "connected to a NATS server that requires NO authentication — the \
100 : ANTARES_CHANGES stream exposes all tenants' change events; require \
101 : nkey/creds/mTLS and network-isolate the JetStream cluster in production"
102 : );
103 0 : }
104 0 : let js = jetstream::new(client.clone());
105 0 : let bus = Self {
106 0 : js,
107 0 : client,
108 0 : reconnects,
109 0 : };
110 0 : bus.ensure_streams().await?;
111 0 : Ok(bus)
112 0 : }
113 :
114 : /// Live connection state for /q/health.
115 0 : pub fn connected(&self) -> bool {
116 0 : self.client.connection_state() == async_nats::connection::State::Connected
117 0 : }
118 :
119 : /// Core NATS request/reply, no JetStream: `payload` to `subject`, the
120 : /// reply within `within`; `None` when nobody answered in time (no
121 : /// subscriber is an immediate `None`).
122 0 : pub async fn request(
123 0 : &self,
124 0 : subject: String,
125 0 : payload: Vec<u8>,
126 0 : within: std::time::Duration,
127 0 : ) -> Option<Vec<u8>> {
128 0 : let req = async_nats::Request::new()
129 0 : .payload(payload.into())
130 0 : .timeout(Some(within));
131 0 : let reply = self.client.send_request(subject, req).await.ok()?;
132 0 : Some(reply.payload.to_vec())
133 0 : }
134 :
135 : /// Answer every core request on `subject` with `answer` of its payload,
136 : /// until the returned task is aborted or `answer` returns `None`.
137 0 : pub fn reply<F>(&self, subject: String, answer: F) -> tokio::task::JoinHandle<()>
138 0 : where
139 0 : F: Fn(&[u8]) -> Option<Vec<u8>> + Send + 'static,
140 : {
141 0 : let client = self.client.clone();
142 0 : tokio::spawn(async move {
143 0 : let mut requests = match client.subscribe(subject.clone()).await {
144 0 : Ok(s) => s,
145 0 : Err(e) => {
146 0 : tracing::warn!("reply subscription on {subject} failed: {e}");
147 0 : return;
148 : }
149 : };
150 0 : while let Some(msg) = requests.next().await {
151 0 : let Some(to) = msg.reply else { continue };
152 0 : let Some(bytes) = answer(&msg.payload) else {
153 0 : return;
154 : };
155 0 : if let Err(e) = client.publish(to, bytes.into()).await {
156 0 : tracing::warn!("reply on {subject} failed: {e}");
157 0 : }
158 : }
159 0 : })
160 0 : }
161 :
162 : /// Successful reconnects since startup. `Event::Connected` fires on
163 : /// EVERY successful connect including the initial one (connector.rs
164 : /// emits it unconditionally), so the first event is subtracted.
165 0 : pub fn reconnects(&self) -> u64 {
166 0 : self.reconnects
167 0 : .load(std::sync::atomic::Ordering::Relaxed)
168 0 : .saturating_sub(1)
169 0 : }
170 :
171 0 : async fn ensure_streams(&self) -> Result<(), BusError> {
172 : // Production runs replicas=3 on a 3-node JetStream cluster. Stream
173 : // replication is a CLIENT-side stream setting, so the deployment
174 : // manifests set ANTARES_NATS_REPLICAS=3; single-node dev/CI keeps 1.
175 : // Garbage is FATAL like every other config value: a typo silently
176 : // running replicas=1 on a 3-node cluster is a durability downgrade
177 : // nobody chose.
178 0 : let replicas = replicas_from(std::env::var("ANTARES_NATS_REPLICAS").ok().as_deref())?;
179 0 : self.js
180 0 : .get_or_create_stream(stream::Config {
181 0 : name: CHANGES_STREAM.into(),
182 0 : subjects: vec!["changes.>".into()],
183 0 : // Interest retention: each durable sees every message; a
184 0 : // message dies once every interested consumer acked it.
185 0 : // (WorkQueue would forbid multiple consumer groups.)
186 0 : retention: stream::RetentionPolicy::Interest,
187 0 : duplicate_window: std::time::Duration::from_secs(120),
188 0 : num_replicas: replicas,
189 0 : ..Default::default()
190 0 : })
191 0 : .await
192 0 : .map_err(err)?;
193 0 : for durable in BALANCED_DURABLES {
194 0 : self.consume_balanced(durable).await?;
195 : }
196 0 : self.js
197 0 : .get_or_create_stream(stream::Config {
198 0 : name: REGISTRY_STREAM.into(),
199 0 : subjects: vec!["registry.>".into()],
200 0 : // Broadcast deltas for per-instance mirrors: ephemeral
201 0 : // consumers carry no interest, so bound by age instead.
202 0 : retention: stream::RetentionPolicy::Limits,
203 0 : max_age: std::time::Duration::from_secs(600),
204 0 : num_replicas: replicas,
205 0 : ..Default::default()
206 0 : })
207 0 : .await
208 0 : .map_err(err)?;
209 0 : self.js
210 0 : .create_key_value(async_nats::jetstream::kv::Config {
211 0 : bucket: SUBS_BUCKET.into(),
212 0 : num_replicas: replicas,
213 0 : ..Default::default()
214 0 : })
215 0 : .await
216 0 : .map_err(err)?;
217 0 : Ok(())
218 0 : }
219 :
220 : /// Publish one change event with `Nats-Msg-Id` dedup. The id is
221 : /// the outbox seq so a drain retry after a crash is absorbed by the
222 : /// stream's duplicate window, not delivered twice.
223 0 : pub async fn publish(&self, ev: &ChangeEvent) -> Result<(), BusError> {
224 0 : let ev = ev.clone().claim_check(crate::CLAIM_CHECK_BYTES);
225 0 : let subject = subjects::change_subject(
226 0 : &ev.tenant,
227 0 : ev.types.first().map(String::as_str).unwrap_or(""),
228 0 : ev.entity_id.as_str(),
229 : );
230 0 : let mut headers = async_nats::HeaderMap::new();
231 0 : headers.insert(
232 : "Nats-Msg-Id",
233 0 : format!("{}:{}", ev.tenant.as_str(), ev.seq).as_str(),
234 : );
235 0 : let bytes = serde_json::to_vec(&ev).map_err(err)?;
236 : // double-ack: await the JetStream publish ack, not just the TCP write
237 0 : self.js
238 0 : .publish_with_headers(subject, headers, bytes.into())
239 0 : .await
240 0 : .map_err(err)?
241 0 : .await
242 0 : .map_err(err)?;
243 0 : Ok(())
244 0 : }
245 :
246 : /// Publish a registration CUD delta: the full registration document
247 : /// (or a `{"deleted": id}` tombstone) on the tenant's registry subject.
248 : /// The tenant is re-validated here because it becomes a subject token.
249 0 : pub async fn publish_registry(
250 0 : &self,
251 0 : tenant: &str,
252 0 : delta: &serde_json::Value,
253 0 : ) -> Result<(), BusError> {
254 0 : let tenant = antares_model::TenantId::new_internal(tenant).map_err(err)?;
255 0 : let bytes = serde_json::to_vec(delta).map_err(err)?;
256 0 : self.js
257 0 : .publish(subjects::registry_subject(&tenant), bytes.into())
258 0 : .await
259 0 : .map_err(err)?
260 0 : .await
261 0 : .map_err(err)?;
262 0 : Ok(())
263 0 : }
264 :
265 : /// The stream sequence of the newest delta held for one registration,
266 : /// `0` when none is (never published, or aged out by the stream's
267 : /// `max_age`): the revision [`Self::publish_registration`] compares
268 : /// against.
269 0 : pub async fn registration_revision(&self, tenant: &str, id: &str) -> Result<u64, BusError> {
270 0 : let tenant = antares_model::TenantId::new_internal(tenant).map_err(err)?;
271 0 : let s = self
272 0 : .js
273 0 : .get_stream_no_info(REGISTRY_STREAM)
274 0 : .await
275 0 : .map_err(err)?;
276 0 : match s
277 0 : .get_last_raw_message_by_subject(&subjects::registration_subject(&tenant, id))
278 0 : .await
279 : {
280 0 : Ok(m) => Ok(m.sequence),
281 0 : Err(e) if e.kind() == stream::LastRawMessageErrorKind::NoMessageFound => Ok(0),
282 0 : Err(e) => Err(err(e)),
283 : }
284 0 : }
285 :
286 : /// Publish one registration's delta on its own subject, only while
287 : /// `revision` is still the newest delta held for it
288 : /// (`Nats-Expected-Last-Subject-Sequence`; `0` = none held). A publish
289 : /// that a newer one overtook fails, and its caller re-reads: a stream
290 : /// has no key to overwrite, so this is what keeps a late delta from
291 : /// landing after a newer one in every mirror.
292 0 : pub async fn publish_registration(
293 0 : &self,
294 0 : tenant: &str,
295 0 : id: &str,
296 0 : delta: &serde_json::Value,
297 0 : revision: u64,
298 0 : ) -> Result<(), BusError> {
299 0 : let tenant = antares_model::TenantId::new_internal(tenant).map_err(err)?;
300 0 : let bytes = serde_json::to_vec(delta).map_err(err)?;
301 0 : let mut headers = async_nats::HeaderMap::new();
302 0 : headers.insert(
303 0 : async_nats::header::NATS_EXPECTED_LAST_SUBJECT_SEQUENCE,
304 0 : revision.to_string().as_str(),
305 : );
306 0 : self.js
307 0 : .publish_with_headers(
308 0 : subjects::registration_subject(&tenant, id),
309 0 : headers,
310 0 : bytes.into(),
311 0 : )
312 0 : .await
313 0 : .map_err(err)?
314 0 : .await
315 0 : .map_err(err)?;
316 0 : Ok(())
317 0 : }
318 :
319 : /// BALANCED consumption (matcher, temporal recorder): a shared durable —
320 : /// instances with the same `durable` name split the work. Explicit-ack,
321 : /// bounded prefetch; the caller acks AFTER processing.
322 0 : pub async fn consume_balanced(
323 0 : &self,
324 0 : durable: &str,
325 0 : ) -> Result<consumer::PullConsumer, BusError> {
326 0 : let s = self.js.get_stream(CHANGES_STREAM).await.map_err(err)?;
327 : // Get-or-create, never create-or-update: rewriting a live durable's
328 : // config from every connecting pod at once crashed a NATS 2.10.29
329 : // server (consumer updateConfig → cleanupNoInterestMessages on this
330 : // Interest stream). A durable an older build made keeps its config
331 : // — the server's 30 s ack_wait, which ACK_WAIT equals, and no
332 : // max_deliver — until an operator edits or recreates it.
333 0 : s.get_or_create_consumer(durable, balanced_config(durable))
334 0 : .await
335 0 : .map_err(err)
336 0 : }
337 :
338 : /// BROADCAST consumption (per-instance registry mirror): an ephemeral
339 : /// consumer — every instance sees every delta, and the consumer dies
340 : /// with the instance.
341 0 : pub async fn consume_registry_broadcast(&self) -> Result<consumer::PullConsumer, BusError> {
342 0 : let s = self.js.get_stream(REGISTRY_STREAM).await.map_err(err)?;
343 0 : s.create_consumer(consumer::pull::Config {
344 0 : // no durable name = ephemeral = broadcast
345 0 : durable_name: None,
346 0 : deliver_policy: consumer::DeliverPolicy::New,
347 0 : ack_policy: consumer::AckPolicy::Explicit,
348 0 : max_ack_pending: MAX_ACK_PENDING,
349 0 : ..Default::default()
350 0 : })
351 0 : .await
352 0 : .map_err(err)
353 0 : }
354 :
355 : /// Drop a consumer of the registry stream, the way a server restart or an
356 : /// inactivity gap drops an ephemeral one. The watcher's reopen path is
357 : /// only exercisable if that gap can be produced on demand.
358 0 : pub async fn delete_registry_consumer(&self, name: &str) -> Result<(), BusError> {
359 0 : let s = self.js.get_stream(REGISTRY_STREAM).await.map_err(err)?;
360 0 : s.delete_consumer(name).await.map_err(err)?;
361 0 : Ok(())
362 0 : }
363 :
364 : /// The KV bucket holding the compiled-subscription mirror.
365 0 : pub async fn subs_kv(&self) -> Result<KvStore, BusError> {
366 0 : self.js.get_key_value(SUBS_BUCKET).await.map_err(err)
367 0 : }
368 :
369 : /// The Scorpio `$[quarkus.uuid}` lesson: assert at startup that every
370 : /// balanced concern really is a shared durable on the server — a typo'd
371 : /// durable name would silently turn work-sharing into a private queue
372 : /// (or broadcast into load-balancing). Fatal on mismatch.
373 0 : pub async fn assert_topology(&self, balanced_durables: &[&str]) -> Result<(), BusError> {
374 0 : let s = self.js.get_stream(CHANGES_STREAM).await.map_err(err)?;
375 0 : for durable in balanced_durables {
376 0 : let info = s.consumer_info(durable).await.map_err(|e| {
377 0 : BusError(format!(
378 0 : "topology: balanced durable '{durable}' missing on {CHANGES_STREAM}: {e}"
379 0 : ))
380 0 : })?;
381 0 : if info.config.durable_name.as_deref() != Some(*durable) {
382 0 : return Err(BusError(format!(
383 0 : "topology: consumer '{durable}' is not durable — balanced work would \
384 0 : not be shared across instances"
385 0 : )));
386 0 : }
387 : }
388 0 : tracing::info!(
389 : durables = ?balanced_durables,
390 : "bus topology asserted: balanced durables present on {CHANGES_STREAM}"
391 : );
392 0 : Ok(())
393 0 : }
394 : }
395 :
396 : /// How long the server waits for a balanced message's ack before it
397 : /// redelivers it — to whichever pod pulls next. Explicit, not the server
398 : /// default, because [`run_balanced`]'s heartbeat is paced from it.
399 : pub const ACK_WAIT: std::time::Duration = std::time::Duration::from_secs(30);
400 :
401 : /// Deliveries of one balanced message before the server gives up on it.
402 : /// With the heartbeat a redelivery means the pod working on it stopped
403 : /// (crash, hang, lost connection); a change that takes ten pods down with it
404 : /// is a poison message, and dropping it beats a fleet in a crash loop.
405 : pub const MAX_DELIVER: i64 = 10;
406 :
407 : /// Messages one pull asks for. Each is held unacked until its turn, so this
408 : /// bounds what one slow message delays on this pod — the rest of the
409 : /// stream stays with the other pods of the durable.
410 : pub const PULL_BATCH: usize = 16;
411 :
412 : /// How long one pull waits for its batch to fill: messages arriving in the
413 : /// meantime are handed over at once, so this is no latency, only the
414 : /// cadence at which an idle consumer re-asks.
415 : const PULL_EXPIRES: std::time::Duration = std::time::Duration::from_secs(5);
416 :
417 : /// The configuration of a balanced durable (`ack_wait` and `max_deliver`
418 : /// above; the caller acks after processing).
419 0 : pub fn balanced_config(durable: &str) -> consumer::pull::Config {
420 0 : consumer::pull::Config {
421 0 : durable_name: Some(durable.into()),
422 0 : ack_policy: consumer::AckPolicy::Explicit,
423 0 : max_ack_pending: MAX_ACK_PENDING,
424 0 : ack_wait: ACK_WAIT,
425 0 : max_deliver: MAX_DELIVER,
426 0 : ..Default::default()
427 0 : }
428 0 : }
429 :
430 : /// Drive a balanced consumer: `work` runs once per message, in delivery
431 : /// order, and the message is acked when it has returned — never before
432 : /// (ADR-0002, at-least-once).
433 : ///
434 : /// Messages are pulled [`PULL_BATCH`] at a time, and every one this pod
435 : /// holds unacked — the one in `work` and the rest of its batch already
436 : /// delivered behind it — gets an `AckKind::Progress` every third of the
437 : /// consumer's `ack_wait`. Without it one slow message (a notification
438 : /// endpoint at its timeout) outlived `ack_wait`, and the server redelivered
439 : /// it and everything queued behind it, so each was matched and notified
440 : /// twice. Now only a pod that stops working lets `ack_wait` pass.
441 : ///
442 : /// Returns on a pull that fails (a deleted consumer, a lost connection); the
443 : /// caller re-opens.
444 0 : pub async fn run_balanced<F, Fut>(
445 0 : consumer: &consumer::PullConsumer,
446 0 : mut work: F,
447 0 : ) -> Result<(), BusError>
448 0 : where
449 0 : F: FnMut(&async_nats::jetstream::Message) -> Fut,
450 0 : Fut: std::future::Future<Output = ()>,
451 0 : {
452 0 : let ack_wait = match consumer.cached_info().config.ack_wait {
453 0 : w if w.is_zero() => ACK_WAIT,
454 0 : w => w,
455 : };
456 0 : let every = ack_wait / 3;
457 : loop {
458 0 : let mut batch = consumer
459 0 : .batch()
460 0 : .max_messages(PULL_BATCH)
461 0 : .expires(PULL_EXPIRES)
462 0 : .messages()
463 0 : .await
464 0 : .map_err(err)?;
465 : // delivered and not yet worked on, in delivery order
466 0 : let mut held = std::collections::VecDeque::new();
467 0 : let mut open = true;
468 0 : let mut failed = None;
469 : // One clock for the whole pull: the messages in `held` were
470 : // delivered when the batch was, so their `ack_wait` runs from there
471 : // whatever happens to each one before them. A clock restarted per
472 : // message never ticked behind a run of messages each shorter than
473 : // a third of `ack_wait`, and the batch behind them was redelivered.
474 0 : let mut beat = tokio::time::interval_at(tokio::time::Instant::now() + every, every);
475 0 : beat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
476 : loop {
477 0 : let msg = match held.pop_front() {
478 0 : Some(m) => m,
479 0 : None if open => match batch.next().await {
480 0 : Some(Ok(m)) => m,
481 0 : Some(Err(e)) => return Err(err(e)),
482 0 : None => break,
483 : },
484 0 : None => break,
485 : };
486 0 : let job = work(&msg);
487 0 : tokio::pin!(job);
488 : loop {
489 0 : tokio::select! {
490 0 : () = &mut job => break,
491 0 : _ = beat.tick() => {
492 0 : for m in std::iter::once(&msg).chain(held.iter()) {
493 0 : if let Err(e) = m.ack_with(jetstream::AckKind::Progress).await {
494 0 : tracing::warn!("progress ack failed: {e}");
495 0 : }
496 : }
497 : }
498 0 : next = batch.next(), if open => match next {
499 0 : Some(Ok(m)) => held.push_back(m),
500 0 : Some(Err(e)) => {
501 0 : open = false;
502 0 : failed = Some(err(e));
503 0 : }
504 0 : None => open = false,
505 : },
506 : }
507 : }
508 0 : if let Err(e) = msg.ack().await {
509 0 : tracing::warn!("ack failed: {e}");
510 0 : }
511 : }
512 0 : if let Some(e) = failed {
513 0 : return Err(e);
514 0 : }
515 : }
516 0 : }
517 :
518 : /// Decode one JetStream message into a `ChangeEvent`. `None` = alien bytes —
519 : /// log-and-ack territory for the consumer (redelivering garbage forever is
520 : /// the alternative). Defence in depth: the subject's tenant segment must agree with
521 : /// the event body — consumers re-verify so a subject-mapping bug can never
522 : /// route one tenant's change into another tenant's processing.
523 : /// `ANTARES_NATS_REPLICAS`: absent = 1; present must be a positive integer.
524 18 : fn replicas_from(v: Option<&str>) -> Result<usize, BusError> {
525 18 : match v {
526 2 : None => Ok(1),
527 16 : Some(raw) => raw
528 16 : .trim()
529 16 : .parse::<usize>()
530 16 : .ok()
531 16 : .filter(|n| *n >= 1)
532 16 : .ok_or_else(|| {
533 12 : BusError(format!(
534 12 : "ANTARES_NATS_REPLICAS must be a positive integer, got {raw:?}"
535 12 : ))
536 12 : }),
537 : }
538 18 : }
539 :
540 0 : pub fn decode(msg: &async_nats::jetstream::Message) -> Option<ChangeEvent> {
541 0 : let ev: ChangeEvent = serde_json::from_slice(&msg.payload).ok()?;
542 0 : if !subject_tenant_agrees(msg.subject.as_str(), &ev) {
543 0 : tracing::error!(
544 0 : subject = %msg.subject,
545 0 : tenant = %ev.tenant.as_str(),
546 : "dropping change event: subject tenant segment disagrees with body"
547 : );
548 0 : return None;
549 0 : }
550 0 : Some(ev)
551 0 : }
552 :
553 : /// The tenant-agreement check, unit-testable: `changes.{tenant}.…` must carry the
554 : /// event's own tenant. Non-`changes` subjects pass (registry deltas carry
555 : /// tenant in the body only).
556 18 : pub fn subject_tenant_agrees(subject: &str, ev: &ChangeEvent) -> bool {
557 18 : let mut parts = subject.split('.');
558 18 : if parts.next() != Some("changes") {
559 2 : return true;
560 16 : }
561 16 : parts.next() == Some(ev.tenant.as_str())
562 18 : }
563 :
564 : /// Drive a pull consumer as a message stream. Thin wrapper so wiring code
565 : /// does not import futures/consumer types everywhere.
566 0 : pub async fn messages(
567 0 : consumer: &consumer::PullConsumer,
568 0 : ) -> Result<
569 0 : impl futures_util::Stream<
570 0 : Item = Result<
571 0 : async_nats::jetstream::Message,
572 0 : async_nats::error::Error<consumer::pull::MessagesErrorKind>,
573 0 : >,
574 0 : > + '_,
575 0 : BusError,
576 0 : > {
577 0 : consumer.messages().await.map_err(err)
578 0 : }
579 :
580 : /// One decoded registry delta from the broadcast consumer, acked in place.
581 0 : pub async fn next_delta(
582 0 : stream: &mut (impl futures_util::Stream<
583 0 : Item = Result<
584 0 : async_nats::jetstream::Message,
585 0 : async_nats::error::Error<consumer::pull::MessagesErrorKind>,
586 0 : >,
587 0 : > + Unpin),
588 0 : ) -> Option<serde_json::Value> {
589 : loop {
590 0 : let msg = stream.next().await?.ok()?;
591 0 : let parsed = serde_json::from_slice(&msg.payload).ok();
592 0 : let _ = msg.ack().await;
593 0 : if parsed.is_some() {
594 0 : return parsed;
595 0 : }
596 : }
597 0 : }
598 :
599 : #[cfg(test)]
600 : mod tests {
601 : use super::*;
602 : use crate::ChangeOp;
603 : use antares_model::{EntityId, TenantId};
604 :
605 : /// Config fatality: a typo'd replica count must never silently run a
606 : /// 3-node deployment at replicas=1 — that is a durability downgrade
607 : /// nobody chose. Absent stays 1; whitespace is tolerated; garbage and
608 : /// zero name the key in the error.
609 : #[test]
610 2 : fn replica_count_garbage_is_fatal_not_a_silent_default() {
611 2 : assert_eq!(replicas_from(None).map_err(|e| e.0), Ok(1));
612 2 : assert_eq!(replicas_from(Some("3")).map_err(|e| e.0), Ok(3));
613 2 : assert_eq!(replicas_from(Some(" 3 ")).map_err(|e| e.0), Ok(3));
614 12 : for bad in ["three", "", "0", "-1", "1.5", "3 nodes"] {
615 12 : let err = replicas_from(Some(bad))
616 12 : .map(|_| ())
617 12 : .expect_err("garbage must be refused")
618 : .0;
619 12 : assert!(
620 12 : err.contains("ANTARES_NATS_REPLICAS") && err.contains(bad.trim()),
621 : "the error must name the key and the value: {err}"
622 : );
623 : }
624 2 : }
625 :
626 : #[test]
627 2 : fn subject_tenant_reverification_drops_mismatches() {
628 2 : let ev = ChangeEvent {
629 2 : tenant: TenantId::new("acme").expect("tenant"),
630 2 : entity_id: EntityId::new("urn:x:1").expect("id"),
631 2 : types: vec!["T".into()],
632 2 : op: ChangeOp::Create,
633 2 : changed_attrs: vec![],
634 2 : payload: None,
635 2 : prev_payload: None,
636 2 : version: 1,
637 2 : incarnation: String::new(),
638 2 : seq: 1,
639 2 : payload_ref: None,
640 2 : prev_payload_ref: None,
641 2 : };
642 2 : assert!(subject_tenant_agrees("changes.acme.aa.bb", &ev));
643 2 : assert!(
644 2 : !subject_tenant_agrees("changes.other.aa.bb", &ev),
645 : "a mis-mapped subject must be dropped, not processed"
646 : );
647 2 : assert!(
648 2 : subject_tenant_agrees("registry.other", &ev),
649 : "non-changes subjects carry tenant in the body only"
650 : );
651 2 : }
652 :
653 : /// The check compares whole tokens: a subject whose tenant segment only
654 : /// starts with, contains or is missing the event's tenant must not pass.
655 : #[test]
656 2 : fn tenant_agreement_is_not_fooled_by_partial_tokens() {
657 2 : let ev = ChangeEvent {
658 2 : tenant: TenantId::new("acme").expect("tenant"),
659 2 : entity_id: EntityId::new("urn:x:1").expect("id"),
660 2 : types: vec![],
661 2 : op: ChangeOp::Delete,
662 2 : changed_attrs: vec![],
663 2 : payload: None,
664 2 : prev_payload: None,
665 2 : version: 1,
666 2 : incarnation: String::new(),
667 2 : seq: 1,
668 2 : payload_ref: None,
669 2 : prev_payload_ref: None,
670 2 : };
671 12 : for subject in [
672 2 : "changes.acmeX.aa.bb", // longer token
673 2 : "changes.acm.aa.bb", // prefix of the tenant
674 2 : "changes..aa.bb", // empty tenant segment
675 2 : "changes.aa.acme.bb", // tenant present, wrong position
676 2 : "changes", // no tenant segment at all
677 2 : "changes.",
678 2 : ] {
679 12 : assert!(
680 12 : !subject_tenant_agrees(subject, &ev),
681 : "must not accept {subject:?}"
682 : );
683 : }
684 2 : }
685 : }
|
