Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Bus wiring: roles × bus. The composition root is the ONLY
3 : //! place that knows both the bus variant and which consumers exist.
4 : //!
5 : //! bus=local — single process, all roles: the store's change hook feeds the
6 : //! in-process matcher (`antares_api::wire`), temporal recording stays synchronous
7 : //! in the write path. Exactly v0's behaviour; the ETSI pipeline runs this.
8 : //!
9 : //! bus=nats — the scale-out spine:
10 : //! api role produces: same-tx outbox rows, the
11 : //! outbox drain publishing to `ANTARES_CHANGES` with
12 : //! `Nats-Msg-Id` dedup, subscription CUD → KV,
13 : //! registration CUD → `ANTARES_REGISTRY`, and the
14 : //! per-instance registration mirror its federation path reads.
15 : //! matcher / one shared DURABLE ("matcher"): decode → process_change →
16 : //! notifier ack AFTER processing; the KV-watched subscription mirror;
17 : //! the interval loop (single-winner by row-lock claim).
18 : //! temporal no bus consumer — auto-recording is synchronous in the
19 : //! write path; the role only carries the
20 : //! plain-mode partition job, wired in main.rs.
21 : //!
22 : //! Concurrent drains on N api pods double-publish only within the stream's
23 : //! duplicate window, where `Nats-Msg-Id` = outbox seq absorbs them — that is
24 : //! the design, not an accident (at-least-once, engineered idempotent).
25 :
26 : use antares_api::AppState;
27 : use antares_bus::nats::{self, NatsBus};
28 : use antares_bus::ChangeEvent;
29 : use antares_model::TenantId;
30 : use antares_sql::store::Kind;
31 : use futures_util::StreamExt;
32 : use std::sync::Arc;
33 :
34 : #[derive(Clone, Copy, Debug)]
35 : pub struct Roles {
36 : pub api: bool,
37 : pub matcher: bool,
38 : pub notifier: bool,
39 : pub temporal: bool,
40 : pub registry: bool,
41 : }
42 :
43 : impl Roles {
44 46 : pub fn parse(spec: &str) -> Result<Self, String> {
45 46 : if spec == "all" {
46 16 : return Ok(Self {
47 16 : api: true,
48 16 : matcher: true,
49 16 : notifier: true,
50 16 : temporal: true,
51 16 : registry: true,
52 16 : });
53 30 : }
54 30 : let mut r = Self {
55 30 : api: false,
56 30 : matcher: false,
57 30 : notifier: false,
58 30 : temporal: false,
59 30 : registry: false,
60 30 : };
61 44 : for part in spec.split(',') {
62 44 : match part.trim() {
63 44 : "api" => r.api = true,
64 36 : "matcher" => r.matcher = true,
65 28 : "notifier" => r.notifier = true,
66 22 : "temporal" => r.temporal = true,
67 18 : "registry" => r.registry = true,
68 16 : other => {
69 16 : return Err(format!(
70 16 : "unknown role {other:?} (api|matcher|notifier|temporal|registry|all)"
71 16 : ))
72 : }
73 : }
74 : }
75 14 : Ok(r)
76 46 : }
77 :
78 26 : pub fn all(&self) -> bool {
79 26 : self.api && self.matcher && self.notifier && self.temporal && self.registry
80 26 : }
81 : }
82 :
83 : /// ANTARES_OUTBOX_DRAIN: `on` (the default) or `off`, and nothing else. A
84 : /// typo'd `of` read as "on" under the old permissive parse, quietly defeating
85 : /// the dedicated-drainer split and the crash drill it exists for.
86 69 : pub fn outbox_drain_enabled() -> Result<bool, String> {
87 69 : match std::env::var("ANTARES_OUTBOX_DRAIN") {
88 51 : Err(std::env::VarError::NotPresent) => Ok(true),
89 0 : Err(e) => Err(format!("ANTARES_OUTBOX_DRAIN is unreadable: {e}")),
90 18 : Ok(v) => match v.as_str() {
91 18 : "on" => Ok(true),
92 16 : "off" => Ok(false),
93 14 : other => Err(format!(
94 14 : "ANTARES_OUTBOX_DRAIN must be on|off, got {other:?}"
95 14 : )),
96 : },
97 : }
98 69 : }
99 :
100 : /// KV key for one mirrored subscription: tenant verbatim (token-safe by
101 : /// construction), id hashed (URNs carry `:` — illegal in KV keys). The VALUE
102 : /// carries the real tenant/id, so the key only needs uniqueness. Kind-scoped:
103 : /// a Subscription and a Context Source Registration Subscription may carry the
104 : /// same client-chosen id in one tenant (5.5.10), and sharing a key would let
105 : /// one overwrite the other's mirror entry.
106 16 : fn kv_key(tenant: &str, kind: Kind, id: &str) -> String {
107 16 : format!(
108 : "{tenant}.{}{:016x}",
109 16 : if kind == Kind::CSourceSubscription {
110 2 : "c"
111 : } else {
112 14 : ""
113 : },
114 16 : antares_bus::subjects::fnv1a64(id.as_bytes())
115 : )
116 16 : }
117 :
118 : /// Wire everything bus=nats needs onto the state. Async: connects, hydrates
119 : /// mirrors, creates consumers, asserts topology — all before the broker
120 : /// starts accepting traffic, so a mis-shapen topology is a startup failure,
121 : /// never a silent runtime drift.
122 0 : pub async fn wire_nats(
123 0 : state: &mut AppState,
124 0 : url: &str,
125 0 : roles: Roles,
126 0 : ) -> Result<(), Box<dyn std::error::Error>> {
127 0 : let bus = Arc::new(NatsBus::connect(url).await?);
128 : // /q/health `bus` member: live connection state + reconnect count.
129 : // Installed HERE so bus=local never carries the member at all.
130 : {
131 0 : let b = bus.clone();
132 0 : state.bus_stats = Some(Arc::new(move || {
133 0 : serde_json::json!({
134 0 : "mode": "nats",
135 0 : "connected": b.connected(),
136 0 : "reconnects": b.reconnects(),
137 : })
138 0 : }));
139 : }
140 : // Multi-process mode: interval firings need the single-winner claim
141 : // — keyed off this flag, not off mirror presence (local mode wires a
142 : // mirror too).
143 0 : state.nats = true;
144 : // Entity writes now enqueue their events in the write transaction.
145 0 : state.store.set_outbox(true);
146 : // Auto-recording stays SYNCHRONOUS in the write path in every bus mode:
147 : // every write goes through an api-role pod that has the
148 : // shared store, so recording in-request gives read-your-writes — the
149 : // ETSI suite asserts history immediately after a write — and kills the
150 : // late-replay resurrection race (a consumer re-applying a pre-delete
151 : // event AFTER a direct temporal delete). The recorder consumer this
152 : // replaced double-applied by design; it bought nothing but the races.
153 :
154 : // The drain nudge: a same-process write pokes its own drain, so publish
155 : // latency is ~1 ms, not the idle-poll interval. Cross-pod writes are
156 : // still covered by each pod's own nudge; the poll below stays as the
157 : // crash-recovery fallback.
158 0 : let nudge = Arc::new(tokio::sync::Notify::new());
159 : {
160 0 : let n = nudge.clone();
161 : // Auto-recording rides the same synchronous hook here as in bus=local
162 : // (one choke point for every write, no handler can forget), then nudges
163 : // the outbox drain.
164 0 : let st_rec = state.clone();
165 0 : state.store.set_change_hook(Arc::new(
166 : move |tenant: &TenantId,
167 : before: Option<serde_json::Value>,
168 : after: Option<serde_json::Value>|
169 0 : -> antares_store::HookFuture<'_> {
170 0 : let st_rec = st_rec.clone();
171 0 : let n = n.clone();
172 0 : Box::pin(async move {
173 0 : antares_api::notify::record_temporal_change(
174 0 : &st_rec,
175 0 : tenant,
176 0 : before.as_ref(),
177 0 : after.as_ref(),
178 0 : )
179 0 : .await;
180 0 : n.notify_one();
181 0 : })
182 0 : },
183 : ));
184 : }
185 :
186 0 : let mut durables: Vec<&'static str> = Vec::new();
187 :
188 0 : if roles.api {
189 : // Subscription write side: subscription CUD → KV (tombstone = null doc).
190 0 : let kv = bus.subs_kv().await?;
191 0 : let kv_for_hook = kv.clone();
192 0 : state.sub_sync = Some(Arc::new(
193 0 : move |tenant: &TenantId, kind: Kind, id: &str, doc| {
194 0 : let kv = kv_for_hook.clone();
195 0 : let key = kv_key(tenant.as_str(), kind, id);
196 0 : let value = serde_json::json!({
197 0 : "tenant": tenant.as_str(), "id": id, "doc": doc,
198 0 : "csub": kind == Kind::CSourceSubscription,
199 : });
200 0 : tokio::spawn(async move {
201 0 : let bytes = serde_json::to_vec(&value).unwrap_or_default();
202 : // The store row is already committed. A lost put leaves this
203 : // subscription invisible to every matcher pod — silently, and
204 : // until the next restart, because mirrors hydrate from the
205 : // store only at process start. So retry rather than warn once.
206 : // Named ceiling: after the last attempt the divergence stands
207 : // until a restart; closing that needs periodic reconciliation.
208 0 : for attempt in 0..MIRROR_SYNC_ATTEMPTS {
209 0 : match kv.put(key.clone(), bytes.clone().into()).await {
210 0 : Ok(_) => return,
211 0 : Err(e) => {
212 0 : tracing::warn!("sub KV sync attempt {} failed: {e}", attempt + 1);
213 0 : tokio::time::sleep(mirror_sync_backoff(attempt)).await;
214 : }
215 : }
216 : }
217 0 : tracing::error!(
218 : "sub KV sync gave up for {key} — this subscription is not mirrored \
219 : until the next restart"
220 : );
221 0 : });
222 0 : },
223 : ));
224 :
225 : // Registration write side: registration CUD → ANTARES_REGISTRY delta.
226 0 : let bus_for_reg = bus.clone();
227 0 : state.reg_sync = Some(Arc::new(move |tenant: &TenantId, id: &str, doc| {
228 0 : let bus = bus_for_reg.clone();
229 0 : let delta = serde_json::json!({
230 0 : "tenant": tenant.as_str(), "id": id, "doc": doc,
231 : });
232 0 : let tenant = tenant.as_str().to_owned();
233 0 : let id = id.to_owned();
234 0 : tokio::spawn(async move {
235 : // Same contract as the subscription mirror: the row is
236 : // committed, so a lost delta makes the registration invisible
237 : // to every federation path until a restart re-hydrates.
238 0 : for attempt in 0..MIRROR_SYNC_ATTEMPTS {
239 0 : match bus.publish_registry(&tenant, &delta).await {
240 0 : Ok(()) => return,
241 0 : Err(e) => {
242 0 : tracing::warn!(
243 : "registry delta publish attempt {} failed: {e}",
244 0 : attempt + 1
245 : );
246 0 : tokio::time::sleep(mirror_sync_backoff(attempt)).await;
247 : }
248 : }
249 : }
250 0 : tracing::error!(
251 : "registry delta publish gave up for {id} — this registration is not \
252 : mirrored until the next restart"
253 : );
254 0 : });
255 0 : }));
256 :
257 : // 5.2.34 write side: a cooldown stamp is broadcast to the other api
258 : // pods on the registry stream (seconds-scale state, deliberately not
259 : // persisted) — per-process stamps re-dial a failed source from every
260 : // pod behind the LB.
261 0 : let bus_for_cool = bus.clone();
262 0 : state.reg_fail_sync = Some(Arc::new(move |reg_id: &str, ok: bool| {
263 0 : let bus = bus_for_cool.clone();
264 0 : let delta = serde_json::json!({"cooldownReg": reg_id, "ok": ok});
265 0 : tokio::spawn(async move {
266 0 : if let Err(e) = bus.publish_registry("cooldown", &delta).await {
267 0 : tracing::warn!("cooldown stamp publish failed: {e}");
268 0 : }
269 0 : });
270 0 : }));
271 :
272 : // Registration read side: the ONE compiled registration mirror this instance's
273 : // federation path reads. Consumer created BEFORE the hydrate so no
274 : // delta can fall between them; last-writer-wins per key converges.
275 0 : let reg_mirror = Arc::new(antares_api::mirror::DocMirror::default());
276 0 : let reg_consumer = bus.consume_registry_broadcast().await?;
277 : // Installed only if it is whole. A mirror that is present and SHORT
278 : // is read as the truth — `reg_docs` asks it and never the store — so
279 : // half a hydrate silently drops Context Sources for the life of the
280 : // process. Left uninstalled, federation matching falls back to the
281 : // store's own indexed narrowing: correct, and merely slower.
282 0 : match antares_api::notify::seed_mirror(
283 0 : &*state.store,
284 0 : reg_mirror.as_ref(),
285 0 : Kind::Registration,
286 : )
287 0 : .await
288 : {
289 0 : Ok(()) => state.reg_mirror = Some(reg_mirror.clone()),
290 0 : Err(e) => tracing::error!(
291 : "registration mirror hydrate failed ({e}); \
292 : federation matching falls back to a store read per request"
293 : ),
294 : }
295 0 : let egress_for_cool = state.egress.clone();
296 0 : let store_for_reg = state.store.clone();
297 0 : tokio::spawn(async move {
298 : // The consumer is ephemeral, so a NATS restart or an inactivity
299 : // gap deletes it server-side and the next pull errors. Ending the
300 : // task there froze this pod's registration mirror — and with it
301 : // all federation matching — for the process lifetime, while
302 : // /q/health still reported the bus connected. Re-open instead,
303 : // and re-hydrate from the store because a fresh consumer starts
304 : // at NEW and never replays what the gap dropped.
305 : loop {
306 0 : let mut msgs = match reg_consumer.messages().await {
307 0 : Ok(m) => m,
308 0 : Err(e) => {
309 0 : tracing::warn!("registry broadcast consumer stream failed: {e}");
310 0 : tokio::time::sleep(std::time::Duration::from_secs(1)).await;
311 0 : continue;
312 : }
313 : };
314 0 : while let Some(delta) = nats::next_delta(&mut msgs).await {
315 : // 5.2.34 read side: a cooldown stamp updates this pod's
316 : // map (a marker delta has no tenant/id — apply_delta
317 : // ignores it on pods that predate the member).
318 0 : if let Some(rid) = delta.get("cooldownReg").and_then(serde_json::Value::as_str)
319 : {
320 0 : egress_for_cool.reg_record(rid, delta["ok"].as_bool().unwrap_or(false));
321 0 : continue;
322 0 : }
323 0 : apply_delta(reg_mirror.as_ref(), &delta);
324 : }
325 0 : tracing::warn!("registry broadcast consumer stream ended — reopening");
326 0 : tokio::time::sleep(std::time::Duration::from_secs(1)).await;
327 : // Already installed, so this one cannot be withheld. A failed
328 : // re-hydrate leaves the mirror holding what it had before the
329 : // gap, which the federation path will serve as current: say
330 : // so at error level rather than let it pass as a warning.
331 0 : if let Err(e) = antares_api::notify::seed_mirror(
332 0 : &*store_for_reg,
333 0 : reg_mirror.as_ref(),
334 0 : Kind::Registration,
335 : )
336 0 : .await
337 : {
338 0 : tracing::error!(
339 : "registration mirror re-hydrate failed ({e}); \
340 : this pod is matching against registrations from before the gap"
341 : );
342 0 : }
343 : }
344 : });
345 :
346 : // The outbox drain. Runs on every api pod; concurrent drains are
347 : // absorbed by Nats-Msg-Id dedup within the duplicate window.
348 : // ANTARES_OUTBOX_DRAIN=off leaves the rows for another pod's drain —
349 : // the crash-drill lever and the dedicated-drainer split.
350 0 : let drain_on = outbox_drain_enabled()?;
351 0 : if !drain_on {
352 0 : tracing::warn!("outbox drain OFF on this pod (ANTARES_OUTBOX_DRAIN=off)");
353 0 : }
354 0 : if drain_on {
355 0 : let store = state.store.clone();
356 0 : let bus_for_drain = bus.clone();
357 0 : tokio::spawn(async move {
358 : loop {
359 0 : let rows = match store.outbox_peek(64).await {
360 0 : Ok(r) => r,
361 0 : Err(e) => {
362 0 : tracing::warn!("outbox peek failed: {e}");
363 0 : tokio::time::sleep(std::time::Duration::from_secs(1)).await;
364 0 : continue;
365 : }
366 : };
367 0 : if rows.is_empty() {
368 : // woken by the same-process write hook, or the fallback
369 : // poll for rows another pod failed to publish
370 0 : let _ = tokio::time::timeout(
371 0 : std::time::Duration::from_millis(250),
372 0 : nudge.notified(),
373 : )
374 0 : .await;
375 0 : continue;
376 0 : }
377 : // Ack the EXACT published seqs — a blanket
378 : // up-to-max delete loses a lower-seq row whose
379 : // transaction commits between peek and ack.
380 0 : let mut acked: Vec<i64> = Vec::new();
381 : // Rows whose bodies were too big for the bus: the message
382 : // carries a reference, and this row is the only copy of
383 : // what it references. Kept, not deleted, and taken out of
384 : // the next page by the same stamp.
385 0 : let mut retained: Vec<(TenantId, i64)> = Vec::new();
386 0 : for (seq, _tenant, event) in rows {
387 0 : match serde_json::from_value::<ChangeEvent>(event) {
388 0 : Ok(mut ev) => {
389 0 : ev.seq = seq;
390 0 : let checked = ev.claim_checked_at(antares_bus::CLAIM_CHECK_BYTES);
391 0 : match bus_for_drain.publish(&ev).await {
392 0 : Ok(()) if checked => retained.push((ev.tenant, seq)),
393 0 : Ok(()) => acked.push(seq),
394 0 : Err(e) => {
395 0 : tracing::warn!("outbox publish of seq {seq} failed: {e}");
396 0 : break; // retry from here next round
397 : }
398 : }
399 : }
400 0 : Err(e) => {
401 : // an undecodable row would wedge the drain forever
402 0 : tracing::error!("outbox row {seq} undecodable ({e}) — skipped");
403 0 : acked.push(seq);
404 : }
405 : }
406 : }
407 : // Retain BEFORE ack: both statements are separate
408 : // transactions, and a crash between them must leave a
409 : // claim-check row alive rather than published and gone.
410 : // One statement per row, under that row's tenant — the
411 : // outbox UPDATE takes no service escape (0005), and an
412 : // event over the bus ceiling is rare enough that grouping
413 : // the page by tenant would cost more code than statements.
414 0 : for (tenant, seq) in &retained {
415 0 : if let Err(e) = store.outbox_retain(tenant, &[*seq]).await {
416 0 : tracing::warn!("outbox retain of seq {seq} failed: {e}");
417 0 : }
418 : }
419 0 : if !acked.is_empty() {
420 0 : if let Err(e) = store.outbox_ack(&acked).await {
421 0 : tracing::warn!("outbox ack {acked:?} failed: {e}");
422 0 : }
423 0 : }
424 : }
425 : });
426 0 : }
427 0 : }
428 :
429 0 : if roles.matcher || roles.notifier {
430 : // Subscription read side: consumer-before-hydrate, same convergence argument.
431 0 : let sub_mirror = Arc::new(antares_api::mirror::SubMirror::default());
432 0 : let kv = bus.subs_kv().await?;
433 0 : let watch = kv.watch_all().await?;
434 : // Same rule as the registration mirror, and the same one `bus=local`
435 : // applies in `antares_api::wire`: a subscription absent from an installed
436 : // mirror never fires again, because the matcher reads candidates
437 : // from the mirror alone.
438 0 : match antares_api::notify::seed_mirror(
439 0 : &*state.store,
440 0 : sub_mirror.as_ref(),
441 0 : Kind::Subscription,
442 : )
443 0 : .await
444 : {
445 0 : Ok(()) => state.sub_mirror = Some(sub_mirror.clone()),
446 0 : Err(e) => tracing::error!(
447 : "subscription mirror hydrate failed ({e}); \
448 : matching falls back to a store scan per change"
449 : ),
450 : }
451 0 : tokio::spawn(async move {
452 : // Same restart contract as the registry consumer: the watch ends
453 : // on a NATS restart, and a task that returns there stops seeing
454 : // every subscription change — i.e. this pod silently stops
455 : // notifying — for the process lifetime. `watch_all` replays the
456 : // bucket's current values, so re-opening also re-converges the
457 : // mirror over the gap.
458 0 : let mut watch = watch;
459 : loop {
460 0 : while let Some(entry) = watch.next().await {
461 0 : let Ok(entry) = entry else { continue };
462 0 : if let Ok(v) = serde_json::from_slice::<serde_json::Value>(&entry.value) {
463 0 : apply_delta(sub_mirror.as_ref(), &v);
464 0 : }
465 : }
466 0 : tracing::warn!("subscription KV watch ended — reopening");
467 0 : tokio::time::sleep(std::time::Duration::from_secs(1)).await;
468 0 : match kv.watch_all().await {
469 0 : Ok(w) => watch = w,
470 0 : Err(e) => tracing::warn!("subscription KV watch reopen failed: {e}"),
471 : }
472 : }
473 : });
474 :
475 : // The balanced matcher durable: decode → process_change → ack AFTER.
476 0 : durables.push("matcher");
477 0 : let consumer = bus.consume_balanced("matcher").await?;
478 0 : let st = state.clone();
479 0 : tokio::spawn(async move {
480 : loop {
481 0 : let mut msgs = match consumer.messages().await {
482 0 : Ok(m) => m,
483 0 : Err(e) => {
484 0 : tracing::warn!("matcher consumer stream failed: {e}");
485 0 : tokio::time::sleep(std::time::Duration::from_secs(1)).await;
486 0 : continue;
487 : }
488 : };
489 0 : while let Some(Ok(msg)) = msgs.next().await {
490 : // Change lag = stream-publish → matcher-processing
491 : // age, from the JetStream metadata timestamp.
492 0 : if let Ok(info) = msg.info() {
493 : // OffsetDateTime → SystemTime (impl in `time`/std),
494 : // so no direct `time` dependency here.
495 0 : let published: std::time::SystemTime = info.published.into();
496 0 : if let Ok(age) = published.elapsed() {
497 0 : metrics::histogram!("antares_change_lag_seconds")
498 0 : .record(age.as_secs_f64());
499 0 : }
500 0 : }
501 0 : if let Some(ev) = nats::decode(&msg) {
502 0 : let (before, after) = resolve_payloads(&st, &ev).await;
503 0 : antares_api::notify::process_change(&st, ev.tenant.as_str(), before, after)
504 0 : .await;
505 0 : }
506 0 : let _ = msg.ack().await;
507 : }
508 : }
509 : });
510 :
511 : // Interval subscriptions: every matcher pod ticks; the row-lock claim
512 : // in interval_tick makes each firing single-winner.
513 0 : let st = state.clone();
514 0 : tokio::spawn(async move {
515 0 : let mut tick = tokio::time::interval(std::time::Duration::from_millis(500));
516 : loop {
517 0 : tick.tick().await;
518 0 : antares_api::notify::interval_tick(&st).await;
519 : }
520 : });
521 0 : }
522 :
523 : // The temporal role carries no bus consumer: auto-recording is
524 : // synchronous in the write path (see above), and plain-mode partition
525 : // maintenance runs from main.rs regardless of bus mode.
526 :
527 : // The server must agree these are shared durables.
528 0 : bus.assert_topology(&durables).await?;
529 0 : tracing::info!(?roles, "bus=nats wired");
530 0 : Ok(())
531 0 : }
532 :
533 : /// Hydrate a mirror from the system of record (Postgres) at startup.
534 : /// Attempts a mirror write gets before the divergence is logged as an error.
535 : const MIRROR_SYNC_ATTEMPTS: u32 = 5;
536 :
537 : /// Exponential backoff between mirror-sync attempts: 0.2 s doubling to 3.2 s,
538 : /// so the whole ladder outlives a bus reconnect without holding a task for
539 : /// minutes.
540 0 : fn mirror_sync_backoff(attempt: u32) -> std::time::Duration {
541 0 : std::time::Duration::from_millis(200u64 << attempt.min(4))
542 0 : }
543 :
544 : /// Apply one `{tenant, id, doc|null, csub}` delta to a mirror. A Context
545 : /// Source Registration Subscription carries no document into the mirror: it
546 : /// is matched against registrations rather than entities, so the delta only
547 : /// wakes the interval sweep (5.11.7).
548 34 : fn apply_delta(mirror: &dyn antares_api::mirror::Mirror, delta: &serde_json::Value) {
549 34 : if delta.get("csub").and_then(serde_json::Value::as_bool) == Some(true) {
550 2 : mirror.csub_written();
551 2 : return;
552 32 : }
553 12 : let (Some(tenant), Some(id)) = (
554 32 : delta.get("tenant").and_then(serde_json::Value::as_str),
555 32 : delta.get("id").and_then(serde_json::Value::as_str),
556 : ) else {
557 20 : return;
558 : };
559 : // Hydration validates the tenant before it touches the mirror; the delta
560 : // path must agree. A name outside the grammar can only add entries that no
561 : // lookup will ever hit, so an unvalidated one is unbounded growth keyed by
562 : // whatever reached the bus. The grammar is the whole check here: what
563 : // arrives is a tenant a peer broker wrote, and a write inside a Snapshot
564 : // (5.5.15) carries the synthetic tenant a client may not name — refusing
565 : // that one would drop the mirror delta the snapshot-scoped subscription
566 : // matches on.
567 12 : if TenantId::new_internal(tenant).is_err() {
568 8 : return;
569 4 : }
570 4 : let doc = delta.get("doc").filter(|d| !d.is_null()).cloned();
571 4 : mirror.apply(tenant, id, doc);
572 34 : }
573 :
574 : /// Resolve claim-check references: bodies the bus could not carry come back
575 : /// from the outbox row the drain kept, read by the event's own `seq`.
576 : ///
577 : /// The store's current row is NOT that source. It answers with the entity as
578 : /// it stands now, which is the after-image: resolving `prev_payload_ref` from
579 : /// it hands the matcher two copies of the same document, `diff` finds nothing
580 : /// changed and the change reaches no subscriber. It stays the fallback for
581 : /// `payload_ref` alone, where being newer than the referenced version is the
582 : /// ordinary at-least-once reality the matcher already tolerates.
583 12 : async fn resolve_payloads(
584 12 : st: &AppState,
585 12 : ev: &ChangeEvent,
586 12 : ) -> (Option<serde_json::Value>, Option<serde_json::Value>) {
587 12 : if ev.payload_ref.is_none() && ev.prev_payload_ref.is_none() {
588 4 : return (ev.prev_payload.clone(), ev.payload.clone());
589 8 : }
590 : // 0 is the local bus, which never claim-checks: it hands the payloads to
591 : // the matcher in process.
592 8 : let kept = match ev.seq {
593 6 : 0 => None,
594 2 : seq => st
595 2 : .store
596 2 : .outbox_event(seq, &ev.tenant)
597 2 : .await
598 2 : .ok()
599 2 : .flatten()
600 2 : .and_then(|v| serde_json::from_value::<ChangeEvent>(v).ok()),
601 : };
602 8 : if let Some(kept) = kept {
603 2 : return (kept.prev_payload, kept.payload);
604 6 : }
605 : // Past the retention window, or a deployment whose store keeps no outbox.
606 : // The after-image is still recoverable; the before-image is not, and a
607 : // guess in its place is a notification that reports a change that did not
608 : // happen.
609 6 : let after = match ev.payload.clone() {
610 0 : Some(v) => Some(v),
611 6 : None => match ev.payload_ref.as_ref() {
612 6 : Some(r) => st
613 6 : .store
614 6 : .get(&ev.tenant, Kind::Entity, r.entity_id.as_str())
615 6 : .await
616 6 : .ok()
617 6 : .flatten(),
618 0 : None => None,
619 : },
620 : };
621 6 : if ev.prev_payload_ref.is_some() {
622 0 : metrics::counter!("antares_claim_check_unresolved_total").increment(1);
623 0 : tracing::warn!(
624 : "claim-check row for seq {} is gone: the change to {} notifies nobody",
625 : ev.seq,
626 0 : ev.entity_id.as_str()
627 : );
628 6 : }
629 6 : (ev.prev_payload.clone(), after)
630 12 : }
631 :
632 : #[cfg(test)]
633 : mod tests {
634 : use super::*;
635 : use antares_api::mirror::DocMirror;
636 : use antares_bus::{ChangeOp, PayloadRef};
637 : use antares_model::EntityId;
638 : use serde_json::json;
639 :
640 : #[test]
641 2 : fn roles_parse_accepts_the_role_set_and_refuses_anything_else() {
642 2 : assert!(Roles::parse("all").expect("all").all());
643 2 : let r = Roles::parse("api").expect("api");
644 2 : assert!(r.api && !r.matcher && !r.notifier && !r.temporal && !r.registry);
645 2 : assert!(
646 2 : !r.all(),
647 : "a single role must never claim to be the full set"
648 : );
649 2 : let r = Roles::parse(" matcher , notifier ").expect("padded list");
650 2 : assert!(r.matcher && r.notifier && !r.api);
651 : // The enumerated full set is the same thing as "all" — a role split
652 : // that happens to name every role must still pass the bus=local gate.
653 2 : assert!(Roles::parse("api,matcher,notifier,temporal,registry")
654 2 : .expect("full list")
655 2 : .all());
656 :
657 14 : for bad in ["", "api,", "ALL", "apis", "api;matcher", "worker", " "] {
658 14 : let err = Roles::parse(bad).expect_err(&format!("ANTARES_ROLES={bad:?} must be fatal"));
659 14 : assert!(err.starts_with("unknown role"), "{bad:?}: {err}");
660 : }
661 2 : }
662 :
663 : /// The KV key must be legal for a NATS KV bucket (`:` from a URN is not),
664 : /// stable across calls, and collision-free per id, per tenant and per
665 : /// kind — 5.5.10 leaves the id to the client, so one URN can name both a
666 : /// Subscription and a Context Source Registration Subscription.
667 : #[test]
668 2 : fn kv_key_is_bucket_legal_and_stable() {
669 2 : let sub = Kind::Subscription;
670 2 : let k = kv_key("default", sub, "urn:ngsi-ld:Subscription:1");
671 2 : assert_eq!(
672 : k,
673 2 : kv_key("default", sub, "urn:ngsi-ld:Subscription:1"),
674 : "stable"
675 : );
676 2 : assert!(
677 2 : k.bytes()
678 48 : .all(|b| b.is_ascii_alphanumeric() || b == b'.' || b == b'_' || b == b'-'),
679 : "illegal KV key character in {k:?}"
680 : );
681 2 : assert!(k.starts_with("default."), "tenant scoping lost: {k}");
682 2 : assert_ne!(
683 2 : kv_key("default", sub, "urn:ngsi-ld:Subscription:1"),
684 2 : kv_key("default", sub, "urn:ngsi-ld:Subscription:2")
685 : );
686 2 : assert_ne!(
687 2 : kv_key("t1", sub, "urn:ngsi-ld:Subscription:1"),
688 2 : kv_key("t2", sub, "urn:ngsi-ld:Subscription:1")
689 : );
690 2 : assert_ne!(
691 2 : kv_key("default", sub, "urn:ngsi-ld:Subscription:1"),
692 2 : kv_key(
693 2 : "default",
694 2 : Kind::CSourceSubscription,
695 2 : "urn:ngsi-ld:Subscription:1"
696 : ),
697 : "one id naming both kinds must not share a mirror entry"
698 : );
699 2 : }
700 :
701 : /// A Context Source Registration Subscription delta carries no document:
702 : /// it wakes the interval sweep and touches nothing the matcher reads.
703 : #[test]
704 2 : fn a_csub_delta_only_wakes_the_sweep() {
705 2 : let m = antares_api::mirror::SubMirror::default();
706 2 : apply_delta(
707 2 : &m,
708 2 : &serde_json::json!({
709 2 : "tenant": "default", "id": "urn:ngsi-ld:CSourceSubscription:1",
710 2 : "doc": {"id": "urn:ngsi-ld:CSourceSubscription:1", "timeInterval": 5},
711 2 : "csub": true,
712 2 : }),
713 : );
714 2 : assert!(
715 2 : m.docs("default").is_empty(),
716 : "a csource subscription must not enter the candidate index"
717 : );
718 2 : }
719 :
720 : /// The KV/registry mirrors are fed from the bus. A malformed or hostile
721 : /// delta must be dropped — never panic a consumer task, and never grow
722 : /// the mirror under a key no request can ever address (the tenant is
723 : /// validated on every request, so an unvalidated one is pure ballast).
724 : #[test]
725 2 : fn apply_delta_ignores_malformed_and_hostile_deltas() {
726 2 : let m = DocMirror::default();
727 2 : apply_delta(
728 2 : &m,
729 2 : &json!({"tenant": "default", "id": "urn:x:1", "doc": {"id": "urn:x:1"}}),
730 : );
731 2 : assert_eq!(m.docs("default").len(), 1, "a good delta must apply");
732 :
733 : // tombstone
734 2 : apply_delta(
735 2 : &m,
736 2 : &json!({"tenant": "default", "id": "urn:x:1", "doc": null}),
737 : );
738 2 : assert!(m.docs("default").is_empty(), "null doc must delete");
739 2 : assert!(m.tenants().is_empty(), "an emptied tenant must not linger");
740 :
741 : // Shapes that must be ignored without panicking.
742 20 : for junk in [
743 2 : json!({}),
744 2 : json!(null),
745 2 : json!(42),
746 2 : json!("scalar"),
747 2 : json!([1, 2, 3]),
748 2 : json!({"tenant": "default"}),
749 2 : json!({"id": "urn:x:1"}),
750 2 : json!({"tenant": 7, "id": "urn:x:1", "doc": {}}),
751 2 : json!({"tenant": "default", "id": null, "doc": {}}),
752 2 : json!({"cooldownReg": "urn:reg:1", "ok": false}),
753 20 : ] {
754 20 : apply_delta(&m, &junk);
755 20 : }
756 2 : assert!(m.tenants().is_empty(), "junk deltas grew the mirror");
757 :
758 : // Hostile tenants: not addressable by any request (the header is
759 : // validated to [A-Za-z0-9_-]{1,64}), so they may not take memory.
760 8 : for hostile in [
761 2 : "a".repeat(4096),
762 2 : "../../etc".into(),
763 2 : "a b".into(),
764 2 : "".into(),
765 8 : ] {
766 8 : apply_delta(
767 8 : &m,
768 8 : &json!({"tenant": hostile, "id": "urn:x:1", "doc": {"id": "urn:x:1"}}),
769 8 : );
770 8 : }
771 2 : assert!(
772 2 : m.tenants().is_empty(),
773 : "an unaddressable tenant grew the mirror without bound: {:?}",
774 0 : m.tenants()
775 : );
776 2 : }
777 :
778 6 : async fn state_with(entity: Option<serde_json::Value>) -> AppState {
779 6 : let st = AppState::new("antares".into());
780 6 : if let Some(doc) = entity {
781 4 : let id = doc["id"].as_str().expect("id").to_owned();
782 4 : st.store
783 4 : .create(&TenantId::default(), Kind::Entity, &id, doc)
784 4 : .await
785 4 : .expect("seed");
786 2 : }
787 6 : st
788 6 : }
789 :
790 10 : fn event(payload: Option<serde_json::Value>, r#ref: Option<PayloadRef>) -> ChangeEvent {
791 10 : ChangeEvent {
792 10 : tenant: TenantId::default(),
793 10 : entity_id: EntityId::new("urn:ngsi-ld:T:1").expect("id"),
794 10 : types: vec!["T".into()],
795 10 : op: ChangeOp::Update,
796 10 : changed_attrs: vec![],
797 10 : payload,
798 10 : prev_payload: None,
799 10 : version: 1,
800 10 : incarnation: String::new(),
801 10 : seq: 0,
802 10 : payload_ref: r#ref,
803 10 : prev_payload_ref: None,
804 10 : }
805 10 : }
806 :
807 : /// Claim-check resolution: inline wins, a reference is fetched, and a
808 : /// reference to a row that is gone resolves to None instead of panicking
809 : /// the matcher task.
810 : #[tokio::test]
811 2 : async fn resolve_payloads_prefers_inline_and_tolerates_a_dangling_reference() {
812 2 : let doc = json!({"id": "urn:ngsi-ld:T:1", "type": "T"});
813 2 : let st = state_with(Some(doc.clone())).await;
814 :
815 2 : let (before, after) =
816 2 : resolve_payloads(&st, &event(Some(json!({"inline": true})), None)).await;
817 2 : assert_eq!(after, Some(json!({"inline": true})), "inline payload wins");
818 2 : assert_eq!(before, None);
819 :
820 2 : let r = PayloadRef {
821 2 : entity_id: EntityId::new("urn:ngsi-ld:T:1").expect("id"),
822 2 : version: 1,
823 2 : };
824 2 : let (_, after) = resolve_payloads(&st, &event(None, Some(r.clone()))).await;
825 2 : assert_eq!(
826 2 : after.as_ref().and_then(|a| a["id"].as_str()),
827 : Some("urn:ngsi-ld:T:1"),
828 : "a claim-check reference must be fetched from the store"
829 : );
830 :
831 : // The row was deleted between publish and consumption.
832 2 : let gone = state_with(None).await;
833 2 : let (before, after) = resolve_payloads(&gone, &event(None, Some(r))).await;
834 2 : assert_eq!(after, None, "a dangling reference must resolve to None");
835 2 : assert_eq!(before, None);
836 :
837 2 : let (before, after) = resolve_payloads(&st, &event(None, None)).await;
838 2 : assert!(before.is_none() && after.is_none(), "no payload, no fetch");
839 2 : }
840 :
841 : /// A claim-check fetch is scoped to the EVENT's tenant: a reference must
842 : /// never resolve against another tenant's row of the same id.
843 : #[tokio::test]
844 2 : async fn resolve_payloads_never_crosses_a_tenant_boundary() {
845 2 : let st = state_with(Some(json!({"id": "urn:ngsi-ld:T:1", "type": "T"}))).await;
846 2 : let mut ev = event(
847 2 : None,
848 2 : Some(PayloadRef {
849 2 : entity_id: EntityId::new("urn:ngsi-ld:T:1").expect("id"),
850 2 : version: 1,
851 2 : }),
852 : );
853 2 : ev.tenant = TenantId::new("other").expect("tenant");
854 2 : let (_, after) = resolve_payloads(&st, &ev).await;
855 2 : assert_eq!(
856 2 : after, None,
857 2 : "a reference resolved another tenant's entity: {after:?}"
858 2 : );
859 2 : }
860 :
861 : /// A change whose bodies were both too big for the bus reaches the
862 : /// matcher with the before-image the write actually replaced. Resolving
863 : /// the reference against the store's current row instead hands back the
864 : /// after-image twice, `diff` finds nothing and the change notifies
865 : /// nobody. Skips without ANTARES_TEST_DATABASE_URL — the outbox is a
866 : /// Postgres table, and the memory arm has none.
867 : #[tokio::test(flavor = "multi_thread")]
868 2 : async fn a_retained_row_gives_the_matcher_the_before_image_the_write_replaced() {
869 2 : let Ok(url) = std::env::var("ANTARES_TEST_DATABASE_URL") else {
870 0 : eprintln!("SKIP: ANTARES_TEST_DATABASE_URL not set");
871 0 : return;
872 : };
873 2 : let pool = antares_sql::store::pg::connect(&url, 5)
874 2 : .await
875 2 : .expect("connect");
876 2 : let tenant = TenantId::new("claimcheck").expect("tenant");
877 2 : antares_sql::store::pg::ensure_tenant(&pool, &tenant)
878 2 : .await
879 2 : .expect("tenant row");
880 :
881 2 : let id = "urn:ngsi-ld:T:oversized";
882 4 : let wide = |v: &str| {
883 4 : json!({"id": id, "type": ["T"],
884 4 : "https://uri.etsi.org/ngsi-ld/default-context/note":
885 4 : [{"type": "Property", "value": format!("{v}{}", "x".repeat(300 * 1024))}]})
886 4 : };
887 2 : let before_doc = wide("before-");
888 2 : let after_doc = wide("after-");
889 :
890 2 : let mut ev = ChangeEvent {
891 2 : tenant: tenant.clone(),
892 2 : entity_id: EntityId::new(id).expect("id"),
893 2 : types: vec!["T".into()],
894 2 : op: ChangeOp::Update,
895 2 : changed_attrs: vec![],
896 2 : payload: Some(after_doc.clone()),
897 2 : prev_payload: Some(before_doc.clone()),
898 2 : version: 2,
899 2 : incarnation: String::new(),
900 2 : seq: 0,
901 2 : payload_ref: None,
902 2 : prev_payload_ref: None,
903 2 : };
904 2 : let mut tx = pool.begin().await.expect("tx");
905 2 : antares_sql::store::pg::set_tenant(&mut tx, &tenant)
906 2 : .await
907 2 : .expect("set tenant");
908 2 : let seq = antares_sql::store::pg::outbox::enqueue(
909 2 : &mut tx,
910 2 : &tenant,
911 2 : &serde_json::to_value(&ev).expect("event json"),
912 2 : )
913 2 : .await
914 2 : .expect("enqueue");
915 2 : tx.commit().await.expect("commit");
916 2 : antares_sql::store::pg::outbox::retain(&pool, &tenant, &[seq])
917 2 : .await
918 2 : .expect("retain");
919 :
920 2 : let st = AppState::with_store(
921 2 : "antares".into(),
922 2 : std::sync::Arc::new(antares_sql::store::any::AnyStore::Pg(
923 2 : antares_sql::store::any::PgBackend::new(pool.clone()),
924 2 : )),
925 2 : "postgres",
926 : );
927 : // The current row is the AFTER image — the document the old
928 : // resolution handed back for both halves.
929 2 : let _ = st.store.delete(&tenant, Kind::Entity, id).await;
930 2 : st.store
931 2 : .create(&tenant, Kind::Entity, id, after_doc.clone())
932 2 : .await
933 2 : .expect("seed current row");
934 :
935 2 : ev.seq = seq;
936 2 : let wire = ev.claim_check(antares_bus::CLAIM_CHECK_BYTES);
937 2 : assert!(
938 2 : wire.prev_payload_ref.is_some() && wire.payload_ref.is_some(),
939 : "the fixture must be over the claim-check ceiling"
940 : );
941 2 : let (before, after) = resolve_payloads(&st, &wire).await;
942 2 : assert_eq!(before.as_ref(), Some(&before_doc), "before-image lost");
943 2 : assert_eq!(after.as_ref(), Some(&after_doc));
944 2 : assert_ne!(
945 : before, after,
946 : "both halves resolved to the current row: the change notifies nobody"
947 : );
948 :
949 2 : let _ = st.store.delete(&tenant, Kind::Entity, id).await;
950 2 : let _ = antares_sql::store::pg::outbox::reap_published(&pool, 0).await;
951 2 : }
952 :
953 : /// The outbox-drain switch is a config value like any other: the two
954 : /// documented spellings decide, anything else is fatal instead of
955 : /// silently leaving the drain on (a typo'd `ANTARES_OUTBOX_DRAIN=of`
956 : /// would otherwise read as "on" and quietly defeat the crash drill).
957 : #[test]
958 2 : fn outbox_drain_switch_is_total() {
959 2 : std::env::remove_var("ANTARES_OUTBOX_DRAIN");
960 2 : assert!(outbox_drain_enabled().expect("default"), "default is on");
961 2 : std::env::set_var("ANTARES_OUTBOX_DRAIN", "off");
962 2 : assert!(!outbox_drain_enabled().expect("off"));
963 2 : std::env::set_var("ANTARES_OUTBOX_DRAIN", "on");
964 2 : assert!(outbox_drain_enabled().expect("on"));
965 12 : for bad in ["", "of", "false", "0", "OFF", "no"] {
966 12 : std::env::set_var("ANTARES_OUTBOX_DRAIN", bad);
967 12 : let err = outbox_drain_enabled()
968 12 : .expect_err(&format!("ANTARES_OUTBOX_DRAIN={bad:?} must be fatal"));
969 12 : assert!(err.contains("ANTARES_OUTBOX_DRAIN"), "{err}");
970 : }
971 2 : std::env::remove_var("ANTARES_OUTBOX_DRAIN");
972 2 : }
973 : }
|
