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 : use async_nats::jetstream::{self, consumer, stream};
22 : use futures_util::StreamExt;
23 :
24 : /// Bus failures. String-typed on purpose: callers either retry (drain loop)
25 : /// or die loudly (startup) — nobody branches on the variant.
26 : #[derive(Debug)]
27 : pub struct BusError(pub String);
28 :
29 : impl std::fmt::Display for BusError {
30 0 : fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
31 0 : write!(f, "bus error: {}", self.0)
32 0 : }
33 : }
34 : impl std::error::Error for BusError {}
35 :
36 0 : fn err<E: std::fmt::Display>(e: E) -> BusError {
37 0 : BusError(e.to_string())
38 0 : }
39 :
40 : pub const CHANGES_STREAM: &str = "ANTARES_CHANGES";
41 : pub const REGISTRY_STREAM: &str = "ANTARES_REGISTRY";
42 : pub const SUBS_BUCKET: &str = "antares_subscriptions";
43 :
44 : /// Bounded prefetch: how many unacked messages one consumer may
45 : /// hold. Any unbounded queue is a 3am page.
46 : pub const MAX_ACK_PENDING: i64 = 256;
47 :
48 : pub struct NatsBus {
49 : js: jetstream::Context,
50 : client: async_nats::Client,
51 : /// `Event::Connected` occurrences (the initial connect included; the
52 : /// getter subtracts it — surfaced on /q/health as `reconnects`).
53 : reconnects: std::sync::Arc<std::sync::atomic::AtomicU64>,
54 : }
55 :
56 : impl NatsBus {
57 : /// Connect and ensure the streams + KV bucket exist (idempotent).
58 0 : pub async fn connect(url: &str) -> Result<Self, BusError> {
59 0 : let reconnects = std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0));
60 0 : let counter = reconnects.clone();
61 0 : let client = async_nats::ConnectOptions::new()
62 0 : .event_callback(move |ev| {
63 0 : let counter = counter.clone();
64 0 : async move {
65 0 : match ev {
66 : // fires per successful (re)connect after the initial
67 : // ConnectOptions::connect returned
68 : async_nats::Event::Connected => {
69 0 : counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
70 0 : tracing::info!("bus reconnected to NATS");
71 : }
72 : async_nats::Event::Disconnected => {
73 0 : tracing::warn!("bus lost the NATS connection — reconnecting");
74 : }
75 0 : other => tracing::debug!("bus event: {other}"),
76 : }
77 0 : }
78 0 : })
79 0 : .connect(url)
80 0 : .await
81 0 : .map_err(err)?;
82 : // The change stream carries every tenant's ChangeEvent bodies
83 : // and MUST stay internal. If the server requires no auth, anything that
84 : // reaches it can read or forge all-tenant events — warn loudly so an
85 : // unauthenticated JetStream cluster is not shipped by accident.
86 0 : if !client.server_info().auth_required {
87 0 : tracing::warn!(
88 : "connected to a NATS server that requires NO authentication — the \
89 : ANTARES_CHANGES stream exposes all tenants' change events; require \
90 : nkey/creds/mTLS and network-isolate the JetStream cluster in production"
91 : );
92 0 : }
93 0 : let js = jetstream::new(client.clone());
94 0 : let bus = Self {
95 0 : js,
96 0 : client,
97 0 : reconnects,
98 0 : };
99 0 : bus.ensure_streams().await?;
100 0 : Ok(bus)
101 0 : }
102 :
103 : /// Live connection state for /q/health.
104 0 : pub fn connected(&self) -> bool {
105 0 : self.client.connection_state() == async_nats::connection::State::Connected
106 0 : }
107 :
108 : /// Successful reconnects since startup. `Event::Connected` fires on
109 : /// EVERY successful connect including the initial one (connector.rs
110 : /// emits it unconditionally), so the first event is subtracted.
111 0 : pub fn reconnects(&self) -> u64 {
112 0 : self.reconnects
113 0 : .load(std::sync::atomic::Ordering::Relaxed)
114 0 : .saturating_sub(1)
115 0 : }
116 :
117 0 : async fn ensure_streams(&self) -> Result<(), BusError> {
118 : // Production runs replicas=3 on a 3-node JetStream cluster. Stream
119 : // replication is a CLIENT-side stream setting, so the deployment
120 : // manifests set ANTARES_NATS_REPLICAS=3; single-node dev/CI keeps 1.
121 : // Garbage is FATAL like every other config value: a typo silently
122 : // running replicas=1 on a 3-node cluster is a durability downgrade
123 : // nobody chose.
124 0 : let replicas = replicas_from(std::env::var("ANTARES_NATS_REPLICAS").ok().as_deref())?;
125 0 : self.js
126 0 : .get_or_create_stream(stream::Config {
127 0 : name: CHANGES_STREAM.into(),
128 0 : subjects: vec!["changes.>".into()],
129 0 : // Interest retention: each durable sees every message; a
130 0 : // message dies once every interested consumer acked it.
131 0 : // (WorkQueue would forbid multiple consumer groups.)
132 0 : retention: stream::RetentionPolicy::Interest,
133 0 : duplicate_window: std::time::Duration::from_secs(120),
134 0 : num_replicas: replicas,
135 0 : ..Default::default()
136 0 : })
137 0 : .await
138 0 : .map_err(err)?;
139 0 : self.js
140 0 : .get_or_create_stream(stream::Config {
141 0 : name: REGISTRY_STREAM.into(),
142 0 : subjects: vec!["registry.>".into()],
143 0 : // Broadcast deltas for per-instance mirrors: ephemeral
144 0 : // consumers carry no interest, so bound by age instead.
145 0 : retention: stream::RetentionPolicy::Limits,
146 0 : max_age: std::time::Duration::from_secs(600),
147 0 : num_replicas: replicas,
148 0 : ..Default::default()
149 0 : })
150 0 : .await
151 0 : .map_err(err)?;
152 0 : self.js
153 0 : .create_key_value(async_nats::jetstream::kv::Config {
154 0 : bucket: SUBS_BUCKET.into(),
155 0 : num_replicas: replicas,
156 0 : ..Default::default()
157 0 : })
158 0 : .await
159 0 : .map_err(err)?;
160 0 : Ok(())
161 0 : }
162 :
163 : /// Publish one change event with `Nats-Msg-Id` dedup. The id is
164 : /// the outbox seq so a drain retry after a crash is absorbed by the
165 : /// stream's duplicate window, not delivered twice.
166 0 : pub async fn publish(&self, ev: &ChangeEvent) -> Result<(), BusError> {
167 0 : let ev = ev.clone().claim_check(crate::CLAIM_CHECK_BYTES);
168 0 : let subject = subjects::change_subject(
169 0 : &ev.tenant,
170 0 : ev.types.first().map(String::as_str).unwrap_or(""),
171 0 : ev.entity_id.as_str(),
172 : );
173 0 : let mut headers = async_nats::HeaderMap::new();
174 0 : headers.insert(
175 : "Nats-Msg-Id",
176 0 : format!("{}:{}", ev.tenant.as_str(), ev.seq).as_str(),
177 : );
178 0 : let bytes = serde_json::to_vec(&ev).map_err(err)?;
179 : // double-ack: await the JetStream publish ack, not just the TCP write
180 0 : self.js
181 0 : .publish_with_headers(subject, headers, bytes.into())
182 0 : .await
183 0 : .map_err(err)?
184 0 : .await
185 0 : .map_err(err)?;
186 0 : Ok(())
187 0 : }
188 :
189 : /// Publish a registration CUD delta: the full registration document
190 : /// (or a `{"deleted": id}` tombstone) on the tenant's registry subject.
191 : /// The tenant is re-validated here because it becomes a subject token.
192 0 : pub async fn publish_registry(
193 0 : &self,
194 0 : tenant: &str,
195 0 : delta: &serde_json::Value,
196 0 : ) -> Result<(), BusError> {
197 0 : let tenant = antares_model::TenantId::new_internal(tenant).map_err(err)?;
198 0 : let bytes = serde_json::to_vec(delta).map_err(err)?;
199 0 : self.js
200 0 : .publish(subjects::registry_subject(&tenant), bytes.into())
201 0 : .await
202 0 : .map_err(err)?
203 0 : .await
204 0 : .map_err(err)?;
205 0 : Ok(())
206 0 : }
207 :
208 : /// BALANCED consumption (matcher, temporal recorder): a shared durable —
209 : /// instances with the same `durable` name split the work. Explicit-ack,
210 : /// bounded prefetch; the caller acks AFTER processing.
211 0 : pub async fn consume_balanced(
212 0 : &self,
213 0 : durable: &str,
214 0 : ) -> Result<consumer::PullConsumer, BusError> {
215 0 : let s = self.js.get_stream(CHANGES_STREAM).await.map_err(err)?;
216 0 : s.get_or_create_consumer(
217 0 : durable,
218 0 : consumer::pull::Config {
219 0 : durable_name: Some(durable.into()),
220 0 : ack_policy: consumer::AckPolicy::Explicit,
221 0 : max_ack_pending: MAX_ACK_PENDING,
222 0 : ..Default::default()
223 0 : },
224 0 : )
225 0 : .await
226 0 : .map_err(err)
227 0 : }
228 :
229 : /// BROADCAST consumption (per-instance registry mirror): an ephemeral
230 : /// consumer — every instance sees every delta, and the consumer dies
231 : /// with the instance.
232 0 : pub async fn consume_registry_broadcast(&self) -> Result<consumer::PullConsumer, BusError> {
233 0 : let s = self.js.get_stream(REGISTRY_STREAM).await.map_err(err)?;
234 0 : s.create_consumer(consumer::pull::Config {
235 0 : // no durable name = ephemeral = broadcast
236 0 : durable_name: None,
237 0 : deliver_policy: consumer::DeliverPolicy::New,
238 0 : ack_policy: consumer::AckPolicy::Explicit,
239 0 : max_ack_pending: MAX_ACK_PENDING,
240 0 : ..Default::default()
241 0 : })
242 0 : .await
243 0 : .map_err(err)
244 0 : }
245 :
246 : /// Drop a consumer of the registry stream, the way a server restart or an
247 : /// inactivity gap drops an ephemeral one. The watcher's reopen path is
248 : /// only exercisable if that gap can be produced on demand.
249 0 : pub async fn delete_registry_consumer(&self, name: &str) -> Result<(), BusError> {
250 0 : let s = self.js.get_stream(REGISTRY_STREAM).await.map_err(err)?;
251 0 : s.delete_consumer(name).await.map_err(err)?;
252 0 : Ok(())
253 0 : }
254 :
255 : /// The KV bucket holding the compiled-subscription mirror.
256 0 : pub async fn subs_kv(&self) -> Result<async_nats::jetstream::kv::Store, BusError> {
257 0 : self.js.get_key_value(SUBS_BUCKET).await.map_err(err)
258 0 : }
259 :
260 : /// The Scorpio `$[quarkus.uuid}` lesson: assert at startup that every
261 : /// balanced concern really is a shared durable on the server — a typo'd
262 : /// durable name would silently turn work-sharing into a private queue
263 : /// (or broadcast into load-balancing). Fatal on mismatch.
264 0 : pub async fn assert_topology(&self, balanced_durables: &[&str]) -> Result<(), BusError> {
265 0 : let s = self.js.get_stream(CHANGES_STREAM).await.map_err(err)?;
266 0 : for durable in balanced_durables {
267 0 : let info = s.consumer_info(durable).await.map_err(|e| {
268 0 : BusError(format!(
269 0 : "topology: balanced durable '{durable}' missing on {CHANGES_STREAM}: {e}"
270 0 : ))
271 0 : })?;
272 0 : if info.config.durable_name.as_deref() != Some(*durable) {
273 0 : return Err(BusError(format!(
274 0 : "topology: consumer '{durable}' is not durable — balanced work would \
275 0 : not be shared across instances"
276 0 : )));
277 0 : }
278 : }
279 0 : tracing::info!(
280 : durables = ?balanced_durables,
281 : "bus topology asserted: balanced durables present on {CHANGES_STREAM}"
282 : );
283 0 : Ok(())
284 0 : }
285 : }
286 :
287 : /// Decode one JetStream message into a `ChangeEvent`. `None` = alien bytes —
288 : /// log-and-ack territory for the consumer (redelivering garbage forever is
289 : /// the alternative). Defence in depth: the subject's tenant segment must agree with
290 : /// the event body — consumers re-verify so a subject-mapping bug can never
291 : /// route one tenant's change into another tenant's processing.
292 : /// `ANTARES_NATS_REPLICAS`: absent = 1; present must be a positive integer.
293 18 : fn replicas_from(v: Option<&str>) -> Result<usize, BusError> {
294 18 : match v {
295 2 : None => Ok(1),
296 16 : Some(raw) => raw
297 16 : .trim()
298 16 : .parse::<usize>()
299 16 : .ok()
300 16 : .filter(|n| *n >= 1)
301 16 : .ok_or_else(|| {
302 12 : BusError(format!(
303 12 : "ANTARES_NATS_REPLICAS must be a positive integer, got {raw:?}"
304 12 : ))
305 12 : }),
306 : }
307 18 : }
308 :
309 0 : pub fn decode(msg: &async_nats::jetstream::Message) -> Option<ChangeEvent> {
310 0 : let ev: ChangeEvent = serde_json::from_slice(&msg.payload).ok()?;
311 0 : if !subject_tenant_agrees(msg.subject.as_str(), &ev) {
312 0 : tracing::error!(
313 0 : subject = %msg.subject,
314 0 : tenant = %ev.tenant.as_str(),
315 : "dropping change event: subject tenant segment disagrees with body"
316 : );
317 0 : return None;
318 0 : }
319 0 : Some(ev)
320 0 : }
321 :
322 : /// The tenant-agreement check, unit-testable: `changes.{tenant}.…` must carry the
323 : /// event's own tenant. Non-`changes` subjects pass (registry deltas carry
324 : /// tenant in the body only).
325 18 : pub fn subject_tenant_agrees(subject: &str, ev: &ChangeEvent) -> bool {
326 18 : let mut parts = subject.split('.');
327 18 : if parts.next() != Some("changes") {
328 2 : return true;
329 16 : }
330 16 : parts.next() == Some(ev.tenant.as_str())
331 18 : }
332 :
333 : /// Drive a pull consumer as a message stream. Thin wrapper so wiring code
334 : /// does not import futures/consumer types everywhere.
335 0 : pub async fn messages(
336 0 : consumer: &consumer::PullConsumer,
337 0 : ) -> Result<
338 0 : impl futures_util::Stream<
339 0 : Item = Result<
340 0 : async_nats::jetstream::Message,
341 0 : async_nats::error::Error<consumer::pull::MessagesErrorKind>,
342 0 : >,
343 0 : > + '_,
344 0 : BusError,
345 0 : > {
346 0 : consumer.messages().await.map_err(err)
347 0 : }
348 :
349 : /// One decoded registry delta from the broadcast consumer, acked in place.
350 0 : pub async fn next_delta(
351 0 : stream: &mut (impl futures_util::Stream<
352 0 : Item = Result<
353 0 : async_nats::jetstream::Message,
354 0 : async_nats::error::Error<consumer::pull::MessagesErrorKind>,
355 0 : >,
356 0 : > + Unpin),
357 0 : ) -> Option<serde_json::Value> {
358 : loop {
359 0 : let msg = stream.next().await?.ok()?;
360 0 : let parsed = serde_json::from_slice(&msg.payload).ok();
361 0 : let _ = msg.ack().await;
362 0 : if parsed.is_some() {
363 0 : return parsed;
364 0 : }
365 : }
366 0 : }
367 :
368 : #[cfg(test)]
369 : mod tests {
370 : use super::*;
371 : use crate::ChangeOp;
372 : use antares_model::{EntityId, TenantId};
373 :
374 : /// Config fatality: a typo'd replica count must never silently run a
375 : /// 3-node deployment at replicas=1 — that is a durability downgrade
376 : /// nobody chose. Absent stays 1; whitespace is tolerated; garbage and
377 : /// zero name the key in the error.
378 : #[test]
379 2 : fn replica_count_garbage_is_fatal_not_a_silent_default() {
380 2 : assert_eq!(replicas_from(None).map_err(|e| e.0), Ok(1));
381 2 : assert_eq!(replicas_from(Some("3")).map_err(|e| e.0), Ok(3));
382 2 : assert_eq!(replicas_from(Some(" 3 ")).map_err(|e| e.0), Ok(3));
383 12 : for bad in ["three", "", "0", "-1", "1.5", "3 nodes"] {
384 12 : let err = replicas_from(Some(bad))
385 12 : .map(|_| ())
386 12 : .expect_err("garbage must be refused")
387 : .0;
388 12 : assert!(
389 12 : err.contains("ANTARES_NATS_REPLICAS") && err.contains(bad.trim()),
390 : "the error must name the key and the value: {err}"
391 : );
392 : }
393 2 : }
394 :
395 : #[test]
396 2 : fn subject_tenant_reverification_drops_mismatches() {
397 2 : let ev = ChangeEvent {
398 2 : tenant: TenantId::new("acme").expect("tenant"),
399 2 : entity_id: EntityId::new("urn:x:1").expect("id"),
400 2 : types: vec!["T".into()],
401 2 : op: ChangeOp::Create,
402 2 : changed_attrs: vec![],
403 2 : payload: None,
404 2 : prev_payload: None,
405 2 : version: 1,
406 2 : incarnation: String::new(),
407 2 : seq: 1,
408 2 : payload_ref: None,
409 2 : prev_payload_ref: None,
410 2 : };
411 2 : assert!(subject_tenant_agrees("changes.acme.aa.bb", &ev));
412 2 : assert!(
413 2 : !subject_tenant_agrees("changes.other.aa.bb", &ev),
414 : "a mis-mapped subject must be dropped, not processed"
415 : );
416 2 : assert!(
417 2 : subject_tenant_agrees("registry.other", &ev),
418 : "non-changes subjects carry tenant in the body only"
419 : );
420 2 : }
421 :
422 : /// The check compares whole tokens: a subject whose tenant segment only
423 : /// starts with, contains or is missing the event's tenant must not pass.
424 : #[test]
425 2 : fn tenant_agreement_is_not_fooled_by_partial_tokens() {
426 2 : let ev = ChangeEvent {
427 2 : tenant: TenantId::new("acme").expect("tenant"),
428 2 : entity_id: EntityId::new("urn:x:1").expect("id"),
429 2 : types: vec![],
430 2 : op: ChangeOp::Delete,
431 2 : changed_attrs: vec![],
432 2 : payload: None,
433 2 : prev_payload: None,
434 2 : version: 1,
435 2 : incarnation: String::new(),
436 2 : seq: 1,
437 2 : payload_ref: None,
438 2 : prev_payload_ref: None,
439 2 : };
440 12 : for subject in [
441 2 : "changes.acmeX.aa.bb", // longer token
442 2 : "changes.acm.aa.bb", // prefix of the tenant
443 2 : "changes..aa.bb", // empty tenant segment
444 2 : "changes.aa.acme.bb", // tenant present, wrong position
445 2 : "changes", // no tenant segment at all
446 2 : "changes.",
447 2 : ] {
448 12 : assert!(
449 12 : !subject_tenant_agrees(subject, &ev),
450 : "must not accept {subject:?}"
451 : );
452 : }
453 2 : }
454 : }
|
