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