Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Subscription matching + HTTP notification delivery (5.8.6, 5.3.1).
3 : //!
4 : //! Change detection: the store's change hook feeds every entity write here as
5 : //! a (before, after) pair; attribute-level changes are derived by diffing —
6 : //! one hook point instead of one call per write handler.
7 : //!
8 : //! Candidate lookup is index-shaped. `SubMirror` keeps inverted
9 : //! (tenant, type) and (tenant, watched-attr) maps next to the docs, so one
10 : //! change evaluates only the subscriptions that could possibly fire — never
11 : //! a scan over all of a tenant's subscriptions. Subscriptions the index
12 : //! cannot classify exactly (4.17 type-selection expressions) fall into a
13 : //! `broad` bucket that is always evaluated: the index may over-select,
14 : //! never under-select. Full evaluation (selector/q/geo/scope/triggers)
15 : //! stays the truth for every candidate.
16 :
17 : use crate::mirror::{index_keys, Change, Keys, Mirror, SubMirror, CSUB_SWEEP_BACKSTOP_MS};
18 : use crate::negotiate::{inject_context, link_header_value};
19 : use crate::state::{now_iso, AppState};
20 : use antares_jsonld::Context;
21 : use antares_model::{dt_key, TenantId};
22 : use antares_store::CurrentStateDriverExt;
23 : use antares_store::Kind;
24 : use antares_store::{TemporalEvent, TemporalOp};
25 : use serde_json::{json, Map, Value};
26 : use std::collections::HashMap;
27 : use std::sync::Arc;
28 :
29 : const DEFAULT_TRIGGERS: &[&str] = &["attributeCreated", "attributeUpdated"];
30 : /// Depth of the change→matcher queue, the same ring size the local bus uses.
31 : /// Held with the other published ceilings so `/q/health` reports it.
32 : use crate::bounds::{CHANGE_QUEUE, DELIVERY_WIDTH, DELIVERY_WIDTH_PER_TENANT};
33 : static CHANGES_DROPPED: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
34 : static TASK_PANICS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
35 : /// Hand a batch to the matcher queue: counted as pending on acceptance,
36 : /// counted as dropped when the queue is full. Not the durable outbox
37 : /// (`antares_sql::store::pg::outbox::enqueue`), which is a row inside the
38 : /// caller's transaction: this ring lives in the process and a full one
39 : /// drops, which is why the drop is counted.
40 6446 : fn queue_for_matching(
41 6446 : tx: &tokio::sync::mpsc::Sender<Vec<Change>>,
42 6446 : pending: &std::sync::atomic::AtomicUsize,
43 6446 : changes: Vec<Change>,
44 6446 : ) {
45 6446 : if tx.try_send(changes).is_ok() {
46 6442 : pending.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
47 6442 : } else {
48 4 : note_drop();
49 4 : }
50 6446 : }
51 :
52 : /// Changes dropped because the matcher queue was full, since process start.
53 4202 : pub(crate) fn changes_dropped() -> u64 {
54 4202 : CHANGES_DROPPED.load(std::sync::atomic::Ordering::Relaxed)
55 4202 : }
56 :
57 : /// Panics absorbed at the notification-task boundary, since process start.
58 90 : pub(crate) fn task_panics() -> u64 {
59 90 : TASK_PANICS.load(std::sync::atomic::Ordering::Relaxed)
60 90 : }
61 :
62 : /// Entity-level members that are NOT Attributes. Table 5.2.12-1 scopes
63 : /// watchedAttributes to "Properties or Relationships", so the entity's own
64 : /// system members (including 4.22 `expiresAt` and `deletedAt`) must never be
65 : /// diffed as attribute-level changes.
66 : #[derive(Clone, Copy, PartialEq, Debug)]
67 : enum ChangeClass {
68 : Created,
69 : Updated,
70 : Deleted,
71 : }
72 :
73 : /// Where the matcher reads candidates from: the indexed mirror (both bus
74 : /// modes wire one), with the store scan only as the never-wired
75 : /// fallback so a missing mirror degrades to correct-but-slow.
76 2373 : async fn subs_for(
77 2373 : st: &AppState,
78 2373 : tenant: &TenantId,
79 2373 : types: &[&str],
80 2373 : changed: &[&str],
81 2373 : ) -> Vec<Arc<Value>> {
82 2373 : match &st.sub_mirror {
83 2373 : Some(m) => m.candidates(tenant.as_str(), types, changed),
84 : // The scan is the fallback, so its own failure may not be read as
85 : // "this tenant has no subscriptions" either: that is the silence
86 : // the mirror seed refuses to install.
87 0 : None => match st.store.list(tenant, Kind::Subscription).await {
88 0 : Ok(subs) => subs.into_iter().map(Arc::new).collect(),
89 0 : Err(e) => {
90 0 : tracing::error!(
91 : "subscription scan failed for tenant {}: {e}; no candidate matched this change",
92 0 : tenant.as_str()
93 : );
94 0 : Vec::new()
95 : }
96 : },
97 : }
98 2373 : }
99 :
100 : /// Fill a mirror from the store, or say why it could not be filled.
101 : ///
102 : /// Every document of every tenant has to be in the mirror before it is
103 : /// installed. `CurrentStateDriver::subscription_tenants` states the rule for
104 : /// the data path — "A SUBSET is a silent outage: a tenant missing here never
105 : /// fires a periodic notification and never reaches the mirror" — and an
106 : /// error absorbed into an empty list is the same subset by another route.
107 : /// A connection failure at startup refuses it, so this is reachable rather
108 : /// than theoretical.
109 : ///
110 : /// Paged, not `list`: 5.5.6 licenses TooManyResults for "a query operation
111 : /// … producing so many results that can potentially exhaust client or
112 : /// server resources", and the seed is not one — it must see every
113 : /// document of every tenant or it is the silent outage above. Reading it
114 : /// through the ceiling `list` carries for client queries made one tenant's
115 : /// stored volume decide whether OTHER tenants are matched at all.
116 : ///
117 : /// One function for every mirror and both bus modes. `bus=local` seeds the
118 : /// subscription mirror here; `bus=nats` seeds a subscription mirror and a
119 : /// registration mirror, and re-seeds the registration one after a consumer
120 : /// gap. Those were a second copy of this walk that kept the ceiling and
121 : /// swallowed the error into an empty list, which is how a rule fixed in one
122 : /// place stayed broken in the others.
123 : ///
124 : /// The domain is `subscription_tenants`, whose contract covers every kind a
125 : /// mirror is built from. A tenant holding nothing of this kind costs one
126 : /// empty page.
127 576 : pub async fn seed_mirror(
128 576 : store: &dyn antares_store::CurrentStateDriver,
129 576 : mirror: &dyn Mirror,
130 576 : kind: Kind,
131 576 : ) -> Result<(), antares_model::NgsiError> {
132 576 : for tenant_str in store.subscription_tenants().await? {
133 : // The store's own enumeration, so the grammar is all that is left to
134 : // check: 5.5.15 permits a Subscription inside a Snapshot, and its
135 : // tenant is the synthetic one no client may name.
136 135 : let tenant = TenantId::new_internal(&tenant_str)?;
137 135 : let mut after: Option<String> = None;
138 : loop {
139 135 : let page = store
140 135 : .list_page(&tenant, kind, after.as_deref(), SEED_PAGE)
141 135 : .await?;
142 135 : let short = page.len() < SEED_PAGE;
143 135 : let before = after.clone();
144 135 : for doc in page {
145 56 : if let Some(id) = doc.get("id").and_then(Value::as_str) {
146 56 : let id = id.to_owned();
147 56 : after = Some(id.clone());
148 56 : mirror.apply(&tenant_str, &id, Some(doc));
149 56 : }
150 : }
151 : // A short page is the end. A cursor that did not move is also the
152 : // end, and it is the load-bearing half: only a document carrying
153 : // an `id` can advance it, so a full page without one would
154 : // otherwise re-read the same page forever. No write path stores
155 : // such a document — which is exactly why the loop may not depend
156 : // on that staying true.
157 135 : if short || after == before {
158 135 : break;
159 0 : }
160 : }
161 : }
162 576 : Ok(())
163 576 : }
164 :
165 : /// Documents per mirror-seed page: the peak transient allocation of the
166 : /// walk, paid once per tenant at startup.
167 : const SEED_PAGE: usize = 1_000;
168 :
169 : /// Install the subscription mirror, the store change hook that feeds the
170 : /// matcher, and the background tasks the pipeline needs. Called once at
171 : /// startup, by `crate::wire`.
172 560 : pub(crate) async fn wire_matcher(state: &mut AppState) {
173 : // bus=local: the same indexed mirror the nats wiring builds, fed
174 : // synchronously by the CUD hook — the matcher never rescans the store.
175 560 : let mirror = Arc::new(SubMirror::default());
176 560 : match seed_mirror(state.store.as_ref(), mirror.as_ref(), Kind::Subscription).await {
177 560 : Ok(()) => state.sub_mirror = Some(mirror.clone()),
178 : // Not installed, so `subs_for` takes the store scan it documents as
179 : // the missing-mirror fallback. Installing what the seed managed to
180 : // read would be the one outcome that is neither correct nor slow:
181 : // the matcher reads candidates from the mirror alone, so a
182 : // subscription absent from it never fires again in this process.
183 0 : Err(e) => tracing::error!(
184 : "subscription mirror seed failed ({e}); \
185 : matching falls back to a store scan per change"
186 : ),
187 : }
188 560 : let m = mirror.clone();
189 560 : state.sub_sync = Some(Arc::new(
190 306 : move |tenant: &TenantId, kind: Kind, id: &str, doc: Option<&Value>| match kind {
191 2 : Kind::CSourceSubscription => m.csub_written(),
192 304 : _ => m.apply(tenant.as_str(), id, doc.cloned()),
193 306 : },
194 : ));
195 :
196 : // The queue carries whole before+after payloads and is drained one
197 : // inline delivery at a time, so behind one slow subscriber an unbounded
198 : // queue grows until the process dies. Bounded instead: a full queue drops
199 : // the change and counts it.
200 560 : let (tx, mut rx) = tokio::sync::mpsc::channel::<Vec<Change>>(CHANGE_QUEUE);
201 560 : let flush_tx = tx.clone();
202 560 : let flush_pending = state.pending_changes.clone();
203 6248 : state.change_flush = Some(Arc::new(move |changes: Vec<Change>| {
204 6248 : queue_for_matching(&flush_tx, &flush_pending, changes)
205 6248 : }));
206 : // Temporal auto-recording runs SYNCHRONOUSLY on the hook (read-your-writes:
207 : // the ETSI suite queries history immediately after a write); the matcher
208 : // work is handed to the async task below. One choke point for every write.
209 560 : let st_rec = state.clone();
210 560 : let hook_pending = state.pending_changes.clone();
211 560 : state.store.set_change_hook(Arc::new(
212 : move |tenant: &TenantId,
213 : before: Option<Value>,
214 : after: Option<Value>|
215 6462 : -> antares_store::HookFuture<'_> {
216 6462 : let st_rec = st_rec.clone();
217 6462 : let tx = tx.clone();
218 6462 : let hook_pending = hook_pending.clone();
219 6462 : Box::pin(async move {
220 6462 : record_temporal_change(&st_rec, tenant, before.as_ref(), after.as_ref()).await;
221 : // inside a request the change rides the request's buffer and
222 : // reaches the matcher with the rest of that request's changes
223 198 : let Some(change) =
224 6462 : crate::history::buffer_change((tenant.as_str().to_owned(), before, after))
225 : else {
226 6264 : return;
227 : };
228 198 : queue_for_matching(&tx, &hook_pending, vec![change]);
229 6462 : })
230 6462 : },
231 : ));
232 560 : let st = state.clone();
233 560 : let pending = state.pending_changes.clone();
234 560 : crate::spawn_loop(async move {
235 2813 : while let Some(mut batch) = rx.recv().await {
236 2326 : let mut taken = 1;
237 : // everything already queued behind it rides the same pass
238 2339 : while batch.len() < CHANGE_BATCH {
239 2339 : match rx.try_recv() {
240 13 : Ok(c) => {
241 13 : batch.extend(c);
242 13 : taken += 1;
243 13 : }
244 2326 : Err(_) => break,
245 : }
246 : }
247 2326 : let st = st.clone();
248 2326 : guarded(async move { process_changes(&st, batch).await }).await;
249 2277 : pending.fetch_sub(taken, std::sync::atomic::Ordering::SeqCst);
250 : }
251 0 : });
252 560 : let st = state.clone();
253 560 : crate::spawn_loop(async move {
254 : loop {
255 : // Tokio's timer natively; the browser's own timer on wasm32
256 : // (tokio time never fires without a reactor there).
257 : #[cfg(not(target_arch = "wasm32"))]
258 3825 : tokio::time::sleep(std::time::Duration::from_millis(500)).await;
259 : #[cfg(target_arch = "wasm32")]
260 : gloo_timers::future::TimeoutFuture::new(500).await;
261 3288 : let st = st.clone();
262 3288 : guarded(async move { interval_tick(&st).await }).await;
263 : }
264 : });
265 560 : }
266 :
267 : /// Run one pipeline step on its own task so a panic inside it cannot end
268 : /// notification delivery for the whole process: the task boundary absorbs
269 : /// the panic, it is counted and logged, and the caller keeps consuming — a
270 : /// later matching change still notifies (5.8.6). The step is awaited, so
271 : /// delivery stays as serial as it was.
272 : #[cfg(not(target_arch = "wasm32"))]
273 5614 : async fn guarded<F>(fut: F)
274 5614 : where
275 5614 : F: std::future::Future<Output = ()>,
276 5614 : {
277 : use futures_util::FutureExt as _;
278 : // Caught here rather than on a spawned task: a task boundary would demand
279 : // Send + 'static of the step, and the interval step holds state that is
280 : // not Sync. Unwinding in place absorbs the panic just as well and keeps
281 : // the step running on this task, so delivery stays exactly as serial.
282 5614 : if std::panic::AssertUnwindSafe(fut)
283 5614 : .catch_unwind()
284 5614 : .await
285 5565 : .is_err()
286 0 : {
287 0 : note_panic();
288 5565 : }
289 5565 : }
290 :
291 : #[cfg(not(target_arch = "wasm32"))]
292 0 : fn note_panic() {
293 0 : TASK_PANICS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
294 0 : metrics::counter!("antares_notification_task_panics_total").increment(1);
295 0 : tracing::error!("notification pipeline task panicked; this change is lost");
296 0 : }
297 :
298 : /// Wasm32 has no task boundary to catch with (single-threaded executor, and
299 : /// the browser profile aborts on panic); the step runs inline.
300 : #[cfg(target_arch = "wasm32")]
301 : async fn guarded<F>(fut: F)
302 : where
303 : F: std::future::Future<Output = ()>,
304 : {
305 : fut.await;
306 : }
307 :
308 : /// Whether two attribute instances are the same once the volatile members
309 : /// — the timestamps the broker itself stamps, and the instance identity —
310 : /// are set aside. Compared in place: this runs per changed attribute per
311 : /// matching subscription per event, and materializing two stripped copies of
312 : /// every JSON tree to throw them away one comparison later is the cost that
313 : /// multiplies by.
314 1471 : fn stable_eq(a: &Value, b: &Value) -> bool {
315 5079 : fn volatile(k: &str) -> bool {
316 5079 : matches!(k, "createdAt" | "modifiedAt" | "instanceId")
317 5079 : }
318 1471 : match (a, b) {
319 424 : (Value::Object(oa), Value::Object(ob)) => {
320 3586 : let count = |o: &Map<String, Value>| o.keys().filter(|k| !volatile(k)).count();
321 : // by lookup rather than by zipped iteration: which order a
322 : // serde_json Map yields is a cargo feature away from changing,
323 : // and features are additive
324 424 : count(oa) == count(ob)
325 396 : && oa
326 396 : .iter()
327 1493 : .filter(|(k, _)| !volatile(k))
328 749 : .all(|(k, va)| ob.get(k).is_some_and(|vb| stable_eq(va, vb)))
329 : }
330 294 : (Value::Array(aa), Value::Array(ab)) => {
331 294 : aa.len() == ab.len() && aa.iter().zip(ab).all(|(x, y)| stable_eq(x, y))
332 : }
333 753 : _ => a == b,
334 : }
335 1471 : }
336 :
337 16738 : fn attr_keys(doc: &Value) -> Vec<String> {
338 16738 : doc.as_object()
339 16738 : .map(|o| {
340 16738 : o.keys()
341 43152 : .filter(|k| !crate::repr::ENTITY_META.contains(&k.as_str()))
342 16738 : .cloned()
343 16738 : .collect()
344 16738 : })
345 16738 : .unwrap_or_default()
346 16738 : }
347 :
348 : /// Per-attribute change classification between two internal docs.
349 8369 : fn diff(before: Option<&Value>, after: Option<&Value>) -> Vec<(String, ChangeClass)> {
350 8369 : let empty = Value::Object(Map::new());
351 8369 : let b = before.unwrap_or(&empty);
352 8369 : let a = after.unwrap_or(&empty);
353 8369 : let mut keys = attr_keys(b);
354 8369 : for k in attr_keys(a) {
355 7296 : if !keys.contains(&k) {
356 7010 : keys.push(k);
357 7010 : }
358 : }
359 8369 : let mut out = Vec::new();
360 8369 : for k in keys {
361 7866 : let bv = b.get(&k);
362 7866 : let av = a.get(&k);
363 7866 : match (bv, av) {
364 7010 : (None, Some(_)) => out.push((k, ChangeClass::Created)),
365 570 : (Some(_), None) => out.push((k, ChangeClass::Deleted)),
366 286 : (Some(x), Some(y)) => {
367 : // instance-level deletion (by datasetId) counts as
368 : // attributeDeleted even when other instances survive (5.8.6)
369 286 : let bx: Vec<&Value> = x.as_array().map(|a| a.iter().collect()).unwrap_or_default();
370 286 : let by: Vec<&Value> = y.as_array().map(|a| a.iter().collect()).unwrap_or_default();
371 286 : let removed = bx
372 286 : .iter()
373 290 : .any(|bi| !by.iter().any(|ai| instance_ds(ai) == instance_ds(bi)));
374 286 : if removed {
375 8 : out.push((k.clone(), ChangeClass::Deleted));
376 8 : let survivors_changed = by.iter().any(|ai| {
377 12 : match bx.iter().find(|bi| instance_ds(bi) == instance_ds(ai)) {
378 8 : Some(bi) => !stable_eq(bi, ai),
379 0 : None => true,
380 : }
381 8 : });
382 8 : if survivors_changed {
383 0 : out.push((k, ChangeClass::Updated));
384 8 : }
385 278 : } else if !stable_eq(x, y) {
386 196 : out.push((k, ChangeClass::Updated));
387 196 : }
388 : }
389 0 : _ => {}
390 : }
391 : }
392 8369 : out
393 8369 : }
394 :
395 : /// Auto-record a current-state change into the temporal representation
396 : /// (5.6.11). Driven by the store's SYNCHRONOUS change hook, which fires inside
397 : /// every entity write (create, update, partial update, merge, replace, batch)
398 : /// for both the memory and postgres stores — so a new write path records
399 : /// without the handler having to remember. The per-handler `mirror_record`
400 : /// this replaced was the forgettable trap that left Partial Attribute Update
401 : /// (5.6.4) and Replace Attribute (5.6.19) silently unrecorded.
402 : ///
403 : /// Append-only, instance-precise: only the instances that are new or changed
404 : /// (by datasetId, ignoring volatile members) are appended, so a multi-instance
405 : /// attribute does not re-record unchanged datasets. Entity and attribute
406 : /// DELETIONS keep their dedicated typed-null mirrors (`mirror_delete_entity` /
407 : /// `mirror_delete_attr`), which the delete handlers still call — their deletion
408 : /// shape is not derivable from a plain append.
409 6462 : pub async fn record_temporal_change(
410 6462 : st: &AppState,
411 6462 : tenant: &TenantId,
412 6462 : before: Option<&Value>,
413 6462 : after: Option<&Value>,
414 6462 : ) {
415 6462 : if !st.record_locally() {
416 0 : return;
417 6462 : }
418 6462 : let Some(after) = after else {
419 474 : return; // entity deletion — handled by mirror_delete_entity
420 : };
421 5988 : let Some(id) = after.get("id").and_then(Value::as_str) else {
422 0 : return;
423 : };
424 5988 : let mut shell = Map::new();
425 29940 : for k in ["id", "type", "createdAt", "modifiedAt", "scope"] {
426 29940 : if let Some(v) = after.get(k) {
427 23866 : shell.insert(k.to_string(), v.clone());
428 23866 : }
429 : }
430 5988 : let shell = Value::Object(shell);
431 5988 : let event = |op, attr: &str, instance| TemporalEvent {
432 5654 : op,
433 5654 : tenant: tenant.clone(),
434 5654 : entity_id: id.to_owned(),
435 5654 : shell: shell.clone(),
436 5654 : attr: attr.to_owned(),
437 5654 : instance,
438 5654 : };
439 : // 4.5.6: the Scope of a Temporal Evolution is represented as a temporal
440 : // Property whose only sub-properties are the non-reified createdAt,
441 : // modifiedAt, deletedAt and observedAt; when it "is updated as the result
442 : // of a change from the Core API, the observedAt sub-Property should be
443 : // set as a copy of the modifiedAt sub-Property".
444 5988 : if before.is_some_and(|b| b.get("scope") != after.get("scope")) {
445 10 : if let Some(scope) = after.get("scope") {
446 2 : let ts = after
447 2 : .get("modifiedAt")
448 2 : .and_then(Value::as_str)
449 2 : .map(String::from)
450 2 : .unwrap_or_else(now_iso);
451 2 : let inst = json!({
452 2 : "type": "Property",
453 2 : "value": scope.clone(),
454 2 : "instanceId": format!("urn:ngsi-ld:Instance:{}", uuid::Uuid::new_v4()),
455 2 : "createdAt": ts, "modifiedAt": ts, "observedAt": ts,
456 : });
457 2 : crate::history::push(st, event(TemporalOp::ScopeChanged, "scope", inst)).await;
458 8 : }
459 5978 : }
460 5988 : for (k, class) in diff(before, Some(after)) {
461 5668 : let op = match class {
462 5546 : ChangeClass::Created => TemporalOp::AttrCreated,
463 98 : ChangeClass::Updated => TemporalOp::AttrModified,
464 24 : ChangeClass::Deleted => continue, // handled by mirror_delete_attr
465 : };
466 5644 : let Some(av) = after.get(&k) else { continue };
467 : // gate 1, value-change: an unchanged instance produces no event
468 5652 : for mut inst in changed_instances(before.and_then(|b| b.get(&k)), av) {
469 5652 : if let Some(o) = inst.as_object_mut() {
470 5652 : let iid = instance_id(id, &k, &*o);
471 5652 : o.entry("instanceId".to_owned())
472 5652 : .or_insert_with(|| Value::String(iid));
473 0 : }
474 5652 : crate::history::push(st, event(op, &k, inst)).await;
475 : }
476 : }
477 6462 : }
478 :
479 : /// 4.5.7: an instance is the Attribute "at a particular point in time",
480 : /// recorded as its observedAt. The id of an observed instance is therefore
481 : /// derived from (entity, attribute, datasetId, observedAt), so a re-send for
482 : /// the same instant lands on the same row — the temporal store's upsert key —
483 : /// and corrects it instead of appending a duplicate. Without observedAt there
484 : /// is no instant to key on: a fresh random id, append-only.
485 : ///
486 : /// The instant is keyed through `dt_key`, not through the stamp as written.
487 : /// 4.6.3 leaves the seconds fraction optional and accepts a comma separator
488 : /// in requests, and the broker stores a DateTime exactly as the client wrote
489 : /// it, so one instant arrives under several spellings; keying the raw text
490 : /// gave each spelling its own instance and left the correction's target in
491 : /// place beside it — the failure 4.5.7 calls severe "in the case of
492 : /// modification or deletion requests for legal reasons".
493 5728 : fn instance_id(entity: &str, attr: &str, inst: &serde_json::Map<String, Value>) -> String {
494 5728 : let u = match inst.get("observedAt").and_then(Value::as_str) {
495 312 : Some(at) => {
496 312 : let ds = inst
497 312 : .get("datasetId")
498 312 : .and_then(Value::as_str)
499 312 : .unwrap_or("@none");
500 312 : uuid::Uuid::new_v5(
501 312 : &uuid::Uuid::NAMESPACE_URL,
502 312 : format!("{entity}\n{attr}\n{ds}\n{}", dt_key(at)).as_bytes(),
503 : )
504 : }
505 5416 : None => uuid::Uuid::new_v4(),
506 : };
507 5728 : format!("urn:ngsi-ld:Instance:{u}")
508 5728 : }
509 :
510 : /// The instances in `after` that are new or changed vs `before` — matched by
511 : /// datasetId, compared via `stable_eq` (volatile members ignored). A newly
512 : /// created attribute (`before` None) contributes every instance.
513 5644 : fn changed_instances(before: Option<&Value>, after: &Value) -> Vec<Value> {
514 5644 : let before_arr: Vec<&Value> = before
515 5644 : .and_then(Value::as_array)
516 5644 : .map(|a| a.iter().collect())
517 5644 : .unwrap_or_default();
518 5644 : after
519 5644 : .as_array()
520 5644 : .cloned()
521 5644 : .unwrap_or_default()
522 5644 : .into_iter()
523 5656 : .filter(|ai| {
524 5656 : match before_arr
525 5656 : .iter()
526 5656 : .find(|bi| instance_ds(bi) == instance_ds(ai))
527 : {
528 5558 : None => true,
529 98 : Some(bi) => !stable_eq(bi, ai),
530 : }
531 5656 : })
532 5644 : .collect()
533 5644 : }
534 :
535 : /// One string-valued member of a stored Subscription, absent or of another
536 : /// JSON type reading the same as absent.
537 1546 : fn sub_str<'a>(sub: &'a Value, key: &str) -> Option<&'a str> {
538 1546 : sub.get(key).and_then(Value::as_str)
539 1546 : }
540 :
541 : /// 4.9 EXAMPLE 13/14: linked-entity q terms (`attr{path}`) resolve through
542 : /// the local store, same tenant. The evaluator is synchronous and the store
543 : /// is not, so `eval` runs against a cache: a URI the cache misses is
544 : /// recorded, reads as absent for that pass, and is fetched before the next
545 : /// one. The first pass that misses nothing is the answer. Every miss of a
546 : /// pass is fetched together, so the pass count is the depth of the link
547 : /// chain rather than the number of entities it touches.
548 862 : pub(crate) async fn linked_eval<F>(st: &AppState, tenant: &TenantId, mut eval: F) -> bool
549 862 : where
550 862 : F: for<'a> FnMut(antares_ql::eval::EntityLookup<'a>) -> bool,
551 862 : {
552 862 : let mut cache: HashMap<String, Option<Value>> = HashMap::new();
553 : // ponytail: the evaluator's own lookup budget bounds how many distinct
554 : // URIs one expression can name, so it bounds the passes too; a chain
555 : // deeper than that reads as unresolved, which is what exhausting the
556 : // budget inside the evaluator already does.
557 862 : for _ in 0..antares_ql::eval::MAX_Q_LINK_LOOKUPS {
558 874 : let missed = std::cell::RefCell::new(Vec::new());
559 874 : let out = eval(&|uri: &str| match cache.get(uri) {
560 12 : Some(doc) => doc.clone(),
561 : None => {
562 12 : missed.borrow_mut().push(uri.to_owned());
563 12 : None
564 : }
565 24 : });
566 874 : let missed = missed.into_inner();
567 874 : if missed.is_empty() {
568 862 : return out;
569 12 : }
570 12 : for uri in missed {
571 12 : let doc = st
572 12 : .store
573 12 : .get(tenant, Kind::Entity, &uri)
574 12 : .await
575 12 : .ok()
576 12 : .flatten();
577 12 : cache.insert(uri, doc);
578 : }
579 : }
580 0 : eval(&|uri: &str| cache.get(uri).cloned().flatten())
581 862 : }
582 :
583 : pub(crate) use antares_matcher::{
584 : conditions_match, geo_params, is_active, selector_match, throttled,
585 : };
586 :
587 : /// The @context governing a subscription's notifications (5.8.6): the
588 : /// jsonldContext member if set, else the @context of the creating request.
589 1155 : pub(crate) async fn sub_context(st: &AppState, tenant: &TenantId, sub: &Value) -> Arc<Context> {
590 : // Borrowed: the resolver only reads it, and this is on the per-candidate
591 : // path, where cloning an inline @context copied the whole document.
592 1155 : let source = sub.get("jsonldContext").or_else(|| sub.get("__context"));
593 671 : match source {
594 : // 5.5.10: the Subscription belongs to one Tenant, so the @context it
595 : // names resolves within that Tenant — a Hosted @context another Tenant
596 : // stored (5.13.1) is not in scope here and falls back to the core
597 : // context rather than compacting this Notification against it.
598 671 : Some(v) if !v.is_null() => st
599 671 : .loader
600 671 : .resolve_quiet_for(tenant, v)
601 671 : .await
602 671 : .unwrap_or_else(|_| st.loader.core()),
603 484 : _ => st.loader.core(),
604 : }
605 1155 : }
606 :
607 : /// A resolved `@context` per Tenant, then per the `@context` URL the
608 : /// Subscription names. Keyed by Tenant first: 5.5.10 scopes a resolution to
609 : /// one Tenant, so one Tenant's resolved context is never reachable under
610 : /// another's identifier even where both name the same URL.
611 : type CtxMemo = HashMap<String, HashMap<String, Arc<Context>>>;
612 :
613 : /// `sub_context` over one drain's memo. Resolving is a cache lookup per call,
614 : /// and this runs once per candidate subscription per change: a drain of
615 : /// `CHANGE_BATCH` changes resolves the same handful of `@context` URLs
616 : /// thousands of times over. The memo lives for the one drain that owns it, so
617 : /// an `@context` a request replaces or deletes meanwhile is picked up by the
618 : /// next drain. A Subscription naming an inline `@context` rather than a URL
619 : /// resolves as before — it has no cheap identity to key on.
620 430 : async fn sub_context_memo(
621 430 : st: &AppState,
622 430 : tenant: &TenantId,
623 430 : tenant_str: &str,
624 430 : sub: &Value,
625 430 : memo: &mut CtxMemo,
626 430 : ) -> Arc<Context> {
627 430 : let url = sub
628 430 : .get("jsonldContext")
629 430 : .or_else(|| sub.get("__context"))
630 430 : .and_then(Value::as_str);
631 430 : let Some(url) = url else {
632 166 : return sub_context(st, tenant, sub).await;
633 : };
634 264 : if let Some(hit) = memo.get(tenant_str).and_then(|by_url| by_url.get(url)) {
635 16 : return Arc::clone(hit);
636 248 : }
637 248 : let ctx = sub_context(st, tenant, sub).await;
638 248 : memo.entry(tenant_str.to_owned())
639 248 : .or_default()
640 248 : .insert(url.to_owned(), Arc::clone(&ctx));
641 248 : ctx
642 430 : }
643 :
644 : /// Per-type NGSI-LD-null member and its showChanges previous-member (5.8.6).
645 76 : fn null_members(atype: &str) -> (&'static str, Value, &'static str) {
646 76 : let null = Value::String("urn:ngsi-ld:null".into());
647 76 : match atype {
648 76 : "Relationship" => ("object", null, "previousObject"),
649 68 : "LanguageProperty" => (
650 24 : "languageMap",
651 24 : json!({"@none": "urn:ngsi-ld:null"}),
652 24 : "previousLanguageMap",
653 24 : ),
654 44 : "JsonProperty" => ("json", null, "previousJson"),
655 40 : "VocabProperty" => ("vocab", null, "previousVocab"),
656 36 : _ => ("value", null, "previousValue"),
657 : }
658 76 : }
659 :
660 0 : fn current_member(atype: &str) -> &'static str {
661 0 : null_members(atype).0
662 0 : }
663 :
664 : /// The deletion tombstone for one former attribute instance (5.8.6 payload
665 : /// forms: typed null + optional datasetId / sysAttrs stamps / previous value).
666 76 : fn tombstone(before_inst: &Value, sys: bool, show: bool, now: &str) -> Value {
667 76 : let atype = before_inst
668 76 : .get("type")
669 76 : .and_then(Value::as_str)
670 76 : .unwrap_or("Property");
671 76 : let (member, null_val, prev_member) = null_members(atype);
672 76 : let mut m = Map::new();
673 76 : m.insert("type".into(), Value::String(atype.to_owned()));
674 76 : m.insert(member.into(), null_val);
675 76 : if let Some(ds) = before_inst.get("datasetId") {
676 8 : m.insert("datasetId".into(), ds.clone());
677 68 : }
678 76 : if sys {
679 24 : for k in ["createdAt", "modifiedAt"] {
680 24 : if let Some(v) = before_inst.get(k) {
681 4 : m.insert(k.into(), v.clone());
682 20 : }
683 : }
684 12 : m.insert("deletedAt".into(), Value::String(now.to_owned()));
685 64 : }
686 76 : if show {
687 12 : if let Some(prev) = before_inst.get(member) {
688 12 : m.insert(prev_member.into(), prev.clone());
689 12 : }
690 64 : }
691 76 : Value::Object(m)
692 76 : }
693 :
694 932 : fn instance_ds(inst: &Value) -> Option<&str> {
695 932 : inst.get("datasetId").and_then(Value::as_str)
696 932 : }
697 :
698 : /// Instances of `before[attr]` that no longer exist in `after[attr]`
699 : /// (matched by datasetId — instance-level deletions count, 046_22_06).
700 76 : fn deleted_instances<'a>(before: &'a Value, after: Option<&Value>, attr: &str) -> Vec<&'a Value> {
701 76 : let b: Vec<&Value> = before
702 76 : .get(attr)
703 76 : .and_then(Value::as_array)
704 76 : .map(|a| a.iter().collect())
705 76 : .unwrap_or_default();
706 76 : let a: Vec<&Value> = after
707 76 : .and_then(|d| d.get(attr))
708 76 : .and_then(Value::as_array)
709 76 : .map(|x| x.iter().collect())
710 76 : .unwrap_or_default();
711 76 : b.into_iter()
712 96 : .filter(|bi| !a.iter().any(|ai| instance_ds(ai) == instance_ds(bi)))
713 76 : .collect()
714 76 : }
715 :
716 : pub(crate) struct NotifShape {
717 : pub(crate) repr: crate::repr::Repr,
718 : pub(crate) show_changes: bool,
719 : pub(crate) join: Option<(String, usize)>,
720 : }
721 :
722 778 : pub(crate) fn notif_shape(sub: &Value, ctx: &Context) -> NotifShape {
723 778 : let n = sub.get("notification").and_then(Value::as_object);
724 778 : let format = n
725 778 : .and_then(|n| n.get("format"))
726 778 : .and_then(Value::as_str)
727 778 : .unwrap_or("normalized");
728 778 : let mut repr = crate::repr::Repr {
729 778 : sys_attrs: n
730 778 : .and_then(|n| n.get("sysAttrs"))
731 778 : .and_then(Value::as_bool)
732 778 : .unwrap_or(false),
733 778 : key_values: matches!(format, "keyValues" | "simplified"),
734 778 : concise: format == "concise",
735 778 : ..Default::default()
736 : };
737 2334 : let names = |key: &str| -> Option<Vec<String>> {
738 2334 : n.and_then(|n| n.get(key))
739 2334 : .and_then(Value::as_array)
740 2334 : .map(|a| {
741 12 : a.iter()
742 12 : .filter_map(Value::as_str)
743 12 : .map(str::to_owned)
744 12 : .collect()
745 12 : })
746 2334 : };
747 778 : if let Some(attrs) = names("attributes") {
748 0 : repr.attrs = Some(attrs.iter().map(|a| ctx.expand_key(a)).collect());
749 778 : }
750 : // 4.21 NGSI-LD Attribute Projection Language: pick/omit values are
751 : // projection language strings, which Table 5.2.14.1-1 requires for the
752 : // notification members too ("a valid attribute projection language string
753 : // as per clause 4.21"). Each term may carry a LinkedEntityTerm
754 : // (`ProjectionTerm = AttrName *1(LinkedEntityTerm)`), which is what
755 : // constrains an Attribute inside a Linked Entity retrieved by join. Building
756 : // the nodes by hand here degraded `refDevice{type}` to a literal Attribute
757 : // name matching nothing, so the term was dropped instead of applied.
758 : // A term that fails to parse is dropped rather than kept flat: for pick that
759 : // withholds the Attribute, which is the safe direction.
760 778 : let nodes = |list: Vec<String>| -> Vec<crate::repr::ProjNode> {
761 12 : list.iter()
762 34 : .filter_map(|term| crate::repr::parse_projection(term, ctx).ok())
763 12 : .flatten()
764 12 : .collect()
765 12 : };
766 778 : if let Some(pick) = names("pick") {
767 10 : repr.pick = Some(nodes(pick));
768 768 : }
769 778 : if let Some(omit) = names("omit") {
770 2 : repr.omit = Some(nodes(omit));
771 776 : }
772 778 : if let Some(ds) = sub.get("datasetId").and_then(Value::as_array) {
773 0 : repr.dataset_id = Some(
774 0 : ds.iter()
775 0 : .filter_map(Value::as_str)
776 0 : .map(str::to_owned)
777 0 : .collect(),
778 0 : );
779 778 : }
780 778 : let join = n
781 778 : .and_then(|n| n.get("join"))
782 778 : .and_then(Value::as_str)
783 778 : .filter(|j| *j == "inline" || *j == "flat")
784 778 : .map(|j| {
785 : // the stored member is bounded at creation, but a Subscription
786 : // written before that bound existed is still on disk, so the
787 : // traversal takes the ceiling from here too
788 12 : let level = (n
789 12 : .and_then(|n| n.get("joinLevel"))
790 12 : .and_then(Value::as_u64)
791 12 : .unwrap_or(1) as usize)
792 12 : .min(crate::bounds::MAX_JOIN_LEVEL);
793 12 : (j.to_owned(), level)
794 12 : });
795 : NotifShape {
796 778 : repr,
797 778 : show_changes: n
798 778 : .and_then(|n| n.get("showChanges"))
799 778 : .and_then(Value::as_bool)
800 778 : .unwrap_or(false),
801 778 : join,
802 : }
803 778 : }
804 :
805 : /// Build the notification `data` array for one change and one subscription.
806 : #[allow(clippy::too_many_arguments)] // one param per 5.8.6 payload input
807 762 : async fn build_data(
808 762 : st: &AppState,
809 762 : tenant: &TenantId,
810 762 : sub: &Value,
811 762 : ctx: &Context,
812 762 : before: Option<&Value>,
813 762 : after: Option<&Value>,
814 762 : relevant_deleted_attrs: &[String],
815 762 : entity_deleted_fired: bool,
816 762 : now: &str,
817 762 : ) -> Vec<Value> {
818 762 : let shape = notif_shape(sub, ctx);
819 762 : let sys = shape.repr.sys_attrs;
820 762 : let show = shape.show_changes;
821 762 : let internal = match after {
822 762 : Some(a) => {
823 762 : let mut doc = a.clone();
824 762 : if show {
825 : // previous* on changed instances (046_31..33). An entity off
826 : // the change feed is an object; one that is not carries
827 : // nothing to decorate and travels on unchanged.
828 8 : if let (Some(b), Some(obj)) = (before, doc.as_object_mut()) {
829 16 : for (k, v) in obj.iter_mut() {
830 16 : if crate::repr::ENTITY_META.contains(&k.as_str()) {
831 16 : continue;
832 0 : }
833 0 : let Some(arr) = v.as_array_mut() else {
834 0 : continue;
835 : };
836 0 : for inst in arr {
837 0 : let Some(bi) = b.get(k).and_then(Value::as_array).and_then(|ba| {
838 0 : ba.iter().find(|x| instance_ds(x) == instance_ds(inst))
839 0 : }) else {
840 0 : continue;
841 : };
842 0 : let atype = inst
843 0 : .get("type")
844 0 : .and_then(Value::as_str)
845 0 : .unwrap_or("Property");
846 0 : let member = current_member(atype);
847 0 : let (_, _, prev_member) = null_members(atype);
848 0 : if bi.get(member) != inst.get(member) {
849 0 : if let (Some(pv), Some(io)) =
850 0 : (bi.get(member).cloned(), inst.as_object_mut())
851 0 : {
852 0 : io.insert(prev_member.into(), pv);
853 0 : }
854 0 : }
855 : }
856 : }
857 0 : }
858 754 : }
859 : // deletion tombstones appended beside surviving instances
860 762 : if let Some(b) = before {
861 76 : for attr in relevant_deleted_attrs {
862 76 : let gone = deleted_instances(b, Some(&doc), attr);
863 76 : if gone.is_empty() {
864 0 : continue;
865 76 : }
866 76 : let attr_absent = doc.get(attr).is_none();
867 76 : let tss: Vec<Value> = if attr_absent {
868 : // whole attribute deleted at once: ONE tombstone,
869 : // no datasetId (046_22_08)
870 68 : let base = gone
871 68 : .iter()
872 68 : .find(|i| instance_ds(i).is_none())
873 68 : .unwrap_or(&gone[0]);
874 68 : let mut ts = tombstone(base, sys, show, now);
875 68 : if let Some(o) = ts.as_object_mut() {
876 68 : o.remove("datasetId");
877 68 : }
878 68 : vec![ts]
879 : } else {
880 8 : gone.iter()
881 8 : .map(|di| tombstone(di, sys, show, now))
882 8 : .collect()
883 : };
884 76 : let Some(target) = doc.as_object_mut() else {
885 0 : continue;
886 : };
887 76 : if let Some(arr) = target
888 76 : .entry(attr.clone())
889 76 : .or_insert_with(|| Value::Array(vec![]))
890 76 : .as_array_mut()
891 76 : {
892 76 : arr.extend(tss);
893 76 : }
894 : }
895 700 : }
896 762 : doc
897 : }
898 : None => {
899 : // entity deleted: tombstone entity (046_21) + per-trigger attrs.
900 : // The caller returns early unless one of before/after is there,
901 : // so this arm has a before; without one there is nothing to
902 : // describe and nothing to notify about.
903 0 : let Some(b) = before else {
904 0 : return Vec::new();
905 : };
906 0 : let mut m = Map::new();
907 0 : for k in ["id", "type"] {
908 0 : if let Some(v) = b.get(k) {
909 0 : m.insert(k.into(), v.clone());
910 0 : }
911 : }
912 0 : if sys {
913 0 : for k in ["createdAt", "modifiedAt"] {
914 0 : if let Some(v) = b.get(k) {
915 0 : m.insert(k.into(), v.clone());
916 0 : }
917 : }
918 0 : }
919 0 : m.insert("deletedAt".into(), Value::String(now.to_owned()));
920 0 : let attrs: Vec<String> = if entity_deleted_fired && show {
921 0 : attr_keys(b) // showChanges: every attribute, tombstoned (046_37)
922 : } else {
923 0 : relevant_deleted_attrs.to_vec()
924 : };
925 0 : for attr in attrs {
926 0 : let insts: Vec<Value> = b
927 0 : .get(&attr)
928 0 : .and_then(Value::as_array)
929 0 : .map(|a| a.iter().map(|i| tombstone(i, sys, show, now)).collect())
930 0 : .unwrap_or_default();
931 0 : if !insts.is_empty() {
932 0 : m.insert(attr, Value::Array(insts));
933 0 : }
934 : }
935 0 : Value::Object(m)
936 : }
937 : };
938 762 : let shaped = crate::repr::apply(&internal, &shape.repr);
939 762 : let mut main = crate::repr::compact_for(&shape.repr, &shaped, ctx);
940 762 : let mut data = Vec::new();
941 4 : match &shape.join {
942 12 : Some((mode, level)) if mode == "inline" => {
943 8 : crate::repr::inline_join(st, tenant, ctx, &shape.repr, &mut main, *level).await;
944 8 : data.push(main);
945 : }
946 4 : Some((mode, level)) if mode == "flat" => {
947 4 : let main_id = internal.get("id").and_then(Value::as_str).unwrap_or("");
948 4 : let mut linked = std::collections::BTreeMap::new();
949 4 : crate::repr::collect_flat(st, tenant, &shape.repr, &internal, *level, &mut linked)
950 4 : .await;
951 4 : data.push(main);
952 4 : for (id, (ldoc, lrepr)) in linked {
953 4 : if id != main_id {
954 4 : data.push(crate::repr::compact_for(
955 4 : &lrepr,
956 4 : &crate::repr::apply(&ldoc, &lrepr),
957 4 : ctx,
958 4 : ));
959 4 : }
960 : }
961 : }
962 750 : _ => data.push(main),
963 : }
964 762 : data
965 762 : }
966 :
967 : /// The notification triggers of one subscription, in the form the matcher
968 : /// compares against (Table 5.2.12-1): absent means the default combination
969 : /// `"attributeCreated"` + `"attributeUpdated"`, and `"entityUpdated"` "is
970 : /// equivalent to the combination `"attributeCreated"`, `"attributeUpdated"`
971 : /// and `"attributeDeleted"`" — so it is expanded here, at the single point
972 : /// the list is read, rather than at each comparison.
973 446 : fn triggers_of(sub: &Value) -> Vec<&str> {
974 : // Borrowed from the Subscription, not copied out of it: this runs once
975 : // per candidate per change, and the default arm used to allocate a
976 : // String for each of DEFAULT_TRIGGERS, which are already 'static.
977 446 : let mut triggers: Vec<&str> = sub
978 446 : .get("notificationTrigger")
979 446 : .and_then(Value::as_array)
980 446 : .map(|a| a.iter().filter_map(Value::as_str).collect())
981 446 : .unwrap_or_else(|| DEFAULT_TRIGGERS.to_vec());
982 446 : if triggers.contains(&"entityUpdated") {
983 36 : for t in DEFAULT_TRIGGERS.iter().copied().chain(["attributeDeleted"]) {
984 36 : if !triggers.contains(&t) {
985 24 : triggers.push(t);
986 24 : }
987 : }
988 434 : }
989 446 : triggers
990 446 : }
991 :
992 : /// Changes one drain of the queue folds into one delivery pass — a batch
993 : /// request's N writes arrive as N events back to back and leave as ONE
994 : /// notification per matching subscription. Bounded so a flood cannot hold
995 : /// the first notification back indefinitely.
996 : const CHANGE_BATCH: usize = 256;
997 :
998 : /// One matched (subscription, entity) pair before delivery.
999 : struct Matched {
1000 : tenant: TenantId,
1001 : /// Shared with the mirror that holds it: a change is evaluated against
1002 : /// every candidate subscription, so owning a copy here made the cost of
1003 : /// a change scale with the size of those documents.
1004 : sub: Arc<Value>,
1005 : ctx: Arc<Context>,
1006 : data: Vec<Value>,
1007 : }
1008 :
1009 0 : pub async fn process_change(
1010 0 : st: &AppState,
1011 0 : tenant_str: &str,
1012 0 : before: Option<Value>,
1013 0 : after: Option<Value>,
1014 0 : ) {
1015 0 : process_changes(st, vec![(tenant_str.to_owned(), before, after)]).await;
1016 0 : }
1017 :
1018 : /// 5.8.6: "the Notification ... data ... shall contain the Entities that
1019 : /// match" — every change of one drain that matches the same subscription
1020 : /// travels in one notification, so a batch of N entities is one POST with N
1021 : /// data entries (and timesSent moves by one), never N POSTs.
1022 2330 : pub(crate) async fn process_changes(st: &AppState, changes: Vec<Change>) {
1023 2330 : let mut groups: Vec<Matched> = Vec::new();
1024 : // (tenant, subscription id) → its group. A scan for the group would be
1025 : // linear in the subscriptions already matched, and a drain of
1026 : // CHANGE_BATCH changes over S matching subscriptions walks it S times per
1027 : // change: quadratic in the one dimension this broker is built to grow
1028 : // (100 000 subscriptions), on the notification hot path.
1029 2330 : let mut index: std::collections::HashMap<(String, String), usize> =
1030 2330 : std::collections::HashMap::new();
1031 2330 : let mut ctx_memo = CtxMemo::new();
1032 2373 : for (tenant_str, before, after) in changes {
1033 2373 : for m in matches_for(st, &tenant_str, before, after, &mut ctx_memo).await {
1034 394 : let key = (
1035 394 : m.tenant.as_str().to_owned(),
1036 394 : m.sub
1037 394 : .get("id")
1038 394 : .and_then(Value::as_str)
1039 394 : .unwrap_or_default()
1040 394 : .to_owned(),
1041 394 : );
1042 394 : match index.get(&key) {
1043 136 : Some(&i) => groups[i].data.extend(m.data),
1044 258 : None => {
1045 258 : index.insert(key, groups.len());
1046 258 : groups.push(m);
1047 258 : }
1048 : }
1049 : }
1050 : }
1051 : // Groups are distinct subscriptions, so they leave concurrently; a
1052 : // subscription's changes in this drain are already one group, and the
1053 : // next drain starts only after this one, so per-subscription order holds.
1054 : // Each group is its own task: a group's bookkeeping reads and writes
1055 : // await the store, and one task's `for_each_concurrent` would still put
1056 : // every group's round-trips on a single task's poll budget.
1057 : #[cfg(not(target_arch = "wasm32"))]
1058 : {
1059 2330 : let mut set = tokio::task::JoinSet::new();
1060 2330 : for g in groups {
1061 258 : let st = st.clone();
1062 258 : set.spawn(async move {
1063 258 : let _permits = delivery_permits(&g.tenant).await;
1064 258 : deliver(&st, &g.tenant, &g.sub, g.data, &g.ctx).await;
1065 217 : });
1066 : }
1067 2539 : while let Some(joined) = set.join_next().await {
1068 209 : if joined.is_err() {
1069 0 : note_panic();
1070 209 : }
1071 : }
1072 : }
1073 : #[cfg(target_arch = "wasm32")]
1074 : {
1075 : use futures_util::StreamExt;
1076 : futures_util::stream::iter(groups)
1077 : .for_each_concurrent(*DELIVERY_WIDTH, |g| async move {
1078 : deliver(st, &g.tenant, &g.sub, g.data, &g.ctx).await;
1079 : })
1080 : .await;
1081 : }
1082 2281 : }
1083 :
1084 : /// A change the full queue refused: counted, and said once per thousand so
1085 : /// a 40 % delivery gap is visible in the log and not only on the counter.
1086 4 : fn note_drop() {
1087 4 : let n = CHANGES_DROPPED.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
1088 4 : metrics::counter!("antares_notification_changes_dropped_total").increment(1);
1089 4 : if n == 1 || n.is_multiple_of(1000) {
1090 4 : tracing::warn!("notification change queue full: {n} changes dropped so far (delivery slower than the write rate)");
1091 0 : }
1092 4 : }
1093 :
1094 : #[cfg(not(target_arch = "wasm32"))]
1095 : static DELIVERY_SLOTS: std::sync::LazyLock<tokio::sync::Semaphore> =
1096 46 : std::sync::LazyLock::new(|| tokio::sync::Semaphore::new(*DELIVERY_WIDTH));
1097 :
1098 : #[cfg(not(target_arch = "wasm32"))]
1099 : type TenantSlots = std::sync::Mutex<std::collections::HashMap<String, Arc<tokio::sync::Semaphore>>>;
1100 : #[cfg(not(target_arch = "wasm32"))]
1101 : static TENANT_SLOTS: std::sync::LazyLock<TenantSlots> =
1102 : std::sync::LazyLock::new(TenantSlots::default);
1103 :
1104 : /// The tenant's share, created on first use. An entry lives only while a
1105 : /// delivery is holding it: a lookup that has to create one first drops every
1106 : /// entry nothing else still references, so this map is bounded by the
1107 : /// delivery width and not by how many of the broker's 10 000 tenants have
1108 : /// ever had a subscription fire.
1109 : #[cfg(not(target_arch = "wasm32"))]
1110 570 : fn tenant_slots(tenant: &TenantId) -> Arc<tokio::sync::Semaphore> {
1111 : // A poisoned map is a panic in some past delivery, not a reason to stop
1112 : // delivering: what it holds is rebuildable and bounded either way.
1113 570 : let mut map = TENANT_SLOTS.lock().unwrap_or_else(|p| p.into_inner());
1114 570 : if let Some(s) = map.get(tenant.as_str()) {
1115 514 : return Arc::clone(s);
1116 56 : }
1117 56 : map.retain(|_, s| Arc::strong_count(s) > 1);
1118 56 : let slots = Arc::new(tokio::sync::Semaphore::new(*DELIVERY_WIDTH_PER_TENANT));
1119 56 : map.insert(tenant.as_str().to_owned(), Arc::clone(&slots));
1120 56 : slots
1121 570 : }
1122 :
1123 : /// Both permits one delivery holds until it settles. The tenant's share is
1124 : /// taken first, so a tenant already at its share waits there instead of
1125 : /// waiting inside one of the broker's slots.
1126 : #[cfg(not(target_arch = "wasm32"))]
1127 570 : async fn delivery_permits(tenant: &TenantId) -> impl Sized {
1128 570 : let share = tenant_slots(tenant).acquire_owned().await;
1129 314 : (share, DELIVERY_SLOTS.acquire().await)
1130 314 : }
1131 :
1132 2373 : async fn matches_for(
1133 2373 : st: &AppState,
1134 2373 : tenant_str: &str,
1135 2373 : before: Option<Value>,
1136 2373 : after: Option<Value>,
1137 2373 : ctx_memo: &mut CtxMemo,
1138 2373 : ) -> Vec<Matched> {
1139 2373 : let mut out = Vec::new();
1140 : // The tenant of the write that fired the change hook — this broker's own
1141 : // value, and a write inside a Snapshot (5.5.15) carries the synthetic one.
1142 2373 : let Ok(tenant) = TenantId::new_internal(tenant_str) else {
1143 0 : return out;
1144 : };
1145 2373 : let changes = diff(before.as_ref(), after.as_ref());
1146 2373 : let entity_trigger = match (&before, &after) {
1147 1696 : (None, Some(_)) => "entityCreated",
1148 472 : (Some(_), None) => "entityDeleted",
1149 205 : _ => "entityUpdated",
1150 : };
1151 2373 : let eval_doc = after.as_ref().or(before.as_ref());
1152 2373 : let Some(eval_doc) = eval_doc else { return out };
1153 : // Candidate lookup by the entity's types and the changed attribute
1154 : // IRIs — no linear scan over all subscriptions.
1155 2373 : let types: Vec<&str> = eval_doc
1156 2373 : .get("type")
1157 2373 : .and_then(Value::as_array)
1158 2373 : .map(|a| a.iter().filter_map(Value::as_str).collect())
1159 2373 : .unwrap_or_default();
1160 2373 : let changed_keys: Vec<&str> = changes.iter().map(|(k, _)| k.as_str()).collect();
1161 2373 : let subs = subs_for(st, &tenant, &types, &changed_keys).await;
1162 2373 : for sub in subs {
1163 446 : if !is_active(&sub) || sub.get("timeInterval").is_some() {
1164 20 : continue;
1165 426 : }
1166 426 : let triggers = triggers_of(&sub);
1167 : // which attribute-level changes this sub cares about
1168 426 : let watched: Option<Vec<&str>> = sub
1169 426 : .get("watchedAttributes")
1170 426 : .and_then(Value::as_array)
1171 426 : .map(|a| a.iter().filter_map(Value::as_str).collect());
1172 426 : let relevant: Vec<&(String, ChangeClass)> = changes
1173 426 : .iter()
1174 450 : .filter(|(k, _)| watched.as_ref().is_none_or(|w| w.contains(&k.as_str())))
1175 426 : .collect();
1176 426 : let attr_trigger_fired = relevant.iter().any(|(_, c)| {
1177 422 : let t = match c {
1178 410 : ChangeClass::Created => "attributeCreated",
1179 8 : ChangeClass::Updated => "attributeUpdated",
1180 4 : ChangeClass::Deleted => "attributeDeleted",
1181 : };
1182 422 : triggers.contains(&t)
1183 422 : });
1184 426 : let entity_trigger_fired =
1185 426 : triggers.contains(&entity_trigger) && (watched.is_none() || !relevant.is_empty());
1186 426 : if !attr_trigger_fired && !entity_trigger_fired {
1187 12 : continue;
1188 414 : }
1189 414 : let ctx = sub_context_memo(st, &tenant, tenant_str, &sub, ctx_memo).await;
1190 414 : if !selector_match(&sub, eval_doc, &ctx) {
1191 0 : continue;
1192 414 : }
1193 426 : if !linked_eval(st, &tenant, |l| conditions_match(&sub, eval_doc, &ctx, l)).await {
1194 14 : continue;
1195 400 : }
1196 400 : if throttled(&sub) {
1197 6 : continue;
1198 394 : }
1199 394 : let deleted: Vec<String> = if triggers.contains(&"attributeDeleted") {
1200 4 : relevant
1201 4 : .iter()
1202 4 : .filter(|(_, c)| *c == ChangeClass::Deleted)
1203 4 : .map(|(k, _)| k.clone())
1204 4 : .collect()
1205 : } else {
1206 390 : Vec::new()
1207 : };
1208 394 : let entity_deleted_fired = after.is_none() && triggers.contains(&"entityDeleted");
1209 394 : let now = now_iso();
1210 394 : let data = build_data(
1211 394 : st,
1212 394 : &tenant,
1213 394 : &sub,
1214 394 : &ctx,
1215 394 : before.as_ref(),
1216 394 : after.as_ref(),
1217 394 : &deleted,
1218 394 : entity_deleted_fired,
1219 394 : &now,
1220 394 : )
1221 394 : .await;
1222 394 : out.push(Matched {
1223 394 : tenant: tenant.clone(),
1224 394 : sub,
1225 394 : ctx,
1226 394 : data,
1227 394 : });
1228 : }
1229 2373 : out
1230 2373 : }
1231 :
1232 : /// When an interval subscription is next due, in epoch millis: one
1233 : /// `timeInterval` after the last Notification it sent (Table 5.2.14.2-1
1234 : /// `lastNotification`), or after its creation while it has sent none. Without
1235 : /// either anchor it is due immediately.
1236 344 : fn due_at_ms(sub: &Value, interval: f64) -> i64 {
1237 344 : let anchor = sub
1238 344 : .get("notification")
1239 344 : .and_then(|n| n.get("lastNotification"))
1240 344 : .and_then(Value::as_str)
1241 344 : .or_else(|| sub.get("createdAt").and_then(Value::as_str));
1242 344 : match anchor.and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok()) {
1243 344 : Some(last) => last
1244 344 : .timestamp_millis()
1245 344 : .saturating_add(interval_offset_ms(interval)),
1246 0 : None => i64::MIN,
1247 : }
1248 344 : }
1249 :
1250 : /// One period of a periodic Subscription (5.2.12 `timeInterval`) in
1251 : /// milliseconds.
1252 : ///
1253 : /// Table 5.2.12-1 bounds the member only as greater than 0, so the seconds a
1254 : /// client names can exceed what epoch milliseconds hold. The cast saturates,
1255 : /// and every caller adds it with `saturating_add`, which puts an interval the
1256 : /// broker cannot schedule at the end of representable time. Adding it plainly
1257 : /// wraps the sum negative, and a negative firing instant reads as permanently
1258 : /// due: the subscription then fires its whole query on every tick, and the
1259 : /// same value poisons the process-wide sweep clock those minima feed.
1260 680 : fn interval_offset_ms(interval: f64) -> i64 {
1261 680 : (interval * 1000.0) as i64
1262 680 : }
1263 :
1264 : /// The exact Entity ids a subscription's `entities` selector pins down, or
1265 : /// `None` when it leaves any of them open. Table 5.2.33-1: `id` is a String
1266 : /// or a String[] and "id takes precedence over idPattern", so an entry
1267 : /// carrying an id constrains the read exactly — while one entry without an id
1268 : /// (a bare type, or an idPattern no store column can answer) admits every id
1269 : /// and forfeits the narrowing for the whole OR-ed selector.
1270 392 : fn selector_ids(sub: &Value) -> Option<Vec<String>> {
1271 392 : let sel = sub.get("entities").and_then(Value::as_array)?;
1272 380 : let mut ids = Vec::new();
1273 388 : for e in sel {
1274 388 : match e.get("id") {
1275 32 : Some(Value::String(i)) => ids.push(i.clone()),
1276 8 : Some(Value::Array(a)) => {
1277 16 : for v in a {
1278 16 : ids.push(v.as_str()?.to_owned());
1279 : }
1280 : }
1281 348 : _ => return None,
1282 : }
1283 : }
1284 32 : (!ids.is_empty()).then_some(ids)
1285 392 : }
1286 :
1287 : /// A store read on a delivery path has no caller to fail: the sweep is a
1288 : /// timer, the fan-out is spawned, and both answer `()`. Silence is what makes
1289 : /// a failure dangerous here — a subscription that stops firing because the
1290 : /// store could not be read looks exactly like one with nothing to send — so
1291 : /// the failure is named and the path continues on the empty set it would
1292 : /// have continued on anyway.
1293 5699 : fn read_or_warn<T>(res: Result<Vec<T>, antares_model::NgsiError>, what: &str) -> Vec<T> {
1294 5699 : res.unwrap_or_else(|e| {
1295 4 : tracing::warn!("notification path: reading {what} failed: {e}");
1296 4 : Vec::new()
1297 4 : })
1298 5699 : }
1299 :
1300 : /// timeInterval subscriptions: fire when due, with all matching entities.
1301 : /// Multi-instance: claim one interval firing under the subscription row
1302 : /// lock — N matcher pods race, exactly one wins (single-winner by
1303 : /// lock, no leader election). The due-check reruns INSIDE the lock; the
1304 : /// winner stamps `lastNotification` as its claim, losers see not-due and
1305 : /// roll back. Only engaged in bus=nats mode — single-process behaviour (and
1306 : /// its 046_12 bookkeeping ordering) is untouched.
1307 : ///
1308 : /// `None` = the firing is not this pod's. `Some(prev)` = claimed, carrying
1309 : /// the `lastNotification` the claim overwrote: a firing that turns out to
1310 : /// have nothing to send gives it back through [`release_interval`], because
1311 : /// Table 5.2.14.2-1 stamps the instant a notification was SENT and 5.8.6
1312 : /// sends none when nothing matches.
1313 4 : async fn claim_interval(
1314 4 : st: &AppState,
1315 4 : tenant: &TenantId,
1316 4 : kind: Kind,
1317 4 : sub: &Value,
1318 4 : interval: f64,
1319 4 : ) -> Option<Option<Value>> {
1320 4 : let id = sub.get("id").and_then(Value::as_str)?;
1321 4 : let mut prev: Option<Value> = None;
1322 4 : let res = st
1323 4 : .store
1324 4 : .mutate(tenant, kind, id, |doc| {
1325 4 : if chrono::Utc::now().timestamp_millis() < due_at_ms(doc, interval) {
1326 0 : return Err(());
1327 4 : }
1328 4 : let Some(sub_doc) = doc.as_object_mut() else {
1329 : // a stored Subscription is an object; one that is not carries no
1330 : // notification member to stamp, and Err(()) is the same "nothing
1331 : // written" the not-due branch above returns
1332 0 : return Err(());
1333 : };
1334 4 : if let Some(n) = sub_doc
1335 4 : .entry("notification")
1336 4 : .or_insert_with(|| json!({}))
1337 4 : .as_object_mut()
1338 4 : {
1339 4 : prev = n.insert("lastNotification".into(), Value::String(now_iso()));
1340 4 : }
1341 4 : Ok(())
1342 4 : })
1343 4 : .await;
1344 4 : matches!(res, Ok(Some(Ok(())))).then_some(prev)
1345 4 : }
1346 :
1347 : /// Give a claimed firing back (5.8.6: nothing matched, so nothing was sent).
1348 : /// The stamp returns to what [`claim_interval`] found, which both keeps
1349 : /// `lastNotification` truthful and leaves the subscription due, exactly as
1350 : /// the single-process path does.
1351 4 : async fn release_interval(
1352 4 : st: &AppState,
1353 4 : tenant: &TenantId,
1354 4 : kind: Kind,
1355 4 : id: &str,
1356 4 : prev: Option<Value>,
1357 4 : ) {
1358 4 : let res = st
1359 4 : .store
1360 4 : .mutate::<(), ()>(tenant, kind, id, |doc| {
1361 4 : if let Some(n) = doc
1362 4 : .as_object_mut()
1363 4 : .and_then(|o| o.get_mut("notification"))
1364 4 : .and_then(Value::as_object_mut)
1365 : {
1366 4 : match &prev {
1367 0 : Some(v) => n.insert("lastNotification".into(), v.clone()),
1368 4 : None => n.remove("lastNotification"),
1369 : };
1370 0 : }
1371 4 : Ok(())
1372 4 : })
1373 4 : .await;
1374 4 : if let Err(e) = res {
1375 0 : tracing::warn!("releasing the interval claim for {id} failed: {e}");
1376 4 : }
1377 4 : }
1378 :
1379 : /// One sweep of the interval subscriptions (5.8.6, 5.11.7): "If a
1380 : /// Subscription defines a timeInterval member, a Notification shall be sent
1381 : /// periodically, when the time interval (in seconds) specified in such value
1382 : /// field is reached, regardless of Attribute changes."
1383 : ///
1384 : /// The sweep runs on a fixed tick, so its idle cost is what has to stay
1385 : /// small. Two things keep it off the store. A tick that cannot fire anything
1386 : /// returns before enumerating tenants: each sweep records the earliest instant
1387 : /// a subscription it saw can next be due, and a write zeroes that clock —
1388 : /// through the mirror for Subscriptions, through `csub_written` for the
1389 : /// Context Source Registration Subscription half, which is mirrored by clock
1390 : /// rather than by document. A lost signal is repaired within
1391 : /// `CSUB_SWEEP_BACKSTOP_MS`. And a due subscription reads only the Entities
1392 : /// its own selector can match instead of its tenant's entity set.
1393 3320 : pub async fn interval_tick(st: &AppState) {
1394 : use std::sync::atomic::Ordering::Relaxed;
1395 3320 : let now_ms = chrono::Utc::now().timestamp_millis();
1396 3320 : let clocks = st.sub_mirror.as_ref().map(|m| {
1397 3304 : (
1398 3304 : m.next_sub_sweep_ms.load(Relaxed),
1399 3304 : m.next_csub_sweep_ms.load(Relaxed),
1400 3304 : )
1401 3304 : });
1402 3320 : let (sweep_subs, sweep_csubs) = match clocks {
1403 3304 : Some((sub_clock, csub_clock)) => (now_ms >= sub_clock, now_ms >= csub_clock),
1404 : // Never-wired fallback: no clock to keep, so every tick sweeps.
1405 16 : None => (true, true),
1406 : };
1407 3320 : if !sweep_subs && !sweep_csubs {
1408 3218 : return;
1409 102 : }
1410 : // Earliest next-due instant seen by this sweep, per half.
1411 102 : let mut next_sub = i64::MAX;
1412 102 : let mut next_csub = i64::MAX;
1413 : // One sweep visits every tenant, and a delivery costs up to the
1414 : // endpoint's whole timeout (Table 5.2.15-1, 30 s at the ceiling). Awaited
1415 : // in turn, one unresponsive endpoint becomes the deadline of every other
1416 : // subscriber's periodic notification. The deliveries of a tick therefore
1417 : // run together, under the same width the change path uses, and the tick
1418 : // still does not return until they have settled: ticks never overlap, so
1419 : // a subscription cannot be fired twice for one period.
1420 : #[cfg(not(target_arch = "wasm32"))]
1421 102 : let mut sending = tokio::task::JoinSet::new();
1422 749 : for tenant_str in read_or_warn(
1423 102 : st.store.subscription_tenants().await,
1424 102 : "the tenants with subscriptions",
1425 : ) {
1426 733 : let Ok(tenant) = TenantId::new_internal(&tenant_str) else {
1427 0 : continue;
1428 : };
1429 : // Same source the matcher reads: the indexed mirror, with the store
1430 : // list only as the never-wired fallback.
1431 733 : let subs = match (&st.sub_mirror, sweep_subs) {
1432 586 : (_, false) => Vec::new(),
1433 131 : (Some(m), _) => m.periodic_docs(tenant.as_str()),
1434 16 : (None, _) => read_or_warn(
1435 16 : st.store.list(&tenant, Kind::Subscription).await,
1436 16 : "the periodic Subscriptions",
1437 : )
1438 16 : .into_iter()
1439 16 : .map(Arc::new)
1440 16 : .collect(),
1441 : };
1442 879 : for sub in subs {
1443 324 : let Some(interval) = sub.get("timeInterval").and_then(Value::as_f64) else {
1444 0 : continue;
1445 : };
1446 324 : if !is_active(&sub) {
1447 0 : continue;
1448 324 : }
1449 324 : let due_at = due_at_ms(&sub, interval);
1450 324 : if now_ms < due_at {
1451 0 : next_sub = next_sub.min(due_at);
1452 0 : continue;
1453 324 : }
1454 : // Due: the following firing is one interval away. Recorded before
1455 : // the claim, so a pod that LOSES the race (another one is firing
1456 : // this subscription right now) keeps sweeping on the interval
1457 : // instead of parking on an anchor only the winner advanced.
1458 324 : next_sub = next_sub.min(now_ms.saturating_add(interval_offset_ms(interval)));
1459 324 : let claim = if st.nats {
1460 4 : match claim_interval(st, &tenant, Kind::Subscription, &sub, interval).await {
1461 4 : Some(prev) => Some(prev),
1462 0 : None => continue,
1463 : }
1464 : } else {
1465 320 : None
1466 : };
1467 324 : let ctx = sub_context(st, &tenant, &sub).await;
1468 324 : let now = now_iso();
1469 : // 5.8.6: the periodic Notification "shall include all the
1470 : // subscribed Entities that match the query, geoquery and Scope
1471 : // query conditions" — so the read is exactly this subscription's
1472 : // own selector (5.2.33) and conditions, never the tenant's entity
1473 : // set. Only predicates a store reproduces without hiding a
1474 : // candidate are offered (ids when every selector entry names one,
1475 : // types when the index proves them plain, q/scopeQ/geoQ under the
1476 : // store's own rule that SQL removes rows and never decides them);
1477 : // the selector_match/conditions_match pair below stays the
1478 : // arbiter, exactly as on the query path.
1479 324 : let type_groups: Vec<Vec<String>> = match index_keys(&sub) {
1480 324 : Keys::Types(ts) => ts.into_iter().map(|t| vec![t]).collect(),
1481 0 : _ => Vec::new(),
1482 : };
1483 324 : let ids = selector_ids(&sub);
1484 : // The filter borrows a term expander that is not Sync, so it lives
1485 : // and dies inside this block: held across the delivery await it
1486 : // would make the whole interval task non-Send.
1487 324 : let rows = {
1488 324 : let expand = |t: &str| ctx.expand_key(t);
1489 324 : let id_refs: Vec<&str> = ids.iter().flatten().map(String::as_str).collect();
1490 : // q values in subscription bodies may be percent-encoded (4.9);
1491 : // parses shared per distinct expression text, as in
1492 : // conditions_match — the sweep re-runs per due subscription
1493 324 : let q_ast = sub_str(&sub, "q").and_then(|q| {
1494 0 : antares_ql::regex::q_node(&crate::negotiate::percent_decode(q.as_bytes()))
1495 0 : });
1496 324 : let geo = sub.get("geoQ").and_then(Value::as_object).and_then(|g| {
1497 0 : let key = serde_json::to_string(g).unwrap_or_default();
1498 0 : antares_ql::regex::geo_query(&key, || {
1499 0 : antares_ql::geo::GeoQuery::from_params(&geo_params(g))
1500 0 : .ok()
1501 0 : .flatten()
1502 0 : })
1503 0 : });
1504 324 : let geo_spec = geo.as_ref().and_then(|g| g.to_sql_spec(&ctx));
1505 324 : let filter = antares_store::filter::EntityFilter {
1506 324 : ids: ids.as_ref().map(|_| id_refs.as_slice()),
1507 324 : types: (!type_groups.is_empty()).then_some(type_groups.as_slice()),
1508 324 : q: q_ast.as_deref(),
1509 324 : scope_q: sub_str(&sub, "scopeQ"),
1510 324 : geo: geo_spec.as_ref(),
1511 324 : expand: &expand,
1512 324 : ..Default::default()
1513 : };
1514 324 : read_or_warn(
1515 324 : st.store
1516 324 : .query_entities(&tenant, &filter)
1517 324 : .await
1518 324 : .map(|o| o.rows),
1519 324 : "the Entities a periodic Subscription notifies about",
1520 : )
1521 : };
1522 324 : let mut matching: Vec<Value> = Vec::new();
1523 340 : for d in rows {
1524 340 : if !selector_match(&sub, &d, &ctx)
1525 312 : || !linked_eval(st, &tenant, |l| conditions_match(&sub, &d, &ctx, l)).await
1526 : {
1527 28 : continue;
1528 312 : }
1529 312 : matching.extend(
1530 312 : build_data(st, &tenant, &sub, &ctx, None, Some(&d), &[], false, &now).await,
1531 : );
1532 : }
1533 324 : if matching.is_empty() {
1534 : // 5.8.6: "If there are no matching Entities, no Notification
1535 : // is sent" — lastNotification stays untouched, so this
1536 : // subscription is still due and every following tick
1537 : // re-checks it. A claim taken to win the firing is given back
1538 : // here, or the multi-pod path would stamp an instant nothing
1539 : // was sent at and park the subscription for a whole interval.
1540 12 : if let (Some(prev), Some(id)) = (claim, sub.get("id").and_then(Value::as_str)) {
1541 4 : release_interval(st, &tenant, Kind::Subscription, id, prev).await;
1542 8 : }
1543 12 : next_sub = next_sub.min(due_at);
1544 12 : continue;
1545 312 : }
1546 : #[cfg(not(target_arch = "wasm32"))]
1547 : {
1548 312 : let (st, tenant, ctx) = (st.clone(), tenant.clone(), Arc::clone(&ctx));
1549 312 : sending.spawn(async move {
1550 312 : let _permits = delivery_permits(&tenant).await;
1551 56 : deliver(&st, &tenant, &sub, matching, &ctx).await;
1552 24 : });
1553 : }
1554 : #[cfg(target_arch = "wasm32")]
1555 : deliver(st, &tenant, &sub, matching, &ctx).await;
1556 : }
1557 733 : if !sweep_csubs {
1558 4 : continue;
1559 729 : }
1560 : // csource timeInterval subs: periodic CSourceNotification with all
1561 : // matching registrations, independent of changes (5.11.7)
1562 729 : for sub in read_or_warn(
1563 729 : st.store.list(&tenant, Kind::CSourceSubscription).await,
1564 729 : "the periodic Context Source Registration Subscriptions",
1565 : ) {
1566 0 : let Some(interval) = sub.get("timeInterval").and_then(Value::as_f64) else {
1567 0 : continue;
1568 : };
1569 0 : if !is_active(&sub) {
1570 0 : continue;
1571 0 : }
1572 0 : let due_at = due_at_ms(&sub, interval);
1573 0 : if now_ms < due_at {
1574 0 : next_csub = next_csub.min(due_at);
1575 0 : continue;
1576 0 : }
1577 0 : next_csub = next_csub.min(now_ms.saturating_add(interval_offset_ms(interval)));
1578 : // 5.11.7 sends the periodic CSourceNotification whatever the
1579 : // matching set is, so this claim is never given back.
1580 0 : if st.nats
1581 0 : && claim_interval(st, &tenant, Kind::CSourceSubscription, &sub, interval)
1582 0 : .await
1583 0 : .is_none()
1584 : {
1585 0 : continue;
1586 0 : }
1587 0 : let ctx = sub_context(st, &tenant, &sub).await;
1588 0 : let spec = crate::registry::spec_for_subscription(&sub);
1589 0 : let data: Vec<Value> = read_or_warn(
1590 0 : st.store.list(&tenant, Kind::Registration).await,
1591 0 : "the registrations a periodic Context Source Notification carries",
1592 : )
1593 0 : .into_iter()
1594 0 : .filter(|r| crate::registry::csr_matches_subscription(&sub, r, &ctx))
1595 0 : .map(|r| {
1596 0 : let mut p = crate::registry::present_registration(
1597 0 : &filter_csr(&spec, &r, &ctx),
1598 0 : &ctx,
1599 : false,
1600 : );
1601 0 : arrayify_entity_types(&mut p);
1602 0 : p
1603 0 : })
1604 0 : .collect();
1605 : #[cfg(not(target_arch = "wasm32"))]
1606 : {
1607 0 : let (st, tenant, ctx) = (st.clone(), tenant.clone(), Arc::clone(&ctx));
1608 0 : sending.spawn(async move {
1609 0 : let _permit = DELIVERY_SLOTS.acquire().await;
1610 0 : deliver_csource(&st, &tenant, &sub, data, &ctx, "newlyMatching").await;
1611 0 : });
1612 : }
1613 : #[cfg(target_arch = "wasm32")]
1614 : deliver_csource(st, &tenant, &sub, data, &ctx, "newlyMatching").await;
1615 : }
1616 : }
1617 : #[cfg(not(target_arch = "wasm32"))]
1618 126 : while let Some(joined) = sending.join_next().await {
1619 24 : if joined.is_err() {
1620 0 : note_panic();
1621 24 : }
1622 : }
1623 98 : if let (Some(m), Some((sub_clock, csub_clock))) = (&st.sub_mirror, clocks) {
1624 : // A periodic subscription written DURING the sweep has already zeroed
1625 : // the clock: the exchange then fails, the zero stands and the next
1626 : // tick sweeps rather than waiting out an interval computed without it.
1627 86 : if sweep_subs {
1628 58 : let _ = m
1629 58 : .next_sub_sweep_ms
1630 58 : .compare_exchange(sub_clock, next_sub, Relaxed, Relaxed);
1631 62 : }
1632 86 : if sweep_csubs {
1633 82 : let _ = m.next_csub_sweep_ms.compare_exchange(
1634 82 : csub_clock,
1635 82 : next_csub.min(now_ms + CSUB_SWEEP_BACKSTOP_MS),
1636 82 : Relaxed,
1637 82 : Relaxed,
1638 82 : );
1639 82 : }
1640 12 : }
1641 3316 : }
1642 :
1643 : /// POST the Notification (5.3.1) and write the 5.2.14.2 bookkeeping back.
1644 344 : pub(crate) async fn deliver(
1645 344 : st: &AppState,
1646 344 : tenant: &TenantId,
1647 344 : sub: &Value,
1648 344 : data: Vec<Value>,
1649 344 : ctx: &Context,
1650 344 : ) {
1651 : // a notification body is bounded the way an inbound body is (6.3.4 wall,
1652 : // MAX_BODY_BYTES): a grouped delivery over the cap leaves as several
1653 : // notifications, each whole entities, never one unbounded POST
1654 344 : for chunk in chunk_by_bytes(data, *crate::bounds::MAX_BODY_BYTES) {
1655 344 : deliver_as(
1656 344 : st,
1657 344 : tenant,
1658 344 : Kind::Subscription,
1659 344 : sub,
1660 344 : "Notification",
1661 344 : chunk,
1662 344 : ctx,
1663 344 : None,
1664 344 : )
1665 344 : .await;
1666 : }
1667 270 : }
1668 :
1669 : /// Split `data` into runs whose serialized sizes stay under `cap`, cutting
1670 : /// only at whole items; one item alone over the cap still travels alone.
1671 413 : fn chunk_by_bytes(data: Vec<Value>, cap: usize) -> Vec<Vec<Value>> {
1672 413 : let mut out: Vec<Vec<Value>> = Vec::new();
1673 413 : let mut size = 0usize;
1674 583 : for item in data {
1675 583 : let n = serde_json::to_vec(&item).map(|b| b.len()).unwrap_or(0);
1676 583 : match out.last_mut() {
1677 174 : Some(run) if size + n <= cap => {
1678 160 : run.push(item);
1679 160 : size += n;
1680 160 : }
1681 423 : _ => {
1682 423 : out.push(vec![item]);
1683 423 : size = n;
1684 423 : }
1685 : }
1686 : }
1687 413 : out
1688 413 : }
1689 :
1690 : /// 5.11.7: which csource subs care about a registration change, and why.
1691 34 : fn csource_trigger(
1692 34 : sub: &Value,
1693 34 : before: Option<&Value>,
1694 34 : after: Option<&Value>,
1695 34 : ctx: &Context,
1696 34 : ) -> Option<&'static str> {
1697 68 : let m = |d: Option<&Value>| {
1698 68 : d.is_some_and(|d| crate::registry::csr_matches_subscription(sub, d, ctx))
1699 68 : };
1700 34 : match (m(before), m(after)) {
1701 16 : (false, true) => Some("newlyMatching"),
1702 4 : (true, true) => Some("updated"),
1703 10 : (true, false) => Some("noLongerMatching"),
1704 4 : (false, false) => None,
1705 : }
1706 34 : }
1707 :
1708 : /// The notification validator reads EntityInfo.type as an array
1709 : /// (`entities[0]["type"][0]`) — normalize to array form in notification data.
1710 81 : fn arrayify_entity_types(reg: &mut Value) {
1711 81 : let Some(infos) = reg.get_mut("information").and_then(Value::as_array_mut) else {
1712 0 : return;
1713 : };
1714 81 : for info in infos {
1715 81 : let Some(es) = info.get_mut("entities").and_then(Value::as_array_mut) else {
1716 0 : continue;
1717 : };
1718 81 : for e in es {
1719 81 : if let Some(t) = e.get("type").filter(|t| t.is_string()).cloned() {
1720 81 : if let Some(o) = e.as_object_mut() {
1721 81 : o.insert("type".into(), Value::Array(vec![t]));
1722 81 : }
1723 0 : }
1724 : }
1725 : }
1726 81 : }
1727 :
1728 : /// One prepared CSource notification, ready to send.
1729 : pub struct CsourceJob {
1730 : sub: Value,
1731 : presented: Value,
1732 : ctx: std::sync::Arc<antares_jsonld::Context>,
1733 : reason: &'static str,
1734 : }
1735 :
1736 : /// Registration create/update/delete → CSourceNotification fan-out (5.11.7),
1737 : /// in two phases: `prepare_csource_jobs` runs IN the request path (store
1738 : /// reads + matching + payload build — so job order is the handlers' commit
1739 : /// order even on a slower store), and the caller spawns `send_csource_jobs`
1740 : /// (network only — the ack must not block on the receiver: the ETSI mock
1741 : /// replies only when the robot side wakes).
1742 2134 : pub(crate) async fn prepare_csource_jobs(
1743 2134 : st: &AppState,
1744 2134 : tenant: &TenantId,
1745 2134 : before: Option<Value>,
1746 2134 : after: Option<Value>,
1747 2134 : ) -> Vec<CsourceJob> {
1748 2134 : let mut jobs = Vec::new();
1749 : // 5.8.1.4: the Registration Subscriptions the distributed half owns are
1750 : // not client resources, so they live under Kind::DistSub beside the
1751 : // mapping documents — which carry no `type`. A document under the
1752 : // client kind in the internal id namespace is a leftover of a release
1753 : // that stored the internal ones there, and drives nothing.
1754 2134 : let client = read_or_warn(
1755 2134 : st.store.list(tenant, Kind::CSourceSubscription).await,
1756 2134 : "the Context Source Registration Subscriptions of a changed registration",
1757 : )
1758 2134 : .into_iter()
1759 2134 : .filter(|d| {
1760 0 : sub_str(d, "id")
1761 0 : .is_some_and(|id| crate::registry::csr_kind(id) == Kind::CSourceSubscription)
1762 0 : });
1763 2134 : let internal = read_or_warn(
1764 2134 : st.store.list(tenant, Kind::DistSub).await,
1765 2134 : "the internal Registration Subscriptions of a changed registration",
1766 : )
1767 2134 : .into_iter()
1768 2134 : .filter(|d| d.get("type").and_then(Value::as_str) == Some("Subscription"));
1769 2134 : for sub in client.chain(internal) {
1770 10 : if !is_active(&sub) || sub.get("timeInterval").is_some() {
1771 0 : continue;
1772 10 : }
1773 10 : let ctx = sub_context(st, tenant, &sub).await;
1774 10 : let Some(reason) = csource_trigger(&sub, before.as_ref(), after.as_ref(), &ctx) else {
1775 0 : continue;
1776 : };
1777 10 : let spec = crate::registry::spec_for_subscription(&sub);
1778 10 : let source = if reason == "noLongerMatching" {
1779 2 : &before
1780 : } else {
1781 8 : &after
1782 : };
1783 10 : let Some(reg) = source.as_ref().or(before.as_ref()) else {
1784 0 : continue;
1785 : };
1786 10 : let filtered = filter_csr(&spec, reg, &ctx);
1787 10 : let mut presented = crate::registry::present_registration(&filtered, &ctx, false);
1788 10 : arrayify_entity_types(&mut presented);
1789 10 : jobs.push(CsourceJob {
1790 10 : sub,
1791 10 : presented,
1792 10 : ctx,
1793 10 : reason,
1794 10 : });
1795 : }
1796 2134 : jobs
1797 2134 : }
1798 :
1799 2022 : pub(crate) async fn send_csource_jobs(st: &AppState, tenant: &TenantId, jobs: Vec<CsourceJob>) {
1800 2022 : for job in jobs {
1801 : // 5.11.7: re-check the subscription still exists right
1802 : // before the send — a deleted subscription must never notify.
1803 6 : let sub_id = job
1804 6 : .sub
1805 6 : .get("id")
1806 6 : .and_then(Value::as_str)
1807 6 : .unwrap_or_default();
1808 6 : let kind = crate::registry::csr_kind(sub_id);
1809 6 : if !matches!(st.store.get(tenant, kind, sub_id).await, Ok(Some(_))) {
1810 0 : continue;
1811 6 : }
1812 6 : deliver_as(
1813 6 : st,
1814 6 : tenant,
1815 6 : kind,
1816 6 : &job.sub,
1817 6 : "ContextSourceNotification",
1818 6 : vec![job.presented],
1819 6 : &job.ctx,
1820 6 : Some(job.reason),
1821 6 : )
1822 6 : .await;
1823 : }
1824 2022 : }
1825 :
1826 : /// Registration writes prepare one job per CSource subscription of the
1827 : /// tenant (5.11.7) and every subscription with localOnly != true owns one,
1828 : /// so a registration stream against many subscriptions queues
1829 : /// subscriptions × registrations jobs faster than the sources drain them —
1830 : /// at 10 000 × 100 the queued jobs were a 3 GB broker peak. The permit is
1831 : /// taken in the request path: a write waits for a fan-out slot instead of
1832 : /// stacking jobs, and the prepare order (the handlers' commit order) holds.
1833 : const CSOURCE_FANOUT_WIDTH: usize = 64;
1834 : static CSOURCE_FANOUT: tokio::sync::Semaphore =
1835 : tokio::sync::Semaphore::const_new(CSOURCE_FANOUT_WIDTH);
1836 :
1837 : /// Registration create/update/delete → prepare in the request path, send
1838 : /// spawned (the ack must not block on the receiver), bounded as above.
1839 2134 : pub(crate) async fn csource_fanout(
1840 2134 : st: &AppState,
1841 2134 : tenant: &TenantId,
1842 2134 : before: Option<Value>,
1843 2134 : after: Option<Value>,
1844 2134 : ) {
1845 : // never closed, so acquire only fails if it were — then send unbounded
1846 2134 : let permit = CSOURCE_FANOUT.acquire().await.ok();
1847 2134 : let jobs = prepare_csource_jobs(st, tenant, before, after).await;
1848 2134 : let (st2, t2) = (st.clone(), tenant.clone());
1849 2134 : crate::spawn(async move {
1850 2022 : send_csource_jobs(&st2, &t2, jobs).await;
1851 2022 : drop(permit);
1852 2022 : });
1853 2134 : }
1854 :
1855 : /// POST a CSourceNotification (5.3.2) under the same body bound as every
1856 : /// other one: 5.11.2.4 sends "all matching Context Source Registrations",
1857 : /// and a broker holding 100 000 of them must not turn that into one
1858 : /// unbounded request. Over the cap the set leaves as several notifications,
1859 : /// each carrying whole registrations — the trade [`deliver`] already makes
1860 : /// for entity Notifications.
1861 57 : async fn deliver_csource(
1862 57 : st: &AppState,
1863 57 : tenant: &TenantId,
1864 57 : sub: &Value,
1865 57 : data: Vec<Value>,
1866 57 : ctx: &Context,
1867 57 : reason: &str,
1868 57 : ) {
1869 57 : let kind = crate::registry::csr_kind(sub_str(sub, "id").unwrap_or_default());
1870 59 : for chunk in chunk_by_bytes(data, *crate::bounds::MAX_BODY_BYTES) {
1871 59 : deliver_as(
1872 59 : st,
1873 59 : tenant,
1874 59 : kind,
1875 59 : sub,
1876 59 : "ContextSourceNotification",
1877 59 : chunk,
1878 59 : ctx,
1879 59 : Some(reason),
1880 59 : )
1881 59 : .await;
1882 : }
1883 55 : }
1884 :
1885 : /// Initial / post-update CSourceNotification with all currently matching
1886 : /// registrations (5.11.2.4 / 5.11.3.4).
1887 260 : pub(crate) async fn csource_initial(st: &AppState, tenant: &TenantId, sub_id: &str) {
1888 260 : let Some(sub) = st
1889 260 : .store
1890 260 : .get(tenant, crate::registry::csr_kind(sub_id), sub_id)
1891 260 : .await
1892 260 : .ok()
1893 260 : .flatten()
1894 : else {
1895 0 : return;
1896 : };
1897 260 : if !is_active(&sub) {
1898 0 : return;
1899 260 : }
1900 260 : let ctx = sub_context(st, tenant, &sub).await;
1901 260 : let spec = crate::registry::spec_for_subscription(&sub);
1902 260 : let data: Vec<Value> = read_or_warn(
1903 260 : st.store.list(tenant, Kind::Registration).await,
1904 260 : "the registrations an initial Context Source Notification carries",
1905 : )
1906 260 : .into_iter()
1907 260 : .filter(|r| crate::registry::csr_matches_subscription(&sub, r, &ctx))
1908 260 : .map(|r| {
1909 71 : let mut p =
1910 71 : crate::registry::present_registration(&filter_csr(&spec, &r, &ctx), &ctx, false);
1911 71 : arrayify_entity_types(&mut p);
1912 71 : p
1913 71 : })
1914 260 : .collect();
1915 260 : if data.is_empty() {
1916 203 : return; // nothing currently matching ⇒ no initial notification
1917 57 : }
1918 57 : deliver_csource(st, tenant, &sub, data, &ctx, "newlyMatching").await;
1919 258 : }
1920 :
1921 : /// Registration copy reduced to the matching RegistrationInfo elements
1922 : /// (5.10.2.5 / 5.11.7 "filtered Context Source Registrations").
1923 81 : fn filter_csr(spec: &crate::registry::CsrSpec, reg: &Value, ctx: &Context) -> Value {
1924 81 : let mut out = reg.clone();
1925 81 : let matching: Vec<Value> = crate::registry::matching_infos(spec, reg, ctx)
1926 81 : .into_iter()
1927 81 : .cloned()
1928 81 : .collect();
1929 81 : if !matching.is_empty() {
1930 81 : if let Some(o) = out.as_object_mut() {
1931 81 : o.insert("information".into(), Value::Array(matching));
1932 81 : }
1933 0 : }
1934 81 : out
1935 81 : }
1936 :
1937 : /// Table 5.2.15-1 `timeout`: per-endpoint delivery deadline in milliseconds.
1938 : /// "The NGSI-LD system can override this value" — clamped to [100 ms, 30 s]
1939 : /// so one subscription cannot park a delivery task for minutes. Default 5 s
1940 : /// (the previous hard-coded deadline). HTTP only: the clause scopes it to
1941 : /// bindings that "always return a response".
1942 400 : fn endpoint_timeout_ms(ep: &serde_json::Map<String, Value>) -> u32 {
1943 400 : ep.get("timeout")
1944 400 : .and_then(Value::as_f64)
1945 400 : .filter(|t| *t > 0.0)
1946 400 : .map(|t| (t as u32).clamp(100, 30_000))
1947 400 : .unwrap_or(5_000)
1948 400 : }
1949 :
1950 : /// Table 5.2.15-1 `cooldown`: "Once a failure has occurred, minimum period of
1951 : /// time in milliseconds which shall elapse before attempting to make a
1952 : /// subsequent notification to the same endpoint after failure. If requests
1953 : /// are received before the cooldown period has expired, no notification is
1954 : /// sent." — i.e. matches inside the window are DROPPED, not queued.
1955 412 : fn in_cooldown(sub: &Value, now: chrono::DateTime<chrono::Utc>) -> bool {
1956 412 : let n = sub.get("notification");
1957 412 : let Some(cd) = n
1958 412 : .and_then(|n| n.get("endpoint"))
1959 412 : .and_then(|e| e.get("cooldown"))
1960 412 : .and_then(Value::as_f64)
1961 412 : .filter(|c| *c > 0.0)
1962 : else {
1963 400 : return false;
1964 : };
1965 : // the gate exists only "once a failure has occurred" and only until a
1966 : // success clears it — notification.status tracks exactly that (5.2.14.2)
1967 12 : if n.and_then(|n| n.get("status")).and_then(Value::as_str) != Some("failed") {
1968 4 : return false;
1969 8 : }
1970 8 : let Some(lf) = n.and_then(|n| n.get("lastFailure")).and_then(Value::as_str) else {
1971 0 : return false;
1972 : };
1973 8 : let Ok(t) = chrono::DateTime::parse_from_rfc3339(lf) else {
1974 0 : return false;
1975 : };
1976 8 : let elapsed = now
1977 8 : .signed_duration_since(t.with_timezone(&chrono::Utc))
1978 8 : .num_milliseconds();
1979 8 : (elapsed as f64) < cd
1980 412 : }
1981 :
1982 : #[allow(clippy::too_many_arguments)] // one param per notification dimension
1983 457 : async fn deliver_as(
1984 457 : st: &AppState,
1985 457 : tenant: &TenantId,
1986 457 : kind: Kind,
1987 457 : sub: &Value,
1988 457 : ntype: &str,
1989 457 : data: Vec<Value>,
1990 457 : ctx: &Context,
1991 457 : trigger_reason: Option<&str>,
1992 457 : ) {
1993 : // 5.2.12: a paused subscription (isActive false) and an expired one send
1994 : // nothing. Every caller that assembles data locally checks this first;
1995 : // the ones that arrive with data already assembled — the 5.8.6 inbound
1996 : // notification among them — did not, so the check belongs here too.
1997 457 : if !is_active(sub) {
1998 0 : return;
1999 457 : }
2000 457 : let sub_id = sub_str(sub, "id").unwrap_or_default().to_owned();
2001 457 : let Some(ep) = sub
2002 457 : .get("notification")
2003 457 : .and_then(|n| n.get("endpoint"))
2004 457 : .and_then(Value::as_object)
2005 : else {
2006 0 : return;
2007 : };
2008 457 : let Some(uri) = ep.get("uri").and_then(Value::as_str) else {
2009 0 : return;
2010 : };
2011 : // 5.8.1.4 consumer half: the internal CSR subscription's notifications
2012 : // are handled in-process (urn:antares:distsub:{tenant}\n{own sub id})
2013 457 : if let Some(own) = uri.strip_prefix("urn:antares:distsub:") {
2014 59 : if let (Some((_, own_id)), Some(handler)) =
2015 61 : (own.split_once('\n'), st.csource_notification.as_ref())
2016 : {
2017 59 : handler(st, tenant, own_id, trigger_reason, &data).await;
2018 2 : }
2019 59 : return;
2020 396 : }
2021 : // 6.3.8: the binding comes from the registry and nowhere else. Creation
2022 : // rejects an endpoint whose scheme no sink serves, so a stored row that
2023 : // still names one was hand-edited — it is dropped, never delivered
2024 : // through some other binding.
2025 396 : if st.sinks.sink_for_uri(uri).is_none() {
2026 0 : tracing::warn!(
2027 : "subscription {sub_id} endpoint {} has no registered binding",
2028 0 : redact_userinfo(uri)
2029 : );
2030 0 : return;
2031 396 : }
2032 : // endpoint.cooldown — drop (never queue) while the window is open.
2033 : // Before any bookkeeping: a suppressed notification was never sent, so
2034 : // timesSent/lastNotification must not move.
2035 396 : if in_cooldown(sub, chrono::Utc::now()) {
2036 0 : tracing::debug!("subscription {sub_id} in cooldown; notification suppressed (5.2.15)");
2037 0 : return;
2038 396 : }
2039 : // An open circuit is the same class of self-inflicted suppression: no
2040 : // request leaves the process, so Table 5.2.14.2-1 timesSent ("number of
2041 : // times that the notification has been sent") and lastNotification ("the
2042 : // instant when the last notification has been sent") must not move
2043 : // either. `is_open` returning false IS the half-open probe, so the
2044 : // check stays exactly once per attempt.
2045 : // A binding that opens no socket has no destination for the policy or
2046 : // the breaker to judge; every network binding is policed below.
2047 396 : let policed = st.sinks.sink_for_uri(uri).is_some_and(|s| s.network());
2048 396 : if policed && st.egress.is_open(tenant.as_str(), uri) {
2049 8 : tracing::debug!(
2050 : "notification to {} short-circuited (breaker open)",
2051 0 : redact_userinfo(uri)
2052 : );
2053 8 : return;
2054 388 : }
2055 388 : let accept = ep
2056 388 : .get("accept")
2057 388 : .and_then(Value::as_str)
2058 388 : .unwrap_or("application/json");
2059 388 : let now = now_iso();
2060 388 : let mut body = json!({
2061 388 : "id": format!("urn:ngsi-ld:{ntype}:{}", uuid::Uuid::new_v4()),
2062 388 : "type": ntype,
2063 388 : "subscriptionId": sub_id,
2064 388 : "notifiedAt": now,
2065 388 : "data": data,
2066 : });
2067 388 : if let Some(r) = trigger_reason {
2068 4 : body["triggerReason"] = Value::String(r.into());
2069 384 : }
2070 : // ADR-0020: the engine sees the notification the broker means to send,
2071 : // before it is encoded — the data entities are still the 5.2.14.1
2072 : // documents the subscription asked for, not a FeatureCollection or a
2073 : // set of @context-injected ones, so a projection lands on entities the
2074 : // engine can name. Before the bookkeeping below, because 5.8.6 moves
2075 : // timesSent for a notification that "shall be sent": one that is
2076 : // dropped here never was, exactly like a cooldown or an open breaker.
2077 : // The subject is the subscriber's, stored when the subscription was
2078 : // created — delivery is broker-initiated and there is no request here
2079 : // to read one off.
2080 : // With no engine attached the answer is Deliver and the subject is
2081 : // never read out of the subscription — this runs once per notification
2082 : // per subscription.
2083 388 : if let Some(engine) = &st.policy {
2084 16 : match crate::policy::pre_notify(
2085 16 : engine.as_ref(),
2086 16 : &crate::policy::stored_subject(tenant, sub),
2087 16 : sub,
2088 16 : &mut body,
2089 16 : ) {
2090 6 : crate::policy::NotifyDecision::Deliver => {}
2091 : crate::policy::NotifyDecision::Drop => {
2092 2 : tracing::debug!("subscription {sub_id} notification dropped by the policy engine");
2093 2 : return;
2094 : }
2095 8 : crate::policy::NotifyDecision::Filter(f) => {
2096 : // a condition the notification path cannot re-run is not a
2097 : // narrowing it can claim to have applied
2098 8 : if f.q.is_some() || f.scope_q.is_some() {
2099 2 : tracing::error!(
2100 : "policy engine {} narrowed a notification with a query; dropping it",
2101 0 : engine.name()
2102 : );
2103 2 : return;
2104 6 : }
2105 6 : let f = crate::repr::compacted_filter(&f, ctx);
2106 6 : if let Some(arr) = body.get_mut("data").and_then(Value::as_array_mut) {
2107 6 : for e in arr.iter_mut() {
2108 6 : f.project(e);
2109 6 : }
2110 0 : }
2111 : }
2112 : }
2113 372 : }
2114 : // 5.8.6: a subscription's ngsildConformance pins the notification format —
2115 : // amend the data entities per the 4.3.6.8 fallbacks.
2116 384 : if let Some(ver) = sub_str(sub, "ngsildConformance").and_then(crate::conformance::parse_version)
2117 : {
2118 0 : if let Some(d) = body.get_mut("data") {
2119 0 : crate::conformance::amend_payload(d, ver);
2120 0 : }
2121 384 : }
2122 384 : if accept == "application/ld+json" {
2123 : // JSON-LD notifications carry the @context inside each data entity
2124 : // (046_14: data[0] must contain @context; no Link header) — same rule
2125 : // over MQTT: with ld+json the @context travels in the body (7.2).
2126 2 : if let Some(arr) = body.get_mut("data").and_then(Value::as_array_mut) {
2127 2 : for e in arr.iter_mut() {
2128 2 : *e = inject_context(e.clone(), ctx);
2129 2 : }
2130 0 : }
2131 382 : }
2132 384 : if accept == "application/geo+json" {
2133 : // Table 5.3.1-1: with endpoint.accept application/geo+json, data is
2134 : // a FeatureCollection (5.2.30); if receiverInfo carries
2135 : // Prefer=body=json the FeatureCollection carries no @context.
2136 4 : let prefer_body_json = ep
2137 4 : .get("receiverInfo")
2138 4 : .and_then(Value::as_array)
2139 4 : .is_some_and(|ri| {
2140 2 : ri.iter().any(|kv| {
2141 2 : kv.get("key").and_then(Value::as_str) == Some("Prefer")
2142 2 : && kv.get("value").and_then(Value::as_str) == Some("body=json")
2143 2 : })
2144 2 : });
2145 4 : let entities = body["data"].as_array().cloned().unwrap_or_default();
2146 4 : let mut fc = crate::repr::to_geojson_collection(entities, None);
2147 4 : if !prefer_body_json {
2148 2 : fc["@context"] = crate::negotiate::served_context(ctx);
2149 2 : }
2150 4 : body["data"] = fc;
2151 380 : }
2152 384 : let receiver_info = kv_pairs(ep.get("receiverInfo"));
2153 :
2154 : // 6.3.22: a subscription living under a snapshot's synthetic tenant
2155 : // notifies with the NGSILD-Snapshot header and the OWNER tenant — the
2156 : // internal "snap-…" tenant never leaks.
2157 384 : let (hdr_tenant, snapshot_id) =
2158 384 : match crate::snapshots::snapshot_of_synth(st, tenant.as_str()).await {
2159 2 : Some((owner, sid)) => (owner, Some(sid)),
2160 382 : None => (tenant.clone(), None),
2161 : };
2162 :
2163 : // Prepared BEFORE the bookkeeping writeback so the optimistic stamp
2164 : // covers only the in-flight attempt (046_12_01 race). The parts are
2165 : // transport-neutral: the sink registered for the endpoint's scheme turns
2166 : // them into HTTP headers (6.3.8) or an MQTT metadata object (Table
2167 : // 7.2-2).
2168 384 : let mut info = receiver_info;
2169 384 : strip_reserved_markers(&mut info);
2170 384 : if hdr_tenant.as_str() != "default" {
2171 54 : info.push(("NGSILD-Tenant".into(), hdr_tenant.as_str().to_owned()));
2172 330 : }
2173 384 : if let Some(sid) = &snapshot_id {
2174 2 : info.push(("NGSILD-Snapshot".into(), sid.clone()));
2175 382 : }
2176 384 : let outbound = Outbound {
2177 384 : body,
2178 384 : accept: accept.to_owned(),
2179 384 : link: link_header_value(ctx),
2180 384 : receiver_info: info,
2181 384 : notifier_info: kv_pairs(ep.get("notifierInfo")),
2182 384 : };
2183 : // Bookkeeping BEFORE the send (5.8.6/5.2.14.2: lastNotification is the
2184 : // instant the notification is sent). The ETSI mock unblocks the test the
2185 : // moment the request ARRIVES, so a post-response-only writeback races the
2186 : // test's immediate Retrieve Subscription (CI flake on 046_12_01).
2187 : // Optimistic ok; a failed attempt is corrected right below — the transient
2188 : // window is the in-flight attempt itself, and the failure TPs wait for the
2189 : // attempt to resolve before asserting.
2190 : // One store call: the stamp is a fixed mutation, so a backend can write
2191 : // it as a single statement instead of locking the row across a round
2192 : // trip. At fan-out that lock is what serializes delivery.
2193 384 : let booked = st
2194 384 : .store
2195 384 : .record_delivery(tenant, kind, &sub_id, &now)
2196 384 : .await
2197 384 : .unwrap_or_else(|e| {
2198 0 : tracing::warn!("bookkeeping writeback failed: {e}");
2199 0 : None
2200 0 : });
2201 : // 5.8.6: notifications are sent for the subscriptions the broker holds.
2202 : // No row to book against means the subscription was deleted (or the
2203 : // store failed) between matching and delivery — nothing may be sent.
2204 384 : let Some(booked) = booked else {
2205 4 : return;
2206 : };
2207 380 : let mut prev_success = booked.prev_success;
2208 380 : mirror_bookkeeping(st, tenant, kind, &sub_id, Some(booked.doc));
2209 : // The notification endpoint is an egress destination like any other
2210 : // — policy check once, breaker consulted before the attempt.
2211 : // A refusal is a delivery failure for bookkeeping (status "failed",
2212 : // lastSuccess rolled back below) but never breaker state: the policy
2213 : // verdict says nothing about the endpoint's health.
2214 380 : let refused = policed
2215 376 : && match st.egress.check_destination(uri).await {
2216 372 : Ok(()) => false,
2217 4 : Err(e) => {
2218 4 : tracing::warn!(
2219 : "notification endpoint {} refused by egress policy: {e}",
2220 0 : redact_userinfo(uri)
2221 : );
2222 4 : true
2223 : }
2224 : };
2225 : // (delivered, timed_out): only a TIMEOUT-class failure feeds the breaker
2226 : // — that protects against peers that eat the deadline. An endpoint
2227 : // that ANSWERS (any status) is alive, costs only its own response time,
2228 : // and 6.3.8 says the notification shall be sent — suppressing sends to a
2229 : // responding host:port starves unrelated subscriptions sharing it.
2230 380 : let timeout_ms = endpoint_timeout_ms(ep);
2231 380 : let first = if refused {
2232 4 : Err((false, "refused by egress policy".to_owned()))
2233 : } else {
2234 376 : send_outbound(st, uri, timeout_ms, &outbound).await
2235 : };
2236 306 : let (ok, timed_out) = match &first {
2237 216 : Ok(()) => (true, false),
2238 90 : Err((t, _)) => (false, *t),
2239 : };
2240 306 : if policed && !refused {
2241 298 : if ok {
2242 212 : st.egress.record_success(tenant.as_str(), uri);
2243 212 : } else if timed_out {
2244 36 : st.egress.record_failure(tenant.as_str(), uri);
2245 50 : } else {
2246 50 : // the destination responded (or refused fast): alive — clear
2247 50 : // any stale consecutive-timeout state
2248 50 : st.egress.record_success(tenant.as_str(), uri);
2249 50 : }
2250 8 : }
2251 : // Delivery counters by binding (facade — no-op without the broker's
2252 : // recorder). The label is the sink's first scheme, so the two members of
2253 : // a family share one series.
2254 306 : let scheme = st
2255 306 : .sinks
2256 306 : .sink_for_uri(uri)
2257 306 : .and_then(|s| s.schemes().first().copied())
2258 306 : .unwrap_or("unknown");
2259 306 : if ok {
2260 216 : metrics::counter!("antares_notifications_sent_total", "scheme" => scheme).increment(1);
2261 216 : } else {
2262 90 : metrics::counter!("antares_notifications_failed_total", "scheme" => scheme).increment(1);
2263 90 : }
2264 306 : if !ok {
2265 : // 5.8.6 / 5.11.7: subscription status → "failed" on delivery failure;
2266 : // roll back the optimistic lastSuccess stamp.
2267 90 : let ts = now_iso();
2268 90 : let mut failed_doc: Option<Value> = None;
2269 90 : st.store
2270 90 : .mutate(tenant, kind, &sub_id, |doc| {
2271 90 : if let Some(o) = doc.as_object_mut() {
2272 90 : o.insert("status".into(), Value::String("failed".into()));
2273 90 : }
2274 90 : if let Some(n) = doc
2275 90 : .as_object_mut()
2276 90 : .and_then(|o| o.get_mut("notification"))
2277 90 : .and_then(Value::as_object_mut)
2278 : {
2279 90 : match prev_success.take() {
2280 0 : Some(v) => n.insert("lastSuccess".into(), v),
2281 90 : None => n.remove("lastSuccess"),
2282 : };
2283 90 : n.insert("lastFailure".into(), Value::String(ts.clone()));
2284 : // Table 5.2.14.2-1 timesFailed: "Number of times an
2285 : // unsuccessful response (or timeout) has been received
2286 : // when delivering the notification" — an output-only
2287 : // member implementations shall generate.
2288 90 : let failed = n.get("timesFailed").and_then(Value::as_i64).unwrap_or(0);
2289 90 : n.insert("timesFailed".into(), json!(failed + 1));
2290 90 : n.insert("status".into(), Value::String("failed".into()));
2291 0 : }
2292 90 : failed_doc = Some(doc.clone());
2293 90 : Ok::<(), antares_model::NgsiError>(())
2294 90 : })
2295 90 : .await
2296 90 : .unwrap_or_else(|e| {
2297 0 : tracing::warn!("failure-status writeback failed: {e}");
2298 0 : None
2299 0 : });
2300 90 : mirror_bookkeeping(st, tenant, kind, &sub_id, failed_doc);
2301 : // Retries are transport, not new notifications: they run on their
2302 : // own task (never on the request path, never delaying another
2303 : // subscription's delivery) and book only the final outcome — a
2304 : // success sets lastSuccess/status ok without touching timesSent;
2305 : // an exhausted policy leaves a dead letter.
2306 : #[cfg(not(target_arch = "wasm32"))]
2307 90 : if !refused && st.delivery.attempts > 1 {
2308 16 : let first_err = first.err().map(|(_, e)| e).unwrap_or_default();
2309 16 : let (st, tenant, uri) = (st.clone(), tenant.clone(), uri.to_owned());
2310 16 : crate::spawn(async move {
2311 16 : retry_and_settle(
2312 16 : &st, &tenant, kind, &sub_id, &uri, timeout_ms, outbound, first_err,
2313 16 : )
2314 16 : .await;
2315 12 : });
2316 74 : }
2317 216 : }
2318 381 : }
2319 :
2320 : /// A `KeyValuePair[]` member of `endpoint` (Table 5.2.15-1) as owned pairs.
2321 : /// A member that is not an array of well-formed pairs contributes nothing:
2322 : /// 5.2.12 validation at creation already refused a malformed one.
2323 768 : fn kv_pairs(v: Option<&Value>) -> Vec<(String, String)> {
2324 768 : v.and_then(Value::as_array)
2325 768 : .map(|a| {
2326 4 : a.iter()
2327 10 : .filter_map(|kv| {
2328 : Some((
2329 10 : kv.get("key")?.as_str()?.to_owned(),
2330 10 : kv.get("value")?.as_str()?.to_owned(),
2331 : ))
2332 10 : })
2333 4 : .collect()
2334 4 : })
2335 768 : .unwrap_or_default()
2336 768 : }
2337 :
2338 : /// 6.3.22 / 6.3.8: `NGSILD-Tenant` and `NGSILD-Snapshot` on a notification
2339 : /// are the broker's own statement of where the data came from, appended to
2340 : /// the `receiverInfo` pairs. The HTTP binding appends every pair it is
2341 : /// handed, so a subscriber naming one of the two in `receiverInfo` would
2342 : /// put a second value of it on the wire beside the broker's, and a receiver
2343 : /// reading "the" tenant of a notification could not tell which one the
2344 : /// broker meant. Ordinary custom headers are untouched.
2345 388 : fn strip_reserved_markers(info: &mut Vec<(String, String)>) {
2346 388 : info.retain(|(k, _)| {
2347 30 : !k.eq_ignore_ascii_case("NGSILD-Tenant") && !k.eq_ignore_ascii_case("NGSILD-Snapshot")
2348 30 : });
2349 388 : }
2350 :
2351 : /// One attempt on the wire, through the binding the registry holds for the
2352 : /// endpoint's scheme (6.3.8). `Err((timed_out, why))`: only a timeout-class
2353 : /// failure feeds the breaker — an endpoint that answers is alive.
2354 394 : async fn send_outbound(
2355 394 : st: &AppState,
2356 394 : uri: &str,
2357 394 : timeout_ms: u32,
2358 394 : outbound: &Outbound,
2359 394 : ) -> Result<(), (bool, String)> {
2360 394 : let Some(sink) = st.sinks.sink_for_uri(uri) else {
2361 2 : return Err((
2362 2 : false,
2363 2 : format!(
2364 2 : "no notification binding registered for {}",
2365 2 : redact_userinfo(uri)
2366 2 : ),
2367 2 : ));
2368 : };
2369 392 : sink.deliver(
2370 392 : uri,
2371 392 : outbound,
2372 392 : std::time::Duration::from_millis(u64::from(timeout_ms)),
2373 392 : )
2374 392 : .await
2375 318 : .map_err(|e| (e.timed_out, e.message))
2376 320 : }
2377 :
2378 : static DEAD_LETTERS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
2379 :
2380 : /// Dead letters written by this process since start (`/q/health`
2381 : /// deadLetters); the letters themselves live in the store.
2382 98 : pub(crate) fn dead_letters_written() -> u64 {
2383 98 : DEAD_LETTERS.load(std::sync::atomic::Ordering::Relaxed)
2384 98 : }
2385 :
2386 : /// The retries the delivery policy grants after a failed first attempt,
2387 : /// then the settlement: lastSuccess/status ok on success, a dead letter
2388 : /// when the policy is exhausted.
2389 : #[cfg(not(target_arch = "wasm32"))]
2390 : #[allow(clippy::too_many_arguments)] // one param per piece of the attempt's state
2391 16 : async fn retry_and_settle(
2392 16 : st: &AppState,
2393 16 : tenant: &TenantId,
2394 16 : kind: Kind,
2395 16 : sub_id: &str,
2396 16 : uri: &str,
2397 16 : timeout_ms: u32,
2398 16 : outbound: Outbound,
2399 16 : first_err: String,
2400 16 : ) {
2401 16 : let policy = st.delivery;
2402 16 : let started = std::time::Instant::now();
2403 16 : let first_at = now_iso();
2404 16 : let mut made = 1u32;
2405 16 : let mut last_err = first_err.clone();
2406 24 : while let Some(delay) = policy.next_delay(made, started.elapsed()) {
2407 20 : tokio::time::sleep(delay).await;
2408 : // the subscription may have gone, or its endpoint may have tripped
2409 : // the breaker meanwhile — a retry is still one more attempt
2410 16 : if st
2411 16 : .store
2412 16 : .get(tenant, kind, sub_id)
2413 16 : .await
2414 16 : .ok()
2415 16 : .flatten()
2416 16 : .is_none()
2417 : {
2418 4 : return;
2419 12 : }
2420 12 : made += 1;
2421 12 : match send_outbound(st, uri, timeout_ms, &outbound).await {
2422 : Ok(()) => {
2423 4 : st.egress.record_success(tenant.as_str(), uri);
2424 4 : metrics::counter!("antares_notifications_retried_total", "outcome" => "ok")
2425 4 : .increment(1);
2426 4 : let ts = now_iso();
2427 4 : let mut retried_doc: Option<Value> = None;
2428 4 : st.store
2429 4 : .mutate(tenant, kind, sub_id, |doc| {
2430 4 : if let Some(o) = doc.as_object_mut() {
2431 4 : o.remove("status");
2432 4 : }
2433 4 : if let Some(n) = doc
2434 4 : .as_object_mut()
2435 4 : .and_then(|o| o.get_mut("notification"))
2436 4 : .and_then(Value::as_object_mut)
2437 4 : {
2438 4 : n.insert("lastSuccess".into(), Value::String(ts.clone()));
2439 4 : n.insert("status".into(), Value::String("ok".into()));
2440 4 : }
2441 4 : retried_doc = Some(doc.clone());
2442 4 : Ok::<(), antares_model::NgsiError>(())
2443 4 : })
2444 4 : .await
2445 4 : .unwrap_or_else(|e| {
2446 0 : tracing::warn!("retry bookkeeping writeback failed: {e}");
2447 0 : None
2448 0 : });
2449 4 : mirror_bookkeeping(st, tenant, kind, sub_id, retried_doc);
2450 4 : return;
2451 : }
2452 8 : Err((timed_out, e)) => {
2453 8 : if timed_out {
2454 0 : st.egress.record_failure(tenant.as_str(), uri);
2455 8 : } else {
2456 8 : st.egress.record_success(tenant.as_str(), uri);
2457 8 : }
2458 8 : last_err = e;
2459 : }
2460 : }
2461 : }
2462 4 : metrics::counter!("antares_notifications_retried_total", "outcome" => "dead").increment(1);
2463 4 : let letter = dead_letter(
2464 4 : sub_id, uri, timeout_ms, &outbound, made, &first_err, &last_err, &first_at,
2465 : );
2466 4 : let id = letter["id"].as_str().unwrap_or_default().to_owned();
2467 4 : match st.store.create(tenant, Kind::DeadLetter, &id, letter).await {
2468 : Ok(_) => {
2469 4 : DEAD_LETTERS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
2470 4 : tracing::warn!(
2471 : "notification for {sub_id} to {} dead-lettered after {made} attempts: {last_err}",
2472 0 : redact_userinfo(uri)
2473 : );
2474 : }
2475 0 : Err(e) => tracing::error!("dead letter for {sub_id} could not be stored: {e}"),
2476 : }
2477 12 : }
2478 :
2479 : /// The dead-letter document: everything a replay needs to send the very
2480 : /// same request again, plus the attempt history.
2481 : #[allow(clippy::too_many_arguments)] // one param per stored letter member
2482 4 : fn dead_letter(
2483 4 : sub_id: &str,
2484 4 : uri: &str,
2485 4 : timeout_ms: u32,
2486 4 : outbound: &Outbound,
2487 4 : attempts: u32,
2488 4 : first_err: &str,
2489 4 : last_err: &str,
2490 4 : first_at: &str,
2491 4 : ) -> Value {
2492 4 : let mut doc = json!({
2493 4 : "id": format!("urn:ngsi-ld:DeadLetter:{}", uuid::Uuid::new_v4()),
2494 4 : "type": "DeadLetter",
2495 4 : "subscriptionId": sub_id,
2496 4 : "uri": uri,
2497 4 : "timeoutMs": timeout_ms,
2498 4 : "attempts": attempts,
2499 4 : "firstError": first_err,
2500 4 : "lastError": last_err,
2501 4 : "firstAt": first_at,
2502 4 : "lastAt": now_iso(),
2503 : });
2504 4 : doc["binding"] = json!(antares_notifier::SinkRegistry::scheme_of(uri));
2505 4 : doc["payload"] = outbound.body.clone();
2506 4 : doc["accept"] = json!(outbound.accept);
2507 4 : doc["link"] = json!(outbound.link);
2508 4 : doc["receiverInfo"] = json!(outbound.receiver_info);
2509 4 : doc["notifierInfo"] = json!(outbound.notifier_info);
2510 4 : doc
2511 4 : }
2512 :
2513 : /// Replay one dead letter through the same binding, once. `Ok` = delivered
2514 : /// (the caller deletes the letter); `Err` carries the failure text.
2515 12 : pub(crate) async fn replay_dead_letter(st: &AppState, letter: &Value) -> Result<(), String> {
2516 12 : let uri = letter["uri"].as_str().ok_or("dead letter without uri")?;
2517 10 : let timeout_ms = letter["timeoutMs"].as_u64().unwrap_or(5_000) as u32;
2518 10 : let outbound = Outbound::from_dead_letter(letter)?;
2519 : // the egress policy of the moment applies, exactly as for a fresh send
2520 8 : if st.sinks.sink_for_uri(uri).is_some_and(|s| s.network()) {
2521 6 : st.egress
2522 6 : .check_destination(uri)
2523 6 : .await
2524 6 : .map_err(|e| e.to_string())?;
2525 2 : }
2526 6 : send_outbound(st, uri, timeout_ms, &outbound)
2527 6 : .await
2528 6 : .map_err(|(_, e)| e)
2529 12 : }
2530 :
2531 : pub(crate) use antares_notifier::{redact_userinfo, Outbound};
2532 :
2533 : /// The matcher reads subscriptions from the
2534 : /// SubMirror, so every notification bookkeeping writeback must be applied
2535 : /// there too — otherwise the mirror copy never gains
2536 : /// `notification.lastNotification` and 5.2.12 `throttling` suppresses
2537 : /// nothing. In-process apply only: a KV write per notification would not
2538 : /// scale to the 100k-sub target, so in bus=nats multi-pod deployments the
2539 : /// throttling window is per-pod approximate.
2540 : /// Known ceiling: exact distributed throttling = per-notification KV sync or a
2541 : /// store read in `throttled()`; add if a deployment needs the strict window.
2542 : /// The mirror learns the counters from the document the writeback just
2543 : /// committed under the row lock; `None` (no row) leaves it untouched.
2544 474 : fn mirror_bookkeeping(
2545 474 : st: &AppState,
2546 474 : tenant: &TenantId,
2547 474 : kind: Kind,
2548 474 : sub_id: &str,
2549 474 : doc: Option<Value>,
2550 474 : ) {
2551 474 : if kind != Kind::Subscription {
2552 4 : return;
2553 470 : }
2554 470 : if let (Some(m), Some(doc)) = (&st.sub_mirror, doc) {
2555 330 : m.apply(tenant.as_str(), sub_id, Some(doc));
2556 330 : }
2557 474 : }
2558 :
2559 : #[cfg(all(test, not(target_arch = "wasm32")))]
2560 : mod deleted_subscription_delivery {
2561 : use super::*;
2562 :
2563 : /// 5.8.6: notifications are sent for the subscriptions a Context Broker
2564 : /// holds — a subscription deleted between matching and delivery no
2565 : /// longer exists, so its endpoint must receive nothing.
2566 : #[tokio::test(flavor = "multi_thread")]
2567 4 : async fn deleted_subscription_receives_no_notification() {
2568 4 : crate::allow_private();
2569 4 : let st = AppState::new("antares-deleted-sub-test".into());
2570 4 : let count = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
2571 4 : let c = count.clone();
2572 4 : let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
2573 4 : .await
2574 4 : .expect("bind");
2575 4 : let addr = listener.local_addr().expect("addr");
2576 4 : let app = axum::Router::new().route(
2577 4 : "/notify",
2578 4 : axum::routing::post(move || {
2579 0 : let c = c.clone();
2580 0 : async move {
2581 0 : c.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
2582 0 : axum::http::StatusCode::OK
2583 0 : }
2584 0 : }),
2585 : );
2586 4 : tokio::spawn(async move {
2587 4 : axum::serve(listener, app).await.expect("serve");
2588 0 : });
2589 4 : let tenant = TenantId::new("default").expect("tenant");
2590 4 : let ctx = antares_jsonld::Loader::new().core();
2591 : // the sub doc is a snapshot whose row is NOT in the store — the
2592 : // deleted-concurrently case
2593 4 : let sub = json!({
2594 4 : "id": "urn:ngsi-ld:Subscription:ghost",
2595 4 : "type": "Subscription",
2596 4 : "entities": [{"type": "Vehicle"}],
2597 4 : "notification": {"endpoint": {"uri": format!("http://{addr}/notify")}},
2598 : });
2599 4 : let data = vec![json!({"id": "urn:ngsi-ld:Vehicle:1", "type": "Vehicle"})];
2600 4 : deliver_as(
2601 4 : &st,
2602 4 : &tenant,
2603 4 : Kind::Subscription,
2604 4 : &sub,
2605 4 : "Notification",
2606 4 : data,
2607 4 : &ctx,
2608 4 : None,
2609 4 : )
2610 4 : .await;
2611 4 : assert_eq!(
2612 4 : count.load(std::sync::atomic::Ordering::SeqCst),
2613 4 : 0,
2614 4 : "a deleted subscription's endpoint must receive NO notification"
2615 4 : );
2616 4 : }
2617 : }
2618 :
2619 : #[cfg(test)]
2620 : mod interval_tests {
2621 : use super::*;
2622 : use serde_json::json;
2623 :
2624 : /// 5.2.12 Table 5.2.12-1 bounds `Subscription.timeInterval` only as
2625 : /// "greater than 0", so a client may name an interval whose milliseconds
2626 : /// do not fit in the epoch arithmetic. A firing instant the broker cannot
2627 : /// represent must read as far in the future, never as far in the past: a
2628 : /// wrapped anchor is a NEGATIVE instant, which reports the subscription
2629 : /// as permanently due and fires it on every tick, each firing running the
2630 : /// subscription's whole query.
2631 : #[test]
2632 4 : fn an_interval_too_large_to_schedule_is_never_due_rather_than_always() {
2633 4 : let sub = json!({
2634 4 : "id": "urn:ngsi-ld:Subscription:huge",
2635 4 : "type": "Subscription",
2636 4 : "createdAt": "2026-01-01T00:00:00Z",
2637 : });
2638 4 : let now = chrono::Utc::now().timestamp_millis();
2639 12 : for interval in [1e18, 1e30, f64::MAX] {
2640 12 : let due = due_at_ms(&sub, interval);
2641 12 : assert!(
2642 12 : due > now,
2643 : "timeInterval {interval} is due at {due}, which is not after {now}"
2644 : );
2645 : }
2646 : // and an interval that DOES fit still schedules exactly one interval
2647 : // past the anchor, so the guard costs the ordinary case nothing
2648 4 : let anchor = chrono::DateTime::parse_from_rfc3339("2026-01-01T00:00:00Z")
2649 4 : .expect("anchor")
2650 4 : .timestamp_millis();
2651 4 : assert_eq!(due_at_ms(&sub, 30.0), anchor + 30_000);
2652 4 : }
2653 :
2654 : /// A subscriber cannot put its own value of a marker the broker sets on
2655 : /// the wire: the pair is dropped before the broker appends its own, and
2656 : /// the drop is by ASCII case-insensitive name, since a header name is
2657 : /// case-insensitive (IETF RFC 9110 clause 5.1) and the pair would
2658 : /// otherwise slip past under a different spelling.
2659 : #[test]
2660 4 : fn a_subscriber_cannot_add_its_own_notification_markers() {
2661 4 : let mut info = vec![
2662 4 : ("Authorization".to_owned(), "Bearer t".to_owned()),
2663 4 : ("ngsild-tenant".to_owned(), "victim".to_owned()),
2664 4 : ("NGSILD-Tenant".to_owned(), "victim".to_owned()),
2665 4 : (
2666 4 : "NGSILD-SNAPSHOT".to_owned(),
2667 4 : "urn:ngsi-ld:Snapshot:x".to_owned(),
2668 4 : ),
2669 4 : ("X-NGSILD-Tenant".to_owned(), "kept".to_owned()),
2670 : ];
2671 4 : strip_reserved_markers(&mut info);
2672 4 : assert_eq!(
2673 : info,
2674 4 : [
2675 4 : ("Authorization".to_owned(), "Bearer t".to_owned()),
2676 4 : ("X-NGSILD-Tenant".to_owned(), "kept".to_owned()),
2677 4 : ]
2678 : );
2679 4 : }
2680 :
2681 : /// The sweep's own clock takes the same offset, so the same overflow
2682 : /// would park (or un-park) every tenant's sweep, not just this
2683 : /// subscription: `next_sub`/`next_csub` are process-wide minima.
2684 : #[test]
2685 4 : fn the_sweep_clock_offset_survives_an_unschedulable_interval() {
2686 4 : let now_ms = chrono::Utc::now().timestamp_millis();
2687 8 : for interval in [1e18, f64::MAX] {
2688 8 : let next = now_ms.saturating_add(interval_offset_ms(interval));
2689 8 : assert!(next > now_ms, "sweep clock went backwards for {interval}");
2690 : }
2691 4 : assert_eq!(interval_offset_ms(1.5), 1500);
2692 4 : }
2693 : }
2694 :
2695 : #[cfg(test)]
2696 : mod endpoint_tests {
2697 : use super::*;
2698 : use serde_json::json;
2699 :
2700 20 : fn ep(v: Value) -> serde_json::Map<String, Value> {
2701 20 : v.as_object().expect("map").clone()
2702 20 : }
2703 :
2704 : /// Clause 4.21 + Table 5.2.14.1-1: notification `pick`/`omit` are
2705 : /// "a valid attribute projection language string as per clause 4.21", so a
2706 : /// LinkedEntityTerm (`ProjectionTerm = AttrName *1(LinkedEntityTerm)`) must
2707 : /// constrain the Linked Entity, exactly as it does on the query path.
2708 : #[test]
2709 4 : fn notification_projection_parses_linked_entity_terms() {
2710 4 : let ctx = antares_jsonld::Loader::new().core();
2711 4 : let sub = json!({
2712 4 : "notification": { "pick": ["id", "type", "refDevice{type}"] }
2713 : });
2714 4 : let shape = notif_shape(&sub, &ctx);
2715 4 : let pick = shape.repr.pick.expect("pick parsed");
2716 :
2717 12 : let linked = pick.iter().find(|n| n.raw == "refDevice").expect(
2718 4 : "refDevice must survive as its own term, not as the literal \"refDevice{type}\"",
2719 : );
2720 4 : let children = linked
2721 4 : .children
2722 4 : .as_ref()
2723 4 : .expect("the {…} term must become children so the Linked Entity is constrained");
2724 4 : assert!(
2725 4 : children.iter().any(|c| c.raw == "type"),
2726 : "refDevice{{type}} must select `type` inside the Linked Entity"
2727 : );
2728 4 : }
2729 :
2730 : /// Table 5.2.15-1 `timeout`: honored, clamped, defaulted.
2731 : #[test]
2732 4 : fn endpoint_timeout_is_honored_clamped_and_defaulted() {
2733 4 : assert_eq!(endpoint_timeout_ms(&ep(json!({"timeout": 1500}))), 1500);
2734 4 : assert_eq!(endpoint_timeout_ms(&ep(json!({}))), 5_000, "default");
2735 : // "The NGSI-LD system can override this value" — the clamp is that
2736 : // override, keeping delivery tasks bounded
2737 4 : assert_eq!(endpoint_timeout_ms(&ep(json!({"timeout": 600000}))), 30_000);
2738 4 : assert_eq!(endpoint_timeout_ms(&ep(json!({"timeout": 1}))), 100);
2739 : // creation rejects <=0, but a hand-edited row must not panic
2740 4 : assert_eq!(endpoint_timeout_ms(&ep(json!({"timeout": -5}))), 5_000);
2741 4 : }
2742 :
2743 : /// Table 5.2.15-1 `cooldown`: gate opens only after a
2744 : /// failure and closes once the window elapses or a success lands.
2745 : #[test]
2746 4 : fn cooldown_gates_only_failed_subscriptions_within_the_window() {
2747 4 : let now = chrono::Utc::now();
2748 4 : let recent = (now - chrono::Duration::milliseconds(500)).to_rfc3339();
2749 4 : let old = (now - chrono::Duration::milliseconds(5_000)).to_rfc3339();
2750 12 : let sub = |status: &str, last_failure: &str| {
2751 12 : json!({
2752 12 : "notification": {
2753 12 : "status": status,
2754 12 : "lastFailure": last_failure,
2755 12 : "endpoint": {"uri": "http://x/n", "cooldown": 2000}
2756 : }
2757 : })
2758 12 : };
2759 4 : assert!(
2760 4 : in_cooldown(&sub("failed", &recent), now),
2761 : "failed 0.5s ago, 2s cooldown ⇒ suppressed"
2762 : );
2763 4 : assert!(
2764 4 : !in_cooldown(&sub("failed", &old), now),
2765 : "failure outside the window ⇒ delivered"
2766 : );
2767 4 : assert!(
2768 4 : !in_cooldown(&sub("ok", &recent), now),
2769 : "a success clears the gate — status is not \"failed\""
2770 : );
2771 4 : let no_cooldown = json!({
2772 4 : "notification": {
2773 4 : "status": "failed", "lastFailure": recent,
2774 4 : "endpoint": {"uri": "http://x/n"}
2775 : }
2776 : });
2777 4 : assert!(
2778 4 : !in_cooldown(&no_cooldown, now),
2779 : "no cooldown member ⇒ no gate"
2780 : );
2781 4 : }
2782 : }
2783 :
2784 : #[cfg(test)]
2785 : mod clause_5_8_1 {
2786 : use super::*;
2787 :
2788 : /// 5.8.1.4: "the status of the Subscription changes automatically to
2789 : /// \"expired\", so that notifications will no longer be sent" — an
2790 : /// expiresAt spelled without a seconds fraction must count as expired
2791 : /// the moment now (spelled with milliseconds) passes it. A raw
2792 : /// lexicographic compare ranks 'Z' above '.' and keeps the
2793 : /// subscription alive for the whole boundary second.
2794 : #[test]
2795 4 : fn expiry_compare_survives_fraction_spellings() {
2796 4 : let now = crate::state::now_iso();
2797 4 : let secs = &now[..19];
2798 4 : std::thread::sleep(std::time::Duration::from_millis(5));
2799 4 : let sub = serde_json::json!({ "expiresAt": format!("{secs}Z") });
2800 4 : assert!(
2801 4 : !is_active(&sub),
2802 : "expiresAt {secs}Z lies in the past and must expire the subscription"
2803 : );
2804 : // a genuinely future expiry stays active
2805 4 : let sub = serde_json::json!({ "expiresAt": "2999-01-01T00:00:00Z" });
2806 4 : assert!(is_active(&sub));
2807 4 : }
2808 : }
2809 :
2810 : #[cfg(test)]
2811 : mod clause_5_3_3 {
2812 : use super::*;
2813 : use serde_json::json;
2814 :
2815 : /// 5.3.2 triggerReason + 5.3.3 TriggerReasonEnumeration: newlyMatching
2816 : /// (did not match -> matches), updated (matched -> still matches),
2817 : /// noLongerMatching (matched -> no longer / deleted); no notification
2818 : /// when neither side matches.
2819 : #[test]
2820 4 : fn trigger_reason_enumeration() {
2821 4 : let ctx = antares_jsonld::Loader::new().core();
2822 4 : let sub = json!({"entities": [
2823 4 : {"type": "https://uri.etsi.org/ngsi-ld/default-context/Building"}]});
2824 8 : let reg = |t: &str| {
2825 8 : json!({"id": "urn:csr:1", "type": "ContextSourceRegistration",
2826 8 : "endpoint": "http://peer:9090",
2827 8 : "information": [{"entities": [
2828 8 : {"type": format!("https://uri.etsi.org/ngsi-ld/default-context/{t}")}]}]})
2829 8 : };
2830 4 : let hit = reg("Building");
2831 4 : let miss = reg("Vehicle");
2832 4 : assert_eq!(
2833 4 : csource_trigger(&sub, None, Some(&hit), &ctx),
2834 : Some("newlyMatching")
2835 : );
2836 4 : assert_eq!(
2837 4 : csource_trigger(&sub, Some(&miss), Some(&hit), &ctx),
2838 : Some("newlyMatching"),
2839 : "an update that STARTS matching is newlyMatching"
2840 : );
2841 4 : assert_eq!(
2842 4 : csource_trigger(&sub, Some(&hit), Some(&hit), &ctx),
2843 : Some("updated")
2844 : );
2845 4 : assert_eq!(
2846 4 : csource_trigger(&sub, Some(&hit), None, &ctx),
2847 : Some("noLongerMatching"),
2848 : "deletion of a matching registration"
2849 : );
2850 4 : assert_eq!(
2851 4 : csource_trigger(&sub, Some(&hit), Some(&miss), &ctx),
2852 : Some("noLongerMatching"),
2853 : "an update that STOPS matching"
2854 : );
2855 4 : assert_eq!(
2856 4 : csource_trigger(&sub, Some(&miss), Some(&miss), &ctx),
2857 : None,
2858 : "never-matching changes produce no notification"
2859 : );
2860 4 : }
2861 : }
2862 :
2863 : #[cfg(test)]
2864 : mod clause_5_2_33 {
2865 : use super::*;
2866 : use serde_json::json;
2867 :
2868 : /// Table 5.2.33-1: id is String or String[]; "id takes precedence over
2869 : /// idPattern" when a selector carries both.
2870 : #[test]
2871 4 : fn selector_id_array_and_precedence() {
2872 4 : let ctx = antares_jsonld::Loader::new().core();
2873 4 : let doc = json!({"id": "urn:x:A", "type": ["T"]});
2874 16 : let sub = |e: serde_json::Value| json!({"entities": [e]});
2875 4 : assert!(!selector_match(
2876 4 : &sub(json!({"type": "T", "idPattern": "^urn:x:B"})),
2877 4 : &doc,
2878 4 : &ctx
2879 4 : ));
2880 4 : assert!(
2881 4 : selector_match(
2882 4 : &sub(json!({"type": "T", "id": "urn:x:A", "idPattern": "^urn:x:B"})),
2883 4 : &doc,
2884 4 : &ctx
2885 : ),
2886 : "id takes precedence over idPattern"
2887 : );
2888 4 : assert!(selector_match(
2889 4 : &sub(json!({"type": "T", "id": ["urn:x:A", "urn:x:C"]})),
2890 4 : &doc,
2891 4 : &ctx
2892 : ));
2893 4 : assert!(
2894 4 : !selector_match(&sub(json!({"type": "T", "id": ["urn:x:B"]})), &doc, &ctx),
2895 : "an id array not containing the entity id must not match"
2896 : );
2897 4 : }
2898 : }
2899 :
2900 : /// Availability of the change→notification pipeline itself: the consumer
2901 : /// task must survive a panic, and its queue must stay bounded.
2902 : #[cfg(all(test, not(target_arch = "wasm32")))]
2903 : mod change_pipeline {
2904 : use super::*;
2905 : use serde_json::json;
2906 : use std::sync::atomic::{AtomicUsize, Ordering};
2907 : use std::sync::Arc;
2908 : use tower::ServiceExt;
2909 :
2910 4228 : async fn post(st: &AppState, uri: &str, body: Value) -> u16 {
2911 4228 : let body = body.to_string();
2912 4228 : crate::router(st.clone())
2913 4228 : .oneshot(
2914 4228 : axum::http::Request::builder()
2915 4228 : .method("POST")
2916 4228 : .uri(uri)
2917 4228 : .header("Content-Type", "application/json")
2918 4228 : .header("Content-Length", body.len())
2919 4228 : .body(axum::body::Body::from(body))
2920 4228 : .expect("request"),
2921 4228 : )
2922 4228 : .await
2923 4228 : .expect("response")
2924 4228 : .status()
2925 4228 : .as_u16()
2926 4228 : }
2927 :
2928 : /// An endpoint that answers 200 and counts the notifications it got.
2929 20 : async fn counting_endpoint() -> (String, Arc<AtomicUsize>) {
2930 20 : let hits: Arc<AtomicUsize> = Arc::default();
2931 20 : let seen = hits.clone();
2932 20 : let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
2933 20 : .await
2934 20 : .expect("bind");
2935 20 : let addr = listener.local_addr().expect("addr");
2936 20 : let app = axum::Router::new().route(
2937 20 : "/notify",
2938 100 : axum::routing::post(move || {
2939 100 : let seen = seen.clone();
2940 100 : async move {
2941 100 : seen.fetch_add(1, Ordering::SeqCst);
2942 100 : axum::http::StatusCode::OK
2943 100 : }
2944 100 : }),
2945 : );
2946 20 : tokio::spawn(async move { axum::serve(listener, app).await.expect("serve") });
2947 20 : (format!("http://{addr}/notify"), hits)
2948 20 : }
2949 :
2950 16 : async fn subscribe(st: &AppState, id: &str, uri: &str, timeout_ms: u32) {
2951 16 : let status = post(
2952 16 : st,
2953 16 : "/ngsi-ld/v1/subscriptions",
2954 16 : json!({
2955 16 : "id": format!("urn:ngsi-ld:Subscription:{id}"),
2956 16 : "type": "Subscription",
2957 16 : "entities": [{"type": "Vehicle"}],
2958 16 : "notification": {"endpoint": {"uri": uri, "timeout": timeout_ms}},
2959 16 : }),
2960 16 : )
2961 16 : .await;
2962 16 : assert_eq!(status, 201, "subscription created");
2963 16 : }
2964 :
2965 4204 : async fn create_vehicle(st: &AppState, n: usize) -> u16 {
2966 4204 : post(
2967 4204 : st,
2968 4204 : "/ngsi-ld/v1/entities",
2969 4204 : json!({
2970 4204 : "id": format!("urn:ngsi-ld:Vehicle:pipe{n}"),
2971 4204 : "type": "Vehicle",
2972 4204 : "speed": {"type": "Property", "value": n},
2973 4204 : }),
2974 4204 : )
2975 4204 : .await
2976 4204 : }
2977 :
2978 : /// Table 5.2.14.2-1 timesSent / lastNotification: the mirror the matcher
2979 : /// reads carries the counters the delivery writeback committed, and the
2980 : /// store row agrees — one writeback, no second read.
2981 : #[tokio::test(flavor = "multi_thread")]
2982 4 : async fn mirror_carries_the_booked_counters() {
2983 4 : crate::allow_private();
2984 4 : let (uri, hits) = counting_endpoint().await;
2985 4 : let mut st = AppState::new("antares-mirror-booked".into());
2986 4 : crate::wire(&mut st).await;
2987 4 : subscribe(&st, "booked", &uri, 30_000).await;
2988 4 : assert_eq!(create_vehicle(&st, 7).await, 201);
2989 4 : let deadline = std::time::Instant::now()
2990 4 : + std::time::Duration::from_secs(10 * crate::state::slow_factor());
2991 8 : while hits.load(Ordering::SeqCst) < 1 && std::time::Instant::now() < deadline {
2992 4 : tokio::time::sleep(std::time::Duration::from_millis(50)).await;
2993 : }
2994 4 : assert_eq!(hits.load(Ordering::SeqCst), 1, "one notification delivered");
2995 4 : let id = "urn:ngsi-ld:Subscription:booked";
2996 4 : let mirrored = st
2997 4 : .sub_mirror
2998 4 : .as_ref()
2999 4 : .expect("mirror wired")
3000 4 : .docs("default")
3001 4 : .into_iter()
3002 4 : .find(|d| d["id"] == id)
3003 4 : .expect("subscription mirrored");
3004 4 : assert_eq!(mirrored["notification"]["timesSent"], json!(1));
3005 4 : assert!(mirrored["notification"]["lastNotification"].is_string());
3006 4 : let tenant = TenantId::new("default").expect("tenant");
3007 4 : let stored = st
3008 4 : .store
3009 4 : .get(&tenant, Kind::Subscription, id)
3010 4 : .await
3011 4 : .expect("store")
3012 4 : .expect("row");
3013 4 : assert_eq!(stored["notification"]["timesSent"], json!(1));
3014 4 : assert_eq!(
3015 4 : stored["notification"]["lastNotification"],
3016 4 : mirrored["notification"]["lastNotification"]
3017 4 : );
3018 4 : }
3019 :
3020 : /// 5.8.6: a matching change notifies, and a panic around ONE change must
3021 : /// not end notification delivery for the process — the NEXT change still
3022 : /// notifies. The lock poisoning here used to make the matcher itself
3023 : /// panic; since the mirrors recover from poisoning, the first change may
3024 : /// legitimately deliver too. The contract under test is therefore that
3025 : /// the SECOND change's notification arrives — not a count, which only
3026 : /// measured the race between the two deliveries and failed either way on
3027 : /// slow machines.
3028 : #[tokio::test(flavor = "multi_thread")]
3029 4 : async fn panicking_change_does_not_stop_the_next_notification() {
3030 4 : crate::allow_private();
3031 4 : let (uri, hits) = counting_endpoint().await;
3032 4 : let mut st = AppState::new("antares-panic-guard".into());
3033 4 : crate::wire(&mut st).await;
3034 4 : subscribe(&st, "guard", &uri, 2_000).await;
3035 : // Poison the mirror lock while a change is in flight: whatever the
3036 : // matcher does with that (recover, or panic into the supervision
3037 : // boundary), the pipeline must keep delivering afterwards.
3038 4 : let mirror = st.sub_mirror.clone().expect("mirror");
3039 4 : mirror.poison();
3040 4 : assert_eq!(create_vehicle(&st, 1).await, 201);
3041 4 : assert_eq!(create_vehicle(&st, 2).await, 201);
3042 4 : let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
3043 6 : while hits.load(Ordering::SeqCst) < 1 && std::time::Instant::now() < deadline {
3044 2 : tokio::time::sleep(std::time::Duration::from_millis(50)).await;
3045 : }
3046 4 : let got = hits.load(Ordering::SeqCst);
3047 4 : assert!(
3048 4 : (1..=2).contains(&got),
3049 : "delivery must survive the poisoned change: expected 1 or 2 notifications, got {got}"
3050 : );
3051 : // and the pipeline is still alive for a THIRD change after the dust
3052 : // settles — the actual supervision contract
3053 4 : let before = hits.load(Ordering::SeqCst);
3054 4 : assert_eq!(create_vehicle(&st, 3).await, 201);
3055 4 : let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
3056 6 : while hits.load(Ordering::SeqCst) <= before && std::time::Instant::now() < deadline {
3057 4 : tokio::time::sleep(std::time::Duration::from_millis(50)).await;
3058 4 : }
3059 4 : assert!(
3060 4 : hits.load(Ordering::SeqCst) > before,
3061 4 : "a change created after the poisoned one must still notify"
3062 4 : );
3063 4 : }
3064 :
3065 : /// The matcher queue is bounded: behind a stalled subscriber the excess
3066 : /// changes are dropped and counted instead of growing without limit.
3067 : #[tokio::test(flavor = "multi_thread")]
3068 4 : async fn overflowing_change_queue_drops_and_counts() {
3069 : // stages a race between producer and a parked consumer; under a
3070 : // sanitizer's slowdown the producer never outruns the queue
3071 : // (same for the file store: fsync per create is slower than the drain)
3072 4 : if std::env::var_os("ANTARES_TEST_SANITIZER").is_some()
3073 4 : || std::env::var("ANTARES_TEST_STORE").is_ok_and(|s| s == "file")
3074 : {
3075 0 : return;
3076 4 : }
3077 4 : crate::allow_private();
3078 : // accepts, reads nothing, never answers — the serial consumer parks
3079 : // on the first delivery for the endpoint's whole timeout
3080 4 : let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
3081 4 : let addr = listener.local_addr().expect("addr");
3082 4 : std::thread::spawn(move || {
3083 4 : let mut held = Vec::new();
3084 4 : for s in listener.incoming().flatten() {
3085 4 : held.push(s);
3086 4 : }
3087 4 : });
3088 4 : let mut st = AppState::new("antares-queue-bound".into());
3089 4 : crate::wire(&mut st).await;
3090 : // the stall is bounded by the outbound client's own 5 s timeout, not
3091 : // by endpoint.timeout: after each timeout the consumer frees another
3092 : // CHANGE_BATCH slots, and on a slow runner a single-queue-depth loop
3093 : // fits inside that window and never overflows (seen in CI). Produce
3094 : // several queue depths and stop at the first counted drop — the
3095 : // producer only has to outpace the drain, not beat one window.
3096 4 : subscribe(&st, "staller", &format!("http://{addr}/notify"), 30_000).await;
3097 4 : let before = changes_dropped();
3098 4104 : for n in 0..(4 * CHANGE_QUEUE) {
3099 4104 : assert_eq!(create_vehicle(&st, 1_000 + n).await, 201);
3100 4104 : if changes_dropped() > before {
3101 4 : break;
3102 4100 : }
3103 4 : }
3104 4 : assert!(
3105 4 : changes_dropped() > before,
3106 4 : "a full matcher queue must drop and count, not grow (dropped {} → {})",
3107 4 : before,
3108 4 : changes_dropped()
3109 4 : );
3110 4 : }
3111 :
3112 : /// Every accepted change is counted until its pass has run, so a drain
3113 : /// that waits for zero closes the pool only after the last delivery.
3114 : #[tokio::test(flavor = "multi_thread")]
3115 4 : async fn pending_changes_return_to_zero_once_delivered() {
3116 4 : crate::allow_private();
3117 4 : let (uri, hits) = counting_endpoint().await;
3118 4 : let mut st = AppState::new("antares-pending-changes".into());
3119 4 : crate::wire(&mut st).await;
3120 4 : subscribe(&st, "counter", &uri, 30_000).await;
3121 80 : for n in 0..20 {
3122 80 : assert_eq!(create_vehicle(&st, 5_000 + n).await, 201);
3123 : }
3124 4 : let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
3125 : // one pass folds queued changes into one notification, so the hit
3126 : // count is at least one, not twenty
3127 12 : let pending = || st.pending_changes.load(Ordering::SeqCst);
3128 8 : while (hits.load(Ordering::SeqCst) == 0 || pending() > 0)
3129 4 : && std::time::Instant::now() < deadline
3130 : {
3131 4 : tokio::time::sleep(std::time::Duration::from_millis(20)).await;
3132 : }
3133 4 : assert!(hits.load(Ordering::SeqCst) > 0, "nothing delivered");
3134 4 : assert_eq!(
3135 4 : pending(),
3136 4 : 0,
3137 4 : "pending must fall back to zero after delivery"
3138 4 : );
3139 4 : }
3140 :
3141 : /// Table 5.2.12-1: "\"entityUpdated\" is equivalent to the combination
3142 : /// \"attributeCreated\", \"attributeUpdated\" and \"attributeDeleted\"",
3143 : /// so such a subscription notifies on a creation exactly like the
3144 : /// spelled-out list does — while "entityDeleted" alone does not.
3145 : #[tokio::test(flavor = "multi_thread")]
3146 4 : async fn entity_updated_trigger_notifies_on_entity_creation() {
3147 : // under a sanitizer 370 concurrent tests starve the endpoint past the
3148 : // outbound client's 5 s cap (seen twice in strict); triggers, not latency
3149 4 : if std::env::var_os("ANTARES_TEST_SANITIZER").is_some() {
3150 0 : return;
3151 4 : }
3152 4 : crate::allow_private();
3153 4 : let (uri, hits) = counting_endpoint().await;
3154 4 : let (quiet_uri, quiet_hits) = counting_endpoint().await;
3155 4 : let mut st = AppState::new("antares-trigger-equivalence".into());
3156 4 : crate::wire(&mut st).await;
3157 8 : let sub = |id: &str, trigger: &str, uri: &str| {
3158 8 : json!({
3159 8 : "id": format!("urn:ngsi-ld:Subscription:{id}"),
3160 8 : "type": "Subscription",
3161 8 : "entities": [{"type": "Vehicle"}],
3162 8 : "notificationTrigger": [trigger],
3163 : // 30 s: a sanitizer runner with 370 concurrent tests took
3164 : // 8 s to deliver once; this test is about triggers, not latency
3165 8 : "notification": {"endpoint": {"uri": uri, "timeout": 30_000}},
3166 : })
3167 8 : };
3168 8 : for body in [
3169 4 : sub("eu", "entityUpdated", &uri),
3170 4 : sub("ed", "entityDeleted", &quiet_uri),
3171 4 : ] {
3172 8 : assert_eq!(post(&st, "/ngsi-ld/v1/subscriptions", body).await, 201);
3173 : }
3174 4 : assert_eq!(create_vehicle(&st, 7).await, 201);
3175 : // a wall-clock bound, not an iteration count: a sanitizer build
3176 : // delivers the same notification an order of magnitude slower
3177 4 : let deadline = std::time::Instant::now() + std::time::Duration::from_secs(60);
3178 8 : while hits.load(Ordering::SeqCst) == 0 && std::time::Instant::now() < deadline {
3179 4 : tokio::time::sleep(std::time::Duration::from_millis(100)).await;
3180 : }
3181 : // the receiver's count alone cannot say WHY nothing arrived: the
3182 : // subscription's own bookkeeping and the pipeline counters can
3183 4 : let sub = st
3184 4 : .store
3185 4 : .get(
3186 4 : &TenantId::default(),
3187 4 : Kind::Subscription,
3188 4 : "urn:ngsi-ld:Subscription:eu",
3189 4 : )
3190 4 : .await
3191 4 : .ok()
3192 4 : .flatten()
3193 4 : .map(|s| s["notification"].to_string())
3194 4 : .unwrap_or_else(|| "subscription missing".into());
3195 4 : let diagnosis = || {
3196 0 : format!(
3197 : "notification={sub} changes_dropped={} task_panics={} dead_letters={}",
3198 0 : changes_dropped(),
3199 0 : task_panics(),
3200 0 : DEAD_LETTERS.load(Ordering::Relaxed)
3201 : )
3202 0 : };
3203 4 : assert_eq!(
3204 4 : hits.load(Ordering::SeqCst),
3205 : 1,
3206 : "entityUpdated implies attributeCreated, so a creation notifies; {}",
3207 0 : diagnosis()
3208 : );
3209 4 : assert_eq!(
3210 4 : quiet_hits.load(Ordering::SeqCst),
3211 4 : 0,
3212 4 : "entityDeleted carries no equivalence and must not fire on a creation"
3213 4 : );
3214 4 : }
3215 : }
3216 :
3217 : #[cfg(test)]
3218 : mod clause_5_2_12_triggers {
3219 : use super::*;
3220 : use serde_json::json;
3221 :
3222 : /// Table 5.2.12-1 notificationTrigger: "If not present, the default is
3223 : /// the combination \"attributeCreated\" and \"attributeUpdated\".
3224 : /// \"entityUpdated\" is equivalent to the combination
3225 : /// \"attributeCreated\", \"attributeUpdated\" and \"attributeDeleted\"."
3226 : #[test]
3227 4 : fn entity_updated_expands_to_its_equivalent_attribute_triggers() {
3228 20 : let has = |v: &[&str], t: &str| v.contains(&t);
3229 :
3230 4 : let empty = json!({});
3231 4 : let default = triggers_of(&empty);
3232 4 : assert_eq!(default, vec!["attributeCreated", "attributeUpdated"]);
3233 4 : assert!(
3234 4 : !has(&default, "attributeDeleted"),
3235 : "the default combination is two triggers, not three"
3236 : );
3237 :
3238 4 : let updated = json!({"notificationTrigger": ["entityUpdated"]});
3239 4 : let expanded = triggers_of(&updated);
3240 12 : for t in ["attributeCreated", "attributeUpdated", "attributeDeleted"] {
3241 12 : assert!(has(&expanded, t), "entityUpdated must imply {t}");
3242 : }
3243 4 : assert!(
3244 4 : has(&expanded, "entityUpdated"),
3245 : "the declared trigger itself survives the expansion"
3246 : );
3247 :
3248 : // Only entityUpdated carries the equivalence: the other two entity
3249 : // triggers must NOT gain attribute triggers.
3250 8 : for t in ["entityCreated", "entityDeleted"] {
3251 8 : let one = json!({ "notificationTrigger": [t] });
3252 8 : let only = triggers_of(&one);
3253 8 : assert_eq!(only, vec![t], "{t} is not an equivalence");
3254 : }
3255 :
3256 : // Idempotent: an explicit list that already spells the combination
3257 : // out gains nothing, and the equivalent forms agree.
3258 4 : let spelled = json!({"notificationTrigger":
3259 : ["entityUpdated", "attributeCreated", "attributeUpdated", "attributeDeleted"]});
3260 4 : let literal = triggers_of(&spelled);
3261 4 : assert_eq!(literal.len(), 4, "no duplicates are appended");
3262 4 : let mut a = expanded.clone();
3263 4 : let mut b = literal.clone();
3264 4 : a.sort();
3265 4 : b.sort();
3266 4 : assert_eq!(a, b, "[\"entityUpdated\"] == the spelled-out combination");
3267 4 : }
3268 :
3269 : /// Table 5.2.12-1 scopes watchedAttributes to "Watched Attributes
3270 : /// (Properties or Relationships)", so a write that only moves the
3271 : /// entity's own system members is no attribute-level change and must
3272 : /// raise no attributeCreated/attributeUpdated trigger.
3273 : #[test]
3274 4 : fn entity_system_members_are_not_attribute_changes() {
3275 4 : let before = json!({
3276 4 : "id": "urn:ngsi-ld:Vehicle:1",
3277 4 : "type": ["Vehicle"],
3278 4 : "expiresAt": "2030-01-01T00:00:00Z",
3279 : });
3280 4 : let after = json!({
3281 4 : "id": "urn:ngsi-ld:Vehicle:1",
3282 4 : "type": ["Vehicle"],
3283 4 : "expiresAt": "2031-01-01T00:00:00Z",
3284 4 : "deletedAt": "2031-01-01T00:00:00Z",
3285 4 : "modifiedAt": "2026-01-01T00:00:00Z",
3286 4 : "scope": "/a",
3287 : });
3288 4 : assert!(
3289 4 : diff(Some(&before), Some(&after)).is_empty(),
3290 : "entity-level system members are not Attributes"
3291 : );
3292 : // Positive control: a real Property change is still reported.
3293 4 : let mut with_attr = after.clone();
3294 4 : with_attr["speed"] = json!([{"type": "Property", "value": 1}]);
3295 4 : assert_eq!(
3296 4 : diff(Some(&before), Some(&with_attr)),
3297 4 : vec![("speed".to_owned(), ChangeClass::Created)]
3298 : );
3299 4 : }
3300 : }
3301 :
3302 : #[cfg(test)]
3303 : mod clause_4_5_7_instance_identity {
3304 : use super::*;
3305 : use serde_json::json;
3306 :
3307 76 : fn id_of(observed: Option<&str>, dataset: Option<&str>) -> String {
3308 76 : let mut inst = Map::new();
3309 76 : if let Some(o) = observed {
3310 68 : inst.insert("observedAt".into(), json!(o));
3311 68 : }
3312 76 : if let Some(d) = dataset {
3313 32 : inst.insert("datasetId".into(), json!(d));
3314 44 : }
3315 76 : instance_id("urn:ngsi-ld:Vehicle:1", "https://a/speed", &inst)
3316 76 : }
3317 :
3318 : /// 4.5.7: an instance is the Property "at a particular point in time,
3319 : /// which is recorded as a Temporal Property of the instance (typically
3320 : /// observedAt)" — so the instant, not the way a client spelled it,
3321 : /// decides which instance a record belongs to. 4.6.3 leaves the seconds
3322 : /// fraction optional and accepts a comma separator in requests, and the
3323 : /// broker stores the stamp as written, so one instant reaches this
3324 : /// function under several spellings.
3325 : ///
3326 : /// The consequence is the one the clause names: "Without such an
3327 : /// instanceId, it is not possible to selectively modify or delete
3328 : /// temporal information via the NGSI-LD API. The consequences of this
3329 : /// may be severe in the case of modification or deletion requests for
3330 : /// legal reasons". A correction re-sent with a different spelling landed
3331 : /// on a second row and left the value it was correcting in place.
3332 : #[test]
3333 4 : fn one_instant_is_one_instance_however_it_is_spelled() {
3334 4 : let spellings = [
3335 4 : "2020-01-01T00:00:00Z",
3336 4 : "2020-01-01T00:00:00.0Z",
3337 4 : "2020-01-01T00:00:00.000Z",
3338 4 : "2020-01-01T00:00:00.000000Z",
3339 4 : "2020-01-01T00:00:00,000Z",
3340 4 : ];
3341 4 : let first = id_of(Some(spellings[0]), None);
3342 16 : for s in &spellings[1..] {
3343 16 : assert_eq!(id_of(Some(s), None), first, "{s} is the same instant");
3344 : }
3345 : // and with a datasetId, which is part of the same identity
3346 4 : let first = id_of(Some(spellings[0]), Some("urn:ds:1"));
3347 16 : for s in &spellings[1..] {
3348 16 : assert_eq!(
3349 16 : id_of(Some(s), Some("urn:ds:1")),
3350 : first,
3351 : "{s} is the same instant"
3352 : );
3353 : }
3354 4 : }
3355 :
3356 : /// The identity still separates what the clause separates: a different
3357 : /// instant, a different dataset and a different fractional value are
3358 : /// different instances.
3359 : #[test]
3360 4 : fn different_instants_and_datasets_stay_different_instances() {
3361 4 : let base = id_of(Some("2020-01-01T00:00:00Z"), None);
3362 16 : for other in [
3363 4 : id_of(Some("2020-01-01T00:00:00.500Z"), None),
3364 4 : id_of(Some("2020-01-01T00:00:01Z"), None),
3365 4 : id_of(Some("2020-01-02T00:00:00Z"), None),
3366 4 : id_of(Some("2020-01-01T00:00:00Z"), Some("urn:ds:1")),
3367 4 : ] {
3368 16 : assert_ne!(other, base);
3369 : }
3370 4 : assert_ne!(
3371 4 : id_of(Some("2020-01-01T00:00:00.500Z"), Some("urn:ds:1")),
3372 4 : id_of(Some("2020-01-01T00:00:00.500Z"), Some("urn:ds:2"))
3373 : );
3374 : // "Without observedAt there is no instant to key on": a fresh id
3375 : // every time, so two unobserved records never collide.
3376 4 : assert_ne!(id_of(None, None), id_of(None, None));
3377 4 : }
3378 : }
3379 :
3380 : #[cfg(test)]
3381 : mod clause_5_8_6_deletion_payload {
3382 : use super::*;
3383 : use serde_json::json;
3384 :
3385 : const DC: &str = "https://uri.etsi.org/ngsi-ld/default-context";
3386 :
3387 56 : fn ctx() -> Arc<Context> {
3388 56 : antares_jsonld::Loader::new().core()
3389 56 : }
3390 :
3391 : /// Instances of one attribute in a notification data entry, whether the
3392 : /// representation collapsed a single instance to an object or kept an
3393 : /// array.
3394 32 : fn insts<'a>(entry: &'a Value, name: &str) -> Vec<&'a Value> {
3395 32 : match entry.get(name) {
3396 4 : Some(Value::Array(a)) => a.iter().collect(),
3397 28 : Some(v) => vec![v],
3398 0 : None => Vec::new(),
3399 : }
3400 32 : }
3401 :
3402 56 : async fn build(
3403 56 : sub: &Value,
3404 56 : before: &Value,
3405 56 : after: Option<&Value>,
3406 56 : deleted: &[String],
3407 56 : entity_deleted: bool,
3408 56 : ) -> Value {
3409 56 : let st = AppState::new("antares-tombstone-test".into());
3410 56 : let tenant = TenantId::new("default").expect("tenant");
3411 56 : let ctx = ctx();
3412 56 : let data = build_data(
3413 56 : &st,
3414 56 : &tenant,
3415 56 : sub,
3416 56 : &ctx,
3417 56 : Some(before),
3418 56 : after,
3419 56 : deleted,
3420 56 : entity_deleted,
3421 56 : "2026-01-01T00:00:00Z",
3422 56 : )
3423 56 : .await;
3424 56 : assert_eq!(data.len(), 1, "one changed entity ⇒ one data entry");
3425 56 : data.into_iter().next().expect("entry")
3426 56 : }
3427 :
3428 : /// 5.8.6: a deleted Attribute is notified as the NGSI-LD null value of
3429 : /// its own type — `object` for a Relationship, the `{"@none": …}`
3430 : /// languageMap form for a LanguageProperty, `json`, `vocab`, `value`.
3431 : #[tokio::test]
3432 4 : async fn typed_null_member_per_attribute_type() {
3433 4 : let cases = [
3434 4 : ("Property", "value", json!("urn:ngsi-ld:null")),
3435 4 : ("Relationship", "object", json!("urn:ngsi-ld:null")),
3436 4 : (
3437 4 : "LanguageProperty",
3438 4 : "languageMap",
3439 4 : json!({"@none": "urn:ngsi-ld:null"}),
3440 4 : ),
3441 4 : ("JsonProperty", "json", json!("urn:ngsi-ld:null")),
3442 4 : ("VocabProperty", "vocab", json!("urn:ngsi-ld:null")),
3443 4 : ];
3444 20 : for (atype, member, null_value) in cases {
3445 20 : let before = json!({
3446 20 : "id": "urn:ngsi-ld:Vehicle:tomb",
3447 20 : "type": [format!("{DC}/Vehicle")],
3448 20 : format!("{DC}/gone"): [{"type": atype, member: json!("previous")}],
3449 4 : });
3450 20 : let after = json!({
3451 20 : "id": "urn:ngsi-ld:Vehicle:tomb",
3452 20 : "type": [format!("{DC}/Vehicle")],
3453 4 : });
3454 20 : let entry = build(
3455 20 : &json!({"notification": {}}),
3456 20 : &before,
3457 20 : Some(&after),
3458 20 : &[format!("{DC}/gone")],
3459 20 : false,
3460 20 : )
3461 20 : .await;
3462 20 : let got = insts(&entry, "gone");
3463 20 : assert_eq!(got.len(), 1, "{atype}: one tombstone");
3464 20 : assert_eq!(got[0].get("type"), Some(&json!(atype)));
3465 20 : assert_eq!(
3466 20 : got[0].get(member),
3467 20 : Some(&null_value),
3468 4 : "{atype} tombstones through its own {member} member"
3469 4 : );
3470 4 : // showChanges false and sysAttrs false ⇒ neither stamp appears
3471 20 : assert!(
3472 20 : got[0].get("deletedAt").is_none(),
3473 4 : "{atype}: no deletedAt without sysAttrs"
3474 4 : );
3475 20 : assert!(
3476 20 : got[0]
3477 20 : .as_object()
3478 40 : .is_some_and(|o| !o.keys().any(|k| k.starts_with("previous"))),
3479 4 : "{atype}: no previous* member without showChanges"
3480 4 : );
3481 4 : }
3482 4 : }
3483 :
3484 : /// 5.8.6: a whole Attribute deleted at once is ONE tombstone with no
3485 : /// datasetId, while losing individual instances tombstones each lost
3486 : /// datasetId and leaves the survivors untouched.
3487 : #[tokio::test]
3488 4 : async fn whole_attribute_versus_per_instance_deletion() {
3489 4 : let three = json!([
3490 4 : {"type": "Property", "value": 1},
3491 4 : {"type": "Property", "value": 2, "datasetId": "urn:ds:a"},
3492 4 : {"type": "Property", "value": 3, "datasetId": "urn:ds:b"},
3493 : ]);
3494 4 : let before = json!({
3495 4 : "id": "urn:ngsi-ld:Vehicle:multi",
3496 4 : "type": [format!("{DC}/Vehicle")],
3497 4 : format!("{DC}/speed"): three,
3498 : });
3499 4 : let attr = vec![format!("{DC}/speed")];
3500 :
3501 : // whole attribute gone
3502 4 : let after = json!({
3503 4 : "id": "urn:ngsi-ld:Vehicle:multi",
3504 4 : "type": [format!("{DC}/Vehicle")],
3505 : });
3506 4 : let entry = build(
3507 4 : &json!({"notification": {}}),
3508 4 : &before,
3509 4 : Some(&after),
3510 4 : &attr,
3511 4 : false,
3512 4 : )
3513 4 : .await;
3514 4 : let got = insts(&entry, "speed");
3515 4 : assert_eq!(got.len(), 1, "a whole-attribute deletion is one tombstone");
3516 4 : assert!(
3517 4 : got[0].get("datasetId").is_none(),
3518 : "the single tombstone carries no datasetId"
3519 : );
3520 :
3521 : // one instance gone, two survive
3522 4 : let after = json!({
3523 4 : "id": "urn:ngsi-ld:Vehicle:multi",
3524 4 : "type": [format!("{DC}/Vehicle")],
3525 4 : format!("{DC}/speed"): [
3526 4 : {"type": "Property", "value": 1},
3527 4 : {"type": "Property", "value": 3, "datasetId": "urn:ds:b"},
3528 : ],
3529 : });
3530 4 : let entry = build(
3531 4 : &json!({"notification": {}}),
3532 4 : &before,
3533 4 : Some(&after),
3534 4 : &attr,
3535 4 : false,
3536 4 : )
3537 4 : .await;
3538 4 : let got = insts(&entry, "speed");
3539 4 : assert_eq!(got.len(), 3, "two survivors plus one tombstone");
3540 4 : let tombs: Vec<&&Value> = got
3541 4 : .iter()
3542 12 : .filter(|i| i.get("value") == Some(&json!("urn:ngsi-ld:null")))
3543 4 : .collect();
3544 4 : assert_eq!(tombs.len(), 1, "exactly the lost instance is tombstoned");
3545 4 : assert_eq!(tombs[0].get("datasetId"), Some(&json!("urn:ds:a")));
3546 8 : for surviving in got
3547 4 : .iter()
3548 12 : .filter(|i| i.get("value") != Some(&json!("urn:ngsi-ld:null")))
3549 4 : {
3550 8 : assert!(
3551 8 : surviving.get("deletedAt").is_none(),
3552 4 : "a surviving instance is not marked deleted"
3553 4 : );
3554 4 : }
3555 4 : }
3556 :
3557 : /// 5.8.6: "If an Attribute has been deleted, only the name of the
3558 : /// attribute as key and the URI `urn:ngsi-ld:null` as value shall be
3559 : /// provided, unless more information is required. The latter is the case,
3560 : /// if: a datasetId needs to be provided; the notification.sysAttrs is set
3561 : /// to true …; notification.showChanges is set to true …. In all such
3562 : /// cases, a JSON object with all the required information is provided,
3563 : /// where the value or the object is set to the URI `urn:ngsi-ld:null`
3564 : /// respectively or, in case of a LanguageProperty, the languageMap is set
3565 : /// to `{"@none": "urn:ngsi-ld:null"}`."
3566 : ///
3567 : /// The bare-key form is the one an interoperability campaign reads, and
3568 : /// it is reached only through the concise collapse — the tombstone itself
3569 : /// is always built as an object, so nothing below this level can tell the
3570 : /// two apart. 5.5.4 confines the bare form to concise: normalized keeps
3571 : /// the object, and a first-level `urn:ngsi-ld:null` is BadRequestData
3572 : /// everywhere else.
3573 : #[tokio::test]
3574 4 : async fn a_deleted_attribute_is_bare_unless_it_has_more_to_say() {
3575 4 : let speed = format!("{DC}/speed");
3576 4 : let label = format!("{DC}/label");
3577 4 : let before = json!({
3578 4 : "id": "urn:ngsi-ld:Vehicle:c",
3579 4 : "type": [format!("{DC}/Vehicle")],
3580 4 : speed.clone(): [{"type": "Property", "value": 1}],
3581 4 : label.clone(): [{"type": "LanguageProperty", "languageMap": {"en": "hi"}}],
3582 : });
3583 4 : let after = json!({
3584 4 : "id": "urn:ngsi-ld:Vehicle:c",
3585 4 : "type": [format!("{DC}/Vehicle")],
3586 : });
3587 4 : let deleted = [speed.clone(), label.clone()];
3588 : // a nested fn, not a closure: the future borrows the arguments and
3589 : // the `sub` it builds, and a closure cannot return a value borrowing
3590 : // its own captures
3591 20 : async fn entry(before: &Value, after: &Value, deleted: &[String], n: Value) -> Value {
3592 20 : build(
3593 20 : &json!({"notification": n}),
3594 20 : before,
3595 20 : Some(after),
3596 20 : deleted,
3597 20 : false,
3598 20 : )
3599 20 : .await
3600 20 : }
3601 :
3602 : // Nothing more is required: the attribute IS the URI.
3603 8 : for fmt in ["concise", "simplified"] {
3604 8 : let e = entry(&before, &after, &deleted, json!({"format": fmt})).await;
3605 8 : assert_eq!(
3606 8 : e["speed"],
3607 8 : json!("urn:ngsi-ld:null"),
3608 : "{fmt}: a plain deletion is the bare URI, not an object"
3609 : );
3610 8 : assert_eq!(
3611 8 : e["label"],
3612 8 : json!({"languageMap": {"@none": "urn:ngsi-ld:null"}}),
3613 : "{fmt}: a LanguageProperty deletion is the @none map"
3614 : );
3615 : }
3616 :
3617 : // Normalized is not the bare form (5.5.4 confines that to concise).
3618 4 : let e = entry(&before, &after, &deleted, json!({"format": "normalized"})).await;
3619 4 : assert_eq!(
3620 4 : e["speed"],
3621 4 : json!({"type": "Property", "value": "urn:ngsi-ld:null"}),
3622 : "normalized keeps the typed object"
3623 : );
3624 :
3625 : // sysAttrs: the system-generated sub-attributes have to be provided,
3626 : // so the deletion becomes an object carrying them.
3627 4 : let e = entry(
3628 4 : &before,
3629 4 : &after,
3630 4 : &deleted,
3631 4 : json!({"format": "concise", "sysAttrs": true}),
3632 4 : )
3633 4 : .await;
3634 4 : assert_eq!(e["speed"]["value"], json!("urn:ngsi-ld:null"));
3635 4 : assert_eq!(e["speed"]["deletedAt"], json!("2026-01-01T00:00:00Z"));
3636 :
3637 : // showChanges: a previous value has to be provided.
3638 4 : let e = entry(
3639 4 : &before,
3640 4 : &after,
3641 4 : &deleted,
3642 4 : json!({"format": "concise", "showChanges": true}),
3643 4 : )
3644 4 : .await;
3645 4 : assert_eq!(e["speed"]["value"], json!("urn:ngsi-ld:null"));
3646 4 : assert_eq!(e["speed"]["previousValue"], json!(1));
3647 4 : assert_eq!(
3648 4 : e["label"]["previousLanguageMap"],
3649 4 : json!({"en": "hi"}),
3650 : "a LanguageProperty reports previousLanguageMap"
3651 : );
3652 :
3653 : // A datasetId needs to be provided: one instance of two goes, so the
3654 : // tombstone must name which — and stays an object to do it.
3655 4 : let before_ds = json!({
3656 4 : "id": "urn:ngsi-ld:Vehicle:c",
3657 4 : "type": [format!("{DC}/Vehicle")],
3658 4 : speed.clone(): [{"type": "Property", "value": 1, "datasetId": "urn:ds:a"},
3659 4 : {"type": "Property", "value": 2, "datasetId": "urn:ds:b"}],
3660 : });
3661 4 : let after_ds = json!({
3662 4 : "id": "urn:ngsi-ld:Vehicle:c",
3663 4 : "type": [format!("{DC}/Vehicle")],
3664 4 : speed.clone(): [{"type": "Property", "value": 2, "datasetId": "urn:ds:b"}],
3665 : });
3666 4 : let e = build(
3667 4 : &json!({"notification": {"format": "concise"}}),
3668 4 : &before_ds,
3669 4 : Some(&after_ds),
3670 4 : &[speed],
3671 4 : false,
3672 4 : )
3673 4 : .await;
3674 4 : let gone = e["speed"]
3675 4 : .as_array()
3676 8 : .and_then(|a| a.iter().find(|i| i["value"] == json!("urn:ngsi-ld:null")))
3677 4 : .expect("the lost instance is tombstoned");
3678 4 : assert_eq!(gone["datasetId"], json!("urn:ds:a"));
3679 4 : assert!(
3680 4 : e["speed"]
3681 4 : .as_array()
3682 4 : .is_some_and(|a| a.iter().any(|i| i["value"] == json!(2))),
3683 4 : "the surviving instance is still reported: {e}"
3684 4 : );
3685 4 : }
3686 :
3687 : /// 5.8.6 with sysAttrs and showChanges: the tombstone carries deletedAt
3688 : /// and the previous value of its own typed member.
3689 : #[tokio::test]
3690 4 : async fn sys_attrs_and_show_changes_stamp_the_tombstone() {
3691 4 : let before = json!({
3692 4 : "id": "urn:ngsi-ld:Vehicle:stamp",
3693 4 : "type": [format!("{DC}/Vehicle")],
3694 4 : format!("{DC}/where"): [{"type": "Relationship", "object": "urn:ngsi-ld:P:1",
3695 4 : "createdAt": "2025-01-01T00:00:00Z"}],
3696 : });
3697 4 : let after = json!({
3698 4 : "id": "urn:ngsi-ld:Vehicle:stamp",
3699 4 : "type": [format!("{DC}/Vehicle")],
3700 : });
3701 4 : let sub = json!({"notification": {"sysAttrs": true, "showChanges": true}});
3702 4 : let entry = build(&sub, &before, Some(&after), &[format!("{DC}/where")], false).await;
3703 4 : let got = insts(&entry, "where");
3704 4 : assert_eq!(got.len(), 1);
3705 4 : assert_eq!(got[0].get("object"), Some(&json!("urn:ngsi-ld:null")));
3706 4 : assert_eq!(
3707 4 : got[0].get("deletedAt"),
3708 4 : Some(&json!("2026-01-01T00:00:00Z"))
3709 : );
3710 4 : assert_eq!(
3711 4 : got[0].get("previousObject"),
3712 4 : Some(&json!("urn:ngsi-ld:P:1")),
3713 : "showChanges reports the previous object, not previousValue"
3714 : );
3715 4 : assert!(
3716 4 : got[0].get("previousValue").is_none(),
3717 4 : "a Relationship never reports previousValue"
3718 4 : );
3719 4 : }
3720 : }
3721 :
3722 : #[cfg(test)]
3723 : mod candidate_index {
3724 : use super::*;
3725 : use serde_json::json;
3726 :
3727 : const DC: &str = "https://uri.etsi.org/ngsi-ld/default-context";
3728 :
3729 : /// The index may over-select, never under-select: every subscription
3730 : /// `selector_match` accepts for a change must come back from
3731 : /// `candidates()` for that change's types and changed attributes —
3732 : /// otherwise it silently stops firing (5.8.6).
3733 : #[test]
3734 4 : fn candidates_never_under_select_for_any_selector_shape() {
3735 4 : let ctx = antares_jsonld::Loader::new().core();
3736 4 : let doc = json!({
3737 4 : "id": "urn:ngsi-ld:Vehicle:1",
3738 4 : "type": [format!("{DC}/Vehicle")],
3739 4 : format!("{DC}/speed"): [{"type": "Property", "value": 1}],
3740 : });
3741 4 : let shapes: Vec<(&str, Value)> = vec![
3742 4 : (
3743 4 : "plain type",
3744 4 : json!({"entities": [{"type": format!("{DC}/Vehicle")}]}),
3745 4 : ),
3746 4 : (
3747 4 : "multiple types",
3748 4 : json!({"entities": [{"type": format!("{DC}/Vehicle")},
3749 4 : {"type": format!("{DC}/Building")}]}),
3750 4 : ),
3751 4 : (
3752 4 : "id only",
3753 4 : json!({"entities": [{"id": "urn:ngsi-ld:Vehicle:1"}]}),
3754 4 : ),
3755 4 : (
3756 4 : "idPattern only",
3757 4 : json!({"entities": [{"idPattern": "^urn:ngsi-ld:Vehicle:"}]}),
3758 4 : ),
3759 4 : (
3760 4 : "type selection expression",
3761 4 : json!({"entities": [{"type": format!("{DC}/Vehicle|{DC}/Building")}]}),
3762 4 : ),
3763 4 : (
3764 4 : "watchedAttributes only",
3765 4 : json!({"watchedAttributes": [format!("{DC}/speed")]}),
3766 4 : ),
3767 4 : (
3768 4 : "watchedAttributes with entities",
3769 4 : json!({"entities": [{"type": format!("{DC}/Vehicle")}],
3770 4 : "watchedAttributes": [format!("{DC}/speed")]}),
3771 4 : ),
3772 4 : (
3773 4 : "type with idPattern",
3774 4 : json!({"entities": [{"type": format!("{DC}/Vehicle"),
3775 4 : "idPattern": "^urn:ngsi-ld:Vehicle:"}]}),
3776 4 : ),
3777 : ];
3778 4 : let mirror = SubMirror::default();
3779 32 : for (i, (_, doc)) in shapes.iter().enumerate() {
3780 32 : let mut sub = doc.clone();
3781 32 : sub["id"] = json!(format!("urn:ngsi-ld:Subscription:{i}"));
3782 32 : mirror.apply(
3783 32 : "default",
3784 32 : &format!("urn:ngsi-ld:Subscription:{i}"),
3785 32 : Some(sub),
3786 32 : );
3787 32 : }
3788 4 : let types = [format!("{DC}/Vehicle")];
3789 4 : let changed = [format!("{DC}/speed")];
3790 4 : let got = mirror.candidates(
3791 4 : "default",
3792 4 : &types.iter().map(String::as_str).collect::<Vec<_>>(),
3793 4 : &changed.iter().map(String::as_str).collect::<Vec<_>>(),
3794 : );
3795 32 : for (i, (name, shape)) in shapes.iter().enumerate() {
3796 32 : if !selector_match(shape, &doc, &ctx) {
3797 0 : continue;
3798 32 : }
3799 32 : let id = format!("urn:ngsi-ld:Subscription:{i}");
3800 32 : assert!(
3801 32 : got.iter()
3802 144 : .any(|c| c.get("id").and_then(Value::as_str) == Some(id.as_str())),
3803 : "{name}: selector_match accepts it, so candidates() must return it"
3804 : );
3805 : }
3806 : // A change touching neither the type nor the watched attribute still
3807 : // yields the shapes the index cannot classify — over-selection is
3808 : // allowed — but never the exactly-classified plain-type subscription.
3809 4 : let other = mirror.candidates("default", &[&format!("{DC}/Device")], &[]);
3810 4 : assert!(
3811 4 : !other
3812 4 : .iter()
3813 12 : .any(|c| c.get("id").and_then(Value::as_str) == Some("urn:ngsi-ld:Subscription:0")),
3814 : "an exactly classified type subscription is not evaluated for other types"
3815 : );
3816 4 : }
3817 :
3818 : /// The two classification sites must stay in the superset relation:
3819 : /// every character `selector_match` reads as a 4.17 type-selection
3820 : /// expression has to send the subscription to the broad bucket, or the
3821 : /// index would under-select.
3822 : #[test]
3823 4 : fn type_selection_expressions_are_always_broad() {
3824 16 : for c in ['|', ',', ';', '('] {
3825 16 : let sub = json!({"entities": [{"type": format!("{DC}/A{c}{DC}/B")}]});
3826 16 : assert!(
3827 16 : matches!(index_keys(&sub), Keys::Broad),
3828 : "a type containing {c:?} is a selection expression for selector_match, \
3829 : so the index must not classify it as a plain type"
3830 : );
3831 : }
3832 : // Table 5.2.33-1's "*" is the other member of that relation: it is
3833 : // stored raw and `selector_match` reads it as every type, so an index
3834 : // that classified it as the literal type "*" would look up a key no
3835 : // change carries and the subscription would never be a candidate.
3836 4 : assert!(
3837 0 : matches!(
3838 4 : index_keys(&json!({"entities": [{"type": "*"}]})),
3839 : Keys::Broad
3840 : ),
3841 : "a \"*\" selector matches every type, so the index must go broad"
3842 : );
3843 4 : assert!(matches!(
3844 4 : index_keys(&json!({"entities": [{"type": format!("{DC}/Vehicle")}]})),
3845 : Keys::Types(_)
3846 : ));
3847 4 : }
3848 : }
3849 :
3850 : /// Table 5.2.14.2-1 bookkeeping around a delivery that fails or is never
3851 : /// attempted at all.
3852 : #[cfg(all(test, not(target_arch = "wasm32")))]
3853 : mod clause_5_2_14_2_bookkeeping {
3854 : use super::*;
3855 : use serde_json::json;
3856 : use std::sync::atomic::{AtomicUsize, Ordering};
3857 :
3858 : const SUB_ID: &str = "urn:ngsi-ld:Subscription:book";
3859 :
3860 : /// An endpoint answering `status`, counting the requests that reach it.
3861 8 : async fn endpoint(status: axum::http::StatusCode) -> (String, Arc<AtomicUsize>) {
3862 8 : let hits: Arc<AtomicUsize> = Arc::default();
3863 8 : let seen = hits.clone();
3864 8 : let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
3865 8 : .await
3866 8 : .expect("bind");
3867 8 : let addr = listener.local_addr().expect("addr");
3868 8 : let app = axum::Router::new().route(
3869 8 : "/notify",
3870 12 : axum::routing::post(move || {
3871 12 : let seen = seen.clone();
3872 12 : async move {
3873 12 : seen.fetch_add(1, Ordering::SeqCst);
3874 12 : status
3875 12 : }
3876 12 : }),
3877 : );
3878 8 : tokio::spawn(async move { axum::serve(listener, app).await.expect("serve") });
3879 8 : (format!("http://{addr}/notify"), hits)
3880 8 : }
3881 :
3882 16 : async fn stored_notification(st: &AppState, tenant: &TenantId) -> Value {
3883 16 : st.store
3884 16 : .get(tenant, Kind::Subscription, SUB_ID)
3885 16 : .await
3886 16 : .expect("store read")
3887 16 : .expect("subscription row")
3888 16 : .get("notification")
3889 16 : .cloned()
3890 16 : .expect("notification member")
3891 16 : }
3892 :
3893 16 : async fn send(st: &AppState, tenant: &TenantId, sub: &Value) {
3894 16 : let ctx = antares_jsonld::Loader::new().core();
3895 16 : deliver_as(
3896 16 : st,
3897 16 : tenant,
3898 16 : Kind::Subscription,
3899 16 : sub,
3900 16 : "Notification",
3901 16 : vec![json!({"id": "urn:ngsi-ld:Vehicle:1", "type": "Vehicle"})],
3902 16 : &ctx,
3903 16 : None,
3904 16 : )
3905 16 : .await;
3906 16 : }
3907 :
3908 8 : async fn subscribe(st: &AppState, tenant: &TenantId, uri: &str) -> Value {
3909 8 : let sub = json!({
3910 8 : "id": SUB_ID,
3911 8 : "type": "Subscription",
3912 8 : "entities": [{"type": "Vehicle"}],
3913 8 : "notification": {"endpoint": {"uri": uri}},
3914 : });
3915 8 : st.store
3916 8 : .create(tenant, Kind::Subscription, SUB_ID, sub.clone())
3917 8 : .await
3918 8 : .expect("subscription row");
3919 8 : sub
3920 8 : }
3921 :
3922 : /// Table 5.2.14.2-1 timesFailed: "Number of times an unsuccessful
3923 : /// response (or timeout) has been received when delivering the
3924 : /// notification" — an output-only member implementations shall generate.
3925 : #[tokio::test(flavor = "multi_thread")]
3926 4 : async fn failed_delivery_generates_and_increments_times_failed() {
3927 4 : crate::allow_private();
3928 4 : let (uri, hits) = endpoint(axum::http::StatusCode::INTERNAL_SERVER_ERROR).await;
3929 4 : let st = AppState::new("antares-times-failed".into());
3930 4 : let tenant = TenantId::new("default").expect("tenant");
3931 4 : let sub = subscribe(&st, &tenant, &uri).await;
3932 :
3933 4 : send(&st, &tenant, &sub).await;
3934 4 : let n = stored_notification(&st, &tenant).await;
3935 4 : assert_eq!(hits.load(Ordering::SeqCst), 1, "the endpoint was tried");
3936 4 : assert_eq!(n.get("timesFailed"), Some(&json!(1)));
3937 4 : assert_eq!(n.get("timesSent"), Some(&json!(1)), "the attempt was sent");
3938 4 : assert_eq!(n.get("status"), Some(&json!("failed")));
3939 4 : assert!(
3940 4 : n.get("lastSuccess").is_none(),
3941 : "a failure rolls the optimistic lastSuccess back"
3942 : );
3943 :
3944 4 : send(&st, &tenant, &sub).await;
3945 4 : let n = stored_notification(&st, &tenant).await;
3946 4 : assert_eq!(
3947 4 : n.get("timesFailed"),
3948 4 : Some(&json!(2)),
3949 4 : "timesFailed counts every unsuccessful response"
3950 4 : );
3951 4 : }
3952 :
3953 : /// Table 5.2.14.2-1 timesSent = "Number of times that the notification
3954 : /// has been sent" and lastNotification = "the instant when the last
3955 : /// notification has been sent": a change suppressed by the open circuit
3956 : /// never reaches the wire, so neither member may move.
3957 : #[tokio::test(flavor = "multi_thread")]
3958 4 : async fn breaker_suppressed_delivery_does_not_move_times_sent() {
3959 4 : crate::allow_private();
3960 4 : let (uri, hits) = endpoint(axum::http::StatusCode::OK).await;
3961 4 : let st = AppState::new("antares-breaker-bookkeeping".into());
3962 4 : let tenant = TenantId::new("default").expect("tenant");
3963 4 : let sub = subscribe(&st, &tenant, &uri).await;
3964 20 : for _ in 0..crate::egress::TRIP_AFTER {
3965 20 : st.egress.record_failure(tenant.as_str(), &uri);
3966 20 : }
3967 4 : assert!(
3968 4 : st.egress.is_open(tenant.as_str(), &uri),
3969 : "the destination is open-circuit"
3970 : );
3971 :
3972 4 : send(&st, &tenant, &sub).await;
3973 4 : let n = stored_notification(&st, &tenant).await;
3974 4 : assert_eq!(hits.load(Ordering::SeqCst), 0, "nothing left the process");
3975 4 : assert!(
3976 4 : n.get("timesSent").is_none(),
3977 : "a suppressed notification was never sent"
3978 : );
3979 4 : assert!(
3980 4 : n.get("lastNotification").is_none(),
3981 : "lastNotification is the instant a notification was sent"
3982 : );
3983 :
3984 : // Positive control: with the circuit closed the same call delivers,
3985 : // so the assertions above cannot pass vacuously.
3986 4 : st.egress.record_success(tenant.as_str(), &uri);
3987 4 : send(&st, &tenant, &sub).await;
3988 4 : let n = stored_notification(&st, &tenant).await;
3989 4 : assert_eq!(hits.load(Ordering::SeqCst), 1);
3990 4 : assert_eq!(n.get("timesSent"), Some(&json!(1)));
3991 4 : assert!(n.get("lastNotification").is_some());
3992 4 : }
3993 : }
3994 :
3995 : /// Delivery policy: retries are transport under one notification (5.8.6
3996 : /// books the notification once), a success by retry sets lastSuccess and
3997 : /// status ok, an exhausted policy leaves exactly one dead letter, and the
3998 : /// default policy is the single attempt the clause describes.
3999 : #[cfg(all(test, not(target_arch = "wasm32")))]
4000 : mod delivery_policy_tests {
4001 : use super::*;
4002 : use antares_notifier::DeliveryPolicy;
4003 : use std::sync::atomic::{AtomicUsize, Ordering};
4004 : use std::time::Duration;
4005 :
4006 : const SUB_ID: &str = "urn:ngsi-ld:Subscription:policy";
4007 :
4008 20 : fn policy(attempts: u32, backoff_ms: u64) -> DeliveryPolicy {
4009 20 : DeliveryPolicy {
4010 20 : attempts,
4011 20 : backoff: Duration::from_millis(backoff_ms),
4012 20 : jitter: 0.0,
4013 20 : max_age: Duration::from_secs(60),
4014 20 : }
4015 20 : }
4016 :
4017 : /// Answers 500 to the first `fail_first` requests, 200 afterwards.
4018 28 : async fn flaky_endpoint(fail_first: usize) -> (String, Arc<AtomicUsize>) {
4019 28 : let hits: Arc<AtomicUsize> = Arc::default();
4020 28 : let seen = hits.clone();
4021 28 : let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
4022 28 : .await
4023 28 : .expect("bind");
4024 28 : let addr = listener.local_addr().expect("addr");
4025 28 : let app = axum::Router::new().route(
4026 28 : "/notify",
4027 36 : axum::routing::post(move || {
4028 36 : let seen = seen.clone();
4029 36 : async move {
4030 36 : let n = seen.fetch_add(1, Ordering::SeqCst);
4031 36 : if n < fail_first {
4032 28 : axum::http::StatusCode::INTERNAL_SERVER_ERROR
4033 : } else {
4034 8 : axum::http::StatusCode::OK
4035 : }
4036 36 : }
4037 36 : }),
4038 : );
4039 28 : tokio::spawn(async move { axum::serve(listener, app).await.expect("serve") });
4040 28 : (format!("http://{addr}/notify"), hits)
4041 28 : }
4042 :
4043 24 : fn state(p: DeliveryPolicy) -> (AppState, TenantId) {
4044 24 : crate::allow_private();
4045 24 : let mut st = AppState::new("antares-policy".into());
4046 24 : st.delivery = p;
4047 24 : (st, TenantId::new("default").expect("tenant"))
4048 24 : }
4049 :
4050 28 : async fn subscribe(st: &AppState, tenant: &TenantId, id: &str, uri: &str) -> Value {
4051 28 : let sub = json!({
4052 28 : "id": id,
4053 28 : "type": "Subscription",
4054 28 : "entities": [{"type": "Vehicle"}],
4055 28 : "notification": {"endpoint": {"uri": uri}},
4056 : });
4057 28 : st.store
4058 28 : .create(tenant, Kind::Subscription, id, sub.clone())
4059 28 : .await
4060 28 : .expect("subscription row");
4061 28 : sub
4062 28 : }
4063 :
4064 28 : async fn send(st: &AppState, tenant: &TenantId, sub: &Value) {
4065 28 : let ctx = antares_jsonld::Loader::new().core();
4066 28 : deliver_as(
4067 28 : st,
4068 28 : tenant,
4069 28 : Kind::Subscription,
4070 28 : sub,
4071 28 : "Notification",
4072 28 : vec![json!({"id": "urn:ngsi-ld:Vehicle:1", "type": "Vehicle"})],
4073 28 : &ctx,
4074 28 : None,
4075 28 : )
4076 28 : .await;
4077 28 : }
4078 :
4079 39 : async fn notification(st: &AppState, tenant: &TenantId, id: &str) -> Value {
4080 39 : st.store
4081 39 : .get(tenant, Kind::Subscription, id)
4082 39 : .await
4083 39 : .expect("store read")
4084 39 : .expect("subscription row")["notification"]
4085 39 : .clone()
4086 39 : }
4087 :
4088 31 : async fn letters(st: &AppState, tenant: &TenantId) -> Vec<Value> {
4089 31 : st.store.list(tenant, Kind::DeadLetter).await.expect("list")
4090 31 : }
4091 :
4092 8 : async fn wait_until<F, Fut>(mut cond: F, what: &str)
4093 8 : where
4094 8 : F: FnMut() -> Fut,
4095 8 : Fut: std::future::Future<Output = bool>,
4096 8 : {
4097 8 : for _ in 0..100 {
4098 30 : if cond().await {
4099 8 : return;
4100 22 : }
4101 22 : tokio::time::sleep(Duration::from_millis(50)).await;
4102 : }
4103 0 : panic!("timed out waiting for {what}");
4104 8 : }
4105 :
4106 : #[tokio::test(flavor = "multi_thread")]
4107 4 : async fn a_retry_that_succeeds_is_one_notification() {
4108 4 : let (st, t) = state(policy(3, 50));
4109 4 : let (uri, hits) = flaky_endpoint(2).await;
4110 4 : let sub = subscribe(&st, &t, SUB_ID, &uri).await;
4111 4 : send(&st, &t, &sub).await;
4112 : // the first attempt is booked at once, as 5.8.6 says
4113 4 : let n = notification(&st, &t, SUB_ID).await;
4114 4 : assert_eq!(hits.load(Ordering::SeqCst), 1);
4115 4 : assert_eq!(n["timesSent"], json!(1));
4116 4 : assert_eq!(n["status"], json!("failed"));
4117 4 : assert!(n.get("lastSuccess").is_none());
4118 4 : wait_until(
4119 38 : || async { notification(&st, &t, SUB_ID).await["status"] == json!("ok") },
4120 4 : "retry success",
4121 : )
4122 4 : .await;
4123 4 : let n = notification(&st, &t, SUB_ID).await;
4124 4 : assert_eq!(hits.load(Ordering::SeqCst), 3, "two retries were made");
4125 4 : assert_eq!(
4126 4 : n["timesSent"],
4127 4 : json!(1),
4128 : "retries never count as a second notification"
4129 : );
4130 4 : assert_eq!(
4131 4 : n["timesFailed"],
4132 4 : json!(1),
4133 : "the failed first attempt was booked once"
4134 : );
4135 4 : assert!(n.get("lastSuccess").is_some());
4136 4 : assert!(
4137 4 : n.get("lastFailure").is_some(),
4138 : "the earlier failure stays recorded"
4139 : );
4140 4 : assert!(
4141 4 : letters(&st, &t).await.is_empty(),
4142 4 : "a delivered notification is no dead letter"
4143 4 : );
4144 4 : }
4145 :
4146 : #[tokio::test(flavor = "multi_thread")]
4147 4 : async fn an_exhausted_policy_leaves_exactly_one_dead_letter() {
4148 4 : let (st, t) = state(policy(2, 50));
4149 4 : let (uri, hits) = flaky_endpoint(usize::MAX).await;
4150 4 : let sub = subscribe(&st, &t, SUB_ID, &uri).await;
4151 4 : let before = dead_letters_written();
4152 4 : send(&st, &t, &sub).await;
4153 4 : wait_until(
4154 22 : || async { !letters(&st, &t).await.is_empty() },
4155 4 : "dead letter",
4156 : )
4157 4 : .await;
4158 4 : tokio::time::sleep(Duration::from_millis(200)).await;
4159 4 : let l = letters(&st, &t).await;
4160 4 : assert_eq!(l.len(), 1, "{l:?}");
4161 4 : assert_eq!(hits.load(Ordering::SeqCst), 2, "attempts = policy.attempts");
4162 4 : let l = &l[0];
4163 4 : assert_eq!(l["subscriptionId"], json!(SUB_ID));
4164 4 : assert_eq!(l["attempts"], json!(2));
4165 4 : assert_eq!(l["binding"], json!("http"));
4166 4 : assert_eq!(l["uri"], json!(uri));
4167 4 : assert_eq!(l["lastError"], json!("HTTP 500"));
4168 4 : assert_eq!(l["payload"]["type"], json!("Notification"));
4169 4 : assert_eq!(l["payload"]["subscriptionId"], json!(SUB_ID));
4170 4 : assert!(l["id"]
4171 4 : .as_str()
4172 4 : .is_some_and(|i| i.starts_with("urn:ngsi-ld:DeadLetter:")));
4173 : // the letter carries the endpoint members the binding renders from,
4174 : // so a replay produces the identical request
4175 4 : assert_eq!(l["accept"], json!("application/json"));
4176 4 : assert!(l["link"].as_str().is_some_and(|v| v.contains("rel=")));
4177 4 : assert!(l["receiverInfo"].is_array());
4178 4 : assert!(dead_letters_written() > before);
4179 4 : let n = notification(&st, &t, SUB_ID).await;
4180 4 : assert_eq!(n["timesSent"], json!(1));
4181 4 : assert_eq!(n["timesFailed"], json!(1));
4182 4 : assert_eq!(n["status"], json!("failed"));
4183 4 : }
4184 :
4185 : #[tokio::test(flavor = "multi_thread")]
4186 4 : async fn the_default_policy_never_retries_and_never_dead_letters() {
4187 4 : let (st, t) = state(DeliveryPolicy::default());
4188 4 : let (uri, hits) = flaky_endpoint(usize::MAX).await;
4189 4 : let sub = subscribe(&st, &t, SUB_ID, &uri).await;
4190 4 : send(&st, &t, &sub).await;
4191 4 : tokio::time::sleep(Duration::from_millis(400)).await;
4192 4 : assert_eq!(hits.load(Ordering::SeqCst), 1);
4193 4 : assert!(letters(&st, &t).await.is_empty());
4194 4 : assert_eq!(
4195 4 : notification(&st, &t, SUB_ID).await["status"],
4196 4 : json!("failed")
4197 4 : );
4198 4 : }
4199 :
4200 : #[tokio::test(flavor = "multi_thread")]
4201 4 : async fn a_backoff_on_one_subscription_does_not_delay_another() {
4202 4 : let (st, t) = state(policy(2, 3_000));
4203 4 : let (dead, _) = flaky_endpoint(usize::MAX).await;
4204 4 : let (live, live_hits) = flaky_endpoint(0).await;
4205 4 : let a = subscribe(&st, &t, "urn:ngsi-ld:Subscription:a", &dead).await;
4206 4 : let b = subscribe(&st, &t, "urn:ngsi-ld:Subscription:b", &live).await;
4207 4 : let started = std::time::Instant::now();
4208 4 : send(&st, &t, &a).await;
4209 4 : send(&st, &t, &b).await;
4210 4 : assert_eq!(live_hits.load(Ordering::SeqCst), 1, "B delivered");
4211 4 : assert!(
4212 4 : started.elapsed() < Duration::from_secs(1),
4213 4 : "A's 3 s backoff must not sit on the delivery path: {:?}",
4214 4 : started.elapsed()
4215 4 : );
4216 4 : }
4217 :
4218 : #[tokio::test(flavor = "multi_thread")]
4219 4 : async fn retries_stop_when_the_subscription_is_deleted() {
4220 4 : let (st, t) = state(policy(4, 100));
4221 4 : let (uri, hits) = flaky_endpoint(usize::MAX).await;
4222 4 : let sub = subscribe(&st, &t, SUB_ID, &uri).await;
4223 4 : send(&st, &t, &sub).await;
4224 4 : st.store
4225 4 : .delete(&t, Kind::Subscription, SUB_ID)
4226 4 : .await
4227 4 : .expect("delete");
4228 4 : tokio::time::sleep(Duration::from_millis(800)).await;
4229 4 : assert_eq!(
4230 4 : hits.load(Ordering::SeqCst),
4231 : 1,
4232 : "no retry for a gone subscription"
4233 : );
4234 4 : assert!(
4235 4 : letters(&st, &t).await.is_empty(),
4236 4 : "no dead letter for a gone subscription"
4237 4 : );
4238 4 : }
4239 :
4240 : #[tokio::test(flavor = "multi_thread")]
4241 4 : async fn an_egress_refusal_is_never_retried() {
4242 4 : let (mut st, t) = state(policy(3, 50));
4243 : // the deny policy is built directly: the environment is shared by
4244 : // every test thread in this process, and a state constructed while
4245 : // the variable read "false" would refuse its loopback endpoint for
4246 : // the rest of its life
4247 4 : st.egress = Arc::new(crate::egress::Egress::new(antares_jsonld::EgressPolicy {
4248 4 : allow_private: false,
4249 4 : }));
4250 4 : let (uri, hits) = flaky_endpoint(0).await;
4251 4 : let sub = subscribe(&st, &t, SUB_ID, &uri).await;
4252 4 : send(&st, &t, &sub).await;
4253 4 : tokio::time::sleep(Duration::from_millis(400)).await;
4254 4 : assert_eq!(
4255 4 : hits.load(Ordering::SeqCst),
4256 : 0,
4257 : "policy refusal: nothing leaves"
4258 : );
4259 4 : assert!(
4260 4 : letters(&st, &t).await.is_empty(),
4261 : "a policy verdict is not a transport failure"
4262 : );
4263 4 : assert_eq!(
4264 4 : notification(&st, &t, SUB_ID).await["status"],
4265 4 : json!("failed")
4266 4 : );
4267 4 : }
4268 : }
4269 :
4270 : /// 5.8.6 periodic notifications: what a due `timeInterval` subscription
4271 : /// reads, what it sends, and which ticks it costs nothing at all.
4272 : #[cfg(all(test, not(target_arch = "wasm32")))]
4273 : mod clause_5_8_6_periodic_sweep {
4274 : use super::*;
4275 : use serde_json::json;
4276 :
4277 : const DC: &str = "https://uri.etsi.org/ngsi-ld/default-context";
4278 :
4279 32 : fn entity(id: &str, ty: &str) -> Value {
4280 32 : json!({"id": id, "type": [format!("{DC}/{ty}")]})
4281 32 : }
4282 :
4283 : /// 5.8.6 narrows the periodic read to "all the subscribed Entities" — and
4284 : /// a narrowing may only ever over-select: every Entity the arbiter
4285 : /// (`selector_match`) accepts has to survive both predicates the sweep
4286 : /// hands the store, or that subscription silently stops reporting it.
4287 : #[test]
4288 4 : fn periodic_read_narrowing_never_under_selects() {
4289 4 : let ctx = antares_jsonld::Loader::new().core();
4290 4 : let docs = [
4291 4 : entity("urn:ngsi-ld:Vehicle:1", "Vehicle"),
4292 4 : entity("urn:ngsi-ld:Vehicle:2", "Vehicle"),
4293 4 : entity("urn:ngsi-ld:Building:1", "Building"),
4294 4 : ];
4295 4 : let subs: Vec<(&str, Value)> = vec![
4296 4 : (
4297 4 : "plain type",
4298 4 : json!({"entities": [{"type": format!("{DC}/Vehicle")}]}),
4299 4 : ),
4300 4 : (
4301 4 : "one id with its type",
4302 4 : json!({"entities": [{"id": "urn:ngsi-ld:Vehicle:2",
4303 4 : "type": format!("{DC}/Vehicle")}]}),
4304 4 : ),
4305 4 : (
4306 4 : "id array",
4307 4 : json!({"entities": [{"id": ["urn:ngsi-ld:Vehicle:1",
4308 4 : "urn:ngsi-ld:Building:1"]}]}),
4309 4 : ),
4310 4 : (
4311 4 : "one id plus a bare-type entry",
4312 4 : json!({"entities": [{"id": "urn:ngsi-ld:Vehicle:1"},
4313 4 : {"type": format!("{DC}/Building")}]}),
4314 4 : ),
4315 4 : (
4316 4 : "idPattern only",
4317 4 : json!({"entities": [{"idPattern": "^urn:ngsi-ld:Vehicle:"}]}),
4318 4 : ),
4319 4 : (
4320 4 : "id overriding a contradicting idPattern",
4321 4 : json!({"entities": [{"id": "urn:ngsi-ld:Vehicle:1",
4322 4 : "idPattern": "^urn:ngsi-ld:Building:"}]}),
4323 4 : ),
4324 4 : (
4325 4 : "type selection expression",
4326 4 : json!({"entities": [{"type": format!("{DC}/Vehicle|{DC}/Building")}]}),
4327 4 : ),
4328 4 : (
4329 4 : "no entities selector at all",
4330 4 : json!({"watchedAttributes": [format!("{DC}/speed")]}),
4331 4 : ),
4332 : ];
4333 32 : for (name, sub) in &subs {
4334 32 : let ids = selector_ids(sub);
4335 32 : let type_groups: Vec<Vec<String>> = match index_keys(sub) {
4336 8 : Keys::Types(ts) => ts.into_iter().map(|t| vec![t]).collect(),
4337 24 : _ => Vec::new(),
4338 : };
4339 96 : for doc in &docs {
4340 96 : if !selector_match(sub, doc, &ctx) {
4341 32 : continue;
4342 64 : }
4343 64 : let id = doc["id"].as_str().expect("id");
4344 64 : if let Some(ids) = &ids {
4345 16 : assert!(
4346 20 : ids.iter().any(|i| i == id),
4347 : "{name}: selector_match accepts {id}, so the id narrowing must keep it"
4348 : );
4349 48 : }
4350 64 : if !type_groups.is_empty() {
4351 12 : let types: Vec<&str> = doc["type"]
4352 12 : .as_array()
4353 12 : .expect("type array")
4354 12 : .iter()
4355 12 : .filter_map(Value::as_str)
4356 12 : .collect();
4357 12 : assert!(
4358 12 : type_groups
4359 12 : .iter()
4360 12 : .any(|g| g.iter().all(|t| types.contains(&t.as_str()))),
4361 : "{name}: selector_match accepts {id}, so the type narrowing must keep it"
4362 : );
4363 52 : }
4364 : }
4365 : }
4366 4 : }
4367 :
4368 : /// Table 5.2.33-1: `id` is a String or a String[] and takes precedence
4369 : /// over `idPattern`, so it pins the read exactly — while any entry
4370 : /// leaving the id open (a bare type, an idPattern, no selector at all)
4371 : /// must yield NO id predicate, since the OR-ed selector then admits
4372 : /// Entities no listed id names.
4373 : #[test]
4374 4 : fn periodic_read_narrows_by_id_only_when_every_selector_entry_pins_one() {
4375 4 : assert_eq!(
4376 4 : selector_ids(&json!({"entities": [{"id": "urn:x:A", "type": "T"}]})),
4377 4 : Some(vec!["urn:x:A".to_owned()])
4378 : );
4379 4 : assert_eq!(
4380 4 : selector_ids(&json!({"entities": [{"id": ["urn:x:A", "urn:x:B"]},
4381 4 : {"id": "urn:x:C"}]})),
4382 4 : Some(vec![
4383 4 : "urn:x:A".to_owned(),
4384 4 : "urn:x:B".to_owned(),
4385 4 : "urn:x:C".to_owned()
4386 4 : ])
4387 : );
4388 28 : for open in [
4389 4 : json!({"entities": [{"id": "urn:x:A"}, {"type": "T"}]}),
4390 4 : json!({"entities": [{"idPattern": "^urn:x:"}]}),
4391 4 : json!({"entities": [{"type": "T"}]}),
4392 4 : json!({"entities": []}),
4393 4 : json!({"watchedAttributes": ["a"]}),
4394 4 : json!({}),
4395 4 : json!({"entities": [{"id": {"not": "a string"}}]}),
4396 4 : ] {
4397 28 : assert_eq!(
4398 28 : selector_ids(&open),
4399 : None,
4400 : "{open} leaves the id open — narrowing by id would drop matching Entities"
4401 : );
4402 : }
4403 4 : }
4404 :
4405 : /// An endpoint that keeps every notification body it receives.
4406 8 : async fn recording_endpoint() -> (String, Arc<std::sync::Mutex<Vec<Value>>>) {
4407 8 : let seen: Arc<std::sync::Mutex<Vec<Value>>> = Arc::default();
4408 8 : let sink = seen.clone();
4409 8 : let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
4410 8 : .await
4411 8 : .expect("bind");
4412 8 : let addr = listener.local_addr().expect("addr");
4413 8 : let app = axum::Router::new().route(
4414 8 : "/notify",
4415 8 : axum::routing::post(move |body: String| {
4416 8 : let sink = sink.clone();
4417 8 : async move {
4418 8 : if let Ok(v) = serde_json::from_str::<Value>(&body) {
4419 8 : sink.lock().expect("recorded bodies").push(v);
4420 8 : }
4421 8 : axum::http::StatusCode::OK
4422 8 : }
4423 8 : }),
4424 : );
4425 8 : tokio::spawn(async move { axum::serve(listener, app).await.expect("serve") });
4426 8 : (format!("http://{addr}/notify"), seen)
4427 8 : }
4428 :
4429 : /// A periodic subscription created `age_s` seconds ago with a one-second
4430 : /// interval — due on the next sweep.
4431 12 : fn periodic(id: &str, uri: &str, entities: Value, age_s: i64) -> Value {
4432 12 : let created = chrono::Utc::now() - chrono::Duration::seconds(age_s);
4433 12 : json!({
4434 12 : "id": id,
4435 12 : "type": "Subscription",
4436 12 : "timeInterval": 1,
4437 12 : "createdAt": created.to_rfc3339(),
4438 12 : "entities": entities,
4439 12 : "notification": {"endpoint": {"uri": uri, "accept": "application/json"}},
4440 : })
4441 12 : }
4442 :
4443 12 : async fn install(st: &AppState, tenant: &TenantId, sub: &Value) {
4444 12 : let id = sub["id"].as_str().expect("sub id");
4445 12 : st.store
4446 12 : .create(tenant, Kind::Subscription, id, sub.clone())
4447 12 : .await
4448 12 : .expect("subscription row");
4449 12 : if let Some(m) = &st.sub_mirror {
4450 12 : m.apply(tenant.as_str(), id, Some(sub.clone()));
4451 12 : }
4452 12 : }
4453 :
4454 16 : fn state_with_mirror(alias: &str) -> (AppState, Arc<SubMirror>, TenantId) {
4455 16 : let mut st = AppState::new(alias.into());
4456 16 : let mirror = Arc::new(SubMirror::default());
4457 16 : st.sub_mirror = Some(mirror.clone());
4458 16 : let tenant = TenantId::new("default").expect("tenant");
4459 16 : (st, mirror, tenant)
4460 16 : }
4461 :
4462 : /// 5.8.6: the periodic Notification carries "all the subscribed Entities
4463 : /// that match the query, geoquery and Scope query conditions" — the one
4464 : /// Entity this selector names out of a populated tenant, and nothing
4465 : /// else. And for the subscription whose selector matches nothing: "If
4466 : /// there are no matching Entities, no Notification is sent."
4467 : #[tokio::test(flavor = "multi_thread")]
4468 4 : async fn periodic_sweep_notifies_only_the_subscribed_entities() {
4469 4 : crate::allow_private();
4470 4 : let (uri, seen) = recording_endpoint().await;
4471 4 : let (st, _mirror, tenant) = state_with_mirror("antares-periodic-narrowing");
4472 16 : for (id, ty) in [
4473 4 : ("urn:ngsi-ld:Vehicle:1", "Vehicle"),
4474 4 : ("urn:ngsi-ld:Vehicle:2", "Vehicle"),
4475 4 : ("urn:ngsi-ld:Vehicle:3", "Vehicle"),
4476 4 : ("urn:ngsi-ld:Building:1", "Building"),
4477 4 : ] {
4478 16 : st.store
4479 16 : .create(&tenant, Kind::Entity, id, entity(id, ty))
4480 16 : .await
4481 16 : .expect("entity row");
4482 : }
4483 4 : install(
4484 4 : &st,
4485 4 : &tenant,
4486 4 : &periodic(
4487 4 : "urn:ngsi-ld:Subscription:one",
4488 4 : &uri,
4489 4 : json!([{"id": "urn:ngsi-ld:Vehicle:2", "type": format!("{DC}/Vehicle")}]),
4490 4 : 10,
4491 4 : ),
4492 4 : )
4493 4 : .await;
4494 4 : install(
4495 4 : &st,
4496 4 : &tenant,
4497 4 : &periodic(
4498 4 : "urn:ngsi-ld:Subscription:none",
4499 4 : &uri,
4500 4 : json!([{"id": "urn:ngsi-ld:Vehicle:404", "type": format!("{DC}/Vehicle")}]),
4501 4 : 10,
4502 4 : ),
4503 4 : )
4504 4 : .await;
4505 :
4506 4 : interval_tick(&st).await;
4507 :
4508 4 : let bodies = seen.lock().expect("recorded bodies").clone();
4509 4 : assert_eq!(bodies.len(), 1, "exactly one subscription had a match");
4510 4 : let body = &bodies[0];
4511 4 : assert_eq!(
4512 4 : body["subscriptionId"], "urn:ngsi-ld:Subscription:one",
4513 : "a subscription matching no Entity sends no Notification"
4514 : );
4515 4 : let data = body["data"].as_array().expect("data array");
4516 4 : assert_eq!(data.len(), 1, "only the subscribed Entity is included");
4517 4 : assert_eq!(data[0]["id"], "urn:ngsi-ld:Vehicle:2");
4518 12 : for other in [
4519 4 : "urn:ngsi-ld:Vehicle:1",
4520 4 : "urn:ngsi-ld:Vehicle:3",
4521 4 : "urn:ngsi-ld:Building:1",
4522 4 : ] {
4523 12 : assert!(
4524 12 : !body.to_string().contains(other),
4525 4 : "{other} is not subscribed and must not appear in the notification"
4526 4 : );
4527 4 : }
4528 4 : }
4529 :
4530 : /// The sweep clock: 5.8.6 sends the periodic Notification "when the time
4531 : /// interval (in seconds) specified in such value field is reached", so a
4532 : /// tick before the earliest such instant cannot fire and must not sweep —
4533 : /// while writing a periodic subscription clears the clock, because a
4534 : /// subscription the previous sweep never saw may be due sooner.
4535 : #[tokio::test(flavor = "multi_thread")]
4536 4 : async fn armed_sweep_clock_skips_the_tick_until_a_subscription_write_clears_it() {
4537 : use std::sync::atomic::Ordering::Relaxed;
4538 4 : crate::allow_private();
4539 4 : let (uri, seen) = recording_endpoint().await;
4540 4 : let (st, mirror, tenant) = state_with_mirror("antares-periodic-clock");
4541 4 : st.store
4542 4 : .create(
4543 4 : &tenant,
4544 4 : Kind::Entity,
4545 4 : "urn:ngsi-ld:Vehicle:1",
4546 4 : entity("urn:ngsi-ld:Vehicle:1", "Vehicle"),
4547 4 : )
4548 4 : .await
4549 4 : .expect("entity row");
4550 4 : let sub = periodic(
4551 4 : "urn:ngsi-ld:Subscription:clock",
4552 4 : &uri,
4553 4 : json!([{"type": format!("{DC}/Vehicle")}]),
4554 : 10,
4555 : );
4556 4 : install(&st, &tenant, &sub).await;
4557 4 : let armed = chrono::Utc::now().timestamp_millis() + 60_000;
4558 4 : mirror.next_sub_sweep_ms.store(armed, Relaxed);
4559 :
4560 4 : interval_tick(&st).await;
4561 4 : assert!(
4562 4 : seen.lock().expect("recorded bodies").is_empty(),
4563 : "a tick before the clock must not fire, due subscription or not"
4564 : );
4565 4 : assert_eq!(
4566 4 : mirror.next_sub_sweep_ms.load(Relaxed),
4567 : armed,
4568 : "a skipped tick leaves the clock alone"
4569 : );
4570 :
4571 : // what a subscription write does
4572 4 : mirror.apply(
4573 4 : tenant.as_str(),
4574 4 : sub["id"].as_str().expect("sub id"),
4575 4 : Some(sub.clone()),
4576 4 : );
4577 4 : assert_eq!(
4578 4 : mirror.next_sub_sweep_ms.load(Relaxed),
4579 : 0,
4580 : "writing a periodic subscription clears the sweep clock"
4581 : );
4582 :
4583 4 : interval_tick(&st).await;
4584 4 : assert_eq!(
4585 4 : seen.lock().expect("recorded bodies").len(),
4586 : 1,
4587 : "with the clock cleared the due subscription fires"
4588 : );
4589 4 : assert!(
4590 4 : mirror.next_sub_sweep_ms.load(Relaxed) > chrono::Utc::now().timestamp_millis(),
4591 4 : "after firing, the clock points at the next due instant"
4592 4 : );
4593 4 : }
4594 :
4595 : /// 5.11.7: a Context Source Registration Subscription with `timeInterval`
4596 : /// fires periodically. Its writes reach the sweep as a signal rather than
4597 : /// as a mirrored document, so a write must clear the clock the way a
4598 : /// Subscription write does — otherwise a new one waits out a clock
4599 : /// computed before it existed.
4600 : #[tokio::test(flavor = "multi_thread")]
4601 4 : async fn a_written_csource_subscription_clears_the_csub_sweep_clock() {
4602 : use std::sync::atomic::Ordering::Relaxed;
4603 4 : let (_st, mirror, _tenant) = state_with_mirror("antares-csub-clock");
4604 4 : let armed = chrono::Utc::now().timestamp_millis() + 600_000;
4605 4 : mirror.next_csub_sweep_ms.store(armed, Relaxed);
4606 4 : assert_eq!(
4607 4 : mirror.next_csub_sweep_ms.load(Relaxed),
4608 : armed,
4609 : "the clock stands until something writes"
4610 : );
4611 :
4612 4 : mirror.csub_written();
4613 :
4614 4 : assert_eq!(
4615 4 : mirror.next_csub_sweep_ms.load(Relaxed),
4616 4 : 0,
4617 4 : "writing a Context Source Registration Subscription clears the sweep clock"
4618 4 : );
4619 4 : }
4620 :
4621 : /// With nothing periodic to serve, the sweep must park past the next few
4622 : /// ticks instead of re-listing every tenant's Context Source Registration
4623 : /// Subscriptions every second: at the tenant target that poll is the
4624 : /// broker's whole idle cost.
4625 : #[tokio::test(flavor = "multi_thread")]
4626 4 : async fn a_sweep_that_finds_nothing_periodic_parks_the_csub_clock() {
4627 : use std::sync::atomic::Ordering::Relaxed;
4628 4 : let (st, mirror, _tenant) = state_with_mirror("antares-csub-park");
4629 4 : mirror.next_csub_sweep_ms.store(0, Relaxed);
4630 :
4631 4 : interval_tick(&st).await;
4632 :
4633 4 : let parked = mirror.next_csub_sweep_ms.load(Relaxed);
4634 4 : let now = chrono::Utc::now().timestamp_millis();
4635 4 : assert!(
4636 4 : parked > now + 1_000,
4637 4 : "an idle sweep parked until {parked} ({} ms out) — the poll is still the fast path",
4638 4 : parked - now
4639 4 : );
4640 4 : }
4641 : }
4642 :
4643 : #[cfg(test)]
4644 : mod clause_5_8_6_grouped_delivery {
4645 : use super::*;
4646 : use serde_json::json;
4647 : use tower::ServiceExt as _;
4648 :
4649 8 : async fn post(st: &AppState, uri: &str, body: Value) -> u16 {
4650 8 : let body = body.to_string();
4651 8 : crate::router(st.clone())
4652 8 : .oneshot(
4653 8 : axum::http::Request::builder()
4654 8 : .method("POST")
4655 8 : .uri(uri)
4656 8 : .header("Content-Type", "application/json")
4657 8 : .header("Content-Length", body.len())
4658 8 : .body(axum::body::Body::from(body))
4659 8 : .expect("request"),
4660 8 : )
4661 8 : .await
4662 8 : .expect("response")
4663 8 : .status()
4664 8 : .as_u16()
4665 8 : }
4666 :
4667 4 : async fn recording_endpoint() -> (String, Arc<std::sync::Mutex<Vec<Value>>>) {
4668 4 : let seen: Arc<std::sync::Mutex<Vec<Value>>> = Arc::default();
4669 4 : let sink = seen.clone();
4670 4 : let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
4671 4 : .await
4672 4 : .expect("bind");
4673 4 : let addr = listener.local_addr().expect("addr");
4674 4 : let app = axum::Router::new().route(
4675 4 : "/notify",
4676 4 : axum::routing::post(move |body: String| {
4677 4 : let sink = sink.clone();
4678 4 : async move {
4679 4 : if let Ok(v) = serde_json::from_str::<Value>(&body) {
4680 4 : sink.lock().expect("recorded bodies").push(v);
4681 4 : }
4682 4 : axum::http::StatusCode::OK
4683 4 : }
4684 4 : }),
4685 : );
4686 4 : tokio::spawn(async move { axum::serve(listener, app).await.expect("serve") });
4687 4 : (format!("http://{addr}/notify"), seen)
4688 4 : }
4689 :
4690 : /// 5.8.6: one batch request writing N matching entities is ONE
4691 : /// notification whose `data` carries the N entities — not N
4692 : /// notifications — and Table 5.2.14.2-1 `timesSent` moves by one.
4693 : #[tokio::test(flavor = "multi_thread")]
4694 4 : async fn a_batch_of_matching_entities_is_one_notification() {
4695 4 : crate::allow_private();
4696 4 : let (uri, seen) = recording_endpoint().await;
4697 4 : let mut st = AppState::new("antares-grouped-delivery".into());
4698 4 : crate::wire(&mut st).await;
4699 4 : assert_eq!(
4700 4 : post(
4701 4 : &st,
4702 4 : "/ngsi-ld/v1/subscriptions",
4703 4 : json!({
4704 4 : "id": "urn:ngsi-ld:Subscription:grouped",
4705 4 : "type": "Subscription",
4706 4 : "entities": [{"type": "Vehicle"}],
4707 4 : "notification": {"endpoint": {"uri": uri, "accept": "application/json"}},
4708 4 : }),
4709 4 : )
4710 4 : .await,
4711 : 201
4712 : );
4713 4 : let batch: Vec<Value> = (1..=3)
4714 12 : .map(|i| {
4715 12 : json!({"id": format!("urn:ngsi-ld:Vehicle:{i}"), "type": "Vehicle",
4716 12 : "speed": {"type": "Property", "value": i}})
4717 12 : })
4718 4 : .collect();
4719 4 : assert_eq!(
4720 4 : post(&st, "/ngsi-ld/v1/entityOperations/create", json!(batch)).await,
4721 : 201
4722 : );
4723 4 : for _ in 0..600 {
4724 4 : tokio::time::sleep(std::time::Duration::from_millis(50)).await;
4725 4 : if !seen.lock().expect("bodies").is_empty() {
4726 4 : break;
4727 0 : }
4728 : }
4729 : // settle: a second POST, if the broker were still splitting, lands here
4730 4 : tokio::time::sleep(std::time::Duration::from_millis(300)).await;
4731 4 : let bodies = seen.lock().expect("bodies").clone();
4732 4 : assert_eq!(bodies.len(), 1, "one POST for the batch, got {bodies:?}");
4733 4 : let data = bodies[0]["data"].as_array().expect("data array");
4734 12 : let mut ids: Vec<&str> = data.iter().filter_map(|e| e["id"].as_str()).collect();
4735 4 : ids.sort_unstable();
4736 4 : assert_eq!(
4737 : ids,
4738 : [
4739 : "urn:ngsi-ld:Vehicle:1",
4740 : "urn:ngsi-ld:Vehicle:2",
4741 : "urn:ngsi-ld:Vehicle:3"
4742 : ]
4743 : );
4744 4 : let tenant = TenantId::new("default").expect("tenant");
4745 4 : let sub = st
4746 4 : .store
4747 4 : .get(
4748 4 : &tenant,
4749 4 : Kind::Subscription,
4750 4 : "urn:ngsi-ld:Subscription:grouped",
4751 4 : )
4752 4 : .await
4753 4 : .expect("store")
4754 4 : .expect("row");
4755 4 : assert_eq!(sub["notification"]["timesSent"], json!(1));
4756 4 : }
4757 : }
4758 :
4759 : #[cfg(test)]
4760 : mod notification_body_bound {
4761 : use super::*;
4762 : use serde_json::json;
4763 :
4764 : /// A grouped notification is cut into whole-entity runs under the byte
4765 : /// cap; nothing is split mid-entity and an oversize single entity still
4766 : /// travels (alone) instead of being dropped.
4767 : #[test]
4768 4 : fn chunks_cut_at_whole_items_under_the_cap() {
4769 32 : let item = |i: usize| json!({"id": format!("urn:ngsi-ld:V:{i}"), "v": "x".repeat(40)});
4770 4 : let one = serde_json::to_vec(&item(0)).expect("json").len();
4771 4 : let runs = chunk_by_bytes((0..5).map(item).collect(), one * 2);
4772 4 : assert_eq!(runs.iter().map(Vec::len).collect::<Vec<_>>(), [2, 2, 1]);
4773 4 : assert_eq!(runs[2][0]["id"], json!("urn:ngsi-ld:V:4"));
4774 4 : let runs = chunk_by_bytes(vec![item(0), item(1)], one / 2);
4775 4 : assert_eq!(runs.len(), 2, "an over-cap item is its own run, never lost");
4776 4 : assert!(chunk_by_bytes(Vec::new(), 1).is_empty());
4777 4 : }
4778 : }
4779 :
4780 : #[cfg(test)]
4781 : mod interval_claim {
4782 : use super::*;
4783 : use serde_json::json;
4784 :
4785 8 : async fn seed(st: &AppState, tenant: &TenantId) {
4786 8 : st.store
4787 8 : .create(
4788 8 : tenant,
4789 8 : Kind::Subscription,
4790 8 : "urn:ngsi-ld:Subscription:tick",
4791 8 : json!({
4792 8 : "id": "urn:ngsi-ld:Subscription:tick",
4793 8 : "type": "Subscription",
4794 8 : "entities": [{"type": "https://uri.etsi.org/ngsi-ld/default-context/Vehicle"}],
4795 8 : "timeInterval": 1,
4796 8 : "status": "active",
4797 8 : "createdAt": "2020-01-01T00:00:00Z",
4798 8 : "notification": {"endpoint": {"uri": "http://127.0.0.1:9/notify"}},
4799 8 : }),
4800 8 : )
4801 8 : .await
4802 8 : .expect("seed");
4803 8 : }
4804 :
4805 8 : async fn stamp(st: &AppState, tenant: &TenantId) -> Option<String> {
4806 8 : st.store
4807 8 : .get(tenant, Kind::Subscription, "urn:ngsi-ld:Subscription:tick")
4808 8 : .await
4809 8 : .expect("store")
4810 8 : .expect("row")["notification"]["lastNotification"]
4811 8 : .as_str()
4812 8 : .map(str::to_owned)
4813 8 : }
4814 :
4815 : /// 5.8.6: "If there are no matching Entities, no Notification is sent",
4816 : /// and Table 5.2.14.2-1 makes `lastNotification` "the timestamp
4817 : /// corresponding to the instant when the last notification was sent".
4818 : /// The multi-pod claim stamps that member to win the firing, so a due
4819 : /// subscription that then matches nothing must have the stamp put back:
4820 : /// otherwise a client reads a notification instant for a notification
4821 : /// that never happened, and the subscription — still owing its firing —
4822 : /// waits out a whole interval that the single-process path does not.
4823 : #[tokio::test]
4824 4 : async fn a_claimed_firing_that_matches_nothing_leaves_no_notification_instant() {
4825 4 : let tenant = TenantId::new("default").expect("tenant");
4826 8 : for nats in [false, true] {
4827 8 : let mut st = AppState::new("me".into());
4828 8 : st.nats = nats;
4829 8 : seed(&st, &tenant).await;
4830 8 : interval_tick(&st).await;
4831 8 : assert_eq!(
4832 8 : stamp(&st, &tenant).await,
4833 4 : None,
4834 4 : "nats={nats}: nothing matched, so nothing was sent"
4835 4 : );
4836 4 : }
4837 4 : }
4838 : }
4839 :
4840 : #[cfg(all(test, not(target_arch = "wasm32")))]
4841 : mod change_grouping {
4842 : use super::*;
4843 : use std::sync::atomic::{AtomicUsize, Ordering};
4844 : use std::sync::Arc as StdArc;
4845 :
4846 : /// 5.8.6: "the Notification … data … shall contain the Entities that
4847 : /// match" — every change of one drain that matches the same Subscription
4848 : /// travels in ONE notification. Grouping is what makes a batch of N
4849 : /// writes one POST with N data entries instead of N POSTs, and it has to
4850 : /// hold when many Subscriptions match the same change: each of them gets
4851 : /// exactly one notification carrying every entity, and no entity lands on
4852 : /// the wrong Subscription.
4853 : #[tokio::test(flavor = "multi_thread")]
4854 4 : async fn every_change_of_a_drain_reaches_each_matching_subscription_once() {
4855 4 : crate::allow_private();
4856 4 : let mut st = AppState::new("antares-grouping-test".into());
4857 4 : crate::wire(&mut st).await;
4858 4 : let tenant = TenantId::new("default").expect("tenant");
4859 4 : let posts: StdArc<AtomicUsize> = StdArc::default();
4860 4 : let entities: StdArc<std::sync::Mutex<Vec<usize>>> = StdArc::default();
4861 4 : let (p, e) = (posts.clone(), entities.clone());
4862 4 : let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
4863 4 : .await
4864 4 : .expect("bind");
4865 4 : let addr = listener.local_addr().expect("addr");
4866 4 : let app = axum::Router::new().route(
4867 4 : "/notify",
4868 32 : axum::routing::post(move |body: String| {
4869 32 : let (p, e) = (p.clone(), e.clone());
4870 32 : async move {
4871 32 : p.fetch_add(1, Ordering::SeqCst);
4872 32 : let v: Value = serde_json::from_str(&body).unwrap_or(Value::Null);
4873 32 : e.lock()
4874 32 : .expect("seen")
4875 32 : .push(v["data"].as_array().map(Vec::len).unwrap_or(0));
4876 32 : axum::http::StatusCode::OK
4877 32 : }
4878 32 : }),
4879 : );
4880 4 : tokio::spawn(async move {
4881 4 : axum::serve(listener, app).await.expect("serve");
4882 0 : });
4883 :
4884 : const SUBS: usize = 8;
4885 : const CHANGES: usize = 5;
4886 32 : for i in 0..SUBS {
4887 32 : let sub = json!({
4888 32 : "id": format!("urn:ngsi-ld:Subscription:group-{i}"),
4889 32 : "type": "Subscription",
4890 32 : "status": "active",
4891 32 : "entities": [{"type": "https://uri.etsi.org/ngsi-ld/default-context/Vehicle"}],
4892 32 : "notification": {"endpoint": {"uri": format!("http://{addr}/notify")}},
4893 : });
4894 32 : st.store
4895 32 : .create(
4896 32 : &tenant,
4897 32 : Kind::Subscription,
4898 32 : &format!("urn:ngsi-ld:Subscription:group-{i}"),
4899 32 : sub.clone(),
4900 32 : )
4901 32 : .await
4902 32 : .expect("seed subscription");
4903 32 : if let Some(m) = &st.sub_mirror {
4904 32 : m.apply(
4905 32 : tenant.as_str(),
4906 32 : &format!("urn:ngsi-ld:Subscription:group-{i}"),
4907 32 : Some(sub),
4908 32 : );
4909 32 : }
4910 : }
4911 4 : let changes: Vec<Change> = (0..CHANGES)
4912 20 : .map(|i| {
4913 20 : (
4914 20 : tenant.as_str().to_owned(),
4915 20 : None,
4916 20 : Some(json!({
4917 20 : "id": format!("urn:ngsi-ld:Vehicle:{i}"),
4918 20 : "type": ["https://uri.etsi.org/ngsi-ld/default-context/Vehicle"],
4919 20 : "https://uri.etsi.org/ngsi-ld/default-context/speed": [
4920 20 : {"type": "Property", "value": i}
4921 20 : ],
4922 20 : })),
4923 20 : )
4924 20 : })
4925 4 : .collect();
4926 4 : process_changes(&st, changes).await;
4927 4 : assert_eq!(
4928 4 : posts.load(Ordering::SeqCst),
4929 : SUBS,
4930 : "one notification per matching subscription, never one per change"
4931 : );
4932 4 : let sizes = entities.lock().expect("seen").clone();
4933 4 : assert_eq!(
4934 4 : sizes,
4935 4 : vec![CHANGES; SUBS],
4936 4 : "each notification carries every entity of the drain"
4937 4 : );
4938 4 : }
4939 : }
4940 :
4941 : #[cfg(all(test, not(target_arch = "wasm32")))]
4942 : mod interval_sweep_concurrency {
4943 : use super::*;
4944 :
4945 : /// 5.8.6 sends a periodic Notification "when the time interval … is
4946 : /// reached". One sweep visits every tenant and every due Subscription, so
4947 : /// whatever it does per Subscription it does 10 000 tenants' worth of —
4948 : /// and a notification endpoint may take its whole `endpoint.timeout` to
4949 : /// answer (Table 5.2.15-1). Awaiting each delivery in turn makes one
4950 : /// unresponsive endpoint the deadline of every other subscriber's
4951 : /// periodic notification, on a broker whose targets are 10 000 tenants
4952 : /// and 100 000 subscriptions.
4953 : #[tokio::test(flavor = "multi_thread")]
4954 4 : async fn one_unresponsive_endpoint_does_not_hold_up_the_other_subscriptions() {
4955 4 : crate::allow_private();
4956 : // a listener that accepts and never answers: every delivery to it
4957 : // costs exactly its endpoint timeout
4958 4 : let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
4959 4 : .await
4960 4 : .expect("bind");
4961 4 : let addr = listener.local_addr().expect("addr");
4962 4 : tokio::spawn(async move {
4963 4 : let mut held = Vec::new();
4964 20 : while let Ok((s, _)) = listener.accept().await {
4965 16 : held.push(s);
4966 16 : }
4967 0 : });
4968 : // no mirror: the sweep reads the subscriptions from the store, which
4969 : // is where this test seeds them
4970 4 : let st = AppState::new("antares-sweep-test".into());
4971 4 : let tenant = TenantId::new("default").expect("tenant");
4972 4 : st.store
4973 4 : .create(
4974 4 : &tenant,
4975 4 : Kind::Entity,
4976 4 : "urn:ngsi-ld:Vehicle:sweep",
4977 4 : json!({
4978 4 : "id": "urn:ngsi-ld:Vehicle:sweep",
4979 4 : "type": ["https://uri.etsi.org/ngsi-ld/default-context/Vehicle"],
4980 4 : }),
4981 4 : )
4982 4 : .await
4983 4 : .expect("seed entity");
4984 : const SUBS: u32 = 4;
4985 : const TIMEOUT_MS: u64 = 300;
4986 16 : for i in 0..SUBS {
4987 16 : let id = format!("urn:ngsi-ld:Subscription:sweep-{i}");
4988 16 : st.store
4989 16 : .create(
4990 16 : &tenant,
4991 16 : Kind::Subscription,
4992 16 : &id,
4993 16 : json!({
4994 16 : "id": id,
4995 16 : "type": "Subscription",
4996 16 : "status": "active",
4997 16 : "timeInterval": 1,
4998 16 : "createdAt": "2020-01-01T00:00:00Z",
4999 16 : "entities": [{
5000 16 : "type": "https://uri.etsi.org/ngsi-ld/default-context/Vehicle"
5001 16 : }],
5002 16 : "notification": {"endpoint": {
5003 16 : "uri": format!("http://{addr}/notify"),
5004 16 : "timeout": TIMEOUT_MS,
5005 16 : }},
5006 16 : }),
5007 16 : )
5008 16 : .await
5009 16 : .expect("seed subscription");
5010 : }
5011 4 : let started = std::time::Instant::now();
5012 4 : interval_tick(&st).await;
5013 4 : let elapsed = started.elapsed().as_millis() as u64;
5014 4 : let serial = TIMEOUT_MS * u64::from(SUBS);
5015 4 : assert!(
5016 4 : elapsed < serial * crate::state::slow_factor(),
5017 4 : "the sweep took {elapsed} ms — {SUBS} deliveries of {TIMEOUT_MS} ms ran one after \
5018 4 : the other instead of together"
5019 4 : );
5020 4 : }
5021 : }
5022 :
5023 : /// What `stable_eq` treats as the same attribute instance (5.8.6: a change
5024 : /// notification carries the attributes that changed, and the members the
5025 : /// broker stamps itself are not a change).
5026 : #[cfg(test)]
5027 : mod stable_comparison {
5028 : use super::*;
5029 : use serde_json::json;
5030 :
5031 : #[test]
5032 4 : fn only_the_volatile_members_may_differ() {
5033 4 : let base = json!({"type": "Property", "value": 1,
5034 4 : "createdAt": "2020-01-01T00:00:00Z",
5035 4 : "modifiedAt": "2020-01-01T00:00:00Z",
5036 4 : "instanceId": "urn:ngsi-ld:Instance:1"});
5037 : // every volatile member differs, or is missing outright
5038 4 : assert!(stable_eq(&base, &json!({"type": "Property", "value": 1})));
5039 4 : assert!(stable_eq(
5040 4 : &base,
5041 4 : &json!({"type": "Property", "value": 1,
5042 4 : "modifiedAt": "2021-06-06T00:00:00Z",
5043 4 : "instanceId": "urn:ngsi-ld:Instance:2"})
5044 : ));
5045 : // a member that is not volatile
5046 4 : assert!(!stable_eq(&base, &json!({"type": "Property", "value": 2})));
5047 4 : assert!(!stable_eq(
5048 4 : &base,
5049 4 : &json!({"type": "Property", "value": 1, "unitCode": "CEL"})
5050 4 : ));
5051 : // same count of non-volatile members, different names
5052 4 : assert!(!stable_eq(
5053 4 : &json!({"a": 1, "b": 2}),
5054 4 : &json!({"a": 1, "c": 2})
5055 4 : ));
5056 : // volatile only counts as a member name, never as a value
5057 4 : assert!(!stable_eq(
5058 4 : &json!({"observedAt": "createdAt"}),
5059 4 : &json!({"observedAt": "modifiedAt"})
5060 4 : ));
5061 4 : }
5062 :
5063 : #[test]
5064 4 : fn nesting_and_arrays_compare_element_by_element() {
5065 4 : assert!(stable_eq(
5066 4 : &json!({"value": {"a": [1, {"b": 2, "createdAt": "x"}]}}),
5067 4 : &json!({"value": {"a": [1, {"b": 2}]}})
5068 : ));
5069 4 : assert!(!stable_eq(
5070 4 : &json!({"value": {"a": [1, {"b": 2}]}}),
5071 4 : &json!({"value": {"a": [1, {"b": 3}]}})
5072 4 : ));
5073 : // order within an array is part of the value
5074 4 : assert!(!stable_eq(&json!([1, 2]), &json!([2, 1])));
5075 4 : assert!(!stable_eq(&json!([1, 2]), &json!([1, 2, 3])));
5076 : // scalars and mismatched kinds
5077 4 : assert!(stable_eq(&json!(null), &json!(null)));
5078 4 : assert!(!stable_eq(&json!({}), &json!([])));
5079 4 : assert!(!stable_eq(&json!(1), &json!("1")));
5080 4 : }
5081 : }
5082 :
5083 : /// How much of the notification width one tenant may hold.
5084 : #[cfg(all(test, not(target_arch = "wasm32")))]
5085 : mod delivery_share {
5086 : use super::*;
5087 :
5088 : /// A Subscription belongs to one tenant (5.2.12), and a delivery to an
5089 : /// endpoint that accepts and never answers holds its slot until the
5090 : /// endpoint's timeout expires — up to 30 s (Table 5.2.15-1). A tenant
5091 : /// that points more subscriptions than the whole width at such an
5092 : /// endpoint would, with one width shared broker-wide, hold every slot
5093 : /// for that long and stop every other tenant's notifications. A tenant
5094 : /// therefore takes a share of the width, never all of it.
5095 : #[tokio::test(flavor = "multi_thread")]
5096 4 : async fn one_tenant_takes_a_share_of_the_delivery_width_not_all_of_it() {
5097 4 : crate::allow_private();
5098 : // accepts and never answers: each connection is held for the whole
5099 : // endpoint timeout, so the connections open at once ARE the
5100 : // deliveries in flight
5101 4 : let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
5102 4 : .await
5103 4 : .expect("bind");
5104 4 : let addr = listener.local_addr().expect("addr");
5105 4 : let opened = Arc::new(std::sync::atomic::AtomicUsize::new(0));
5106 4 : let counter = Arc::clone(&opened);
5107 4 : tokio::spawn(async move {
5108 4 : let mut held = Vec::new();
5109 36 : while let Ok((s, _)) = listener.accept().await {
5110 32 : held.push(s);
5111 32 : counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
5112 32 : }
5113 0 : });
5114 4 : let st = AppState::new("antares-share-test".into());
5115 4 : let tenant = TenantId::new("hog").expect("tenant");
5116 4 : st.store
5117 4 : .create(
5118 4 : &tenant,
5119 4 : Kind::Entity,
5120 4 : "urn:ngsi-ld:Vehicle:share",
5121 4 : json!({
5122 4 : "id": "urn:ngsi-ld:Vehicle:share",
5123 4 : "type": ["https://uri.etsi.org/ngsi-ld/default-context/Vehicle"],
5124 4 : }),
5125 4 : )
5126 4 : .await
5127 4 : .expect("seed entity");
5128 : // more subscriptions than the whole width, so an unbounded tenant
5129 : // would hold every slot
5130 4 : let subs = *DELIVERY_WIDTH + *DELIVERY_WIDTH_PER_TENANT;
5131 288 : for i in 0..subs {
5132 288 : let id = format!("urn:ngsi-ld:Subscription:share-{i}");
5133 288 : st.store
5134 288 : .create(
5135 288 : &tenant,
5136 288 : Kind::Subscription,
5137 288 : &id,
5138 288 : json!({
5139 288 : "id": id,
5140 288 : "type": "Subscription",
5141 288 : "status": "active",
5142 288 : "timeInterval": 1,
5143 288 : "createdAt": "2020-01-01T00:00:00Z",
5144 288 : "entities": [{
5145 288 : "type": "https://uri.etsi.org/ngsi-ld/default-context/Vehicle"
5146 288 : }],
5147 288 : // long enough that nothing in the first round has
5148 288 : // timed out and freed its slot while this is counted
5149 288 : "notification": {"endpoint": {
5150 288 : "uri": format!("http://{addr}/notify"),
5151 288 : "timeout": 30_000,
5152 288 : }},
5153 288 : }),
5154 288 : )
5155 288 : .await
5156 288 : .expect("seed subscription");
5157 : }
5158 4 : let sweep = tokio::spawn(async move { interval_tick(&st).await });
5159 4 : let settle = 400 * crate::state::slow_factor();
5160 4 : tokio::time::sleep(std::time::Duration::from_millis(settle)).await;
5161 4 : let in_flight = opened.load(std::sync::atomic::Ordering::SeqCst);
5162 4 : sweep.abort();
5163 4 : assert!(in_flight > 0, "the sweep delivered nothing at all");
5164 4 : assert!(
5165 4 : in_flight <= *DELIVERY_WIDTH_PER_TENANT,
5166 4 : "one tenant held {in_flight} of the {} delivery slots",
5167 4 : *DELIVERY_WIDTH
5168 4 : );
5169 4 : }
5170 : }
5171 :
5172 : /// The drain's memo of resolved `@contexts`.
5173 : #[cfg(all(test, not(target_arch = "wasm32")))]
5174 : mod context_memo {
5175 : use super::*;
5176 :
5177 : /// 5.5.10 scopes an `@context` resolution to one Tenant, so the drain's
5178 : /// memo is keyed by Tenant before URL: a resolution one Tenant reached is
5179 : /// not reachable under another Tenant naming the same URL.
5180 : #[tokio::test]
5181 4 : async fn one_tenants_memoised_context_is_not_served_to_another() {
5182 4 : let st = AppState::new("antares-ctx-memo-tenant".into());
5183 4 : let url = "https://uri.etsi.org/ngsi-ld/v1/ngsi-ld-core-context-v1.9.jsonld";
5184 4 : let sub = json!({"id": "urn:ngsi-ld:Subscription:1", "__context": url});
5185 4 : let planted = Arc::new(Context::default());
5186 4 : let mut memo = CtxMemo::new();
5187 4 : memo.entry("alpha".to_owned())
5188 4 : .or_default()
5189 4 : .insert(url.to_owned(), Arc::clone(&planted));
5190 :
5191 4 : let beta = TenantId::new_internal("beta").expect("tenant");
5192 4 : let got = sub_context_memo(&st, &beta, "beta", &sub, &mut memo).await;
5193 :
5194 4 : assert!(
5195 4 : !Arc::ptr_eq(&got, &planted),
5196 : "Tenant beta was served Tenant alpha's resolved @context"
5197 : );
5198 4 : assert!(
5199 4 : memo.get("alpha").and_then(|m| m.get(url)).is_some(),
5200 4 : "alpha's entry was overwritten by another Tenant's resolution"
5201 4 : );
5202 4 : }
5203 :
5204 : /// The memo exists to spare the resolver a lookup per candidate: a second
5205 : /// candidate naming the same `@context` in the same drain reuses the
5206 : /// resolution the first one reached rather than resolving again.
5207 : #[tokio::test]
5208 4 : async fn a_second_candidate_reuses_the_resolution_of_the_first() {
5209 4 : let st = AppState::new("antares-ctx-memo-reuse".into());
5210 4 : let url = "https://uri.etsi.org/ngsi-ld/v1/ngsi-ld-core-context-v1.9.jsonld";
5211 4 : let tenant = TenantId::new_internal("alpha").expect("tenant");
5212 4 : let mut memo = CtxMemo::new();
5213 :
5214 4 : let first = sub_context_memo(
5215 4 : &st,
5216 4 : &tenant,
5217 4 : "alpha",
5218 4 : &json!({"id": "urn:ngsi-ld:Subscription:1", "__context": url}),
5219 4 : &mut memo,
5220 4 : )
5221 4 : .await;
5222 4 : let second = sub_context_memo(
5223 4 : &st,
5224 4 : &tenant,
5225 4 : "alpha",
5226 4 : &json!({"id": "urn:ngsi-ld:Subscription:2", "__context": url}),
5227 4 : &mut memo,
5228 4 : )
5229 4 : .await;
5230 :
5231 4 : assert!(
5232 4 : Arc::ptr_eq(&first, &second),
5233 4 : "the second candidate resolved its @context again"
5234 4 : );
5235 4 : }
5236 :
5237 : /// A Subscription carrying an inline `@context` rather than a URL has no
5238 : /// cheap identity to key on: it resolves as it did before the memo, and
5239 : /// leaves no entry behind that a later URL lookup could collide with.
5240 : #[tokio::test]
5241 4 : async fn an_inline_context_is_resolved_rather_than_memoised() {
5242 4 : let st = AppState::new("antares-ctx-memo-inline".into());
5243 4 : let tenant = TenantId::new_internal("alpha").expect("tenant");
5244 4 : let mut memo = CtxMemo::new();
5245 4 : let sub = json!({
5246 4 : "id": "urn:ngsi-ld:Subscription:1",
5247 4 : "__context": {"Vehicle": "https://example.org/Vehicle"},
5248 : });
5249 :
5250 4 : let ctx = sub_context_memo(&st, &tenant, "alpha", &sub, &mut memo).await;
5251 :
5252 4 : assert_eq!(ctx.expand_key("Vehicle"), "https://example.org/Vehicle");
5253 4 : assert!(memo.is_empty(), "an inline @context left a memo entry");
5254 4 : }
5255 : }
|
