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::pg::doc::DocKind;
31 : use antares_sql::store::pg::seal::{self, SecretKey};
32 : use antares_sql::store::Kind;
33 : use futures_util::StreamExt;
34 : use std::sync::Arc;
35 :
36 : #[derive(Clone, Copy, Debug)]
37 : pub struct Roles {
38 : pub api: bool,
39 : pub matcher: bool,
40 : pub notifier: bool,
41 : pub temporal: bool,
42 : pub registry: bool,
43 : }
44 :
45 : impl Roles {
46 46 : pub fn parse(spec: &str) -> Result<Self, String> {
47 46 : if spec == "all" {
48 16 : return Ok(Self {
49 16 : api: true,
50 16 : matcher: true,
51 16 : notifier: true,
52 16 : temporal: true,
53 16 : registry: true,
54 16 : });
55 30 : }
56 30 : let mut r = Self {
57 30 : api: false,
58 30 : matcher: false,
59 30 : notifier: false,
60 30 : temporal: false,
61 30 : registry: false,
62 30 : };
63 44 : for part in spec.split(',') {
64 44 : match part.trim() {
65 44 : "api" => r.api = true,
66 36 : "matcher" => r.matcher = true,
67 28 : "notifier" => r.notifier = true,
68 22 : "temporal" => r.temporal = true,
69 18 : "registry" => r.registry = true,
70 16 : other => {
71 16 : return Err(format!(
72 16 : "unknown role {other:?} (api|matcher|notifier|temporal|registry|all)"
73 16 : ))
74 : }
75 : }
76 : }
77 14 : Ok(r)
78 46 : }
79 :
80 26 : pub fn all(&self) -> bool {
81 26 : self.api && self.matcher && self.notifier && self.temporal && self.registry
82 26 : }
83 : }
84 :
85 : /// ANTARES_OUTBOX_DRAIN: `on` (the default) or `off`, and nothing else. A
86 : /// typo'd `of` read as "on" under the old permissive parse, quietly defeating
87 : /// the dedicated-drainer split and the crash drill it exists for.
88 71 : pub fn outbox_drain_enabled() -> Result<bool, String> {
89 71 : match std::env::var("ANTARES_OUTBOX_DRAIN") {
90 53 : Err(std::env::VarError::NotPresent) => Ok(true),
91 0 : Err(e) => Err(format!("ANTARES_OUTBOX_DRAIN is unreadable: {e}")),
92 18 : Ok(v) => match v.as_str() {
93 18 : "on" => Ok(true),
94 16 : "off" => Ok(false),
95 14 : other => Err(format!(
96 14 : "ANTARES_OUTBOX_DRAIN must be on|off, got {other:?}"
97 14 : )),
98 : },
99 : }
100 71 : }
101 :
102 : /// KV key for one mirrored subscription: tenant verbatim (token-safe by
103 : /// construction), id hashed (URNs carry `:` — illegal in KV keys). The VALUE
104 : /// carries the real tenant/id, so the key only needs uniqueness. Kind-scoped:
105 : /// a Subscription and a Context Source Registration Subscription may carry the
106 : /// same client-chosen id in one tenant (5.5.10), and sharing a key would let
107 : /// one overwrite the other's mirror entry.
108 16 : fn kv_key(tenant: &str, kind: Kind, id: &str) -> String {
109 16 : format!(
110 : "{tenant}.{}{:016x}",
111 16 : if kind == Kind::CSourceSubscription {
112 2 : "c"
113 : } else {
114 14 : ""
115 : },
116 16 : antares_bus::subjects::fnv1a64(id.as_bytes())
117 : )
118 16 : }
119 :
120 : /// Wire everything bus=nats needs onto the state. Async: connects, hydrates
121 : /// mirrors, creates consumers, asserts topology — all before the broker
122 : /// starts accepting traffic, so a mis-shapen topology is a startup failure,
123 : /// never a silent runtime drift. The bus is handed back for the channels
124 : /// that ride its core request/reply (a relay between pods).
125 : ///
126 : /// `secret_key` is `ANTARES_SECRET_KEY`: the subscription KV mirror carries
127 : /// each document's notification credentials sealed under it, as the store's
128 : /// row does ([`seal`]), and the matcher side opens them where it applies an
129 : /// entry.
130 0 : pub async fn wire_nats(
131 0 : state: &mut AppState,
132 0 : url: &str,
133 0 : roles: Roles,
134 0 : secret_key: Option<SecretKey>,
135 0 : ) -> Result<Arc<NatsBus>, Box<dyn std::error::Error>> {
136 0 : let secret_key = secret_key.map(Arc::new);
137 0 : let bus = Arc::new(NatsBus::connect(url).await?);
138 : // /q/health `bus` member: live connection state + reconnect count.
139 : // Installed HERE so bus=local never carries the member at all.
140 : {
141 0 : let b = bus.clone();
142 0 : state.bus_stats = Some(Arc::new(move || {
143 0 : serde_json::json!({
144 0 : "mode": "nats",
145 0 : "connected": b.connected(),
146 0 : "reconnects": b.reconnects(),
147 : })
148 0 : }));
149 : }
150 : // Multi-process mode: interval firings need the single-winner claim
151 : // — keyed off this flag, not off mirror presence (local mode wires a
152 : // mirror too).
153 0 : state.nats = true;
154 : // Entity writes now enqueue their events in the write transaction.
155 0 : state.store.set_outbox(true);
156 : // Auto-recording stays SYNCHRONOUS in the write path in every bus mode:
157 : // every write goes through an api-role pod that has the
158 : // shared store, so recording in-request gives read-your-writes — the
159 : // ETSI suite asserts history immediately after a write — and kills the
160 : // late-replay resurrection race (a consumer re-applying a pre-delete
161 : // event AFTER a direct temporal delete). The recorder consumer this
162 : // replaced double-applied by design; it bought nothing but the races.
163 :
164 : // The drain nudge: a same-process write pokes its own drain, so publish
165 : // latency is ~1 ms, not the idle-poll interval. Cross-pod writes are
166 : // still covered by each pod's own nudge; the poll below stays as the
167 : // crash-recovery fallback.
168 0 : let nudge = Arc::new(tokio::sync::Notify::new());
169 : {
170 0 : let n = nudge.clone();
171 : // Auto-recording rides the same synchronous hook here as in bus=local
172 : // (one choke point for every write, no handler can forget), then nudges
173 : // the outbox drain.
174 0 : let st_rec = state.clone();
175 0 : state.store.set_change_hook(Arc::new(
176 : move |tenant: &TenantId,
177 : before: Option<serde_json::Value>,
178 : after: Option<serde_json::Value>,
179 : _version: antares_store::ChangeVersion|
180 0 : -> antares_store::HookFuture<'_> {
181 0 : let st_rec = st_rec.clone();
182 0 : let n = n.clone();
183 0 : Box::pin(async move {
184 0 : antares_api::notify::record_temporal_change(
185 0 : &st_rec,
186 0 : tenant,
187 0 : before.as_ref(),
188 0 : after.as_ref(),
189 0 : )
190 0 : .await;
191 0 : n.notify_one();
192 0 : })
193 0 : },
194 : ));
195 : }
196 :
197 0 : let mut durables: Vec<&'static str> = Vec::new();
198 :
199 0 : if roles.api {
200 : // Subscription write side: subscription CUD → KV (tombstone = null doc).
201 0 : let kv = bus.subs_kv().await?;
202 0 : let kv_for_hook = kv.clone();
203 0 : let store_for_hook = state.store.clone();
204 0 : let key_for_hook = secret_key.clone();
205 0 : state.sub_sync = Some(Arc::new(
206 0 : move |tenant: &TenantId, kind: Kind, id: &str, _doc| {
207 0 : let kv = kv_for_hook.clone();
208 0 : let store = store_for_hook.clone();
209 0 : let secret = key_for_hook.clone();
210 0 : let key = kv_key(tenant.as_str(), kind, id);
211 0 : let (tenant, id) = (tenant.clone(), id.to_owned());
212 0 : tokio::spawn(async move {
213 : // The store row is already committed. A lost put leaves this
214 : // subscription invisible to every matcher pod — silently, and
215 : // until the next restart, because mirrors hydrate from the
216 : // store only at process start. So retry rather than warn once.
217 : // Named ceiling: after the last attempt the divergence stands
218 : // until a restart; closing that needs periodic reconciliation.
219 : //
220 : // What is put is the row as the store holds it NOW, under a
221 : // compare-and-set on the revision read before the store: each
222 : // write's sync is its own task with its own retries, and a
223 : // document captured at write time, put late, overwrote a
224 : // newer write — in the KV every matcher pod replays after its
225 : // hydrate. A sync that loses the race re-reads both, so the
226 : // last one to succeed carries the store's last write.
227 0 : for attempt in 0..MIRROR_SYNC_ATTEMPTS {
228 0 : let revision = match kv.entry(key.clone()).await {
229 0 : Ok(e) => e.map_or(0, |e| e.revision),
230 0 : Err(e) => {
231 0 : tracing::warn!("sub KV sync attempt {} failed: {e}", attempt + 1);
232 0 : tokio::time::sleep(mirror_sync_backoff(attempt)).await;
233 0 : continue;
234 : }
235 : };
236 : // A Context Source Registration Subscription's entry
237 : // only wakes the sweep (`apply_delta`): it carries no
238 : // document, so none is read, sealed or shipped.
239 0 : let read = if kind == Kind::CSourceSubscription {
240 0 : Ok(None)
241 : } else {
242 0 : store.get(&tenant, kind, &id).await
243 : };
244 0 : let doc = match read {
245 0 : Ok(d) => d,
246 0 : Err(e) => {
247 0 : tracing::warn!("sub KV sync attempt {} failed: {e}", attempt + 1);
248 0 : tokio::time::sleep(mirror_sync_backoff(attempt)).await;
249 0 : continue;
250 : }
251 : };
252 : // The store answered the credentials opened; the
253 : // bucket gets them sealed again, as the row holds them.
254 0 : let row = seal::Row {
255 0 : kind: sub_doc_kind(kind == Kind::CSourceSubscription),
256 0 : tenant: tenant.as_str(),
257 0 : id: &id,
258 0 : };
259 0 : let doc = match doc
260 0 : .map(|d| {
261 0 : seal::sealed(secret.as_deref(), row, &d).map(|c| c.into_owned())
262 0 : })
263 0 : .transpose()
264 : {
265 0 : Ok(d) => d,
266 : Err(_) => {
267 0 : tracing::warn!(
268 : "sub KV sync attempt {} failed: the credentials could \
269 : not be sealed",
270 0 : attempt + 1
271 : );
272 0 : tokio::time::sleep(mirror_sync_backoff(attempt)).await;
273 0 : continue;
274 : }
275 : };
276 0 : let value = serde_json::json!({
277 0 : "tenant": tenant.as_str(), "id": id, "doc": doc,
278 0 : "csub": kind == Kind::CSourceSubscription,
279 : });
280 0 : let bytes = serde_json::to_vec(&value).unwrap_or_default();
281 0 : match kv.update(key.clone(), bytes.into(), revision).await {
282 0 : Ok(_) => return,
283 0 : Err(e) => {
284 0 : tracing::warn!("sub KV sync attempt {} failed: {e}", attempt + 1);
285 0 : tokio::time::sleep(mirror_sync_backoff(attempt)).await;
286 : }
287 : }
288 : }
289 0 : tracing::error!(
290 : "sub KV sync gave up for {key} — this subscription is not mirrored \
291 : until the next restart"
292 : );
293 0 : });
294 0 : },
295 : ));
296 :
297 : // An entry put before ANTARES_SECRET_KEY was set still holds its
298 : // credentials in plaintext, and `sub_sync` rewrites an entry only
299 : // when its subscription is written again: seal them once, here.
300 0 : if let Some(secret) = secret_key.clone() {
301 0 : let (kv, store) = (kv.clone(), state.store.clone());
302 0 : tokio::spawn(async move {
303 0 : let keys: Vec<String> = match kv.keys().await {
304 0 : Ok(keys) => keys.filter_map(|k| async move { k.ok() }).collect().await,
305 0 : Err(e) => {
306 0 : tracing::warn!("sealing the subscription KV: listing failed: {e}");
307 0 : return;
308 : }
309 : };
310 0 : seal_kv_entries(&kv, &*store, &secret, keys).await;
311 0 : });
312 0 : }
313 :
314 : // Registration write side: registration CUD → ANTARES_REGISTRY delta.
315 0 : let bus_for_reg = bus.clone();
316 0 : let store_for_reg = state.store.clone();
317 0 : state.reg_sync = Some(Arc::new(move |tenant: &TenantId, id: &str, _doc| {
318 0 : let bus = bus_for_reg.clone();
319 0 : let store = store_for_reg.clone();
320 0 : let (tenant, id) = (tenant.clone(), id.to_owned());
321 0 : tokio::spawn(async move {
322 : // Same contract as the subscription mirror: the row is
323 : // committed, so a lost delta makes the registration invisible
324 : // to every federation path until a restart re-hydrates.
325 : //
326 : // And the same ordering fix: what is published is the row as
327 : // the store holds it NOW, under a compare-and-set on the
328 : // registration's newest delta read before the store. A
329 : // document captured at write time, published late, landed
330 : // after a newer write's in every pod's mirror. A publish that
331 : // loses the race re-reads both, so the last one to succeed
332 : // carries the store's last write.
333 0 : for attempt in 0..MIRROR_SYNC_ATTEMPTS {
334 0 : let revision = match bus.registration_revision(tenant.as_str(), &id).await {
335 0 : Ok(r) => r,
336 0 : Err(e) => {
337 0 : tracing::warn!(
338 : "registry delta publish attempt {} failed: {e}",
339 0 : attempt + 1
340 : );
341 0 : tokio::time::sleep(mirror_sync_backoff(attempt)).await;
342 0 : continue;
343 : }
344 : };
345 0 : let doc = match store.get(&tenant, Kind::Registration, &id).await {
346 0 : Ok(d) => d,
347 0 : Err(e) => {
348 0 : tracing::warn!(
349 : "registry delta publish attempt {} failed: {e}",
350 0 : attempt + 1
351 : );
352 0 : tokio::time::sleep(mirror_sync_backoff(attempt)).await;
353 0 : continue;
354 : }
355 : };
356 0 : let delta = serde_json::json!({
357 0 : "tenant": tenant.as_str(), "id": id, "doc": doc,
358 : });
359 0 : match bus
360 0 : .publish_registration(tenant.as_str(), &id, &delta, revision)
361 0 : .await
362 : {
363 0 : Ok(()) => return,
364 0 : Err(e) => {
365 0 : tracing::warn!(
366 : "registry delta publish attempt {} failed: {e}",
367 0 : attempt + 1
368 : );
369 0 : tokio::time::sleep(mirror_sync_backoff(attempt)).await;
370 : }
371 : }
372 : }
373 0 : tracing::error!(
374 : "registry delta publish gave up for {id} — this registration is not \
375 : mirrored until the next restart"
376 : );
377 0 : });
378 0 : }));
379 :
380 : // 5.2.34 write side: a cooldown stamp is broadcast to the other api
381 : // pods on the registry stream (seconds-scale state, deliberately not
382 : // persisted) — per-process stamps re-dial a failed source from every
383 : // pod behind the LB.
384 0 : let bus_for_cool = bus.clone();
385 0 : state.reg_fail_sync = Some(Arc::new(move |reg_id: &str, ok: bool| {
386 0 : let bus = bus_for_cool.clone();
387 0 : let delta = serde_json::json!({"cooldownReg": reg_id, "ok": ok});
388 0 : tokio::spawn(async move {
389 0 : if let Err(e) = bus.publish_registry("cooldown", &delta).await {
390 0 : tracing::warn!("cooldown stamp publish failed: {e}");
391 0 : }
392 0 : });
393 0 : }));
394 :
395 : // Registration read side: the ONE compiled registration mirror this instance's
396 : // federation path reads. Consumer created BEFORE the hydrate so no
397 : // delta can fall between them; last-writer-wins per key converges.
398 0 : let reg_mirror = Arc::new(antares_api::mirror::DocMirror::default());
399 0 : let reg_consumer = bus.consume_registry_broadcast().await?;
400 : // Installed only if it is whole. A mirror that is present and SHORT
401 : // is read as the truth — `reg_docs` asks it and never the store — so
402 : // half a hydrate silently drops Context Sources for the life of the
403 : // process. Left uninstalled, federation matching falls back to the
404 : // store's own indexed narrowing: correct, and merely slower.
405 0 : match antares_api::notify::seed_mirror(
406 0 : &*state.store,
407 0 : reg_mirror.as_ref(),
408 0 : Kind::Registration,
409 : )
410 0 : .await
411 : {
412 0 : Ok(()) => state.reg_mirror = Some(reg_mirror.clone()),
413 0 : Err(e) => tracing::error!(
414 : "registration mirror hydrate failed ({e}); \
415 : federation matching falls back to a store read per request"
416 : ),
417 : }
418 0 : tokio::spawn(watch_registry(
419 0 : bus.clone(),
420 0 : reg_consumer,
421 0 : reg_mirror,
422 0 : state.store.clone(),
423 0 : state.egress.clone(),
424 : ));
425 :
426 : // The outbox drain. Runs on every api pod; concurrent drains are
427 : // absorbed by Nats-Msg-Id dedup within the duplicate window.
428 : // ANTARES_OUTBOX_DRAIN=off leaves the rows for another pod's drain —
429 : // the crash-drill lever and the dedicated-drainer split.
430 0 : let drain_on = outbox_drain_enabled()?;
431 0 : if !drain_on {
432 0 : tracing::warn!("outbox drain OFF on this pod (ANTARES_OUTBOX_DRAIN=off)");
433 0 : }
434 0 : if drain_on {
435 0 : let store = state.store.clone();
436 0 : let bus_for_drain = bus.clone();
437 0 : tokio::spawn(async move {
438 : loop {
439 0 : let rows = match store.outbox_peek(64).await {
440 0 : Ok(r) => r,
441 0 : Err(e) => {
442 0 : tracing::warn!("outbox peek failed: {e}");
443 0 : tokio::time::sleep(std::time::Duration::from_secs(1)).await;
444 0 : continue;
445 : }
446 : };
447 0 : if rows.is_empty() {
448 : // woken by the same-process write hook, or the fallback
449 : // poll for rows another pod failed to publish
450 0 : let _ = tokio::time::timeout(
451 0 : std::time::Duration::from_millis(250),
452 0 : nudge.notified(),
453 : )
454 0 : .await;
455 0 : continue;
456 0 : }
457 : // Ack the EXACT published seqs — a blanket
458 : // up-to-max delete loses a lower-seq row whose
459 : // transaction commits between peek and ack.
460 0 : let mut acked: Vec<i64> = Vec::new();
461 : // Rows whose bodies were too big for the bus: the message
462 : // carries a reference, and this row is the only copy of
463 : // what it references. Kept, not deleted, and taken out of
464 : // the next page by the same stamp.
465 0 : let mut retained: Vec<(TenantId, i64)> = Vec::new();
466 0 : for (seq, _tenant, event) in rows {
467 0 : match serde_json::from_value::<ChangeEvent>(event) {
468 0 : Ok(mut ev) => {
469 0 : ev.seq = seq;
470 0 : let checked = ev.claim_checked_at(antares_bus::CLAIM_CHECK_BYTES);
471 0 : match bus_for_drain.publish(&ev).await {
472 0 : Ok(()) if checked => retained.push((ev.tenant, seq)),
473 0 : Ok(()) => acked.push(seq),
474 0 : Err(e) => {
475 0 : tracing::warn!("outbox publish of seq {seq} failed: {e}");
476 0 : break; // retry from here next round
477 : }
478 : }
479 : }
480 0 : Err(e) => {
481 : // an undecodable row would wedge the drain forever
482 0 : tracing::error!("outbox row {seq} undecodable ({e}) — skipped");
483 0 : acked.push(seq);
484 : }
485 : }
486 : }
487 : // Retain BEFORE ack: both statements are separate
488 : // transactions, and a crash between them must leave a
489 : // claim-check row alive rather than published and gone.
490 : // One statement per row, under that row's tenant — the
491 : // outbox UPDATE takes no service escape (0005), and an
492 : // event over the bus ceiling is rare enough that grouping
493 : // the page by tenant would cost more code than statements.
494 0 : for (tenant, seq) in &retained {
495 0 : if let Err(e) = store.outbox_retain(tenant, &[*seq]).await {
496 0 : tracing::warn!("outbox retain of seq {seq} failed: {e}");
497 0 : }
498 : }
499 0 : if !acked.is_empty() {
500 0 : if let Err(e) = store.outbox_ack(&acked).await {
501 0 : tracing::warn!("outbox ack {acked:?} failed: {e}");
502 0 : }
503 0 : }
504 : }
505 : });
506 0 : }
507 0 : }
508 :
509 0 : if roles.matcher || roles.notifier {
510 : // Subscription read side: consumer-before-hydrate, same convergence argument.
511 0 : let sub_mirror = Arc::new(antares_api::mirror::SubMirror::default());
512 0 : let kv = bus.subs_kv().await?;
513 0 : let watch = kv.watch_all().await?;
514 0 : let key = secret_key.clone();
515 : // Same rule as the registration mirror, and the same one `bus=local`
516 : // applies in `antares_api::wire`: a subscription absent from an installed
517 : // mirror never fires again, because the matcher reads candidates
518 : // from the mirror alone.
519 0 : match antares_api::notify::seed_mirror(
520 0 : &*state.store,
521 0 : sub_mirror.as_ref(),
522 0 : Kind::Subscription,
523 : )
524 0 : .await
525 : {
526 0 : Ok(()) => state.sub_mirror = Some(sub_mirror.clone()),
527 0 : Err(e) => tracing::error!(
528 : "subscription mirror hydrate failed ({e}); \
529 : matching falls back to a store scan per change"
530 : ),
531 : }
532 0 : tokio::spawn(async move {
533 : // Same restart contract as the registry consumer: the watch ends
534 : // on a NATS restart, and a task that returns there stops seeing
535 : // every subscription change — i.e. this pod silently stops
536 : // notifying — for the process lifetime. `watch_all` replays the
537 : // bucket's current values, so re-opening also re-converges the
538 : // mirror over the gap.
539 0 : let mut watch = watch;
540 : loop {
541 0 : while let Some(entry) = watch.next().await {
542 0 : let Ok(entry) = entry else { continue };
543 0 : if let Ok(mut v) = serde_json::from_slice::<serde_json::Value>(&entry.value) {
544 0 : open_sub_delta(key.as_deref(), &mut v);
545 0 : apply_delta(sub_mirror.as_ref(), &v);
546 0 : }
547 : }
548 0 : tracing::warn!("subscription KV watch ended — reopening");
549 0 : tokio::time::sleep(std::time::Duration::from_secs(1)).await;
550 0 : match kv.watch_all().await {
551 0 : Ok(w) => watch = w,
552 0 : Err(e) => tracing::warn!("subscription KV watch reopen failed: {e}"),
553 : }
554 : }
555 : });
556 :
557 : // The balanced matcher durable: decode → process_change → ack AFTER,
558 : // with progress heartbeats for every message this pod holds while
559 : // one of them is processed (a slow endpoint must not let ack_wait
560 : // pass and have the server hand them all out a second time).
561 0 : durables.push("matcher");
562 0 : let consumer = bus.consume_balanced("matcher").await?;
563 0 : let st = state.clone();
564 0 : tokio::spawn(async move {
565 : loop {
566 0 : let run = nats::run_balanced(&consumer, |msg| {
567 : // Change lag = stream-publish → matcher-processing
568 : // age, from the JetStream metadata timestamp.
569 0 : if let Ok(info) = msg.info() {
570 0 : note_final_delivery(
571 0 : msg.subject.as_str(),
572 0 : info.stream_sequence,
573 0 : info.delivered,
574 : );
575 : // OffsetDateTime → SystemTime (impl in `time`/std),
576 : // so no direct `time` dependency here.
577 0 : let published: std::time::SystemTime = info.published.into();
578 0 : if let Ok(age) = published.elapsed() {
579 0 : metrics::histogram!("antares_change_lag_seconds")
580 0 : .record(age.as_secs_f64());
581 0 : }
582 0 : }
583 0 : let ev = nats::decode(msg);
584 0 : let st = st.clone();
585 0 : async move {
586 0 : if let Some(ev) = ev {
587 0 : let (before, after) = resolve_payloads(&st, &ev).await;
588 0 : antares_api::notify::process_change(
589 0 : &st,
590 0 : ev.tenant.as_str(),
591 0 : before,
592 0 : after,
593 0 : )
594 0 : .await;
595 0 : }
596 0 : }
597 0 : })
598 0 : .await;
599 0 : if let Err(e) = run {
600 0 : tracing::warn!("matcher consumer stream failed: {e}");
601 0 : }
602 0 : tokio::time::sleep(std::time::Duration::from_secs(1)).await;
603 : }
604 : });
605 :
606 : // Interval subscriptions: every matcher pod ticks; the row-lock claim
607 : // in interval_tick makes each firing single-winner.
608 0 : let st = state.clone();
609 0 : tokio::spawn(async move {
610 0 : let mut tick = tokio::time::interval(std::time::Duration::from_millis(500));
611 : loop {
612 0 : tick.tick().await;
613 0 : antares_api::notify::interval_tick(&st).await;
614 : }
615 : });
616 0 : }
617 :
618 : // The temporal role carries no bus consumer: auto-recording is
619 : // synchronous in the write path (see above), and plain-mode partition
620 : // maintenance runs from main.rs regardless of bus mode.
621 :
622 : // The server must agree these are shared durables.
623 0 : bus.assert_topology(&durables).await?;
624 0 : tracing::info!(?roles, "bus=nats wired");
625 0 : Ok(bus)
626 0 : }
627 :
628 : /// Whether a change message is on its last delivery: the matcher durable
629 : /// gives a message up after `nats::MAX_DELIVER` deliveries, so if this
630 : /// attempt does not finish the server drops the change and the
631 : /// notifications it should produce. Logged at error level (subject and
632 : /// stream sequence, never the payload) and counted, before the work, so such
633 : /// a loss is never silent.
634 6 : fn note_final_delivery(subject: &str, stream_seq: u64, delivered: i64) -> bool {
635 6 : if delivered < nats::MAX_DELIVER {
636 4 : return false;
637 2 : }
638 2 : tracing::error!(
639 : subject,
640 : stream_seq,
641 : delivered,
642 : "change message on its final delivery: if this attempt fails, the server drops it"
643 : );
644 2 : metrics::counter!("antares_change_final_delivery_total").increment(1);
645 2 : true
646 6 : }
647 :
648 : /// The registration mirror's read side (bus=nats, api role): every
649 : /// registry delta into `mirror`, every 5.2.34 cooldown stamp into `egress`.
650 0 : async fn watch_registry(
651 0 : bus: Arc<NatsBus>,
652 0 : reg_consumer: nats::PullConsumer,
653 0 : reg_mirror: Arc<antares_api::mirror::DocMirror>,
654 0 : store: Arc<dyn antares_store::CurrentStateDriver>,
655 0 : egress: Arc<antares_api::egress::Egress>,
656 0 : ) {
657 : // The consumer is ephemeral, so a NATS restart or an inactivity
658 : // gap deletes it server-side and the next pull errors. Ending the
659 : // task there froze this pod's registration mirror — and with it
660 : // all federation matching — for the process lifetime, while
661 : // /q/health still reported the bus connected. Re-open instead,
662 : // and re-hydrate from the store because a fresh consumer starts
663 : // at NEW and never replays what the gap dropped.
664 : // The gap ends on a NEW consumer: the handle of a deleted one pulls
665 : // nothing ever again.
666 0 : let mut reg_consumer = reg_consumer;
667 : loop {
668 0 : match reg_consumer.messages().await {
669 0 : Ok(mut msgs) => {
670 0 : while let Some(delta) = nats::next_delta(&mut msgs).await {
671 : // 5.2.34 read side: a cooldown stamp updates this pod's
672 : // map (a marker delta has no tenant/id — apply_delta
673 : // ignores it on pods that predate the member).
674 0 : if let Some(rid) = delta.get("cooldownReg").and_then(serde_json::Value::as_str)
675 : {
676 0 : egress.reg_record(rid, delta["ok"].as_bool().unwrap_or(false));
677 0 : continue;
678 0 : }
679 0 : apply_delta(reg_mirror.as_ref(), &delta);
680 : }
681 0 : tracing::warn!("registry broadcast consumer stream ended — reopening");
682 : }
683 0 : Err(e) => tracing::warn!("registry broadcast consumer stream failed: {e}"),
684 : }
685 : // Consumer before re-hydrate, the startup order: no delta can
686 : // fall between the two.
687 : loop {
688 0 : tokio::time::sleep(std::time::Duration::from_secs(1)).await;
689 0 : match bus.consume_registry_broadcast().await {
690 0 : Ok(c) => {
691 0 : reg_consumer = c;
692 0 : break;
693 : }
694 0 : Err(e) => tracing::warn!("registry broadcast consumer reopen failed: {e}"),
695 : }
696 : }
697 : // Already installed, so this one cannot be withheld. A failed
698 : // re-hydrate leaves the mirror holding what it had before the
699 : // gap, which the federation path will serve as current: say
700 : // so at error level rather than let it pass as a warning.
701 0 : if let Err(e) = rehydrate_registrations(&*store, ®_mirror).await {
702 0 : tracing::error!(
703 : "registration mirror re-hydrate failed ({e}); \
704 : this pod is matching against registrations from before the gap"
705 : );
706 0 : }
707 : }
708 : }
709 :
710 : /// Seal the subscription KV entries named by `keys` that still hold a
711 : /// plaintext credential, from the store's row and under a compare-and-set
712 : /// on the entry read: a sync that wrote meanwhile put its own sealed entry,
713 : /// and wins. An entry whose subscription this store does not hold is left to
714 : /// the next write's sync. Answers how many entries it sealed.
715 : ///
716 : /// ponytail: one pass at boot, sequential, one entry read per key of the
717 : /// bucket (one key per subscription).
718 0 : async fn seal_kv_entries(
719 0 : kv: &nats::KvStore,
720 0 : store: &dyn antares_store::CurrentStateDriver,
721 0 : secret: &SecretKey,
722 0 : keys: Vec<String>,
723 0 : ) -> usize {
724 0 : let mut sealed = 0;
725 0 : for key in keys {
726 0 : let Ok(Some(entry)) = kv.entry(key.clone()).await else {
727 0 : continue;
728 : };
729 0 : let Ok(v) = serde_json::from_slice::<serde_json::Value>(&entry.value) else {
730 0 : continue;
731 : };
732 0 : let csub = v.get("csub").and_then(serde_json::Value::as_bool) == Some(true);
733 0 : let (Some(tenant), Some(id), Some(doc)) = (
734 0 : v.get("tenant").and_then(serde_json::Value::as_str),
735 0 : v.get("id").and_then(serde_json::Value::as_str),
736 0 : v.get("doc").filter(|d| !d.is_null()),
737 : ) else {
738 0 : continue;
739 : };
740 0 : if !seal::holds_plaintext(sub_doc_kind(csub), doc) {
741 0 : continue;
742 0 : }
743 0 : let Ok(tenant_id) = TenantId::new_internal(tenant) else {
744 0 : continue;
745 : };
746 : // a Context Source Registration Subscription's entry needs no document
747 0 : let doc = if csub {
748 0 : None
749 : } else {
750 0 : match store.get(&tenant_id, Kind::Subscription, id).await {
751 0 : Ok(Some(d)) => {
752 0 : let row = seal::Row {
753 0 : kind: sub_doc_kind(false),
754 0 : tenant,
755 0 : id,
756 0 : };
757 0 : match seal::sealed(Some(secret), row, &d) {
758 0 : Ok(d) => Some(d.into_owned()),
759 0 : Err(_) => continue,
760 : }
761 : }
762 0 : _ => continue,
763 : }
764 : };
765 0 : let value = serde_json::json!({"tenant": tenant, "id": id, "doc": doc, "csub": csub});
766 0 : let bytes = serde_json::to_vec(&value).unwrap_or_default();
767 0 : match kv.update(key.clone(), bytes.into(), entry.revision).await {
768 0 : Ok(_) => sealed += 1,
769 0 : Err(e) => tracing::warn!("sealing the subscription KV entry {key} failed: {e}"),
770 : }
771 : }
772 0 : sealed
773 0 : }
774 :
775 : /// Re-seed the registration mirror after a gap in its deltas. The gap may
776 : /// have swallowed deletions too, so the store's set is seeded into a fresh
777 : /// mirror and replaces the old contents whole; seeding over the old mirror
778 : /// kept every registration deleted during the gap, and federation kept
779 : /// forwarding to it. On an error the old contents stay.
780 2 : async fn rehydrate_registrations(
781 2 : store: &dyn antares_store::CurrentStateDriver,
782 2 : reg_mirror: &antares_api::mirror::DocMirror,
783 2 : ) -> Result<(), antares_model::NgsiError> {
784 2 : let fresh = antares_api::mirror::DocMirror::default();
785 2 : antares_api::notify::seed_mirror(store, &fresh, Kind::Registration).await?;
786 2 : reg_mirror.replace_with(fresh);
787 2 : Ok(())
788 2 : }
789 :
790 : /// Hydrate a mirror from the system of record (Postgres) at startup.
791 : /// Attempts a mirror write gets before the divergence is logged as an error.
792 : const MIRROR_SYNC_ATTEMPTS: u32 = 5;
793 :
794 : /// Exponential backoff between mirror-sync attempts: 0.2 s doubling to 3.2 s,
795 : /// so the whole ladder outlives a bus reconnect without holding a task for
796 : /// minutes.
797 0 : fn mirror_sync_backoff(attempt: u32) -> std::time::Duration {
798 0 : std::time::Duration::from_millis(200u64 << attempt.min(4))
799 0 : }
800 :
801 : /// Apply one `{tenant, id, doc|null, csub}` delta to a mirror. A Context
802 : /// Source Registration Subscription carries no document into the mirror: it
803 : /// is matched against registrations rather than entities, so the delta only
804 : /// wakes the interval sweep (5.11.7).
805 34 : fn apply_delta(mirror: &dyn antares_api::mirror::Mirror, delta: &serde_json::Value) {
806 34 : if delta.get("csub").and_then(serde_json::Value::as_bool) == Some(true) {
807 2 : mirror.csub_written();
808 2 : return;
809 32 : }
810 12 : let (Some(tenant), Some(id)) = (
811 32 : delta.get("tenant").and_then(serde_json::Value::as_str),
812 32 : delta.get("id").and_then(serde_json::Value::as_str),
813 : ) else {
814 20 : return;
815 : };
816 : // Hydration validates the tenant before it touches the mirror; the delta
817 : // path must agree. A name outside the grammar can only add entries that no
818 : // lookup will ever hit, so an unvalidated one is unbounded growth keyed by
819 : // whatever reached the bus. The grammar is the whole check here: what
820 : // arrives is a tenant a peer broker wrote, and a write inside a Snapshot
821 : // (5.5.15) carries the synthetic tenant a client may not name — refusing
822 : // that one would drop the mirror delta the snapshot-scoped subscription
823 : // matches on.
824 12 : if TenantId::new_internal(tenant).is_err() {
825 8 : return;
826 4 : }
827 4 : let doc = delta.get("doc").filter(|d| !d.is_null()).cloned();
828 4 : mirror.apply(tenant, id, doc);
829 34 : }
830 :
831 : /// The doc kind a subscription KV entry's credentials are sealed for.
832 0 : fn sub_doc_kind(csub: bool) -> DocKind {
833 0 : if csub {
834 0 : DocKind::CSourceSubscription
835 : } else {
836 0 : DocKind::Subscription
837 : }
838 0 : }
839 :
840 : /// Open the credentials of one subscription KV entry in place, before
841 : /// [`apply_delta`]. An entry this pod cannot open (no key, another key, a
842 : /// damaged value) becomes a tombstone: the subscription leaves this pod's
843 : /// mirror rather than notify with ciphertext in its headers or its URI.
844 : /// A plaintext entry (one put before the key was set) opens as it is.
845 0 : fn open_sub_delta(key: Option<&SecretKey>, delta: &mut serde_json::Value) {
846 0 : let csub = delta.get("csub").and_then(serde_json::Value::as_bool) == Some(true);
847 0 : let (Some(tenant), Some(id)) = (
848 0 : delta.get("tenant").and_then(serde_json::Value::as_str),
849 0 : delta.get("id").and_then(serde_json::Value::as_str),
850 : ) else {
851 0 : return;
852 : };
853 0 : let (tenant, id) = (tenant.to_owned(), id.to_owned());
854 0 : let Some(doc) = delta.get_mut("doc").filter(|d| !d.is_null()) else {
855 0 : return;
856 : };
857 0 : let row = seal::Row {
858 0 : kind: sub_doc_kind(csub),
859 0 : tenant: &tenant,
860 0 : id: &id,
861 0 : };
862 0 : if seal::open(key, row, doc).is_err() {
863 0 : tracing::error!(
864 : %tenant, %id,
865 : "a mirrored subscription's notification credentials cannot be opened: \
866 : ANTARES_SECRET_KEY is unset or differs from the key it was sealed with; \
867 : this pod drops the subscription from its mirror"
868 : );
869 0 : *doc = serde_json::Value::Null;
870 0 : }
871 0 : }
872 :
873 : /// Resolve claim-check references: bodies the bus could not carry come back
874 : /// from the outbox row the drain kept, read by the event's own `seq`.
875 : ///
876 : /// The store's current row is NOT that source. It answers with the entity as
877 : /// it stands now, which is the after-image: resolving `prev_payload_ref` from
878 : /// it hands the matcher two copies of the same document, `diff` finds nothing
879 : /// changed and the change reaches no subscriber. It stays the fallback for
880 : /// `payload_ref` alone, where being newer than the referenced version is the
881 : /// ordinary at-least-once reality the matcher already tolerates.
882 12 : async fn resolve_payloads(
883 12 : st: &AppState,
884 12 : ev: &ChangeEvent,
885 12 : ) -> (Option<serde_json::Value>, Option<serde_json::Value>) {
886 12 : if ev.payload_ref.is_none() && ev.prev_payload_ref.is_none() {
887 4 : return (ev.prev_payload.clone(), ev.payload.clone());
888 8 : }
889 : // 0 is the local bus, which never claim-checks: it hands the payloads to
890 : // the matcher in process.
891 8 : let kept = match ev.seq {
892 6 : 0 => None,
893 2 : seq => st
894 2 : .store
895 2 : .outbox_event(seq, &ev.tenant)
896 2 : .await
897 2 : .ok()
898 2 : .flatten()
899 2 : .and_then(|v| serde_json::from_value::<ChangeEvent>(v).ok()),
900 : };
901 8 : if let Some(kept) = kept {
902 2 : return (kept.prev_payload, kept.payload);
903 6 : }
904 : // Past the retention window, or a deployment whose store keeps no outbox.
905 : // The after-image is still recoverable; the before-image is not, and a
906 : // guess in its place is a notification that reports a change that did not
907 : // happen.
908 6 : let after = match ev.payload.clone() {
909 0 : Some(v) => Some(v),
910 6 : None => match ev.payload_ref.as_ref() {
911 6 : Some(r) => st
912 6 : .store
913 6 : .get(&ev.tenant, Kind::Entity, r.entity_id.as_str())
914 6 : .await
915 6 : .ok()
916 6 : .flatten(),
917 0 : None => None,
918 : },
919 : };
920 6 : if ev.prev_payload_ref.is_some() {
921 0 : metrics::counter!("antares_claim_check_unresolved_total").increment(1);
922 0 : tracing::warn!(
923 : "claim-check row for seq {} is gone: the change to {} notifies nobody",
924 : ev.seq,
925 0 : ev.entity_id.as_str()
926 : );
927 6 : }
928 6 : (ev.prev_payload.clone(), after)
929 12 : }
930 :
931 : #[cfg(test)]
932 : mod tests {
933 : use super::*;
934 : use antares_api::mirror::DocMirror;
935 : use antares_bus::{ChangeOp, PayloadRef};
936 : use antares_model::EntityId;
937 : use serde_json::json;
938 :
939 : /// The matcher durable drops a change after MAX_DELIVER deliveries: the
940 : /// last one is reported, the ones before it are not.
941 : #[test]
942 2 : fn only_the_final_delivery_of_a_change_is_reported() {
943 2 : assert!(!note_final_delivery("antares.changes.t", 7, 1));
944 2 : assert!(!note_final_delivery(
945 2 : "antares.changes.t",
946 2 : 7,
947 2 : nats::MAX_DELIVER - 1
948 2 : ));
949 2 : assert!(note_final_delivery(
950 2 : "antares.changes.t",
951 : 7,
952 : nats::MAX_DELIVER
953 : ));
954 2 : }
955 :
956 : #[test]
957 2 : fn roles_parse_accepts_the_role_set_and_refuses_anything_else() {
958 2 : assert!(Roles::parse("all").expect("all").all());
959 2 : let r = Roles::parse("api").expect("api");
960 2 : assert!(r.api && !r.matcher && !r.notifier && !r.temporal && !r.registry);
961 2 : assert!(
962 2 : !r.all(),
963 : "a single role must never claim to be the full set"
964 : );
965 2 : let r = Roles::parse(" matcher , notifier ").expect("padded list");
966 2 : assert!(r.matcher && r.notifier && !r.api);
967 : // The enumerated full set is the same thing as "all" — a role split
968 : // that happens to name every role must still pass the bus=local gate.
969 2 : assert!(Roles::parse("api,matcher,notifier,temporal,registry")
970 2 : .expect("full list")
971 2 : .all());
972 :
973 14 : for bad in ["", "api,", "ALL", "apis", "api;matcher", "worker", " "] {
974 14 : let err = Roles::parse(bad).expect_err(&format!("ANTARES_ROLES={bad:?} must be fatal"));
975 14 : assert!(err.starts_with("unknown role"), "{bad:?}: {err}");
976 : }
977 2 : }
978 :
979 : /// The KV key must be legal for a NATS KV bucket (`:` from a URN is not),
980 : /// stable across calls, and collision-free per id, per tenant and per
981 : /// kind — 5.5.10 leaves the id to the client, so one URN can name both a
982 : /// Subscription and a Context Source Registration Subscription.
983 : #[test]
984 2 : fn kv_key_is_bucket_legal_and_stable() {
985 2 : let sub = Kind::Subscription;
986 2 : let k = kv_key("default", sub, "urn:ngsi-ld:Subscription:1");
987 2 : assert_eq!(
988 : k,
989 2 : kv_key("default", sub, "urn:ngsi-ld:Subscription:1"),
990 : "stable"
991 : );
992 2 : assert!(
993 2 : k.bytes()
994 48 : .all(|b| b.is_ascii_alphanumeric() || b == b'.' || b == b'_' || b == b'-'),
995 : "illegal KV key character in {k:?}"
996 : );
997 2 : assert!(k.starts_with("default."), "tenant scoping lost: {k}");
998 2 : assert_ne!(
999 2 : kv_key("default", sub, "urn:ngsi-ld:Subscription:1"),
1000 2 : kv_key("default", sub, "urn:ngsi-ld:Subscription:2")
1001 : );
1002 2 : assert_ne!(
1003 2 : kv_key("t1", sub, "urn:ngsi-ld:Subscription:1"),
1004 2 : kv_key("t2", sub, "urn:ngsi-ld:Subscription:1")
1005 : );
1006 2 : assert_ne!(
1007 2 : kv_key("default", sub, "urn:ngsi-ld:Subscription:1"),
1008 2 : kv_key(
1009 2 : "default",
1010 2 : Kind::CSourceSubscription,
1011 2 : "urn:ngsi-ld:Subscription:1"
1012 : ),
1013 : "one id naming both kinds must not share a mirror entry"
1014 : );
1015 2 : }
1016 :
1017 : /// A Context Source Registration Subscription delta carries no document:
1018 : /// it wakes the interval sweep and touches nothing the matcher reads.
1019 : #[test]
1020 2 : fn a_csub_delta_only_wakes_the_sweep() {
1021 2 : let m = antares_api::mirror::SubMirror::default();
1022 2 : apply_delta(
1023 2 : &m,
1024 2 : &serde_json::json!({
1025 2 : "tenant": "default", "id": "urn:ngsi-ld:CSourceSubscription:1",
1026 2 : "doc": {"id": "urn:ngsi-ld:CSourceSubscription:1", "timeInterval": 5},
1027 2 : "csub": true,
1028 2 : }),
1029 : );
1030 2 : assert!(
1031 2 : m.docs("default").is_empty(),
1032 : "a csource subscription must not enter the candidate index"
1033 : );
1034 2 : }
1035 :
1036 : /// The KV/registry mirrors are fed from the bus. A malformed or hostile
1037 : /// delta must be dropped — never panic a consumer task, and never grow
1038 : /// the mirror under a key no request can ever address (the tenant is
1039 : /// validated on every request, so an unvalidated one is pure ballast).
1040 : #[test]
1041 2 : fn apply_delta_ignores_malformed_and_hostile_deltas() {
1042 2 : let m = DocMirror::default();
1043 2 : apply_delta(
1044 2 : &m,
1045 2 : &json!({"tenant": "default", "id": "urn:x:1", "doc": {"id": "urn:x:1"}}),
1046 : );
1047 2 : assert_eq!(m.docs("default").len(), 1, "a good delta must apply");
1048 :
1049 : // tombstone
1050 2 : apply_delta(
1051 2 : &m,
1052 2 : &json!({"tenant": "default", "id": "urn:x:1", "doc": null}),
1053 : );
1054 2 : assert!(m.docs("default").is_empty(), "null doc must delete");
1055 2 : assert!(m.tenants().is_empty(), "an emptied tenant must not linger");
1056 :
1057 : // Shapes that must be ignored without panicking.
1058 20 : for junk in [
1059 2 : json!({}),
1060 2 : json!(null),
1061 2 : json!(42),
1062 2 : json!("scalar"),
1063 2 : json!([1, 2, 3]),
1064 2 : json!({"tenant": "default"}),
1065 2 : json!({"id": "urn:x:1"}),
1066 2 : json!({"tenant": 7, "id": "urn:x:1", "doc": {}}),
1067 2 : json!({"tenant": "default", "id": null, "doc": {}}),
1068 2 : json!({"cooldownReg": "urn:reg:1", "ok": false}),
1069 20 : ] {
1070 20 : apply_delta(&m, &junk);
1071 20 : }
1072 2 : assert!(m.tenants().is_empty(), "junk deltas grew the mirror");
1073 :
1074 : // Hostile tenants: not addressable by any request (the header is
1075 : // validated to [A-Za-z0-9_-]{1,64}), so they may not take memory.
1076 8 : for hostile in [
1077 2 : "a".repeat(4096),
1078 2 : "../../etc".into(),
1079 2 : "a b".into(),
1080 2 : "".into(),
1081 8 : ] {
1082 8 : apply_delta(
1083 8 : &m,
1084 8 : &json!({"tenant": hostile, "id": "urn:x:1", "doc": {"id": "urn:x:1"}}),
1085 8 : );
1086 8 : }
1087 2 : assert!(
1088 2 : m.tenants().is_empty(),
1089 : "an unaddressable tenant grew the mirror without bound: {:?}",
1090 0 : m.tenants()
1091 : );
1092 2 : }
1093 :
1094 6 : async fn state_with(entity: Option<serde_json::Value>) -> AppState {
1095 6 : let st = AppState::new("antares".into());
1096 6 : if let Some(doc) = entity {
1097 4 : let id = doc["id"].as_str().expect("id").to_owned();
1098 4 : st.store
1099 4 : .create(&TenantId::default(), Kind::Entity, &id, doc)
1100 4 : .await
1101 4 : .expect("seed");
1102 2 : }
1103 6 : st
1104 6 : }
1105 :
1106 10 : fn event(payload: Option<serde_json::Value>, r#ref: Option<PayloadRef>) -> ChangeEvent {
1107 10 : ChangeEvent {
1108 10 : tenant: TenantId::default(),
1109 10 : entity_id: EntityId::new("urn:ngsi-ld:T:1").expect("id"),
1110 10 : types: vec!["T".into()],
1111 10 : op: ChangeOp::Update,
1112 10 : changed_attrs: vec![],
1113 10 : payload,
1114 10 : prev_payload: None,
1115 10 : version: 1,
1116 10 : incarnation: String::new(),
1117 10 : seq: 0,
1118 10 : payload_ref: r#ref,
1119 10 : prev_payload_ref: None,
1120 10 : }
1121 10 : }
1122 :
1123 : /// Claim-check resolution: inline wins, a reference is fetched, and a
1124 : /// reference to a row that is gone resolves to None instead of panicking
1125 : /// the matcher task.
1126 : #[tokio::test]
1127 2 : async fn resolve_payloads_prefers_inline_and_tolerates_a_dangling_reference() {
1128 2 : let doc = json!({"id": "urn:ngsi-ld:T:1", "type": "T"});
1129 2 : let st = state_with(Some(doc.clone())).await;
1130 :
1131 2 : let (before, after) =
1132 2 : resolve_payloads(&st, &event(Some(json!({"inline": true})), None)).await;
1133 2 : assert_eq!(after, Some(json!({"inline": true})), "inline payload wins");
1134 2 : assert_eq!(before, None);
1135 :
1136 2 : let r = PayloadRef {
1137 2 : entity_id: EntityId::new("urn:ngsi-ld:T:1").expect("id"),
1138 2 : version: 1,
1139 2 : };
1140 2 : let (_, after) = resolve_payloads(&st, &event(None, Some(r.clone()))).await;
1141 2 : assert_eq!(
1142 2 : after.as_ref().and_then(|a| a["id"].as_str()),
1143 : Some("urn:ngsi-ld:T:1"),
1144 : "a claim-check reference must be fetched from the store"
1145 : );
1146 :
1147 : // The row was deleted between publish and consumption.
1148 2 : let gone = state_with(None).await;
1149 2 : let (before, after) = resolve_payloads(&gone, &event(None, Some(r))).await;
1150 2 : assert_eq!(after, None, "a dangling reference must resolve to None");
1151 2 : assert_eq!(before, None);
1152 :
1153 2 : let (before, after) = resolve_payloads(&st, &event(None, None)).await;
1154 2 : assert!(before.is_none() && after.is_none(), "no payload, no fetch");
1155 2 : }
1156 :
1157 : /// A claim-check fetch is scoped to the EVENT's tenant: a reference must
1158 : /// never resolve against another tenant's row of the same id.
1159 : #[tokio::test]
1160 2 : async fn resolve_payloads_never_crosses_a_tenant_boundary() {
1161 2 : let st = state_with(Some(json!({"id": "urn:ngsi-ld:T:1", "type": "T"}))).await;
1162 2 : let mut ev = event(
1163 2 : None,
1164 2 : Some(PayloadRef {
1165 2 : entity_id: EntityId::new("urn:ngsi-ld:T:1").expect("id"),
1166 2 : version: 1,
1167 2 : }),
1168 : );
1169 2 : ev.tenant = TenantId::new("other").expect("tenant");
1170 2 : let (_, after) = resolve_payloads(&st, &ev).await;
1171 2 : assert_eq!(
1172 2 : after, None,
1173 2 : "a reference resolved another tenant's entity: {after:?}"
1174 2 : );
1175 2 : }
1176 :
1177 : /// A change whose bodies were both too big for the bus reaches the
1178 : /// matcher with the before-image the write actually replaced. Resolving
1179 : /// the reference against the store's current row instead hands back the
1180 : /// after-image twice, `diff` finds nothing and the change notifies
1181 : /// nobody. Skips without ANTARES_TEST_DATABASE_URL — the outbox is a
1182 : /// Postgres table, and the memory arm has none.
1183 : #[tokio::test(flavor = "multi_thread")]
1184 2 : async fn a_retained_row_gives_the_matcher_the_before_image_the_write_replaced() {
1185 2 : let Ok(url) = std::env::var("ANTARES_TEST_DATABASE_URL") else {
1186 0 : eprintln!("SKIP: ANTARES_TEST_DATABASE_URL not set");
1187 0 : return;
1188 : };
1189 2 : let pool = antares_sql::store::pg::connect(&url, 5)
1190 2 : .await
1191 2 : .expect("connect");
1192 2 : let tenant = TenantId::new("claimcheck").expect("tenant");
1193 2 : antares_sql::store::pg::ensure_tenant(&pool, &tenant)
1194 2 : .await
1195 2 : .expect("tenant row");
1196 :
1197 2 : let id = "urn:ngsi-ld:T:oversized";
1198 4 : let wide = |v: &str| {
1199 4 : json!({"id": id, "type": ["T"],
1200 4 : "https://uri.etsi.org/ngsi-ld/default-context/note":
1201 4 : [{"type": "Property", "value": format!("{v}{}", "x".repeat(300 * 1024))}]})
1202 4 : };
1203 2 : let before_doc = wide("before-");
1204 2 : let after_doc = wide("after-");
1205 :
1206 2 : let mut ev = ChangeEvent {
1207 2 : tenant: tenant.clone(),
1208 2 : entity_id: EntityId::new(id).expect("id"),
1209 2 : types: vec!["T".into()],
1210 2 : op: ChangeOp::Update,
1211 2 : changed_attrs: vec![],
1212 2 : payload: Some(after_doc.clone()),
1213 2 : prev_payload: Some(before_doc.clone()),
1214 2 : version: 2,
1215 2 : incarnation: String::new(),
1216 2 : seq: 0,
1217 2 : payload_ref: None,
1218 2 : prev_payload_ref: None,
1219 2 : };
1220 2 : let mut tx = pool.begin().await.expect("tx");
1221 2 : antares_sql::store::pg::set_tenant(&mut tx, &tenant)
1222 2 : .await
1223 2 : .expect("set tenant");
1224 2 : let seq = antares_sql::store::pg::outbox::enqueue(
1225 2 : &mut tx,
1226 2 : &tenant,
1227 2 : &serde_json::to_value(&ev).expect("event json"),
1228 2 : )
1229 2 : .await
1230 2 : .expect("enqueue");
1231 2 : tx.commit().await.expect("commit");
1232 2 : antares_sql::store::pg::outbox::retain(&pool, &tenant, &[seq])
1233 2 : .await
1234 2 : .expect("retain");
1235 :
1236 2 : let st = AppState::with_store(
1237 2 : "antares".into(),
1238 2 : std::sync::Arc::new(antares_sql::store::any::AnyStore::Pg(
1239 2 : antares_sql::store::any::PgBackend::new(pool.clone()),
1240 2 : )),
1241 2 : "postgres",
1242 : );
1243 : // The current row is the AFTER image — the document the old
1244 : // resolution handed back for both halves.
1245 2 : let _ = st.store.delete(&tenant, Kind::Entity, id).await;
1246 2 : st.store
1247 2 : .create(&tenant, Kind::Entity, id, after_doc.clone())
1248 2 : .await
1249 2 : .expect("seed current row");
1250 :
1251 2 : ev.seq = seq;
1252 2 : let wire = ev.claim_check(antares_bus::CLAIM_CHECK_BYTES);
1253 2 : assert!(
1254 2 : wire.prev_payload_ref.is_some() && wire.payload_ref.is_some(),
1255 : "the fixture must be over the claim-check ceiling"
1256 : );
1257 2 : let (before, after) = resolve_payloads(&st, &wire).await;
1258 2 : assert_eq!(before.as_ref(), Some(&before_doc), "before-image lost");
1259 2 : assert_eq!(after.as_ref(), Some(&after_doc));
1260 2 : assert_ne!(
1261 : before, after,
1262 : "both halves resolved to the current row: the change notifies nobody"
1263 : );
1264 :
1265 2 : let _ = st.store.delete(&tenant, Kind::Entity, id).await;
1266 2 : let _ = antares_sql::store::pg::outbox::reap_published(&pool, 0).await;
1267 2 : }
1268 :
1269 : /// The outbox-drain switch is a config value like any other: the two
1270 : /// documented spellings decide, anything else is fatal instead of
1271 : /// silently leaving the drain on (a typo'd `ANTARES_OUTBOX_DRAIN=of`
1272 : /// would otherwise read as "on" and quietly defeat the crash drill).
1273 : #[test]
1274 2 : fn outbox_drain_switch_is_total() {
1275 2 : std::env::remove_var("ANTARES_OUTBOX_DRAIN");
1276 2 : assert!(outbox_drain_enabled().expect("default"), "default is on");
1277 2 : std::env::set_var("ANTARES_OUTBOX_DRAIN", "off");
1278 2 : assert!(!outbox_drain_enabled().expect("off"));
1279 2 : std::env::set_var("ANTARES_OUTBOX_DRAIN", "on");
1280 2 : assert!(outbox_drain_enabled().expect("on"));
1281 12 : for bad in ["", "of", "false", "0", "OFF", "no"] {
1282 12 : std::env::set_var("ANTARES_OUTBOX_DRAIN", bad);
1283 12 : let err = outbox_drain_enabled()
1284 12 : .expect_err(&format!("ANTARES_OUTBOX_DRAIN={bad:?} must be fatal"));
1285 12 : assert!(err.contains("ANTARES_OUTBOX_DRAIN"), "{err}");
1286 : }
1287 2 : std::env::remove_var("ANTARES_OUTBOX_DRAIN");
1288 2 : }
1289 :
1290 : /// Live NATS for the mirror-sync tests; they skip without it.
1291 12 : fn nats_url() -> Option<String> {
1292 12 : let url = std::env::var("ANTARES_TEST_NATS_URL").ok();
1293 12 : if url.is_none() {
1294 12 : eprintln!("SKIP: ANTARES_TEST_NATS_URL not set");
1295 12 : }
1296 12 : url
1297 12 : }
1298 :
1299 0 : async fn wait_until<F: std::future::Future<Output = bool>>(
1300 0 : what: &str,
1301 0 : mut f: impl FnMut() -> F,
1302 0 : ) {
1303 0 : let rounds = 100 * antares_api::state::slow_factor();
1304 0 : for _ in 0..rounds {
1305 0 : if f().await {
1306 0 : return;
1307 0 : }
1308 0 : tokio::time::sleep(std::time::Duration::from_millis(100)).await;
1309 : }
1310 0 : panic!("timed out waiting for {what}");
1311 0 : }
1312 :
1313 : /// The registry broadcast consumer is ephemeral: a NATS restart or an
1314 : /// inactivity gap deletes it on the server. The watcher has to come back
1315 : /// on a NEW consumer; pulling again on the deleted one never delivers,
1316 : /// so every later delta (a deleted registration included) is lost to
1317 : /// this pod's mirror for the life of the process.
1318 : #[tokio::test(flavor = "multi_thread")]
1319 2 : async fn the_registry_watcher_survives_its_consumer_being_deleted() {
1320 2 : let Some(url) = nats_url() else { return };
1321 0 : let bus = Arc::new(NatsBus::connect(&url).await.expect("connect"));
1322 0 : let st = AppState::new("antares".into());
1323 0 : let mirror = Arc::new(DocMirror::default());
1324 0 : let consumer = bus.consume_registry_broadcast().await.expect("consumer");
1325 0 : let name = consumer.cached_info().name.clone();
1326 0 : let watcher = tokio::spawn(watch_registry(
1327 0 : bus.clone(),
1328 0 : consumer,
1329 0 : mirror.clone(),
1330 0 : st.store.clone(),
1331 0 : st.egress.clone(),
1332 : ));
1333 0 : let tenant = format!("regwatch{}", std::process::id());
1334 0 : let delta = |id: &str| json!({"tenant": tenant, "id": id, "doc": {"id": id, "type": "ContextSourceRegistration"}});
1335 0 : let held = |id: &'static str| {
1336 0 : let (m, t) = (mirror.clone(), tenant.clone());
1337 0 : move || {
1338 0 : let (m, t) = (m.clone(), t.clone());
1339 0 : async move { m.docs(&t).iter().any(|d| d["id"] == id) }
1340 0 : }
1341 0 : };
1342 0 : bus.publish_registry(&tenant, &delta("urn:reg:before"))
1343 0 : .await
1344 0 : .expect("publish");
1345 0 : wait_until("the delta before the gap", held("urn:reg:before")).await;
1346 :
1347 0 : bus.delete_registry_consumer(&name).await.expect("delete");
1348 : // the delta keeps being published until a consumer exists to see it:
1349 : // a fresh consumer starts at NEW
1350 0 : let deadline = std::time::Instant::now()
1351 0 : + std::time::Duration::from_secs(20 * antares_api::state::slow_factor());
1352 2 : loop {
1353 2 : bus.publish_registry(&tenant, &delta("urn:reg:after"))
1354 0 : .await
1355 2 : .expect("publish");
1356 2 : if held("urn:reg:after")().await {
1357 2 : break;
1358 2 : }
1359 2 : assert!(
1360 2 : std::time::Instant::now() < deadline,
1361 2 : "no delta reached the mirror after its consumer was deleted"
1362 2 : );
1363 2 : tokio::time::sleep(std::time::Duration::from_millis(500)).await;
1364 2 : }
1365 2 : watcher.abort();
1366 2 : }
1367 :
1368 : /// A gap in the registry deltas can swallow a deletion: the re-hydrate
1369 : /// that ends it must drop the registration as well as add the ones it
1370 : /// missed, or federation keeps forwarding to a deleted Context Source.
1371 : #[tokio::test]
1372 2 : async fn a_rehydrate_drops_a_registration_deleted_during_the_gap() {
1373 2 : let st = AppState::new("antares".into());
1374 2 : let tenant = antares_model::TenantId::new("rehydrategap").expect("tenant");
1375 6 : let reg = |id: &str| {
1376 6 : json!({"id": id, "type": "ContextSourceRegistration",
1377 6 : "information": [{"entities": [{"type": "Vehicle"}]}],
1378 6 : "endpoint": "http://127.0.0.1:9"})
1379 6 : };
1380 2 : st.store
1381 2 : .create(
1382 2 : &tenant,
1383 2 : Kind::Registration,
1384 2 : "urn:reg:kept",
1385 2 : reg("urn:reg:kept"),
1386 2 : )
1387 2 : .await
1388 2 : .expect("kept");
1389 2 : let mirror = DocMirror::default();
1390 2 : mirror.apply(tenant.as_str(), "urn:reg:kept", Some(reg("urn:reg:kept")));
1391 : // deleted from the store while no delta reached this pod
1392 2 : mirror.apply(tenant.as_str(), "urn:reg:gone", Some(reg("urn:reg:gone")));
1393 2 : rehydrate_registrations(&*st.store, &mirror)
1394 2 : .await
1395 2 : .expect("rehydrate");
1396 2 : let ids: Vec<serde_json::Value> = mirror
1397 2 : .docs(tenant.as_str())
1398 2 : .iter()
1399 2 : .map(|d| d["id"].clone())
1400 2 : .collect();
1401 2 : assert_eq!(ids, vec![json!("urn:reg:kept")]);
1402 2 : }
1403 :
1404 : /// The KV holds what the store holds. Each write's sync runs as its own
1405 : /// task with its own retries, so one that lands late (a retried put, or
1406 : /// a task scheduled after a newer write's) used to overwrite the newer
1407 : /// value with the document it captured — and the KV is what every
1408 : /// matcher pod replays, after its store hydrate, on every (re)start.
1409 : #[tokio::test(flavor = "multi_thread")]
1410 2 : async fn a_late_subscription_sync_never_overwrites_a_newer_write() {
1411 : use antares_store::CurrentStateDriverExt;
1412 2 : let Some(url) = nats_url() else { return };
1413 0 : let mut st = AppState::new("antares".into());
1414 0 : let _bus = wire_nats(&mut st, &url, Roles::parse("api").expect("roles"), None)
1415 0 : .await
1416 0 : .expect("wire");
1417 0 : let tenant = TenantId::default();
1418 0 : let id = format!("urn:ngsi-ld:Subscription:late{}", std::process::id());
1419 0 : let doc = |v: i64| {
1420 0 : json!({"id": id, "type": "Subscription", "v": v,
1421 0 : "entities": [{"type": "https://x/T"}]})
1422 0 : };
1423 0 : st.store
1424 0 : .create(&tenant, Kind::Subscription, &id, doc(1))
1425 0 : .await
1426 0 : .expect("create");
1427 0 : st.store
1428 0 : .mutate::<(), ()>(&tenant, Kind::Subscription, &id, |d| {
1429 0 : *d = doc(2);
1430 0 : Ok(())
1431 0 : })
1432 0 : .await
1433 0 : .expect("update");
1434 0 : let sync = st.sub_sync.clone().expect("api role installs the sync");
1435 0 : let kv = NatsBus::connect(&url)
1436 0 : .await
1437 0 : .expect("connect")
1438 0 : .subs_kv()
1439 0 : .await
1440 0 : .expect("kv");
1441 0 : let key = kv_key(tenant.as_str(), Kind::Subscription, &id);
1442 : // the subscription's `v` as the KV holds it
1443 : macro_rules! kv_v {
1444 : () => {{
1445 : let (kv, key) = (kv.clone(), key.clone());
1446 0 : async move {
1447 0 : kv.get(key)
1448 0 : .await
1449 0 : .ok()
1450 0 : .flatten()
1451 0 : .and_then(|b| serde_json::from_slice::<serde_json::Value>(&b).ok())
1452 0 : .and_then(|v| v["doc"]["v"].as_i64())
1453 0 : }
1454 : }};
1455 : }
1456 : // the newer write's sync lands first ...
1457 0 : sync(&tenant, Kind::Subscription, &id, Some(&doc(2)));
1458 0 : wait_until("the newer write in the KV", || {
1459 0 : let v = kv_v!();
1460 0 : async move { v.await == Some(2) }
1461 0 : })
1462 0 : .await;
1463 : // ... and the older write's sync after it
1464 0 : sync(&tenant, Kind::Subscription, &id, Some(&doc(1)));
1465 0 : tokio::time::sleep(std::time::Duration::from_millis(
1466 0 : 1_000 * antares_api::state::slow_factor(),
1467 0 : ))
1468 0 : .await;
1469 0 : assert_eq!(
1470 2 : kv_v!().await,
1471 2 : Some(2),
1472 2 : "the KV went back to the document an older write captured"
1473 2 : );
1474 2 : }
1475 :
1476 : /// 32 zero bytes in base64: a test key that is plainly not a secret.
1477 : const TEST_KEY: &str = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=";
1478 :
1479 : /// With ANTARES_SECRET_KEY the subscription KV mirror holds what the
1480 : /// store's row holds: the receiverInfo values and a URI with userinfo
1481 : /// sealed, never the plaintext. The matcher side opens them into its
1482 : /// mirror; a pod without the key, or with another, drops the entry
1483 : /// instead of mirroring ciphertext it would send as headers.
1484 : #[tokio::test(flavor = "multi_thread")]
1485 2 : async fn the_subscription_kv_holds_the_credentials_sealed() {
1486 2 : let Some(url) = nats_url() else { return };
1487 0 : let mut st = AppState::new("antares".into());
1488 0 : let _bus = wire_nats(
1489 0 : &mut st,
1490 0 : &url,
1491 0 : Roles::parse("api").expect("roles"),
1492 0 : Some(SecretKey::from_base64(TEST_KEY).expect("key")),
1493 0 : )
1494 0 : .await
1495 0 : .expect("wire");
1496 0 : let tenant = TenantId::default();
1497 0 : let id = format!("urn:ngsi-ld:Subscription:sealedkv{}", std::process::id());
1498 0 : let doc = json!({"id": id, "type": "Subscription",
1499 0 : "entities": [{"type": "https://x/T"}],
1500 0 : "notification": {"endpoint": {
1501 0 : "uri": "http://hook:pa55word@localhost:1/n",
1502 0 : "receiverInfo": [{"key": "Authorization", "value": "Bearer kv-token"}]}}});
1503 0 : st.store
1504 0 : .create(&tenant, Kind::Subscription, &id, doc.clone())
1505 0 : .await
1506 0 : .expect("create");
1507 0 : let kv = NatsBus::connect(&url)
1508 0 : .await
1509 0 : .expect("connect")
1510 0 : .subs_kv()
1511 0 : .await
1512 0 : .expect("kv");
1513 0 : let key = kv_key(tenant.as_str(), Kind::Subscription, &id);
1514 0 : let sync = st.sub_sync.clone().expect("api role installs the sync");
1515 0 : sync(&tenant, Kind::Subscription, &id, Some(&doc));
1516 0 : wait_until("the subscription in the KV", || {
1517 0 : let (kv, key) = (kv.clone(), key.clone());
1518 0 : async move { kv.get(key).await.ok().flatten().is_some() }
1519 0 : })
1520 0 : .await;
1521 0 : let raw = kv.get(key.clone()).await.expect("get").expect("entry");
1522 0 : let text = String::from_utf8_lossy(&raw).into_owned();
1523 0 : assert!(
1524 0 : !text.contains("kv-token") && !text.contains("pa55word"),
1525 : "plaintext credential in the KV: {text}"
1526 : );
1527 0 : assert!(text.contains(seal::PREFIX), "{text}");
1528 :
1529 0 : let entry: serde_json::Value = serde_json::from_slice(&raw).expect("json");
1530 0 : let mirrored = |key: Option<&SecretKey>| {
1531 0 : let m = antares_api::mirror::SubMirror::default();
1532 0 : let mut v = entry.clone();
1533 0 : open_sub_delta(key, &mut v);
1534 0 : apply_delta(&m, &v);
1535 0 : m.docs(tenant.as_str())
1536 0 : .into_iter()
1537 0 : .map(|d| (*d).clone())
1538 0 : .collect::<Vec<_>>()
1539 0 : };
1540 0 : let own = SecretKey::from_base64(TEST_KEY).expect("key");
1541 0 : let opened = mirrored(Some(&own));
1542 0 : assert_eq!(opened.len(), 1);
1543 0 : assert_eq!(
1544 0 : opened[0]["notification"]["endpoint"],
1545 0 : doc["notification"]["endpoint"]
1546 : );
1547 0 : let other =
1548 0 : SecretKey::from_base64("AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=").expect("key");
1549 0 : assert!(mirrored(None).is_empty(), "mirrored without the key");
1550 0 : assert!(
1551 0 : mirrored(Some(&other)).is_empty(),
1552 : "mirrored under another key"
1553 : );
1554 2 : let _ = kv.purge(key).await;
1555 2 : }
1556 :
1557 : /// A Context Source Registration Subscription's KV entry only wakes the
1558 : /// sweep (`apply_delta` reads no document from it), so it ships none:
1559 : /// neither the read nor the credentials it would carry.
1560 : #[tokio::test(flavor = "multi_thread")]
1561 2 : async fn a_csource_subscription_kv_entry_carries_no_document() {
1562 2 : let Some(url) = nats_url() else { return };
1563 0 : let mut st = AppState::new("antares".into());
1564 0 : let _bus = wire_nats(&mut st, &url, Roles::parse("api").expect("roles"), None)
1565 0 : .await
1566 0 : .expect("wire");
1567 0 : let tenant = TenantId::new("csubnodoc").expect("tenant");
1568 0 : let id = format!("urn:ngsi-ld:Subscription:csubnodoc{}", std::process::id());
1569 0 : let doc = json!({"id": id, "type": "ContextSourceRegistrationSubscription",
1570 0 : "entities": [{"type": "https://x/T"}],
1571 0 : "notification": {"endpoint": {
1572 0 : "uri": "http://localhost:1/n",
1573 0 : "receiverInfo": [{"key": "Authorization", "value": "Bearer test-token"}]}}});
1574 0 : st.store
1575 0 : .create(&tenant, Kind::CSourceSubscription, &id, doc.clone())
1576 0 : .await
1577 0 : .expect("create");
1578 0 : let kv = NatsBus::connect(&url)
1579 0 : .await
1580 0 : .expect("connect")
1581 0 : .subs_kv()
1582 0 : .await
1583 0 : .expect("kv");
1584 0 : let key = kv_key(tenant.as_str(), Kind::CSourceSubscription, &id);
1585 0 : let sync = st.sub_sync.clone().expect("api role installs the sync");
1586 0 : sync(&tenant, Kind::CSourceSubscription, &id, Some(&doc));
1587 0 : wait_until("the entry in the KV", || {
1588 0 : let (kv, key) = (kv.clone(), key.clone());
1589 0 : async move { kv.get(key).await.ok().flatten().is_some() }
1590 0 : })
1591 0 : .await;
1592 0 : let raw = kv.get(key.clone()).await.expect("get").expect("entry");
1593 0 : let _ = kv.purge(key).await;
1594 0 : let entry: serde_json::Value = serde_json::from_slice(&raw).expect("json");
1595 0 : assert_eq!(entry["csub"], true);
1596 0 : assert!(entry["doc"].is_null(), "a document was shipped");
1597 2 : }
1598 :
1599 : /// An entry put before ANTARES_SECRET_KEY was set holds its credentials
1600 : /// in plaintext; the boot pass seals it from the store's row, and the
1601 : /// entry still opens into the same document.
1602 : #[tokio::test(flavor = "multi_thread")]
1603 2 : async fn the_boot_pass_seals_a_plaintext_subscription_kv_entry() {
1604 2 : let Some(url) = nats_url() else { return };
1605 0 : let st = AppState::new("antares".into());
1606 0 : let tenant = TenantId::new("sealbootpass").expect("tenant");
1607 0 : let id = format!("urn:ngsi-ld:Subscription:sealboot{}", std::process::id());
1608 0 : let doc = json!({"id": id, "type": "Subscription",
1609 0 : "entities": [{"type": "https://x/T"}],
1610 0 : "notification": {"endpoint": {
1611 0 : "uri": "http://hook:pa55word@localhost:1/n",
1612 0 : "receiverInfo": [{"key": "Authorization", "value": "Bearer test-token"}]}}});
1613 0 : st.store
1614 0 : .create(&tenant, Kind::Subscription, &id, doc.clone())
1615 0 : .await
1616 0 : .expect("create");
1617 0 : let kv = NatsBus::connect(&url)
1618 0 : .await
1619 0 : .expect("connect")
1620 0 : .subs_kv()
1621 0 : .await
1622 0 : .expect("kv");
1623 0 : let key = kv_key(tenant.as_str(), Kind::Subscription, &id);
1624 0 : let plain = json!({"tenant": tenant.as_str(), "id": id, "doc": doc, "csub": false});
1625 0 : kv.put(
1626 0 : key.clone(),
1627 0 : serde_json::to_vec(&plain).expect("json").into(),
1628 0 : )
1629 0 : .await
1630 0 : .expect("put");
1631 0 : let secret = SecretKey::from_base64(TEST_KEY).expect("key");
1632 0 : let n = seal_kv_entries(&kv, &*st.store, &secret, vec![key.clone()]).await;
1633 0 : let raw = kv.get(key.clone()).await.expect("get").expect("entry");
1634 0 : let _ = kv.purge(key).await;
1635 0 : assert_eq!(n, 1);
1636 0 : let text = String::from_utf8_lossy(&raw).into_owned();
1637 0 : assert!(
1638 0 : !text.contains("test-token") && !text.contains("pa55word"),
1639 : "plaintext credential left in the KV entry"
1640 : );
1641 0 : let mut entry: serde_json::Value = serde_json::from_slice(&raw).expect("json");
1642 0 : open_sub_delta(Some(&secret), &mut entry);
1643 0 : assert_eq!(
1644 2 : entry["doc"]["notification"]["endpoint"],
1645 2 : doc["notification"]["endpoint"]
1646 2 : );
1647 2 : }
1648 :
1649 : /// The registration mirror converges on what the store holds. Each
1650 : /// write's delta was its own task publishing the document captured at
1651 : /// write time, so one that landed late (a retried publish, or a task
1652 : /// scheduled after a newer write's) left every api pod's mirror holding
1653 : /// the older registration, and the federation path reads that mirror as
1654 : /// the truth until a restart re-hydrates it.
1655 : #[tokio::test(flavor = "multi_thread")]
1656 2 : async fn a_late_registration_sync_never_overwrites_a_newer_write() {
1657 : use antares_store::CurrentStateDriverExt;
1658 2 : let Some(url) = nats_url() else { return };
1659 0 : let mut st = AppState::new("antares".into());
1660 0 : let bus = wire_nats(&mut st, &url, Roles::parse("api").expect("roles"), None)
1661 0 : .await
1662 0 : .expect("wire");
1663 0 : let tenant = TenantId::new(&format!("reglate{}", std::process::id())).expect("tenant");
1664 0 : let id = format!(
1665 : "urn:ngsi-ld:ContextSourceRegistration:late{}",
1666 0 : std::process::id()
1667 : );
1668 0 : let doc = |v: i64| json!({"id": id, "type": "ContextSourceRegistration", "v": v});
1669 0 : st.store
1670 0 : .create(&tenant, Kind::Registration, &id, doc(1))
1671 0 : .await
1672 0 : .expect("create");
1673 0 : st.store
1674 0 : .mutate::<(), ()>(&tenant, Kind::Registration, &id, |d| {
1675 0 : *d = doc(2);
1676 0 : Ok(())
1677 0 : })
1678 0 : .await
1679 0 : .expect("update");
1680 : // a second pod's mirror, fed from the stream
1681 0 : let mirror = Arc::new(DocMirror::default());
1682 0 : let watcher = tokio::spawn(watch_registry(
1683 0 : bus.clone(),
1684 0 : bus.consume_registry_broadcast().await.expect("consumer"),
1685 0 : mirror.clone(),
1686 0 : st.store.clone(),
1687 0 : st.egress.clone(),
1688 : ));
1689 0 : let held = || {
1690 0 : let (m, t) = (mirror.clone(), tenant.as_str().to_owned());
1691 0 : async move {
1692 0 : m.docs(&t)
1693 0 : .iter()
1694 0 : .find_map(|d| d.get("v").and_then(serde_json::Value::as_i64))
1695 0 : }
1696 0 : };
1697 0 : let sync = st.reg_sync.clone().expect("api role installs the sync");
1698 : // the newer write's delta lands first ...
1699 0 : sync(&tenant, &id, Some(&doc(2)));
1700 0 : wait_until("the newer write in the mirror", || {
1701 0 : let v = held();
1702 0 : async move { v.await == Some(2) }
1703 0 : })
1704 0 : .await;
1705 : // ... and the older write's delta after it
1706 0 : sync(&tenant, &id, Some(&doc(1)));
1707 0 : tokio::time::sleep(std::time::Duration::from_millis(
1708 0 : 1_000 * antares_api::state::slow_factor(),
1709 0 : ))
1710 0 : .await;
1711 0 : assert_eq!(
1712 0 : held().await,
1713 : Some(2),
1714 : "the mirror went back to the registration an older write captured"
1715 : );
1716 0 : watcher.abort();
1717 2 : }
1718 : }
|
