Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! The CONSUMER half of distributed subscriptions (5.8.1.4 / 5.8.2.4 /
3 : //! 5.8.5.4): an entity Subscription with localOnly != true creates an
4 : //! internal Context Source Registration Subscription (5.11.2) whose
5 : //! CSource notifications drive per-registration remote subscription
6 : //! create/update/delete (triggerReason newlyMatching / updated /
7 : //! noLongerMatching), with subscriptionId mappings stored so inbound
8 : //! remote notifications forward to the original subscriber.
9 :
10 : use crate::negotiate::{ApiError, ApiResult};
11 : use crate::state::{now_iso, AppState};
12 : use antares_model::{NgsiError, TenantId};
13 : use antares_store::CurrentStateDriverExt;
14 : use antares_store::Kind;
15 : use axum::body::Bytes;
16 : use axum::extract::State;
17 : use axum::http::{HeaderMap, StatusCode};
18 : use axum::response::{IntoResponse, Response};
19 : use serde_json::{json, Value};
20 :
21 270 : fn ds_index_tenant() -> Option<TenantId> {
22 270 : TenantId::new_internal("distsub-index").ok()
23 270 : }
24 :
25 2020 : async fn ds_get(st: &AppState, tenant: &TenantId, own_id: &str) -> Value {
26 2020 : st.store
27 2020 : .get(tenant, Kind::DistSub, own_id)
28 2020 : .await
29 2020 : .ok()
30 2020 : .flatten()
31 2020 : .unwrap_or_else(|| json!({}))
32 2020 : }
33 :
34 1002 : async fn ds_put(st: &AppState, tenant: &TenantId, own_id: &str, doc: Value) {
35 1002 : let updated = st
36 1002 : .store
37 1002 : .mutate(tenant, Kind::DistSub, own_id, |d| {
38 0 : *d = doc.clone();
39 0 : Ok::<_, std::convert::Infallible>(())
40 0 : })
41 1002 : .await
42 1002 : .ok()
43 1002 : .flatten()
44 1002 : .is_some();
45 1002 : if !updated {
46 1002 : if let Err(e) = st.store.create(tenant, Kind::DistSub, own_id, doc).await {
47 0 : tracing::warn!("subscription {own_id}: distributed mapping not stored: {e}");
48 1002 : }
49 0 : }
50 1002 : }
51 :
52 : /// remotes of one own Subscription: reg id → (endpoint, remote sub id)
53 886 : fn ds_remotes(doc: &Value) -> Vec<(String, (String, String))> {
54 886 : doc.get("remotes")
55 886 : .and_then(Value::as_object)
56 886 : .map(|m| {
57 130 : m.iter()
58 146 : .filter_map(|(reg, v)| {
59 : Some((
60 146 : reg.clone(),
61 : (
62 146 : v.get(0)?.as_str()?.to_owned(),
63 134 : v.get(1)?.as_str()?.to_owned(),
64 : ),
65 : ))
66 146 : })
67 130 : .collect()
68 130 : })
69 886 : .unwrap_or_default()
70 886 : }
71 :
72 : /// Insert or remove ONE `remotes` entry under the store's own lock, and
73 : /// only if the document still exists. A targeted mutate, never a
74 : /// read-modify-write of the whole document: the per-registration branches of
75 : /// 5.8.1.4 interleave at their forward `await`s, so a full-document write
76 : /// would drop a sibling registration's mapping, and a Delete Subscription
77 : /// (5.8.5.4) that lands mid-forward would be undone by a write that
78 : /// resurrects the deleted document. `false` = the Subscription's mapping
79 : /// document is gone, or the registration already holds a remote subscription
80 : /// — 5.8.1.4 stores ONE subscriptionId per Context Source Registration, and
81 : /// two notifications racing for the same pair would otherwise each create a
82 : /// remote copy and orphan the loser's at the source. That check runs inside
83 : /// the closure the store executes under its write lock, so the same lock
84 : /// decides and acts.
85 136 : async fn ds_set_remote(
86 136 : st: &AppState,
87 136 : tenant: &TenantId,
88 136 : own_id: &str,
89 136 : reg_id: &str,
90 136 : entry: Option<Value>,
91 136 : ) -> bool {
92 136 : st.store
93 136 : .mutate(tenant, Kind::DistSub, own_id, |d| {
94 132 : if let Some(o) = d.as_object_mut() {
95 132 : match &entry {
96 116 : Some(v) => {
97 116 : if !o.get("remotes").is_some_and(Value::is_object) {
98 108 : o.insert("remotes".into(), json!({}));
99 108 : }
100 116 : if let Some(m) = o.get_mut("remotes").and_then(Value::as_object_mut) {
101 116 : if m.contains_key(reg_id) {
102 4 : return Err(());
103 112 : }
104 112 : m.insert(reg_id.to_owned(), v.clone());
105 0 : }
106 : }
107 : None => {
108 16 : if let Some(m) = o.get_mut("remotes").and_then(Value::as_object_mut) {
109 16 : m.remove(reg_id);
110 16 : }
111 : }
112 : }
113 0 : }
114 128 : Ok(())
115 132 : })
116 136 : .await
117 136 : .ok()
118 136 : .flatten()
119 136 : .is_some_and(|r: Result<(), ()>| r.is_ok())
120 136 : }
121 :
122 120 : async fn inbound_put(st: &AppState, remote_id: &str, tenant: &TenantId, own_id: &str) {
123 120 : if let Some(idx) = ds_index_tenant() {
124 : // Without this index every notification the source sends is answered
125 : // 404 and the subscription silently never notifies — a store failure
126 : // here has to be visible to an operator.
127 120 : if let Err(e) = st
128 120 : .store
129 120 : .create(
130 120 : &idx,
131 120 : Kind::DistSub,
132 120 : remote_id,
133 120 : json!({"tenant": tenant.as_str(), "own": own_id}),
134 120 : )
135 120 : .await
136 : {
137 : // Not the remote id: it is the key /ex/v1/remote-notify accepts
138 : // a notification by, and a log line is no place for it.
139 0 : tracing::warn!("subscription {own_id}: inbound mapping not stored: {e}");
140 120 : }
141 0 : }
142 120 : }
143 :
144 102 : async fn inbound_get(st: &AppState, remote_id: &str) -> Option<(String, String)> {
145 102 : let idx = ds_index_tenant()?;
146 102 : let doc = st
147 102 : .store
148 102 : .get(&idx, Kind::DistSub, remote_id)
149 102 : .await
150 102 : .ok()
151 102 : .flatten()?;
152 : Some((
153 74 : doc.get("tenant")?.as_str()?.to_owned(),
154 74 : doc.get("own")?.as_str()?.to_owned(),
155 : ))
156 102 : }
157 :
158 : impl AppState {
159 : /// The Tenant whose distributed Subscription receives the notifications
160 : /// of the copy `subscription_id` (5.8.1.4), read from this broker's own
161 : /// inbound index, never from the notification.
162 4 : pub async fn notification_receiver(&self, subscription_id: &str) -> Option<String> {
163 4 : inbound_get(self, subscription_id)
164 4 : .await
165 4 : .map(|(tenant, _)| tenant)
166 4 : }
167 : }
168 :
169 48 : async fn inbound_delete(st: &AppState, remote_id: &str) {
170 48 : if let Some(idx) = ds_index_tenant() {
171 48 : let _ = st.store.delete(&idx, Kind::DistSub, remote_id).await;
172 0 : }
173 48 : }
174 :
175 1182 : fn distributed(sub: &Value) -> bool {
176 1182 : sub.get("localOnly").and_then(Value::as_bool) != Some(true)
177 1182 : }
178 :
179 : /// 6.3.17/6.3.18: the Via chain the Subscription arrived with, rebuilt as
180 : /// the header the outbound forward extends (`federation::forward` appends
181 : /// this broker's alias). Stored on the Subscription as the broker-internal
182 : /// `__via` member (`__context` is the precedent) by create.
183 1082 : fn sub_via_headers(sub: &Value) -> HeaderMap {
184 1082 : let mut h = HeaderMap::new();
185 1082 : if let Some(v) = sub
186 1082 : .get("__via")
187 1082 : .and_then(Value::as_str)
188 1082 : .and_then(|v| axum::http::HeaderValue::from_str(v).ok())
189 52 : {
190 52 : h.insert("via", v);
191 1030 : }
192 1082 : h
193 1082 : }
194 :
195 : /// The Subscription members 5.11.2.4 matches a Context Source Registration
196 : /// on, plus the @context the internal Registration Subscription is read
197 : /// under. Kept in one place: create and update must offer the very same
198 : /// document, or an update silently widens which sources are matched.
199 : const CSR_MATCH_MEMBERS: [&str; 6] = [
200 : "entities",
201 : "watchedAttributes",
202 : "csf",
203 : "geoQ",
204 : "scopeQ",
205 : "temporalQ",
206 : ];
207 :
208 : /// 5.8.1.4: "Based on the content of the Subscription, a Context Source
209 : /// Registration Subscription shall be created (clause 5.11.2)" — internal,
210 : /// with the urn:antares:distsub endpoint handled in-process by notify.
211 974 : pub(crate) async fn on_subscription_created(st: &AppState, tenant: &TenantId, sub: &Value) {
212 974 : if !distributed(sub) {
213 8 : return;
214 966 : }
215 : // 6.3.18 ("to avoid infinite loops"): a forwarded copy whose Via chain
216 : // already names this broker has come full circle — it serves locally,
217 : // and the distributed half is not created, so mutually registered
218 : // brokers cannot re-forward copies of copies without bound. via_loop
219 : // also enforces the MAX_VIA_HOPS ceiling on a forged chain.
220 966 : if crate::federation::via_loop(
221 966 : &sub_via_headers(sub),
222 966 : &crate::federation::alias_for(&st.host_alias, tenant),
223 : ) {
224 12 : return;
225 954 : }
226 954 : let Some(own_id) = sub.get("id").and_then(Value::as_str) else {
227 0 : return;
228 : };
229 954 : let csr_id = format!(
230 : "{}{}",
231 : crate::registry::INTERNAL_CSR_PREFIX,
232 954 : uuid::Uuid::new_v4()
233 : );
234 954 : let ts = now_iso();
235 954 : let mut doc = json!({
236 954 : "id": csr_id,
237 954 : "type": "Subscription",
238 954 : "isActive": true,
239 954 : "status": "active",
240 954 : "createdAt": ts,
241 954 : "modifiedAt": ts,
242 954 : "notification": {"endpoint": {"uri":
243 954 : format!("urn:antares:distsub:{}\n{own_id}", tenant.as_str())}},
244 : });
245 : // 5.11.2.4 matches a registration on all of these, so the Registration
246 : // Subscription carries every member that decides the match — a copy
247 : // built from the entity selectors alone offers the Subscription to
248 : // sources the subscriber's csf, geoQ, scopeQ or temporalQ excluded.
249 : // `q` is deliberately absent: on a Context Source Registration
250 : // Subscription it would filter registration properties, not Entity
251 : // Attributes.
252 954 : if let Some(o) = doc.as_object_mut() {
253 6678 : for k in CSR_MATCH_MEMBERS.iter().chain(["__context"].iter()) {
254 6678 : if let Some(v) = sub.get(k) {
255 2070 : o.insert((*k).to_owned(), v.clone());
256 4608 : }
257 : }
258 0 : }
259 60 : if let (Some(a), Some(n)) = (
260 954 : sub.get("notification").and_then(|n| n.get("attributes")),
261 954 : doc.get_mut("notification").and_then(Value::as_object_mut),
262 60 : ) {
263 60 : n.insert("attributes".into(), a.clone());
264 894 : }
265 954 : let created = st
266 954 : .store
267 954 : .create(tenant, Kind::DistSub, &csr_id, doc)
268 954 : .await
269 954 : .unwrap_or_else(|e| {
270 0 : tracing::warn!("subscription {own_id}: Registration Subscription not created: {e}");
271 0 : false
272 0 : });
273 954 : if created {
274 954 : let mut doc = ds_get(st, tenant, own_id).await;
275 954 : if let Some(o) = doc.as_object_mut() {
276 954 : o.insert("csr_sub".into(), Value::String(csr_id.clone()));
277 954 : }
278 954 : ds_put(st, tenant, own_id, doc).await;
279 : // 5.11.2.4 initial notification with all matching registrations —
280 : // this is what turns already-known registrations into newlyMatching
281 954 : let (st2, t2) = (st.clone(), tenant.clone());
282 954 : crate::spawn(async move {
283 882 : crate::notify::csource_initial(&st2, &t2, &csr_id).await;
284 878 : });
285 0 : }
286 974 : }
287 :
288 : /// 5.8.2.4: keep the internal CSR subscription in step and forward reduced
289 : /// updates to every mapped remote supporting updateSubscription (5.11.3).
290 92 : pub(crate) async fn on_subscription_updated(st: &AppState, tenant: &TenantId, own_id: &str) {
291 92 : let Some(sub) = st
292 92 : .store
293 92 : .get(tenant, Kind::Subscription, own_id)
294 92 : .await
295 92 : .ok()
296 92 : .flatten()
297 : else {
298 0 : return;
299 : };
300 : // 5.8.1.4 gates the distributed half on "If localOnly=false": a
301 : // Subscription updated to localOnly=true is torn down like a deleted
302 : // one — the internal Registration Subscription goes, and every remote
303 : // copy already created is deleted at its source.
304 92 : if !distributed(&sub) {
305 4 : on_subscription_deleted(st, tenant, own_id).await;
306 4 : return;
307 88 : }
308 88 : let csr_id = ds_get(st, tenant, own_id)
309 88 : .await
310 88 : .get("csr_sub")
311 88 : .and_then(Value::as_str)
312 88 : .map(str::to_owned);
313 : // the initial Context Source Notification of the updated Registration
314 : // Subscription, started once the mapped copies are read below
315 88 : let mut initial: Option<String> = None;
316 88 : match csr_id {
317 : None => {
318 : // became distributed only now (e.g. localOnly flipped off)
319 0 : on_subscription_created(st, tenant, &sub).await;
320 : }
321 88 : Some(csr_id) => {
322 88 : let _ = st
323 88 : .store
324 88 : .mutate(tenant, Kind::DistSub, &csr_id, |doc| {
325 : // The mapping comes out of the store, and `Value`'s index
326 : // panics on anything that is not an object — inside a
327 : // closure `mutate` runs holding the write lock, so the shape
328 : // is decided once, here, and a mapping that is not one is
329 : // left exactly as it was found.
330 88 : let Some(map) = doc.as_object_mut() else {
331 16 : return Ok(());
332 : };
333 432 : for k in CSR_MATCH_MEMBERS {
334 432 : match sub.get(k) {
335 76 : Some(v) => {
336 76 : map.insert(k.to_owned(), v.clone());
337 76 : }
338 356 : None => {
339 356 : map.remove(k);
340 356 : }
341 : }
342 : }
343 72 : let attrs = sub.get("notification").and_then(|n| n.get("attributes"));
344 72 : match map.get_mut("notification").and_then(Value::as_object_mut) {
345 64 : Some(n) => match attrs {
346 4 : Some(a) => {
347 4 : n.insert("attributes".to_owned(), a.clone());
348 4 : }
349 60 : None => {
350 60 : n.remove("attributes");
351 60 : }
352 : },
353 : // no notification object to carry them: the copy this
354 : // mapping stands for is the one create built, so a
355 : // mapping without it is not one an update can complete
356 8 : None => return Ok(()),
357 : }
358 64 : map.insert("modifiedAt".to_owned(), Value::String(now_iso()));
359 64 : Ok::<(), NgsiError>(())
360 88 : })
361 88 : .await;
362 : // 5.11.3.4 after the update: every registration matching now is
363 : // notified, so one the update newly matches gets its copy
364 : // (newlyMatching); one already mapped is left to the PATCH below
365 88 : initial = Some(csr_id);
366 : }
367 : }
368 : // the updated Registration Subscription decides which mapped copies
369 : // still match
370 88 : let csr = match ds_get(st, tenant, own_id)
371 88 : .await
372 88 : .get("csr_sub")
373 88 : .and_then(Value::as_str)
374 : {
375 88 : Some(csr_id) => st
376 88 : .store
377 88 : .get(tenant, Kind::DistSub, csr_id)
378 88 : .await
379 88 : .ok()
380 88 : .flatten(),
381 0 : None => None,
382 : };
383 88 : let csr_ctx = match &csr {
384 88 : Some(c) => Some(crate::notify::sub_context(st, tenant, c).await),
385 0 : None => None,
386 : };
387 88 : let remotes: Vec<(String, (String, String))> = ds_remotes(&ds_get(st, tenant, own_id).await);
388 : // Read before the initial notification runs: a registration it maps now
389 : // is getting its copy by POST, and a PATCH raced against that create
390 : // reaches a remote subscription that may not exist yet.
391 88 : if let Some(csr_id) = initial {
392 88 : let (st2, t2) = (st.clone(), tenant.clone());
393 88 : crate::spawn(async move {
394 56 : crate::notify::csource_initial(&st2, &t2, &csr_id).await;
395 56 : });
396 0 : }
397 88 : for (reg_id, (endpoint, remote_id)) in remotes {
398 8 : let Some(reg) = st
399 8 : .store
400 8 : .get(tenant, Kind::Registration, ®_id)
401 8 : .await
402 8 : .ok()
403 8 : .flatten()
404 : else {
405 0 : continue;
406 : };
407 : // 5.11.2.4 noLongerMatching: the copy at a registration the updated
408 : // Subscription no longer matches is deleted there and unmapped
409 8 : if let (Some(c), Some(ctx)) = (&csr, &csr_ctx) {
410 8 : if !crate::registry::csr_matches_subscription(c, ®, ctx) {
411 4 : let (st2, t2, own) = (st.clone(), tenant.clone(), own_id.to_owned());
412 4 : crate::spawn(async move {
413 4 : on_csource_notification(&st2, &t2, &own, Some("noLongerMatching"), &[reg])
414 4 : .await;
415 4 : });
416 4 : continue;
417 4 : }
418 0 : }
419 4 : let (st2, t2, sub2) = (st.clone(), tenant.clone(), sub.clone());
420 4 : let ctx_url = sub_ctx_url(st, &sub);
421 4 : let own = own_id.to_owned();
422 4 : crate::spawn(async move {
423 4 : let ctx = crate::notify::sub_context(&st2, &t2, &sub2).await;
424 4 : let Some(mut copy) = reduced_copy(&st2, &sub2, ®, &remote_id, &ctx) else {
425 : // nothing this registration covers is watched any more:
426 : // the copy there has nothing left to notify. Decided before
427 : // the updateSubscription gate: a source without it still
428 : // has its copy deleted (and unmapped) here.
429 0 : on_csource_notification(&st2, &t2, &own, Some("noLongerMatching"), &[reg]).await;
430 0 : return;
431 : };
432 4 : if !crate::federation::doc_supports(®, "updateSubscription") {
433 0 : return;
434 4 : }
435 4 : copy.as_object_mut().map(|o| o.remove("id"));
436 4 : forward_sub(
437 4 : &st2,
438 4 : &t2,
439 4 : reqwest::Method::PATCH,
440 4 : format!("{endpoint}/ngsi-ld/v1/subscriptions/{remote_id}"),
441 4 : ®_id,
442 4 : ®,
443 4 : &ctx_url,
444 4 : &sub_via_headers(&sub2),
445 4 : Some(copy),
446 4 : )
447 4 : .await;
448 4 : });
449 : }
450 92 : }
451 :
452 : /// 5.8.5.4: delete the internal CSR subscription (5.11.6) and forward the
453 : /// delete to every mapped remote supporting deleteSubscription.
454 582 : pub(crate) async fn on_subscription_deleted(st: &AppState, tenant: &TenantId, own_id: &str) {
455 582 : let doc = ds_get(st, tenant, own_id).await;
456 582 : let csr_id = doc
457 582 : .get("csr_sub")
458 582 : .and_then(Value::as_str)
459 582 : .map(str::to_owned);
460 582 : let remotes = ds_remotes(&doc);
461 582 : for (_, (_, remote_id)) in &remotes {
462 28 : inbound_delete(st, remote_id).await;
463 : }
464 582 : let _ = st.store.delete(tenant, Kind::DistSub, own_id).await;
465 582 : if let Some(csr_id) = csr_id {
466 574 : let _ = st.store.delete(tenant, Kind::DistSub, &csr_id).await;
467 8 : }
468 582 : for (reg_id, (endpoint, remote_id)) in remotes {
469 28 : let stored = st
470 28 : .store
471 28 : .get(tenant, Kind::Registration, ®_id)
472 28 : .await
473 28 : .ok()
474 28 : .flatten();
475 28 : if stored
476 28 : .as_ref()
477 28 : .is_some_and(|reg| !crate::federation::doc_supports(reg, "deleteSubscription"))
478 : {
479 0 : continue;
480 28 : }
481 28 : let (st2, t2) = (st.clone(), tenant.clone());
482 28 : let ctx_url = sub_ctx_url(st, &Value::Null);
483 28 : crate::spawn(async move {
484 28 : let reg = forward_reg(stored, ®_id, &endpoint);
485 : // the Subscription is already deleted, so its stored chain is
486 : // gone — a delete-forward cannot create a copy, so a fresh
487 : // one-hop chain is loop-safe
488 28 : forward_sub(
489 28 : &st2,
490 28 : &t2,
491 28 : reqwest::Method::DELETE,
492 28 : format!("{endpoint}/ngsi-ld/v1/subscriptions/{remote_id}"),
493 28 : ®_id,
494 28 : ®,
495 28 : &ctx_url,
496 28 : &HeaderMap::new(),
497 28 : None,
498 28 : )
499 28 : .await;
500 24 : });
501 : }
502 582 : }
503 :
504 : /// 4.14 / 5.8.5.4: the document a forwarded delete travels with. The stored
505 : /// Context Source Registration carries the tenant, the contextSourceInfo and
506 : /// the timeout/cooldown the forward needs; the id/endpoint pair is the
507 : /// fallback for a registration that is already gone.
508 36 : fn forward_reg(stored: Option<Value>, reg_id: &str, endpoint: &str) -> Value {
509 36 : stored.unwrap_or_else(|| json!({"id": reg_id, "endpoint": endpoint}))
510 36 : }
511 :
512 : /// 5.8.1.4: "The @context to be used for sending Notifications related to
513 : /// this Subscription shall be the one specified in the jsonldContext field."
514 : /// The forwarded copy carries the subscriber's own terms, so it is shipped
515 : /// under the Subscription's @context, falling back to the core context.
516 160 : fn sub_ctx_url(st: &AppState, sub: &Value) -> String {
517 160 : let source = sub
518 160 : .get("jsonldContext")
519 160 : .or_else(|| sub.get("__context"))
520 160 : .filter(|v| !v.is_null())
521 160 : .cloned()
522 160 : .unwrap_or_else(|| st.loader.core().source.clone());
523 160 : crate::federation::ctx_link_url(&HeaderMap::new(), &source)
524 160 : }
525 :
526 : /// The 5.8.1.4 localOnly=false block: one CSource notification for the
527 : /// internal CSR subscription, dispatched per registration and triggerReason.
528 116 : pub(crate) async fn on_csource_notification(
529 116 : st: &AppState,
530 116 : tenant: &TenantId,
531 116 : own_id: &str,
532 116 : reason: Option<&str>,
533 116 : regs: &[Value],
534 116 : ) {
535 116 : let Some(sub) = st
536 116 : .store
537 116 : .get(tenant, Kind::Subscription, own_id)
538 116 : .await
539 116 : .ok()
540 116 : .flatten()
541 : else {
542 0 : return;
543 : };
544 : // 5.8.1.4: "If localOnly=false, each time a Context Source Notification
545 : // … is received" — a Subscription that has since been flipped to
546 : // localOnly creates no further remote copy.
547 116 : if !distributed(&sub) {
548 4 : return;
549 112 : }
550 112 : let ctx = crate::notify::sub_context(st, tenant, &sub).await;
551 112 : let ctx_url = sub_ctx_url(st, &sub);
552 112 : let via = sub_via_headers(&sub);
553 112 : let seen = crate::federation::via_tokens(&via);
554 112 : let reason = reason.unwrap_or("updated");
555 120 : for reg in regs {
556 120 : let Some(reg_id) = reg.get("id").and_then(Value::as_str) else {
557 0 : continue;
558 : };
559 : // auxiliary registrations take no part (5.8.1.4 lists exclusive,
560 : // redirect and inclusive only)
561 120 : if reg.get("mode").and_then(Value::as_str) == Some("auxiliary") {
562 0 : continue;
563 120 : }
564 : // Table 6.3.18-2: the Via listing "is used when determining matching
565 : // registrations" — a Context Source the Subscription already
566 : // travelled through must not receive a copy of it back.
567 120 : if reg
568 120 : .get("contextSourceAlias")
569 120 : .and_then(Value::as_str)
570 120 : .is_some_and(|a| seen.iter().any(|t| t == a))
571 : {
572 4 : continue;
573 116 : }
574 116 : let Some(endpoint) = reg
575 116 : .get("endpoint")
576 116 : .and_then(Value::as_str)
577 116 : .map(crate::federation::endpoint_base)
578 : else {
579 0 : continue;
580 : };
581 116 : let mapped = ds_remotes(&ds_get(st, tenant, own_id).await)
582 116 : .into_iter()
583 116 : .find(|(r, _)| r == reg_id)
584 116 : .map(|(_, v)| v);
585 116 : match (reason, mapped) {
586 : // an unmapped registration that starts (or keeps) matching gets
587 : // the reduced copy — newlyMatching, or "updated" reported by an
588 : // initial notification when no mapping exists yet
589 100 : ("newlyMatching" | "updated", None)
590 0 : if crate::federation::doc_supports(reg, "createSubscription") =>
591 : {
592 96 : let remote_id =
593 96 : format!("urn:ngsi-ld:Subscription:distsub:{}", uuid::Uuid::new_v4());
594 96 : let Some(copy) = reduced_copy(st, &sub, reg, &remote_id, &ctx) else {
595 : // nothing this registration covers is watched — there is
596 : // no reduced copy to forward to it
597 0 : continue;
598 : };
599 : // 5.8.1.4/5.8.5.4: the mapping is stored BEFORE the forward —
600 : // the remote id is broker-generated, and a delete Subscription
601 : // arriving while the create-forward's response is still in
602 : // flight must find the mapping or the delete-forward is lost
603 : // (the ETSI 5814_01_01 pg race). A failed forward rolls the
604 : // mapping back below.
605 96 : inbound_put(st, &remote_id, tenant, own_id).await;
606 96 : if !ds_set_remote(
607 96 : st,
608 96 : tenant,
609 96 : own_id,
610 96 : reg_id,
611 96 : Some(json!([endpoint.clone(), remote_id])),
612 96 : )
613 96 : .await
614 : {
615 : // the Subscription was deleted while this notification
616 : // was in flight — no remote copy is created for it
617 0 : inbound_delete(st, &remote_id).await;
618 0 : continue;
619 96 : }
620 96 : let (status, _) = forward_sub(
621 96 : st,
622 96 : tenant,
623 96 : reqwest::Method::POST,
624 96 : format!("{endpoint}/ngsi-ld/v1/subscriptions"),
625 96 : reg_id,
626 96 : reg,
627 96 : &ctx_url,
628 96 : &via,
629 96 : Some(copy),
630 : )
631 96 : .await;
632 92 : if !(200..300).contains(&status) {
633 0 : inbound_delete(st, &remote_id).await;
634 0 : ds_set_remote(st, tenant, own_id, reg_id, None).await;
635 92 : }
636 : }
637 16 : ("updated", Some((_, remote_id)))
638 0 : if crate::federation::doc_supports(reg, "updateSubscription") =>
639 : {
640 0 : let Some(mut copy) = reduced_copy(st, &sub, reg, &remote_id, &ctx) else {
641 0 : continue;
642 : };
643 0 : copy.as_object_mut().map(|o| o.remove("id"));
644 0 : forward_sub(
645 0 : st,
646 0 : tenant,
647 0 : reqwest::Method::PATCH,
648 0 : format!("{endpoint}/ngsi-ld/v1/subscriptions/{remote_id}"),
649 0 : reg_id,
650 0 : reg,
651 0 : &ctx_url,
652 0 : &via,
653 0 : Some(copy),
654 0 : )
655 0 : .await;
656 : }
657 16 : ("noLongerMatching", Some((_, remote_id))) => {
658 : // The copy no longer serves this Subscription, whatever the
659 : // source supports: unmapped first, so a notification it
660 : // still sends is not forwarded to the subscriber while the
661 : // delete is in flight, or forever where the source cannot
662 : // delete it.
663 12 : ds_set_remote(st, tenant, own_id, reg_id, None).await;
664 12 : inbound_delete(st, &remote_id).await;
665 : // 5.8.1.4: the delete is forwarded where the registration
666 : // "indicates support for the delete Subscription operation"
667 12 : if crate::federation::doc_supports(reg, "deleteSubscription") {
668 8 : forward_sub(
669 8 : st,
670 8 : tenant,
671 8 : reqwest::Method::DELETE,
672 8 : format!("{endpoint}/ngsi-ld/v1/subscriptions/{remote_id}"),
673 8 : reg_id,
674 8 : reg,
675 8 : &ctx_url,
676 8 : &via,
677 8 : None,
678 8 : )
679 8 : .await;
680 4 : }
681 : }
682 8 : _ => {}
683 : }
684 : }
685 112 : }
686 :
687 : /// 5.8.1.4: "a copy of the original Subscription shall be reduced to what
688 : /// is matched by the registration information"; with splitEntities the
689 : /// q/geoQ/scopeQ members are removed; the notification attributes/pick/omit
690 : /// members are removed; the endpoint is set to the local broker.
691 : ///
692 : /// 5.5.7 Term to URI expansion: the Subscription comes out of the store with
693 : /// its names and types EXPANDED, while a registration delivered by a Context
694 : /// Source Notification carries them COMPACTED. Every registration-derived
695 : /// name is expanded against `ctx` before it is compared or inserted, so the
696 : /// reduction happens in one representation — `expand_key` is idempotent on
697 : /// an absolute IRI, which is what the stored registration already holds.
698 120 : fn reduced_copy(
699 120 : st: &AppState,
700 120 : sub: &Value,
701 120 : reg: &Value,
702 120 : remote_id: &str,
703 120 : ctx: &antares_jsonld::Context,
704 120 : ) -> Option<Value> {
705 120 : let mut copy = sub.clone();
706 120 : let Some(o) = copy.as_object_mut() else {
707 0 : return Some(copy);
708 : };
709 : // __via travels as the Via HTTP header the forward extends (6.3.17),
710 : // never as a body member
711 960 : for k in [
712 120 : "status",
713 120 : "timesSent",
714 120 : "lastNotification",
715 120 : "lastSuccess",
716 120 : "lastFailure",
717 120 : "createdAt",
718 120 : "modifiedAt",
719 120 : "localOnly",
720 960 : ] {
721 960 : o.remove(k);
722 960 : }
723 : // every broker-internal member, not a list of the ones known today:
724 : // __context, __via and the ADR-0020 __subject are this broker's own
725 : // record and none of them may cross to a Context Source
726 858 : o.retain(|k, _| !k.starts_with("__"));
727 120 : o.insert("id".into(), Value::String(remote_id.to_owned()));
728 : // reduce the entity selectors to the registration information
729 120 : let reg_entities: Vec<Value> = reg
730 120 : .get("information")
731 120 : .and_then(Value::as_array)
732 120 : .into_iter()
733 120 : .flatten()
734 120 : .filter_map(|i| i.get("entities").and_then(Value::as_array))
735 120 : .flatten()
736 120 : .map(|e| {
737 120 : let mut e = e.clone();
738 : // notification presentation arrayifies type — EntitySelector
739 : // wants the plain form back
740 120 : if let Some(o) = e.as_object_mut() {
741 120 : if let Some(t) = o
742 120 : .get("type")
743 120 : .and_then(Value::as_array)
744 120 : .and_then(|a| a.first())
745 120 : .cloned()
746 116 : {
747 116 : o.insert("type".into(), t);
748 116 : }
749 120 : if let Some(t) = o.get("type").and_then(Value::as_str) {
750 120 : let t = Value::String(ctx.expand_key(t));
751 120 : o.insert("type".into(), t);
752 120 : }
753 0 : }
754 120 : e
755 120 : })
756 120 : .collect();
757 120 : if !reg_entities.is_empty() {
758 120 : o.insert("entities".into(), Value::Array(reg_entities));
759 120 : }
760 : // watchedAttributes ∩ the registration's attribute scope
761 120 : let reg_attrs: Vec<String> = reg
762 120 : .get("information")
763 120 : .and_then(Value::as_array)
764 120 : .into_iter()
765 120 : .flatten()
766 120 : .flat_map(|i| {
767 120 : antares_model::registration_attribute_names(i)
768 120 : .unwrap_or_default()
769 120 : .into_iter()
770 120 : .map(|n| ctx.expand_key(n))
771 120 : })
772 120 : .collect();
773 : // a periodic Subscription watches nothing (5.2.12 forbids the pair):
774 : // its copy gets no watchedAttributes either, or the source refuses it
775 120 : let periodic = o.contains_key("timeInterval");
776 120 : if !reg_attrs.is_empty() && !periodic {
777 16 : let watch: Vec<Value> = match o.get("watchedAttributes").and_then(Value::as_array) {
778 12 : Some(w) => w
779 12 : .iter()
780 24 : .filter(|a| a.as_str().is_some_and(|a| reg_attrs.iter().any(|r| r == a)))
781 12 : .cloned()
782 12 : .collect(),
783 : // 5.8.1.4 "reduced to what is matched by the registration
784 : // information": a watch-everything Subscription still only
785 : // watches the registered names at the source — otherwise an
786 : // unregistered attribute change notifies through the chain.
787 4 : None => reg_attrs.iter().map(|a| Value::String(a.clone())).collect(),
788 : };
789 : // Nothing the subscriber watches is matched by this registration
790 : // information: there is no reduced copy to forward — and an empty
791 : // watchedAttributes is a payload 5.2.12 forbids.
792 16 : if watch.is_empty() {
793 0 : return None;
794 16 : }
795 16 : o.insert("watchedAttributes".into(), Value::Array(watch));
796 104 : }
797 : // 5.8.1.4: with splitEntities the remote sees only fragments — the
798 : // q/geoQ/scopeQ conditions are evaluated LOCALLY after the 5.8.6 merge
799 : // (splitEntities is a Subscription member, 5.2.12)
800 120 : if sub.get("splitEntities").and_then(Value::as_bool) == Some(true) {
801 60 : for k in ["q", "geoQ", "scopeQ"] {
802 60 : o.remove(k);
803 60 : }
804 100 : }
805 120 : if let Some(n) = o.get_mut("notification").and_then(Value::as_object_mut) {
806 360 : for k in ["attributes", "pick", "omit"] {
807 360 : n.remove(k);
808 360 : }
809 : // Table 5.2.14.2-1: the delivery bookkeeping is this broker's own
810 : // record of notifying ITS subscriber. It lives inside `notification`,
811 : // it is output-only on a create ("implementations shall ignore
812 : // them"), and it says nothing the Context Source may act on — so it
813 : // stays here, like the top-level members stripped above.
814 720 : for k in [
815 120 : "status",
816 120 : "timesSent",
817 120 : "timesFailed",
818 120 : "lastNotification",
819 120 : "lastSuccess",
820 120 : "lastFailure",
821 720 : ] {
822 720 : n.remove(k);
823 720 : }
824 120 : n.insert(
825 120 : "endpoint".into(),
826 120 : json!({"uri": format!("{}/ex/v1/remote-notify", st.public_url)}),
827 : );
828 0 : }
829 120 : Some(copy)
830 120 : }
831 :
832 : /// One forwarded subscription operation, through the shared federation
833 : /// forward (egress policy, Via, contextSourceInfo, tenant mapping).
834 : /// `via` is the chain the Subscription arrived with ([`sub_via_headers`]);
835 : /// the forward appends this broker's alias to it (6.3.17), so downstream
836 : /// brokers see the full path and can cut a loop.
837 : #[allow(clippy::too_many_arguments)] // mirrors the wire: one param per forwarded request part
838 136 : async fn forward_sub(
839 136 : st: &AppState,
840 136 : tenant: &TenantId,
841 136 : method: reqwest::Method,
842 136 : url: String,
843 136 : reg_id: &str,
844 136 : reg: &Value,
845 136 : ctx_url: &str,
846 136 : via: &HeaderMap,
847 136 : body: Option<Value>,
848 136 : ) -> (u16, Value) {
849 136 : let fed = crate::federation::fed_reg_of(reg_id, reg);
850 128 : let (status, body, _) =
851 136 : crate::federation::forward(st, method, url, &[], via, tenant, &fed, ctx_url, body).await;
852 128 : (status, body)
853 128 : }
854 :
855 : /// 5.8.6 splitEntities=true inbound merge: each notified Entity "shall be
856 : /// retrieved locally and from all Context Sources that have information
857 : /// about these Entities, except for the one from which the Notification has
858 : /// been received", merged with the notified fragment, and "all Entities
859 : /// that do not match the query, geoquery and Scope query conditions of the
860 : /// local Subscription shall be removed".
861 20 : async fn split_merge(
862 20 : st: &AppState,
863 20 : tenant: &TenantId,
864 20 : sub: &Value,
865 20 : origin_reg: Option<&str>,
866 20 : data: Vec<Value>,
867 20 : ) -> Result<Vec<Value>, NgsiError> {
868 20 : let ctx = crate::notify::sub_context(st, tenant, sub).await;
869 20 : let headers = HeaderMap::new();
870 20 : let mut out = Vec::new();
871 24 : for ent in data {
872 24 : let Some(obj) = ent.as_object() else { continue };
873 : // inbound notification presentation → the expanded storage form.
874 : // 5.5.4 bans "urn:ngsi-ld:null" as a first level member value "with
875 : // the exception of NGSI-LD Fragments … or to represent deleted
876 : // Properties in concise representation as part of notifications", so
877 : // a notified deletion (4.5.7) is a valid payload here and a
878 : // validating expansion would drop the whole Entity. `sys` keeps the
879 : // timestamps a sysAttrs subscriber asked for; this is a translation
880 : // of a document the Context Source already produced, not an
881 : // admission check on client input.
882 24 : let Ok(mut merged) = antares_jsonld::expand_entity(
883 24 : obj,
884 24 : &ctx,
885 24 : antares_jsonld::ExpandOpts {
886 24 : allow_null: true,
887 24 : sys: true,
888 24 : ..Default::default()
889 24 : },
890 24 : ) else {
891 0 : continue;
892 : };
893 24 : let Some(id) = merged.get("id").and_then(Value::as_str).map(str::to_owned) else {
894 0 : continue;
895 : };
896 24 : if let Ok(Some(local)) = st.store.get(tenant, Kind::Entity, &id).await {
897 8 : crate::federation::merge_docs(&mut merged, &local, false);
898 16 : }
899 24 : let mut warnings = Vec::new();
900 24 : let fed = crate::federation::fed_retrieve(
901 24 : st,
902 24 : tenant,
903 24 : &headers,
904 24 : &ctx,
905 24 : &id,
906 24 : None,
907 24 : origin_reg,
908 24 : &mut warnings,
909 24 : &mut None,
910 24 : )
911 24 : .await?;
912 48 : for aux_pass in [false, true] {
913 48 : for (aux, doc) in &fed {
914 16 : if *aux == aux_pass {
915 8 : crate::federation::merge_docs(&mut merged, doc, *aux);
916 8 : }
917 : }
918 : }
919 24 : if crate::notify::linked_eval(st, tenant, None, |l| {
920 24 : crate::notify::conditions_match(sub, &merged, &ctx, l)
921 24 : })
922 24 : .await
923 16 : {
924 16 : // 5.3.1/5.8.6: notification data carries Entities in their
925 16 : // API representation — shape and compact the merged storage
926 16 : // form exactly like the local notify path.
927 16 : let shape = crate::notify::notif_shape(sub, &ctx);
928 16 : let shaped = crate::repr::apply(&merged, &shape.repr);
929 16 : out.push(crate::repr::compact_for(&shape.repr, &shaped, &ctx));
930 16 : }
931 : }
932 20 : Ok(out)
933 20 : }
934 :
935 : /// POST /ex/v1/remote-notify — the local broker's endpoint for
936 : /// notifications from forwarded subscription copies. 5.8.1.4: "the mapping
937 : /// of the received subscriptionId with the own Subscription identifier …
938 : /// to enable forwarding received notifications to the original subscriber."
939 86 : pub async fn remote_notify(State(st): State<AppState>, body: Bytes) -> Response {
940 86 : match remote_notify_inner(&st, &body).await {
941 60 : Ok(r) => r,
942 26 : Err(e) => e.into_response(),
943 : }
944 86 : }
945 :
946 86 : async fn remote_notify_inner(st: &AppState, body: &[u8]) -> ApiResult<Response> {
947 86 : let v: Value = serde_json::from_slice(body)
948 86 : .map_err(|e| NgsiError::InvalidRequest(format!("body is not valid JSON: {e}")))?;
949 : // Peer-facing entry point: the Entity count is capped before any store
950 : // touch or per-Entity work, under the same ceiling a client batch gets
951 : // (ANTARES_MAX_BATCH_ITEMS). One notification drives one local retrieve
952 : // and one federated fan-out per Entity in the 5.8.6 merge, so an
953 : // uncapped data array is an amplification lever.
954 82 : let cap = *crate::bounds::MAX_BATCH_ITEMS;
955 82 : if v.get("data")
956 82 : .and_then(Value::as_array)
957 82 : .is_some_and(|a| a.len() > cap)
958 : {
959 4 : return Err(NgsiError::BadRequestData(format!(
960 4 : "notification data carries more than {cap} Entities"
961 4 : ))
962 4 : .into());
963 78 : }
964 78 : let sid = v
965 78 : .get("subscriptionId")
966 78 : .and_then(Value::as_str)
967 78 : .ok_or_else(|| NgsiError::BadRequestData("notification without subscriptionId".into()))?;
968 78 : let Some((tenant, own_id)) = inbound_get(st, sid).await else {
969 14 : return Err(ApiError::from(NgsiError::ResourceNotFound(format!(
970 14 : "no distributed subscription maps {sid}"
971 14 : ))));
972 : };
973 : // read back from the broker's own inbound index, not from the request
974 64 : let tenant = TenantId::new_internal(&tenant)
975 64 : .map_err(|_| NgsiError::InternalError("stored tenant invalid".into()))?;
976 64 : let Some(sub) = st
977 64 : .store
978 64 : .get(&tenant, Kind::Subscription, &own_id)
979 64 : .await
980 64 : .ok()
981 64 : .flatten()
982 : else {
983 : // the subscriber is gone: prune the mapping on touch so a remote
984 : // that keeps notifying cannot pin a dead index entry forever, and
985 : // answer about the peer's own id — the local Subscription id is not
986 : // the peer's to learn
987 4 : inbound_delete(st, sid).await;
988 4 : return Err(ApiError::from(NgsiError::ResourceNotFound(format!(
989 4 : "no distributed subscription maps {sid}"
990 4 : ))));
991 : };
992 : // the origin of this notification is the registration its remote
993 : // subscription was created at
994 60 : let origin_reg_id = ds_remotes(&ds_get(st, &tenant, &own_id).await)
995 60 : .into_iter()
996 60 : .find(|(_, (_, rid))| rid.as_str() == sid)
997 60 : .map(|(reg_id, _)| reg_id);
998 : // 5.8.6: "if a Context Source filter is defined, then only the
999 : // subscribed Entities whose origin Context Source matches the referred
1000 : // filter shall be included".
1001 60 : if let Some(csf) = sub.get("csf").and_then(Value::as_str) {
1002 8 : if let Ok(ast) = antares_ql::parse_q(csf) {
1003 8 : let origin_reg = match origin_reg_id.as_ref() {
1004 8 : Some(reg_id) => st
1005 8 : .store
1006 8 : .get(&tenant, Kind::Registration, reg_id)
1007 8 : .await
1008 8 : .ok()
1009 8 : .flatten(),
1010 0 : None => None,
1011 : };
1012 : // 5.8.1.4: the csf is written in the Subscription's own @context,
1013 : // and 5.11.2.4 already read it in that @context when it chose the
1014 : // sources this Subscription was forwarded to. Reading it in the
1015 : // core context here would let the two disagree, and a source the
1016 : // broker deliberately subscribed to would have every notification
1017 : // dropped.
1018 8 : let ctx = crate::notify::sub_context(st, &tenant, &sub).await;
1019 8 : let matches =
1020 8 : origin_reg.is_some_and(|reg| crate::registry::csf_matches(&ast, ®, &ctx));
1021 8 : if !matches {
1022 : // origin gated out — acknowledged, nothing forwarded
1023 0 : return Ok(StatusCode::OK.into_response());
1024 8 : }
1025 0 : }
1026 52 : }
1027 60 : let mut data: Vec<Value> = v
1028 60 : .get("data")
1029 60 : .and_then(Value::as_array)
1030 60 : .cloned()
1031 60 : .unwrap_or_default();
1032 : // 5.2.33 / 5.8.1.4: the remote copy carries the REGISTRATION's entity
1033 : // scope, which may be broader than the original Subscription's own
1034 : // entities selector — re-filter inbound entities against the original
1035 : // selector (id over idPattern precedence included) before forwarding.
1036 60 : let sel_ctx = crate::notify::sub_context(st, &tenant, &sub).await;
1037 68 : data.retain(|e| {
1038 68 : let id = e.get("id").and_then(Value::as_str).unwrap_or("");
1039 68 : let types: Vec<Value> = match e.get("type") {
1040 68 : Some(Value::String(t)) => vec![Value::String(sel_ctx.expand_key(t))],
1041 0 : Some(Value::Array(a)) => a
1042 0 : .iter()
1043 0 : .filter_map(Value::as_str)
1044 0 : .map(|t| Value::String(sel_ctx.expand_key(t)))
1045 0 : .collect(),
1046 0 : _ => Vec::new(),
1047 : };
1048 68 : let shim = serde_json::json!({"id": id, "type": types});
1049 68 : crate::notify::selector_match(&sub, &shim, &sel_ctx)
1050 68 : });
1051 60 : if data.is_empty() {
1052 : // acknowledged; nothing the original Subscription selected
1053 4 : return Ok(StatusCode::OK.into_response());
1054 56 : }
1055 : // 5.8.6 splitEntities=true: the notified Entities are fragments —
1056 : // retrieve them locally and from all other Context Sources (except the
1057 : // origin), merge, and re-filter by the local Subscription's conditions.
1058 56 : if sub.get("splitEntities").and_then(Value::as_bool) == Some(true) {
1059 20 : data = split_merge(st, &tenant, &sub, origin_reg_id.as_deref(), data).await?;
1060 20 : if data.is_empty() {
1061 : // "If there are Entities in the data member of the Notification
1062 : // copy, the Notification copy shall be forwarded" — none left
1063 4 : return Ok(StatusCode::OK.into_response());
1064 16 : }
1065 36 : }
1066 : // 5.8.6: forward to the original subscriber under the OWN subscriptionId
1067 52 : let split = sub.get("splitEntities").and_then(Value::as_bool) == Some(true);
1068 52 : let (st2, sub2, t2) = (st.clone(), sub.clone(), tenant.clone());
1069 52 : crate::spawn(async move {
1070 52 : let ctx = crate::notify::sub_context(&st2, &t2, &sub2).await;
1071 : // split_merge shaped its Entities already
1072 52 : let data = if split {
1073 16 : data
1074 : } else {
1075 36 : project_forwarded(&sub2, data, &ctx)
1076 : };
1077 52 : crate::notify::deliver(&st2, &t2, &sub2, data, &ctx).await;
1078 50 : });
1079 52 : Ok(StatusCode::OK.into_response())
1080 86 : }
1081 :
1082 : /// The local Subscription's projection (attributes, pick, omit, format)
1083 : /// over Entities a Context Source notified: the source answers
1084 : /// its own copy's projection, which need not be the subscriber's. Each is
1085 : /// read back into the storage form, shaped and compacted as the local
1086 : /// notify path does; one that does not read is dropped.
1087 40 : fn project_forwarded(sub: &Value, data: Vec<Value>, ctx: &antares_jsonld::Context) -> Vec<Value> {
1088 40 : let shape = crate::notify::notif_shape(sub, ctx);
1089 40 : data.iter()
1090 40 : .filter_map(Value::as_object)
1091 40 : .filter_map(|obj| {
1092 40 : antares_jsonld::expand_entity(
1093 40 : obj,
1094 40 : ctx,
1095 40 : antares_jsonld::ExpandOpts {
1096 40 : allow_null: true,
1097 40 : sys: true,
1098 40 : ..Default::default()
1099 40 : },
1100 : )
1101 40 : .ok()
1102 40 : })
1103 40 : .map(|doc| {
1104 40 : let shaped = crate::repr::apply(&doc, &shape.repr);
1105 40 : crate::repr::compact_for(&shape.repr, &shaped, ctx)
1106 40 : })
1107 40 : .collect()
1108 40 : }
1109 :
1110 : #[cfg(test)]
1111 : mod tests {
1112 : use super::*;
1113 :
1114 : /// a forwarded notification carries what the local
1115 : /// Subscription projects, not what the source sent.
1116 : #[test]
1117 4 : fn a_forwarded_notification_takes_the_local_projection() {
1118 4 : let ctx = antares_jsonld::core_context();
1119 4 : let sub = json!({"notification": {"attributes": [format!("{DC}speed")]}});
1120 4 : let data = vec![json!({"id": "urn:ngsi-ld:Vehicle:1", "type": "Vehicle",
1121 4 : "speed": {"type": "Property", "value": 1},
1122 4 : "brand": {"type": "Property", "value": "x"}})];
1123 4 : let out = project_forwarded(&sub, data, &ctx);
1124 4 : assert_eq!(out.len(), 1);
1125 4 : assert_eq!(out[0]["speed"]["value"], 1, "{}", out[0]);
1126 4 : assert!(out[0].get("brand").is_none(), "{}", out[0]);
1127 4 : }
1128 :
1129 : /// The default @context prefix every Term expands to (5.5.7).
1130 : const DC: &str = "https://uri.etsi.org/ngsi-ld/default-context/";
1131 :
1132 : /// A registration exactly as `on_csource_notification` receives it: out
1133 : /// of a Context Source Notification, whose names and types
1134 : /// `present_registration` has COMPACTED.
1135 28 : fn reg_doc() -> Value {
1136 28 : json!({
1137 28 : "id": "urn:ngsi-ld:ContextSourceRegistration:r1",
1138 28 : "endpoint": "http://source.example.org",
1139 28 : "operations": ["createSubscription", "updateSubscription", "deleteSubscription"],
1140 28 : "information": [{
1141 28 : "entities": [{"type": ["Vehicle"], "id": "urn:ngsi-ld:Vehicle:1"}],
1142 28 : "propertyNames": ["speed"],
1143 : }],
1144 : })
1145 28 : }
1146 :
1147 : /// A Subscription exactly as the store holds it: names and types
1148 : /// EXPANDED by `normalize_subscription` (5.5.7).
1149 40 : fn sub_doc() -> Value {
1150 40 : json!({
1151 40 : "id": "urn:ngsi-ld:Subscription:own",
1152 40 : "type": "Subscription",
1153 40 : "entities": [{"type": format!("{DC}Vehicle")}, {"type": format!("{DC}Device")}],
1154 40 : "watchedAttributes": [format!("{DC}speed"), format!("{DC}brand")],
1155 40 : "q": "speed>10",
1156 40 : "geoQ": {"georel": "near;maxDistance==1000"},
1157 40 : "scopeQ": "/A",
1158 40 : "localOnly": false,
1159 40 : "status": "active",
1160 40 : "timesSent": 7,
1161 40 : "createdAt": "2026-01-01T00:00:00Z",
1162 40 : "modifiedAt": "2026-01-02T00:00:00Z",
1163 40 : "__context": "http://example.org/ctx.jsonld",
1164 40 : "__via": "1.1 upstream-broker",
1165 : // Table 5.2.14.2-1 keeps the delivery bookkeeping HERE, not at
1166 : // the top level: this is the shape `record_delivery` leaves
1167 : // behind, and the shape a forward is built from.
1168 40 : "notification": {
1169 40 : "attributes": [format!("{DC}speed")],
1170 40 : "pick": ["speed"],
1171 40 : "omit": ["brand"],
1172 40 : "endpoint": {"uri": "http://subscriber.example.org/cb"},
1173 40 : "status": "ok",
1174 40 : "timesSent": 7,
1175 40 : "timesFailed": 2,
1176 40 : "lastNotification": "2026-01-03T00:00:00Z",
1177 40 : "lastSuccess": "2026-01-03T00:00:00Z",
1178 40 : "lastFailure": "2026-01-02T00:00:00Z",
1179 : },
1180 : })
1181 40 : }
1182 :
1183 : /// a periodic Subscription that watches everything gets no
1184 : /// watchedAttributes in its copy: 5.2.12 forbids the pair with
1185 : /// timeInterval, so the source refused every such copy.
1186 : #[test]
1187 4 : fn a_periodic_copy_watches_nothing() {
1188 4 : let st = AppState::new("antares-ds-periodic".into());
1189 4 : let mut sub = sub_doc();
1190 4 : if let Some(o) = sub.as_object_mut() {
1191 4 : o.remove("watchedAttributes");
1192 4 : o.insert("timeInterval".into(), json!(10));
1193 4 : }
1194 4 : let copy = reduced_copy(
1195 4 : &st,
1196 4 : &sub,
1197 4 : ®_doc(),
1198 4 : "urn:ngsi-ld:Subscription:remote1",
1199 4 : &st.loader.core(),
1200 : )
1201 4 : .expect("a copy the registration covers");
1202 4 : assert_eq!(copy.get("watchedAttributes"), None, "{copy}");
1203 4 : assert_eq!(copy.get("timeInterval"), Some(&json!(10)));
1204 4 : }
1205 :
1206 : /// 5.8.1.4: "a copy of the original Subscription shall be reduced to
1207 : /// what is matched by the registration information … Also from the
1208 : /// notification member, the attributes, pick and omit members are to be
1209 : /// removed. The copied Subscription is then forwarded to the Context
1210 : /// Source as a new Subscription where the notification endpoint is set
1211 : /// to that of the local Broker." Nothing outside the registration's
1212 : /// scope, and no local bookkeeping, may travel with the copy.
1213 : #[test]
1214 4 : fn clause_5_8_1_reduced_copy_carries_only_the_registration_scope() {
1215 4 : let st = AppState::new("antares-ds-reduce".into());
1216 4 : let copy = reduced_copy(
1217 4 : &st,
1218 4 : &sub_doc(),
1219 4 : ®_doc(),
1220 4 : "urn:ngsi-ld:Subscription:remote1",
1221 4 : &st.loader.core(),
1222 : )
1223 4 : .expect("a copy the registration covers");
1224 4 : assert_eq!(copy["id"], json!("urn:ngsi-ld:Subscription:remote1"));
1225 4 : assert_ne!(
1226 4 : copy["id"],
1227 4 : json!("urn:ngsi-ld:Subscription:own"),
1228 : "the remote must be told the broker-generated id, not the local one"
1229 : );
1230 4 : let ents = copy["entities"].as_array().expect("entities");
1231 4 : assert_eq!(ents.len(), 1, "{copy}");
1232 4 : assert_eq!(ents[0]["type"], json!(format!("{DC}Vehicle")));
1233 4 : assert!(
1234 4 : !copy.to_string().contains(&format!("{DC}Device")),
1235 : "a selector the registration does not cover must NOT be forwarded: {copy}"
1236 : );
1237 4 : assert_eq!(
1238 4 : copy["watchedAttributes"],
1239 4 : json!([format!("{DC}speed")]),
1240 : "watchedAttributes are intersected with the registered names"
1241 : );
1242 : // local-only bookkeeping never leaves the broker — the Via chain
1243 : // travels as the HTTP header the forward extends, never in the body
1244 40 : for k in [
1245 4 : "status",
1246 4 : "timesSent",
1247 4 : "lastNotification",
1248 4 : "lastSuccess",
1249 4 : "lastFailure",
1250 4 : "createdAt",
1251 4 : "modifiedAt",
1252 4 : "__context",
1253 4 : "__via",
1254 4 : "localOnly",
1255 4 : ] {
1256 40 : assert!(copy.get(k).is_none(), "{k} must not be forwarded: {copy}");
1257 : }
1258 : // Table 5.2.14.2-1: the same bookkeeping the store keeps under
1259 : // `notification` is this broker's alone — it says how often IT has
1260 : // notified ITS subscriber, is output-only on a create, and has no
1261 : // meaning to the Context Source receiving the copy.
1262 24 : for k in [
1263 4 : "status",
1264 4 : "timesSent",
1265 4 : "timesFailed",
1266 4 : "lastNotification",
1267 4 : "lastSuccess",
1268 4 : "lastFailure",
1269 4 : ] {
1270 24 : assert!(
1271 24 : copy["notification"].get(k).is_none(),
1272 : "notification.{k} must not be forwarded: {copy}"
1273 : );
1274 : }
1275 4 : let n = ©["notification"];
1276 12 : for k in ["attributes", "pick", "omit"] {
1277 12 : assert!(
1278 12 : n.get(k).is_none(),
1279 : "notification.{k} must be removed: {copy}"
1280 : );
1281 : }
1282 4 : let uri = n["endpoint"]["uri"].as_str().expect("endpoint uri");
1283 4 : assert!(uri.ends_with("/ex/v1/remote-notify"), "{uri}");
1284 4 : assert_ne!(
1285 : uri, "http://subscriber.example.org/cb",
1286 : "the source must never learn the original subscriber's endpoint"
1287 : );
1288 : // splitEntities is absent here, so the filters stay on the copy
1289 4 : assert_eq!(copy["q"], json!("speed>10"));
1290 4 : }
1291 :
1292 : /// 5.8.1.4: "If the splitEntities member is explicitly set to true …
1293 : /// the members q, geoQ and scopeQ shall be removed from the created
1294 : /// copy"; and a Subscription with no watchedAttributes is still reduced
1295 : /// to the registered names, so an unregistered Attribute change cannot
1296 : /// notify through the chain.
1297 : #[test]
1298 4 : fn clause_5_8_1_reduced_copy_split_and_watch_everything() {
1299 4 : let st = AppState::new("antares-ds-split".into());
1300 4 : let mut sub = sub_doc();
1301 4 : sub["splitEntities"] = json!(true);
1302 4 : let copy = reduced_copy(
1303 4 : &st,
1304 4 : &sub,
1305 4 : ®_doc(),
1306 4 : "urn:ngsi-ld:Subscription:remote2",
1307 4 : &st.loader.core(),
1308 : )
1309 4 : .expect("copy");
1310 12 : for k in ["q", "geoQ", "scopeQ"] {
1311 12 : assert!(
1312 12 : copy.get(k).is_none(),
1313 : "{k} is evaluated locally after the 5.8.6 merge: {copy}"
1314 : );
1315 : }
1316 4 : let mut watch_all = sub_doc();
1317 4 : watch_all
1318 4 : .as_object_mut()
1319 4 : .expect("object")
1320 4 : .remove("watchedAttributes");
1321 4 : let copy = reduced_copy(
1322 4 : &st,
1323 4 : &watch_all,
1324 4 : ®_doc(),
1325 4 : "urn:ngsi-ld:Subscription:remote3",
1326 4 : &st.loader.core(),
1327 : )
1328 4 : .expect("copy");
1329 4 : assert_eq!(
1330 4 : copy["watchedAttributes"],
1331 4 : json!([format!("{DC}speed")]),
1332 : "a watch-everything Subscription still only watches the \
1333 : registered names at the source"
1334 : );
1335 4 : }
1336 :
1337 : /// 5.8.1.4 "reduced to what is matched by the registration information",
1338 : /// with 5.5.7 Term to URI expansion: the registration arrives through a
1339 : /// Context Source Notification (names COMPACTED), the Subscription comes
1340 : /// out of the store (names EXPANDED). The intersection must be taken in
1341 : /// ONE representation — an empty `watchedAttributes` is a payload 5.2.12
1342 : /// forbids, and it narrows the forwarded copy to nothing.
1343 : #[test]
1344 4 : fn clause_5_8_1_reduced_copy_intersects_across_representations() {
1345 4 : let st = AppState::new("antares-ds-expand".into());
1346 4 : let copy = reduced_copy(
1347 4 : &st,
1348 4 : &sub_doc(),
1349 4 : ®_doc(),
1350 4 : "urn:ngsi-ld:Subscription:remote4",
1351 4 : &st.loader.core(),
1352 : )
1353 4 : .expect("a copy the registration covers");
1354 4 : assert_eq!(
1355 4 : copy["watchedAttributes"],
1356 4 : json!([format!("{DC}speed")]),
1357 : "the compacted registered name must be expanded before the \
1358 : intersection: {copy}"
1359 : );
1360 4 : assert_ne!(
1361 4 : copy["watchedAttributes"],
1362 4 : json!([]),
1363 : "5.2.12: watchedAttributes, when present, is a non-empty array"
1364 : );
1365 4 : assert!(
1366 4 : !copy.to_string().contains(&format!("{DC}brand")),
1367 : "an unregistered watched name must not be forwarded: {copy}"
1368 : );
1369 4 : assert_eq!(
1370 4 : copy["entities"],
1371 4 : json!([{"type": format!("{DC}Vehicle"), "id": "urn:ngsi-ld:Vehicle:1"}]),
1372 : "the registration's selector travels in the Subscription's own \
1373 : representation: {copy}"
1374 : );
1375 4 : }
1376 :
1377 : /// 5.8.1.4: "Based on the content of the Subscription, a Context Source
1378 : /// Registration Subscription shall be created (clause 5.11.2)" — 5.11.2.4
1379 : /// matches registrations on csf, geoQ, scopeQ, temporalQ and the
1380 : /// notification attributes as well, so a Registration Subscription built
1381 : /// from entities and watchedAttributes alone offers the Subscription to
1382 : /// sources the subscriber excluded.
1383 : #[tokio::test]
1384 4 : async fn clause_5_8_1_csr_subscription_carries_every_matching_member() {
1385 4 : let st = AppState::new("antares-ds-csr".into());
1386 4 : let t = TenantId::default();
1387 4 : let mut sub = sub_doc();
1388 4 : sub["csf"] = json!("name==\"SourceA\"");
1389 4 : sub["temporalQ"] = json!({"timerel": "before", "timeAt": "2026-01-01T00:00:00Z"});
1390 4 : let own = sub["id"].as_str().expect("id").to_owned();
1391 4 : st.store
1392 4 : .create(&t, Kind::Subscription, &own, sub.clone())
1393 4 : .await
1394 4 : .expect("create");
1395 4 : on_subscription_created(&st, &t, &sub).await;
1396 4 : let csr_id = ds_get(&st, &t, &own).await["csr_sub"]
1397 4 : .as_str()
1398 4 : .expect("csr_sub")
1399 4 : .to_owned();
1400 4 : let csr = st
1401 4 : .store
1402 4 : .get(&t, Kind::DistSub, &csr_id)
1403 4 : .await
1404 4 : .ok()
1405 4 : .flatten()
1406 4 : .expect("csr subscription");
1407 24 : for k in [
1408 4 : "entities",
1409 4 : "watchedAttributes",
1410 4 : "csf",
1411 4 : "geoQ",
1412 4 : "scopeQ",
1413 4 : "temporalQ",
1414 4 : ] {
1415 24 : assert!(
1416 24 : csr.get(k).is_some(),
1417 : "{k} decides which registrations match: {csr}"
1418 : );
1419 : }
1420 4 : assert_eq!(
1421 4 : csr["notification"]["attributes"],
1422 4 : json!([format!("{DC}speed")]),
1423 : "5.11.2.4 unions notification.attributes into the match spec: {csr}"
1424 : );
1425 4 : assert!(
1426 4 : csr.get("q").is_none(),
1427 4 : "q filters Entity Attributes, not registration properties: {csr}"
1428 4 : );
1429 4 : }
1430 :
1431 : /// 6.3.18: the Via header exists "to avoid infinite loops". A forwarded
1432 : /// Subscription copy (5.8.1.4) arrives with the Via chain of the brokers
1433 : /// it has already passed through; a chain that names THIS broker means
1434 : /// the copy has looped back, so the Subscription serves locally and the
1435 : /// distributed half is NOT created — otherwise two mutually registered
1436 : /// brokers re-forward copies of copies without bound.
1437 : #[tokio::test]
1438 4 : async fn clause_6_3_18_looping_via_chain_suppresses_the_distributed_half() {
1439 4 : let st = AppState::new("antares-ds-viahost".into());
1440 4 : let t = TenantId::default();
1441 : // the copy's chain already names this broker's own alias
1442 4 : let mut looped = sub_doc();
1443 4 : looped["__via"] = json!("1.1 sourceX, 1.1 antares-ds-viahost");
1444 4 : let own = looped["id"].as_str().expect("id").to_owned();
1445 4 : st.store
1446 4 : .create(&t, Kind::Subscription, &own, looped.clone())
1447 4 : .await
1448 4 : .expect("create");
1449 4 : on_subscription_created(&st, &t, &looped).await;
1450 4 : assert!(
1451 4 : ds_get(&st, &t, &own).await.get("csr_sub").is_none(),
1452 : "a looped copy must not create the internal Registration Subscription"
1453 : );
1454 : // positive control: a chain naming only OTHER brokers is not a loop
1455 4 : let mut chained = sub_doc();
1456 4 : chained["id"] = json!("urn:ngsi-ld:Subscription:chained");
1457 4 : chained["__via"] = json!("1.1 sourceX");
1458 4 : st.store
1459 4 : .create(
1460 4 : &t,
1461 4 : Kind::Subscription,
1462 4 : "urn:ngsi-ld:Subscription:chained",
1463 4 : chained.clone(),
1464 4 : )
1465 4 : .await
1466 4 : .expect("create");
1467 4 : on_subscription_created(&st, &t, &chained).await;
1468 4 : assert!(
1469 4 : ds_get(&st, &t, "urn:ngsi-ld:Subscription:chained")
1470 4 : .await
1471 4 : .get("csr_sub")
1472 4 : .is_some(),
1473 4 : "a pass-through chain (A->B->C) must keep the distributed half"
1474 4 : );
1475 4 : }
1476 :
1477 : /// 5.8.1.4 gates the whole distributed block on "If localOnly=false": a
1478 : /// Subscription updated to localOnly=true forwards no further copy, and
1479 : /// the internal Context Source Registration Subscription plus the
1480 : /// mappings it already holds are torn down.
1481 : #[tokio::test]
1482 4 : async fn clause_5_8_1_local_only_flip_tears_the_distributed_half_down() {
1483 4 : let st = AppState::new("antares-ds-local".into());
1484 4 : let t = TenantId::default();
1485 4 : let mut sub = sub_doc();
1486 4 : sub["localOnly"] = json!(true);
1487 4 : let own = sub["id"].as_str().expect("id").to_owned();
1488 4 : let csr_id = "urn:ngsi-ld:CSourceSubscription:distsub:x";
1489 4 : st.store
1490 4 : .create(&t, Kind::Subscription, &own, sub.clone())
1491 4 : .await
1492 4 : .expect("create");
1493 4 : st.store
1494 4 : .create(&t, Kind::DistSub, csr_id, json!({"id": csr_id}))
1495 4 : .await
1496 4 : .expect("create");
1497 4 : ds_put(&st, &t, &own, json!({"csr_sub": csr_id})).await;
1498 : // a Context Source Notification arriving after the flip creates
1499 : // nothing at the source
1500 4 : on_csource_notification(&st, &t, &own, Some("newlyMatching"), &[reg_doc()]).await;
1501 4 : assert!(
1502 4 : ds_remotes(&ds_get(&st, &t, &own).await).is_empty(),
1503 : "a local-only Subscription forwards no copy"
1504 : );
1505 4 : on_subscription_updated(&st, &t, &own).await;
1506 4 : assert!(
1507 4 : st.store
1508 4 : .get(&t, Kind::DistSub, csr_id)
1509 4 : .await
1510 4 : .ok()
1511 4 : .flatten()
1512 4 : .is_none(),
1513 : "the internal Registration Subscription must not survive the flip"
1514 : );
1515 4 : assert!(
1516 4 : st.store
1517 4 : .get(&t, Kind::DistSub, &own)
1518 4 : .await
1519 4 : .ok()
1520 4 : .flatten()
1521 4 : .is_none(),
1522 4 : "the mapping document must not survive the flip"
1523 4 : );
1524 4 : }
1525 :
1526 : /// 5.8.1.4 noLongerMatching: a registration that cannot delete the
1527 : /// remote copy still loses its mapping, so a notification the orphaned
1528 : /// copy keeps sending reaches nobody.
1529 : #[tokio::test]
1530 4 : async fn clause_5_8_1_no_longer_matching_unmaps_a_source_without_delete() {
1531 4 : let st = AppState::new("antares-ds-nolonger".into());
1532 4 : let t = TenantId::default();
1533 4 : let sub = sub_doc();
1534 4 : let own = sub["id"].as_str().expect("id").to_owned();
1535 4 : st.store
1536 4 : .create(&t, Kind::Subscription, &own, sub)
1537 4 : .await
1538 4 : .expect("create");
1539 4 : ds_put(&st, &t, &own, json!({"csr_sub": "urn:csr:nolonger"})).await;
1540 4 : let mut reg = reg_doc();
1541 4 : reg["operations"] = json!(["createSubscription", "updateSubscription"]);
1542 4 : let reg_id = reg["id"].as_str().expect("id").to_owned();
1543 4 : assert!(
1544 4 : ds_set_remote(
1545 4 : &st,
1546 4 : &t,
1547 4 : &own,
1548 4 : ®_id,
1549 4 : Some(json!(["http://source.example.org", "urn:remote:nolonger"]))
1550 4 : )
1551 4 : .await
1552 : );
1553 4 : inbound_put(&st, "urn:remote:nolonger", &t, &own).await;
1554 4 : on_csource_notification(&st, &t, &own, Some("noLongerMatching"), &[reg]).await;
1555 4 : assert!(
1556 4 : ds_remotes(&ds_get(&st, &t, &own).await).is_empty(),
1557 : "the mapping outlived noLongerMatching"
1558 : );
1559 4 : assert!(
1560 4 : inbound_get(&st, "urn:remote:nolonger").await.is_none(),
1561 4 : "the inbound key outlived noLongerMatching"
1562 4 : );
1563 4 : }
1564 :
1565 : /// 5.8.1.4 stores ONE remote subscriptionId per (Subscription,
1566 : /// registration) pair. Two Context Source Notifications for the same pair
1567 : /// interleave at their forward, so the second insert is refused under the
1568 : /// store's own lock instead of overwriting — an overwritten mapping
1569 : /// orphans a live remote subscription at the source and doubles every
1570 : /// notification the subscriber receives.
1571 : #[tokio::test]
1572 4 : async fn clause_5_8_1_second_mapping_for_a_registration_is_refused() {
1573 4 : let st = AppState::new("antares-ds-cas".into());
1574 4 : let t = TenantId::default();
1575 4 : let own = "urn:ngsi-ld:Subscription:own";
1576 4 : ds_put(&st, &t, own, json!({"csr_sub": "urn:csr:1"})).await;
1577 4 : assert!(
1578 4 : ds_set_remote(
1579 4 : &st,
1580 4 : &t,
1581 4 : own,
1582 4 : "urn:reg:1",
1583 4 : Some(json!(["http://s", "urn:remote:1"]))
1584 4 : )
1585 4 : .await
1586 : );
1587 4 : assert!(
1588 4 : !ds_set_remote(
1589 4 : &st,
1590 4 : &t,
1591 4 : own,
1592 4 : "urn:reg:1",
1593 4 : Some(json!(["http://s", "urn:remote:2"]))
1594 4 : )
1595 4 : .await,
1596 : "the registration already has a remote subscription"
1597 : );
1598 4 : let got = ds_remotes(&ds_get(&st, &t, own).await);
1599 4 : assert_eq!(got.len(), 1);
1600 4 : assert_eq!(
1601 4 : got[0].1 .1, "urn:remote:1",
1602 4 : "the first mapping survives — the loser rolls its own copy back"
1603 4 : );
1604 4 : }
1605 :
1606 : /// 4.14: "the Tenant information from the Context Source Registration has
1607 : /// to be used" — the 5.8.5.4 delete-forward must travel with the stored
1608 : /// registration, not a synthetic id/endpoint pair, or it lands in the
1609 : /// peer's default tenant and the remote subscription is never deleted.
1610 : /// The synthetic fallback stays for the already-deleted registration.
1611 : #[test]
1612 4 : fn clause_5_8_5_delete_forward_carries_the_stored_registration() {
1613 4 : let stored = json!({
1614 4 : "id": "urn:ngsi-ld:ContextSourceRegistration:r1",
1615 4 : "endpoint": "http://source.example.org",
1616 4 : "tenant": "cityB",
1617 4 : "contextSourceInfo": [{"key": "Authorization", "value": "Bearer t"}],
1618 : });
1619 4 : let reg = forward_reg(
1620 4 : Some(stored.clone()),
1621 4 : "urn:ngsi-ld:ContextSourceRegistration:r1",
1622 4 : "http://source.example.org",
1623 : );
1624 4 : assert_eq!(reg["tenant"], json!("cityB"), "{reg}");
1625 4 : assert_eq!(reg["contextSourceInfo"], stored["contextSourceInfo"]);
1626 4 : let gone = forward_reg(
1627 4 : None,
1628 4 : "urn:ngsi-ld:ContextSourceRegistration:r1",
1629 4 : "http://source.example.org",
1630 : );
1631 4 : assert_eq!(gone["endpoint"], json!("http://source.example.org"));
1632 4 : assert!(
1633 4 : gone.get("tenant").is_none(),
1634 : "a deleted registration carries nothing to forward with: {gone}"
1635 : );
1636 4 : }
1637 :
1638 : /// 5.8.1.4: "The @context to be used for sending Notifications related to
1639 : /// this Subscription shall be the one specified in the jsonldContext
1640 : /// field." The forwarded copy carries the subscriber's own terms (`q` is
1641 : /// stored verbatim), so it must be shipped under the Subscription's
1642 : /// @context — under the core context those terms name Attributes that do
1643 : /// not exist at the source.
1644 : #[test]
1645 4 : fn clause_5_8_1_forwarded_copy_is_shipped_under_the_subscription_context() {
1646 4 : let st = AppState::new("antares-ds-ctx".into());
1647 4 : let core = st.loader.core().source.clone();
1648 4 : let hosted = json!({"jsonldContext": "http://broker.example.org/jsonldContexts/abc",
1649 4 : "__context": "http://example.org/ctx.jsonld"});
1650 4 : assert_eq!(
1651 4 : sub_ctx_url(&st, &hosted),
1652 : "http://broker.example.org/jsonldContexts/abc",
1653 : "jsonldContext wins — it is the member 5.8.1.4 names"
1654 : );
1655 4 : let own = json!({"__context": "http://example.org/ctx.jsonld"});
1656 4 : assert_eq!(sub_ctx_url(&st, &own), "http://example.org/ctx.jsonld");
1657 4 : assert_ne!(
1658 4 : sub_ctx_url(&st, &own),
1659 4 : crate::federation::ctx_link_url(&HeaderMap::new(), &core),
1660 : "a Subscription with its own vocabulary is not forwarded under \
1661 : the core context"
1662 : );
1663 4 : assert_eq!(
1664 4 : sub_ctx_url(&st, &json!({})),
1665 4 : crate::federation::ctx_link_url(&HeaderMap::new(), &core),
1666 : "with no context of its own the core context is the fallback"
1667 : );
1668 4 : }
1669 :
1670 : /// 5.8.1.4 stores three mappings; every store touch takes the tenant
1671 : /// first (4.14: "an NGSI-LD system shall behave as if the tenants were
1672 : /// separate systems"). The inbound index is keyed by the
1673 : /// broker-generated remote subscriptionId under its own reserved
1674 : /// tenant, so it never collides with a tenant's own mapping documents.
1675 : #[tokio::test]
1676 4 : async fn clause_5_8_1_mappings_are_tenant_scoped() {
1677 4 : let st = AppState::new("antares-ds-tenant".into());
1678 4 : let a = TenantId::new("alpha").expect("tenant");
1679 4 : let b = TenantId::new("beta").expect("tenant");
1680 4 : let own = "urn:ngsi-ld:Subscription:own";
1681 4 : ds_put(
1682 4 : &st,
1683 4 : &a,
1684 4 : own,
1685 4 : json!({"csr_sub": "urn:csr:1",
1686 4 : "remotes": {"urn:reg:1": ["http://s", "urn:remote:1"]}}),
1687 4 : )
1688 4 : .await;
1689 4 : assert_eq!(ds_remotes(&ds_get(&st, &a, own).await).len(), 1);
1690 4 : assert!(
1691 4 : ds_remotes(&ds_get(&st, &b, own).await).is_empty(),
1692 : "another tenant must not read the remote mapping"
1693 : );
1694 4 : assert!(ds_get(&st, &b, own).await.get("csr_sub").is_none());
1695 4 : inbound_put(&st, "urn:remote:1", &a, own).await;
1696 4 : assert_eq!(
1697 4 : inbound_get(&st, "urn:remote:1").await,
1698 4 : Some(("alpha".to_owned(), own.to_owned()))
1699 : );
1700 4 : assert!(
1701 4 : st.store
1702 4 : .get(&a, Kind::DistSub, "urn:remote:1")
1703 4 : .await
1704 4 : .ok()
1705 4 : .flatten()
1706 4 : .is_none(),
1707 : "the index entry must not land in the subscriber's own namespace"
1708 : );
1709 4 : inbound_delete(&st, "urn:remote:1").await;
1710 4 : assert!(inbound_get(&st, "urn:remote:1").await.is_none());
1711 4 : }
1712 :
1713 : /// 4.14 — "an NGSI-LD system shall behave as if the tenants were
1714 : /// separate systems" — through the one index every tenant's inbound
1715 : /// mappings share. Its key is the remote subscriptionId the Context
1716 : /// Source echoes back on every notification (5.8.1.4), and the tenant
1717 : /// the index names is the tenant the data is delivered to: nothing later
1718 : /// on that path re-derives it. So a mapping claiming a key another
1719 : /// tenant already holds must lose. An index where the newer write won
1720 : /// would let a Context Source hand one tenant's notifications to another
1721 : /// tenant's subscriber by returning a subscriptionId already in use.
1722 : #[tokio::test]
1723 4 : async fn clause_5_8_1_an_inbound_key_is_never_taken_from_the_tenant_holding_it() {
1724 4 : let st = AppState::new("antares-ds-claim".into());
1725 4 : let a = TenantId::new("alpha").expect("tenant");
1726 4 : let b = TenantId::new("beta").expect("tenant");
1727 4 : let shared = "urn:ngsi-ld:Subscription:distsub:collide";
1728 4 : inbound_put(&st, shared, &a, "urn:ngsi-ld:Subscription:a").await;
1729 4 : inbound_put(&st, shared, &b, "urn:ngsi-ld:Subscription:b").await;
1730 4 : assert_eq!(
1731 4 : inbound_get(&st, shared).await,
1732 4 : Some(("alpha".to_owned(), "urn:ngsi-ld:Subscription:a".to_owned())),
1733 4 : "the second tenant took the first tenant's inbound route"
1734 4 : );
1735 4 : }
1736 :
1737 : /// 5.8.2.4 keeps the Registration Subscription in step by mutating the
1738 : /// stored mapping document. That document comes out of the store, and
1739 : /// `Value`'s index panics on anything that is not an object or Null, so
1740 : /// the two shapes a driver can hand back — a mapping that is not an
1741 : /// object, and one whose `notification` member is not an object — decide
1742 : /// whether an update of an ordinary Subscription takes the process down.
1743 : /// The panic would land inside the closure `mutate` runs under the
1744 : /// store's write lock, so it is not one request that is lost.
1745 : #[tokio::test]
1746 4 : async fn clause_5_8_2_a_stored_mapping_of_the_wrong_shape_is_not_indexed() {
1747 24 : for stored in [
1748 4 : json!("not an object"),
1749 4 : json!([1, 2, 3]),
1750 4 : json!(7),
1751 4 : Value::Null,
1752 4 : json!({"notification": "not an object"}),
1753 4 : json!({"notification": [1]}),
1754 4 : ] {
1755 24 : let st = AppState::new("antares-ds-shape".into());
1756 24 : let tenant = TenantId::new("shape").expect("tenant");
1757 24 : let own = "urn:ngsi-ld:Subscription:shape";
1758 24 : st.store
1759 24 : .create(
1760 24 : &tenant,
1761 24 : Kind::Subscription,
1762 24 : own,
1763 24 : json!({"id": own, "type": "Subscription",
1764 24 : "entities": [{"type": "Vehicle"}],
1765 24 : "notification": {"endpoint": {"uri": "http://127.0.0.1:9/n"},
1766 24 : "attributes": ["speed"]}}),
1767 24 : )
1768 24 : .await
1769 24 : .expect("seed the subscription");
1770 24 : ds_put(&st, &tenant, own, json!({"csr_sub": "urn:csr:shape"})).await;
1771 24 : st.store
1772 24 : .create(&tenant, Kind::DistSub, "urn:csr:shape", stored.clone())
1773 24 : .await
1774 24 : .expect("seed the mapping");
1775 4 :
1776 24 : on_subscription_updated(&st, &tenant, own).await;
1777 4 :
1778 4 : // the update completes, and a mapping that could not be brought
1779 4 : // into step is left as it was rather than half-written
1780 24 : let after = st
1781 24 : .store
1782 24 : .get(&tenant, Kind::DistSub, "urn:csr:shape")
1783 24 : .await
1784 24 : .expect("read the mapping");
1785 24 : assert!(after.is_some(), "{stored}: the mapping was dropped");
1786 4 : }
1787 4 : }
1788 :
1789 : /// The same path on a well-formed mapping still carries the update
1790 : /// through: the guard above refuses a shape, it does not refuse the work.
1791 : #[tokio::test]
1792 4 : async fn clause_5_8_2_a_well_formed_mapping_is_brought_into_step() {
1793 4 : let st = AppState::new("antares-ds-step".into());
1794 4 : let tenant = TenantId::new("step").expect("tenant");
1795 4 : let own = "urn:ngsi-ld:Subscription:step";
1796 4 : st.store
1797 4 : .create(
1798 4 : &tenant,
1799 4 : Kind::Subscription,
1800 4 : own,
1801 4 : json!({"id": own, "type": "Subscription",
1802 4 : "entities": [{"type": "Bus"}],
1803 4 : "scopeQ": "/a/b",
1804 4 : "notification": {"endpoint": {"uri": "http://127.0.0.1:9/n"},
1805 4 : "attributes": ["speed"]}}),
1806 4 : )
1807 4 : .await
1808 4 : .expect("seed the subscription");
1809 4 : ds_put(&st, &tenant, own, json!({"csr_sub": "urn:csr:step"})).await;
1810 4 : st.store
1811 4 : .create(
1812 4 : &tenant,
1813 4 : Kind::DistSub,
1814 4 : "urn:csr:step",
1815 4 : json!({"id": "urn:csr:step", "type": "Subscription",
1816 4 : "entities": [{"type": "Tram"}],
1817 4 : "notification": {"endpoint": {"uri": "urn:antares:distsub:step"}}}),
1818 4 : )
1819 4 : .await
1820 4 : .expect("seed the mapping");
1821 :
1822 4 : on_subscription_updated(&st, &tenant, own).await;
1823 :
1824 4 : let after = st
1825 4 : .store
1826 4 : .get(&tenant, Kind::DistSub, "urn:csr:step")
1827 4 : .await
1828 4 : .ok()
1829 4 : .flatten()
1830 4 : .expect("the mapping is still there");
1831 4 : assert_eq!(
1832 4 : after["entities"],
1833 4 : json!([{"type": "Bus"}]),
1834 : "the match members follow the Subscription: {after}"
1835 : );
1836 4 : assert_eq!(after["scopeQ"], "/a/b", "{after}");
1837 4 : assert_eq!(
1838 4 : after["notification"]["attributes"],
1839 4 : json!(["speed"]),
1840 : "5.8.2.4: the notification attributes follow too: {after}"
1841 : );
1842 4 : assert!(
1843 4 : after["notification"]["endpoint"]["uri"]
1844 4 : .as_str()
1845 4 : .is_some_and(|u| u.starts_with("urn:antares:distsub:")),
1846 4 : "the endpoint the copy points at is not rewritten: {after}"
1847 4 : );
1848 4 : }
1849 :
1850 : /// The remotes index is read back from storage, so every malformed
1851 : /// shape is dropped rather than indexed or unwrapped.
1852 : #[test]
1853 4 : fn clause_5_8_1_remotes_index_drops_malformed_entries() {
1854 4 : let doc = json!({"remotes": {
1855 4 : "urn:reg:ok": ["http://s", "urn:remote:1"],
1856 4 : "urn:reg:short": ["http://s"],
1857 4 : "urn:reg:nonstring": [1, 2],
1858 4 : "urn:reg:notarray": "http://s",
1859 4 : "urn:reg:empty": [],
1860 : }});
1861 4 : let got = ds_remotes(&doc);
1862 4 : assert_eq!(got.len(), 1, "{got:?}");
1863 4 : assert_eq!(got[0].0, "urn:reg:ok");
1864 4 : assert_eq!(got[0].1 .1, "urn:remote:1");
1865 4 : assert!(ds_remotes(&json!({"remotes": "nope"})).is_empty());
1866 4 : assert!(ds_remotes(&json!({})).is_empty());
1867 4 : }
1868 :
1869 : /// 5.8.1.4 / 5.8.5.4: the per-registration branches interleave at their
1870 : /// forward, so one registration's mapping write must not drop another's,
1871 : /// and a mapping document deleted by Delete Subscription must stay
1872 : /// deleted — an in-flight branch may not write it back.
1873 : #[tokio::test]
1874 4 : async fn clause_5_8_5_mapping_writes_never_resurrect_a_deleted_subscription() {
1875 4 : let st = AppState::new("antares-ds-resurrect".into());
1876 4 : let t = TenantId::default();
1877 4 : let own = "urn:ngsi-ld:Subscription:own";
1878 4 : ds_put(&st, &t, own, json!({"csr_sub": "urn:csr:1"})).await;
1879 4 : assert!(
1880 4 : ds_set_remote(
1881 4 : &st,
1882 4 : &t,
1883 4 : own,
1884 4 : "urn:reg:1",
1885 4 : Some(json!(["http://s1", "urn:remote:1"]))
1886 4 : )
1887 4 : .await
1888 : );
1889 4 : assert!(
1890 4 : ds_set_remote(
1891 4 : &st,
1892 4 : &t,
1893 4 : own,
1894 4 : "urn:reg:2",
1895 4 : Some(json!(["http://s2", "urn:remote:2"]))
1896 4 : )
1897 4 : .await
1898 : );
1899 4 : assert_eq!(
1900 4 : ds_remotes(&ds_get(&st, &t, own).await).len(),
1901 : 2,
1902 : "a second registration's write must not drop the first"
1903 : );
1904 4 : assert!(ds_set_remote(&st, &t, own, "urn:reg:1", None).await);
1905 4 : assert_eq!(ds_remotes(&ds_get(&st, &t, own).await).len(), 1);
1906 4 : let _ = st.store.delete(&t, Kind::DistSub, own).await;
1907 4 : assert!(
1908 4 : !ds_set_remote(
1909 4 : &st,
1910 4 : &t,
1911 4 : own,
1912 4 : "urn:reg:3",
1913 4 : Some(json!(["http://s3", "urn:remote:3"]))
1914 4 : )
1915 4 : .await,
1916 : "there is no mapping document left to write to"
1917 : );
1918 4 : assert!(
1919 4 : st.store
1920 4 : .get(&t, Kind::DistSub, own)
1921 4 : .await
1922 4 : .ok()
1923 4 : .flatten()
1924 4 : .is_none(),
1925 4 : "a deleted mapping document must stay deleted"
1926 4 : );
1927 4 : }
1928 :
1929 : /// The inbound notification endpoint is peer-facing: the Entity count is
1930 : /// capped before any store touch, because the 5.8.6 merge runs one local
1931 : /// retrieve and one federated fan-out per notified Entity.
1932 : #[tokio::test]
1933 4 : async fn clause_5_8_6_inbound_notification_entity_count_is_capped() {
1934 4 : let st = AppState::new("antares-ds-cap".into());
1935 4 : let cap = *crate::bounds::MAX_BATCH_ITEMS;
1936 8 : let body = |n: usize| {
1937 8 : json!({
1938 8 : "type": "Notification",
1939 8 : "subscriptionId": "urn:ngsi-ld:Subscription:remote1",
1940 8 : "data": vec![json!({"id": "urn:ngsi-ld:Vehicle:1", "type": "Vehicle"}); n],
1941 : })
1942 8 : .to_string()
1943 8 : };
1944 4 : let over = remote_notify(State(st.clone()), Bytes::from(body(cap + 1))).await;
1945 4 : assert_eq!(
1946 4 : over.status(),
1947 : StatusCode::BAD_REQUEST,
1948 : "an over-cap notification is rejected before the mapping lookup"
1949 : );
1950 4 : let at_cap = remote_notify(State(st.clone()), Bytes::from(body(cap))).await;
1951 4 : assert_eq!(
1952 4 : at_cap.status(),
1953 4 : StatusCode::NOT_FOUND,
1954 4 : "the ceiling itself is accepted — the unknown mapping is what stops it"
1955 4 : );
1956 4 : }
1957 :
1958 : /// 5.8.1.4 stores the mapping "to enable forwarding received
1959 : /// notifications to the original subscriber": once that subscriber is
1960 : /// gone the mapping is dead weight, so it is pruned on touch, and the
1961 : /// peer is answered about its own id — never told the local one.
1962 : #[tokio::test]
1963 4 : async fn clause_5_8_1_mapping_to_a_deleted_subscription_is_pruned() {
1964 4 : let st = AppState::new("antares-ds-prune".into());
1965 4 : let t = TenantId::new("alpha").expect("tenant");
1966 4 : let remote = "urn:ngsi-ld:Subscription:remote9";
1967 4 : inbound_put(&st, remote, &t, "urn:ngsi-ld:Subscription:gone").await;
1968 4 : let body =
1969 4 : json!({"type": "Notification", "subscriptionId": remote, "data": []}).to_string();
1970 4 : let resp = remote_notify(State(st.clone()), Bytes::from(body)).await;
1971 4 : assert_eq!(resp.status(), StatusCode::NOT_FOUND);
1972 4 : assert!(
1973 4 : inbound_get(&st, remote).await.is_none(),
1974 : "the dead mapping must not survive the touch"
1975 : );
1976 4 : let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
1977 4 : .await
1978 4 : .expect("body");
1979 4 : let text = String::from_utf8_lossy(&bytes);
1980 4 : assert!(
1981 4 : !text.contains("urn:ngsi-ld:Subscription:gone"),
1982 4 : "the local Subscription id must not leak to the peer: {text}"
1983 4 : );
1984 4 : }
1985 :
1986 : /// 5.8.1.4 consumer half, as a seam: the delivery path hands a
1987 : /// notification on the internal endpoint to the handler `wire` installs
1988 : /// and to nothing else. An unwired state has no handler, so such a
1989 : /// notification is dropped rather than delivered somewhere — the seam
1990 : /// fails closed, and a caller cannot reach `distsub` by bypassing it.
1991 : #[tokio::test]
1992 4 : async fn the_internal_endpoint_is_reached_only_through_the_installed_handler() {
1993 4 : let bare = AppState::new("antares".into());
1994 4 : assert!(
1995 4 : bare.csource_notification.is_none(),
1996 : "a state nobody wired carries no handler"
1997 : );
1998 4 : let wired = crate::wired_state("antares");
1999 4 : assert!(
2000 4 : wired.await.csource_notification.is_some(),
2001 4 : "wire installs the consumer half"
2002 4 : );
2003 4 : }
2004 :
2005 : /// 5.8.6 forwards an inbound notification to the original subscriber,
2006 : /// and that delivery is a notification like any other: it passes
2007 : /// `pre_notify` once, under the subject stored with the ORIGINAL
2008 : /// Subscription, so a narrowing the subscriber is under holds for data a
2009 : /// remote source sent (operation matrix section 8 item 13).
2010 : #[tokio::test(flavor = "multi_thread")]
2011 4 : async fn a_remote_notification_is_narrowed_for_the_original_subscriber() {
2012 : use crate::policy::{
2013 : Decision, DecisionFuture, NotifyDecision, Operation, PolicyEngine, Subject,
2014 : };
2015 : use std::sync::{Arc, Mutex};
2016 :
2017 : /// Strips `secret` from every notified Entity and records the
2018 : /// subject it was asked about.
2019 : struct OmitSecret(Mutex<Vec<Vec<(String, String)>>>);
2020 : impl PolicyEngine for OmitSecret {
2021 0 : fn name(&self) -> &str {
2022 0 : "omit-secret"
2023 0 : }
2024 0 : fn decide<'a>(&'a self, _s: &'a Subject, _op: &'a Operation<'a>) -> DecisionFuture<'a> {
2025 0 : Box::pin(std::future::ready(Decision::Allow))
2026 0 : }
2027 4 : fn pre_notify(&self, s: &Subject, _sub: &Value, n: &mut Value) -> NotifyDecision {
2028 4 : self.0.lock().expect("subjects").push(s.headers.clone());
2029 4 : for e in n
2030 4 : .get_mut("data")
2031 4 : .and_then(Value::as_array_mut)
2032 4 : .into_iter()
2033 4 : .flatten()
2034 : {
2035 4 : if let Some(o) = e.as_object_mut() {
2036 4 : o.remove("secret");
2037 4 : }
2038 : }
2039 4 : NotifyDecision::Deliver
2040 4 : }
2041 : }
2042 :
2043 4 : let seen: Arc<Mutex<Vec<Value>>> = Arc::default();
2044 4 : let sink = seen.clone();
2045 4 : let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
2046 4 : .await
2047 4 : .expect("bind");
2048 4 : let addr = listener.local_addr().expect("addr");
2049 4 : let app = axum::Router::new().route(
2050 4 : "/notify",
2051 4 : axum::routing::post(move |body: String| {
2052 4 : let sink = sink.clone();
2053 4 : async move {
2054 4 : if let Ok(v) = serde_json::from_str::<Value>(&body) {
2055 4 : sink.lock().expect("bodies").push(v);
2056 4 : }
2057 4 : StatusCode::OK
2058 4 : }
2059 4 : }),
2060 : );
2061 4 : tokio::spawn(async move { axum::serve(listener, app).await.expect("serve") });
2062 :
2063 4 : let engine = Arc::new(OmitSecret(Mutex::default()));
2064 4 : let st = AppState::new("antares-ds-narrow".into()).with_policy(engine.clone());
2065 4 : let t = TenantId::new("alpha").expect("tenant");
2066 4 : let own = "urn:ngsi-ld:Subscription:own-narrowed";
2067 4 : st.store
2068 4 : .create(
2069 4 : &t,
2070 4 : Kind::Subscription,
2071 4 : own,
2072 4 : json!({
2073 4 : "id": own,
2074 4 : "type": "Subscription",
2075 4 : "entities": [{"type": format!("{DC}Vehicle")}],
2076 4 : "notification": {"endpoint": {
2077 4 : "uri": format!("http://{addr}/notify"),
2078 4 : "accept": "application/json",
2079 4 : }},
2080 4 : crate::policy::SUBJECT_MEMBER: [["x-user", "alice"]],
2081 4 : }),
2082 4 : )
2083 4 : .await
2084 4 : .expect("subscription row");
2085 4 : let remote = "urn:ngsi-ld:Subscription:remote-narrowed";
2086 4 : inbound_put(&st, remote, &t, own).await;
2087 :
2088 4 : let body = json!({
2089 4 : "type": "Notification",
2090 4 : "subscriptionId": remote,
2091 4 : "data": [{
2092 4 : "id": "urn:ngsi-ld:Vehicle:1",
2093 4 : "type": "Vehicle",
2094 4 : "speed": {"type": "Property", "value": 50},
2095 4 : "secret": {"type": "Property", "value": "hidden"},
2096 : }],
2097 : })
2098 4 : .to_string();
2099 4 : let resp = remote_notify(State(st.clone()), Bytes::from(body)).await;
2100 4 : assert_eq!(resp.status(), StatusCode::OK);
2101 :
2102 4 : let mut delivered = None;
2103 4 : for _ in 0..100 {
2104 8 : if let Some(b) = seen.lock().expect("bodies").first().cloned() {
2105 4 : delivered = Some(b);
2106 4 : break;
2107 4 : }
2108 4 : tokio::time::sleep(std::time::Duration::from_millis(50)).await;
2109 : }
2110 4 : let delivered = delivered.expect("the original subscriber was notified");
2111 4 : let entity = &delivered["data"][0];
2112 4 : assert_eq!(entity["id"], "urn:ngsi-ld:Vehicle:1");
2113 4 : assert!(
2114 4 : entity.get("speed").is_some(),
2115 : "visible data travels: {delivered}"
2116 : );
2117 4 : assert!(
2118 4 : entity.get("secret").is_none(),
2119 : "the subscriber's narrowing holds for remote data: {delivered}"
2120 : );
2121 4 : assert_eq!(
2122 4 : delivered["subscriptionId"], own,
2123 : "delivered under the subscriber's own id"
2124 : );
2125 4 : assert_eq!(
2126 4 : *engine.0.lock().expect("subjects"),
2127 4 : vec![vec![("x-user".to_owned(), "alice".to_owned())]],
2128 4 : "asked once, about the original subscriber"
2129 4 : );
2130 4 : }
2131 : }
|
