Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Distributed operations (4.3.6, 5.12 matching, 6.3.17–6.3.19).
3 : //!
4 : //! Registration modes: inclusive (local + forward), auxiliary (read-only
5 : //! supplement, local wins), exclusive/redirect (proxied — registered data is
6 : //! never held locally). Forwarded requests carry `Via: 1.1 <hostAlias>` and
7 : //! the request @context as a Link header; bodies travel as application/json
8 : //! without an inline @context.
9 : //!
10 : //! Three contracts hold across this module and `csource.rs`: candidate
11 : //! matching is SQL over `csource_index`, never a scan of a tenant's
12 : //! registrations; a forwarded query is narrowed to the registration's scope
13 : //! (4.3.6.1 — spec-mandated, not a bug to be fixed away); and fan-out is
14 : //! bounded, by a semaphore, a per-source timeout and an aggregate deadline.
15 :
16 : use crate::negotiate::*;
17 : use crate::paging::pct_encode;
18 : use crate::state::AppState;
19 : use antares_jsonld::Context;
20 : use antares_model::operations::{group_members, DEFAULT_OPERATION_GROUP};
21 : use antares_model::{NgsiError, TenantId};
22 : #[cfg(test)]
23 : use antares_store::Kind;
24 : use axum::http::{HeaderMap, StatusCode};
25 : use axum::response::{IntoResponse, Response};
26 : use serde_json::{json, Map, Value};
27 : use std::collections::HashMap;
28 :
29 : /// One matching registration, compiled for forwarding.
30 : #[derive(Clone, Debug, Default)]
31 : pub struct FedReg {
32 : /// The registration's @id — carried into `NotUpdatedDetails.registrationId`
33 : /// (5.2.19) and `BatchEntityError.registrationId` (5.2.17).
34 : pub reg_id: String,
35 : pub endpoint: String,
36 : pub mode: String, // inclusive | auxiliary | exclusive | redirect
37 : pub(crate) ops: Vec<String>,
38 : /// Expanded attribute IRIs the matched RegistrationInfo covers; None ⇒ all.
39 : pub attrs: Option<Vec<String>>,
40 : /// EntityInfo ids/types of the matched RegistrationInfo elements.
41 : pub ent_ids: Vec<String>,
42 : pub ent_types: Vec<String>,
43 : /// EntityInfo idPattern values (5.2.8 IEEE 1003.2 regexes).
44 : pub ent_patterns: Vec<String>,
45 : /// True when any matched RegistrationInfo carries an EntityInfo with
46 : /// neither id nor idPattern (or no entities at all) — the registration
47 : /// imposes no id restriction (5.12 condition 1).
48 : pub ent_unrestricted: bool,
49 : /// 5.2.9 `tenant`: the Tenant to specify in all requests to this Context
50 : /// Source. None ⇒ the requesting tenant is carried through unchanged.
51 : pub tenant: Option<String>,
52 : /// 5.2.9 `contextSourceAlias`: "a previously retrieved unique id for a
53 : /// registered Context Source which is used to identify loops", tenant-
54 : /// specific per Table 5.2.9-1. A registration whose alias is already in
55 : /// the inbound Via chain names a source this request has visited, so it
56 : /// is not a matching registration (Table 6.3.18-2).
57 : pub alias: Option<String>,
58 : /// 4.3.6.5 `contextSourceInfo` key/value pairs, conveyed as headers on
59 : /// every forward to this source (string values only — headers).
60 : pub csi: Vec<(String, String)>,
61 : /// 5.2.9 `localOnly` (4.3.6.4): distributed operations for this
62 : /// registration "will act only on data held directly by the registered
63 : /// Context Source itself" — every forward carries `local=true`.
64 : pub local_only: bool,
65 : /// 5.2.34 timeout: "Maximum period of time in milliseconds which may
66 : /// elapse before a forwarded request is assumed to have failed."
67 : pub timeout_ms: Option<u64>,
68 : /// 5.2.34 cooldown: "Minimum period of time in milliseconds which shall
69 : /// elapse before attempting to make a subsequent forwarded request to
70 : /// the same endpoint after failure."
71 : pub cooldown_ms: Option<u64>,
72 : }
73 :
74 : impl FedReg {
75 : /// 4.3.6.1: registered Context Sources "may indicate that they are only
76 : /// willing to respond to a limited subset of API operations. Context
77 : /// Brokers shall respect this, to avoid unnecessarily sending distributed
78 : /// operation requests which are always guaranteed to fail." Matches the
79 : /// registration's `operations` list (5.2.9) by name or operation group;
80 : /// default when absent is federationOps.
81 27213 : pub fn supports(&self, op: &str) -> bool {
82 27213 : self.ops
83 27213 : .iter()
84 40689 : .any(|o| o == op || group_members(o).is_some_and(|members| members.contains(&op)))
85 27213 : }
86 576 : pub fn is_proxy(&self) -> bool {
87 576 : self.mode == "exclusive" || self.mode == "redirect"
88 576 : }
89 : /// 4.3.6.1 ("all constraints specified in the registration shall be
90 : /// respected" — including Entity IDs): can this registration's
91 : /// EntityInfo id constraints match `id`? Patterns use regex find,
92 : /// mirroring `entity_info_matches` (5.12).
93 632 : pub fn can_match_id(&self, id: &str) -> bool {
94 632 : self.ent_unrestricted
95 96 : || self.ent_ids.iter().any(|i| i == id)
96 44 : || self
97 44 : .ent_patterns
98 44 : .iter()
99 44 : .any(|p| antares_ql::regex::compile(p).is_ok_and(|re| re.find(id).is_some()))
100 632 : }
101 : /// Does this registration cover the given expanded attribute IRI?
102 222 : pub fn covers_attr(&self, iri: &str) -> bool {
103 222 : self.attrs
104 222 : .as_ref()
105 222 : .is_none_or(|a| a.iter().any(|x| x == iri))
106 222 : }
107 24 : pub fn read_op(&self) -> Option<&'static str> {
108 24 : ["retrieveEntity", "queryEntity", "queryBatch"]
109 24 : .into_iter()
110 24 : .find(|op| self.supports(op))
111 24 : }
112 : /// 4.20 Table 4.20-1: queryEntity/queryBatch implement 5.7.2 Query
113 : /// Entities — retrieveEntity implements only 5.7.1, so a query is never
114 : /// forwarded to a source that offers retrieveEntity alone (4.3.6.1:
115 : /// "Context Brokers shall respect this").
116 372 : pub fn query_op(&self) -> Option<&'static str> {
117 372 : ["queryEntity", "queryBatch"]
118 372 : .into_iter()
119 388 : .find(|op| self.supports(op))
120 372 : }
121 : /// 4.3.6.1: the registration's EntityInfo constraints gate which payload
122 : /// ITEMS a distributed write may carry — an item whose present id/type
123 : /// the registration does not name is not this source's data. An item
124 : /// without a `type` member (attribute fragments) cannot be disproven and
125 : /// stays covered.
126 110 : pub fn covers_item(&self, obj: &Map<String, Value>, ctx: &Context) -> bool {
127 110 : if !self.ent_ids.is_empty() || !self.ent_patterns.is_empty() {
128 62 : if let Some(id) = obj.get("id").and_then(Value::as_str) {
129 46 : if !self.can_match_id(id) {
130 4 : return false;
131 42 : }
132 16 : }
133 48 : }
134 106 : if !self.ent_types.is_empty() {
135 86 : let types: Vec<String> = match obj.get("type") {
136 74 : Some(Value::String(t)) => vec![ctx.expand_key(t)],
137 0 : Some(Value::Array(a)) => a
138 0 : .iter()
139 0 : .filter_map(Value::as_str)
140 0 : .map(|t| ctx.expand_key(t))
141 0 : .collect(),
142 12 : _ => Vec::new(),
143 : };
144 86 : if !types.is_empty() && !types.iter().any(|t| self.ent_types.contains(t)) {
145 8 : return false;
146 78 : }
147 20 : }
148 98 : true
149 110 : }
150 : }
151 :
152 : /// Is federation active for this request? (6.3.18 local param; 5.5.13:
153 : /// a local-scope request executes only on information available locally —
154 : /// no Context Source Registrations are considered)
155 : ///
156 : /// Table 6.4.3.2-1: for `type=*`, "local is implicitly set to true and shall
157 : /// not be explicitly set to false" — so the wildcard alone disables forwarding.
158 16476 : pub fn active(params: &HashMap<String, String>) -> bool {
159 16476 : if params.get("type").map(String::as_str) == Some("*") {
160 20 : return false;
161 16456 : }
162 16456 : params.get("local").map(String::as_str) != Some("true")
163 16476 : }
164 :
165 : /// The full inbound Via chain, joined across header FIELDS.
166 : ///
167 : /// RFC 7230 sections 3.2.2 and 5.7.1: Via is a list header — senders may split the
168 : /// chain over any number of `Via:` field lines, and the two forms are
169 : /// equivalent. Reading only the first field (`headers.get`) made a loop
170 : /// pseudonym in a later field invisible (undetected cycle) AND rebuilt the
171 : /// outbound chain from the truncated view, deleting upstream hop history
172 : /// that downstream brokers need for THEIR loop detection.
173 19298 : pub fn inbound_via(headers: &HeaderMap) -> Option<String> {
174 19298 : let fields: Vec<&str> = headers
175 19298 : .get_all("via")
176 19298 : .iter()
177 19298 : .filter_map(|v| v.to_str().ok())
178 19298 : .collect();
179 19298 : if fields.is_empty() {
180 19024 : None
181 : } else {
182 274 : Some(fields.join(", "))
183 : }
184 19298 : }
185 :
186 : /// This broker's Via pseudonym **for one Tenant**.
187 : ///
188 : /// Table 5.2.40-1: the alias is "a unique id for a Context Source which can
189 : /// be used to identify loops. In the multi-tenancy use case (see clause
190 : /// 4.14), this id **shall** be identifying a specific Tenant within a
191 : /// registered Context Source." One static per-process alias therefore makes
192 : /// every tenant of this broker look like the same Context Source: a request
193 : /// in tenant B whose registration points back here for tenant A is a
194 : /// different (source, tenant) pair, but a tenant-blind chain reads it as a
195 : /// loop and drops the registration.
196 : ///
197 : /// Format `{alias}~{tenant}`, and the bare alias for the default tenant —
198 : /// mirroring 6.3.14, where the tenant header is omitted rather than sent as
199 : /// `default`. `~` is an RFC 7230 token character that cannot occur in a
200 : /// `TenantId` (`[A-Za-z0-9_-]{1,64}`), so the two halves never blur; the
201 : /// broker rejects a configured `ANTARES_HOST_ALIAS` containing `~` at
202 : /// startup, so `a~b` in the default tenant cannot collide with `a` in
203 : /// tenant `b`.
204 : ///
205 : /// The value is stable for the life of a deployment because peers **register**
206 : /// it: a Context Source Registration's `contextSourceAlias` (Table 5.2.9-1) is
207 : /// "a previously retrieved unique id" — retrieved from this broker's
208 : /// `/info/sourceIdentity` for that tenant. Changing an alias silently breaks
209 : /// every peer's loop detection, so treat it as a published identifier.
210 16259 : pub fn alias_for(host_alias: &str, tenant: &TenantId) -> String {
211 16259 : if tenant.as_str() == TenantId::DEFAULT {
212 15452 : host_alias.to_owned()
213 : } else {
214 807 : format!("{host_alias}~{}", tenant.as_str())
215 : }
216 16259 : }
217 :
218 : /// The `received-by` pseudonyms of the inbound Via chain — the Context
219 : /// Sources this request has already passed through (Table 6.3.18-2).
220 : ///
221 : /// RFC 7230: `Via = 1#( received-protocol RWS received-by [ RWS comment ] )`
222 : /// — received-by is the SECOND whitespace token of each element and is a
223 : /// token compared for equality, never by suffix (`ends_with`
224 : /// made alias `b1` match peer `sub-b1`). A malformed element with no
225 : /// protocol falls back to its first token.
226 18559 : pub fn via_tokens(headers: &HeaderMap) -> Vec<String> {
227 18559 : inbound_via(headers)
228 18559 : .map(|v| {
229 234 : v.split(',')
230 882 : .filter_map(|t| {
231 882 : let mut toks = t.split_whitespace();
232 882 : let first = toks.next();
233 882 : toks.next().or(first).map(str::to_owned)
234 882 : })
235 234 : .collect()
236 234 : })
237 18559 : .unwrap_or_default()
238 18559 : }
239 :
240 : /// How many `Via` elements this broker will process on one inbound request.
241 : ///
242 : /// 6.3.18 fixes the header's purpose ("to avoid infinite loops") and Table
243 : /// 6.3.18-2 makes its listing part of registration matching, so every
244 : /// element is compared against every candidate registration: the chain is
245 : /// attacker-supplied input whose length multiplies the work of the request.
246 : /// RFC 7230 section 3.2.5 lets a recipient refuse a field longer than it is
247 : /// willing to process. A cascade deeper than this is a loop the alias
248 : /// pseudonyms failed to name, not a deployment.
249 : const MAX_VIA_HOPS: usize = 32;
250 :
251 : /// The number of `Via` elements received, counted across header fields
252 : /// without building the token list (RFC 7230 sections 3.2.2 and 5.7.1: the
253 : /// elements of a list header may be split over any number of field lines).
254 32062 : fn via_hops(headers: &HeaderMap) -> usize {
255 32062 : headers
256 32062 : .get_all("via")
257 32062 : .iter()
258 32062 : .filter_map(|v| v.to_str().ok())
259 32062 : .map(|v| v.split(',').count())
260 32062 : .sum()
261 32062 : }
262 :
263 : /// Does the inbound Via chain already name this broker, in this tenant?
264 : /// (loop, 6.3.18) — `alias` is always [`alias_for`]'s tenant-qualified value.
265 2558 : pub fn via_loop(headers: &HeaderMap, alias: &str) -> bool {
266 2558 : via_hops(headers) > MAX_VIA_HOPS || via_tokens(headers).iter().any(|t| t == alias)
267 2558 : }
268 :
269 : /// 6.3.17/6.3.18 loop handling for operations with matching registrations.
270 : /// 508 Loop Detected is mandated ONLY "in the case of an exclusive or
271 : /// redirect registration, where all of the data is held outside of the
272 : /// Context Broker and held in a single registered source ... registered to
273 : /// redirect back on to the Context Broker". Any other loop clears `regs` —
274 : /// the Via listing "is used when determining matching registrations"
275 : /// (Table 6.3.18-2), so the operation proceeds locally without re-forwarding.
276 13540 : pub fn handle_via_loop(
277 13540 : headers: &HeaderMap,
278 13540 : alias: &str,
279 13540 : tenant: &TenantId,
280 13540 : regs: &mut Vec<FedReg>,
281 13540 : ) -> Option<Response> {
282 : // A chain past [`MAX_VIA_HOPS`] is refused outright: the operation is
283 : // not re-forwarded and not run locally either, because a request that
284 : // deep is a cascade the pseudonyms failed to close.
285 13540 : if via_hops(headers) > MAX_VIA_HOPS {
286 18 : regs.clear();
287 18 : return Some(loop_508(tenant));
288 13522 : }
289 13522 : if regs.is_empty() || !via_loop(headers, alias) {
290 13480 : return None;
291 42 : }
292 42 : if regs.len() == 1 && regs[0].is_proxy() {
293 24 : return Some(loop_508(tenant));
294 18 : }
295 18 : regs.clear();
296 18 : None
297 13540 : }
298 :
299 : /// One `NGSILD-Warning` header value (6.3.17), in RFC 7234 warn form:
300 : /// `warn-code SP warn-agent SP quoted warn-text`.
301 32 : pub fn warning(code: u16, alias: &str, text: &str) -> String {
302 32 : format!("{code} {alias} \"{text}\"")
303 32 : }
304 :
305 : /// Classify one forwarded-read outcome per Table 6.3.17-1. A registration
306 : /// endpoint answering 404 with no data "should not be considered as abnormal
307 : /// behaviour"; 503/504 means no response arrived within the timeout (199);
308 : /// any other error status IS a received error response (299); a 2xx whose
309 : /// payload could not be parsed as NGSI-LD is 111.
310 200 : fn read_warning(status: u16, body: &Value) -> Option<(u16, &'static str)> {
311 186 : match status {
312 0 : 404 => None,
313 8 : 503 | 504 => Some((
314 8 : 199,
315 8 : "no response was received from the registration endpoint within the timeout period",
316 8 : )),
317 192 : s if s >= 400 => Some((
318 6 : 299,
319 6 : "an error response was received from the registration endpoint",
320 6 : )),
321 186 : s if (200..300).contains(&s) && body.is_null() => {
322 4 : Some((111, "the payload of the response was invalid"))
323 : }
324 182 : _ => None,
325 : }
326 200 : }
327 :
328 421 : fn outbound_via(headers: &HeaderMap, alias: &str) -> String {
329 421 : match inbound_via(headers) {
330 24 : Some(v) => format!("{v}, 1.1 {alias}"),
331 397 : None => format!("1.1 {alias}"),
332 : }
333 421 : }
334 :
335 : /// Percent-encode one client-controlled value for use as a single path
336 : /// segment of a forwarded URL (RFC 3986 clause 3.3: a segment is made of
337 : /// `pchar`, and `/`, `?`, `#` end it). Entity ids and attribute names arrive
338 : /// already percent-decoded from the request path, so splicing them raw would
339 : /// let `#` or `?` truncate the forwarded path and re-target the peer's
340 : /// resource — `.../entities/urn:x%23/attrs/speed` would reach the peer as
341 : /// Delete Entity (5.6.6) instead of Delete Attribute (5.6.5).
342 8252 : pub(crate) fn path_segment(s: &str) -> String {
343 33606 : pct_encode(s, |b| {
344 148 : matches!(
345 33606 : b,
346 : b'-' | b'.'
347 : | b'_'
348 : | b'~'
349 : | b'!'
350 : | b'$'
351 : | b'&'
352 : | b'\''
353 : | b'('
354 : | b')'
355 : | b'*'
356 : | b'+'
357 : | b','
358 : | b';'
359 : | b'='
360 : | b':'
361 : | b'@'
362 : )
363 33606 : })
364 8252 : }
365 :
366 : /// The @context URL to advertise on forwarded requests.
367 2481 : pub fn ctx_link_url(headers: &HeaderMap, source: &Value) -> String {
368 : // the request was validated on the way in, so an ambiguous Link is
369 : // already a 400 and cannot reach a forward
370 2481 : if let Ok(Some(url)) = link_context(headers) {
371 18 : return url;
372 2463 : }
373 0 : match source {
374 2459 : Value::String(s) => s.clone(),
375 : // 5.5.7/6.3.5 fidelity: an inline @context has no dereferenceable
376 : // URL — serialize it so forward() can embed it in the body as
377 : // application/ld+json instead of dropping the term mappings.
378 0 : Value::Array(a) if a.iter().any(|e| !e.is_string()) => {
379 0 : serde_json::to_string(source).unwrap_or_else(|_| antares_jsonld::CORE_CONTEXT.into())
380 : }
381 0 : Value::Array(a) => a
382 0 : .iter()
383 0 : .find_map(|e| e.as_str())
384 0 : .unwrap_or(antares_jsonld::CORE_CONTEXT)
385 0 : .to_owned(),
386 : Value::Object(_) => {
387 4 : serde_json::to_string(source).unwrap_or_else(|_| antares_jsonld::CORE_CONTEXT.into())
388 : }
389 0 : _ => antares_jsonld::CORE_CONTEXT.to_owned(),
390 : }
391 2481 : }
392 :
393 : /// The registration documents of one tenant: the ONE compiled mirror when
394 : /// wired (bus=nats), the store otherwise — narrowed there by the ids and
395 : /// Entity Types this operation names, so a broker holding a large
396 : /// registration set does not read all of it per distributed request.
397 : ///
398 : /// A store that cannot answer is an error, never an empty set: "no Context
399 : /// Source is registered" and "the registrations could not be read" lead to
400 : /// opposite answers, and only the first of them is one the client may be
401 : /// shown as complete (Table 6.3.2-1 InternalError).
402 : ///
403 : /// The narrowing may only ever drop registrations [`reg_candidate`] would
404 : /// reject anyway; it is a prefilter, never the decision. So the type
405 : /// dimension is dropped whenever a member is a 4.17 Entity Type Selection
406 : /// (`A|B`, `(A;B)`, `*`) rather than a single type: the index compares types
407 : /// by equality and cannot evaluate a selection, and a narrowing that
408 : /// mis-decides would silently lose a Context Source. A plain term is
409 : /// expanded first — the spec carries the parameter as the client wrote it
410 : /// (`Vehicle`), the index stores what [`reg_candidate`] compares: the IRI.
411 15946 : async fn reg_docs(
412 15946 : st: &AppState,
413 15946 : tenant: &TenantId,
414 15946 : spec: &crate::registry::CsrSpec,
415 15946 : ctx: &Context,
416 15946 : ) -> Result<Vec<std::sync::Arc<Value>>, NgsiError> {
417 15946 : let types: Option<Vec<String>> = spec
418 15946 : .types
419 15946 : .as_ref()
420 15946 : .filter(|ts| {
421 9544 : !ts.iter()
422 14448 : .any(|t| t.contains([',', ';', '|', '(', ')', '*']))
423 9544 : })
424 16202 : .map(|ts| ts.iter().map(|t| ctx.expand_key(t)).collect());
425 : // The id dimension is dropped whenever the query ALSO carries an
426 : // idPattern: `entity_info_matches` lets a registration's own entity id
427 : // match that pattern, so narrowing to the id list alone would drop a
428 : // Context Source 5.12 matches — the one thing a prefilter may never do.
429 15946 : let ids = spec.ids.as_deref().filter(|_| spec.id_pattern.is_none());
430 15946 : match &st.reg_mirror {
431 14 : Some(m) => Ok(m.matching(tenant.as_str(), ids, types.as_deref())),
432 15932 : None => Ok(st
433 15932 : .store
434 15932 : .matching_registrations(tenant, ids, types.as_deref())
435 15932 : .await?
436 15924 : .into_iter()
437 15924 : .map(std::sync::Arc::new)
438 15924 : .collect()),
439 : }
440 15946 : }
441 :
442 : /// Does one stored registration take part in this operation, and through
443 : /// which RegistrationInfos (5.12)? Every condition is decided from the
444 : /// borrowed document — expiry, csf, datasetId, location, intervals, the Via
445 : /// chain — so a caller that only needs the verdict ([`would_federate`])
446 : /// stops here instead of compiling a `FedReg` per registration. Expiry is
447 : /// filtered HERE and only here: the single yield point.
448 31366 : fn reg_candidate<'a>(
449 31366 : doc: &'a Value,
450 31366 : spec: &crate::registry::CsrSpec,
451 31366 : ctx: &Context,
452 31366 : seen: &[String],
453 31366 : ) -> Option<Vec<&'a Value>> {
454 31366 : if crate::registry::reg_expired(doc) {
455 0 : return None;
456 31366 : }
457 : // 5.7.2.4/5.7.4.4/5.6.21.4: a csf gates which Context Sources
458 : // are considered (evaluated over the registration's own
459 : // Context Source Properties, 5.10.2.4 semantics).
460 31366 : if let Some(csf) = &spec.csf {
461 6 : if !crate::registry::csf_matches(csf, doc, ctx) {
462 4 : return None;
463 2 : }
464 31360 : }
465 : // 5.12 datasetId condition (should-level): both sides specifying
466 : // datasetId match only with a value in common; one side alone always
467 : // matches.
468 31362 : if let Some(ds) = &spec.dataset_ids {
469 4 : if let Some(reg_ds) = doc.get("datasetId").and_then(Value::as_array) {
470 4 : if !reg_ds
471 4 : .iter()
472 4 : .filter_map(Value::as_str)
473 4 : .any(|d| ds.iter().any(|q| q == d))
474 : {
475 2 : return None;
476 2 : }
477 0 : }
478 31358 : }
479 : // 5.2.9 location + 4.3.6.1: a geo-scoped registration is only consulted
480 : // when the query's geo filter matches its geometry; a registration
481 : // without `location` is unconstrained.
482 31360 : if let Some(gq) = &spec.geo {
483 4 : if let Some(geom) = doc.get("location") {
484 0 : if !gq.matches_geometry(geom) {
485 0 : return None;
486 0 : }
487 4 : }
488 31356 : }
489 : // 5.2.9: a declared observation/management interval gates temporal
490 : // fan-out on overlap with the temporal query; without any declared
491 : // interval the registration is unconstrained.
492 31360 : if let Some(tq) = &spec.temporal {
493 16 : if (doc.get("observationInterval").is_some() || doc.get("managementInterval").is_some())
494 0 : && !crate::registry::temporal_interval_matches(doc, tq)
495 : {
496 0 : return None;
497 16 : }
498 31344 : }
499 : // Table 6.3.18-2 / 5.2.9: this source already handled the request.
500 31360 : if doc
501 31360 : .get("contextSourceAlias")
502 31360 : .and_then(Value::as_str)
503 31360 : .is_some_and(|a| seen.iter().any(|t| t == a))
504 : {
505 14 : return None;
506 31346 : }
507 31346 : let infos = crate::registry::matching_infos(spec, doc, ctx);
508 31346 : if infos.is_empty() {
509 30754 : None
510 : } else {
511 592 : Some(infos)
512 : }
513 31366 : }
514 :
515 : /// Registrations matching an entity spec (5.12), compiled for forwarding.
516 : ///
517 : /// Table 6.3.18-2 makes the inbound `Via` listing part of matching itself —
518 : /// "the listing of previously encountered Context Sources supplied is used
519 : /// when determining matching registrations" — so a registration whose
520 : /// `contextSourceAlias` is already in the chain is filtered out HERE, at the
521 : /// one place every read and write path resolves its candidates. Keeping it
522 : /// out of the call sites is deliberate: a loop check the callers own is a
523 : /// loop check some caller forgets.
524 15686 : pub async fn matching_regs(
525 15686 : st: &AppState,
526 15686 : tenant: &TenantId,
527 15686 : spec: &crate::registry::CsrSpec,
528 15686 : ctx: &Context,
529 15686 : headers: &HeaderMap,
530 15686 : ) -> Result<Vec<FedReg>, NgsiError> {
531 15686 : if via_hops(headers) > MAX_VIA_HOPS {
532 18 : return Ok(Vec::new());
533 15668 : }
534 15668 : let seen = via_tokens(headers);
535 15668 : let regs: Vec<FedReg> = reg_docs(st, tenant, spec, ctx)
536 15668 : .await?
537 15660 : .iter()
538 37110 : .filter_map(|doc| {
539 31334 : let infos = reg_candidate(doc, spec, ctx, &seen)?;
540 568 : let alias = doc
541 568 : .get("contextSourceAlias")
542 568 : .and_then(Value::as_str)
543 568 : .map(str::to_owned);
544 568 : let endpoint = doc
545 568 : .get("endpoint")
546 568 : .and_then(Value::as_str)?
547 568 : .trim_end_matches('/');
548 : // registrations may name the API root itself (…/ngsi-ld/v1) —
549 : // normalize so forward URLs never double the prefix (IOP fixtures)
550 568 : let endpoint = endpoint
551 568 : .strip_suffix("/ngsi-ld/v1")
552 568 : .unwrap_or(endpoint)
553 568 : .to_owned();
554 568 : let mode = doc
555 568 : .get("mode")
556 568 : .and_then(Value::as_str)
557 568 : .unwrap_or("inclusive")
558 568 : .to_owned();
559 568 : let ops = doc
560 568 : .get("operations")
561 568 : .and_then(Value::as_array)
562 568 : .map(|a| {
563 494 : a.iter()
564 494 : .filter_map(Value::as_str)
565 494 : .map(str::to_owned)
566 494 : .collect()
567 494 : })
568 568 : .unwrap_or_else(|| vec![DEFAULT_OPERATION_GROUP.into()]);
569 568 : let mut attrs: Option<Vec<String>> = Some(Vec::new());
570 568 : let mut ent_ids = Vec::new();
571 568 : let mut ent_types = Vec::new();
572 568 : let mut ent_patterns = Vec::new();
573 568 : let mut ent_unrestricted = false;
574 568 : for info in &infos {
575 568 : let props = info.get("propertyNames").and_then(Value::as_array);
576 568 : let rels = info.get("relationshipNames").and_then(Value::as_array);
577 568 : if props.is_none() && rels.is_none() {
578 548 : attrs = None; // an unscoped info covers everything
579 548 : } else if let Some(list) = &mut attrs {
580 20 : for src in [props, rels].into_iter().flatten() {
581 20 : list.extend(src.iter().filter_map(Value::as_str).map(str::to_owned));
582 20 : }
583 0 : }
584 568 : if let Some(es) = info.get("entities").and_then(Value::as_array) {
585 552 : for e in es {
586 552 : if let Some(i) = e.get("id").and_then(Value::as_str) {
587 198 : ent_ids.push(i.to_owned());
588 354 : }
589 552 : if let Some(p) = e.get("idPattern").and_then(Value::as_str) {
590 26 : ent_patterns.push(p.to_owned());
591 526 : }
592 : // 5.12 condition 1: neither id nor idPattern ⇒ the
593 : // element restricts by type only, never by id
594 552 : if e.get("id").is_none() && e.get("idPattern").is_none() {
595 328 : ent_unrestricted = true;
596 328 : }
597 : // 5.2.8: type may be a String or String[]
598 552 : match e.get("type") {
599 540 : Some(Value::String(t)) => ent_types.push(t.clone()),
600 0 : Some(Value::Array(ts)) => ent_types
601 0 : .extend(ts.iter().filter_map(Value::as_str).map(str::to_owned)),
602 12 : _ => {}
603 : }
604 : }
605 16 : } else {
606 16 : // an attributes-only RegistrationInfo imposes no id scope
607 16 : ent_unrestricted = true;
608 16 : }
609 : }
610 568 : let tenant = doc.get("tenant").and_then(Value::as_str).map(str::to_owned);
611 568 : let csi = csi_of(doc);
612 : Some(FedReg {
613 568 : reg_id: doc
614 568 : .get("id")
615 568 : .and_then(Value::as_str)
616 568 : .unwrap_or_default()
617 568 : .to_owned(),
618 568 : endpoint,
619 568 : mode,
620 568 : ops,
621 568 : attrs,
622 568 : ent_ids,
623 568 : ent_types,
624 568 : ent_patterns,
625 568 : ent_unrestricted,
626 568 : tenant,
627 568 : alias,
628 568 : csi,
629 568 : local_only: local_only_of(doc),
630 568 : timeout_ms: doc
631 568 : .get("management")
632 568 : .and_then(|m| m.get("timeout"))
633 568 : .and_then(Value::as_u64),
634 568 : cooldown_ms: doc
635 568 : .get("management")
636 568 : .and_then(|m| m.get("cooldown"))
637 568 : .and_then(Value::as_u64),
638 : })
639 31334 : })
640 15660 : .collect();
641 15660 : Ok(merge_same_source(regs))
642 15686 : }
643 :
644 : /// The API operations a registration is willing to answer, EXPANDED: 5.2.9
645 : /// lets `operations` name groups as well as operations, so `["federationOps"]`
646 : /// and the names it stands for are the same subset written two ways, and only
647 : /// the expansion can tell two registrations apart by what they will serve.
648 612 : fn op_set(r: &FedReg) -> std::collections::BTreeSet<&'static str> {
649 612 : antares_model::operations::OPERATION_NAMES
650 612 : .iter()
651 612 : .copied()
652 26316 : .filter(|op| r.supports(op))
653 612 : .collect()
654 612 : }
655 :
656 : /// Registrations naming the same Context Source (same endpoint, mode,
657 : /// tenant, contextSourceAlias, contextSourceInfo, localOnly) fold into ONE
658 : /// forwarded request (5.2.9: the alias identifies a source, so a different
659 : /// alias is a different source even behind one endpoint):
660 : /// attribute and entity scopes union (an unscoped one covers everything),
661 : /// the first registration's id and timing stay. Two calls to one source for
662 : /// one query would return the same data twice.
663 : ///
664 : /// Registrations that declare DIFFERENT operations are not the same source
665 : /// for this purpose. 4.3.6.1: a source "may indicate that they are only
666 : /// willing to respond to a limited subset of API operations. Context Brokers
667 : /// shall respect this" — and the fold unions the entity and attribute scope,
668 : /// so an operation only one of them declared would then travel for the
669 : /// other's Entities. They stay separate instead, each filtered by `supports`
670 : /// on its own. The cost is one extra request to a source that registered
671 : /// itself twice with different operation lists and both lists cover the
672 : /// operation at hand; 4.5.5 merges the two answers by Entity id, so the
673 : /// caller sees the same data either way.
674 15680 : fn merge_same_source(regs: Vec<FedReg>) -> Vec<FedReg> {
675 15680 : let mut out: Vec<(FedReg, std::collections::BTreeSet<&'static str>)> = Vec::new();
676 15680 : for r in regs {
677 612 : let ops = op_set(&r);
678 612 : let same = out.iter_mut().find(|(o, o_ops)| {
679 148 : o.endpoint == r.endpoint
680 52 : && o.mode == r.mode
681 44 : && o.tenant == r.tenant
682 44 : && o.alias == r.alias
683 28 : && o.csi == r.csi
684 28 : && o.local_only == r.local_only
685 24 : && *o_ops == ops
686 148 : });
687 612 : let Some((o, _)) = same else {
688 592 : out.push((r, ops));
689 592 : continue;
690 : };
691 20 : o.attrs = match (o.attrs.take(), r.attrs) {
692 4 : (Some(mut a), Some(b)) => {
693 8 : for x in b {
694 8 : if !a.contains(&x) {
695 4 : a.push(x);
696 4 : }
697 : }
698 4 : Some(a)
699 : }
700 16 : _ => None,
701 : };
702 20 : o.ent_ids.extend(r.ent_ids);
703 20 : o.ent_types.extend(r.ent_types);
704 20 : o.ent_patterns.extend(r.ent_patterns);
705 20 : o.ent_unrestricted |= r.ent_unrestricted;
706 : }
707 15680 : out.into_iter().map(|(r, _)| r).collect()
708 15680 : }
709 :
710 : /// contextSourceInfo keys the forward must NOT copy into headers: the tenant
711 : /// travels via the registration's own `tenant` member (4.3.6.5 "shall not be
712 : /// part of contextSourceInfo"), connection/binding-managed headers cannot be
713 : /// overridden ("shall be ignored"), and the 4.3.6.6 processed keys (accept,
714 : /// contentType, jsonldContext, ngsildConformance) are TRANSFORMED by
715 : /// `forward` rather than passed through raw, which would corrupt
716 : /// negotiation instead.
717 : const CSI_SKIP: &[&str] = &[
718 : "ngsild-tenant",
719 : "content-length",
720 : "content-type",
721 : "host",
722 : "via",
723 : "link",
724 : "connection",
725 : "accept",
726 : "contenttype",
727 : "jsonldcontext",
728 : "ngsildconformance",
729 : ];
730 :
731 : /// Bounds wall on the forwarded-read path: a peer response larger than
732 : /// ANTARES_MAX_FED_RESPONSE_BYTES is never held in memory — the part fails
733 : /// exactly like an unparseable payload (Table 6.3.17-1, warning 111 via the
734 : /// 2xx-with-null-body arm of `read_warning`).
735 388 : async fn read_body_capped(resp: reqwest::Response) -> Value {
736 388 : let cap = *crate::bounds::MAX_FED_RESPONSE_BYTES;
737 388 : if resp.content_length().is_some_and(|l| l > cap as u64) {
738 2 : tracing::warn!("federation response over the {cap}-byte cap (declared length), skipped");
739 2 : return Value::Null;
740 386 : }
741 : #[cfg(not(target_arch = "wasm32"))]
742 384 : let bytes = {
743 386 : let mut resp = resp;
744 386 : let mut buf: Vec<u8> = Vec::new();
745 : loop {
746 1038 : match resp.chunk().await {
747 654 : Ok(Some(c)) => {
748 654 : if buf.len() + c.len() > cap {
749 2 : tracing::warn!("federation response over the {cap}-byte cap, skipped");
750 2 : return Value::Null;
751 652 : }
752 652 : buf.extend_from_slice(&c);
753 : }
754 384 : Ok(None) => break,
755 0 : Err(_) => return Value::Null,
756 : }
757 : }
758 384 : buf
759 : };
760 : // the browser fetch API hands the body over whole — cap after the read
761 : #[cfg(target_arch = "wasm32")]
762 : let bytes = match resp.bytes().await {
763 : Ok(b) if b.len() <= cap => b.to_vec(),
764 : _ => return Value::Null,
765 : };
766 384 : serde_json::from_slice(&bytes).unwrap_or(Value::Null)
767 388 : }
768 :
769 : /// 5.7.2.4: "If split entities flag is explicitly set to true or, if not
770 : /// explicitly set, the default setting of the deployment allows split
771 : /// entities" — this deployment's default is OFF, so only the explicit flag
772 : /// engages the split branch.
773 1168 : pub(crate) fn split_entities(params: &HashMap<String, String>) -> bool {
774 1168 : params.get("splitEntities").map(String::as_str) == Some("true")
775 1168 : }
776 :
777 : /// 4.3.6.6: a registration carrying a jsonldContext contextSourceInfo key —
778 : /// forwards to it are recompacted term-by-term (attrs/type/geoproperty only).
779 178 : fn has_reg_context(reg: &FedReg) -> bool {
780 178 : reg.csi
781 178 : .iter()
782 178 : .any(|(k, _)| k.eq_ignore_ascii_case("jsonldContext"))
783 178 : }
784 :
785 : /// 4.3.6.1 fan-out: forwards to matching registrations run concurrently —
786 : /// the clause fixes the merge order (4.5.5 non-aux before aux), never the
787 : /// request order, and cross-source result ordering does not exist (the
788 : /// `ordering` parameter is a 400 outside local scope, 5.7.2.4). Results
789 : /// return in registration order so warning and merge processing stay
790 : /// deterministic. Concurrency per request is bounded by
791 : /// bounds::MAX_FED_FANOUT.
792 2142 : async fn fan_out<I, T, F, Fut>(items: Vec<I>, per_item: F) -> Vec<T>
793 2142 : where
794 2142 : F: FnMut(I) -> Fut,
795 2142 : Fut: std::future::Future<Output = T>,
796 2142 : {
797 : use futures_util::StreamExt;
798 2142 : futures_util::stream::iter(items.into_iter().map(per_item))
799 2142 : .buffered(*crate::bounds::MAX_FED_FANOUT)
800 2142 : .collect()
801 2142 : .await
802 2142 : }
803 :
804 : /// 6.6 Entity Attribute and 6.7 Entity Attribute Instance name one Attribute
805 : /// in the PATH rather than in the payload. The range is the encoded segment
806 : /// inside `url`, so a translated name can be written back over it; the name
807 : /// comes back percent-decoded, which is the form an `@context` expands. The
808 : /// segment stops at the next `/`, so `/attrs/{name}/value` and
809 : /// `/attrs/{name}/{instanceId}` both name `{name}` and nothing more.
810 52 : fn path_attr_segment(url: &str) -> Option<(std::ops::Range<usize>, String)> {
811 52 : let path_end = url.find('?').unwrap_or(url.len());
812 52 : let at = url[..path_end].find("/attrs/")? + "/attrs/".len();
813 38 : let rest = &url[at..path_end];
814 38 : let end = at + rest.find('/').unwrap_or(rest.len());
815 38 : (end > at).then(|| {
816 30 : (
817 30 : at..end,
818 30 : crate::negotiate::percent_decode(&url.as_bytes()[at..end]),
819 30 : )
820 30 : })
821 52 : }
822 :
823 : /// Table 5.2.9-2 and 5.2.34: one forward's outcome, booked on the
824 : /// registration it was made for. `ok` is the table's own failure
825 : /// definition — "an HTTP response code other than 2xx".
826 : ///
827 : /// Only an operation that reached the wire is booked. The three ways out of
828 : /// `forward` above this point — a destination the egress policy refuses, an
829 : /// open breaker, the declared 5.2.34 cooldown — are this broker declining to
830 : /// perform the operation, and the table counts operations that were
831 : /// performed: `lastFailure` is when a failure "was returned". Nothing is
832 : /// lost by that: the breaker opens only after failures that WERE attempted
833 : /// and booked, so a registration whose Context Source is down already reads
834 : /// `"failed"` before the first forward is suppressed.
835 : ///
836 : /// The stamp never fails the forward. The answer is decided by the time this
837 : /// runs, and a store that cannot take a counter must not turn a served
838 : /// response into an error. The registration mirror is not refreshed either:
839 : /// it exists to decide which registrations match, and no member here does.
840 417 : async fn note_forward(st: &AppState, tenant: &TenantId, reg: &FedReg, ok: bool) {
841 417 : if reg.cooldown_ms.is_some() {
842 0 : st.reg_cooldown_stamp(tenant, ®.reg_id, ok);
843 417 : }
844 417 : if let Err(e) = st
845 417 : .store
846 417 : .record_forward(tenant, ®.reg_id, &crate::state::now_iso(), ok)
847 417 : .await
848 : {
849 0 : tracing::warn!(
850 : "forward bookkeeping for registration {} failed: {e}",
851 : reg.reg_id
852 : );
853 417 : }
854 417 : }
855 :
856 : /// One forwarded request. `body` is compacted JSON (no @context member).
857 : #[allow(clippy::too_many_arguments)] // mirrors the wire: one param per forwarded request part
858 421 : pub async fn forward(
859 421 : st: &AppState,
860 421 : method: reqwest::Method,
861 421 : mut url: String,
862 421 : query: &[(String, String)],
863 421 : headers: &HeaderMap,
864 421 : tenant: &TenantId,
865 421 : reg: &FedReg,
866 421 : ctx_url: &str,
867 421 : mut body: Option<Value>,
868 421 : ) -> (u16, Value, Vec<String>) {
869 : // 6.3.17: NGSILD-Warning values received from the peer are returned to
870 : // the caller — abnormal behaviour detected downstream in a cascade
871 : // (4.3.6.4) must surface on the aggregated response, not vanish here.
872 : // Process-wide slot first (bounds::MAX_FED_INFLIGHT): the buffers and
873 : // connections below exist only while a slot is held.
874 421 : let _slot = crate::bounds::FED_INFLIGHT.acquire().await;
875 : // One policy for every outbound class — scheme allowlist,
876 : // private-range deny, per-destination circuit breaker.
877 421 : if let Err(e) = st.egress.check_url(&url).await {
878 : // 5.2.9 allows any URI as the registered endpoint, and a URI's
879 : // authority may carry credentials reqwest sends as basic auth, so a
880 : // peer URL loses its userinfo on the way into the log.
881 0 : tracing::warn!(
882 : "federation forward to {} refused: {e}",
883 0 : antares_notifier::redact_userinfo(&url)
884 : );
885 0 : return (502, Value::Null, Vec::new());
886 421 : }
887 421 : if st.egress.is_open(tenant.as_str(), &url) {
888 0 : tracing::debug!(
889 : "federation forward to {} short-circuited (breaker open)",
890 0 : antares_notifier::redact_userinfo(&url)
891 : );
892 0 : return (503, Value::Null, Vec::new());
893 421 : }
894 : // 5.2.34 cooldown (per REGISTRATION, distinct from the host:port
895 : // breaker): inside the declared window "a timeout error response for
896 : // the registration is automatically returned" — the source is not
897 : // contacted.
898 421 : if let Some(cd) = reg.cooldown_ms {
899 0 : if st
900 0 : .egress
901 0 : .reg_in_cooldown(&crate::egress::reg_key(tenant.as_str(), ®.reg_id), cd)
902 : {
903 0 : return (504, Value::Null, Vec::new());
904 0 : }
905 421 : }
906 : // 4.3.6.6: the four contextSourceInfo keys with processing
907 : // semantics. Values were validated at registration time (5.9.2).
908 1398 : let csi_get = |key: &str| {
909 1398 : reg.csi
910 1398 : .iter()
911 1398 : .find(|(k, _)| k.eq_ignore_ascii_case(key))
912 1398 : .map(|(_, v)| v.as_str())
913 1398 : };
914 : // "accept": the response shall come back in this format (the read path
915 : // strips any body @context before expanding, so both forms import).
916 421 : let accept = csi_get("accept").unwrap_or("application/json");
917 : // "ngsildConformance": amend the payload to the pinned version (4.3.6.8).
918 421 : if let Some(ver) = csi_get("ngsildConformance").and_then(crate::conformance::parse_version) {
919 0 : if let Some(b) = body.as_mut() {
920 0 : crate::conformance::amend_payload(b, ver);
921 0 : }
922 421 : }
923 : // "jsonldContext": recompact payload and term-bearing query parameters
924 : // with the registered context, forward THAT context, Content-Type
925 : // application/json, no @context member in the payload. Entity-shaped
926 : // bodies only — a POST-query body has no entity terms to recompact; if
927 : // either context fails to load the forward degrades to the original
928 : // context rather than sending terms compacted against the wrong one.
929 421 : let mut link_ctx: String = ctx_url.to_owned();
930 421 : let mut query: Vec<(String, String)> = query.to_vec();
931 : // 4.3.6.4: "a binding-specific mechanism to request operations only on
932 : // the registered endpoint itself" — a localOnly registration (5.2.9)
933 : // must not cascade, so the forward carries the 6.3.18 local parameter.
934 421 : if reg.local_only && !query.iter().any(|(k, _)| k == "local") {
935 2 : query.push(("local".into(), "true".into()));
936 419 : }
937 421 : if let Some(reg_ctx_url) = csi_get("jsonldContext") {
938 : // 5.5.10: both URLs are Tenant data — the caller's own @context and one
939 : // the Registration names — so they resolve within the forwarding
940 : // Tenant, never against a Hosted @context another Tenant stored.
941 12 : let orig = st
942 12 : .loader
943 12 : .resolve_quiet_for(tenant, &Value::String(ctx_url.to_owned()))
944 12 : .await;
945 12 : let target = st
946 12 : .loader
947 12 : .resolve_quiet_for(tenant, &Value::String(reg_ctx_url.to_owned()))
948 12 : .await;
949 12 : if let (Ok(orig), Ok(target)) = (orig, target) {
950 : // The payload reaching here has already been validated by the
951 : // operation's own handler; this expansion only re-reads it to
952 : // translate its terms, so it must accept everything that handler
953 : // accepted and drop nothing. A merge fragment carries NGSI-LD
954 : // Nulls (5.5.12), a temporal payload repeats a datasetId across
955 : // instances (4.5.6), and a 4.5.7 tombstone is a deletedAt an
956 : // expansion without `sys` discards — re-validating or thinning
957 : // the body here changes the write the client asked for.
958 12 : let mut translated: Option<Value> = None;
959 12 : let mut can_switch = true;
960 : // 6.6/6.7 name their Attribute in the PATH, which the Context
961 : // Source expands with the @context the forward advertises — so
962 : // the segment is translated with the payload or switching the
963 : // context renames the target Attribute. 4.3.6.6 compacts
964 : // "payload and query parameters" and the path is neither, but it
965 : // carries a term the same compaction has to reach, or the rule
966 : // cannot be applied to these resources at all. A name that does
967 : // not expand to an absolute IRI is no term to translate, and the
968 : // whole request stays in the @context its names are already in.
969 12 : let path_swap = match path_attr_segment(&url) {
970 6 : Some((range, name)) => match antares_jsonld::expand_attr_name(&name, &orig) {
971 : // The compacted term is written into the request PATH, so
972 : // it is held to the rule the client's own name was held to
973 : // at the door (`antares_model::check_attr_name`): a `.` or
974 : // `..` segment names a different resource of the Context
975 : // Source endpoint, and a URL parser resolves it before the
976 : // request is sent. A registered @context is client-supplied
977 : // and may bind any term, so a term that is a dot segment
978 : // means the request cannot be expressed in that context.
979 6 : Ok(iri) => {
980 6 : let term = target.compact_iri(&iri);
981 6 : if antares_model::has_dot_segment(&term) {
982 2 : can_switch = false;
983 2 : None
984 : } else {
985 4 : Some((range, path_segment(&term)))
986 : }
987 : }
988 : Err(_) => {
989 0 : can_switch = false;
990 0 : None
991 : }
992 : },
993 6 : None => None,
994 : };
995 12 : match body.as_ref() {
996 : // nothing to translate
997 6 : None => {}
998 : // 6.6/6.7 take an Attribute Fragment (5.6.4, 5.6.14, 5.6.19),
999 : // not an Entity — its `type` is an Attribute type and its
1000 : // `value` a value, so reading it as an Entity would turn the
1001 : // value into a sub-Attribute. `expand_attr_fragment` is the
1002 : // reader the partial-update handler itself uses, and
1003 : // `compact_instance` its inverse.
1004 6 : Some(v) if path_swap.is_some() => match v.as_object() {
1005 2 : Some(o) => match antares_jsonld::expand_attr_fragment(o, &orig) {
1006 2 : Ok(exp) => {
1007 2 : let mut re = antares_jsonld::compact::compact_instance(&exp, &target);
1008 2 : if let Some(m) = re.as_object_mut() {
1009 2 : m.remove("@context");
1010 2 : }
1011 2 : translated = Some(re);
1012 : }
1013 0 : Err(_) => can_switch = false,
1014 : },
1015 0 : None => can_switch = false,
1016 : },
1017 4 : Some(v) => match v.as_object() {
1018 : // a 5.2.23 Query body carries no entity terms to recompact
1019 4 : Some(o) if o.get("type").and_then(Value::as_str) == Some("Query") => {}
1020 4 : Some(o) => {
1021 4 : let opts = antares_jsonld::ExpandOpts {
1022 4 : fragment: o.get("id").is_none(),
1023 4 : allow_null: true,
1024 4 : merge: true,
1025 4 : temporal: true,
1026 4 : sys: true,
1027 4 : };
1028 4 : match antares_jsonld::expand_entity(o, &orig, opts) {
1029 4 : Ok(exp) => {
1030 4 : let mut re = antares_jsonld::compact::compact_entity(&exp, &target);
1031 4 : if let Some(m) = re.as_object_mut() {
1032 4 : m.remove("@context");
1033 4 : }
1034 4 : translated = Some(re);
1035 : }
1036 0 : Err(_) => can_switch = false,
1037 : }
1038 : }
1039 : // any other shape (a batch array) has no entity-document
1040 : // translation here, so it must not claim to be in the
1041 : // registered @context either
1042 0 : None => can_switch = false,
1043 : },
1044 : }
1045 : // 4.3.6.6 states one rule, not two: compact the payload with the
1046 : // registered @context "and forward with this JSON-LD context". A
1047 : // payload that could not be translated must therefore travel in
1048 : // the @context its terms ARE in — advertising the registered one
1049 : // over untranslated terms makes the Context Source expand them to
1050 : // different Fully Qualified Names (5.5.7) and write Attributes
1051 : // the client never named.
1052 12 : if can_switch {
1053 10 : if let Some((range, seg)) = path_swap {
1054 4 : url.replace_range(range, &seg);
1055 6 : }
1056 10 : if let Some(re) = translated {
1057 6 : body = Some(re);
1058 6 : }
1059 10 : for (k, v) in query.iter_mut() {
1060 4 : if matches!(k.as_str(), "attrs" | "type" | "geoproperty") {
1061 2 : *v = v
1062 2 : .split(',')
1063 2 : .map(|t| target.compact_iri(&orig.expand_key(t.trim())))
1064 2 : .collect::<Vec<_>>()
1065 2 : .join(",");
1066 2 : }
1067 : }
1068 10 : link_ctx = reg_ctx_url.to_owned();
1069 : } else {
1070 2 : tracing::warn!(
1071 : "payload could not be expressed in the registered jsonldContext \
1072 : {reg_ctx_url}; forwarding with the original context"
1073 : );
1074 : }
1075 : } else {
1076 0 : tracing::warn!(
1077 : "registered jsonldContext {reg_ctx_url} (or the request context) \
1078 : failed to load; forwarding with the original context"
1079 : );
1080 : }
1081 409 : }
1082 : // An inline @context (serialized JSON from ctx_link_url) cannot travel
1083 : // as a Link header — it is embedded in the body below (5.5.7/6.3.5).
1084 421 : let inline_ctx: Option<Value> = if link_ctx.starts_with('[') || link_ctx.starts_with('{') {
1085 2 : serde_json::from_str(&link_ctx).ok()
1086 : } else {
1087 419 : None
1088 : };
1089 421 : let mut req = st
1090 421 : .fed_http
1091 421 : .request(method, &url)
1092 421 : .header("Accept", accept)
1093 421 : .header(
1094 : "Via",
1095 421 : outbound_via(headers, &alias_for(&st.host_alias, tenant)),
1096 : );
1097 421 : if inline_ctx.is_none() {
1098 419 : req = req.header(
1099 419 : "Link",
1100 419 : format!("<{link_ctx}>; rel=\"http://www.w3.org/ns/json-ld#context\"; type=\"application/ld+json\""),
1101 419 : );
1102 419 : }
1103 421 : if !query.is_empty() {
1104 236 : req = req.query(&query);
1105 262 : }
1106 : // 4.14: "the Tenant information from the Context Source Registration has
1107 : // to be used. If no Tenant information is present in the Context Source
1108 : // Registration, no Tenant information is to be used and thus the default
1109 : // Tenant is targeted" — the requesting tenant never flows through; 6.3.14
1110 : // omits the header for the default Tenant.
1111 421 : if let Some(peer_tenant) = reg.tenant.as_deref().filter(|t| *t != "default") {
1112 8 : req = req.header("NGSILD-Tenant", peer_tenant);
1113 413 : }
1114 : // 4.3.6.5 contextSourceInfo ⇒ extra headers; the special value
1115 : // "urn:ngsi-ld:request" copies the header from the triggering request
1116 : // (dropped when the triggering request did not carry it).
1117 421 : for (k, v) in ®.csi {
1118 26 : let lower = k.to_ascii_lowercase();
1119 26 : if CSI_SKIP.contains(&lower.as_str()) {
1120 14 : continue;
1121 12 : }
1122 : // ADR-0020: the subject never leaves this process. A registration is
1123 : // client-supplied, so without this a registration that names the
1124 : // subject header with "urn:ngsi-ld:request" would copy the identity
1125 : // of every request that fans out to it onto the wire — the peer
1126 : // would be handed a credential the broker was only ever given to
1127 : // ask its own engine about. Dropped rather than refused at 5.9.2:
1128 : // which headers a deployment made subject headers is not something
1129 : // a client gets to probe by watching which registrations fail.
1130 12 : if crate::policy::SUBJECT_HEADERS.contains(&lower) {
1131 2 : tracing::warn!(
1132 : "registration {} asks for the policy subject header {k:?}; not forwarded",
1133 : reg.reg_id
1134 : );
1135 2 : continue;
1136 10 : }
1137 10 : let val = if v == "urn:ngsi-ld:request" {
1138 4 : match headers.get(k.as_str()).and_then(|h| h.to_str().ok()) {
1139 2 : Some(h) => h.to_owned(),
1140 2 : None => continue,
1141 : }
1142 : } else {
1143 6 : v.clone()
1144 : };
1145 8 : req = req.header(k.as_str(), val);
1146 : }
1147 421 : if let Some(mut b) = body {
1148 : // 4.3.6.6 "contentType": provide request + @context as the MIME type
1149 : // requires — ld+json carries the context inline. When "jsonldContext"
1150 : // is also defined its own mandate wins ("the Content-Type of the
1151 : // forwarded request shall be application/json").
1152 135 : let want_ld = csi_get("contentType") == Some("application/ld+json")
1153 0 : && csi_get("jsonldContext").is_none();
1154 135 : if let Some(ic) = &inline_ctx {
1155 : // inline request @context: the only lossless carrier is the
1156 : // body itself, as application/ld+json (5.5.7/6.3.5).
1157 2 : if let Some(o) = b.as_object_mut() {
1158 2 : o.insert("@context".into(), ic.clone());
1159 2 : }
1160 2 : req = req.header("Content-Type", "application/ld+json");
1161 133 : } else if want_ld {
1162 0 : if let Some(o) = b.as_object_mut() {
1163 0 : o.insert("@context".into(), Value::String(link_ctx.clone()));
1164 0 : }
1165 0 : req = req.header("Content-Type", "application/ld+json");
1166 133 : } else {
1167 133 : req = req.header("Content-Type", "application/json");
1168 133 : }
1169 135 : req = req.body(serde_json::to_vec(&b).unwrap_or_default());
1170 286 : }
1171 : // The whole HTTP interaction is one Send unit (http_interaction) so
1172 : // the handler futures above stay Send on wasm32 too.
1173 421 : antares_jsonld::http_interaction(async {
1174 : // wasm has no client-level timeout — bound the forward per request
1175 : // (mirrors the native fed_http 8 s total); a timed-out
1176 : // forward is the only failure class that feeds the breaker.
1177 : // 5.2.34 timeout bounds the forward below the 8 s ceiling.
1178 : // Natively io_deadline is a passthrough (the client owns the 8 s
1179 : // default), so the per-registration budget rides on the request.
1180 421 : let ceiling = 8_000 * crate::state::slow_factor();
1181 421 : let deadline: u32 = reg.timeout_ms.map_or(ceiling, |t| t.min(ceiling)) as u32;
1182 : #[cfg(not(target_arch = "wasm32"))]
1183 421 : let req = req.timeout(std::time::Duration::from_millis(deadline as u64));
1184 421 : let sent = antares_jsonld::io_deadline(req.send(), deadline).await;
1185 417 : let sent = match sent {
1186 417 : Some(r) => r,
1187 : None => {
1188 0 : st.egress.record_failure(tenant.as_str(), &url);
1189 0 : note_forward(st, tenant, reg, false).await;
1190 0 : return (504, Value::Null, Vec::new());
1191 : }
1192 : };
1193 29 : match sent {
1194 388 : Ok(resp) => {
1195 : // Any response — even 5xx — proves the peer answers within
1196 : // its own response time: no deadline cost, so no breaker
1197 : // state. Only TIMEOUT-class failures trip; a
1198 : // responding-but-erroring peer must keep being attempted,
1199 : // else unrelated registrations sharing its host:port starve.
1200 388 : let status = resp.status().as_u16();
1201 388 : st.egress.record_success(tenant.as_str(), &url);
1202 388 : note_forward(st, tenant, reg, (200..300).contains(&status)).await;
1203 : // 6.3.17: the peer's own values travel on (4.3.6.4), but
1204 : // only up to the cap — the list is written by the peer and
1205 : // sent by this broker, and past the cap one source would
1206 : // both outgrow the fan-out and crowd out the warnings the
1207 : // clause obliges this broker to raise about the others.
1208 : // The head is kept: a source states its own outcome before
1209 : // the ones it relays.
1210 388 : let all = resp.headers().get_all("NGSILD-Warning");
1211 388 : let peer_warnings: Vec<String> = all
1212 388 : .iter()
1213 388 : .filter_map(|v| v.to_str().ok().map(str::to_owned))
1214 388 : .take(crate::bounds::MAX_PEER_WARNINGS)
1215 388 : .collect();
1216 388 : if all.iter().count() > peer_warnings.len() {
1217 4 : tracing::warn!(
1218 : "registration {} sent more than {} NGSILD-Warning values; the rest were dropped",
1219 : reg.reg_id,
1220 : crate::bounds::MAX_PEER_WARNINGS
1221 : );
1222 384 : }
1223 388 : let body = read_body_capped(resp).await;
1224 388 : (status, body, peer_warnings)
1225 : }
1226 29 : Err(e) if e.is_timeout() => {
1227 2 : st.egress.record_failure(tenant.as_str(), &url);
1228 2 : note_forward(st, tenant, reg, false).await;
1229 2 : (504, Value::Null, Vec::new())
1230 : }
1231 : Err(_) => {
1232 : // connect refused/reset: fails in milliseconds — no deadline cost;
1233 : // clearing avoids stale suppression of a restarted peer.
1234 : // 503 (not 502): NO HTTP response was received, so the read
1235 : // path classifies it under Table 6.3.17-1 code 199 ("No
1236 : // response was received from the registration endpoint"),
1237 : // never 299 ("An error response ... was received").
1238 27 : st.egress.record_success(tenant.as_str(), &url);
1239 27 : note_forward(st, tenant, reg, false).await;
1240 27 : (503, Value::Null, Vec::new())
1241 : }
1242 : }
1243 417 : })
1244 421 : .await
1245 417 : }
1246 :
1247 : /// 6.3.17: a forwarded response contributes the NGSILD-Warning values the
1248 : /// peer sent AND, for an abnormal outcome, one warning of this broker's own
1249 : /// — whether or not the payload can be used. Only a 2xx payload can be:
1250 : /// 4.3.6.4 makes a failed forward a warning on a successful overall
1251 : /// response, never a failure of it, so a non-2xx registration is dropped
1252 : /// from the answer after it has been reported.
1253 200 : fn usable_payload(
1254 200 : st: &AppState,
1255 200 : tenant: &TenantId,
1256 200 : status: u16,
1257 200 : body: &Value,
1258 200 : peer_warns: Vec<String>,
1259 200 : warnings: &mut Vec<String>,
1260 200 : ) -> bool {
1261 200 : warnings.extend(peer_warns);
1262 200 : if let Some((code, text)) = read_warning(status, body) {
1263 18 : warnings.push(warning(code, &alias_for(&st.host_alias, tenant), text));
1264 182 : }
1265 200 : (200..300).contains(&status)
1266 200 : }
1267 :
1268 : /// 5.2.9 `attrs`: a registration's scope narrows what a forwarded read asks
1269 : /// for. The names go out compacted against the request @context, because the
1270 : /// peer is asked in terms, not in IRIs.
1271 24 : fn scope_attrs(reg: &FedReg, ctx: &Context) -> Option<(String, String)> {
1272 24 : let scope = reg.attrs.as_ref()?;
1273 2 : let names: Vec<String> = scope.iter().map(|a| ctx.compact_iri(a)).collect();
1274 2 : Some(("attrs".into(), names.join(",")))
1275 24 : }
1276 :
1277 : /// The 4.11 temporal window a forwarded temporal read carries (5.7.3.4 and
1278 : /// 5.7.4.4 ask the peer over the SAME window), plus sysAttrs on every
1279 : /// forwarded read: conflicting instances resolve by most recent
1280 : /// observedAt/modifiedAt (4.5.5.3), and without the remote modifiedAt the
1281 : /// winner would be arrival order.
1282 12 : fn temporal_window(params: &HashMap<String, String>) -> Vec<(String, String)> {
1283 12 : let mut query: Vec<(String, String)> = vec![("options".into(), "sysAttrs".into())];
1284 60 : for k in ["timerel", "timeAt", "endTimeAt", "timeproperty", "lastN"] {
1285 60 : if let Some(v) = params.get(k) {
1286 20 : query.push((k.into(), v.clone()));
1287 40 : }
1288 : }
1289 12 : query
1290 12 : }
1291 :
1292 : /// 5.2.9 `attrs` again, on the way back: the scope narrows ATTRIBUTES, so the
1293 : /// entity-level members of 4.5.1 stay whatever the scope says — dropping
1294 : /// them would let 4.5.5.3 read a missing `expiresAt` as "absent from a
1295 : /// received version" and delete the local one.
1296 482 : fn narrow_to_scope(expanded: Value, reg: &FedReg) -> Option<Value> {
1297 482 : let Some(scope) = ®.attrs else {
1298 476 : return Some(expanded);
1299 : };
1300 6 : let mut out = Map::new();
1301 26 : for (k, v) in expanded.as_object()? {
1302 26 : if crate::repr::ENTITY_META.contains(&k.as_str()) || scope.iter().any(|s| s == k) {
1303 22 : out.insert(k.clone(), v.clone());
1304 22 : }
1305 : }
1306 6 : Some(Value::Object(out))
1307 482 : }
1308 :
1309 : /// Expand + registration-scope-filter one remote compacted entity.
1310 470 : pub fn import_entity(remote: &Value, reg: &FedReg, ctx: &Context) -> Option<Value> {
1311 470 : let mut obj = remote.as_object()?.clone();
1312 470 : obj.remove("@context");
1313 470 : let expanded = antares_jsonld::expand_entity(
1314 470 : &obj,
1315 470 : ctx,
1316 470 : antares_jsonld::ExpandOpts {
1317 470 : sys: true,
1318 470 : ..Default::default()
1319 470 : },
1320 : )
1321 470 : .ok()?;
1322 470 : narrow_to_scope(expanded, reg)
1323 470 : }
1324 :
1325 : /// Expand + registration-scope-filter one remote temporal entity (5.7.3.4).
1326 : /// Instances of one datasetId legally repeat in a Temporal Evolution, so
1327 : /// expansion runs in temporal mode.
1328 12 : fn import_temporal(remote: &Value, reg: &FedReg, ctx: &Context) -> Option<Value> {
1329 12 : let mut obj = remote.as_object()?.clone();
1330 12 : obj.remove("@context");
1331 12 : let expanded = antares_jsonld::expand_entity(
1332 12 : &obj,
1333 12 : ctx,
1334 12 : antares_jsonld::ExpandOpts {
1335 12 : sys: true,
1336 12 : temporal: true,
1337 12 : // 4.5.7: deletion instances (value urn:ngsi-ld:null +
1338 12 : // deletedAt) are part of a Temporal Evolution — a remote
1339 12 : // tombstone must import, not be dropped as an invalid payload.
1340 12 : allow_null: true,
1341 12 : ..Default::default()
1342 12 : },
1343 : )
1344 12 : .ok()?;
1345 12 : narrow_to_scope(expanded, reg)
1346 12 : }
1347 :
1348 : /// 5.7.3.4: forward Retrieve Temporal Evolution to matching registrations
1349 : /// that support the retrieveTemporal operation; registrations without it
1350 : /// are not contacted. Returns (auxiliary, expanded doc) pairs for the
1351 : /// caller's 4.5.5 merge.
1352 : /// 4.20: does a raw registration document support `op`? Same group tables
1353 : /// as FedReg::supports; default when the operations member is absent is
1354 : /// federationOps (5.2.9).
1355 87 : pub(crate) fn doc_supports(reg: &Value, op: &str) -> bool {
1356 87 : fed_reg_of(reg.get("id").and_then(Value::as_str).unwrap_or(""), reg).supports(op)
1357 87 : }
1358 :
1359 : /// 5.2.34 `management.localOnly`; the top-level spelling is kept for
1360 : /// compatibility (4.3.6.4 wording / older payloads). Read here so every
1361 : /// FedReg of one document agrees on it — a view that reads it and a view
1362 : /// that does not send the same peer two different questions.
1363 764 : fn local_only_of(reg: &Value) -> bool {
1364 764 : reg.get("localOnly")
1365 764 : .and_then(Value::as_bool)
1366 764 : .or_else(|| {
1367 756 : reg.get("management")
1368 756 : .and_then(|m| m.get("localOnly"))
1369 756 : .and_then(Value::as_bool)
1370 756 : })
1371 764 : .unwrap_or(false)
1372 764 : }
1373 :
1374 : /// A minimal FedReg view of a raw registration document — enough for
1375 : /// `forward` (endpoint/tenant/csi/alias) and `supports`.
1376 : /// 4.3.6.6 contextSourceInfo, as the key/value pairs a forward applies.
1377 764 : fn csi_of(reg: &Value) -> Vec<(String, String)> {
1378 764 : reg.get("contextSourceInfo")
1379 764 : .and_then(Value::as_array)
1380 764 : .map(|a| {
1381 28 : a.iter()
1382 34 : .filter_map(|kv| {
1383 : Some((
1384 34 : kv.get("key")?.as_str()?.to_owned(),
1385 34 : kv.get("value")?.as_str()?.to_owned(),
1386 : ))
1387 34 : })
1388 28 : .collect()
1389 28 : })
1390 764 : .unwrap_or_default()
1391 764 : }
1392 :
1393 196 : pub(crate) fn fed_reg_of(reg_id: &str, reg: &Value) -> FedReg {
1394 196 : let endpoint = reg
1395 196 : .get("endpoint")
1396 196 : .and_then(Value::as_str)
1397 196 : .map(|e| {
1398 196 : let e = e.trim_end_matches('/');
1399 196 : e.strip_suffix("/ngsi-ld/v1").unwrap_or(e).to_owned()
1400 196 : })
1401 196 : .unwrap_or_default();
1402 : FedReg {
1403 196 : reg_id: reg_id.to_owned(),
1404 196 : endpoint,
1405 196 : mode: reg
1406 196 : .get("mode")
1407 196 : .and_then(Value::as_str)
1408 196 : .unwrap_or("inclusive")
1409 196 : .to_owned(),
1410 196 : ops: reg
1411 196 : .get("operations")
1412 196 : .and_then(Value::as_array)
1413 196 : .map(|a| {
1414 186 : a.iter()
1415 186 : .filter_map(Value::as_str)
1416 186 : .map(str::to_owned)
1417 186 : .collect()
1418 186 : })
1419 196 : .unwrap_or_else(|| vec![DEFAULT_OPERATION_GROUP.into()]),
1420 196 : attrs: None,
1421 196 : ent_ids: Vec::new(),
1422 196 : ent_types: Vec::new(),
1423 196 : ent_patterns: Vec::new(),
1424 : // minimal view: no EntityInfo scope loaded ⇒ never narrow by it
1425 : ent_unrestricted: true,
1426 196 : tenant: reg.get("tenant").and_then(Value::as_str).map(str::to_owned),
1427 196 : alias: reg
1428 196 : .get("contextSourceAlias")
1429 196 : .and_then(Value::as_str)
1430 196 : .map(str::to_owned),
1431 : // 4.3.6.6: the registered headers (auth among them) travel with every
1432 : // forward, including the subscription operations that build their
1433 : // registration view from this function.
1434 196 : csi: csi_of(reg),
1435 196 : local_only: local_only_of(reg),
1436 196 : timeout_ms: reg
1437 196 : .get("management")
1438 196 : .and_then(|m| m.get("timeout"))
1439 196 : .and_then(Value::as_u64),
1440 196 : cooldown_ms: reg
1441 196 : .get("management")
1442 196 : .and_then(|m| m.get("cooldown"))
1443 196 : .and_then(Value::as_u64),
1444 : }
1445 196 : }
1446 :
1447 : /// 5.7.1.4 / 5.7.3.4: with an EntityMap in use, "only the retrieved Entity
1448 : /// Map shall be used to determine which Context Source Registrations match
1449 : /// the Entity ID" — a registration not listed in the entry does not match;
1450 : /// "the location of the linked EntityMap shall be passed as part of any
1451 : /// forwarded request" (conveyed as an extra header on the forward).
1452 22 : fn map_gate(mut reg: FedReg, map: Option<&Value>, id: &str) -> Option<FedReg> {
1453 22 : let Some(m) = map else { return Some(reg) };
1454 2 : let listed = m
1455 2 : .get("entityMap")
1456 2 : .and_then(|e| e.get(id))
1457 2 : .and_then(Value::as_array)
1458 2 : .is_some_and(|a| a.iter().any(|v| v.as_str() == Some(reg.reg_id.as_str())));
1459 2 : if !listed {
1460 2 : return None;
1461 0 : }
1462 0 : if let Some(mid) = m
1463 0 : .get("linkedMaps")
1464 0 : .and_then(|l| l.get(®.reg_id))
1465 0 : .and_then(Value::as_str)
1466 0 : {
1467 0 : reg.csi.push(("NGSILD-EntityMap".into(), mid.to_owned()));
1468 0 : }
1469 0 : Some(reg)
1470 22 : }
1471 :
1472 : #[allow(clippy::too_many_arguments)] // mirrors the wire: one param per forwarded request part
1473 246 : pub async fn fed_retrieve_temporal(
1474 246 : st: &AppState,
1475 246 : tenant: &TenantId,
1476 246 : headers: &HeaderMap,
1477 246 : ctx: &Context,
1478 246 : id: &str,
1479 246 : params: &HashMap<String, String>,
1480 246 : map: Option<&Value>,
1481 246 : warnings: &mut Vec<String>,
1482 246 : ) -> Result<Vec<(bool, Value)>, NgsiError> {
1483 246 : let spec = crate::registry::CsrSpec {
1484 246 : ids: Some(vec![id.to_owned()]),
1485 246 : temporal: crate::temporalq::TemporalQ::from_params(params, false)
1486 246 : .ok()
1487 246 : .flatten(),
1488 246 : ..Default::default()
1489 246 : };
1490 246 : let ctx_url = ctx_link_url(headers, &ctx.source);
1491 246 : let ctx_url = &ctx_url;
1492 246 : let regs: Vec<FedReg> = matching_regs(st, tenant, &spec, ctx, headers)
1493 246 : .await?
1494 246 : .into_iter()
1495 : // 5.7.3.4: a live EntityMap in use is the ONLY source of matching
1496 : // registrations; its linked map location travels with the forward.
1497 246 : .filter_map(|reg| map_gate(reg, map, id))
1498 246 : .filter(|reg| reg.supports("retrieveTemporal"))
1499 246 : .collect();
1500 246 : let fetched = fan_out(regs, move |reg| async move {
1501 : // the temporal window travels with the forward; sysAttrs for the
1502 : // 4.5.5.3 recency arbitration
1503 2 : let mut query = temporal_window(params);
1504 2 : query.extend(scope_attrs(®, ctx));
1505 2 : let (status, body, peer_warns) = forward(
1506 2 : st,
1507 2 : reqwest::Method::GET,
1508 2 : format!(
1509 : "{}/ngsi-ld/v1/temporal/entities/{}",
1510 : reg.endpoint,
1511 2 : path_segment(id)
1512 : ),
1513 2 : &query,
1514 2 : headers,
1515 2 : tenant,
1516 2 : ®,
1517 2 : ctx_url,
1518 2 : None,
1519 : )
1520 2 : .await;
1521 2 : (reg, status, body, peer_warns)
1522 4 : })
1523 246 : .await;
1524 246 : let mut out = Vec::new();
1525 246 : for (reg, status, body, peer_warns) in fetched {
1526 2 : if !usable_payload(st, tenant, status, &body, peer_warns, warnings) {
1527 0 : continue;
1528 2 : }
1529 2 : if body.get("id").and_then(Value::as_str) != Some(id) {
1530 0 : continue;
1531 2 : }
1532 2 : match import_temporal(&body, ®, ctx) {
1533 2 : Some(doc) => out.push((reg.mode == "auxiliary", doc)),
1534 0 : None => warnings.push(warning(
1535 : 111,
1536 0 : &alias_for(&st.host_alias, tenant),
1537 0 : "the payload of the response was invalid",
1538 : )),
1539 : }
1540 : }
1541 246 : Ok(out)
1542 246 : }
1543 :
1544 : /// The DateTime 4.5.5.3 arbitrates on, as a comparable key: 4.6.3 admits the
1545 : /// same instant with or without a fraction, so a raw string compare would
1546 : /// rank `…:01.500Z` below `…:01Z`.
1547 32 : fn recency(inst: &Value) -> String {
1548 32 : antares_model::dt_key(
1549 32 : inst.get("observedAt")
1550 32 : .or_else(|| inst.get("modifiedAt"))
1551 32 : .and_then(Value::as_str)
1552 32 : .unwrap_or(""),
1553 : )
1554 32 : }
1555 :
1556 : /// 4.5.5.2 Processing of Conflicting Transient Entities: for each received
1557 : /// Entity version with an entity-level expiresAt, add it as a non-reified
1558 : /// expiresAt on every Attribute instance that lacks one, and cap any
1559 : /// Attribute-level expiresAt further in the future to the entity's (earlier)
1560 : /// DateTime.
1561 70 : fn push_down_expires(doc: &mut Value) {
1562 70 : let Some(o) = doc.as_object_mut() else { return };
1563 70 : let Some(exp) = o.get("expiresAt").and_then(Value::as_str).map(String::from) else {
1564 54 : return;
1565 : };
1566 52 : for (k, v) in o.iter_mut() {
1567 48 : if matches!(
1568 52 : k.as_str(),
1569 52 : "id" | "type" | "scope" | "expiresAt" | "createdAt" | "modifiedAt" | "deletedAt"
1570 : ) {
1571 48 : continue;
1572 4 : }
1573 4 : let Some(instances) = v.as_array_mut() else {
1574 0 : continue;
1575 : };
1576 12 : for inst in instances.iter_mut().filter_map(Value::as_object_mut) {
1577 12 : match inst.get("expiresAt").and_then(Value::as_str) {
1578 8 : Some(ae) if antares_model::dt_key(ae) <= antares_model::dt_key(&exp) => {}
1579 8 : _ => {
1580 8 : inst.insert("expiresAt".into(), Value::String(exp.clone()));
1581 8 : }
1582 : }
1583 : }
1584 : }
1585 70 : }
1586 :
1587 : /// 4.5.5.3 first step: "if an expiresAt DateTime is present on
1588 : /// the Attribute and the date lies in the past, it shall be discarded" —
1589 : /// BEFORE any recency comparison.
1590 44 : fn expired(inst: &Value, now: &str) -> bool {
1591 44 : inst.get("expiresAt")
1592 44 : .and_then(Value::as_str)
1593 44 : .is_some_and(|e| antares_model::dt_key(e) < antares_model::dt_key(now))
1594 44 : }
1595 :
1596 : /// 4.3.6.2: "An auxiliary Context Source Registration never overrides data
1597 : /// held directly within a Context Broker. […] Context data from auxiliary
1598 : /// context sources is only included if it is supplementary."
1599 : /// Merge attributes of `add` into `base` (auxiliary sources never override —
1600 : /// base wins; otherwise conflicting instances resolve per 4.5.5.3: discard
1601 : /// past-expiresAt instances first, then most recent observedAt/modifiedAt).
1602 70 : pub fn merge_docs(base: &mut Value, add: &Value, aux: bool) {
1603 70 : let now = crate::state::now_iso();
1604 70 : let mut add = add.clone();
1605 70 : push_down_expires(&mut add);
1606 70 : let Some(bo) = base.as_object_mut() else {
1607 0 : return;
1608 : };
1609 70 : let Some(ao) = add.as_object() else { return };
1610 : // 4.5.5.3: entity-level expiresAt — "missing from at least one version of
1611 : // the Entity received" → removed; present in all versions → the DateTime
1612 : // furthest in the future. 4.3.6.2 keeps auxiliary versions out of this:
1613 : // they never override data the broker holds directly, and removing or
1614 : // extending the lifetime is an override.
1615 70 : if !aux {
1616 : match (
1617 56 : bo.get("expiresAt").and_then(Value::as_str),
1618 56 : ao.get("expiresAt").and_then(Value::as_str),
1619 : ) {
1620 4 : (Some(b), Some(a)) => {
1621 4 : if antares_model::dt_key(a) > antares_model::dt_key(b) {
1622 4 : bo.insert("expiresAt".into(), Value::String(a.to_owned()));
1623 4 : }
1624 : }
1625 52 : _ => {
1626 52 : bo.remove("expiresAt");
1627 52 : }
1628 : }
1629 14 : }
1630 236 : for (k, v) in ao {
1631 236 : if k == "expiresAt" {
1632 16 : continue; // resolved above
1633 220 : }
1634 220 : match bo.get_mut(k) {
1635 48 : None => {
1636 48 : bo.insert(k.clone(), v.clone());
1637 48 : }
1638 172 : Some(cur) if !aux && !matches!(k.as_str(), "id" | "type" | "scope") => {
1639 24 : let (Some(ca), Some(aa)) = (cur.as_array_mut(), v.as_array()) else {
1640 0 : continue;
1641 : };
1642 24 : for ai in aa {
1643 24 : if expired(ai, &now) {
1644 4 : continue; // 4.5.5.3: discarded before comparison
1645 20 : }
1646 20 : let ds = ai.get("datasetId").and_then(Value::as_str);
1647 20 : match ca
1648 20 : .iter_mut()
1649 20 : .find(|ci| ci.get("datasetId").and_then(Value::as_str) == ds)
1650 : {
1651 0 : None => ca.push(ai.clone()),
1652 20 : Some(ci) => {
1653 20 : if expired(ci, &now) || recency(ai) > recency(ci) {
1654 12 : *ci = ai.clone();
1655 12 : }
1656 : }
1657 : }
1658 : }
1659 : }
1660 148 : _ => {}
1661 : }
1662 : }
1663 70 : }
1664 :
1665 : // ---------- distributed reads ----------
1666 :
1667 : /// Federated retrieve: internal docs from every matching registration,
1668 : /// (aux, doc) pairs so callers can order the merge.
1669 : #[allow(clippy::too_many_arguments)] // mirrors the wire: one param per forwarded request part
1670 528 : pub async fn fed_retrieve(
1671 528 : st: &AppState,
1672 528 : tenant: &TenantId,
1673 528 : headers: &HeaderMap,
1674 528 : ctx: &Context,
1675 528 : id: &str,
1676 528 : map: Option<&Value>,
1677 528 : except_reg: Option<&str>,
1678 528 : warnings: &mut Vec<String>,
1679 528 : ) -> Result<Vec<(bool, Value)>, NgsiError> {
1680 528 : let spec = crate::registry::CsrSpec {
1681 528 : ids: Some(vec![id.to_owned()]),
1682 528 : ..Default::default()
1683 528 : };
1684 528 : let ctx_url = ctx_link_url(headers, &ctx.source);
1685 528 : let ctx_url = &ctx_url;
1686 528 : let regs: Vec<FedReg> = matching_regs(st, tenant, &spec, ctx, headers)
1687 528 : .await?
1688 526 : .into_iter()
1689 : // 5.8.6 splitEntities merge: "except for the one from which the
1690 : // Notification has been received"
1691 526 : .filter(|reg| !except_reg.is_some_and(|x| x == reg.reg_id))
1692 : // 5.7.1.4: a live EntityMap in use is the ONLY source of matching
1693 : // registrations; its linked map location travels with the forward.
1694 526 : .filter_map(|reg| map_gate(reg, map, id))
1695 526 : .filter(|reg| reg.read_op().is_some())
1696 526 : .collect();
1697 526 : let fetched = fan_out(regs, move |reg| async move {
1698 12 : let Some(op) = reg.read_op() else {
1699 0 : return (reg, 0, Value::Null, Vec::new());
1700 : };
1701 : // sysAttrs on every forwarded read: conflicting instances resolve by
1702 : // most recent observedAt/modifiedAt (4.5.5.3) — without the remote
1703 : // modifiedAt the winner would be arrival order, i.e. indeterminate.
1704 12 : let mut query: Vec<(String, String)> = vec![("options".into(), "sysAttrs".into())];
1705 12 : query.extend(scope_attrs(®, ctx));
1706 12 : let (status, body, peer_warns) = match op {
1707 12 : "retrieveEntity" => {
1708 12 : forward(
1709 12 : st,
1710 12 : reqwest::Method::GET,
1711 12 : format!("{}/ngsi-ld/v1/entities/{}", reg.endpoint, path_segment(id)),
1712 12 : &query,
1713 12 : headers,
1714 12 : tenant,
1715 12 : ®,
1716 12 : ctx_url,
1717 12 : None,
1718 12 : )
1719 12 : .await
1720 : }
1721 0 : "queryEntity" => {
1722 0 : let t = reg
1723 0 : .ent_types
1724 0 : .first()
1725 0 : .map(|t| ctx.compact_iri(t))
1726 0 : .unwrap_or_else(|| "*".into());
1727 0 : query.push(("type".into(), t));
1728 0 : query.push(("id".into(), id.to_owned()));
1729 0 : forward(
1730 0 : st,
1731 0 : reqwest::Method::GET,
1732 0 : format!("{}/ngsi-ld/v1/entities", reg.endpoint),
1733 0 : &query,
1734 0 : headers,
1735 0 : tenant,
1736 0 : ®,
1737 0 : ctx_url,
1738 0 : None,
1739 0 : )
1740 0 : .await
1741 : }
1742 : _ => {
1743 : // queryBatch
1744 0 : let mut sel = Map::new();
1745 0 : if let Some(t) = reg.ent_types.first() {
1746 0 : sel.insert("type".into(), Value::String(ctx.compact_iri(t)));
1747 0 : }
1748 0 : sel.insert("id".into(), Value::String(id.to_owned()));
1749 0 : forward(
1750 0 : st,
1751 0 : reqwest::Method::POST,
1752 0 : format!("{}/ngsi-ld/v1/entityOperations/query", reg.endpoint),
1753 0 : &query,
1754 0 : headers,
1755 0 : tenant,
1756 0 : ®,
1757 0 : ctx_url,
1758 0 : Some(json!({"type": "Query", "entities": [Value::Object(sel)]})),
1759 0 : )
1760 0 : .await
1761 : }
1762 : };
1763 12 : (reg, status, body, peer_warns)
1764 24 : })
1765 526 : .await;
1766 526 : let mut out = Vec::new();
1767 526 : for (reg, status, body, peer_warns) in fetched {
1768 12 : if !usable_payload(st, tenant, status, &body, peer_warns, warnings) {
1769 4 : continue;
1770 8 : }
1771 8 : let candidates: Vec<&Value> = match &body {
1772 0 : Value::Array(a) => a.iter().collect(),
1773 8 : Value::Object(_) => vec![&body],
1774 0 : _ => continue,
1775 : };
1776 8 : for c in candidates {
1777 8 : if c.get("id").and_then(Value::as_str) != Some(id) {
1778 0 : continue;
1779 8 : }
1780 8 : match import_entity(c, ®, ctx) {
1781 8 : Some(doc) => out.push((reg.mode == "auxiliary", doc)),
1782 : // received in time, but not a parseable NGSI-LD entity (111)
1783 0 : None => warnings.push(warning(
1784 : 111,
1785 0 : &alias_for(&st.host_alias, tenant),
1786 0 : "the payload of the response was invalid",
1787 : )),
1788 : }
1789 : }
1790 : }
1791 526 : Ok(out)
1792 528 : }
1793 :
1794 : /// Federated query: internal docs matching a type query, per registration.
1795 : /// The `CsrSpec` a Query Entities request matches registrations against.
1796 1672 : fn query_spec(ctx: &Context, params: &HashMap<String, String>) -> crate::registry::CsrSpec {
1797 : // 4.17: the parameter is ONE Entity Type Selection, so it travels whole —
1798 : // splitting it on commas turned a conjunction like (A;B) into two terms
1799 : // that match nothing, and no registration was consulted for it.
1800 1672 : let types: Option<Vec<String>> = params.get("type").cloned().map(|s| vec![s]);
1801 1672 : let ids: Option<Vec<String>> = params
1802 1672 : .get("id")
1803 1672 : .map(|s| s.split(',').map(str::to_owned).collect());
1804 : crate::registry::CsrSpec {
1805 1672 : types,
1806 1672 : ids,
1807 : // 5.12: "the id pattern (if present)" is part of the query-side
1808 : // Entity specification matched against EntityInfo elements
1809 1672 : id_pattern: params.get("idPattern").cloned(),
1810 : // 5.12 attribute conditions: the "list of Attribute names (if
1811 : // present)" gates which RegistrationInfos match
1812 1672 : attrs: params
1813 1672 : .get("attrs")
1814 1672 : .map(|s| s.split(',').map(|a| ctx.expand_key(a.trim())).collect()),
1815 : // 5.12 datasetId condition (should-level): disjoint sets don't match
1816 1672 : dataset_ids: params
1817 1672 : .get("datasetId")
1818 1672 : .map(|s| s.split(',').map(|d| d.trim().to_owned()).collect()),
1819 1672 : csf: params.get("csf").and_then(|c| antares_ql::parse_q(c).ok()),
1820 1672 : geo: antares_ql::geo::GeoQuery::from_params(params)
1821 1672 : .ok()
1822 1672 : .flatten(),
1823 1672 : ..Default::default()
1824 : }
1825 1672 : }
1826 :
1827 : /// Will this query actually fan out to a Context Source?
1828 : ///
1829 : /// 5.7.2.4 forbids ordering when "the execution of the operation is not
1830 : /// limited to the local scope", and 4.23.1 says "Sort ordering is never
1831 : /// applied to distributed operations". The subject is the *execution*, not the
1832 : /// presence of `local=true`: a query no registration matches executes locally
1833 : /// whether or not the client said so. Reading it as "local=true is mandatory
1834 : /// for orderBy" would fail ETSI's own 019_19, which orders without it.
1835 328 : pub async fn would_federate(
1836 328 : st: &AppState,
1837 328 : tenant: &TenantId,
1838 328 : ctx: &Context,
1839 328 : params: &HashMap<String, String>,
1840 328 : headers: &HeaderMap,
1841 328 : ) -> Result<bool, NgsiError> {
1842 328 : if !active(params) || via_hops(headers) > MAX_VIA_HOPS {
1843 50 : return Ok(false);
1844 278 : }
1845 : // the verdict only — no forwarding set is compiled for it
1846 278 : let spec = query_spec(ctx, params);
1847 278 : let seen = via_tokens(headers);
1848 278 : Ok(reg_docs(st, tenant, &spec, ctx)
1849 278 : .await?
1850 278 : .iter()
1851 278 : .any(|doc| reg_candidate(doc, &spec, ctx, &seen).is_some()))
1852 328 : }
1853 :
1854 : /// May a fan-out response contribute the entity `id` under registration
1855 : /// `reg`, for a request whose id selection is `spec`?
1856 : ///
1857 : /// A source only speaks for what its registration covers — an id outside
1858 : /// that scope is dropped rather than merged, or a peer could overwrite
1859 : /// unrelated local attributes on recency (4.5.5.3). But 5.12's matching is
1860 : /// deliberately over-broad for patterns: condition 5 forwards when "both a
1861 : /// specified id pattern and an idPattern in the Entity Info are present
1862 : /// (since in the general case it is not easily feasible to determine if
1863 : /// there can be identifiers matching both patterns)". The same
1864 : /// undecidability applies to the ANSWER: an id the CLIENT's own selection
1865 : /// admits cannot be refused on the registration's pattern. A peer therefore
1866 : /// contributes only ids inside its registration scope or inside what the
1867 : /// client itself asked to see — never a third thing.
1868 492 : fn admits_import(reg: &FedReg, spec: &crate::registry::CsrSpec, id: &str) -> bool {
1869 492 : reg.can_match_id(id)
1870 20 : || spec
1871 20 : .ids
1872 20 : .as_ref()
1873 20 : .is_some_and(|ids| ids.iter().any(|i| i == id))
1874 16 : || spec
1875 16 : .id_pattern
1876 16 : .as_deref()
1877 16 : .is_some_and(|p| antares_ql::regex::compile(p).is_ok_and(|re| re.find(id).is_some()))
1878 492 : }
1879 :
1880 724 : pub async fn fed_query(
1881 724 : st: &AppState,
1882 724 : tenant: &TenantId,
1883 724 : headers: &HeaderMap,
1884 724 : ctx: &Context,
1885 724 : params: &HashMap<String, String>,
1886 724 : warnings: &mut Vec<String>,
1887 724 : ) -> Result<Vec<(bool, Value)>, NgsiError> {
1888 724 : let spec = query_spec(ctx, params);
1889 724 : let ctx_url = ctx_link_url(headers, &ctx.source);
1890 724 : let ctx_url = &ctx_url;
1891 724 : let regs: Vec<FedReg> = matching_regs(st, tenant, &spec, ctx, headers)
1892 724 : .await?
1893 720 : .into_iter()
1894 720 : .filter(|r| r.query_op().is_some())
1895 720 : .collect();
1896 720 : let fetched = fan_out(regs, move |reg| async move {
1897 176 : let Some(op) = reg.query_op() else {
1898 0 : return (reg, 0, Value::Null, Vec::new());
1899 : };
1900 : // The forwarded selection is decided ONCE and then rendered either as
1901 : // query parameters (Query Entities, 5.7.2) or as a Query body
1902 : // (5.2.23) — the two must ask the peer the same question.
1903 : //
1904 : // 4.3.6.1: the forwarded id list carries only ids this registration
1905 : // can match — never the full client list.
1906 176 : let ids: Option<String> = match params.get("id") {
1907 40 : Some(ids) => {
1908 40 : let keep: Vec<&str> = ids
1909 40 : .split(',')
1910 90 : .filter(|i| reg.can_match_id(i.trim()))
1911 40 : .collect();
1912 40 : (!keep.is_empty()).then(|| keep.join(","))
1913 : }
1914 : // the registration is scoped to exact ids only — ask for those
1915 136 : None if !reg.ent_unrestricted
1916 14 : && reg.ent_patterns.is_empty()
1917 6 : && !reg.ent_ids.is_empty() =>
1918 : {
1919 6 : Some(reg.ent_ids.join(","))
1920 : }
1921 130 : None => None,
1922 : };
1923 176 : let attrs: Option<String> = match ®.attrs {
1924 18 : Some(scope) => Some(
1925 18 : scope
1926 18 : .iter()
1927 18 : .map(|a| ctx.compact_iri(a))
1928 18 : .collect::<Vec<String>>()
1929 18 : .join(","),
1930 : ),
1931 158 : None => params.get("attrs").cloned(),
1932 : };
1933 : // 5.7.2.4: with split entities the filters "shall be removed
1934 : // before forwarding" and re-applied on the aggregate (which the
1935 : // local re-check always does); otherwise the request is forwarded
1936 : // WITH its filters, so the peer returns its filtered subset
1937 : // instead of everything. A registered jsonldContext (4.3.6.6)
1938 : // recompacts only attrs/type/geoproperty — q/scopeQ terms cannot
1939 : // be recompacted, so push-down is skipped there rather than
1940 : // filtering at the remote against the wrong terms.
1941 176 : let push_filters = !split_entities(params) && !has_reg_context(®);
1942 1056 : let filter = |k: &str| {
1943 1056 : push_filters
1944 1056 : .then(|| params.get(k))
1945 1056 : .flatten()
1946 1056 : .map(String::to_owned)
1947 1056 : };
1948 176 : let (status, body, peer_warns) = if op == "queryBatch" {
1949 4 : let mut sel = Map::new();
1950 4 : if let Some(t) = params.get("type") {
1951 4 : sel.insert("type".into(), Value::String(t.clone()));
1952 4 : }
1953 0 : match &ids {
1954 : // 5.2.33: `id` is one URI or an array of them, and it takes
1955 : // precedence over idPattern — so a pattern only travels when
1956 : // no id list survived the narrowing.
1957 0 : Some(list) if list.contains(',') => {
1958 0 : let arr: Vec<Value> =
1959 0 : list.split(',').map(|i| Value::String(i.into())).collect();
1960 0 : sel.insert("id".into(), Value::Array(arr));
1961 : }
1962 0 : Some(one) => {
1963 0 : sel.insert("id".into(), Value::String(one.clone()));
1964 0 : }
1965 : None => {
1966 4 : if let Some(p) = params.get("idPattern") {
1967 2 : sel.insert("idPattern".into(), Value::String(p.clone()));
1968 2 : }
1969 : }
1970 : }
1971 4 : let mut q_body = Map::new();
1972 4 : q_body.insert("type".into(), Value::String("Query".into()));
1973 4 : if !sel.is_empty() {
1974 4 : q_body.insert("entities".into(), json!([Value::Object(sel)]));
1975 4 : }
1976 4 : if let Some(a) = &attrs {
1977 2 : let list: Vec<Value> = a.split(',').map(|n| Value::String(n.into())).collect();
1978 2 : q_body.insert("attrs".into(), Value::Array(list));
1979 2 : }
1980 4 : if let Some(q) = filter("q") {
1981 2 : q_body.insert("q".into(), Value::String(q));
1982 2 : }
1983 4 : if let Some(s) = filter("scopeQ") {
1984 2 : q_body.insert("scopeQ".into(), Value::String(s));
1985 2 : }
1986 4 : let mut geo = Map::new();
1987 16 : for k in ["georel", "geometry", "coordinates", "geoproperty"] {
1988 16 : if let Some(v) = filter(k) {
1989 : // 5.2.13 GeoQuery carries coordinates as the GeoJSON
1990 : // value, not as the query-string spelling of it.
1991 6 : let parsed = if k == "coordinates" {
1992 2 : serde_json::from_str(&v).unwrap_or(Value::String(v))
1993 : } else {
1994 4 : Value::String(v)
1995 : };
1996 6 : geo.insert(k.into(), parsed);
1997 10 : }
1998 : }
1999 4 : if !geo.is_empty() {
2000 2 : q_body.insert("geoQ".into(), Value::Object(geo));
2001 2 : }
2002 4 : forward(
2003 4 : st,
2004 4 : reqwest::Method::POST,
2005 4 : format!("{}/ngsi-ld/v1/entityOperations/query", reg.endpoint),
2006 4 : &[("options".into(), "sysAttrs".into())],
2007 4 : headers,
2008 4 : tenant,
2009 4 : ®,
2010 4 : ctx_url,
2011 4 : Some(Value::Object(q_body)),
2012 4 : )
2013 4 : .await
2014 : } else {
2015 172 : let mut query: Vec<(String, String)> = vec![("options".into(), "sysAttrs".into())];
2016 172 : if let Some(t) = params.get("type") {
2017 172 : query.push(("type".into(), t.clone()));
2018 172 : }
2019 : // Table 6.4.3.2-1, and 5.2.33 for the body rendering below:
2020 : // `id` takes precedence over `idPattern`, so a pattern travels
2021 : // only when no id list survived the narrowing. The two
2022 : // renderings ask the peer the same question.
2023 172 : match &ids {
2024 38 : Some(list) => query.push(("id".into(), list.clone())),
2025 : None => {
2026 134 : if let Some(p) = params.get("idPattern") {
2027 22 : query.push(("idPattern".into(), p.clone()));
2028 112 : }
2029 : }
2030 : }
2031 172 : if let Some(a) = &attrs {
2032 20 : query.push(("attrs".into(), a.clone()));
2033 152 : }
2034 1032 : for k in [
2035 172 : "q",
2036 172 : "georel",
2037 172 : "geometry",
2038 172 : "coordinates",
2039 172 : "geoproperty",
2040 172 : "scopeQ",
2041 172 : ] {
2042 1032 : if let Some(v) = filter(k) {
2043 22 : query.push((k.into(), v));
2044 1010 : }
2045 : }
2046 172 : forward(
2047 172 : st,
2048 172 : reqwest::Method::GET,
2049 172 : format!("{}/ngsi-ld/v1/entities", reg.endpoint),
2050 172 : &query,
2051 172 : headers,
2052 172 : tenant,
2053 172 : ®,
2054 172 : ctx_url,
2055 172 : None,
2056 172 : )
2057 172 : .await
2058 : };
2059 176 : (reg, status, body, peer_warns)
2060 352 : })
2061 720 : .await;
2062 720 : let mut out = Vec::new();
2063 720 : for (reg, status, body, peer_warns) in fetched {
2064 176 : if !usable_payload(st, tenant, status, &body, peer_warns, warnings) {
2065 10 : continue;
2066 166 : }
2067 166 : if let Value::Array(a) = &body {
2068 : // Scope gate — see admits_import.
2069 458 : for c in a.iter().filter(|c| {
2070 458 : c.get("id")
2071 458 : .and_then(Value::as_str)
2072 458 : .is_some_and(|i| admits_import(®, &spec, i))
2073 458 : }) {
2074 458 : match import_entity(c, ®, ctx) {
2075 458 : Some(doc) => out.push((reg.mode == "auxiliary", doc)),
2076 0 : None => warnings.push(warning(
2077 : 111,
2078 0 : &alias_for(&st.host_alias, tenant),
2079 0 : "the payload of the response was invalid",
2080 : )),
2081 : }
2082 : }
2083 4 : }
2084 : }
2085 720 : Ok(out)
2086 724 : }
2087 :
2088 : /// 5.14.4.4 / 5.14.5.4: forward EntityMap creation to matching registrations
2089 : /// that support `op` (createEntityMapQueryEntity / …QueryTemporal, 4.20);
2090 : /// with split entities in play the value/geo/scope filters are removed
2091 : /// before forwarding. Returns (registration id, returned EntityMap) pairs —
2092 : /// the caller merges them into the local map's entityMap/linkedMaps.
2093 : #[allow(clippy::too_many_arguments)] // mirrors the wire: one param per forwarded request part
2094 172 : pub(crate) async fn fed_entity_maps(
2095 172 : st: &AppState,
2096 172 : tenant: &TenantId,
2097 172 : headers: &HeaderMap,
2098 172 : ctx: &Context,
2099 172 : params: &HashMap<String, String>,
2100 172 : split: bool,
2101 172 : op: &str,
2102 172 : path: &str,
2103 172 : ) -> Result<Vec<(String, Value)>, NgsiError> {
2104 172 : let spec = query_spec(ctx, params);
2105 172 : let ctx_url = ctx_link_url(headers, &ctx.source);
2106 172 : let ctx_url = &ctx_url;
2107 172 : let regs: Vec<FedReg> = matching_regs(st, tenant, &spec, ctx, headers)
2108 172 : .await?
2109 172 : .into_iter()
2110 172 : .filter(|reg| reg.supports(op))
2111 172 : .collect();
2112 172 : let fetched = fan_out(regs, move |reg| async move {
2113 10 : let mut query: Vec<(String, String)> = Vec::new();
2114 80 : for k in [
2115 10 : "id",
2116 10 : "idPattern",
2117 10 : "type",
2118 10 : "timerel",
2119 10 : "timeAt",
2120 10 : "endTimeAt",
2121 10 : "timeproperty",
2122 10 : "lastN",
2123 10 : ] {
2124 80 : if let Some(v) = params.get(k) {
2125 10 : query.push((k.to_owned(), v.clone()));
2126 70 : }
2127 : }
2128 10 : if !split {
2129 80 : for k in [
2130 10 : "attrs",
2131 10 : "q",
2132 10 : "georel",
2133 10 : "geometry",
2134 10 : "coordinates",
2135 10 : "geoproperty",
2136 10 : "scopeQ",
2137 10 : "lang",
2138 10 : ] {
2139 80 : if let Some(v) = params.get(k) {
2140 0 : query.push((k.to_owned(), v.clone()));
2141 80 : }
2142 : }
2143 0 : }
2144 10 : let (status, body, _) = forward(
2145 10 : st,
2146 10 : reqwest::Method::GET,
2147 10 : format!("{}/ngsi-ld/v1/{path}", reg.endpoint),
2148 10 : &query,
2149 10 : headers,
2150 10 : tenant,
2151 10 : ®,
2152 10 : ctx_url,
2153 10 : None,
2154 : )
2155 10 : .await;
2156 10 : (reg, status, body)
2157 20 : })
2158 172 : .await;
2159 172 : let mut out = Vec::new();
2160 172 : for (reg, status, body) in fetched {
2161 10 : if (200..300).contains(&status) && body.get("entityMap").is_some() {
2162 10 : out.push((reg.reg_id.clone(), body));
2163 10 : }
2164 : }
2165 172 : Ok(out)
2166 172 : }
2167 :
2168 : /// 5.7.4.4: forward the temporal query to matching registrations that
2169 : /// support the queryTemporal operation; registrations without it are not
2170 : /// contacted. Returns (auxiliary, expanded doc) pairs.
2171 478 : pub async fn fed_query_temporal(
2172 478 : st: &AppState,
2173 478 : tenant: &TenantId,
2174 478 : headers: &HeaderMap,
2175 478 : ctx: &Context,
2176 478 : params: &HashMap<String, String>,
2177 478 : warnings: &mut Vec<String>,
2178 478 : ) -> Result<Vec<(bool, Value)>, NgsiError> {
2179 478 : let mut spec = query_spec(ctx, params);
2180 478 : spec.temporal = crate::temporalq::TemporalQ::from_params(params, false)
2181 478 : .ok()
2182 478 : .flatten();
2183 478 : let ctx_url = ctx_link_url(headers, &ctx.source);
2184 478 : let ctx_url = &ctx_url;
2185 478 : let regs: Vec<FedReg> = matching_regs(st, tenant, &spec, ctx, headers)
2186 478 : .await?
2187 478 : .into_iter()
2188 478 : .filter(|reg| reg.supports("queryTemporal"))
2189 478 : .collect();
2190 478 : let fetched = fan_out(regs, move |reg| async move {
2191 10 : let mut query = temporal_window(params);
2192 30 : for k in ["type", "id", "idPattern"] {
2193 30 : if let Some(v) = params.get(k) {
2194 10 : query.push((k.into(), v.clone()));
2195 20 : }
2196 : }
2197 : // 5.7.4.4 mirrors 5.7.2.4: with split entities the value/geo/scope
2198 : // filters are stripped from the forward and applied on the
2199 : // aggregate; a registered jsonldContext cannot recompact q/scopeQ
2200 : // terms, so push-down is skipped there too.
2201 10 : if !split_entities(params) && !has_reg_context(®) {
2202 48 : for k in [
2203 8 : "q",
2204 8 : "georel",
2205 8 : "geometry",
2206 8 : "coordinates",
2207 8 : "geoproperty",
2208 8 : "scopeQ",
2209 8 : ] {
2210 48 : if let Some(v) = params.get(k) {
2211 2 : query.push((k.into(), v.clone()));
2212 46 : }
2213 : }
2214 2 : }
2215 10 : if let Some(a) = scope_attrs(®, ctx) {
2216 2 : query.push(a);
2217 8 : } else if let Some(a) = params.get("attrs") {
2218 0 : query.push(("attrs".into(), a.clone()));
2219 8 : }
2220 10 : let (status, body, peer_warns) = forward(
2221 10 : st,
2222 10 : reqwest::Method::GET,
2223 10 : format!("{}/ngsi-ld/v1/temporal/entities", reg.endpoint),
2224 10 : &query,
2225 10 : headers,
2226 10 : tenant,
2227 10 : ®,
2228 10 : ctx_url,
2229 10 : None,
2230 : )
2231 10 : .await;
2232 10 : (reg, status, body, peer_warns)
2233 20 : })
2234 478 : .await;
2235 478 : let mut out = Vec::new();
2236 478 : for (reg, status, body, peer_warns) in fetched {
2237 10 : if !usable_payload(st, tenant, status, &body, peer_warns, warnings) {
2238 0 : continue;
2239 10 : }
2240 10 : if let Value::Array(a) = &body {
2241 : // Same scope gate as the non-temporal query fan-out.
2242 10 : for c in a.iter().filter(|c| {
2243 10 : c.get("id")
2244 10 : .and_then(Value::as_str)
2245 10 : .is_some_and(|i| admits_import(®, &spec, i))
2246 10 : }) {
2247 10 : match import_temporal(c, ®, ctx) {
2248 10 : Some(doc) => out.push((reg.mode == "auxiliary", doc)),
2249 0 : None => warnings.push(warning(
2250 : 111,
2251 0 : &alias_for(&st.host_alias, tenant),
2252 0 : "the payload of the response was invalid",
2253 : )),
2254 : }
2255 : }
2256 0 : }
2257 : }
2258 478 : Ok(out)
2259 478 : }
2260 :
2261 : /// Merge federated docs into a local candidate set (keyed by id). Local docs
2262 : /// win; non-aux remote attrs merge before aux ones.
2263 982 : pub fn merge_candidates(local: Vec<Value>, fed: Vec<(bool, Value)>) -> Vec<Value> {
2264 982 : let mut order: Vec<String> = Vec::new();
2265 982 : let mut by_id: HashMap<String, Value> = HashMap::new();
2266 18726 : for doc in local {
2267 18726 : if let Some(id) = doc.get("id").and_then(Value::as_str) {
2268 18726 : order.push(id.to_owned());
2269 18726 : by_id.insert(id.to_owned(), doc);
2270 18726 : }
2271 : }
2272 1964 : for aux_pass in [false, true] {
2273 1964 : for (aux, doc) in &fed {
2274 916 : if *aux != aux_pass {
2275 458 : continue;
2276 458 : }
2277 458 : let Some(id) = doc.get("id").and_then(Value::as_str) else {
2278 0 : continue;
2279 : };
2280 458 : match by_id.get_mut(id) {
2281 6 : Some(base) => merge_docs(base, doc, *aux),
2282 452 : None => {
2283 452 : order.push(id.to_owned());
2284 452 : by_id.insert(id.to_owned(), doc.clone());
2285 452 : }
2286 : }
2287 : }
2288 : }
2289 982 : order
2290 982 : .into_iter()
2291 19178 : .filter_map(|id| by_id.remove(&id))
2292 982 : .collect()
2293 982 : }
2294 :
2295 : // ---------- distributed writes ----------
2296 :
2297 : /// Outcome of one part of a distributed write.
2298 : pub struct Part {
2299 : pub status: u16,
2300 : pub detail: String,
2301 : }
2302 :
2303 : impl Part {
2304 476 : pub fn ok(&self) -> bool {
2305 : // 207 from a forwarded source is a partial failure, not a success
2306 476 : (200..300).contains(&self.status) && self.status != 207
2307 476 : }
2308 : }
2309 :
2310 : /// Combine local + forwarded parts (6.3.17/6.4.3.1): all-success ⇒ `ok`,
2311 : /// single failing part ⇒ its own error, mixed ⇒ 207 Multi-Status.
2312 172 : pub fn combine(parts: Vec<Part>, ok: Response, tenant: &TenantId) -> Response {
2313 172 : if parts.iter().all(Part::ok) {
2314 98 : return ok;
2315 74 : }
2316 74 : if parts.len() == 1 {
2317 44 : let p = &parts[0];
2318 44 : let status = StatusCode::from_u16(p.status).unwrap_or(StatusCode::BAD_GATEWAY);
2319 44 : let (etype, title) = match p.status {
2320 : // 5.6.1.4/5.6.2…: an unsupported-operation part is the Conflict
2321 : // error type; an entity-exists part stays AlreadyExists.
2322 10 : 409 if p.detail.contains("does not accept") => ("Conflict", "Conflict"),
2323 0 : 409 => ("AlreadyExists", "Conflict"),
2324 10 : 404 => ("ResourceNotFound", "Not Found"),
2325 24 : _ => ("InternalError", "Error"),
2326 : };
2327 44 : let body = json!({
2328 44 : "type": format!("https://uri.etsi.org/ngsi-ld/errors/{etype}"),
2329 44 : "title": title,
2330 44 : "detail": p.detail,
2331 44 : "status": p.status,
2332 : });
2333 44 : let mut resp = (
2334 44 : status,
2335 44 : [(axum::http::header::CONTENT_TYPE, "application/json")],
2336 44 : axum::Json(body),
2337 44 : )
2338 44 : .into_response();
2339 44 : echo_tenant(tenant, &mut resp);
2340 44 : return resp;
2341 30 : }
2342 30 : let errors: Vec<Value> = parts
2343 30 : .iter()
2344 60 : .filter(|p| !p.ok())
2345 34 : .map(|p| {
2346 34 : json!({
2347 34 : "error": {
2348 34 : "status": p.status,
2349 34 : "type": "https://uri.etsi.org/ngsi-ld/errors/InternalError",
2350 34 : "title": "distributed operation failed",
2351 34 : "detail": p.detail,
2352 : }
2353 : })
2354 34 : })
2355 30 : .collect();
2356 : // 6.3.17: "the error response should be as informative as possible" — a
2357 : // 207 that hides the succeeded halves (did my local delete happen?)
2358 : // isn't. Parts carry no entity id, so success entries are the details.
2359 30 : let success: Vec<&str> = parts
2360 30 : .iter()
2361 60 : .filter(|p| p.ok())
2362 30 : .map(|p| p.detail.as_str())
2363 30 : .collect();
2364 30 : let body = json!({"success": success, "errors": errors});
2365 30 : let mut resp = (
2366 30 : StatusCode::MULTI_STATUS,
2367 30 : [(axum::http::header::CONTENT_TYPE, "application/json")],
2368 30 : axum::Json(body),
2369 30 : )
2370 30 : .into_response();
2371 30 : echo_tenant(tenant, &mut resp);
2372 30 : resp
2373 172 : }
2374 :
2375 : /// 4.3.6.1: "It is the responsibility of the Context Broker to respect the
2376 : /// registration parameters when issuing distributed requests. […] Ultimately,
2377 : /// all constraints specified in the registration shall be respected."
2378 : /// Reduce a compacted entity/fragment to the members a registration covers
2379 : /// (plus id/type); returns None if no attribute member remains.
2380 54 : pub fn reduce_to_scope(obj: &Map<String, Value>, reg: &FedReg, ctx: &Context) -> Option<Value> {
2381 : // an item outside the registration's EntityInfo ids/types is not this
2382 : // source's data at all — nothing of it may be forwarded there
2383 54 : if !reg.covers_item(obj, ctx) {
2384 4 : return None;
2385 50 : }
2386 50 : let Some(_) = ®.attrs else {
2387 42 : let mut full = obj.clone();
2388 42 : full.remove("@context");
2389 42 : return Some(Value::Object(full));
2390 : };
2391 8 : let mut out = Map::new();
2392 8 : let mut any = false;
2393 32 : for (k, v) in obj {
2394 32 : if k == "@context" {
2395 4 : continue;
2396 28 : }
2397 28 : if ["id", "type", "scope"].contains(&k.as_str()) {
2398 16 : out.insert(k.clone(), v.clone());
2399 16 : continue;
2400 12 : }
2401 12 : if reg.covers_attr(&ctx.expand_key(k)) {
2402 4 : out.insert(k.clone(), v.clone());
2403 4 : any = true;
2404 8 : }
2405 : }
2406 8 : any.then_some(Value::Object(out))
2407 54 : }
2408 :
2409 : /// 508 Loop Detected (6.3.17): the inbound Via chain already names us and a
2410 : /// registration would forward the operation right back.
2411 42 : pub fn loop_508(tenant: &TenantId) -> Response {
2412 42 : let body = json!({
2413 42 : "type": "https://uri.etsi.org/ngsi-ld/errors/InternalError",
2414 42 : "title": "Loop Detected",
2415 42 : "detail": "the Via chain already contains this broker",
2416 42 : "status": 508,
2417 : });
2418 42 : let mut resp = (
2419 42 : StatusCode::LOOP_DETECTED,
2420 42 : [(axum::http::header::CONTENT_TYPE, "application/json")],
2421 42 : axum::Json(body),
2422 42 : )
2423 42 : .into_response();
2424 42 : echo_tenant(tenant, &mut resp);
2425 42 : resp
2426 42 : }
2427 :
2428 : /// What a distributed write does with its registrations once the 6.3.18
2429 : /// loop rule has spoken: forward to these, or answer with the chain's own
2430 : /// response and touch nothing locally.
2431 : pub enum WritePlan {
2432 : /// The non-auxiliary registrations that match the operation's entity
2433 : /// specification, `Via` chain already applied; empty means local only.
2434 : Forward(Vec<FedReg>),
2435 : /// The loop chain's answer (Table 6.3.18-2, or 508 past the hop cap):
2436 : /// the operation ends here, before anything local happens.
2437 : Answered(Box<Response>),
2438 : }
2439 :
2440 : /// The prologue every distributed write shares (4.3.6, 6.3.18): the
2441 : /// registrations of [`write_regs`] with [`handle_via_loop`] applied, so
2442 : /// no operation re-derives the pair and the two cannot disagree.
2443 13514 : pub async fn write_plan(
2444 13514 : st: &AppState,
2445 13514 : tenant: &TenantId,
2446 13514 : spec: &crate::registry::CsrSpec,
2447 13514 : ctx: &Context,
2448 13514 : params: &HashMap<String, String>,
2449 13514 : headers: &HeaderMap,
2450 13514 : ) -> Result<WritePlan, NgsiError> {
2451 13514 : let mut regs = write_regs(st, tenant, spec, ctx, params, headers).await?;
2452 : Ok(
2453 13512 : match handle_via_loop(
2454 13512 : headers,
2455 13512 : &alias_for(&st.host_alias, tenant),
2456 13512 : tenant,
2457 13512 : &mut regs,
2458 13512 : ) {
2459 30 : Some(answer) => WritePlan::Answered(Box::new(answer)),
2460 13482 : None => WritePlan::Forward(regs),
2461 : },
2462 : )
2463 13514 : }
2464 :
2465 : /// 4.3.6.2: "Auxiliary distributed operations are limited to context
2466 : /// information consumption operations (see clause 5.7)" — so a write op
2467 : /// only ever considers non-auxiliary matching registrations.
2468 13518 : pub async fn write_regs(
2469 13518 : st: &AppState,
2470 13518 : tenant: &TenantId,
2471 13518 : spec: &crate::registry::CsrSpec,
2472 13518 : ctx: &Context,
2473 13518 : params: &HashMap<String, String>,
2474 13518 : headers: &HeaderMap,
2475 13518 : ) -> Result<Vec<FedReg>, NgsiError> {
2476 13518 : if !active(params) {
2477 44 : return Ok(Vec::new());
2478 13474 : }
2479 13474 : Ok(matching_regs(st, tenant, spec, ctx, headers)
2480 13474 : .await?
2481 13472 : .into_iter()
2482 13472 : .filter(|r| r.mode != "auxiliary")
2483 13472 : .collect())
2484 13518 : }
2485 :
2486 : /// Execute one forwarded write and turn it into a Part.
2487 : #[allow(clippy::too_many_arguments)] // mirrors the wire: one param per forwarded request part
2488 104 : pub async fn forward_part(
2489 104 : st: &AppState,
2490 104 : method: reqwest::Method,
2491 104 : url: String,
2492 104 : query: &[(String, String)],
2493 104 : headers: &HeaderMap,
2494 104 : tenant: &TenantId,
2495 104 : reg: &FedReg,
2496 104 : ctx_url: &str,
2497 104 : body: Option<Value>,
2498 104 : ) -> Part {
2499 104 : let (status, _, _) = forward(
2500 104 : st,
2501 104 : method,
2502 104 : url.clone(),
2503 104 : query,
2504 104 : headers,
2505 104 : tenant,
2506 104 : reg,
2507 104 : ctx_url,
2508 104 : body,
2509 : )
2510 104 : .await;
2511 : // 6.3.17: "the error response should be as informative as possible" —
2512 : // informative about the operation, not about the deployment. The part is
2513 : // named by its Context Source Registration id (5.2.9), which the client
2514 : // can already read from /csourceRegistrations; the registered endpoint
2515 : // is internal topology and stays in the log, so that a client able to
2516 : // provoke a partial failure cannot enumerate the peers.
2517 104 : tracing::debug!(
2518 : "distributed operation to {} returned {status}",
2519 0 : antares_notifier::redact_userinfo(&url)
2520 : );
2521 104 : let detail = format!(
2522 : "distributed operation to registration {} returned {status}",
2523 : reg.reg_id
2524 : );
2525 : // 6.3.17 p.278: for a proxied (exclusive/redirect) source the error
2526 : // vocabulary is fixed — 508 loop, 504 timeout, 404 not found, and
2527 : // "502 Bad Gateway — if the single forwarded request fails for any other
2528 : // reason such as the Context Broker itself having insufficient access
2529 : // rights". 404/504/508 pass through, 409 keeps AlreadyExists semantics
2530 : // (the peer speaks NGSI-LD) and a 207 stays the partial verdict it is;
2531 : // every other failure — auth-class 401/403, a peer's 500/503, a 400 on
2532 : // the inter-broker request — surfaces as 502. The original status stays
2533 : // in `detail` for diagnosis.
2534 104 : let status = if reg.is_proxy()
2535 92 : && !(200..300).contains(&status)
2536 10 : && !matches!(status, 207 | 404 | 409 | 504 | 508)
2537 : {
2538 2 : 502
2539 : } else {
2540 102 : status
2541 : };
2542 104 : Part { status, detail }
2543 104 : }
2544 :
2545 : /// Forward one attribute-level write to every matching registration.
2546 : #[allow(clippy::too_many_arguments)] // mirrors the wire: one param per forwarded request part
2547 54 : pub async fn fed_attr_parts(
2548 54 : st: &AppState,
2549 54 : headers: &HeaderMap,
2550 54 : tenant: &TenantId,
2551 54 : ctx_source: &Value,
2552 54 : regs: &[FedReg],
2553 54 : op: &str,
2554 54 : method: reqwest::Method,
2555 54 : path: &str,
2556 54 : query: &[(String, String)],
2557 54 : body: Option<Value>,
2558 54 : ) -> Vec<Part> {
2559 54 : let ctx_url = ctx_link_url(headers, ctx_source);
2560 54 : let mut parts = Vec::new();
2561 54 : for reg in regs {
2562 : // 5.6.2.4 (and the sibling attribute operations): a proxy-mode
2563 : // registration not supporting the operation is an error of type
2564 : // Conflict and is never contacted; an inclusive one is simply not
2565 : // forwarded.
2566 54 : if !reg.supports(op) {
2567 4 : if reg.is_proxy() {
2568 2 : parts.push(conflict_part(op));
2569 2 : } else {
2570 2 : // status 0 = "not forwarded" sentinel: keeps the parts list
2571 2 : // 1:1 with regs (combine_attr_parts zips them) without
2572 2 : // counting as success or failure.
2573 2 : parts.push(Part {
2574 2 : status: 0,
2575 2 : detail: format!("not forwarded: {op} not supported"),
2576 2 : });
2577 2 : }
2578 4 : continue;
2579 50 : }
2580 50 : parts.push(
2581 50 : forward_part(
2582 50 : st,
2583 50 : method.clone(),
2584 50 : format!("{}/ngsi-ld/v1{path}", reg.endpoint),
2585 50 : query,
2586 50 : headers,
2587 50 : tenant,
2588 50 : reg,
2589 50 : &ctx_url,
2590 50 : body.clone(),
2591 50 : )
2592 50 : .await,
2593 : );
2594 : }
2595 54 : parts
2596 54 : }
2597 :
2598 : /// Conflict part for an exclusive registration that does not accept the op.
2599 18 : pub fn conflict_part(op: &str) -> Part {
2600 18 : Part {
2601 18 : status: 409,
2602 18 : detail: format!("exclusive registration does not accept {op}"),
2603 18 : }
2604 18 : }
2605 :
2606 : /// Remove proxy-covered attributes from an EXPANDED fragment so the local
2607 : /// write never stores exclusively/redirect-registered data (4.3.6.3).
2608 190 : pub fn strip_covered_expanded(fragment: &Value, regs: &[FedReg]) -> Value {
2609 190 : let mut f = fragment.clone();
2610 190 : if let Some(o) = f.as_object_mut() {
2611 190 : let covered: Vec<String> = o
2612 190 : .keys()
2613 194 : .filter(|k| {
2614 2 : !matches!(
2615 194 : k.as_str(),
2616 194 : "id" | "type" | "scope" | "createdAt" | "modifiedAt"
2617 192 : ) && regs.iter().any(|r| r.is_proxy() && r.covers_attr(k))
2618 194 : })
2619 190 : .cloned()
2620 190 : .collect();
2621 190 : for k in covered {
2622 18 : o.remove(&k);
2623 18 : }
2624 0 : }
2625 190 : f
2626 190 : }
2627 :
2628 : /// Strip the members proxied registrations cover from a compacted object;
2629 : /// returns (remainder, had_attrs_left).
2630 122 : pub fn strip_proxied(
2631 122 : obj: &Map<String, Value>,
2632 122 : proxies: &[&FedReg],
2633 122 : ctx: &Context,
2634 122 : ) -> (Map<String, Value>, bool) {
2635 122 : let mut out = Map::new();
2636 122 : let mut any_attr = false;
2637 344 : for (k, v) in obj {
2638 344 : if k == "@context" {
2639 0 : continue;
2640 344 : }
2641 344 : if ["id", "type", "scope"].contains(&k.as_str()) {
2642 224 : out.insert(k.clone(), v.clone());
2643 224 : continue;
2644 120 : }
2645 120 : let iri = ctx.expand_key(k);
2646 : // a proxy only owns the attribute if this ITEM is within its
2647 : // EntityInfo constraints (4.3.6.1) — a type-scoped registration
2648 : // must not strip attributes from an unrelated entity
2649 120 : if proxies
2650 120 : .iter()
2651 120 : .any(|r| r.covers_item(obj, ctx) && r.covers_attr(&iri))
2652 : {
2653 40 : continue;
2654 80 : }
2655 80 : out.insert(k.clone(), v.clone());
2656 80 : any_attr = true;
2657 : }
2658 122 : (out, any_attr)
2659 122 : }
2660 :
2661 : #[cfg(test)]
2662 : mod tests {
2663 : use super::*;
2664 : use serde_json::json;
2665 :
2666 84 : fn hdrs(via: Option<&str>) -> HeaderMap {
2667 84 : let mut h = HeaderMap::new();
2668 84 : if let Some(v) = via {
2669 60 : h.insert("via", v.parse().expect("via"));
2670 60 : }
2671 84 : h
2672 84 : }
2673 :
2674 : /// 4.3.6.6: contextSourceInfo carries the headers a source needs to
2675 : /// answer at all (an API key, say). The minimal registration view built
2676 : /// for the subscription operations dropped them, so every forwarded
2677 : /// subscription create, update and delete went out unauthenticated.
2678 : #[test]
2679 4 : fn the_minimal_registration_view_keeps_context_source_info() {
2680 4 : let doc = json!({
2681 4 : "id": "urn:ngsi-ld:ContextSourceRegistration:csi",
2682 4 : "endpoint": "http://peer:9090",
2683 4 : "contextSourceInfo": [
2684 4 : {"key": "X-API-Key", "value": "s3cret"},
2685 4 : {"key": "jsonldContext", "value": "https://example.org/ctx.jsonld"}
2686 : ]
2687 : });
2688 4 : let reg = fed_reg_of("urn:ngsi-ld:ContextSourceRegistration:csi", &doc);
2689 4 : assert_eq!(
2690 : reg.csi,
2691 4 : vec![
2692 4 : ("X-API-Key".to_owned(), "s3cret".to_owned()),
2693 4 : (
2694 4 : "jsonldContext".to_owned(),
2695 4 : "https://example.org/ctx.jsonld".to_owned()
2696 4 : ),
2697 : ]
2698 : );
2699 4 : }
2700 :
2701 : /// 4.17: `type` is one Entity Type Selection. Splitting it on commas
2702 : /// destroys a conjunction, and the shared evaluator (which csource tests
2703 : /// against real registrations) never sees the expression the client sent.
2704 : #[test]
2705 4 : fn the_type_selection_reaches_registration_matching_whole() {
2706 4 : let ctx = antares_jsonld::Loader::new().core();
2707 4 : let mut params = HashMap::new();
2708 4 : params.insert("type".to_owned(), "(Home;Vehicle),Building".to_owned());
2709 4 : let spec = query_spec(&ctx, ¶ms);
2710 4 : assert_eq!(
2711 : spec.types,
2712 4 : Some(vec!["(Home;Vehicle),Building".to_owned()]),
2713 : "the selection travels as one expression"
2714 : );
2715 4 : }
2716 :
2717 : /// 6.6/6.7: which part of a forwarded URL names the Attribute. The
2718 : /// segment ends at the next `/`, so the `value` sub-resource and a 6.7
2719 : /// instanceId are outside it, and a query string is never part of a path.
2720 : #[test]
2721 4 : fn the_path_attribute_segment_is_the_one_between_attrs_and_the_next_slash() {
2722 36 : let name = |u: &str| path_attr_segment(u).map(|(_, n)| n);
2723 4 : assert_eq!(
2724 4 : name("/entities/urn:x/attrs/speed"),
2725 4 : Some("speed".to_owned())
2726 : );
2727 4 : assert_eq!(
2728 4 : name("/entities/urn:x/attrs/speed/value"),
2729 4 : Some("speed".to_owned())
2730 : );
2731 4 : assert_eq!(
2732 4 : name("/temporal/entities/urn:x/attrs/speed/urn:ngsi-ld:instance:1"),
2733 4 : Some("speed".to_owned())
2734 : );
2735 4 : assert_eq!(
2736 4 : name("/entities/urn:x/attrs/speed?type=Vehicle"),
2737 4 : Some("speed".to_owned())
2738 : );
2739 : // percent-decoded, because that is the form an @context expands
2740 4 : assert_eq!(
2741 4 : name("/entities/urn:x/attrs/http%3A%2F%2Fa.example%2Fs"),
2742 4 : Some("http://a.example/s".to_owned())
2743 : );
2744 : // resources that name no Attribute
2745 4 : assert_eq!(name("/entities/urn:x"), None);
2746 4 : assert_eq!(name("/entities/urn:x/attrs"), None);
2747 4 : assert_eq!(name("/entities/urn:x/attrs/"), None);
2748 4 : assert_eq!(name("/entities/urn:x/attrs/?type=V"), None);
2749 : // the range addresses the ENCODED segment, so writing a translated
2750 : // name back over it cannot corrupt the rest of the URL
2751 4 : let url = "/entities/urn:x/attrs/speed/value?q=1";
2752 4 : let (range, _) = path_attr_segment(url).expect("segment");
2753 4 : assert_eq!(&url[range], "speed");
2754 4 : }
2755 :
2756 : /// 4.3.6.6 recompaction where the request @context and the registered one
2757 : /// are the same document is the identity: the Context Source receives the
2758 : /// Attribute Fragment the client sent, sub-Attributes and all. Any other
2759 : /// answer would rewrite a payload the clause only asked to re-spell.
2760 : #[test]
2761 4 : fn an_attribute_fragment_round_trips_through_one_context_unchanged() {
2762 4 : let ctx = antares_jsonld::core_context();
2763 4 : let frag = serde_json::json!({
2764 4 : "type": "Property",
2765 4 : "value": 56,
2766 4 : "source": {"type": "Property", "value": "Speedometer"},
2767 : });
2768 4 : let exp = antares_jsonld::expand_attr_fragment(frag.as_object().expect("obj"), &ctx)
2769 4 : .expect("the fragment expands");
2770 4 : assert_eq!(
2771 4 : antares_jsonld::compact::compact_instance(&exp, &ctx),
2772 : frag,
2773 : "the same @context in and out must change nothing"
2774 : );
2775 4 : }
2776 :
2777 : /// RFC 3986 clause 3.3: `#`, `?` and `/` end a path segment, so a client
2778 : /// id carrying one would re-target the peer's resource. The characters an
2779 : /// NGSI-LD id legitimately uses (`urn:`, `:`, `-`) must survive unchanged
2780 : /// or every forward would address a different entity than the client did.
2781 : #[test]
2782 4 : fn path_segment_encodes_what_would_end_the_segment() {
2783 4 : assert_eq!(
2784 4 : path_segment("urn:ngsi-ld:Vehicle:A4567-W"),
2785 : "urn:ngsi-ld:Vehicle:A4567-W"
2786 : );
2787 4 : assert_eq!(path_segment("urn:x#"), "urn:x%23");
2788 4 : assert_eq!(path_segment("urn:x?q=1"), "urn:x%3Fq=1");
2789 4 : assert_eq!(path_segment("a/b"), "a%2Fb");
2790 : // an id already carrying a percent must not decode twice at the peer
2791 4 : assert_eq!(path_segment("a%2Fb"), "a%252Fb");
2792 : // `.` is unreserved, so dot segments pass through unchanged here —
2793 : // they are refused at the door instead (EntityId::new, check_attr_name)
2794 4 : assert_eq!(path_segment(".."), "..");
2795 : // non-ASCII is percent-encoded per its UTF-8 bytes
2796 4 : assert_eq!(path_segment("é"), "%C3%A9");
2797 4 : }
2798 :
2799 : /// RFC 7230 received-by is a TOKEN compared for equality:
2800 : /// `ends_with` made alias `b1` match pseudonym `sub-b1` (spurious loop)
2801 : /// and could never catch the converse.
2802 : #[test]
2803 4 : fn via_loop_compares_tokens_not_suffixes() {
2804 4 : assert!(via_loop(&hdrs(Some("1.1 b1")), "b1"));
2805 4 : assert!(via_loop(&hdrs(Some("1.1 b2, 1.1 b1")), "b1"));
2806 4 : assert!(via_loop(&hdrs(Some("HTTP/1.1 b1")), "b1"));
2807 4 : assert!(
2808 4 : !via_loop(&hdrs(Some("1.1 sub-b1")), "b1"),
2809 : "suffix must not match — a former suffix-match false positive"
2810 : );
2811 4 : assert!(!via_loop(&hdrs(Some("1.1 b10")), "b1"));
2812 4 : assert!(!via_loop(&hdrs(None), "b1"));
2813 : // malformed element carrying only a pseudonym still detects
2814 4 : assert!(via_loop(&hdrs(Some("b1")), "b1"));
2815 4 : }
2816 :
2817 52 : fn reg(mode: &str) -> FedReg {
2818 52 : FedReg {
2819 52 : reg_id: "urn:ngsi-ld:ContextSourceRegistration:test".into(),
2820 52 : endpoint: "http://peer:9090".into(),
2821 52 : mode: mode.into(),
2822 52 : ops: vec!["federationOps".into()],
2823 52 : attrs: None,
2824 52 : ent_ids: vec![],
2825 52 : ent_types: vec![],
2826 52 : ent_patterns: vec![],
2827 52 : ent_unrestricted: false,
2828 52 : tenant: None,
2829 52 : alias: None,
2830 52 : csi: vec![],
2831 52 : local_only: false,
2832 52 : timeout_ms: None,
2833 52 : cooldown_ms: None,
2834 52 : }
2835 52 : }
2836 :
2837 : /// 5.12 condition 5, response side: a query idPattern was forwarded to a
2838 : /// pattern-scoped registration BECAUSE the two patterns cannot be
2839 : /// compared — so the answer cannot be refused on the registration's
2840 : /// pattern when the CLIENT's own selection admits the id. An id outside
2841 : /// BOTH scopes stays refused (a peer must not inject unrelated entities
2842 : /// that would win on recency).
2843 : #[test]
2844 4 : fn a_response_id_the_query_selects_survives_the_registration_pattern() {
2845 4 : let mut r = reg("inclusive");
2846 4 : r.ent_patterns = vec!["^urn:ngsi-ld:V:sk_bb:.*$".into()];
2847 4 : let spec = crate::registry::CsrSpec {
2848 4 : id_pattern: Some("^urn:ngsi-ld:V:sk_zvolen:.*$".into()),
2849 4 : ..Default::default()
2850 4 : };
2851 4 : assert!(
2852 4 : admits_import(&r, &spec, "urn:ngsi-ld:V:sk_zvolen:7"),
2853 : "the client's own pattern admits the id the peer answered with"
2854 : );
2855 4 : assert!(
2856 4 : admits_import(&r, &spec, "urn:ngsi-ld:V:sk_bb:1"),
2857 : "the registration's own scope still admits"
2858 : );
2859 4 : assert!(
2860 4 : !admits_import(&r, &spec, "urn:ngsi-ld:V:sk_presov:9"),
2861 : "an id outside both the registration and the query is refused"
2862 : );
2863 : // no query selection at all: only the registration scope admits
2864 4 : let none = crate::registry::CsrSpec::default();
2865 4 : assert!(!admits_import(&r, &none, "urn:ngsi-ld:V:sk_zvolen:7"));
2866 : // exact query ids admit exactly themselves
2867 4 : let exact = crate::registry::CsrSpec {
2868 4 : ids: Some(vec!["urn:ngsi-ld:V:sk_zvolen:7".into()]),
2869 4 : ..Default::default()
2870 4 : };
2871 4 : assert!(admits_import(&r, &exact, "urn:ngsi-ld:V:sk_zvolen:7"));
2872 4 : assert!(!admits_import(&r, &exact, "urn:ngsi-ld:V:sk_zvolen:8"));
2873 4 : }
2874 :
2875 : /// 4.3.6.1/5.12: an idPattern-scoped registration gates payload items
2876 : /// exactly like an exact-id one — a foreign-razidlo item is not this
2877 : /// source's data; an id-less fragment cannot be disproven.
2878 : #[test]
2879 4 : fn covers_item_honours_entityinfo_id_patterns() {
2880 4 : let mut r = reg("redirect");
2881 4 : r.ent_patterns = vec!["^urn:ngsi-ld:V:sk_bb:.*$".into()];
2882 4 : let st = AppState::new("me".into());
2883 4 : let ctx = st.loader.core();
2884 12 : let item = |id: Option<&str>| {
2885 12 : let mut m = Map::new();
2886 12 : if let Some(id) = id {
2887 8 : m.insert("id".into(), Value::String(id.into()));
2888 8 : }
2889 12 : m
2890 12 : };
2891 4 : assert!(r.covers_item(&item(Some("urn:ngsi-ld:V:sk_bb:1")), &ctx));
2892 4 : assert!(
2893 4 : !r.covers_item(&item(Some("urn:ngsi-ld:V:sk_po:1")), &ctx),
2894 : "a foreign-razidlo item must not be covered"
2895 : );
2896 4 : assert!(
2897 4 : r.covers_item(&item(None), &ctx),
2898 : "id-less fragments stay covered"
2899 : );
2900 4 : }
2901 :
2902 : /// Table 5.2.40-1: "In the multi-tenancy use case (see clause 4.14), this
2903 : /// id shall be identifying a specific Tenant within a registered Context
2904 : /// Source." One alias for every tenant of a broker makes cross-tenant
2905 : /// federation inside that broker look like a loop.
2906 : #[test]
2907 4 : fn alias_identifies_the_tenant_not_just_the_broker() {
2908 20 : let tenant = |s: &str| antares_model::TenantId::new(s).expect("tenant");
2909 : // the default tenant keeps the bare alias — 6.3.14's own convention
2910 : // (the header is omitted, not sent as "default"), and the wire format
2911 : // every single-tenant peer already registered
2912 4 : assert_eq!(alias_for("antares1", &tenant("default")), "antares1");
2913 4 : assert_eq!(alias_for("antares1", &tenant("zvolen")), "antares1~zvolen");
2914 : // a chain naming this broker in ANOTHER tenant is not a loop: the
2915 : // registration points at a different (source, tenant) pair
2916 4 : let h = hdrs(Some("1.1 antares1~zvolen"));
2917 4 : assert!(via_loop(&h, &alias_for("antares1", &tenant("zvolen"))));
2918 4 : assert!(!via_loop(
2919 4 : &h,
2920 4 : &alias_for("antares1", &tenant("banskabystrica"))
2921 4 : ));
2922 4 : assert!(
2923 4 : !via_loop(&h, &alias_for("antares1", &tenant("default"))),
2924 : "the default tenant of this broker is its own Context Source"
2925 : );
2926 : // `~` cannot occur in a TenantId and is rejected in a configured
2927 : // alias, so the two halves can never blur into each other
2928 4 : assert!(antares_model::TenantId::new("a~b").is_err());
2929 4 : }
2930 :
2931 : /// Table 6.3.18-2: "the listing of previously encountered Context Sources
2932 : /// supplied is used when determining matching registrations", and 5.2.9
2933 : /// gives a registration the peer's `contextSourceAlias` "which is used to
2934 : /// identify loops". A source already in the chain is therefore not a
2935 : /// match — and the tenant-specific alias keeps that per (source, tenant).
2936 : #[tokio::test]
2937 4 : async fn registered_alias_in_the_via_chain_is_not_a_matching_registration() {
2938 4 : let st = AppState::new("me".into());
2939 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
2940 4 : let ctx = st.loader.core();
2941 12 : for (id, alias) in [
2942 4 : ("urn:ngsi-ld:ContextSourceRegistration:visited", "peer1"),
2943 4 : ("urn:ngsi-ld:ContextSourceRegistration:fresh", "peer2"),
2944 4 : ("urn:ngsi-ld:ContextSourceRegistration:anon", ""),
2945 4 : ] {
2946 12 : let mut doc = json!({
2947 12 : "id": id,
2948 12 : "type": "ContextSourceRegistration",
2949 12 : "endpoint": "http://peer:9090",
2950 12 : "information": [{"entities": [{"type": "https://uri.etsi.org/ngsi-ld/default-context/Vehicle"}]}],
2951 : });
2952 12 : if !alias.is_empty() {
2953 8 : doc["contextSourceAlias"] = json!(alias);
2954 8 : }
2955 12 : st.store
2956 12 : .create(&tenant, Kind::Registration, id, doc)
2957 12 : .await
2958 12 : .expect("seed registration");
2959 : }
2960 4 : let spec = crate::registry::CsrSpec {
2961 4 : types: Some(vec![
2962 4 : "https://uri.etsi.org/ngsi-ld/default-context/Vehicle".into()
2963 4 : ]),
2964 4 : ..Default::default()
2965 4 : };
2966 4 : let all = matching_regs(&st, &tenant, &spec, &ctx, &hdrs(None))
2967 4 : .await
2968 4 : .expect("registrations");
2969 4 : assert_eq!(all.len(), 3, "no Via ⇒ every registration matches");
2970 4 : let via = hdrs(Some("1.1 peer1"));
2971 4 : let left = matching_regs(&st, &tenant, &spec, &ctx, &via)
2972 4 : .await
2973 4 : .expect("registrations");
2974 8 : let ids: Vec<&str> = left.iter().map(|r| r.reg_id.as_str()).collect();
2975 4 : assert!(
2976 4 : !ids.contains(&"urn:ngsi-ld:ContextSourceRegistration:visited"),
2977 : "a source already in the Via chain must not match"
2978 : );
2979 4 : assert_eq!(
2980 4 : ids.len(),
2981 4 : 2,
2982 4 : "an unvisited peer and a registration with no alias still match"
2983 4 : );
2984 4 : }
2985 :
2986 : /// 6.3.17 p.278 scopes 508 to "an exclusive or redirect registration,
2987 : /// where all of the data is held ... in a single registered source";
2988 : /// any other loop clears the forward set and proceeds locally
2989 : /// (Table 6.3.18-2: the Via listing amends registration matching).
2990 : #[test]
2991 4 : fn loop_508_only_for_a_single_proxy_registration() {
2992 4 : let t = antares_model::TenantId::new("default").expect("tenant");
2993 4 : let h = hdrs(Some("1.1 me"));
2994 : // single exclusive source looping back → 508
2995 4 : let mut regs = vec![reg("exclusive")];
2996 4 : assert!(handle_via_loop(&h, "me", &t, &mut regs).is_some());
2997 4 : let mut regs = vec![reg("redirect")];
2998 4 : assert!(handle_via_loop(&h, "me", &t, &mut regs).is_some());
2999 : // inclusive loop → no 508, forwards cleared, local execution proceeds
3000 4 : let mut regs = vec![reg("inclusive")];
3001 4 : assert!(handle_via_loop(&h, "me", &t, &mut regs).is_none());
3002 4 : assert!(regs.is_empty(), "looping forwards must be dropped");
3003 : // a mixed set is not "a single registered source"
3004 4 : let mut regs = vec![reg("exclusive"), reg("inclusive")];
3005 4 : assert!(handle_via_loop(&h, "me", &t, &mut regs).is_none());
3006 4 : assert!(regs.is_empty());
3007 : // no loop → untouched
3008 4 : let mut regs = vec![reg("exclusive")];
3009 4 : assert!(handle_via_loop(&hdrs(None), "me", &t, &mut regs).is_none());
3010 4 : assert_eq!(regs.len(), 1);
3011 4 : }
3012 :
3013 : /// 4.3.6.4 / 5.2.9 localOnly: "distributed operations associated to this
3014 : /// Context Source Registration will act only on data held directly by
3015 : /// the registered Context Source itself" — the flag must survive
3016 : /// registration compilation so every forward can carry local=true.
3017 : #[tokio::test]
3018 4 : async fn local_only_survives_registration_compilation() {
3019 4 : let st = AppState::new("me".into());
3020 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
3021 4 : let ctx = st.loader.core();
3022 8 : for (id, local_only) in [
3023 4 : ("urn:ngsi-ld:ContextSourceRegistration:lo", true),
3024 4 : ("urn:ngsi-ld:ContextSourceRegistration:casc", false),
3025 4 : ] {
3026 8 : let mut doc = json!({
3027 8 : "id": id,
3028 8 : "type": "ContextSourceRegistration",
3029 8 : "endpoint": "http://peer:9090",
3030 8 : "information": [{"entities": [{"type": "https://uri.etsi.org/ngsi-ld/default-context/Vehicle"}]}],
3031 : });
3032 8 : if local_only {
3033 4 : doc["localOnly"] = json!(true);
3034 4 : }
3035 8 : st.store
3036 8 : .create(&tenant, Kind::Registration, id, doc)
3037 8 : .await
3038 8 : .expect("seed registration");
3039 : }
3040 4 : let spec = crate::registry::CsrSpec {
3041 4 : types: Some(vec![
3042 4 : "https://uri.etsi.org/ngsi-ld/default-context/Vehicle".into()
3043 4 : ]),
3044 4 : ..Default::default()
3045 4 : };
3046 4 : let regs = matching_regs(&st, &tenant, &spec, &ctx, &HeaderMap::new())
3047 4 : .await
3048 4 : .expect("registrations");
3049 8 : let lo = |id: &str| {
3050 8 : regs.iter()
3051 12 : .find(|r| r.reg_id == id)
3052 8 : .expect("compiled")
3053 : .local_only
3054 8 : };
3055 4 : assert!(lo("urn:ngsi-ld:ContextSourceRegistration:lo"));
3056 4 : assert!(!lo("urn:ngsi-ld:ContextSourceRegistration:casc"));
3057 4 : }
3058 :
3059 : /// Table 5.2.34-1: management.localOnly — "distributed operations
3060 : /// associated to this Context Source Registration will act only on data
3061 : /// held directly by the registered Context Source itself".
3062 : #[tokio::test]
3063 4 : async fn management_local_only_survives_registration_compilation() {
3064 4 : let st = AppState::new("me".into());
3065 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
3066 4 : let ctx = st.loader.core();
3067 4 : let id = "urn:ngsi-ld:ContextSourceRegistration:mgmt-lo";
3068 4 : let doc = json!({
3069 4 : "id": id,
3070 4 : "type": "ContextSourceRegistration",
3071 4 : "endpoint": "http://peer:9090",
3072 4 : "information": [{"entities": [{"type": "https://uri.etsi.org/ngsi-ld/default-context/Vehicle"}]}],
3073 4 : "management": {"localOnly": true}
3074 : });
3075 4 : st.store
3076 4 : .create(&tenant, Kind::Registration, id, doc)
3077 4 : .await
3078 4 : .expect("seed registration");
3079 4 : let spec = crate::registry::CsrSpec {
3080 4 : types: Some(vec![
3081 4 : "https://uri.etsi.org/ngsi-ld/default-context/Vehicle".into()
3082 4 : ]),
3083 4 : ..Default::default()
3084 4 : };
3085 4 : let regs = matching_regs(&st, &tenant, &spec, &ctx, &HeaderMap::new())
3086 4 : .await
3087 4 : .expect("registrations");
3088 4 : assert!(
3089 4 : regs.iter()
3090 4 : .find(|r| r.reg_id == id)
3091 4 : .expect("compiled")
3092 4 : .local_only,
3093 4 : "management.localOnly must compile into the forward flag"
3094 4 : );
3095 4 : }
3096 :
3097 : /// 4.3.6.2: "An auxiliary Context Source Registration never overrides
3098 : /// data held directly within a Context Broker" — supplementary attributes
3099 : /// are included, conflicting ones lose to the base regardless of recency.
3100 : #[test]
3101 4 : fn auxiliary_merge_supplements_but_never_overrides() {
3102 4 : let attr = "https://uri.etsi.org/ngsi-ld/default-context/speed";
3103 4 : let extra = "https://uri.etsi.org/ngsi-ld/default-context/color";
3104 4 : let mut base = json!({
3105 4 : "id": "urn:x", "type": ["T"],
3106 4 : attr: [{"type": "Property", "value": 1, "modifiedAt": "2020-01-01T00:00:00Z"}]
3107 : });
3108 : // aux instance is FRESHER and would win a 4.5.5.3 recency merge —
3109 : // auxiliary mode must still lose the conflict, yet supplement `color`
3110 4 : let add = json!({
3111 4 : "id": "urn:x", "type": ["T"],
3112 4 : attr: [{"type": "Property", "value": 2, "modifiedAt": "2026-01-01T00:00:00Z"}],
3113 4 : extra: [{"type": "Property", "value": "red"}]
3114 : });
3115 4 : merge_docs(&mut base, &add, true);
3116 4 : assert_eq!(base[attr][0]["value"], 1, "aux must not override local");
3117 4 : assert_eq!(base[extra][0]["value"], "red", "aux supplement is included");
3118 : // the same add as a non-aux inclusive source DOES win on recency
3119 4 : let mut base2 = json!({
3120 4 : "id": "urn:x", "type": ["T"],
3121 4 : attr: [{"type": "Property", "value": 1, "modifiedAt": "2020-01-01T00:00:00Z"}]
3122 : });
3123 4 : merge_docs(&mut base2, &add, false);
3124 4 : assert_eq!(base2[attr][0]["value"], 2);
3125 4 : }
3126 :
3127 : /// 4.20: retrieveEntity implements only 5.7.1 — a source offering it
3128 : /// alone is never a query target; queryEntity/queryBatch are.
3129 : #[test]
3130 4 : fn query_op_requires_query_support() {
3131 4 : let reg =
3132 20 : |ops: &[&str]| fed_reg_of("urn:r", &json!({"endpoint": "http://x", "operations": ops}));
3133 4 : assert_eq!(reg(&["retrieveEntity"]).query_op(), None);
3134 4 : assert_eq!(reg(&["queryEntity"]).query_op(), Some("queryEntity"));
3135 4 : assert_eq!(reg(&["queryBatch"]).query_op(), Some("queryBatch"));
3136 4 : assert_eq!(reg(&["federationOps"]).query_op(), Some("queryEntity"));
3137 4 : assert_eq!(reg(&["retrieveOps"]).query_op(), Some("queryEntity"));
3138 4 : }
3139 :
3140 12 : async fn seed_reg(
3141 12 : st: &AppState,
3142 12 : tenant: &antares_model::TenantId,
3143 12 : id: &str,
3144 12 : mode: &str,
3145 12 : entities: Value,
3146 12 : ) {
3147 12 : let doc = json!({
3148 12 : "id": id,
3149 12 : "type": "ContextSourceRegistration",
3150 12 : "mode": mode,
3151 12 : "operations": ["redirectionOps"],
3152 12 : "endpoint": "http://peer:9090",
3153 12 : "information": [{"entities": entities}],
3154 : });
3155 12 : st.store
3156 12 : .create(tenant, Kind::Registration, id, doc)
3157 12 : .await
3158 12 : .expect("seed registration");
3159 12 : }
3160 :
3161 : /// 5.2.9 / 5.12: an `idPattern` in the registration's EntityInfo matches
3162 : /// a write addressed by id, so the write forwards to that source; a
3163 : /// pattern the id does not match keeps the write local.
3164 : #[tokio::test]
3165 4 : async fn write_plan_forwards_a_write_by_id_to_an_id_pattern_registration() {
3166 4 : let st = AppState::new("me".into());
3167 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
3168 4 : let ctx = st.loader.core();
3169 4 : seed_reg(
3170 4 : &st,
3171 4 : &tenant,
3172 4 : "urn:ngsi-ld:ContextSourceRegistration:cars",
3173 4 : "exclusive",
3174 4 : json!([{"idPattern": "urn:ngsi-ld:Vehicle:.*"}]),
3175 4 : )
3176 4 : .await;
3177 4 : seed_reg(
3178 4 : &st,
3179 4 : &tenant,
3180 4 : "urn:ngsi-ld:ContextSourceRegistration:bikes",
3181 4 : "exclusive",
3182 4 : json!([{"idPattern": "urn:ngsi-ld:Bike:.*"}]),
3183 4 : )
3184 4 : .await;
3185 4 : let spec = crate::registry::CsrSpec {
3186 4 : ids: Some(vec!["urn:ngsi-ld:Vehicle:1".into()]),
3187 4 : ..Default::default()
3188 4 : };
3189 4 : let plan = write_plan(
3190 4 : &st,
3191 4 : &tenant,
3192 4 : &spec,
3193 4 : &ctx,
3194 4 : &HashMap::new(),
3195 4 : &HeaderMap::new(),
3196 4 : )
3197 4 : .await
3198 4 : .expect("plan");
3199 4 : let WritePlan::Forward(regs) = plan else {
3200 0 : panic!("no Via chain, so nothing answers early");
3201 : };
3202 4 : let ids: Vec<&str> = regs.iter().map(|r| r.reg_id.as_str()).collect();
3203 4 : assert_eq!(ids, vec!["urn:ngsi-ld:ContextSourceRegistration:cars"]);
3204 4 : }
3205 :
3206 : /// 6.3.18: a `Via` chain that already names this broker is answered by
3207 : /// the chain rule before any registration is used — with a single
3208 : /// exclusive source that is the 508 of Table 6.3.18-2 — and a chain
3209 : /// past the hop cap is 508 regardless of what matched.
3210 : #[tokio::test]
3211 4 : async fn write_plan_answers_a_via_loop_before_forwarding() {
3212 4 : let st = AppState::new("me".into());
3213 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
3214 4 : let ctx = st.loader.core();
3215 4 : seed_reg(
3216 4 : &st,
3217 4 : &tenant,
3218 4 : "urn:ngsi-ld:ContextSourceRegistration:cars",
3219 4 : "exclusive",
3220 4 : json!([{"idPattern": "urn:ngsi-ld:Vehicle:.*"}]),
3221 4 : )
3222 4 : .await;
3223 4 : let spec = crate::registry::CsrSpec {
3224 4 : ids: Some(vec!["urn:ngsi-ld:Vehicle:1".into()]),
3225 4 : ..Default::default()
3226 4 : };
3227 4 : let looped = write_plan(
3228 4 : &st,
3229 4 : &tenant,
3230 4 : &spec,
3231 4 : &ctx,
3232 4 : &HashMap::new(),
3233 4 : &hdrs(Some("1.1 me")),
3234 4 : )
3235 4 : .await
3236 4 : .expect("plan");
3237 4 : let WritePlan::Answered(resp) = looped else {
3238 0 : panic!("a Via chain naming this broker with one exclusive source is answered, not forwarded");
3239 : };
3240 4 : assert_eq!(resp.status(), 508);
3241 4 : let deep: String = (0..=MAX_VIA_HOPS)
3242 132 : .map(|i| format!("1.1 hop{i}"))
3243 4 : .collect::<Vec<_>>()
3244 4 : .join(", ");
3245 4 : let WritePlan::Answered(resp) = write_plan(
3246 4 : &st,
3247 4 : &tenant,
3248 4 : &spec,
3249 4 : &ctx,
3250 4 : &HashMap::new(),
3251 4 : &hdrs(Some(&deep)),
3252 4 : )
3253 4 : .await
3254 4 : .expect("plan") else {
3255 0 : panic!("a chain past the hop cap is refused outright");
3256 : };
3257 4 : assert_eq!(resp.status(), 508);
3258 : // no chain, no early answer: the same registration forwards
3259 4 : assert!(matches!(
3260 4 : write_plan(&st, &tenant, &spec, &ctx, &HashMap::new(), &HeaderMap::new()).await.expect("plan"),
3261 4 : WritePlan::Forward(regs) if regs.len() == 1
3262 4 : ));
3263 4 : }
3264 :
3265 : /// 4.3.6.2: "Auxiliary distributed operations are limited to context
3266 : /// information consumption operations" — write_regs must drop a matching
3267 : /// auxiliary registration while keeping an inclusive one.
3268 : #[tokio::test]
3269 4 : async fn write_regs_exclude_auxiliary_registrations() {
3270 4 : let st = AppState::new("me".into());
3271 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
3272 4 : let ctx = st.loader.core();
3273 8 : for (id, mode) in [
3274 4 : ("urn:ngsi-ld:ContextSourceRegistration:aux", "auxiliary"),
3275 4 : ("urn:ngsi-ld:ContextSourceRegistration:inc", "inclusive"),
3276 4 : ] {
3277 8 : let doc = json!({
3278 8 : "id": id,
3279 8 : "type": "ContextSourceRegistration",
3280 8 : "mode": mode,
3281 8 : "operations": ["redirectionOps"],
3282 8 : "endpoint": "http://peer:9090",
3283 8 : "information": [{"entities": [{"type": "https://uri.etsi.org/ngsi-ld/default-context/Vehicle"}]}],
3284 : });
3285 8 : st.store
3286 8 : .create(&tenant, Kind::Registration, id, doc)
3287 8 : .await
3288 8 : .expect("seed registration");
3289 : }
3290 4 : let spec = crate::registry::CsrSpec {
3291 4 : types: Some(vec![
3292 4 : "https://uri.etsi.org/ngsi-ld/default-context/Vehicle".into()
3293 4 : ]),
3294 4 : ..Default::default()
3295 4 : };
3296 4 : let regs = write_regs(
3297 4 : &st,
3298 4 : &tenant,
3299 4 : &spec,
3300 4 : &ctx,
3301 4 : &HashMap::new(),
3302 4 : &HeaderMap::new(),
3303 4 : )
3304 4 : .await
3305 4 : .expect("registrations");
3306 4 : let ids: Vec<&str> = regs.iter().map(|r| r.reg_id.as_str()).collect();
3307 4 : assert_eq!(ids, vec!["urn:ngsi-ld:ContextSourceRegistration:inc"]);
3308 4 : }
3309 :
3310 : /// 4.3.6.1: "Context Brokers shall respect" a Context Source's declared
3311 : /// operations subset — explicit operation names and operation groups
3312 : /// (5.2.9) both gate; anything outside the list must not be forwarded.
3313 : #[test]
3314 4 : fn operations_subset_gates_forwarding() {
3315 4 : let mut r = reg("inclusive"); // ops = ["federationOps"], the 5.2.9 default
3316 4 : assert!(r.supports("queryEntity"));
3317 4 : assert!(r.supports("createSubscription"));
3318 4 : assert!(
3319 4 : !r.supports("createEntity"),
3320 : "federationOps carries no provision operations"
3321 : );
3322 4 : r.ops = vec!["updateOps".into()];
3323 4 : assert!(r.supports("updateAttrs"));
3324 4 : assert!(!r.supports("queryEntity"));
3325 4 : r.ops = vec!["createEntity".into()];
3326 4 : assert!(r.supports("createEntity"), "explicit op name matches");
3327 4 : assert!(!r.supports("deleteEntity"));
3328 4 : r.ops = vec![];
3329 4 : assert!(!r.supports("queryEntity"), "empty subset forwards nothing");
3330 4 : }
3331 :
3332 : /// 4.3.6.1: "all constraints specified in the registration shall be
3333 : /// respected" — a forwarded fragment is reduced to the attributes the
3334 : /// RegistrationInfo covers; when nothing covered remains there is no
3335 : /// forward at all (None).
3336 : #[test]
3337 4 : fn forwarded_fragment_reduced_to_registration_scope() {
3338 4 : let st = AppState::new("me".into());
3339 4 : let ctx = st.loader.core();
3340 4 : let speed = "https://uri.etsi.org/ngsi-ld/default-context/speed";
3341 4 : let mut r = reg("inclusive");
3342 4 : r.attrs = Some(vec![speed.into()]);
3343 4 : let obj = json!({
3344 4 : "id": "urn:x", "type": "Vehicle",
3345 4 : "speed": {"type": "Property", "value": 3},
3346 4 : "color": {"type": "Property", "value": "red"},
3347 4 : "@context": "https://example.org/ctx.jsonld"
3348 : });
3349 4 : let out = reduce_to_scope(obj.as_object().expect("obj"), &r, &ctx).expect("covered");
3350 4 : let out = out.as_object().expect("out");
3351 4 : assert!(out.contains_key("speed"));
3352 4 : assert!(out.contains_key("id") && out.contains_key("type"));
3353 4 : assert!(
3354 4 : !out.contains_key("color"),
3355 : "uncovered attribute must be dropped"
3356 : );
3357 4 : assert!(!out.contains_key("@context"));
3358 : // nothing covered ⇒ no forwarded fragment
3359 4 : let only_color = json!({
3360 4 : "id": "urn:x", "type": "Vehicle",
3361 4 : "color": {"type": "Property", "value": "red"}
3362 : });
3363 4 : assert!(reduce_to_scope(only_color.as_object().expect("obj"), &r, &ctx).is_none());
3364 : // an unscoped registration (attrs: None) passes everything but @context
3365 4 : r.attrs = None;
3366 4 : let full = reduce_to_scope(obj.as_object().expect("obj"), &r, &ctx).expect("all");
3367 4 : let full = full.as_object().expect("full");
3368 4 : assert!(full.contains_key("color") && !full.contains_key("@context"));
3369 4 : }
3370 :
3371 : /// 4.5.5.3 p.60: "if an expiresAt DateTime is present on the
3372 : /// Attribute and the date lies in the past, it shall be discarded" —
3373 : /// BEFORE the observedAt/modifiedAt recency comparison.
3374 : #[test]
3375 4 : fn merge_discards_expired_instances_before_recency() {
3376 4 : let attr = "https://uri.etsi.org/ngsi-ld/default-context/speed";
3377 4 : let mut base = json!({
3378 4 : "id": "urn:x", "type": ["T"],
3379 4 : attr: [{"type": "Property", "value": 1, "modifiedAt": "2026-01-01T00:00:00Z"}]
3380 : });
3381 : // fresher instance, but expired → must NOT win
3382 4 : let add = json!({
3383 4 : "id": "urn:x", "type": ["T"],
3384 4 : attr: [{"type": "Property", "value": 2,
3385 4 : "modifiedAt": "2026-06-01T00:00:00Z",
3386 4 : "expiresAt": "2020-01-01T00:00:00Z"}]
3387 : });
3388 4 : merge_docs(&mut base, &add, false);
3389 4 : assert_eq!(base[attr][0]["value"], 1, "expired instance was discarded");
3390 : // an expired BASE instance loses to a live remote one even if fresher
3391 4 : let mut base2 = json!({
3392 4 : "id": "urn:x", "type": ["T"],
3393 4 : attr: [{"type": "Property", "value": 1,
3394 4 : "modifiedAt": "2026-06-01T00:00:00Z",
3395 4 : "expiresAt": "2020-01-01T00:00:00Z"}]
3396 : });
3397 4 : let add2 = json!({
3398 4 : "id": "urn:x", "type": ["T"],
3399 4 : attr: [{"type": "Property", "value": 2, "modifiedAt": "2026-01-01T00:00:00Z"}]
3400 : });
3401 4 : merge_docs(&mut base2, &add2, false);
3402 4 : assert_eq!(base2[attr][0]["value"], 2, "live instance replaces expired");
3403 4 : }
3404 :
3405 : /// 4.5.5.3: entity-level expiresAt — missing from at least one received
3406 : /// version → removed; present in all versions → furthest in the future.
3407 : #[test]
3408 4 : fn merge_entity_expires_at_intersection_and_max() {
3409 4 : let mut base = json!({"id": "urn:x", "type": ["T"], "expiresAt": "2030-01-01T00:00:00Z"});
3410 4 : let add = json!({"id": "urn:x", "type": ["T"], "expiresAt": "2031-01-01T00:00:00Z"});
3411 4 : merge_docs(&mut base, &add, false);
3412 4 : assert_eq!(base["expiresAt"], "2031-01-01T00:00:00Z");
3413 : // one version without expiresAt → removed
3414 4 : let add2 = json!({"id": "urn:x", "type": ["T"]});
3415 4 : merge_docs(&mut base, &add2, false);
3416 4 : assert!(base.get("expiresAt").is_none(), "expiresAt must be removed");
3417 : // and never re-introduced by a later version that has one
3418 4 : let add3 = json!({"id": "urn:x", "type": ["T"], "expiresAt": "2032-01-01T00:00:00Z"});
3419 4 : merge_docs(&mut base, &add3, false);
3420 4 : assert!(base.get("expiresAt").is_none());
3421 4 : }
3422 :
3423 : /// 4.3.6.2: "An auxiliary Context Source Registration never overrides
3424 : /// data held directly within a Context Broker." The entity-level
3425 : /// expiresAt reconciliation is part of that data, so an auxiliary
3426 : /// version can neither remove it nor push it further out.
3427 : #[test]
3428 4 : fn auxiliary_merge_never_touches_entity_expires_at() {
3429 4 : let mut base = json!({"id": "urn:x", "type": ["T"], "expiresAt": "2030-01-01T00:00:00Z"});
3430 4 : let aux_without = json!({"id": "urn:x", "type": ["T"]});
3431 4 : merge_docs(&mut base, &aux_without, true);
3432 4 : assert_eq!(
3433 4 : base["expiresAt"], "2030-01-01T00:00:00Z",
3434 : "an auxiliary version lacking expiresAt must not remove the broker's own"
3435 : );
3436 4 : let aux_later = json!({"id": "urn:x", "type": ["T"], "expiresAt": "2031-01-01T00:00:00Z"});
3437 4 : merge_docs(&mut base, &aux_later, true);
3438 4 : assert_eq!(
3439 4 : base["expiresAt"], "2030-01-01T00:00:00Z",
3440 : "an auxiliary version must not extend the broker's own expiresAt"
3441 : );
3442 4 : }
3443 :
3444 : /// 4.5.5.3 arbitrates on the most recent DateTime, and 4.6.3 lets the
3445 : /// same instant be written with or without a fraction — so the winner
3446 : /// must be chosen on the instant, never on the spelling.
3447 : #[test]
3448 4 : fn recency_arbitrates_on_the_instant_not_the_spelling() {
3449 4 : let attr = "https://uri.etsi.org/ngsi-ld/default-context/speed";
3450 4 : let mut base = json!({
3451 4 : "id": "urn:x", "type": ["T"],
3452 4 : attr: [{"type": "Property", "value": 1, "observedAt": "2026-01-01T00:00:01Z"}]
3453 : });
3454 4 : let add = json!({
3455 4 : "id": "urn:x", "type": ["T"],
3456 4 : attr: [{"type": "Property", "value": 2, "observedAt": "2026-01-01T00:00:01.500Z"}]
3457 : });
3458 4 : merge_docs(&mut base, &add, false);
3459 4 : assert_eq!(
3460 4 : base[attr][0]["value"], 2,
3461 : "the later instant wins even though its string sorts lower"
3462 : );
3463 : // and the converse: a fraction that is EARLIER must not win
3464 4 : let older = json!({
3465 4 : "id": "urn:x", "type": ["T"],
3466 4 : attr: [{"type": "Property", "value": 3, "observedAt": "2026-01-01T00:00:01.250Z"}]
3467 : });
3468 4 : merge_docs(&mut base, &older, false);
3469 4 : assert_eq!(base[attr][0]["value"], 2, "an earlier instant must not win");
3470 4 : }
3471 :
3472 : /// 4.5.5.2/4.5.7: the registration-scope filter narrows ATTRIBUTES, so
3473 : /// the entity-level lifetime members must cross it — dropping expiresAt
3474 : /// here made the 4.5.5.3 reconciliation delete the local one.
3475 : #[test]
3476 4 : fn scoped_import_keeps_the_entity_level_lifetime_members() {
3477 4 : let speed = "https://uri.etsi.org/ngsi-ld/default-context/speed";
3478 4 : let reg = FedReg {
3479 4 : attrs: Some(vec![speed.to_owned()]),
3480 4 : ..FedReg::default()
3481 4 : };
3482 4 : let remote = json!({
3483 4 : "id": "urn:ngsi-ld:Vehicle:1",
3484 4 : "type": "Vehicle",
3485 4 : "expiresAt": "2030-01-01T00:00:00Z",
3486 4 : "speed": {"type": "Property", "value": 1},
3487 4 : "brandName": {"type": "Property", "value": "x"}
3488 : });
3489 4 : let ctx = antares_jsonld::Loader::new().core();
3490 4 : let imported = import_entity(&remote, ®, &ctx).expect("import");
3491 4 : assert_eq!(
3492 4 : imported["expiresAt"], "2030-01-01T00:00:00Z",
3493 : "entity-level expiresAt must survive the scope filter"
3494 : );
3495 4 : assert!(
3496 4 : imported
3497 4 : .get("https://uri.etsi.org/ngsi-ld/default-context/brandName")
3498 4 : .is_none(),
3499 : "an out-of-scope attribute must not be imported"
3500 : );
3501 4 : }
3502 :
3503 : /// 4.5.5.2: a received version's entity-level expiresAt is pushed onto
3504 : /// each Attribute instance — added where absent, capped where the
3505 : /// Attribute's own expiresAt lies further in the future.
3506 : #[test]
3507 4 : fn merge_pushes_entity_expires_at_onto_attributes() {
3508 4 : let attr = "https://uri.etsi.org/ngsi-ld/default-context/speed";
3509 4 : let mut base = json!({"id": "urn:x", "type": ["T"]});
3510 4 : let add = json!({
3511 4 : "id": "urn:x", "type": ["T"],
3512 4 : "expiresAt": "2030-01-01T00:00:00Z",
3513 4 : attr: [
3514 4 : {"type": "Property", "value": 1},
3515 4 : {"type": "Property", "value": 2, "datasetId": "urn:ngsi-ld:Dataset:1",
3516 4 : "expiresAt": "2035-01-01T00:00:00Z"},
3517 4 : {"type": "Property", "value": 3, "datasetId": "urn:ngsi-ld:Dataset:2",
3518 4 : "expiresAt": "2029-01-01T00:00:00Z"}
3519 : ]
3520 : });
3521 4 : merge_docs(&mut base, &add, false);
3522 4 : let inst = base[attr].as_array().expect("attr array");
3523 4 : assert_eq!(inst[0]["expiresAt"], "2030-01-01T00:00:00Z", "added");
3524 4 : assert_eq!(inst[1]["expiresAt"], "2030-01-01T00:00:00Z", "capped");
3525 4 : assert_eq!(inst[2]["expiresAt"], "2029-01-01T00:00:00Z", "earlier kept");
3526 4 : }
3527 :
3528 : /// 6.3.17: "the error response should be as informative as possible" —
3529 : /// informative about the OPERATION, not about the deployment. The part
3530 : /// that failed is identified by its Context Source Registration id
3531 : /// (5.2.9), which the client can retrieve from /csourceRegistrations;
3532 : /// the registration `endpoint` is not part of any client-facing payload,
3533 : /// and a client able to provoke a partial failure must not be able to
3534 : /// enumerate the address of every registered Context Source.
3535 : #[tokio::test]
3536 4 : async fn partial_failure_detail_omits_the_peer_endpoint() {
3537 : use std::io::{Read, Write};
3538 4 : crate::allow_private();
3539 : // a Context Source that refuses every forwarded write
3540 4 : let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
3541 4 : let port = listener.local_addr().expect("addr").port();
3542 4 : std::thread::spawn(move || {
3543 4 : for stream in listener.incoming() {
3544 4 : let Ok(mut s) = stream else { continue };
3545 4 : let mut buf = [0u8; 4096];
3546 4 : let _ = s.read(&mut buf);
3547 4 : let _ = s.write_all(
3548 4 : b"HTTP/1.1 400 Bad Request\r\nConnection: close\r\nContent-Length: 0\r\n\r\n",
3549 4 : );
3550 : }
3551 4 : });
3552 4 : let st = AppState::new("me".into());
3553 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
3554 4 : let mut r = reg("inclusive");
3555 4 : r.endpoint = format!("http://127.0.0.1:{port}");
3556 4 : let part = forward_part(
3557 4 : &st,
3558 4 : reqwest::Method::POST,
3559 4 : format!("{}/ngsi-ld/v1/entities", r.endpoint),
3560 4 : &[],
3561 4 : &HeaderMap::new(),
3562 4 : &tenant,
3563 4 : &r,
3564 4 : antares_jsonld::CORE_CONTEXT,
3565 4 : Some(json!({"id": "urn:ngsi-ld:V:1", "type": "Vehicle"})),
3566 4 : )
3567 4 : .await;
3568 4 : assert_eq!(part.status, 400, "the peer refused the write");
3569 : // one failed forward + one succeeded local part ⇒ 207 Multi-Status
3570 4 : let local = Part {
3571 4 : status: 204,
3572 4 : detail: "local write applied".into(),
3573 4 : };
3574 4 : let resp = combine(
3575 4 : vec![local, part],
3576 4 : StatusCode::NO_CONTENT.into_response(),
3577 4 : &tenant,
3578 : );
3579 4 : assert_eq!(resp.status(), StatusCode::MULTI_STATUS);
3580 4 : let bytes = axum::body::to_bytes(resp.into_body(), 1 << 20)
3581 4 : .await
3582 4 : .expect("body");
3583 4 : let body = String::from_utf8_lossy(&bytes).into_owned();
3584 4 : assert!(
3585 4 : !body.contains(&format!("127.0.0.1:{port}")),
3586 : "the peer host:port must never reach the client, got {body}"
3587 : );
3588 4 : assert!(
3589 4 : body.contains(&r.reg_id),
3590 4 : "the registration id is the client-safe identifier, got {body}"
3591 4 : );
3592 4 : }
3593 :
3594 : /// 6.3.18: the Via header exists "to avoid infinite loops", and Table
3595 : /// 6.3.18-2 makes its listing part of registration matching — every
3596 : /// element is compared against every candidate registration. A chain
3597 : /// longer than any real cascade is therefore both a loop symptom and a
3598 : /// work amplifier, and is refused before the registrations are read.
3599 : #[tokio::test]
3600 4 : async fn via_chain_beyond_the_hop_ceiling_is_refused() {
3601 8 : let chain = |n: usize| {
3602 8 : (0..n)
3603 260 : .map(|i| format!("1.1 hop{i}"))
3604 8 : .collect::<Vec<_>>()
3605 8 : .join(", ")
3606 8 : };
3607 4 : let t = antares_model::TenantId::new("default").expect("tenant");
3608 4 : let over = hdrs(Some(&chain(MAX_VIA_HOPS + 1)));
3609 4 : assert!(
3610 4 : via_loop(&over, "me"),
3611 : "a chain past the ceiling is treated as a loop"
3612 : );
3613 4 : let mut regs = vec![reg("inclusive")];
3614 4 : let resp = handle_via_loop(&over, "me", &t, &mut regs)
3615 4 : .expect("an over-long Via chain must be refused");
3616 4 : assert_eq!(resp.status(), StatusCode::LOOP_DETECTED);
3617 : // and no registration is consulted: the candidate set is empty
3618 : // without a single registration document being examined
3619 4 : let st = AppState::new("me".into());
3620 4 : let ctx = st.loader.core();
3621 4 : let id = "urn:ngsi-ld:ContextSourceRegistration:hops";
3622 4 : st.store
3623 4 : .create(
3624 4 : &t,
3625 4 : Kind::Registration,
3626 4 : id,
3627 4 : json!({
3628 4 : "id": id,
3629 4 : "type": "ContextSourceRegistration",
3630 4 : "endpoint": "http://peer:9090",
3631 4 : "information": [{"entities": [{"type": "https://uri.etsi.org/ngsi-ld/default-context/Vehicle"}]}],
3632 4 : }),
3633 4 : ).await
3634 4 : .expect("seed registration");
3635 4 : let spec = crate::registry::CsrSpec {
3636 4 : types: Some(vec![
3637 4 : "https://uri.etsi.org/ngsi-ld/default-context/Vehicle".into()
3638 4 : ]),
3639 4 : ..Default::default()
3640 4 : };
3641 4 : assert!(
3642 4 : matching_regs(&st, &t, &spec, &ctx, &over)
3643 4 : .await
3644 4 : .expect("registrations")
3645 4 : .is_empty(),
3646 : "no registration is matched past the hop ceiling"
3647 : );
3648 : // at the ceiling the chain is still processed normally
3649 4 : let at = hdrs(Some(&chain(MAX_VIA_HOPS)));
3650 4 : assert!(!via_loop(&at, "me"));
3651 4 : assert!(handle_via_loop(&at, "me", &t, &mut vec![reg("inclusive")]).is_none());
3652 4 : assert_eq!(
3653 4 : matching_regs(&st, &t, &spec, &ctx, &at)
3654 4 : .await
3655 4 : .expect("registrations")
3656 4 : .len(),
3657 4 : 1
3658 4 : );
3659 4 : }
3660 :
3661 : /// 5.7.2.4/4.23.1: `would_federate` only answers WHETHER a query leaves
3662 : /// the local scope, so it never compiles a forwarding set — but its
3663 : /// verdict must stay identical to the set's emptiness for every shape
3664 : /// that gates matching (type, id, the Via chain, local scope).
3665 : #[tokio::test]
3666 4 : async fn would_federate_agrees_with_the_compiled_forward_set() {
3667 4 : let st = AppState::new("me".into());
3668 4 : let t = antares_model::TenantId::new("default").expect("tenant");
3669 4 : let ctx = st.loader.core();
3670 4 : let id = "urn:ngsi-ld:ContextSourceRegistration:wf";
3671 4 : st.store
3672 4 : .create(
3673 4 : &t,
3674 4 : Kind::Registration,
3675 4 : id,
3676 4 : json!({
3677 4 : "id": id,
3678 4 : "type": "ContextSourceRegistration",
3679 4 : "endpoint": "http://peer:9090",
3680 4 : "contextSourceAlias": "peer1",
3681 4 : "information": [{"entities": [{"type": "https://uri.etsi.org/ngsi-ld/default-context/Vehicle"}]}],
3682 4 : }),
3683 4 : ).await
3684 4 : .expect("seed registration");
3685 20 : let params = |kv: &[(&str, &str)]| {
3686 20 : kv.iter()
3687 24 : .map(|(k, v)| ((*k).to_owned(), (*v).to_owned()))
3688 20 : .collect::<HashMap<String, String>>()
3689 20 : };
3690 20 : for (p, headers, expected) in [
3691 4 : (params(&[("type", "Vehicle")]), hdrs(None), true),
3692 4 : (params(&[("type", "Parking")]), hdrs(None), false),
3693 4 : (
3694 4 : params(&[("type", "Vehicle"), ("local", "true")]),
3695 4 : hdrs(None),
3696 4 : false,
3697 4 : ),
3698 4 : (
3699 4 : params(&[("type", "Vehicle")]),
3700 4 : hdrs(Some("1.1 peer1")),
3701 4 : false,
3702 4 : ),
3703 4 : (
3704 4 : params(&[("type", "Vehicle")]),
3705 4 : hdrs(Some("1.1 other")),
3706 4 : true,
3707 4 : ),
3708 4 : ] {
3709 20 : assert_eq!(
3710 20 : would_federate(&st, &t, &ctx, &p, &headers)
3711 20 : .await
3712 20 : .expect("registrations"),
3713 4 : expected,
3714 4 : "would_federate verdict for {p:?}"
3715 4 : );
3716 20 : assert_eq!(
3717 20 : active(&p)
3718 16 : && !matching_regs(&st, &t, &query_spec(&ctx, &p), &ctx, &headers)
3719 16 : .await
3720 16 : .expect("registrations")
3721 16 : .is_empty(),
3722 4 : expected,
3723 4 : "the compiled forward set must agree for {p:?}"
3724 4 : );
3725 4 : }
3726 4 : }
3727 : }
3728 :
3729 : #[cfg(test)]
3730 : mod clause_4_20 {
3731 : use super::*;
3732 :
3733 48 : fn reg(ops: &[&str]) -> FedReg {
3734 : FedReg {
3735 48 : ops: ops.iter().map(|s| (*s).to_owned()).collect(),
3736 48 : ..FedReg::default()
3737 : }
3738 48 : }
3739 :
3740 : /// Table 4.20-2: associationOps is federationOps WITHOUT the EntityMap
3741 : /// support operations (and without createEntityMapQueryTemporal, which is
3742 : /// in neither group).
3743 : #[test]
3744 4 : fn association_ops_exclude_the_entity_map_operations() {
3745 4 : let r = reg(&["associationOps"]);
3746 16 : for op in [
3747 4 : "retrieveEntity",
3748 4 : "queryEntity",
3749 4 : "deleteSubscription",
3750 4 : "retrieveContextSourceIdentity",
3751 4 : ] {
3752 16 : assert!(r.supports(op), "{op} is in associationOps");
3753 : }
3754 16 : for op in [
3755 4 : "retrieveEntityMap",
3756 4 : "updateEntityMap",
3757 4 : "deleteEntityMap",
3758 4 : "createEntityMapQueryEntity",
3759 4 : ] {
3760 16 : assert!(
3761 16 : !r.supports(op),
3762 : "{op} is NOT in associationOps (Table 4.20-2)"
3763 : );
3764 : }
3765 4 : }
3766 :
3767 : /// Table 4.20-1/2: individual names match themselves; groups match their
3768 : /// members; nothing matches createEntityMapQueryTemporal except itself.
3769 : #[test]
3770 4 : fn groups_and_individual_names() {
3771 4 : assert!(reg(&["federationOps"]).supports("retrieveEntityMap"));
3772 4 : assert!(reg(&["redirectionOps"]).supports("purgeEntity"));
3773 4 : assert!(!reg(&["redirectionOps"]).supports("createSubscription"));
3774 4 : assert!(reg(&["updateOps"]).supports("replaceAttrs"));
3775 4 : assert!(!reg(&["updateOps"]).supports("deleteEntity"));
3776 4 : assert!(reg(&["retrieveOps"]).supports("queryEntity"));
3777 4 : assert!(!reg(&["retrieveOps"]).supports("retrieveTemporal"));
3778 4 : assert!(reg(&["createEntityMapQueryTemporal"]).supports("createEntityMapQueryTemporal"));
3779 12 : for group in ["federationOps", "associationOps", "redirectionOps"] {
3780 12 : assert!(
3781 12 : !reg(&[group]).supports("createEntityMapQueryTemporal"),
3782 : "{group} does not include createEntityMapQueryTemporal"
3783 : );
3784 : }
3785 4 : }
3786 : }
3787 :
3788 : #[cfg(test)]
3789 : mod same_source_merge {
3790 : use super::*;
3791 :
3792 44 : fn reg(id: &str, endpoint: &str, attrs: Option<&[&str]>, types: &[&str]) -> FedReg {
3793 : FedReg {
3794 44 : reg_id: id.into(),
3795 44 : endpoint: endpoint.into(),
3796 44 : mode: "inclusive".into(),
3797 44 : ops: vec!["federationOps".into()],
3798 44 : attrs: attrs.map(|a| a.iter().map(|s| (*s).to_owned()).collect()),
3799 44 : ent_types: types.iter().map(|s| (*s).to_owned()).collect(),
3800 44 : ..FedReg::default()
3801 : }
3802 44 : }
3803 :
3804 : /// Two registrations for one Context Source become one forward whose
3805 : /// scopes are the union; a different endpoint or mode stays separate.
3806 : #[test]
3807 4 : fn registrations_of_one_source_fold_into_one_forward() {
3808 4 : let merged = merge_same_source(vec![
3809 4 : reg("urn:r:1", "http://a", Some(&["speed"]), &["Vehicle"]),
3810 4 : reg(
3811 4 : "urn:r:2",
3812 4 : "http://a",
3813 4 : Some(&["heading", "speed"]),
3814 4 : &["Bike"],
3815 : ),
3816 4 : reg("urn:r:3", "http://b", Some(&["speed"]), &["Vehicle"]),
3817 : ]);
3818 4 : assert_eq!(merged.len(), 2);
3819 4 : assert_eq!(merged[0].reg_id, "urn:r:1");
3820 4 : assert_eq!(
3821 4 : merged[0].attrs.as_deref(),
3822 4 : Some(&["speed".to_owned(), "heading".to_owned()][..])
3823 : );
3824 4 : assert_eq!(merged[0].ent_types, ["Vehicle", "Bike"]);
3825 4 : assert_eq!(merged[1].endpoint, "http://b");
3826 :
3827 : // an unscoped registration widens the merged one to everything
3828 4 : let merged = merge_same_source(vec![
3829 4 : reg("urn:r:1", "http://a", Some(&["speed"]), &["Vehicle"]),
3830 4 : reg("urn:r:2", "http://a", None, &[]),
3831 : ]);
3832 4 : assert_eq!(merged.len(), 1);
3833 4 : assert!(merged[0].attrs.is_none());
3834 :
3835 4 : let mut other_mode = reg("urn:r:2", "http://a", None, &[]);
3836 4 : other_mode.mode = "exclusive".into();
3837 4 : let merged = merge_same_source(vec![reg("urn:r:1", "http://a", None, &[]), other_mode]);
3838 4 : assert_eq!(merged.len(), 2, "a different mode is a different forward");
3839 4 : }
3840 :
3841 : /// 4.3.6.1: a source "may indicate that they are only willing to respond
3842 : /// to a limited subset of API operations. Context Brokers shall respect
3843 : /// this". The fold unions the entity and attribute scope, so folding two
3844 : /// registrations that declare different operations would send an
3845 : /// operation only one of them offered for the OTHER's Entities.
3846 : #[test]
3847 4 : fn registrations_of_one_source_that_offer_different_operations_do_not_fold() {
3848 4 : let mut reads = reg("urn:r:1", "http://a", None, &["Vehicle"]);
3849 4 : reads.ops = vec!["queryEntity".into()];
3850 4 : let mut writes = reg("urn:r:2", "http://a", None, &["Bike"]);
3851 4 : writes.ops = vec!["createEntity".into()];
3852 4 : let merged = merge_same_source(vec![reads, writes]);
3853 4 : assert_eq!(
3854 4 : merged.len(),
3855 : 2,
3856 : "a different operation subset is a different forward"
3857 : );
3858 4 : assert!(
3859 4 : merged[0].supports("queryEntity") && !merged[0].supports("createEntity"),
3860 : "the read registration must not gain the write registration's operation"
3861 : );
3862 4 : assert_eq!(
3863 4 : merged[0].ent_types,
3864 : ["Vehicle"],
3865 : "and must not gain its Entity scope either"
3866 : );
3867 :
3868 : // the same subset written two ways is still one source: 5.2.9 lets
3869 : // `operations` name a group, and the default IS a group
3870 4 : let mut spelled_out = reg("urn:r:2", "http://a", None, &["Bike"]);
3871 4 : spelled_out.ops = antares_model::operations::group_members("federationOps")
3872 4 : .expect("the default group")
3873 4 : .iter()
3874 76 : .map(|s| (*s).to_owned())
3875 4 : .collect();
3876 4 : let merged = merge_same_source(vec![
3877 4 : reg("urn:r:1", "http://a", None, &["Vehicle"]),
3878 4 : spelled_out,
3879 : ]);
3880 4 : assert_eq!(
3881 4 : merged.len(),
3882 : 1,
3883 : "federationOps and the names it stands for are one subset"
3884 : );
3885 4 : assert_eq!(merged[0].ent_types, ["Vehicle", "Bike"]);
3886 4 : }
3887 : }
|
