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 80874 : pub fn supports(&self, op: &str) -> bool {
82 80874 : self.ops
83 80874 : .iter()
84 120406 : .any(|o| o == op || group_members(o).is_some_and(|members| members.contains(&op)))
85 80874 : }
86 1802 : pub fn is_proxy(&self) -> bool {
87 1802 : self.mode == "exclusive" || self.mode == "redirect"
88 1802 : }
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 1578 : pub fn can_match_id(&self, id: &str) -> bool {
94 1578 : self.ent_unrestricted
95 948 : || self.ent_ids.iter().any(|i| i == id)
96 484 : || self
97 484 : .ent_patterns
98 484 : .iter()
99 484 : .any(|p| antares_ql::regex::compile(p).is_ok_and(|re| re.find(id).is_some()))
100 1578 : }
101 : /// Does this registration cover the given expanded attribute IRI?
102 1050 : pub fn covers_attr(&self, iri: &str) -> bool {
103 1050 : self.attrs
104 1050 : .as_ref()
105 1050 : .is_none_or(|a| a.iter().any(|x| x == iri))
106 1050 : }
107 368 : pub fn read_op(&self) -> Option<&'static str> {
108 368 : ["retrieveEntity", "queryEntity", "queryBatch"]
109 368 : .into_iter()
110 416 : .find(|op| self.supports(op))
111 368 : }
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 718 : pub fn query_op(&self) -> Option<&'static str> {
117 718 : ["queryEntity", "queryBatch"]
118 718 : .into_iter()
119 770 : .find(|op| self.supports(op))
120 718 : }
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 826 : pub fn covers_item(&self, obj: &Map<String, Value>, ctx: &Context) -> bool {
127 826 : if !self.ent_ids.is_empty() || !self.ent_patterns.is_empty() {
128 722 : if let Some(id) = obj.get("id").and_then(Value::as_str) {
129 622 : if !self.can_match_id(id) {
130 4 : return false;
131 618 : }
132 100 : }
133 104 : }
134 822 : self.covers_types(obj, ctx)
135 826 : }
136 : /// The type half of `covers_item`: does the item's `type` name one of the
137 : /// registration's EntityInfo types? No type to check ⇒ covered.
138 1386 : pub fn covers_types(&self, obj: &Map<String, Value>, ctx: &Context) -> bool {
139 1386 : if !self.ent_types.is_empty() {
140 1362 : let types: Vec<String> = match obj.get("type") {
141 1268 : Some(Value::String(t)) => vec![ctx.expand_key(t)],
142 4 : Some(Value::Array(a)) => a
143 4 : .iter()
144 4 : .filter_map(Value::as_str)
145 4 : .map(|t| ctx.expand_key(t))
146 4 : .collect(),
147 90 : _ => Vec::new(),
148 : };
149 1362 : if !types.is_empty() && !types.iter().any(|t| self.ent_types.contains(t)) {
150 22 : return false;
151 1340 : }
152 24 : }
153 1364 : true
154 1386 : }
155 : }
156 :
157 : /// Is federation active for this request? (6.3.18 local param; 5.5.13:
158 : /// a local-scope request executes only on information available locally —
159 : /// no Context Source Registrations are considered)
160 : ///
161 : /// Table 6.4.3.2-1: for `type=*`, "local is implicitly set to true and shall
162 : /// not be explicitly set to false" — so the wildcard alone disables forwarding.
163 45809 : pub fn active(params: &HashMap<String, String>) -> bool {
164 45809 : if params.get("type").map(String::as_str) == Some("*") {
165 32 : return false;
166 45777 : }
167 45777 : params.get("local").map(String::as_str) != Some("true")
168 45809 : }
169 :
170 : /// The full inbound Via chain, joined across header FIELDS.
171 : ///
172 : /// RFC 7230 sections 3.2.2 and 5.7.1: Via is a list header — senders may split the
173 : /// chain over any number of `Via:` field lines, and the two forms are
174 : /// equivalent. Reading only the first field (`headers.get`) made a loop
175 : /// pseudonym in a later field invisible (undetected cycle) AND rebuilt the
176 : /// outbound chain from the truncated view, deleting upstream hop history
177 : /// that downstream brokers need for THEIR loop detection.
178 89107 : pub fn inbound_via(headers: &HeaderMap) -> Option<String> {
179 89107 : let fields: Vec<&str> = headers
180 89107 : .get_all("via")
181 89107 : .iter()
182 89107 : .filter_map(|v| v.to_str().ok())
183 89107 : .collect();
184 89107 : if fields.is_empty() {
185 88747 : None
186 : } else {
187 360 : Some(fields.join(", "))
188 : }
189 89107 : }
190 :
191 : /// This broker's Via pseudonym **for one Tenant**.
192 : ///
193 : /// Table 5.2.40-1: the alias is "a unique id for a Context Source which can
194 : /// be used to identify loops. In the multi-tenancy use case (see clause
195 : /// 4.14), this id **shall** be identifying a specific Tenant within a
196 : /// registered Context Source." One static per-process alias therefore makes
197 : /// every tenant of this broker look like the same Context Source: a request
198 : /// in tenant B whose registration points back here for tenant A is a
199 : /// different (source, tenant) pair, but a tenant-blind chain reads it as a
200 : /// loop and drops the registration.
201 : ///
202 : /// Format `{alias}~{tenant}`, and the bare alias for the default tenant —
203 : /// mirroring 6.3.14, where the tenant header is omitted rather than sent as
204 : /// `default`. `~` is an RFC 7230 token character that cannot occur in a
205 : /// `TenantId` (`[A-Za-z0-9_-]{1,64}`), so the two halves never blur; the
206 : /// broker rejects a configured `ANTARES_HOST_ALIAS` containing `~` at
207 : /// startup, so `a~b` in the default tenant cannot collide with `a` in
208 : /// tenant `b`.
209 : ///
210 : /// The value is stable for the life of a deployment because peers **register**
211 : /// it: a Context Source Registration's `contextSourceAlias` (Table 5.2.9-1) is
212 : /// "a previously retrieved unique id" — retrieved from this broker's
213 : /// `/info/sourceIdentity` for that tenant. Changing an alias silently breaks
214 : /// every peer's loop detection, so treat it as a published identifier.
215 44920 : pub fn alias_for(host_alias: &str, tenant: &TenantId) -> String {
216 44920 : if tenant.as_str() == TenantId::DEFAULT {
217 43564 : host_alias.to_owned()
218 : } else {
219 1356 : format!("{host_alias}~{}", tenant.as_str())
220 : }
221 44920 : }
222 :
223 : /// The `received-by` pseudonyms of the inbound Via chain — the Context
224 : /// Sources this request has already passed through (Table 6.3.18-2).
225 : ///
226 : /// RFC 7230: `Via = 1#( received-protocol RWS received-by [ RWS comment ] )`
227 : /// — received-by is the SECOND whitespace token of each element and is a
228 : /// token compared for equality, never by suffix (`ends_with`
229 : /// made alias `b1` match peer `sub-b1`). A malformed element with no
230 : /// protocol falls back to its first token.
231 86503 : pub fn via_tokens(headers: &HeaderMap) -> Vec<String> {
232 86503 : inbound_via(headers)
233 86503 : .map(|v| {
234 308 : v.split(',')
235 956 : .filter_map(|t| {
236 956 : let mut toks = t.split_whitespace();
237 956 : let first = toks.next();
238 956 : toks.next().or(first).map(str::to_owned)
239 956 : })
240 308 : .collect()
241 308 : })
242 86503 : .unwrap_or_default()
243 86503 : }
244 :
245 : /// How many `Via` elements this broker will process on one inbound request.
246 : ///
247 : /// 6.3.18 fixes the header's purpose ("to avoid infinite loops") and Table
248 : /// 6.3.18-2 makes its listing part of registration matching, so every
249 : /// element is compared against every candidate registration: the chain is
250 : /// attacker-supplied input whose length multiplies the work of the request.
251 : /// RFC 7230 section 3.2.5 lets a recipient refuse a field longer than it is
252 : /// willing to process. A cascade deeper than this is a loop the alias
253 : /// pseudonyms failed to name, not a deployment.
254 : const MAX_VIA_HOPS: usize = 32;
255 :
256 : /// The number of `Via` elements received, counted across header fields
257 : /// without building the token list (RFC 7230 sections 3.2.2 and 5.7.1: the
258 : /// elements of a list header may be split over any number of field lines).
259 86431 : fn via_hops(headers: &HeaderMap) -> usize {
260 86431 : headers
261 86431 : .get_all("via")
262 86431 : .iter()
263 86431 : .filter_map(|v| v.to_str().ok())
264 86431 : .map(|v| v.split(',').count())
265 86431 : .sum()
266 86431 : }
267 :
268 : /// Does the inbound Via chain already name this broker, in this tenant?
269 : /// (loop, 6.3.18) — `alias` is always [`alias_for`]'s tenant-qualified value.
270 43328 : pub fn via_loop(headers: &HeaderMap, alias: &str) -> bool {
271 43328 : via_hops(headers) > MAX_VIA_HOPS || via_tokens(headers).iter().any(|t| t == alias)
272 43328 : }
273 :
274 : /// Loop handling for a distributed write (6.3.17, 6.3.18, TS 104 175 9.7).
275 : /// 9.7: "If the requested endpoint is also listed as previously encountered
276 : /// when a Context Source receives a call to that endpoint, this is an
277 : /// indication that a new loop has been detected. A suitable error shall be
278 : /// returned". A write whose `Via`
279 : /// already names this broker in this tenant is answered 508 Loop Detected
280 : /// whatever its registrations, the single exclusive or redirect source of
281 : /// TS 104 176 6.3.5 included, and nothing is forwarded; a read with the same
282 : /// chain is served locally with a 199 warning by its own handler.
283 35916 : pub fn handle_via_loop(
284 35916 : headers: &HeaderMap,
285 35916 : alias: &str,
286 35916 : tenant: &TenantId,
287 35916 : regs: &mut Vec<FedReg>,
288 35916 : ) -> Option<Response> {
289 : // [`via_loop`] also refuses a chain past [`MAX_VIA_HOPS`] outright: the
290 : // operation is not re-forwarded and not run locally either, because a
291 : // request that deep is a cascade the pseudonyms failed to close.
292 35916 : if !via_loop(headers, alias) {
293 35846 : return None;
294 70 : }
295 70 : regs.clear();
296 70 : Some(loop_508(tenant))
297 35916 : }
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 208 : pub fn warning(code: u16, alias: &str, text: &str) -> String {
302 208 : format!("{code} {alias} \"{text}\"")
303 208 : }
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 556 : fn read_warning(status: u16, body: &Value) -> Option<(u16, &'static str)> {
311 398 : match status {
312 6 : 404 => None,
313 138 : 503 | 504 => Some((
314 138 : 199,
315 138 : "no response was received from the registration endpoint within the timeout period",
316 138 : )),
317 412 : s if s >= 400 => Some((
318 14 : 299,
319 14 : "an error response was received from the registration endpoint",
320 14 : )),
321 398 : s if (200..300).contains(&s) && body.is_null() => {
322 26 : Some((111, "the payload of the response was invalid"))
323 : }
324 372 : _ => None,
325 : }
326 556 : }
327 :
328 1396 : fn outbound_via(headers: &HeaderMap, alias: &str) -> String {
329 1396 : match inbound_via(headers) {
330 28 : Some(v) => format!("{v}, 1.1 {alias}"),
331 1368 : None => format!("1.1 {alias}"),
332 : }
333 1396 : }
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 15230 : pub(crate) fn path_segment(s: &str) -> String {
343 63032 : pct_encode(s, |b| {
344 148 : matches!(
345 63032 : 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 63032 : })
364 15230 : }
365 :
366 : /// The @context URL to advertise on forwarded requests.
367 7461 : 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 7461 : if let Ok(Some(url)) = link_context(headers) {
371 2410 : return url;
372 5051 : }
373 176 : match source {
374 4857 : 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 176 : 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 176 : Value::Array(a) => a
382 176 : .iter()
383 176 : .find_map(|e| e.as_str())
384 176 : .unwrap_or(antares_jsonld::CORE_CONTEXT)
385 176 : .to_owned(),
386 : Value::Object(_) => {
387 18 : serde_json::to_string(source).unwrap_or_else(|_| antares_jsonld::CORE_CONTEXT.into())
388 : }
389 0 : _ => antares_jsonld::CORE_CONTEXT.to_owned(),
390 : }
391 7461 : }
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 43085 : async fn reg_docs(
412 43085 : st: &AppState,
413 43085 : tenant: &TenantId,
414 43085 : spec: &crate::registry::CsrSpec,
415 43085 : ctx: &Context,
416 43085 : ) -> Result<Vec<std::sync::Arc<Value>>, NgsiError> {
417 43085 : let types: Option<Vec<String>> = spec
418 43085 : .types
419 43085 : .as_ref()
420 43085 : .filter(|ts| {
421 17027 : !ts.iter()
422 22663 : .any(|t| t.contains([',', ';', '|', '(', ')', '*']))
423 17027 : })
424 43085 : .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 43085 : let ids = spec.ids.as_deref().filter(|_| spec.id_pattern.is_none());
430 43085 : let mut docs: Vec<std::sync::Arc<Value>> = match &st.reg_mirror {
431 38246 : Some(m) => m.matching(tenant.as_str(), ids, types.as_deref()),
432 4839 : None => st
433 4839 : .store
434 4839 : .matching_registrations(tenant, ids, types.as_deref())
435 4839 : .await?
436 4831 : .into_iter()
437 4831 : .map(std::sync::Arc::new)
438 4831 : .collect(),
439 : };
440 : // The mirror answers in hash order, Postgres in its collation's id order
441 : // and the memory store by its own keys: one order for all of them (id,
442 : // bytewise) keeps forwards, 207 reasons and the first failing
443 : // registration the same whichever one answered.
444 43077 : docs.sort_by(|a, b| doc_id(a).cmp(doc_id(b)));
445 43077 : Ok(docs)
446 43085 : }
447 :
448 61684 : fn doc_id(doc: &Value) -> &str {
449 61684 : doc.get("id").and_then(Value::as_str).unwrap_or_default()
450 61684 : }
451 :
452 : /// Does one stored registration take part in this operation, and through
453 : /// which RegistrationInfos (5.12)? Every condition is decided from the
454 : /// borrowed document — expiry, csf, datasetId, location, intervals, the Via
455 : /// chain — so a caller that only needs the verdict ([`would_federate`])
456 : /// stops here instead of compiling a `FedReg` per registration. Expiry is
457 : /// filtered HERE and only here: the single yield point.
458 32594 : fn reg_candidate<'a>(
459 32594 : doc: &'a Value,
460 32594 : spec: &crate::registry::CsrSpec,
461 32594 : ctx: &Context,
462 32594 : seen: &[String],
463 32594 : ) -> Option<Vec<&'a Value>> {
464 32594 : if crate::registry::reg_expired(doc) {
465 0 : return None;
466 32594 : }
467 : // 5.7.2.4/5.7.4.4/5.6.21.4: a csf gates which Context Sources
468 : // are considered (evaluated over the registration's own
469 : // Context Source Properties, 5.10.2.4 semantics).
470 32594 : if let Some(csf) = &spec.csf {
471 14 : if !crate::registry::csf_matches(csf, doc, ctx) {
472 8 : return None;
473 6 : }
474 32580 : }
475 : // 5.12 datasetId condition (should-level): both sides specifying
476 : // datasetId match only with a value in common; one side alone always
477 : // matches.
478 32586 : if let Some(ds) = &spec.dataset_ids {
479 4 : if let Some(reg_ds) = doc.get("datasetId").and_then(Value::as_array) {
480 4 : if !reg_ds
481 4 : .iter()
482 4 : .filter_map(Value::as_str)
483 4 : .any(|d| ds.iter().any(|q| q == d))
484 : {
485 2 : return None;
486 2 : }
487 0 : }
488 32582 : }
489 : // 5.2.9 location + 4.3.6.1: a geo-scoped registration is only consulted
490 : // when the query's geo filter matches its geometry; a registration
491 : // without `location` is unconstrained.
492 32584 : if let Some(gq) = &spec.geo {
493 4 : if let Some(geom) = doc.get("location") {
494 0 : if !gq.matches_geometry(geom) {
495 0 : return None;
496 0 : }
497 4 : }
498 32580 : }
499 : // 5.2.9: a declared observation/management interval gates temporal
500 : // fan-out on overlap with the temporal query; without any declared
501 : // interval the registration is unconstrained.
502 32584 : if let Some(tq) = &spec.temporal {
503 30 : if (doc.get("observationInterval").is_some() || doc.get("managementInterval").is_some())
504 0 : && !crate::registry::temporal_interval_matches(doc, tq)
505 : {
506 0 : return None;
507 30 : }
508 32554 : }
509 : // Table 6.3.18-2 / 5.2.9: this source already handled the request.
510 32584 : if doc
511 32584 : .get("contextSourceAlias")
512 32584 : .and_then(Value::as_str)
513 32584 : .is_some_and(|a| seen.iter().any(|t| t == a))
514 : {
515 14 : return None;
516 32570 : }
517 32570 : let infos = crate::registry::matching_infos(spec, doc, ctx);
518 32570 : if infos.is_empty() {
519 30770 : None
520 : } else {
521 1800 : Some(infos)
522 : }
523 32594 : }
524 :
525 : /// Registrations matching an entity spec (5.12), compiled for forwarding.
526 : ///
527 : /// Table 6.3.18-2 makes the inbound `Via` listing part of matching itself —
528 : /// "the listing of previously encountered Context Sources supplied is used
529 : /// when determining matching registrations" — so a registration whose
530 : /// `contextSourceAlias` is already in the chain is filtered out HERE, at the
531 : /// one place every read and write path resolves its candidates. Keeping it
532 : /// out of the call sites is deliberate: a loop check the callers own is a
533 : /// loop check some caller forgets.
534 42217 : pub async fn matching_regs(
535 42217 : st: &AppState,
536 42217 : tenant: &TenantId,
537 42217 : spec: &crate::registry::CsrSpec,
538 42217 : ctx: &Context,
539 42217 : headers: &HeaderMap,
540 42217 : ) -> Result<Vec<FedReg>, NgsiError> {
541 42217 : if via_hops(headers) > MAX_VIA_HOPS {
542 18 : return Ok(Vec::new());
543 42199 : }
544 42199 : let seen = via_tokens(headers);
545 42199 : let regs: Vec<FedReg> = reg_docs(st, tenant, spec, ctx)
546 42199 : .await?
547 42191 : .iter()
548 42538 : .filter_map(|doc| {
549 32550 : let infos = reg_candidate(doc, spec, ctx, &seen)?;
550 1764 : let alias = doc
551 1764 : .get("contextSourceAlias")
552 1764 : .and_then(Value::as_str)
553 1764 : .map(str::to_owned);
554 1764 : let endpoint = endpoint_base(doc.get("endpoint").and_then(Value::as_str)?);
555 1764 : let mode = doc
556 1764 : .get("mode")
557 1764 : .and_then(Value::as_str)
558 1764 : .unwrap_or("inclusive")
559 1764 : .to_owned();
560 1764 : let ops = doc
561 1764 : .get("operations")
562 1764 : .and_then(Value::as_array)
563 1764 : .map(|a| {
564 1522 : a.iter()
565 1522 : .filter_map(Value::as_str)
566 1522 : .map(str::to_owned)
567 1522 : .collect()
568 1522 : })
569 1764 : .unwrap_or_else(|| vec![DEFAULT_OPERATION_GROUP.into()]);
570 1764 : let mut attrs: Option<Vec<String>> = Some(Vec::new());
571 1764 : let mut ent_ids = Vec::new();
572 1764 : let mut ent_types = Vec::new();
573 1764 : let mut ent_patterns = Vec::new();
574 1764 : let mut ent_unrestricted = false;
575 : // an info that names no type admits every type; ent_types empty
576 : // is how every reader of it spells "no type restriction"
577 1764 : let mut any_type = false;
578 1766 : for info in &infos {
579 : // TS 104 175 9.4.2: attributeNames, else the deprecated pair
580 1766 : match antares_model::registration_attribute_names(info) {
581 1504 : None => attrs = None, // an unscoped info covers everything
582 262 : Some(names) => {
583 262 : if let Some(list) = &mut attrs {
584 260 : list.extend(names.into_iter().map(str::to_owned));
585 260 : }
586 : }
587 : }
588 1766 : if let Some(es) = info.get("entities").and_then(Value::as_array) {
589 1812 : for e in es {
590 1812 : if let Some(i) = e.get("id").and_then(Value::as_str) {
591 858 : ent_ids.push(i.to_owned());
592 962 : }
593 1812 : if let Some(p) = e.get("idPattern").and_then(Value::as_str) {
594 310 : ent_patterns.push(p.to_owned());
595 1502 : }
596 : // 5.12 condition 1: neither id nor idPattern ⇒ the
597 : // element restricts by type only, never by id
598 1812 : if e.get("id").is_none() && e.get("idPattern").is_none() {
599 644 : ent_unrestricted = true;
600 1168 : }
601 : // 5.2.8: type may be a String or String[]
602 1812 : match e.get("type") {
603 1800 : Some(Value::String(t)) => ent_types.push(t.clone()),
604 0 : Some(Value::Array(ts)) => ent_types
605 0 : .extend(ts.iter().filter_map(Value::as_str).map(str::to_owned)),
606 12 : _ => any_type = true,
607 : }
608 : }
609 18 : } else {
610 18 : // an attributes-only RegistrationInfo imposes no id scope
611 18 : ent_unrestricted = true;
612 18 : any_type = true;
613 18 : }
614 : }
615 1764 : if any_type {
616 30 : ent_types.clear();
617 1734 : }
618 1764 : let tenant = doc.get("tenant").and_then(Value::as_str).map(str::to_owned);
619 1764 : let csi = csi_of(doc);
620 : Some(FedReg {
621 1764 : reg_id: doc
622 1764 : .get("id")
623 1764 : .and_then(Value::as_str)
624 1764 : .unwrap_or_default()
625 1764 : .to_owned(),
626 1764 : endpoint,
627 1764 : mode,
628 1764 : ops,
629 1764 : attrs,
630 1764 : ent_ids,
631 1764 : ent_types,
632 1764 : ent_patterns,
633 1764 : ent_unrestricted,
634 1764 : tenant,
635 1764 : alias,
636 1764 : csi,
637 1764 : local_only: local_only_of(doc),
638 1764 : timeout_ms: doc
639 1764 : .get("management")
640 1764 : .and_then(|m| m.get("timeout"))
641 1764 : .and_then(Value::as_u64),
642 1764 : cooldown_ms: doc
643 1764 : .get("management")
644 1764 : .and_then(|m| m.get("cooldown"))
645 1764 : .and_then(Value::as_u64),
646 : })
647 32550 : })
648 42191 : .collect();
649 42191 : Ok(merge_same_source(regs))
650 42217 : }
651 :
652 : /// The API operations a registration is willing to answer, EXPANDED: 5.2.9
653 : /// lets `operations` name groups as well as operations, so `["federationOps"]`
654 : /// and the names it stands for are the same subset written two ways, and only
655 : /// the expansion can tell two registrations apart by what they will serve.
656 1824 : fn op_set(r: &FedReg) -> std::collections::BTreeSet<&'static str> {
657 1824 : antares_model::operations::OPERATION_NAMES
658 1824 : .iter()
659 1824 : .copied()
660 78432 : .filter(|op| r.supports(op))
661 1824 : .collect()
662 1824 : }
663 :
664 14 : fn same_entity_scope(a: &FedReg, b: &FedReg) -> bool {
665 14 : a.ent_ids == b.ent_ids
666 14 : && a.ent_types == b.ent_types
667 4 : && a.ent_patterns == b.ent_patterns
668 4 : && a.ent_unrestricted == b.ent_unrestricted
669 14 : }
670 :
671 20 : fn covers_everything(r: &FedReg) -> bool {
672 20 : r.attrs.is_none()
673 10 : && r.ent_types.is_empty()
674 4 : && r.ent_ids.is_empty()
675 4 : && r.ent_patterns.is_empty()
676 4 : && r.ent_unrestricted
677 20 : }
678 :
679 : /// Registrations naming the same Context Source (same endpoint, mode,
680 : /// tenant, contextSourceAlias, contextSourceInfo, localOnly) fold into ONE
681 : /// forwarded request (5.2.9: the alias identifies a source, so a different
682 : /// alias is a different source even behind one endpoint):
683 : /// attribute and entity scopes union (an unscoped one covers everything),
684 : /// the first registration's id and timing stay. Two calls to one source for
685 : /// one query would return the same data twice.
686 : ///
687 : /// Registrations that declare DIFFERENT operations are not the same source
688 : /// for this purpose. 4.3.6.1: a source "may indicate that they are only
689 : /// willing to respond to a limited subset of API operations. Context Brokers
690 : /// shall respect this" — and the fold unions the entity and attribute scope,
691 : /// so an operation only one of them declared would then travel for the
692 : /// other's Entities. They stay separate instead, each filtered by `supports`
693 : /// on its own. The cost is one extra request to a source that registered
694 : /// itself twice with different operation lists and both lists cover the
695 : /// operation at hand; 4.5.5 merges the two answers by Entity id, so the
696 : /// caller sees the same data either way.
697 : ///
698 : /// For the same reason they fold only when the union is exact: (Vehicle ×
699 : /// all) and (OffStreetParking × two attributes) would fold to
700 : /// (both types × all), and the source's whole OffStreetParking would be
701 : /// imported (ETSI IOP_003_02_02). Equal attribute scopes, equal entity
702 : /// scopes, or one side covering everything keep the union exact.
703 42219 : fn merge_same_source(regs: Vec<FedReg>) -> Vec<FedReg> {
704 42219 : let mut out: Vec<(FedReg, std::collections::BTreeSet<&'static str>)> = Vec::new();
705 42219 : for r in regs {
706 1824 : let ops = op_set(&r);
707 1824 : let same = out.iter_mut().find(|(o, o_ops)| {
708 792 : o.endpoint == r.endpoint
709 66 : && o.mode == r.mode
710 58 : && o.tenant == r.tenant
711 58 : && o.alias == r.alias
712 42 : && o.csi == r.csi
713 42 : && o.local_only == r.local_only
714 38 : && *o_ops == ops
715 34 : && (o.attrs == r.attrs
716 14 : || same_entity_scope(o, &r)
717 10 : || covers_everything(o)
718 10 : || covers_everything(&r))
719 792 : });
720 1824 : let Some((o, _)) = same else {
721 1796 : out.push((r, ops));
722 1796 : continue;
723 : };
724 28 : o.attrs = match (o.attrs.take(), r.attrs) {
725 8 : (Some(mut a), Some(b)) => {
726 12 : for x in b {
727 12 : if !a.contains(&x) {
728 4 : a.push(x);
729 8 : }
730 : }
731 8 : Some(a)
732 : }
733 20 : _ => None,
734 : };
735 28 : o.ent_ids.extend(r.ent_ids);
736 28 : if o.ent_types.is_empty() || r.ent_types.is_empty() {
737 4 : o.ent_types.clear(); // either side admits every type
738 24 : } else {
739 24 : o.ent_types.extend(r.ent_types);
740 24 : }
741 28 : o.ent_patterns.extend(r.ent_patterns);
742 28 : o.ent_unrestricted |= r.ent_unrestricted;
743 : }
744 42219 : out.into_iter().map(|(r, _)| r).collect()
745 42219 : }
746 :
747 : /// contextSourceInfo keys the forward must NOT copy into headers: the tenant
748 : /// travels via the registration's own `tenant` member (4.3.6.5 "shall not be
749 : /// part of contextSourceInfo"), connection/binding-managed headers cannot be
750 : /// overridden ("shall be ignored"), and the 4.3.6.6 processed keys (accept,
751 : /// contentType, jsonldContext, ngsildConformance) are TRANSFORMED by
752 : /// `forward` rather than passed through raw, which would corrupt
753 : /// negotiation instead.
754 : /// Headers that carry the triggering caller's own credential: never copied
755 : /// from the request by a `urn:ngsi-ld:request` contextSourceInfo entry.
756 : const REQUEST_CREDENTIALS: &[&str] = &["authorization", "proxy-authorization", "cookie"];
757 :
758 : const CSI_SKIP: &[&str] = &[
759 : "ngsild-tenant",
760 : "content-length",
761 : "content-type",
762 : "host",
763 : "via",
764 : "link",
765 : "connection",
766 : "accept",
767 : "contenttype",
768 : "jsonldcontext",
769 : "ngsildconformance",
770 : ];
771 :
772 : /// Bounds wall on the forwarded-read path: a peer response larger than
773 : /// ANTARES_MAX_FED_RESPONSE_BYTES is never held in memory — the part fails
774 : /// exactly like an unparseable payload (Table 6.3.17-1, warning 111 via the
775 : /// 2xx-with-null-body arm of `read_warning`).
776 1130 : async fn read_body_capped(resp: reqwest::Response) -> Value {
777 1130 : let cap = *crate::bounds::MAX_FED_RESPONSE_BYTES;
778 1130 : if resp.content_length().is_some_and(|l| l > cap as u64) {
779 6 : tracing::warn!("federation response over the {cap}-byte cap (declared length), skipped");
780 6 : return Value::Null;
781 1124 : }
782 : #[cfg(not(target_arch = "wasm32"))]
783 1122 : let bytes = {
784 1124 : let mut resp = resp;
785 1124 : let mut buf: Vec<u8> = Vec::new();
786 : loop {
787 2016 : match resp.chunk().await {
788 894 : Ok(Some(c)) => {
789 894 : if buf.len().saturating_add(c.len()) > cap {
790 2 : tracing::warn!("federation response over the {cap}-byte cap, skipped");
791 2 : return Value::Null;
792 892 : }
793 892 : buf.extend_from_slice(&c);
794 : }
795 1122 : Ok(None) => break,
796 0 : Err(_) => return Value::Null,
797 : }
798 : }
799 1122 : buf
800 : };
801 : // the browser fetch API hands the body over whole — cap after the read
802 : #[cfg(target_arch = "wasm32")]
803 : let bytes = match resp.bytes().await {
804 : Ok(b) if b.len() <= cap => b.to_vec(),
805 : _ => return Value::Null,
806 : };
807 1122 : serde_json::from_slice(&bytes).unwrap_or(Value::Null)
808 1130 : }
809 :
810 : /// 5.7.2.4: "If split entities flag is explicitly set to true or, if not
811 : /// explicitly set, the default setting of the deployment allows split
812 : /// entities" — this deployment's default is OFF, so only the explicit flag
813 : /// engages the split branch.
814 3353 : pub(crate) fn split_entities(params: &HashMap<String, String>) -> bool {
815 3353 : params.get("splitEntities").map(String::as_str) == Some("true")
816 3353 : }
817 :
818 : /// 4.3.6.6: a registration carrying a jsonldContext contextSourceInfo key —
819 : /// forwards to it are recompacted term-by-term (attrs/type/geoproperty only).
820 358 : fn has_reg_context(reg: &FedReg) -> bool {
821 358 : reg.csi
822 358 : .iter()
823 358 : .any(|(k, _)| k.eq_ignore_ascii_case("jsonldContext"))
824 358 : }
825 :
826 : /// 4.3.6.1 fan-out: forwards to matching registrations run concurrently —
827 : /// the clause fixes the merge order (4.5.5 non-aux before aux), never the
828 : /// request order, and cross-source result ordering does not exist (the
829 : /// `ordering` parameter is a 400 outside local scope, 5.7.2.4). Results
830 : /// return in registration order so warning and merge processing stay
831 : /// deterministic. Concurrency per request is bounded by
832 : /// bounds::MAX_FED_FANOUT.
833 6463 : async fn fan_out<I, T, F, Fut>(items: Vec<I>, per_item: F) -> Vec<T>
834 6463 : where
835 6463 : F: FnMut(I) -> Fut,
836 6463 : Fut: std::future::Future<Output = T>,
837 6463 : {
838 : use futures_util::StreamExt;
839 6463 : futures_util::stream::iter(items.into_iter().map(per_item))
840 6463 : .buffered(*crate::bounds::MAX_FED_FANOUT)
841 6463 : .collect()
842 6463 : .await
843 6463 : }
844 :
845 : /// 6.6 Entity Attribute and 6.7 Entity Attribute Instance name one Attribute
846 : /// in the PATH rather than in the payload. The range is the encoded segment
847 : /// inside `url`, so a translated name can be written back over it; the name
848 : /// comes back percent-decoded, which is the form an `@context` expands. The
849 : /// segment stops at the next `/`, so `/attrs/{name}/value` and
850 : /// `/attrs/{name}/{instanceId}` both name `{name}` and nothing more.
851 64 : fn path_attr_segment(url: &str) -> Option<(std::ops::Range<usize>, String)> {
852 64 : let path = url.split_once('?').map_or(url, |(path, _)| path);
853 64 : let (before, rest) = path.split_once("/attrs/")?;
854 42 : let name = rest.split_once('/').map_or(rest, |(name, _)| name);
855 : // `before` and "/attrs/" are a prefix of `url`, so the range lies inside it
856 42 : let at = before.len().checked_add("/attrs/".len())?;
857 42 : let end = at.checked_add(name.len())?;
858 42 : (!name.is_empty()).then(|| (at..end, crate::negotiate::percent_decode(name.as_bytes())))
859 64 : }
860 :
861 : /// Table 5.2.9-2 and 5.2.34: one forward's outcome, booked on the
862 : /// registration it was made for. `ok` is the table's own failure
863 : /// definition — "an HTTP response code other than 2xx".
864 : ///
865 : /// Only an operation that reached the wire is booked. The three ways out of
866 : /// `forward` above this point — a destination the egress policy refuses, an
867 : /// open breaker, the declared 5.2.34 cooldown — are this broker declining to
868 : /// perform the operation, and the table counts operations that were
869 : /// performed: `lastFailure` is when a failure "was returned". Nothing is
870 : /// lost by that: the breaker opens only after failures that WERE attempted
871 : /// and booked, so a registration whose Context Source is down already reads
872 : /// `"failed"` before the first forward is suppressed.
873 : ///
874 : /// The stamp never fails the forward. The answer is decided by the time this
875 : /// runs, and a store that cannot take a counter must not turn a served
876 : /// response into an error. The registration mirror is not refreshed either:
877 : /// it exists to decide which registrations match, and no member here does.
878 1386 : async fn note_forward(st: &AppState, tenant: &TenantId, reg: &FedReg, ok: bool) {
879 1386 : if reg.cooldown_ms.is_some() {
880 2 : st.reg_cooldown_stamp(tenant, ®.reg_id, ok);
881 1384 : }
882 1386 : if let Err(e) = st
883 1386 : .store
884 1386 : .record_forward(tenant, ®.reg_id, &crate::state::now_iso(), ok)
885 1386 : .await
886 : {
887 0 : tracing::warn!(
888 : "forward bookkeeping for registration {} failed: {e}",
889 : reg.reg_id
890 : );
891 1386 : }
892 1386 : }
893 :
894 : /// One forward's answer: the status, the body (`Null` when there is none
895 : /// or it does not parse), and the `NGSILD-Warning` values the source sent.
896 : pub type ForwardAnswer = (u16, Value, Vec<String>);
897 :
898 : /// The answer a [`ForwardTransport`] is working on.
899 : pub type ForwardFuture<'a> =
900 : std::pin::Pin<Box<dyn std::future::Future<Output = ForwardAnswer> + Send + 'a>>;
901 :
902 : /// A transport that serves some forwards without a socket (the forward
903 : /// transport seam). Every distributed operation reaches
904 : /// it through `federation::forward`, after the request is built and before the
905 : /// egress check; vanilla registers none
906 : /// ([`AppState::with_forward_transport`]).
907 : pub trait ForwardTransport: Send + Sync {
908 : /// `Some(answer)` when this transport serves `req`; `None` sends it over
909 : /// HTTP as before. The 5.2.34 cooldown and timeout, the bookkeeping of
910 : /// Table 5.2.9-2 and the cap on the source's warnings apply to the
911 : /// answer as they do over HTTP; the egress guard and the breaker do
912 : /// not, since no socket is opened.
913 : fn forward<'a>(&'a self, req: BuiltForward<'a>) -> Option<ForwardFuture<'a>>;
914 :
915 : /// Whether Tenant `source` may store a registration whose forwards go
916 : /// to `endpoint` in Tenant `target` (`None`: the default Tenant), asked
917 : /// at 12.2.2 and 12.2.4 before anything is stored.
918 : /// `Ok` for an endpoint this transport does not serve: its forwards go
919 : /// over HTTP, as before. `Err` carries the reason of the 403.
920 26 : fn admit<'a>(
921 26 : &'a self,
922 26 : _source: &'a TenantId,
923 26 : _endpoint: &'a str,
924 26 : _target: Option<&'a str>,
925 26 : ) -> AdmitFuture<'a> {
926 26 : Box::pin(std::future::ready(Ok(())))
927 26 : }
928 :
929 : /// Whether this transport delivers the notifications sent to `uri`.
930 : /// Such a delivery opens no socket, so the egress
931 : /// guard and the breaker do not judge it; `false` leaves it to the
932 : /// binding of the URI's scheme, as before.
933 8 : fn notifies(&self, _uri: &str) -> bool {
934 8 : false
935 8 : }
936 :
937 : /// One attempt to deliver `notification`, which a Subscription of
938 : /// Tenant `tenant` sends to `uri`, an endpoint [`Self::notifies`] took.
939 : /// `Err` carries why it failed: a failed attempt under the 5.8.6 retry
940 : /// and dead-letter policy, as an HTTP one is. No subject is carried:
941 : /// delivery is broker-initiated (ADR-0020).
942 2 : fn notify<'a>(
943 2 : &'a self,
944 2 : _tenant: &'a TenantId,
945 2 : _uri: &'a str,
946 2 : _notification: &'a Value,
947 2 : ) -> NotifyFuture<'a> {
948 2 : Box::pin(std::future::ready(Err(
949 2 : "this transport delivers no notification".into(),
950 2 : )))
951 2 : }
952 : }
953 :
954 : /// What [`ForwardTransport::notify`] answers.
955 : pub type NotifyFuture<'a> =
956 : std::pin::Pin<Box<dyn std::future::Future<Output = Result<(), String>> + Send + 'a>>;
957 :
958 : /// What [`ForwardTransport::admit`] answers.
959 : pub type AdmitFuture<'a> =
960 : std::pin::Pin<Box<dyn std::future::Future<Output = Result<(), String>> + Send + 'a>>;
961 :
962 : /// The registered transport's [`ForwardTransport::admit`] for the
963 : /// registration document `doc` of `tenant`; `Ok` without a transport.
964 4300 : pub(crate) async fn admit_registration(
965 4300 : st: &AppState,
966 4300 : tenant: &TenantId,
967 4300 : doc: &serde_json::Map<String, Value>,
968 4300 : ) -> Result<(), crate::policy::Denied> {
969 26 : let (Some(transport), Some(endpoint)) = (
970 4300 : st.forward_transport.get(),
971 4300 : doc.get("endpoint").and_then(Value::as_str),
972 : ) else {
973 4274 : return Ok(());
974 : };
975 26 : let target = doc.get("tenant").and_then(Value::as_str);
976 26 : transport
977 26 : .admit(tenant, endpoint, target)
978 26 : .await
979 26 : .map_err(crate::policy::Denied)
980 4300 : }
981 :
982 : /// The request `federation::forward` built, as the HTTP client would send it: a
983 : /// transport reads it and cannot change what is sent, only how.
984 : #[derive(Clone)]
985 : pub struct BuiltForward<'a> {
986 : method: reqwest::Method,
987 : url: String,
988 : query: Vec<(String, String)>,
989 : headers: Vec<(String, String)>,
990 : body: Option<Vec<u8>>,
991 : tenant: &'a TenantId,
992 : reg: &'a FedReg,
993 : subject: HeaderMap,
994 : /// The registered @context the forward was switched to (4.3.6.6).
995 : answered_in: Option<std::sync::Arc<Context>>,
996 : }
997 :
998 : impl BuiltForward<'_> {
999 : /// The method.
1000 : #[must_use]
1001 20 : pub fn method(&self) -> &reqwest::Method {
1002 20 : &self.method
1003 20 : }
1004 :
1005 : /// The URL on the registration's endpoint, without the query.
1006 : #[must_use]
1007 0 : pub fn url(&self) -> &str {
1008 0 : &self.url
1009 0 : }
1010 :
1011 : /// The URL with the query encoded as the HTTP client encodes it, or
1012 : /// `None` when the URL does not parse (the HTTP client refuses it too).
1013 : #[must_use]
1014 20 : pub fn target(&self) -> Option<String> {
1015 20 : let mut url = reqwest::Url::parse(&self.url).ok()?;
1016 20 : if !self.query.is_empty() {
1017 10 : url.query_pairs_mut().extend_pairs(&self.query);
1018 10 : }
1019 20 : Some(url.into())
1020 20 : }
1021 :
1022 : /// The query parameters, in order.
1023 : #[must_use]
1024 0 : pub fn query(&self) -> &[(String, String)] {
1025 0 : &self.query
1026 0 : }
1027 :
1028 : /// The headers, in the order they are sent: `Accept`, `Via`, `Link`,
1029 : /// `NGSILD-Tenant`, the 4.3.6.5 `contextSourceInfo` ones, `Content-Type`.
1030 : #[must_use]
1031 20 : pub fn headers(&self) -> &[(String, String)] {
1032 20 : &self.headers
1033 20 : }
1034 :
1035 : /// The payload, serialized.
1036 : #[must_use]
1037 22 : pub fn body(&self) -> Option<&[u8]> {
1038 22 : self.body.as_deref()
1039 22 : }
1040 :
1041 : /// The forwarding Tenant: the one the triggering request ran in.
1042 : #[must_use]
1043 0 : pub fn tenant(&self) -> &TenantId {
1044 0 : self.tenant
1045 0 : }
1046 :
1047 : /// The registration this forward is made for.
1048 : #[must_use]
1049 0 : pub fn registration_id(&self) -> &str {
1050 0 : &self.reg.reg_id
1051 0 : }
1052 :
1053 : /// Whether the registration is the `exclusive` or `redirect` kind
1054 : /// (9.3.3), whose single failure is the operation's answer (6.3.17).
1055 : #[must_use]
1056 0 : pub fn proxied(&self) -> bool {
1057 0 : self.reg.is_proxy()
1058 0 : }
1059 :
1060 : /// The registration's 5.2.9 `tenant`: the Tenant the forward targets,
1061 : /// `None` for the default Tenant (4.14).
1062 : #[must_use]
1063 0 : pub fn registration_tenant(&self) -> Option<&str> {
1064 0 : self.reg.tenant.as_deref()
1065 0 : }
1066 :
1067 : /// The policy subject headers of the triggering request (ADR-0020):
1068 : /// never on the wire, for a transport that asks the target's engine
1069 : /// about the caller in process.
1070 : #[must_use]
1071 0 : pub fn subject(&self) -> &HeaderMap {
1072 0 : &self.subject
1073 0 : }
1074 : }
1075 :
1076 : /// 5.2.34 cooldown (per REGISTRATION, distinct from the host:port breaker):
1077 : /// inside the declared window "a timeout error response for the
1078 : /// registration is automatically returned" — the source is not contacted.
1079 1396 : fn in_cooldown(st: &AppState, tenant: &TenantId, reg: &FedReg) -> bool {
1080 1396 : reg.cooldown_ms.is_some_and(|cd| {
1081 4 : st.egress
1082 4 : .reg_in_cooldown(&crate::egress::reg_key(tenant.as_str(), ®.reg_id), cd)
1083 4 : })
1084 1396 : }
1085 :
1086 : /// 5.2.34 timeout: the per-registration budget, below the 8 s ceiling.
1087 1394 : fn deadline_ms(reg: &FedReg) -> u32 {
1088 1394 : let ceiling = crate::state::slow_factor().saturating_mul(8_000);
1089 1394 : reg.timeout_ms.map_or(ceiling, |t| t.min(ceiling)) as u32
1090 1394 : }
1091 :
1092 : /// `fut`, or `None` once `ms` have passed.
1093 : #[cfg(not(target_arch = "wasm32"))]
1094 6 : async fn within<T>(ms: u32, fut: impl std::future::Future<Output = T>) -> Option<T> {
1095 6 : tokio::time::timeout(std::time::Duration::from_millis(u64::from(ms)), fut)
1096 6 : .await
1097 6 : .ok()
1098 6 : }
1099 :
1100 : /// The browser timer is not Send, so the wait is one Send unit, as an
1101 : /// HTTP interaction is.
1102 : #[cfg(target_arch = "wasm32")]
1103 : async fn within<T>(ms: u32, fut: impl std::future::Future<Output = T>) -> Option<T> {
1104 : antares_jsonld::http_interaction(antares_jsonld::io_deadline(fut, ms)).await
1105 : }
1106 :
1107 : /// One forwarded request. `body` is compacted JSON (no @context member).
1108 : /// The registered [`ForwardTransport`] is asked first; without one, or
1109 : /// when it does not serve this forward, it goes over HTTP.
1110 : #[allow(clippy::too_many_arguments)] // mirrors the wire: one param per forwarded request part
1111 838 : pub async fn forward(
1112 838 : st: &AppState,
1113 838 : method: reqwest::Method,
1114 838 : url: String,
1115 838 : query: &[(String, String)],
1116 838 : headers: &HeaderMap,
1117 838 : tenant: &TenantId,
1118 838 : reg: &FedReg,
1119 838 : ctx_url: &str,
1120 838 : body: Option<Value>,
1121 838 : ) -> ForwardAnswer {
1122 838 : forward_in(st, method, url, query, headers, tenant, reg, ctx_url, body)
1123 838 : .await
1124 : .0
1125 830 : }
1126 :
1127 : /// [`forward`], and the @context its answer is in when that is not the
1128 : /// request's own: a forward 4.3.6.6 switched to the registered
1129 : /// `jsonldContext` is answered in that context's terms, so a read expands
1130 : /// the answer with it (`None`: with the request's).
1131 : #[allow(clippy::too_many_arguments)] // mirrors the wire: one param per forwarded request part
1132 1396 : pub(crate) async fn forward_in(
1133 1396 : st: &AppState,
1134 1396 : method: reqwest::Method,
1135 1396 : url: String,
1136 1396 : query: &[(String, String)],
1137 1396 : headers: &HeaderMap,
1138 1396 : tenant: &TenantId,
1139 1396 : reg: &FedReg,
1140 1396 : ctx_url: &str,
1141 1396 : body: Option<Value>,
1142 1396 : ) -> (ForwardAnswer, Option<std::sync::Arc<Context>>) {
1143 : // Process-wide slot first (bounds::MAX_FED_INFLIGHT): the buffers and
1144 : // connections below exist only while a slot is held.
1145 1396 : let _slot = crate::bounds::FED_INFLIGHT.acquire().await;
1146 1396 : let built = build_forward(st, method, url, query, headers, tenant, reg, ctx_url, body).await;
1147 1396 : let answered_in = built.answered_in.clone();
1148 1396 : if let Some(answer) = ask_transport(st, &built).await {
1149 8 : return (answer, answered_in);
1150 1388 : }
1151 1388 : (send_http(st, built).await, answered_in)
1152 1388 : }
1153 :
1154 : /// The forward through the registered transport, or `None` when there is
1155 : /// none or it does not serve this one.
1156 1396 : async fn ask_transport(st: &AppState, built: &BuiltForward<'_>) -> Option<ForwardAnswer> {
1157 1396 : let answer = st.forward_transport.get()?.forward(built.clone())?;
1158 8 : let (tenant, reg) = (built.tenant, built.reg);
1159 8 : if in_cooldown(st, tenant, reg) {
1160 2 : return Some((504, Value::Null, Vec::new()));
1161 6 : }
1162 6 : let Some((status, body, mut warnings)) = within(deadline_ms(reg), answer).await else {
1163 2 : note_forward(st, tenant, reg, false).await;
1164 2 : return Some((504, Value::Null, Vec::new()));
1165 : };
1166 4 : warnings.truncate(crate::bounds::MAX_PEER_WARNINGS);
1167 4 : note_forward(st, tenant, reg, (200..300).contains(&status)).await;
1168 4 : Some((status, body, warnings))
1169 1396 : }
1170 :
1171 : /// The request one forward sends: the 4.3.6.6 processing, the headers and
1172 : /// the payload, exactly as the HTTP client puts them on the wire.
1173 : #[allow(clippy::too_many_arguments)] // mirrors the wire: one param per forwarded request part
1174 1396 : async fn build_forward<'a>(
1175 1396 : st: &AppState,
1176 1396 : method: reqwest::Method,
1177 1396 : mut url: String,
1178 1396 : query: &[(String, String)],
1179 1396 : headers: &HeaderMap,
1180 1396 : tenant: &'a TenantId,
1181 1396 : reg: &'a FedReg,
1182 1396 : ctx_url: &str,
1183 1396 : mut body: Option<Value>,
1184 1396 : ) -> BuiltForward<'a> {
1185 : // 4.3.6.6: the four contextSourceInfo keys with processing
1186 : // semantics. Values were validated at registration time (5.9.2).
1187 4800 : let csi_get = |key: &str| {
1188 4800 : reg.csi
1189 4800 : .iter()
1190 4800 : .find(|(k, _)| k.eq_ignore_ascii_case(key))
1191 4800 : .map(|(_, v)| v.as_str())
1192 4800 : };
1193 : // "accept": the response shall come back in this format (the read path
1194 : // strips any body @context before expanding, so both forms import).
1195 1396 : let accept = csi_get("accept").unwrap_or("application/json");
1196 : // "ngsildConformance": amend the payload to the pinned version (4.3.6.8).
1197 1396 : if let Some(ver) = csi_get("ngsildConformance").and_then(crate::conformance::parse_version) {
1198 0 : if let Some(b) = body.as_mut() {
1199 0 : crate::conformance::amend_payload(b, ver);
1200 0 : }
1201 1396 : }
1202 : // "jsonldContext": recompact payload and term-bearing query parameters
1203 : // with the registered context, forward THAT context, Content-Type
1204 : // application/json, no @context member in the payload. Entity-shaped
1205 : // bodies only — a POST-query body is not translated and so keeps the
1206 : // original context; if either context fails to load the forward
1207 : // degrades to the original context rather than sending terms compacted
1208 : // against the wrong one.
1209 1396 : let mut link_ctx: String = ctx_url.to_owned();
1210 1396 : let mut answered_in = None;
1211 1396 : let mut query: Vec<(String, String)> = query.to_vec();
1212 : // 4.3.6.4: "a binding-specific mechanism to request operations only on
1213 : // the registered endpoint itself" — a localOnly registration (5.2.9)
1214 : // must not cascade, so the forward carries the 6.3.18 local parameter.
1215 1396 : if reg.local_only && !query.iter().any(|(k, _)| k == "local") {
1216 8 : query.push(("local".into(), "true".into()));
1217 1388 : }
1218 1396 : if let Some(reg_ctx_url) = csi_get("jsonldContext") {
1219 : // 5.5.10: both URLs are Tenant data — the caller's own @context and one
1220 : // the Registration names — so they resolve within the forwarding
1221 : // Tenant, never against a Hosted @context another Tenant stored.
1222 24 : let orig = st
1223 24 : .loader
1224 24 : .resolve_quiet_for(tenant, &Value::String(ctx_url.to_owned()))
1225 24 : .await;
1226 24 : let target = st
1227 24 : .loader
1228 24 : .resolve_quiet_for(tenant, &Value::String(reg_ctx_url.to_owned()))
1229 24 : .await;
1230 24 : if let (Ok(orig), Ok(target)) = (orig, target) {
1231 : // The payload reaching here has already been validated by the
1232 : // operation's own handler; this expansion only re-reads it to
1233 : // translate its terms, so it must accept everything that handler
1234 : // accepted and drop nothing. A merge fragment carries NGSI-LD
1235 : // Nulls (5.5.12), a temporal payload repeats a datasetId across
1236 : // instances (4.5.6), and a 4.5.7 tombstone is a deletedAt an
1237 : // expansion without `sys` discards — re-validating or thinning
1238 : // the body here changes the write the client asked for.
1239 24 : let mut translated: Option<Value> = None;
1240 24 : let mut can_switch = true;
1241 : // 6.6/6.7 name their Attribute in the PATH, which the Context
1242 : // Source expands with the @context the forward advertises — so
1243 : // the segment is translated with the payload or switching the
1244 : // context renames the target Attribute. 4.3.6.6 compacts
1245 : // "payload and query parameters" and the path is neither, but it
1246 : // carries a term the same compaction has to reach, or the rule
1247 : // cannot be applied to these resources at all. A name that does
1248 : // not expand to an absolute IRI is no term to translate, and the
1249 : // whole request stays in the @context its names are already in.
1250 24 : let path_swap = match path_attr_segment(&url) {
1251 10 : Some((range, name)) => match antares_jsonld::expand_attr_name(&name, &orig) {
1252 : // The compacted term is written into the request PATH, so
1253 : // it is held to the rule the client's own name was held to
1254 : // at the door (`antares_model::check_attr_name`): a `.` or
1255 : // `..` segment names a different resource of the Context
1256 : // Source endpoint, and a URL parser resolves it before the
1257 : // request is sent. A registered @context is client-supplied
1258 : // and may bind any term, so a term that is a dot segment
1259 : // means the request cannot be expressed in that context.
1260 10 : Ok(iri) => {
1261 10 : let term = target.compact_iri(&iri);
1262 10 : if antares_model::has_dot_segment(&term) {
1263 2 : can_switch = false;
1264 2 : None
1265 : } else {
1266 8 : Some((range, path_segment(&term)))
1267 : }
1268 : }
1269 : Err(_) => {
1270 0 : can_switch = false;
1271 0 : None
1272 : }
1273 : },
1274 14 : None => None,
1275 : };
1276 24 : match body.as_ref() {
1277 : // nothing to translate
1278 8 : None => {}
1279 : // 6.6/6.7 take an Attribute Fragment (5.6.4, 5.6.14, 5.6.19),
1280 : // not an Entity — its `type` is an Attribute type and its
1281 : // `value` a value, so reading it as an Entity would turn the
1282 : // value into a sub-Attribute. `expand_attr_fragment` is the
1283 : // reader the partial-update handler itself uses, and
1284 : // `compact_instance` its inverse.
1285 16 : Some(v) if path_swap.is_some() => match v.as_object() {
1286 6 : Some(o) => match antares_jsonld::expand_attr_fragment(o, &orig) {
1287 6 : Ok(exp) => {
1288 6 : let mut re = antares_jsonld::compact::compact_instance(&exp, &target);
1289 6 : if let Some(m) = re.as_object_mut() {
1290 6 : m.remove("@context");
1291 6 : }
1292 6 : translated = Some(re);
1293 : }
1294 0 : Err(_) => can_switch = false,
1295 : },
1296 0 : None => can_switch = false,
1297 : },
1298 10 : Some(v) => match v.as_object() {
1299 : // A 5.2.23 Query body names terms too (entities[].type,
1300 : // attrs, geoQ.geoproperty, q) and has no translation
1301 : // here, so it travels in the @context its terms are in.
1302 10 : Some(o) if o.get("type").and_then(Value::as_str) == Some("Query") => {
1303 2 : can_switch = false;
1304 2 : }
1305 8 : Some(o) => {
1306 8 : let opts = antares_jsonld::ExpandOpts {
1307 8 : fragment: o.get("id").is_none(),
1308 8 : allow_null: true,
1309 8 : merge: true,
1310 8 : temporal: true,
1311 8 : sys: true,
1312 8 : };
1313 8 : match antares_jsonld::expand_entity(o, &orig, opts) {
1314 8 : Ok(exp) => {
1315 8 : let mut re = antares_jsonld::compact::compact_entity(&exp, &target);
1316 8 : if let Some(m) = re.as_object_mut() {
1317 8 : m.remove("@context");
1318 8 : }
1319 8 : translated = Some(re);
1320 : }
1321 0 : Err(_) => can_switch = false,
1322 : }
1323 : }
1324 : // any other shape (a batch array) has no entity-document
1325 : // translation here, so it must not claim to be in the
1326 : // registered @context either
1327 0 : None => can_switch = false,
1328 : },
1329 : }
1330 : // 4.3.6.6 states one rule, not two: compact the payload with the
1331 : // registered @context "and forward with this JSON-LD context". A
1332 : // payload that could not be translated must therefore travel in
1333 : // the @context its terms ARE in — advertising the registered one
1334 : // over untranslated terms makes the Context Source expand them to
1335 : // different Fully Qualified Names (5.5.7) and write Attributes
1336 : // the client never named.
1337 24 : if can_switch {
1338 20 : if let Some((range, seg)) = path_swap {
1339 8 : url.replace_range(range, &seg);
1340 12 : }
1341 20 : if let Some(re) = translated {
1342 14 : body = Some(re);
1343 14 : }
1344 20 : for (k, v) in query.iter_mut() {
1345 10 : if matches!(k.as_str(), "attrs" | "type" | "geoproperty") {
1346 4 : *v = v
1347 4 : .split(',')
1348 4 : .map(|t| target.compact_iri(&orig.expand_key(t.trim())))
1349 4 : .collect::<Vec<_>>()
1350 4 : .join(",");
1351 6 : }
1352 : }
1353 20 : link_ctx = reg_ctx_url.to_owned();
1354 20 : answered_in = Some(target);
1355 : } else {
1356 4 : tracing::warn!(
1357 : "payload could not be expressed in the registered jsonldContext \
1358 : {reg_ctx_url}; forwarding with the original context"
1359 : );
1360 : }
1361 : } else {
1362 0 : tracing::warn!(
1363 : "registered jsonldContext {reg_ctx_url} (or the request context) \
1364 : failed to load; forwarding with the original context"
1365 : );
1366 : }
1367 1372 : }
1368 : // An inline @context (serialized JSON from ctx_link_url) cannot travel
1369 : // as a Link header — it is embedded in the body below (5.5.7/6.3.5).
1370 1396 : let inline_ctx: Option<Value> = if link_ctx.starts_with('[') || link_ctx.starts_with('{') {
1371 0 : serde_json::from_str(&link_ctx).ok()
1372 : } else {
1373 1396 : None
1374 : };
1375 1396 : let mut sent: Vec<(String, String)> = vec![
1376 1396 : ("Accept".into(), accept.to_owned()),
1377 1396 : (
1378 1396 : "Via".into(),
1379 1396 : outbound_via(headers, &alias_for(&st.host_alias, tenant)),
1380 1396 : ),
1381 : ];
1382 1396 : if inline_ctx.is_none() {
1383 1396 : sent.push((
1384 1396 : "Link".into(),
1385 1396 : format!("<{link_ctx}>; rel=\"http://www.w3.org/ns/json-ld#context\"; type=\"application/ld+json\""),
1386 1396 : ));
1387 1396 : }
1388 : // 4.14: "the Tenant information from the Context Source Registration has
1389 : // to be used. If no Tenant information is present in the Context Source
1390 : // Registration, no Tenant information is to be used and thus the default
1391 : // Tenant is targeted" — the requesting tenant never flows through; 6.3.14
1392 : // omits the header for the default Tenant.
1393 1396 : if let Some(peer_tenant) = reg.tenant.as_deref().filter(|t| *t != "default") {
1394 16 : sent.push(("NGSILD-Tenant".into(), peer_tenant.to_owned()));
1395 1380 : }
1396 : // 4.3.6.5 contextSourceInfo ⇒ extra headers; the special value
1397 : // "urn:ngsi-ld:request" copies the header from the triggering request
1398 : // (dropped when the triggering request did not carry it).
1399 1396 : for (k, v) in ®.csi {
1400 82 : let lower = k.to_ascii_lowercase();
1401 82 : if CSI_SKIP.contains(&lower.as_str()) {
1402 38 : continue;
1403 44 : }
1404 : // ADR-0020: the subject never leaves this process. A registration is
1405 : // client-supplied, so without this a registration that names the
1406 : // subject header with "urn:ngsi-ld:request" would copy the identity
1407 : // of every request that fans out to it onto the wire — the peer
1408 : // would be handed a credential the broker was only ever given to
1409 : // ask its own engine about. Dropped rather than refused at 5.9.2:
1410 : // which headers a deployment made subject headers is not something
1411 : // a client gets to probe by watching which registrations fail.
1412 44 : if crate::policy::SUBJECT_HEADERS.contains(&lower) {
1413 2 : tracing::warn!(
1414 : "registration {} asks for the policy subject header {k:?}; not forwarded",
1415 : reg.reg_id
1416 : );
1417 2 : continue;
1418 42 : }
1419 42 : let val = if v == "urn:ngsi-ld:request" {
1420 : // the triggering request's own credential is the
1421 : // caller's, not the registrant's: copying it would hand every
1422 : // caller's token or session to whoever registered the source.
1423 : // A value the registration states itself is its own and goes.
1424 18 : if REQUEST_CREDENTIALS.contains(&lower.as_str()) {
1425 6 : tracing::warn!(
1426 : "registration {} asks for the request's credential header {k:?}; not forwarded",
1427 : reg.reg_id
1428 : );
1429 6 : continue;
1430 12 : }
1431 12 : match headers.get(k.as_str()).and_then(|h| h.to_str().ok()) {
1432 6 : Some(h) => h.to_owned(),
1433 6 : None => continue,
1434 : }
1435 : } else {
1436 24 : v.clone()
1437 : };
1438 30 : sent.push((k.clone(), val));
1439 : }
1440 1396 : let mut bytes = None;
1441 1396 : if let Some(mut b) = body {
1442 : // 4.3.6.6 "contentType": provide request + @context as the MIME type
1443 : // requires — ld+json carries the context inline. When "jsonldContext"
1444 : // is also defined its own mandate wins ("the Content-Type of the
1445 : // forwarded request shall be application/json").
1446 608 : let want_ld = csi_get("contentType") == Some("application/ld+json")
1447 4 : && csi_get("jsonldContext").is_none();
1448 608 : if let Some(ic) = &inline_ctx {
1449 : // inline request @context: the only lossless carrier is the
1450 : // body itself, as application/ld+json (5.5.7/6.3.5).
1451 0 : if let Some(o) = b.as_object_mut() {
1452 0 : o.insert("@context".into(), ic.clone());
1453 0 : }
1454 0 : sent.push(("Content-Type".into(), "application/ld+json".into()));
1455 608 : } else if want_ld {
1456 4 : if let Some(o) = b.as_object_mut() {
1457 4 : o.insert("@context".into(), Value::String(link_ctx.clone()));
1458 4 : }
1459 4 : sent.push(("Content-Type".into(), "application/ld+json".into()));
1460 604 : } else {
1461 604 : sent.push(("Content-Type".into(), "application/json".into()));
1462 604 : }
1463 608 : bytes = Some(serde_json::to_vec(&b).unwrap_or_default());
1464 788 : }
1465 1396 : let mut subject = HeaderMap::new();
1466 1396 : for name in crate::policy::SUBJECT_HEADERS.iter() {
1467 8 : let Ok(name) = axum::http::HeaderName::from_bytes(name.as_bytes()) else {
1468 0 : continue;
1469 : };
1470 8 : for v in headers.get_all(&name) {
1471 6 : subject.append(name.clone(), v.clone());
1472 6 : }
1473 : }
1474 1396 : BuiltForward {
1475 1396 : method,
1476 1396 : url,
1477 1396 : query,
1478 1396 : headers: sent,
1479 1396 : body: bytes,
1480 1396 : tenant,
1481 1396 : reg,
1482 1396 : subject,
1483 1396 : answered_in,
1484 1396 : }
1485 1396 : }
1486 :
1487 : /// One forward over HTTP: the egress policy, the breaker and the cooldown
1488 : /// first, then the request `built` describes.
1489 1388 : async fn send_http(st: &AppState, built: BuiltForward<'_>) -> ForwardAnswer {
1490 : let BuiltForward {
1491 1388 : method,
1492 1388 : url,
1493 1388 : query,
1494 1388 : headers: sent,
1495 1388 : body,
1496 1388 : tenant,
1497 1388 : reg,
1498 : ..
1499 1388 : } = built;
1500 : // 6.3.17: NGSILD-Warning values received from the peer are returned to
1501 : // the caller — abnormal behaviour detected downstream in a cascade
1502 : // (4.3.6.4) must surface on the aggregated response, not vanish here.
1503 : // One policy for every outbound class — scheme allowlist,
1504 : // private-range deny, per-destination circuit breaker.
1505 1388 : if let Err(e) = st.egress.check_url(&url).await {
1506 : // 5.2.9 allows any URI as the registered endpoint, and a URI's
1507 : // authority may carry credentials reqwest sends as basic auth, so a
1508 : // peer URL loses its userinfo on the way into the log.
1509 0 : tracing::warn!(
1510 : "federation forward to {} refused: {e}",
1511 0 : antares_notifier::redact_userinfo(&url)
1512 : );
1513 0 : return (502, Value::Null, Vec::new());
1514 1388 : }
1515 1388 : if st.egress.is_open(tenant.as_str(), &url) {
1516 0 : tracing::debug!(
1517 : "federation forward to {} short-circuited (breaker open)",
1518 0 : antares_notifier::redact_userinfo(&url)
1519 : );
1520 0 : return (503, Value::Null, Vec::new());
1521 1388 : }
1522 : // 5.2.34 cooldown: the source is not contacted.
1523 1388 : if in_cooldown(st, tenant, reg) {
1524 0 : return (504, Value::Null, Vec::new());
1525 1388 : }
1526 1388 : let mut req = st.fed_http.request(method, &url);
1527 4816 : for (k, v) in &sent {
1528 4816 : req = req.header(k.as_str(), v.as_str());
1529 4816 : }
1530 1388 : if !query.is_empty() {
1531 722 : req = req.query(&query);
1532 784 : }
1533 1388 : if let Some(b) = body {
1534 606 : req = req.body(b);
1535 812 : }
1536 : // The whole HTTP interaction is one Send unit (http_interaction) so
1537 : // the handler futures above stay Send on wasm32 too.
1538 1388 : antares_jsonld::http_interaction(async {
1539 : // wasm has no client-level timeout — bound the forward per request
1540 : // (mirrors the native fed_http 8 s total); a timed-out
1541 : // forward is the only failure class that feeds the breaker.
1542 : // 5.2.34 timeout bounds the forward below the 8 s ceiling.
1543 : // Natively io_deadline is a passthrough (the client owns the 8 s
1544 : // default), so the per-registration budget rides on the request.
1545 1388 : let deadline = deadline_ms(reg);
1546 : #[cfg(not(target_arch = "wasm32"))]
1547 1388 : let req = req.timeout(std::time::Duration::from_millis(deadline as u64));
1548 1388 : let sent = antares_jsonld::io_deadline(req.send(), deadline).await;
1549 1380 : let sent = match sent {
1550 1380 : Some(r) => r,
1551 : None => {
1552 0 : st.egress.record_failure(tenant.as_str(), &url);
1553 0 : note_forward(st, tenant, reg, false).await;
1554 0 : return (504, Value::Null, Vec::new());
1555 : }
1556 : };
1557 250 : match sent {
1558 1130 : Ok(resp) => {
1559 : // Any response — even 5xx — proves the peer answers within
1560 : // its own response time: no deadline cost, so no breaker
1561 : // state. Only TIMEOUT-class failures trip; a
1562 : // responding-but-erroring peer must keep being attempted,
1563 : // else unrelated registrations sharing its host:port starve.
1564 1130 : let status = resp.status().as_u16();
1565 1130 : st.egress.record_success(tenant.as_str(), &url);
1566 1130 : note_forward(st, tenant, reg, (200..300).contains(&status)).await;
1567 : // 6.3.17: the peer's own values travel on (4.3.6.4), but
1568 : // only up to the cap — the list is written by the peer and
1569 : // sent by this broker, and past the cap one source would
1570 : // both outgrow the fan-out and crowd out the warnings the
1571 : // clause obliges this broker to raise about the others.
1572 : // The head is kept: a source states its own outcome before
1573 : // the ones it relays.
1574 1130 : let all = resp.headers().get_all("NGSILD-Warning");
1575 1130 : let peer_warnings: Vec<String> = all
1576 1130 : .iter()
1577 1130 : .filter_map(|v| v.to_str().ok().map(str::to_owned))
1578 1130 : .take(crate::bounds::MAX_PEER_WARNINGS)
1579 1130 : .collect();
1580 1130 : if all.iter().count() > peer_warnings.len() {
1581 4 : tracing::warn!(
1582 : "registration {} sent more than {} NGSILD-Warning values; the rest were dropped",
1583 : reg.reg_id,
1584 : crate::bounds::MAX_PEER_WARNINGS
1585 : );
1586 1126 : }
1587 1130 : let body = read_body_capped(resp).await;
1588 1130 : (status, body, peer_warnings)
1589 : }
1590 250 : Err(e) if e.is_timeout() => {
1591 6 : st.egress.record_failure(tenant.as_str(), &url);
1592 6 : note_forward(st, tenant, reg, false).await;
1593 6 : (504, Value::Null, Vec::new())
1594 : }
1595 : Err(_) => {
1596 : // connect refused/reset: fails in milliseconds — no deadline cost;
1597 : // clearing avoids stale suppression of a restarted peer.
1598 : // 503 (not 502): NO HTTP response was received, so the read
1599 : // path classifies it under Table 6.3.17-1 code 199 ("No
1600 : // response was received from the registration endpoint"),
1601 : // never 299 ("An error response ... was received").
1602 244 : st.egress.record_success(tenant.as_str(), &url);
1603 244 : note_forward(st, tenant, reg, false).await;
1604 244 : (503, Value::Null, Vec::new())
1605 : }
1606 : }
1607 1380 : })
1608 1388 : .await
1609 1380 : }
1610 :
1611 : /// 6.3.17: a forwarded response contributes the NGSILD-Warning values the
1612 : /// peer sent AND, for an abnormal outcome, one warning of this broker's own
1613 : /// — whether or not the payload can be used. Only a 2xx payload can be:
1614 : /// 4.3.6.4 makes a failed forward a warning on a successful overall
1615 : /// response, never a failure of it, so a non-2xx registration is dropped
1616 : /// from the answer after it has been reported.
1617 556 : fn usable_payload(
1618 556 : st: &AppState,
1619 556 : tenant: &TenantId,
1620 556 : status: u16,
1621 556 : body: &Value,
1622 556 : peer_warns: Vec<String>,
1623 556 : warnings: &mut Vec<String>,
1624 556 : ) -> bool {
1625 556 : warnings.extend(peer_warns);
1626 556 : if let Some((code, text)) = read_warning(status, body) {
1627 178 : warnings.push(warning(code, &alias_for(&st.host_alias, tenant), text));
1628 378 : }
1629 556 : (200..300).contains(&status)
1630 556 : }
1631 :
1632 : /// 5.2.9 `attrs`: a registration's scope narrows what a forwarded read asks
1633 : /// for. The names go out compacted against the request @context, because the
1634 : /// peer is asked in terms, not in IRIs.
1635 208 : fn scope_attrs(reg: &FedReg, ctx: &Context) -> Option<(String, String)> {
1636 208 : let scope = reg.attrs.as_ref()?;
1637 66 : let names: Vec<String> = scope.iter().map(|a| ctx.compact_iri(a)).collect();
1638 66 : Some(("attrs".into(), names.join(",")))
1639 208 : }
1640 :
1641 : /// The 4.11 temporal window a forwarded temporal read carries (5.7.3.4 and
1642 : /// 5.7.4.4 ask the peer over the SAME window), plus sysAttrs on every
1643 : /// forwarded read: conflicting instances resolve by most recent
1644 : /// observedAt/modifiedAt (4.5.5.3), and without the remote modifiedAt the
1645 : /// winner would be arrival order.
1646 30 : fn temporal_window(params: &HashMap<String, String>) -> Vec<(String, String)> {
1647 30 : let mut query: Vec<(String, String)> = vec![("options".into(), "sysAttrs".into())];
1648 120 : for k in ["timerel", "timeAt", "endTimeAt", "timeproperty"] {
1649 120 : if let Some(v) = params.get(k) {
1650 48 : query.push((k.into(), v.clone()));
1651 72 : }
1652 : }
1653 30 : query.extend(instance_cap(params));
1654 30 : query
1655 30 : }
1656 :
1657 : /// firstN/lastN forwarded to a peer: every instance up to the end of the
1658 : /// asked page, never offsetN. The merged evolution is paged here, so a peer
1659 : /// that skipped offsetN itself would have this broker skip them twice.
1660 48 : fn instance_cap(params: &HashMap<String, String>) -> Option<(String, String)> {
1661 48 : let offset: usize = params
1662 48 : .get("offsetN")
1663 48 : .and_then(|o| o.parse().ok())
1664 48 : .unwrap_or(0);
1665 96 : ["firstN", "lastN"].into_iter().find_map(|k| {
1666 96 : let n: usize = params.get(k)?.parse().ok()?;
1667 0 : Some((k.to_owned(), n.saturating_add(offset).to_string()))
1668 96 : })
1669 48 : }
1670 :
1671 : /// 5.2.9 `attrs` again, on the way back: the scope narrows ATTRIBUTES, so the
1672 : /// entity-level members of 4.5.1 stay whatever the scope says — dropping
1673 : /// them would let 4.5.5.3 read a missing `expiresAt` as "absent from a
1674 : /// received version" and delete the local one.
1675 648 : fn narrow_to_scope(expanded: Value, reg: &FedReg) -> Option<Value> {
1676 648 : let Some(scope) = ®.attrs else {
1677 586 : return Some(expanded);
1678 : };
1679 62 : let mut out = Map::new();
1680 196 : for (k, v) in expanded.as_object()? {
1681 196 : if crate::repr::ENTITY_META.contains(&k.as_str()) || scope.iter().any(|s| s == k) {
1682 190 : out.insert(k.clone(), v.clone());
1683 190 : }
1684 : }
1685 62 : Some(Value::Object(out))
1686 648 : }
1687 :
1688 : /// Expand + registration-scope-filter one remote compacted entity.
1689 628 : pub fn import_entity(remote: &Value, reg: &FedReg, ctx: &Context) -> Option<Value> {
1690 628 : let mut obj = remote.as_object()?.clone();
1691 628 : obj.remove("@context");
1692 628 : let expanded = antares_jsonld::expand_entity(
1693 628 : &obj,
1694 628 : ctx,
1695 628 : antares_jsonld::ExpandOpts {
1696 628 : sys: true,
1697 628 : ..Default::default()
1698 628 : },
1699 : )
1700 628 : .ok()?;
1701 628 : narrow_to_scope(expanded, reg)
1702 628 : }
1703 :
1704 : /// Expand + registration-scope-filter one remote temporal entity (5.7.3.4).
1705 : /// Instances of one datasetId legally repeat in a Temporal Evolution, so
1706 : /// expansion runs in temporal mode.
1707 20 : fn import_temporal(remote: &Value, reg: &FedReg, ctx: &Context) -> Option<Value> {
1708 20 : let mut obj = remote.as_object()?.clone();
1709 20 : obj.remove("@context");
1710 20 : let expanded = antares_jsonld::expand_entity(
1711 20 : &obj,
1712 20 : ctx,
1713 20 : antares_jsonld::ExpandOpts {
1714 20 : sys: true,
1715 20 : temporal: true,
1716 20 : // 4.5.7: deletion instances (value urn:ngsi-ld:null +
1717 20 : // deletedAt) are part of a Temporal Evolution — a remote
1718 20 : // tombstone must import, not be dropped as an invalid payload.
1719 20 : allow_null: true,
1720 20 : ..Default::default()
1721 20 : },
1722 : )
1723 20 : .ok()?;
1724 20 : narrow_to_scope(expanded, reg)
1725 20 : }
1726 :
1727 : /// 5.7.3.4: forward Retrieve Temporal Evolution to matching registrations
1728 : /// that support the retrieveTemporal operation; registrations without it
1729 : /// are not contacted. Returns (auxiliary, expanded doc) pairs for the
1730 : /// caller's 4.5.5 merge.
1731 : /// 4.20: does a raw registration document support `op`? Same group tables
1732 : /// as FedReg::supports; default when the operations member is absent is
1733 : /// federationOps (5.2.9).
1734 144 : pub(crate) fn doc_supports(reg: &Value, op: &str) -> bool {
1735 144 : fed_reg_of(reg.get("id").and_then(Value::as_str).unwrap_or(""), reg).supports(op)
1736 144 : }
1737 :
1738 : /// 5.2.34 `management.localOnly`; the top-level spelling is kept for
1739 : /// compatibility (4.3.6.4 wording / older payloads). Read here so every
1740 : /// FedReg of one document agrees on it — a view that reads it and a view
1741 : /// that does not send the same peer two different questions.
1742 2068 : fn local_only_of(reg: &Value) -> bool {
1743 2068 : reg.get("localOnly")
1744 2068 : .and_then(Value::as_bool)
1745 2068 : .or_else(|| {
1746 2052 : reg.get("management")
1747 2052 : .and_then(|m| m.get("localOnly"))
1748 2052 : .and_then(Value::as_bool)
1749 2052 : })
1750 2068 : .unwrap_or(false)
1751 2068 : }
1752 :
1753 : /// A minimal FedReg view of a raw registration document — enough for
1754 : /// `forward` (endpoint/tenant/csi/alias) and `supports`.
1755 : /// 4.3.6.6 contextSourceInfo, as the key/value pairs a forward applies.
1756 2068 : fn csi_of(reg: &Value) -> Vec<(String, String)> {
1757 2068 : crate::notify::kv_pairs(reg.get("contextSourceInfo"))
1758 2068 : }
1759 :
1760 : /// A registration's endpoint without a trailing `/` or the API root it may name
1761 : /// (`…/ngsi-ld/v1`), so a forward URL never doubles the prefix (IOP fixtures).
1762 2184 : pub(crate) fn endpoint_base(endpoint: &str) -> String {
1763 2184 : let e = endpoint.trim_end_matches('/');
1764 2184 : e.strip_suffix("/ngsi-ld/v1").unwrap_or(e).to_owned()
1765 2184 : }
1766 :
1767 304 : pub(crate) fn fed_reg_of(reg_id: &str, reg: &Value) -> FedReg {
1768 304 : let endpoint = reg
1769 304 : .get("endpoint")
1770 304 : .and_then(Value::as_str)
1771 304 : .map(endpoint_base)
1772 304 : .unwrap_or_default();
1773 : FedReg {
1774 304 : reg_id: reg_id.to_owned(),
1775 304 : endpoint,
1776 304 : mode: reg
1777 304 : .get("mode")
1778 304 : .and_then(Value::as_str)
1779 304 : .unwrap_or("inclusive")
1780 304 : .to_owned(),
1781 304 : ops: reg
1782 304 : .get("operations")
1783 304 : .and_then(Value::as_array)
1784 304 : .map(|a| {
1785 300 : a.iter()
1786 300 : .filter_map(Value::as_str)
1787 300 : .map(str::to_owned)
1788 300 : .collect()
1789 300 : })
1790 304 : .unwrap_or_else(|| vec![DEFAULT_OPERATION_GROUP.into()]),
1791 304 : attrs: None,
1792 304 : ent_ids: Vec::new(),
1793 304 : ent_types: Vec::new(),
1794 304 : ent_patterns: Vec::new(),
1795 : // minimal view: no EntityInfo scope loaded ⇒ never narrow by it
1796 : ent_unrestricted: true,
1797 304 : tenant: reg.get("tenant").and_then(Value::as_str).map(str::to_owned),
1798 304 : alias: reg
1799 304 : .get("contextSourceAlias")
1800 304 : .and_then(Value::as_str)
1801 304 : .map(str::to_owned),
1802 : // 4.3.6.6: the registered headers (auth among them) travel with every
1803 : // forward, including the subscription operations that build their
1804 : // registration view from this function.
1805 304 : csi: csi_of(reg),
1806 304 : local_only: local_only_of(reg),
1807 304 : timeout_ms: reg
1808 304 : .get("management")
1809 304 : .and_then(|m| m.get("timeout"))
1810 304 : .and_then(Value::as_u64),
1811 304 : cooldown_ms: reg
1812 304 : .get("management")
1813 304 : .and_then(|m| m.get("cooldown"))
1814 304 : .and_then(Value::as_u64),
1815 : }
1816 304 : }
1817 :
1818 : /// 5.7.1.4 / 5.7.3.4: with an EntityMap in use, "only the retrieved Entity
1819 : /// Map shall be used to determine which Context Source Registrations match
1820 : /// the Entity ID" — a registration not listed in the entry does not match;
1821 : /// "the location of the linked EntityMap shall be passed as part of any
1822 : /// forwarded request" (conveyed as an extra header on the forward).
1823 206 : fn map_gate(mut reg: FedReg, map: Option<&Value>, id: &str) -> Option<FedReg> {
1824 206 : let Some(m) = map else { return Some(reg) };
1825 2 : let listed = m
1826 2 : .get("entityMap")
1827 2 : .and_then(|e| e.get(id))
1828 2 : .and_then(Value::as_array)
1829 2 : .is_some_and(|a| a.iter().any(|v| v.as_str() == Some(reg.reg_id.as_str())));
1830 2 : if !listed {
1831 2 : return None;
1832 0 : }
1833 0 : if let Some(mid) = m
1834 0 : .get("linkedMaps")
1835 0 : .and_then(|l| l.get(®.reg_id))
1836 0 : .and_then(Value::as_str)
1837 0 : {
1838 0 : reg.csi.push(("NGSILD-EntityMap".into(), mid.to_owned()));
1839 0 : }
1840 0 : Some(reg)
1841 206 : }
1842 :
1843 : #[allow(clippy::too_many_arguments)] // mirrors the wire: one param per forwarded request part
1844 980 : pub async fn fed_retrieve_temporal(
1845 980 : st: &AppState,
1846 980 : tenant: &TenantId,
1847 980 : headers: &HeaderMap,
1848 980 : ctx: &Context,
1849 980 : id: &str,
1850 980 : params: &HashMap<String, String>,
1851 980 : map: Option<&Value>,
1852 980 : warnings: &mut Vec<String>,
1853 980 : proxies: &mut Vec<FedReg>,
1854 980 : ) -> Result<Vec<(bool, Value)>, NgsiError> {
1855 980 : let spec = crate::registry::CsrSpec {
1856 980 : ids: Some(vec![id.to_owned()]),
1857 980 : temporal: crate::temporalq::TemporalQ::from_params(params, false)
1858 980 : .ok()
1859 980 : .flatten(),
1860 980 : ..Default::default()
1861 980 : };
1862 980 : let ctx_url = ctx_link_url(headers, &ctx.source);
1863 980 : let ctx_url = &ctx_url;
1864 980 : let matched = matching_regs(st, tenant, &spec, ctx, headers).await?;
1865 980 : proxies.extend(matched.iter().filter(|r| r.is_proxy()).cloned());
1866 980 : let regs: Vec<FedReg> = matched
1867 980 : .into_iter()
1868 : // 5.7.3.4: a live EntityMap in use is the ONLY source of matching
1869 : // registrations; its linked map location travels with the forward.
1870 980 : .filter_map(|reg| map_gate(reg, map, id))
1871 980 : .filter(|reg| reg.supports("retrieveTemporal"))
1872 980 : .collect();
1873 980 : let fetched = fan_out(regs, move |reg| async move {
1874 : // the temporal window travels with the forward; sysAttrs for the
1875 : // 4.5.5.3 recency arbitration
1876 8 : let mut query = temporal_window(params);
1877 8 : query.extend(scope_attrs(®, ctx));
1878 8 : let ((status, body, peer_warns), actx) = forward_in(
1879 8 : st,
1880 8 : reqwest::Method::GET,
1881 8 : format!(
1882 : "{}/ngsi-ld/v1/temporal/entities/{}",
1883 : reg.endpoint,
1884 8 : path_segment(id)
1885 : ),
1886 8 : &query,
1887 8 : headers,
1888 8 : tenant,
1889 8 : ®,
1890 8 : ctx_url,
1891 8 : None,
1892 : )
1893 8 : .await;
1894 8 : (reg, status, body, peer_warns, actx)
1895 16 : })
1896 980 : .await;
1897 980 : let mut out = Vec::new();
1898 980 : for (reg, status, body, peer_warns, actx) in fetched {
1899 8 : if !usable_payload(st, tenant, status, &body, peer_warns, warnings) {
1900 0 : continue;
1901 8 : }
1902 8 : if body.get("id").and_then(Value::as_str) != Some(id) {
1903 4 : continue;
1904 4 : }
1905 4 : match import_temporal(&body, ®, actx.as_deref().unwrap_or(ctx)) {
1906 4 : Some(doc) => out.push((reg.mode == "auxiliary", doc)),
1907 0 : None => warnings.push(warning(
1908 : 111,
1909 0 : &alias_for(&st.host_alias, tenant),
1910 0 : "the payload of the response was invalid",
1911 : )),
1912 : }
1913 : }
1914 980 : Ok(out)
1915 980 : }
1916 :
1917 : /// The rank 4.5.5.3 arbitrates on: TS 104 175 8.5.3 has any instance with
1918 : /// an observedAt take precedence over every instance without one, the most
1919 : /// recent observedAt among them; modifiedAt decides only between instances
1920 : /// that observed nothing. The DateTime is a comparable key: 4.6.3 admits the
1921 : /// same instant with or without a fraction, so a raw string compare would
1922 : /// rank `…:01.500Z` below `…:01Z`.
1923 220 : fn recency(inst: &Value) -> (bool, String) {
1924 368 : let at = |k| inst.get(k).and_then(Value::as_str);
1925 220 : match at("observedAt") {
1926 72 : Some(observed) => (true, antares_model::dt_key(observed)),
1927 148 : None => (false, antares_model::dt_key(at("modifiedAt").unwrap_or(""))),
1928 : }
1929 220 : }
1930 :
1931 : /// 4.5.5.2 Processing of Conflicting Transient Entities: for each received
1932 : /// Entity version with an entity-level expiresAt, add it as a non-reified
1933 : /// expiresAt on every Attribute instance that lacks one, and cap any
1934 : /// Attribute-level expiresAt further in the future to the entity's (earlier)
1935 : /// DateTime.
1936 176 : fn push_down_expires(doc: &mut Value) {
1937 176 : let Some(o) = doc.as_object_mut() else { return };
1938 176 : let Some(exp) = o.get("expiresAt").and_then(Value::as_str).map(String::from) else {
1939 160 : return;
1940 : };
1941 52 : for (k, v) in o.iter_mut() {
1942 48 : if matches!(
1943 52 : k.as_str(),
1944 52 : "id" | "type" | "scope" | "expiresAt" | "createdAt" | "modifiedAt" | "deletedAt"
1945 : ) {
1946 48 : continue;
1947 4 : }
1948 4 : let Some(instances) = v.as_array_mut() else {
1949 0 : continue;
1950 : };
1951 12 : for inst in instances.iter_mut().filter_map(Value::as_object_mut) {
1952 12 : match inst.get("expiresAt").and_then(Value::as_str) {
1953 8 : Some(ae) if antares_model::dt_key(ae) <= antares_model::dt_key(&exp) => {}
1954 8 : _ => {
1955 8 : inst.insert("expiresAt".into(), Value::String(exp.clone()));
1956 8 : }
1957 : }
1958 : }
1959 : }
1960 176 : }
1961 :
1962 : /// 4.5.5.3 first step: "if an expiresAt DateTime is present on
1963 : /// the Attribute and the date lies in the past, it shall be discarded" —
1964 : /// BEFORE any recency comparison.
1965 232 : fn expired(inst: &Value, now: &str) -> bool {
1966 232 : inst.get("expiresAt")
1967 232 : .and_then(Value::as_str)
1968 232 : .is_some_and(|e| antares_model::dt_key(e) < antares_model::dt_key(now))
1969 232 : }
1970 :
1971 : /// 4.3.6.2: "An auxiliary Context Source Registration never overrides data
1972 : /// held directly within a Context Broker. […] Context data from auxiliary
1973 : /// context sources is only included if it is supplementary."
1974 : /// Merge attributes of `add` into `base` (auxiliary sources never override —
1975 : /// base wins; otherwise conflicting instances resolve per 4.5.5.3: discard
1976 : /// past-expiresAt instances first, then most recent observedAt/modifiedAt).
1977 176 : pub fn merge_docs(base: &mut Value, add: &Value, aux: bool) {
1978 176 : let now = crate::state::now_iso();
1979 176 : let mut add = add.clone();
1980 176 : push_down_expires(&mut add);
1981 176 : let Some(bo) = base.as_object_mut() else {
1982 0 : return;
1983 : };
1984 176 : let Some(ao) = add.as_object() else { return };
1985 : // 4.5.5.3: entity-level expiresAt — "missing from at least one version of
1986 : // the Entity received" → removed; present in all versions → the DateTime
1987 : // furthest in the future. 4.3.6.2 keeps auxiliary versions out of this:
1988 : // they never override data the broker holds directly, and removing or
1989 : // extending the lifetime is an override.
1990 176 : if !aux {
1991 : match (
1992 154 : bo.get("expiresAt").and_then(Value::as_str),
1993 154 : ao.get("expiresAt").and_then(Value::as_str),
1994 : ) {
1995 4 : (Some(b), Some(a)) => {
1996 4 : if antares_model::dt_key(a) > antares_model::dt_key(b) {
1997 4 : bo.insert("expiresAt".into(), Value::String(a.to_owned()));
1998 4 : }
1999 : }
2000 150 : _ => {
2001 150 : bo.remove("expiresAt");
2002 150 : }
2003 : }
2004 22 : }
2005 590 : for (k, v) in ao {
2006 590 : if k == "expiresAt" {
2007 16 : continue; // resolved above
2008 574 : }
2009 574 : match bo.get_mut(k) {
2010 88 : None => {
2011 88 : bo.insert(k.clone(), v.clone());
2012 88 : }
2013 486 : Some(cur) if !aux && !matches!(k.as_str(), "id" | "type" | "scope") => {
2014 118 : let (Some(ca), Some(aa)) = (cur.as_array_mut(), v.as_array()) else {
2015 0 : continue;
2016 : };
2017 118 : for ai in aa {
2018 118 : if expired(ai, &now) {
2019 4 : continue; // 4.5.5.3: discarded before comparison
2020 114 : }
2021 114 : let ds = ai.get("datasetId").and_then(Value::as_str);
2022 114 : match ca
2023 114 : .iter_mut()
2024 114 : .find(|ci| ci.get("datasetId").and_then(Value::as_str) == ds)
2025 : {
2026 0 : None => ca.push(ai.clone()),
2027 114 : Some(ci) => {
2028 114 : if expired(ci, &now) || recency(ai) > recency(ci) {
2029 24 : *ci = ai.clone();
2030 90 : }
2031 : }
2032 : }
2033 : }
2034 : }
2035 368 : _ => {}
2036 : }
2037 : }
2038 176 : }
2039 :
2040 : // ---------- distributed reads ----------
2041 :
2042 : /// Federated retrieve: internal docs from every matching registration,
2043 : /// (aux, doc) pairs so callers can order the merge. `sole_failure` receives
2044 : /// the failed answer when the one registration asked is exclusive or
2045 : /// redirect: 6.3.17 p.278 gives that single source its own error statuses.
2046 : #[allow(clippy::too_many_arguments)] // mirrors the wire: one param per forwarded request part
2047 2008 : pub async fn fed_retrieve(
2048 2008 : st: &AppState,
2049 2008 : tenant: &TenantId,
2050 2008 : headers: &HeaderMap,
2051 2008 : ctx: &Context,
2052 2008 : id: &str,
2053 2008 : map: Option<&Value>,
2054 2008 : except_reg: Option<&str>,
2055 2008 : warnings: &mut Vec<String>,
2056 2008 : sole_failure: &mut Option<Part>,
2057 2008 : ) -> Result<Vec<(bool, Value)>, NgsiError> {
2058 2008 : let spec = crate::registry::CsrSpec {
2059 2008 : ids: Some(vec![id.to_owned()]),
2060 2008 : ..Default::default()
2061 2008 : };
2062 2008 : let ctx_url = ctx_link_url(headers, &ctx.source);
2063 2008 : let ctx_url = &ctx_url;
2064 2008 : let regs: Vec<FedReg> = matching_regs(st, tenant, &spec, ctx, headers)
2065 2008 : .await?
2066 2006 : .into_iter()
2067 : // 5.8.6 splitEntities merge: "except for the one from which the
2068 : // Notification has been received"
2069 2006 : .filter(|reg| !except_reg.is_some_and(|x| x == reg.reg_id))
2070 : // 5.7.1.4: a live EntityMap in use is the ONLY source of matching
2071 : // registrations; its linked map location travels with the forward.
2072 2006 : .filter_map(|reg| map_gate(reg, map, id))
2073 2006 : .filter(|reg| reg.read_op().is_some())
2074 2006 : .collect();
2075 2006 : let fetched = fan_out(regs, move |reg| async move {
2076 178 : let Some(op) = reg.read_op() else {
2077 0 : return (reg, 0, Value::Null, Vec::new(), None);
2078 : };
2079 : // sysAttrs on every forwarded read: conflicting instances resolve by
2080 : // most recent observedAt/modifiedAt (4.5.5.3) — without the remote
2081 : // modifiedAt the winner would be arrival order, i.e. indeterminate.
2082 178 : let mut query: Vec<(String, String)> = vec![("options".into(), "sysAttrs".into())];
2083 178 : query.extend(scope_attrs(®, ctx));
2084 178 : let ((status, body, peer_warns), actx) = match op {
2085 178 : "retrieveEntity" => {
2086 170 : forward_in(
2087 170 : st,
2088 170 : reqwest::Method::GET,
2089 170 : format!("{}/ngsi-ld/v1/entities/{}", reg.endpoint, path_segment(id)),
2090 170 : &query,
2091 170 : headers,
2092 170 : tenant,
2093 170 : ®,
2094 170 : ctx_url,
2095 170 : None,
2096 170 : )
2097 170 : .await
2098 : }
2099 8 : "queryEntity" => {
2100 4 : let t = reg
2101 4 : .ent_types
2102 4 : .first()
2103 4 : .map(|t| ctx.compact_iri(t))
2104 4 : .unwrap_or_else(|| "*".into());
2105 4 : query.push(("type".into(), t));
2106 4 : query.push(("id".into(), id.to_owned()));
2107 4 : forward_in(
2108 4 : st,
2109 4 : reqwest::Method::GET,
2110 4 : format!("{}/ngsi-ld/v1/entities", reg.endpoint),
2111 4 : &query,
2112 4 : headers,
2113 4 : tenant,
2114 4 : ®,
2115 4 : ctx_url,
2116 4 : None,
2117 4 : )
2118 4 : .await
2119 : }
2120 : _ => {
2121 : // queryBatch
2122 4 : let mut sel = Map::new();
2123 4 : if let Some(t) = reg.ent_types.first() {
2124 4 : sel.insert("type".into(), Value::String(ctx.compact_iri(t)));
2125 4 : }
2126 4 : sel.insert("id".into(), Value::String(id.to_owned()));
2127 4 : forward_in(
2128 4 : st,
2129 4 : reqwest::Method::POST,
2130 4 : format!("{}/ngsi-ld/v1/entityOperations/query", reg.endpoint),
2131 4 : &query,
2132 4 : headers,
2133 4 : tenant,
2134 4 : ®,
2135 4 : ctx_url,
2136 4 : Some(json!({"type": "Query", "entities": [Value::Object(sel)]})),
2137 4 : )
2138 4 : .await
2139 : }
2140 : };
2141 178 : (reg, status, body, peer_warns, actx)
2142 356 : })
2143 2006 : .await;
2144 2006 : let mut out = Vec::new();
2145 2006 : let single = fetched.len() == 1;
2146 2006 : for (reg, status, body, peer_warns, actx) in fetched {
2147 178 : if !usable_payload(st, tenant, status, &body, peer_warns, warnings) {
2148 88 : if single && reg.is_proxy() {
2149 12 : *sole_failure = Some(Part {
2150 12 : status,
2151 12 : detail: format!(
2152 12 : "distributed operation to registration {} returned {status}",
2153 12 : reg.reg_id
2154 12 : ),
2155 12 : registration: Some(reg.reg_id.clone()),
2156 12 : proxy: true,
2157 12 : update: None,
2158 12 : conflict: false,
2159 12 : });
2160 76 : }
2161 88 : continue;
2162 90 : }
2163 90 : let candidates: Vec<&Value> = match &body {
2164 4 : Value::Array(a) => a.iter().collect(),
2165 78 : Value::Object(_) => vec![&body],
2166 8 : _ => continue,
2167 : };
2168 82 : for c in candidates {
2169 78 : if c.get("id").and_then(Value::as_str) != Some(id) {
2170 0 : continue;
2171 78 : }
2172 78 : match import_entity(c, ®, actx.as_deref().unwrap_or(ctx)) {
2173 78 : Some(doc) => out.push((reg.mode == "auxiliary", doc)),
2174 : // received in time, but not a parseable NGSI-LD entity (111)
2175 0 : None => warnings.push(warning(
2176 : 111,
2177 0 : &alias_for(&st.host_alias, tenant),
2178 0 : "the payload of the response was invalid",
2179 : )),
2180 : }
2181 : }
2182 : }
2183 2006 : Ok(out)
2184 2008 : }
2185 :
2186 : /// Federated query: internal docs matching a type query, per registration.
2187 : /// The `CsrSpec` a Query Entities request matches registrations against.
2188 4387 : fn query_spec(ctx: &Context, params: &HashMap<String, String>) -> crate::registry::CsrSpec {
2189 : // 4.17: the parameter is ONE Entity Type Selection, so it travels whole —
2190 : // splitting it on commas turned a conjunction like (A;B) into two terms
2191 : // that match nothing, and no registration was consulted for it.
2192 4387 : let types: Option<Vec<String>> = params.get("type").cloned().map(|s| vec![s]);
2193 4387 : let ids: Option<Vec<String>> = params
2194 4387 : .get("id")
2195 4387 : .map(|s| s.split(',').map(str::to_owned).collect());
2196 : crate::registry::CsrSpec {
2197 4387 : types,
2198 4387 : ids,
2199 : // 5.12: "the id pattern (if present)" is part of the query-side
2200 : // Entity specification matched against EntityInfo elements
2201 4387 : id_pattern: params.get("idPattern").cloned(),
2202 : // 5.12 attribute conditions: the "list of Attribute names (if
2203 : // present)" gates which RegistrationInfos match
2204 4387 : attrs: params
2205 4387 : .get("attrs")
2206 4387 : .map(|s| s.split(',').map(|a| ctx.expand_key(a.trim())).collect()),
2207 : // 5.12 datasetId condition (should-level): disjoint sets don't match
2208 4387 : dataset_ids: params
2209 4387 : .get("datasetId")
2210 4387 : .map(|s| s.split(',').map(|d| d.trim().to_owned()).collect()),
2211 4387 : csf: params.get("csf").and_then(|c| antares_ql::parse_q(c).ok()),
2212 4387 : geo: antares_ql::geo::GeoQuery::from_params(params)
2213 4387 : .ok()
2214 4387 : .flatten(),
2215 4387 : ..Default::default()
2216 : }
2217 4387 : }
2218 :
2219 : /// Will this query actually fan out to a Context Source?
2220 : ///
2221 : /// 5.7.2.4 forbids ordering when "the execution of the operation is not
2222 : /// limited to the local scope", and 4.23.1 says "Sort ordering is never
2223 : /// applied to distributed operations". The subject is the *execution*, not the
2224 : /// presence of `local=true`: a query no registration matches executes locally
2225 : /// whether or not the client said so. Reading it as "local=true is mandatory
2226 : /// for orderBy" would fail ETSI's own 019_19, which orders without it.
2227 950 : pub async fn would_federate(
2228 950 : st: &AppState,
2229 950 : tenant: &TenantId,
2230 950 : ctx: &Context,
2231 950 : params: &HashMap<String, String>,
2232 950 : headers: &HeaderMap,
2233 950 : ) -> Result<bool, NgsiError> {
2234 950 : if !active(params) || via_hops(headers) > MAX_VIA_HOPS {
2235 64 : return Ok(false);
2236 886 : }
2237 : // the verdict only — no forwarding set is compiled for it
2238 886 : let spec = query_spec(ctx, params);
2239 886 : let seen = via_tokens(headers);
2240 886 : Ok(reg_docs(st, tenant, &spec, ctx)
2241 886 : .await?
2242 886 : .iter()
2243 886 : .any(|doc| reg_candidate(doc, &spec, ctx, &seen).is_some()))
2244 950 : }
2245 :
2246 : /// May a fan-out response contribute the entity `id` under registration
2247 : /// `reg`, for a request whose id selection is `spec`?
2248 : ///
2249 : /// A source only speaks for what its registration covers — an id outside
2250 : /// that scope is dropped rather than merged, or a peer could overwrite
2251 : /// unrelated local attributes on recency (4.5.5.3). But 5.12's matching is
2252 : /// deliberately over-broad for patterns: condition 5 forwards when "both a
2253 : /// specified id pattern and an idPattern in the Entity Info are present
2254 : /// (since in the general case it is not easily feasible to determine if
2255 : /// there can be identifiers matching both patterns)". The same
2256 : /// undecidability applies to the ANSWER: an id the CLIENT's own selection
2257 : /// admits cannot be refused on the registration's pattern. A peer therefore
2258 : /// contributes only ids inside its registration scope or inside what the
2259 : /// client itself asked to see — never a third thing.
2260 582 : fn admits_import(reg: &FedReg, spec: &crate::registry::CsrSpec, id: &str) -> bool {
2261 582 : reg.can_match_id(id)
2262 20 : || spec
2263 20 : .ids
2264 20 : .as_ref()
2265 20 : .is_some_and(|ids| ids.iter().any(|i| i == id))
2266 16 : || spec
2267 16 : .id_pattern
2268 16 : .as_deref()
2269 16 : .is_some_and(|p| antares_ql::regex::compile(p).is_ok_and(|re| re.find(id).is_some()))
2270 582 : }
2271 :
2272 : /// The scope gate of a fan-out answer: `c` is this source's data (4.3.6.1:
2273 : /// a type outside the registration's EntityInfo is not, whatever the client
2274 : /// asked for) and [`admits_import`] admits its id.
2275 564 : fn in_scope(reg: &FedReg, spec: &crate::registry::CsrSpec, ctx: &Context, c: &Value) -> bool {
2276 564 : c.as_object().is_some_and(|o| reg.covers_types(o, ctx))
2277 558 : && c.get("id")
2278 558 : .and_then(Value::as_str)
2279 558 : .is_some_and(|i| admits_import(reg, spec, i))
2280 564 : }
2281 :
2282 : /// Whether a query parameter travels on a forward. A narrowed `q` (the
2283 : /// caller's joined with a policy's, `antares_ql::parse_narrowed_q`) can be
2284 : /// past the caps a peer counts it with, `antares_ql::parse_q`'s; the peer's
2285 : /// TooComplexQuery would then tell the caller that a policy predicate is
2286 : /// there and how large (threat V23). Such a `q` stays home: the
2287 : /// local re-check applies it to the aggregate (5.7.2.4, 5.7.4.4), as it
2288 : /// does for split entities.
2289 : // The peer is modelled with our own caps; a peer with smaller ones
2290 : // can still refuse the conjunction. Forward only the caller's own q then.
2291 72 : fn forwardable(k: &str, v: &str) -> bool {
2292 72 : k != "q" || antares_ql::parse_q(v).is_ok()
2293 72 : }
2294 :
2295 2301 : pub async fn fed_query(
2296 2301 : st: &AppState,
2297 2301 : tenant: &TenantId,
2298 2301 : headers: &HeaderMap,
2299 2301 : ctx: &Context,
2300 2301 : params: &HashMap<String, String>,
2301 2301 : warnings: &mut Vec<String>,
2302 2301 : ) -> Result<Vec<(bool, Value)>, NgsiError> {
2303 2301 : let spec = query_spec(ctx, params);
2304 2301 : let ctx_url = ctx_link_url(headers, &ctx.source);
2305 2301 : let ctx_url = &ctx_url;
2306 2301 : let regs: Vec<FedReg> = matching_regs(st, tenant, &spec, ctx, headers)
2307 2301 : .await?
2308 2297 : .into_iter()
2309 2297 : .filter(|r| r.query_op().is_some())
2310 2297 : .collect();
2311 2297 : let fetched = fan_out(regs, move |reg| async move {
2312 348 : let Some(op) = reg.query_op() else {
2313 0 : return (reg, 0, Value::Null, Vec::new(), None);
2314 : };
2315 : // The forwarded selection is decided ONCE and then rendered either as
2316 : // query parameters (Query Entities, 5.7.2) or as a Query body
2317 : // (5.2.23) — the two must ask the peer the same question.
2318 : //
2319 : // 4.3.6.1: the forwarded id list carries only ids this registration
2320 : // can match — never the full client list.
2321 348 : let ids: Option<String> = match params.get("id") {
2322 104 : Some(ids) => {
2323 104 : let keep: Vec<&str> = ids
2324 104 : .split(',')
2325 198 : .filter(|i| reg.can_match_id(i.trim()))
2326 104 : .collect();
2327 104 : (!keep.is_empty()).then(|| keep.join(","))
2328 : }
2329 : // the registration is scoped to exact ids only — ask for those
2330 244 : None if !reg.ent_unrestricted
2331 26 : && reg.ent_patterns.is_empty()
2332 18 : && !reg.ent_ids.is_empty() =>
2333 : {
2334 18 : Some(reg.ent_ids.join(","))
2335 : }
2336 226 : None => None,
2337 : };
2338 348 : let attrs: Option<String> = match ®.attrs {
2339 38 : Some(scope) => Some(
2340 38 : scope
2341 38 : .iter()
2342 38 : .map(|a| ctx.compact_iri(a))
2343 38 : .collect::<Vec<String>>()
2344 38 : .join(","),
2345 : ),
2346 310 : None => params.get("attrs").cloned(),
2347 : };
2348 : // 5.7.2.4: with split entities the filters "shall be removed
2349 : // before forwarding" and re-applied on the aggregate (which the
2350 : // local re-check always does); otherwise the request is forwarded
2351 : // WITH its filters, so the peer returns its filtered subset
2352 : // instead of everything. A registered jsonldContext (4.3.6.6)
2353 : // recompacts only attrs/type/geoproperty — q/scopeQ terms cannot
2354 : // be recompacted, so push-down is skipped there rather than
2355 : // filtering at the remote against the wrong terms.
2356 348 : let push_filters = !split_entities(params) && !has_reg_context(®);
2357 2088 : let filter = |k: &str| {
2358 2088 : push_filters
2359 2088 : .then(|| params.get(k))
2360 2088 : .flatten()
2361 2088 : .filter(|v| forwardable(k, v))
2362 2088 : .map(String::to_owned)
2363 2088 : };
2364 348 : let ((status, body, peer_warns), actx) = if op == "queryBatch" {
2365 22 : let mut sel = Map::new();
2366 22 : if let Some(t) = params.get("type") {
2367 22 : sel.insert("type".into(), Value::String(t.clone()));
2368 22 : }
2369 14 : match &ids {
2370 : // 5.2.33: `id` is one URI or an array of them, and it takes
2371 : // precedence over idPattern — so a pattern only travels when
2372 : // no id list survived the narrowing.
2373 14 : Some(list) if list.contains(',') => {
2374 0 : let arr: Vec<Value> =
2375 0 : list.split(',').map(|i| Value::String(i.into())).collect();
2376 0 : sel.insert("id".into(), Value::Array(arr));
2377 : }
2378 14 : Some(one) => {
2379 14 : sel.insert("id".into(), Value::String(one.clone()));
2380 14 : }
2381 : None => {
2382 8 : if let Some(p) = params.get("idPattern") {
2383 2 : sel.insert("idPattern".into(), Value::String(p.clone()));
2384 6 : }
2385 : }
2386 : }
2387 22 : let mut q_body = Map::new();
2388 22 : q_body.insert("type".into(), Value::String("Query".into()));
2389 22 : if !sel.is_empty() {
2390 22 : q_body.insert("entities".into(), json!([Value::Object(sel)]));
2391 22 : }
2392 22 : if let Some(a) = &attrs {
2393 8 : let list: Vec<Value> = a.split(',').map(|n| Value::String(n.into())).collect();
2394 8 : q_body.insert("attrs".into(), Value::Array(list));
2395 14 : }
2396 22 : if let Some(q) = filter("q") {
2397 2 : q_body.insert("q".into(), Value::String(q));
2398 20 : }
2399 22 : if let Some(s) = filter("scopeQ") {
2400 2 : q_body.insert("scopeQ".into(), Value::String(s));
2401 20 : }
2402 22 : let mut geo = Map::new();
2403 88 : for k in ["georel", "geometry", "coordinates", "geoproperty"] {
2404 88 : if let Some(v) = filter(k) {
2405 : // 5.2.13 GeoQuery carries coordinates as the GeoJSON
2406 : // value, not as the query-string spelling of it.
2407 6 : let parsed = if k == "coordinates" {
2408 2 : serde_json::from_str(&v).unwrap_or(Value::String(v))
2409 : } else {
2410 4 : Value::String(v)
2411 : };
2412 6 : geo.insert(k.into(), parsed);
2413 82 : }
2414 : }
2415 22 : if !geo.is_empty() {
2416 2 : q_body.insert("geoQ".into(), Value::Object(geo));
2417 20 : }
2418 22 : forward_in(
2419 22 : st,
2420 22 : reqwest::Method::POST,
2421 22 : format!("{}/ngsi-ld/v1/entityOperations/query", reg.endpoint),
2422 22 : &[("options".into(), "sysAttrs".into())],
2423 22 : headers,
2424 22 : tenant,
2425 22 : ®,
2426 22 : ctx_url,
2427 22 : Some(Value::Object(q_body)),
2428 22 : )
2429 22 : .await
2430 : } else {
2431 326 : let mut query: Vec<(String, String)> = vec![("options".into(), "sysAttrs".into())];
2432 326 : if let Some(t) = params.get("type") {
2433 326 : query.push(("type".into(), t.clone()));
2434 326 : }
2435 : // Table 6.4.3.2-1, and 5.2.33 for the body rendering below:
2436 : // `id` takes precedence over `idPattern`, so a pattern travels
2437 : // only when no id list survived the narrowing. The two
2438 : // renderings ask the peer the same question.
2439 326 : match &ids {
2440 100 : Some(list) => query.push(("id".into(), list.clone())),
2441 : None => {
2442 226 : if let Some(p) = params.get("idPattern") {
2443 22 : query.push(("idPattern".into(), p.clone()));
2444 204 : }
2445 : }
2446 : }
2447 326 : if let Some(a) = &attrs {
2448 48 : query.push(("attrs".into(), a.clone()));
2449 278 : }
2450 1956 : for k in [
2451 326 : "q",
2452 326 : "georel",
2453 326 : "geometry",
2454 326 : "coordinates",
2455 326 : "geoproperty",
2456 326 : "scopeQ",
2457 326 : ] {
2458 1956 : if let Some(v) = filter(k) {
2459 32 : query.push((k.into(), v));
2460 1924 : }
2461 : }
2462 326 : forward_in(
2463 326 : st,
2464 326 : reqwest::Method::GET,
2465 326 : format!("{}/ngsi-ld/v1/entities", reg.endpoint),
2466 326 : &query,
2467 326 : headers,
2468 326 : tenant,
2469 326 : ®,
2470 326 : ctx_url,
2471 326 : None,
2472 326 : )
2473 326 : .await
2474 : };
2475 348 : (reg, status, body, peer_warns, actx)
2476 696 : })
2477 2297 : .await;
2478 2297 : let mut out = Vec::new();
2479 2297 : for (reg, status, body, peer_warns, actx) in fetched {
2480 348 : if !usable_payload(st, tenant, status, &body, peer_warns, warnings) {
2481 70 : continue;
2482 278 : }
2483 278 : if let Value::Array(a) = &body {
2484 : // Scope gate — see in_scope.
2485 544 : for c in a
2486 260 : .iter()
2487 550 : .filter(|c| in_scope(®, &spec, actx.as_deref().unwrap_or(ctx), c))
2488 : {
2489 542 : match import_entity(c, ®, actx.as_deref().unwrap_or(ctx)) {
2490 542 : Some(doc) => out.push((reg.mode == "auxiliary", doc)),
2491 0 : None => warnings.push(warning(
2492 : 111,
2493 0 : &alias_for(&st.host_alias, tenant),
2494 0 : "the payload of the response was invalid",
2495 : )),
2496 : }
2497 : }
2498 18 : }
2499 : }
2500 2297 : Ok(out)
2501 2301 : }
2502 :
2503 : /// 5.14.4.4 / 5.14.5.4: forward EntityMap creation to matching registrations
2504 : /// that support `op` (createEntityMapQueryEntity / …QueryTemporal, 4.20);
2505 : /// with split entities in play the value/geo/scope filters are removed
2506 : /// before forwarding. Returns (registration id, returned EntityMap) pairs —
2507 : /// the caller merges them into the local map's entityMap/linkedMaps.
2508 : #[allow(clippy::too_many_arguments)] // mirrors the wire: one param per forwarded request part
2509 224 : pub(crate) async fn fed_entity_maps(
2510 224 : st: &AppState,
2511 224 : tenant: &TenantId,
2512 224 : headers: &HeaderMap,
2513 224 : ctx: &Context,
2514 224 : params: &HashMap<String, String>,
2515 224 : split: bool,
2516 224 : op: &str,
2517 224 : path: &str,
2518 224 : ) -> Result<Vec<(String, Value)>, NgsiError> {
2519 224 : let spec = query_spec(ctx, params);
2520 224 : let ctx_url = ctx_link_url(headers, &ctx.source);
2521 224 : let ctx_url = &ctx_url;
2522 224 : let regs: Vec<FedReg> = matching_regs(st, tenant, &spec, ctx, headers)
2523 224 : .await?
2524 224 : .into_iter()
2525 224 : .filter(|reg| reg.supports(op))
2526 224 : .collect();
2527 : // a narrowed q the peer's caps refuse stays home, and the map the peer
2528 : // answers without it is narrowed here
2529 224 : let held_back = params
2530 224 : .get("q")
2531 224 : .filter(|q| !split && !forwardable("q", q))
2532 224 : .map(|q| antares_ql::parse_narrowed_q(q))
2533 224 : .transpose()?;
2534 224 : let held_back = &held_back;
2535 224 : let fetched = fan_out(regs, move |reg| async move {
2536 18 : let mut query: Vec<(String, String)> = Vec::new();
2537 126 : for k in [
2538 18 : "id",
2539 18 : "idPattern",
2540 18 : "type",
2541 18 : "timerel",
2542 18 : "timeAt",
2543 18 : "endTimeAt",
2544 18 : "timeproperty",
2545 18 : ] {
2546 126 : if let Some(v) = params.get(k) {
2547 18 : query.push((k.to_owned(), v.clone()));
2548 108 : }
2549 : }
2550 18 : query.extend(instance_cap(params));
2551 18 : if !split {
2552 144 : for k in [
2553 18 : "attrs",
2554 18 : "q",
2555 18 : "georel",
2556 18 : "geometry",
2557 18 : "coordinates",
2558 18 : "geoproperty",
2559 18 : "scopeQ",
2560 18 : "lang",
2561 18 : ] {
2562 144 : if let Some(v) = params.get(k).filter(|v| forwardable(k, v)) {
2563 2 : query.push((k.to_owned(), v.clone()));
2564 142 : }
2565 : }
2566 0 : }
2567 18 : let (status, mut body, _) = forward(
2568 18 : st,
2569 18 : reqwest::Method::GET,
2570 18 : format!("{}/ngsi-ld/v1/{path}", reg.endpoint),
2571 18 : &query,
2572 18 : headers,
2573 18 : tenant,
2574 18 : ®,
2575 18 : ctx_url,
2576 18 : None,
2577 : )
2578 18 : .await;
2579 18 : if let Some(q) = held_back.as_ref().filter(|_| (200..300).contains(&status)) {
2580 2 : let peer = Peer {
2581 2 : st,
2582 2 : tenant,
2583 2 : headers,
2584 2 : ctx_url,
2585 2 : reg: ®,
2586 2 : };
2587 2 : narrow_map_ids(peer, ctx, params.get("type"), q, &mut body).await;
2588 16 : }
2589 18 : (reg, status, body)
2590 36 : })
2591 224 : .await;
2592 224 : let mut out = Vec::new();
2593 224 : for (reg, status, body) in fetched {
2594 18 : if (200..300).contains(&status) && body.get("entityMap").is_some() {
2595 14 : out.push((reg.reg_id.clone(), body));
2596 14 : }
2597 : }
2598 224 : Ok(out)
2599 224 : }
2600 :
2601 : /// What every forward to one registered peer shares.
2602 : struct Peer<'a> {
2603 : st: &'a AppState,
2604 : tenant: &'a TenantId,
2605 : headers: &'a HeaderMap,
2606 : ctx_url: &'a str,
2607 : reg: &'a FedReg,
2608 : }
2609 :
2610 : /// A peer's EntityMap asked without a narrowed `q` its caps refuse (
2611 : /// threat V23): the ids it lists are a superset of the narrowing's, and the
2612 : /// map is served to the caller, so each id stays only when the peer's own
2613 : /// Entity is selected by `q` here. The Entities are read from the same peer
2614 : /// by id, as Query Entities forwards them; a linked path does not resolve,
2615 : /// so it never holds, and a peer that cannot be read contributes no id.
2616 2 : async fn narrow_map_ids(
2617 2 : peer: Peer<'_>,
2618 2 : ctx: &Context,
2619 2 : type_param: Option<&String>,
2620 2 : q: &antares_ql::QNode,
2621 2 : body: &mut Value,
2622 2 : ) {
2623 : let Peer {
2624 2 : st,
2625 2 : tenant,
2626 2 : headers,
2627 2 : ctx_url,
2628 2 : reg,
2629 2 : } = peer;
2630 : // One read per 100 ids, sequential; a peer map is capped at
2631 : // the broker's limit, so this is a few reads at most
2632 : const CHUNK: usize = 100;
2633 2 : let Some(map) = body.get_mut("entityMap").and_then(Value::as_object_mut) else {
2634 0 : return;
2635 : };
2636 2 : let listed: Vec<String> = map.keys().cloned().collect();
2637 2 : let mut kept = std::collections::HashSet::new();
2638 2 : if reg.query_op().is_some() {
2639 2 : for ids in listed.chunks(CHUNK) {
2640 2 : let mut query = vec![
2641 2 : ("id".to_owned(), ids.join(",")),
2642 2 : ("limit".to_owned(), ids.len().to_string()),
2643 : ];
2644 2 : if let Some(t) = type_param {
2645 2 : query.push(("type".to_owned(), t.clone()));
2646 2 : }
2647 2 : let ((status, found, _), actx) = forward_in(
2648 2 : st,
2649 2 : reqwest::Method::GET,
2650 2 : format!("{}/ngsi-ld/v1/entities", reg.endpoint),
2651 2 : &query,
2652 2 : headers,
2653 2 : tenant,
2654 2 : reg,
2655 2 : ctx_url,
2656 2 : None,
2657 : )
2658 2 : .await;
2659 2 : if !(200..300).contains(&status) {
2660 0 : continue;
2661 2 : }
2662 4 : for e in found.as_array().into_iter().flatten() {
2663 4 : let Some(id) = e.get("id").and_then(Value::as_str) else {
2664 0 : continue;
2665 : };
2666 6 : if ids.iter().any(|i| i == id)
2667 4 : && import_entity(e, reg, actx.as_deref().unwrap_or(ctx))
2668 4 : .is_some_and(|doc| antares_ql::eval::eval_q(q, &doc, ctx, &|_| None))
2669 2 : {
2670 2 : kept.insert(id.to_owned());
2671 2 : }
2672 : }
2673 : }
2674 0 : }
2675 4 : map.retain(|id, _| kept.contains(id));
2676 2 : }
2677 :
2678 : /// 5.7.4.4: forward the temporal query to matching registrations that
2679 : /// support the queryTemporal operation; registrations without it are not
2680 : /// contacted. Returns (auxiliary, expanded doc) pairs.
2681 956 : pub async fn fed_query_temporal(
2682 956 : st: &AppState,
2683 956 : tenant: &TenantId,
2684 956 : headers: &HeaderMap,
2685 956 : ctx: &Context,
2686 956 : params: &HashMap<String, String>,
2687 956 : warnings: &mut Vec<String>,
2688 956 : proxies: &mut Vec<FedReg>,
2689 956 : ) -> Result<Vec<(bool, Value)>, NgsiError> {
2690 956 : let mut spec = query_spec(ctx, params);
2691 956 : spec.temporal = crate::temporalq::TemporalQ::from_params(params, false)
2692 956 : .ok()
2693 956 : .flatten();
2694 956 : let ctx_url = ctx_link_url(headers, &ctx.source);
2695 956 : let ctx_url = &ctx_url;
2696 956 : let matched = matching_regs(st, tenant, &spec, ctx, headers).await?;
2697 956 : proxies.extend(matched.iter().filter(|r| r.is_proxy()).cloned());
2698 956 : let regs: Vec<FedReg> = matched
2699 956 : .into_iter()
2700 956 : .filter(|reg| reg.supports("queryTemporal"))
2701 956 : .collect();
2702 956 : let fetched = fan_out(regs, move |reg| async move {
2703 22 : let mut query = temporal_window(params);
2704 66 : for k in ["type", "id", "idPattern"] {
2705 66 : if let Some(v) = params.get(k) {
2706 22 : query.push((k.into(), v.clone()));
2707 44 : }
2708 : }
2709 : // 5.7.4.4 mirrors 5.7.2.4: with split entities the value/geo/scope
2710 : // filters are stripped from the forward and applied on the
2711 : // aggregate; a registered jsonldContext cannot recompact q/scopeQ
2712 : // terms, so push-down is skipped there too.
2713 22 : if !split_entities(params) && !has_reg_context(®) {
2714 120 : for k in [
2715 20 : "q",
2716 20 : "georel",
2717 20 : "geometry",
2718 20 : "coordinates",
2719 20 : "geoproperty",
2720 20 : "scopeQ",
2721 20 : ] {
2722 120 : if let Some(v) = params.get(k).filter(|v| forwardable(k, v)) {
2723 4 : query.push((k.into(), v.clone()));
2724 116 : }
2725 : }
2726 2 : }
2727 22 : if let Some(a) = scope_attrs(®, ctx) {
2728 4 : query.push(a);
2729 18 : } else if let Some(a) = params.get("attrs") {
2730 0 : query.push(("attrs".into(), a.clone()));
2731 18 : }
2732 22 : let ((status, body, peer_warns), actx) = forward_in(
2733 22 : st,
2734 22 : reqwest::Method::GET,
2735 22 : format!("{}/ngsi-ld/v1/temporal/entities", reg.endpoint),
2736 22 : &query,
2737 22 : headers,
2738 22 : tenant,
2739 22 : ®,
2740 22 : ctx_url,
2741 22 : None,
2742 : )
2743 22 : .await;
2744 22 : (reg, status, body, peer_warns, actx)
2745 44 : })
2746 956 : .await;
2747 956 : let mut out = Vec::new();
2748 956 : for (reg, status, body, peer_warns, actx) in fetched {
2749 22 : if !usable_payload(st, tenant, status, &body, peer_warns, warnings) {
2750 0 : continue;
2751 22 : }
2752 22 : if let Value::Array(a) = &body {
2753 : // Same scope gate as the non-temporal query fan-out.
2754 22 : for c in a
2755 22 : .iter()
2756 22 : .filter(|c| in_scope(®, &spec, actx.as_deref().unwrap_or(ctx), c))
2757 : {
2758 16 : match import_temporal(c, ®, actx.as_deref().unwrap_or(ctx)) {
2759 16 : Some(doc) => out.push((reg.mode == "auxiliary", doc)),
2760 0 : None => warnings.push(warning(
2761 : 111,
2762 0 : &alias_for(&st.host_alias, tenant),
2763 0 : "the payload of the response was invalid",
2764 : )),
2765 : }
2766 : }
2767 0 : }
2768 : }
2769 956 : Ok(out)
2770 956 : }
2771 :
2772 : /// Merge federated docs into a local candidate set (keyed by id). Local docs
2773 : /// win; non-aux remote attrs merge before aux ones.
2774 3011 : pub fn merge_candidates(local: Vec<Value>, fed: Vec<(bool, Value)>) -> Vec<Value> {
2775 3011 : let mut order: Vec<String> = Vec::new();
2776 3011 : let mut by_id: HashMap<String, Value> = HashMap::new();
2777 33560 : for doc in local {
2778 33560 : if let Some(id) = doc.get("id").and_then(Value::as_str) {
2779 33560 : order.push(id.to_owned());
2780 33560 : by_id.insert(id.to_owned(), doc);
2781 33560 : }
2782 : }
2783 6022 : for aux_pass in [false, true] {
2784 6022 : for (aux, doc) in &fed {
2785 1084 : if *aux != aux_pass {
2786 542 : continue;
2787 542 : }
2788 542 : let Some(id) = doc.get("id").and_then(Value::as_str) else {
2789 0 : continue;
2790 : };
2791 542 : match by_id.get_mut(id) {
2792 26 : Some(base) => merge_docs(base, doc, *aux),
2793 516 : None => {
2794 516 : order.push(id.to_owned());
2795 516 : by_id.insert(id.to_owned(), doc.clone());
2796 516 : }
2797 : }
2798 : }
2799 : }
2800 3011 : order
2801 3011 : .into_iter()
2802 34076 : .filter_map(|id| by_id.remove(&id))
2803 3011 : .collect()
2804 3011 : }
2805 :
2806 : // ---------- distributed writes ----------
2807 :
2808 : /// Outcome of one part of a distributed write.
2809 : #[derive(Clone, Debug, Default)]
2810 : pub struct Part {
2811 : /// The status the local write or the source answered.
2812 : pub status: u16,
2813 : pub detail: String,
2814 : /// 5.2.17 BatchEntityError registrationId: the Context Source
2815 : /// Registration a forwarded part went to; `None` for the local part.
2816 : pub registration: Option<String>,
2817 : /// The part went to an exclusive or redirect registration (6.3.17).
2818 : pub proxy: bool,
2819 : /// A source's 207 UpdateResult (5.2.18), Attribute names expanded.
2820 : pub update: Option<RemoteUpdate>,
2821 : /// An exclusive registration refused the operation ([`conflict_part`]): a 409 Conflict, not
2822 : /// AlreadyExists.
2823 : pub conflict: bool,
2824 : }
2825 :
2826 : /// 5.2.18 UpdateResult a source answered for a forwarded attribute write.
2827 : #[derive(Clone, Debug, Default, PartialEq)]
2828 : pub struct RemoteUpdate {
2829 : pub updated: Vec<String>,
2830 : /// (Attribute IRI, reason)
2831 : pub not_updated: Vec<(String, String)>,
2832 : }
2833 :
2834 : impl Part {
2835 : /// A part decided by this broker itself.
2836 498 : pub fn local(status: u16, detail: impl Into<String>) -> Self {
2837 498 : Part {
2838 498 : status,
2839 498 : detail: detail.into(),
2840 498 : ..Part::default()
2841 498 : }
2842 498 : }
2843 1606 : pub fn ok(&self) -> bool {
2844 : // 207 from a forwarded source is a partial failure, not a success
2845 1606 : (200..300).contains(&self.status) && self.status != 207
2846 1606 : }
2847 : /// The status this part answers alone. 6.3.17 p.278: for a proxied
2848 : /// (exclusive/redirect) source the error vocabulary is fixed: 508 loop,
2849 : /// 504 timeout, 404 not found, and "502 Bad Gateway - if the single
2850 : /// forwarded request fails for any other reason such as the Context
2851 : /// Broker itself having insufficient access rights". 409 keeps
2852 : /// AlreadyExists semantics (the peer speaks NGSI-LD) and a 207 stays the
2853 : /// partial verdict it is; a 401/403, a peer's 500/503 or a 400 on the
2854 : /// inter-broker request surfaces as 502. Inside a 207 each error keeps
2855 : /// the source's own status, the most informative one.
2856 146 : pub fn alone_status(&self) -> u16 {
2857 146 : if self.proxy
2858 42 : && !(200..300).contains(&self.status)
2859 42 : && !matches!(self.status, 207 | 404 | 409 | 504 | 508)
2860 : {
2861 12 : 502
2862 : } else {
2863 134 : self.status
2864 : }
2865 146 : }
2866 : /// The ProblemDetails type and title of a failed part. 5.6.1.4/5.6.2…:
2867 : /// an unsupported-operation part is the Conflict error type; an
2868 : /// entity-exists part stays AlreadyExists.
2869 356 : fn problem(&self, status: u16) -> (&'static str, &'static str) {
2870 74 : match status {
2871 58 : 409 if self.conflict => ("Conflict", "Conflict"),
2872 16 : 409 => ("AlreadyExists", "Conflict"),
2873 184 : 404 => ("ResourceNotFound", "Not Found"),
2874 98 : _ => ("InternalError", "Error"),
2875 : }
2876 356 : }
2877 : }
2878 :
2879 : /// One failed part answered alone, as the ProblemDetails of its status.
2880 136 : pub fn part_error(p: &Part, tenant: &TenantId) -> Response {
2881 136 : problem_response(p, tenant, false)
2882 136 : }
2883 :
2884 : /// The failure of the single registered source of a distributed read (6.3.17
2885 : /// p.278), naming the Context Source Registration it was asked through.
2886 10 : pub fn source_error(p: &Part, tenant: &TenantId) -> Response {
2887 10 : problem_response(p, tenant, true)
2888 10 : }
2889 :
2890 146 : fn problem_response(p: &Part, tenant: &TenantId, name_registration: bool) -> Response {
2891 146 : let code = p.alone_status();
2892 146 : let status = StatusCode::from_u16(code).unwrap_or(StatusCode::BAD_GATEWAY);
2893 146 : let (etype, title) = p.problem(code);
2894 146 : let mut body = json!({
2895 146 : "type": format!("https://uri.etsi.org/ngsi-ld/errors/{etype}"),
2896 146 : "title": title,
2897 146 : "detail": p.detail,
2898 146 : "status": code,
2899 : });
2900 146 : if let (true, Some(rid), Some(o)) = (name_registration, &p.registration, body.as_object_mut()) {
2901 10 : o.insert("registrationId".into(), Value::String(rid.clone()));
2902 136 : }
2903 146 : let mut resp = (
2904 146 : status,
2905 146 : [(axum::http::header::CONTENT_TYPE, "application/json")],
2906 146 : axum::Json(body),
2907 146 : )
2908 146 : .into_response();
2909 146 : echo_tenant(tenant, &mut resp);
2910 146 : resp
2911 146 : }
2912 :
2913 : /// Combine local + forwarded parts of a write on Entity `id` (`None` for a
2914 : /// set operation such as Purge) (6.3.17/6.4.3.1): all-success ⇒ `ok`, single failing part ⇒ its own
2915 : /// error, mixed ⇒ 207 Multi-Status with a BatchOperationResult.
2916 508 : pub fn combine(
2917 508 : profile: crate::Profile,
2918 508 : id: Option<&str>,
2919 508 : mut parts: Vec<Part>,
2920 508 : ok: Response,
2921 508 : tenant: &TenantId,
2922 508 : ) -> Response {
2923 : // 6.3.17 calls a source's 404 not abnormal only for HTTP GET; every
2924 : // part here is an unsafe write, whose error response "should be as
2925 : // informative as possible": a 404 beside a success stays in the 207
2926 : // (ETSI D002_02_01_inc, D002_02_02_inc). 404 everywhere is the one 404.
2927 : // ANTARES_FED_WRITE_SOURCE_404=ignore reads the 404 as "no data" for
2928 : // writes too (forge all-fixes, Coraine): beside a success it drops out.
2929 508 : if profile.fed_write_ignore_404 && parts.iter().any(Part::ok) {
2930 32 : parts.retain(|p| p.status != 404);
2931 494 : }
2932 508 : if parts.iter().all(Part::ok) {
2933 238 : return ok;
2934 270 : }
2935 270 : if parts.len() == 1 || parts.iter().all(|p| p.status == 404) {
2936 112 : let Some(p) = parts.first() else {
2937 0 : return ok;
2938 : };
2939 112 : return part_error(p, tenant);
2940 158 : }
2941 : // TS 104 176 7.2.3.2: the 207 body is a BatchOperationResult. 5.2.16
2942 : // success: "Array of Entity IDs corresponding to the Elements that were
2943 : // successfully treated"; 5.2.17 BatchEntityError: entityId "Entity ID
2944 : // corresponding to the Entity in error", registrationId "Registration Id
2945 : // corresponding to a failed distributed operation".
2946 158 : let errors: Vec<Value> = parts
2947 158 : .iter()
2948 316 : .filter(|p| !p.ok())
2949 210 : .map(|p| {
2950 210 : let (etype, title) = p.problem(p.status);
2951 210 : let mut e = json!({
2952 210 : "error": {
2953 210 : "status": p.status,
2954 210 : "type": format!("https://uri.etsi.org/ngsi-ld/errors/{etype}"),
2955 210 : "title": title,
2956 210 : "detail": p.detail,
2957 : }
2958 : });
2959 210 : if let Some(o) = e.as_object_mut() {
2960 210 : if let Some(id) = id {
2961 208 : o.insert("entityId".into(), Value::String(id.to_owned()));
2962 208 : }
2963 210 : if let Some(rid) = &p.registration {
2964 106 : o.insert("registrationId".into(), Value::String(rid.clone()));
2965 108 : }
2966 0 : }
2967 210 : e
2968 210 : })
2969 158 : .collect();
2970 158 : let success: Vec<&str> = id
2971 158 : .filter(|_| parts.iter().any(Part::ok))
2972 158 : .into_iter()
2973 158 : .collect();
2974 158 : let body = json!({"success": success, "errors": errors});
2975 158 : let mut resp = (
2976 158 : StatusCode::MULTI_STATUS,
2977 158 : [(axum::http::header::CONTENT_TYPE, "application/json")],
2978 158 : axum::Json(body),
2979 158 : )
2980 158 : .into_response();
2981 158 : echo_tenant(tenant, &mut resp);
2982 158 : resp
2983 508 : }
2984 :
2985 : /// 4.3.6.1: "It is the responsibility of the Context Broker to respect the
2986 : /// registration parameters when issuing distributed requests. […] Ultimately,
2987 : /// all constraints specified in the registration shall be respected."
2988 : /// Reduce a compacted entity/fragment to the members a registration covers
2989 : /// (plus id/type); returns None if no attribute member remains.
2990 456 : pub fn reduce_to_scope(obj: &Map<String, Value>, reg: &FedReg, ctx: &Context) -> Option<Value> {
2991 : // an item outside the registration's EntityInfo ids/types is not this
2992 : // source's data at all — nothing of it may be forwarded there
2993 456 : if !reg.covers_item(obj, ctx) {
2994 8 : return None;
2995 448 : }
2996 448 : let Some(_) = ®.attrs else {
2997 348 : let mut full = obj.clone();
2998 348 : full.remove("@context");
2999 348 : return Some(Value::Object(full));
3000 : };
3001 100 : let mut out = Map::new();
3002 100 : let mut any = false;
3003 424 : for (k, v) in obj {
3004 424 : if k == "@context" {
3005 68 : continue;
3006 356 : }
3007 356 : if ["id", "type", "scope"].contains(&k.as_str()) {
3008 182 : out.insert(k.clone(), v.clone());
3009 182 : continue;
3010 174 : }
3011 174 : if reg.covers_attr(&ctx.expand_key(k)) {
3012 98 : out.insert(k.clone(), v.clone());
3013 98 : any = true;
3014 98 : }
3015 : }
3016 100 : any.then_some(Value::Object(out))
3017 456 : }
3018 :
3019 84 : fn without_id(obj: &Map<String, Value>) -> Map<String, Value> {
3020 84 : let mut o = obj.clone();
3021 84 : o.remove("id");
3022 84 : o
3023 84 : }
3024 :
3025 : /// 508 Loop Detected (6.3.17): the inbound Via chain already names us and a
3026 : /// registration would forward the operation right back.
3027 70 : pub fn loop_508(tenant: &TenantId) -> Response {
3028 70 : let body = json!({
3029 70 : "type": "https://uri.etsi.org/ngsi-ld/errors/InternalError",
3030 70 : "title": "Loop Detected",
3031 70 : "detail": "the Via chain already contains this broker",
3032 70 : "status": 508,
3033 : });
3034 70 : let mut resp = (
3035 70 : StatusCode::LOOP_DETECTED,
3036 70 : [(axum::http::header::CONTENT_TYPE, "application/json")],
3037 70 : axum::Json(body),
3038 70 : )
3039 70 : .into_response();
3040 70 : echo_tenant(tenant, &mut resp);
3041 70 : resp
3042 70 : }
3043 :
3044 : /// What a distributed write does with its registrations once the 6.3.18
3045 : /// loop rule has spoken: forward to these, or answer with the chain's own
3046 : /// response and touch nothing locally.
3047 : pub enum WritePlan {
3048 : /// The non-auxiliary registrations that match the operation's entity
3049 : /// specification, `Via` chain already applied; empty means local only.
3050 : Forward(Vec<FedReg>),
3051 : /// The loop chain's answer (Table 6.3.18-2, or 508 past the hop cap):
3052 : /// the operation ends here, before anything local happens.
3053 : Answered(Box<Response>),
3054 : }
3055 :
3056 : /// The prologue every distributed write shares (4.3.6, 6.3.18): the
3057 : /// registrations of [`write_regs`] with [`handle_via_loop`] applied, so
3058 : /// no operation re-derives the pair and the two cannot disagree.
3059 35886 : pub async fn write_plan(
3060 35886 : st: &AppState,
3061 35886 : tenant: &TenantId,
3062 35886 : spec: &crate::registry::CsrSpec,
3063 35886 : ctx: &Context,
3064 35886 : params: &HashMap<String, String>,
3065 35886 : headers: &HeaderMap,
3066 35886 : ) -> Result<WritePlan, NgsiError> {
3067 35886 : let mut regs = write_regs(st, tenant, spec, ctx, params, headers).await?;
3068 : Ok(
3069 35884 : match handle_via_loop(
3070 35884 : headers,
3071 35884 : &alias_for(&st.host_alias, tenant),
3072 35884 : tenant,
3073 35884 : &mut regs,
3074 35884 : ) {
3075 46 : Some(answer) => WritePlan::Answered(Box::new(answer)),
3076 35838 : None => WritePlan::Forward(regs),
3077 : },
3078 : )
3079 35886 : }
3080 :
3081 : /// [`write_plan`] for the registrations of one Entity id.
3082 4996 : pub async fn write_plan_for_id(
3083 4996 : st: &AppState,
3084 4996 : tenant: &TenantId,
3085 4996 : id: &str,
3086 4996 : ctx: &Context,
3087 4996 : params: &HashMap<String, String>,
3088 4996 : headers: &HeaderMap,
3089 4996 : ) -> Result<WritePlan, NgsiError> {
3090 4996 : let spec = crate::registry::CsrSpec {
3091 4996 : ids: Some(vec![id.to_owned()]),
3092 4996 : ..Default::default()
3093 4996 : };
3094 4996 : write_plan(st, tenant, &spec, ctx, params, headers).await
3095 4996 : }
3096 :
3097 : /// Where a fragment write `op` goes: every registration that supports it,
3098 : /// with the share of `obj` its scope covers, and the Conflict part of each
3099 : /// proxy registration that does not (an inclusive one is not forwarded).
3100 168 : pub fn scoped_fragments(
3101 168 : regs: &[FedReg],
3102 168 : op: &str,
3103 168 : obj: &Map<String, Value>,
3104 168 : ctx: &Context,
3105 168 : ) -> (Vec<Part>, Vec<(FedReg, Value)>) {
3106 168 : let mut conflicts = Vec::new();
3107 168 : let mut fwd = Vec::new();
3108 174 : for reg in regs {
3109 174 : if !reg.supports(op) {
3110 92 : if reg.is_proxy() {
3111 16 : conflicts.push(conflict_part(op, reg));
3112 76 : }
3113 92 : continue;
3114 82 : }
3115 82 : if let Some(frag) = reduce_to_scope(obj, reg, ctx) {
3116 82 : fwd.push((reg.clone(), frag));
3117 82 : }
3118 : }
3119 168 : (conflicts, fwd)
3120 168 : }
3121 :
3122 : /// 4.3.6.2: "Auxiliary distributed operations are limited to context
3123 : /// information consumption operations (see clause 5.7)" — so a write op
3124 : /// only ever considers non-auxiliary matching registrations.
3125 35890 : pub async fn write_regs(
3126 35890 : st: &AppState,
3127 35890 : tenant: &TenantId,
3128 35890 : spec: &crate::registry::CsrSpec,
3129 35890 : ctx: &Context,
3130 35890 : params: &HashMap<String, String>,
3131 35890 : headers: &HeaderMap,
3132 35890 : ) -> Result<Vec<FedReg>, NgsiError> {
3133 35890 : if !active(params) {
3134 224 : return Ok(Vec::new());
3135 35666 : }
3136 35666 : Ok(matching_regs(st, tenant, spec, ctx, headers)
3137 35666 : .await?
3138 35664 : .into_iter()
3139 35664 : .filter(|r| r.mode != "auxiliary")
3140 35664 : .collect())
3141 35890 : }
3142 :
3143 : /// Execute one forwarded write and turn it into a Part.
3144 : #[allow(clippy::too_many_arguments)] // mirrors the wire: one param per forwarded request part
3145 466 : pub async fn forward_part(
3146 466 : st: &AppState,
3147 466 : method: reqwest::Method,
3148 466 : url: String,
3149 466 : query: &[(String, String)],
3150 466 : headers: &HeaderMap,
3151 466 : tenant: &TenantId,
3152 466 : reg: &FedReg,
3153 466 : ctx_url: &str,
3154 466 : body: Option<Value>,
3155 466 : ) -> Part {
3156 466 : let (status, answer, _) = forward(
3157 466 : st,
3158 466 : method,
3159 466 : url.clone(),
3160 466 : query,
3161 466 : headers,
3162 466 : tenant,
3163 466 : reg,
3164 466 : ctx_url,
3165 466 : body,
3166 : )
3167 466 : .await;
3168 : // 6.3.17: "the error response should be as informative as possible" —
3169 : // informative about the operation, not about the deployment. The part is
3170 : // named by its Context Source Registration id (5.2.9), which the client
3171 : // can already read from /csourceRegistrations; the registered endpoint
3172 : // is internal topology and stays in the log, so that a client able to
3173 : // provoke a partial failure cannot enumerate the peers.
3174 466 : tracing::debug!(
3175 : "distributed operation to {} returned {status}",
3176 0 : antares_notifier::redact_userinfo(&url)
3177 : );
3178 466 : let detail = format!(
3179 : "distributed operation to registration {} returned {status}",
3180 : reg.reg_id
3181 : );
3182 : Part {
3183 466 : status,
3184 466 : detail,
3185 466 : registration: Some(reg.reg_id.clone()),
3186 466 : proxy: reg.is_proxy(),
3187 466 : update: (status == 207).then(|| remote_update(&answer)).flatten(),
3188 : conflict: false,
3189 : }
3190 466 : }
3191 :
3192 : /// A source's 207 body read as a 5.2.18 UpdateResult, names as the source
3193 : /// wrote them; `None` when it is not one.
3194 20 : fn remote_update(body: &Value) -> Option<RemoteUpdate> {
3195 20 : let o = body.as_object()?;
3196 8 : let names = |v: Option<&Value>| -> Option<Vec<Value>> {
3197 8 : match v {
3198 0 : None => Some(Vec::new()),
3199 8 : Some(Value::Array(a)) => Some(a.clone()),
3200 0 : Some(_) => None,
3201 : }
3202 8 : };
3203 4 : let updated = names(o.get("updated"))?
3204 4 : .iter()
3205 6 : .map(|n| n.as_str().map(str::to_owned))
3206 4 : .collect::<Option<Vec<_>>>()?;
3207 4 : let not_updated = names(o.get("notUpdated"))?
3208 4 : .iter()
3209 4 : .map(|d| {
3210 2 : let name = d.get("attributeName")?.as_str()?.to_owned();
3211 2 : let reason = d.get("reason").and_then(Value::as_str).unwrap_or_default();
3212 2 : Some((name, reason.to_owned()))
3213 2 : })
3214 4 : .collect::<Option<Vec<_>>>()?;
3215 4 : Some(RemoteUpdate {
3216 4 : updated,
3217 4 : not_updated,
3218 4 : })
3219 20 : }
3220 :
3221 : /// Forward one attribute-level write to every matching registration. A
3222 : /// `fragment` body (Append, Update Attributes) is an Attribute map, reduced
3223 : /// per registration to the Attributes it covers: 5.6.3.4 "If an exclusive or
3224 : /// redirect Context Source Registration matches against the input data, the
3225 : /// Attributes from matching input data are forwarded", and 4.3.6.1 "all
3226 : /// constraints specified in the registration shall be respected".
3227 : #[allow(clippy::too_many_arguments)] // mirrors the wire: one param per forwarded request part
3228 166 : pub async fn fed_attr_parts(
3229 166 : st: &AppState,
3230 166 : headers: &HeaderMap,
3231 166 : tenant: &TenantId,
3232 166 : ctx: &Context,
3233 166 : regs: &[FedReg],
3234 166 : op: &str,
3235 166 : method: reqwest::Method,
3236 166 : path: &str,
3237 166 : query: &[(String, String)],
3238 166 : body: Option<Value>,
3239 166 : fragment: bool,
3240 166 : ) -> Vec<Part> {
3241 166 : let ctx_url = ctx_link_url(headers, &ctx.source);
3242 166 : let mut parts = Vec::new();
3243 196 : for reg in regs {
3244 : // 5.6.2.4 (and the sibling attribute operations): a proxy-mode
3245 : // registration not supporting the operation is an error of type
3246 : // Conflict and is never contacted; an inclusive one is simply not
3247 : // forwarded.
3248 196 : if !reg.supports(op) {
3249 8 : if reg.is_proxy() {
3250 6 : parts.push(conflict_part(op, reg));
3251 6 : } else {
3252 2 : parts.push(not_forwarded(format!("{op} not supported")));
3253 2 : }
3254 8 : continue;
3255 188 : }
3256 188 : let body = match (&body, fragment) {
3257 : // The path names the Entity, and `write_plan` matched the
3258 : // registrations on it; an `id` inside the fragment names
3259 : // nothing (the ETSI payloads carry "urn:ngsi-ld:Vehicle:randomUUID"),
3260 : // so it neither decides the scope nor goes to the source.
3261 84 : (Some(Value::Object(o)), true) => match reduce_to_scope(&without_id(o), reg, ctx) {
3262 84 : Some(b) => Some(b),
3263 : None => {
3264 0 : parts.push(not_forwarded("no Attribute of the registration".into()));
3265 0 : continue;
3266 : }
3267 : },
3268 104 : _ => body.clone(),
3269 : };
3270 188 : let mut part = forward_part(
3271 188 : st,
3272 188 : method.clone(),
3273 188 : format!("{}/ngsi-ld/v1{path}", reg.endpoint),
3274 188 : query,
3275 188 : headers,
3276 188 : tenant,
3277 188 : reg,
3278 188 : &ctx_url,
3279 188 : body,
3280 188 : )
3281 188 : .await;
3282 : // the source answered in the Context it was sent (ctx_url)
3283 188 : if let Some(u) = part.update.as_mut() {
3284 6 : for n in u.updated.iter_mut() {
3285 6 : *n = ctx.expand_key(n);
3286 6 : }
3287 4 : for (n, _) in u.not_updated.iter_mut() {
3288 2 : *n = ctx.expand_key(n);
3289 2 : }
3290 184 : }
3291 188 : parts.push(part);
3292 : }
3293 166 : parts
3294 166 : }
3295 :
3296 : /// status 0 = "not forwarded" sentinel: keeps the parts list 1:1 with regs
3297 : /// (combine_attr_parts zips them) without counting as success or failure.
3298 2 : fn not_forwarded(why: String) -> Part {
3299 2 : Part::local(0, format!("not forwarded: {why}"))
3300 2 : }
3301 :
3302 : /// Conflict part for an exclusive registration that does not accept the op.
3303 64 : pub fn conflict_part(op: &str, reg: &FedReg) -> Part {
3304 64 : Part {
3305 64 : status: 409,
3306 64 : detail: format!("exclusive registration does not accept {op}"),
3307 64 : registration: Some(reg.reg_id.clone()),
3308 64 : proxy: true,
3309 64 : update: None,
3310 64 : conflict: true,
3311 64 : }
3312 64 : }
3313 :
3314 : /// Remove proxy-covered attributes from an EXPANDED fragment so the local
3315 : /// write never stores exclusively/redirect-registered data (4.3.6.3).
3316 13590 : pub fn strip_covered_expanded(fragment: &Value, regs: &[FedReg]) -> Value {
3317 13590 : let mut f = fragment.clone();
3318 13590 : if let Some(o) = f.as_object_mut() {
3319 13590 : let covered: Vec<String> = o
3320 13590 : .keys()
3321 51700 : .filter(|k| {
3322 64 : !matches!(
3323 51700 : k.as_str(),
3324 51700 : "id" | "type" | "scope" | "createdAt" | "modifiedAt"
3325 51636 : ) && regs.iter().any(|r| r.is_proxy() && r.covers_attr(k))
3326 51700 : })
3327 13590 : .cloned()
3328 13590 : .collect();
3329 13590 : for k in covered {
3330 76 : o.remove(&k);
3331 76 : }
3332 0 : }
3333 13590 : f
3334 13590 : }
3335 :
3336 : /// 4.3.6.3 (TS 104 175 9.3.3): with an exclusive or redirect registration
3337 : /// the "Context Broker itself is not permitted to hold context data about the
3338 : /// registered Entities and Attributes locally (thus all registered context
3339 : /// data is obtained from the external registered sources)". Drop from a
3340 : /// local EXPANDED document the Attributes such a registration claims for it.
3341 3192 : pub fn drop_proxied_local(doc: &mut Value, proxies: &[FedReg], ctx: &Context) {
3342 3192 : let Some(o) = doc.as_object() else {
3343 0 : return;
3344 : };
3345 3192 : let claims: Vec<FedReg> = proxies
3346 3192 : .iter()
3347 3192 : .filter(|r| r.is_proxy() && r.covers_item(o, ctx))
3348 3192 : .cloned()
3349 3192 : .collect();
3350 3192 : if !claims.is_empty() {
3351 4 : *doc = strip_covered_expanded(doc, &claims);
3352 3188 : }
3353 3192 : }
3354 :
3355 : /// Strip the members proxied registrations cover from a compacted object;
3356 : /// returns (remainder, had_attrs_left).
3357 338 : pub fn strip_proxied(
3358 338 : obj: &Map<String, Value>,
3359 338 : proxies: &[&FedReg],
3360 338 : ctx: &Context,
3361 338 : ) -> (Map<String, Value>, bool) {
3362 338 : let mut out = Map::new();
3363 338 : let mut any_attr = false;
3364 1376 : for (k, v) in obj {
3365 1376 : if k == "@context" {
3366 180 : continue;
3367 1196 : }
3368 1196 : if ["id", "type", "scope"].contains(&k.as_str()) {
3369 688 : out.insert(k.clone(), v.clone());
3370 688 : continue;
3371 508 : }
3372 508 : let iri = ctx.expand_key(k);
3373 : // a proxy only owns the attribute if this ITEM is within its
3374 : // EntityInfo constraints (4.3.6.1) — a type-scoped registration
3375 : // must not strip attributes from an unrelated entity
3376 508 : if proxies
3377 508 : .iter()
3378 508 : .any(|r| r.covers_item(obj, ctx) && r.covers_attr(&iri))
3379 : {
3380 294 : continue;
3381 214 : }
3382 214 : out.insert(k.clone(), v.clone());
3383 214 : any_attr = true;
3384 : }
3385 338 : (out, any_attr)
3386 338 : }
3387 :
3388 : #[cfg(test)]
3389 : mod tests {
3390 : use super::*;
3391 : use serde_json::json;
3392 :
3393 92 : fn hdrs(via: Option<&str>) -> HeaderMap {
3394 92 : let mut h = HeaderMap::new();
3395 92 : if let Some(v) = via {
3396 60 : h.insert("via", v.parse().expect("via"));
3397 60 : }
3398 92 : h
3399 92 : }
3400 :
3401 : /// 6.3.17: on a write a 404 beside a success is reported in the 207
3402 : /// (ETSI D002_02_01_inc local 204 + source 404, D002_02_02_inc local 404
3403 : /// + source 204); 404 on every part is one 404.
3404 : #[test]
3405 4 : fn a_404_beside_a_success_is_a_multi_status_on_a_write() {
3406 4 : let t = antares_model::TenantId::new(TenantId::DEFAULT).expect("tenant");
3407 32 : let part = |status: u16| Part::local(status, String::new());
3408 16 : let ok = || StatusCode::NO_CONTENT.into_response();
3409 4 : assert_eq!(
3410 4 : combine(
3411 4 : crate::Profile::default(),
3412 4 : Some("urn:ngsi-ld:V:1"),
3413 4 : vec![part(204), part(404)],
3414 4 : ok(),
3415 4 : &t
3416 4 : )
3417 4 : .status(),
3418 : 207
3419 : );
3420 4 : assert_eq!(
3421 4 : combine(
3422 4 : crate::Profile::default(),
3423 4 : Some("urn:ngsi-ld:V:1"),
3424 4 : vec![part(404), part(204)],
3425 4 : ok(),
3426 4 : &t
3427 4 : )
3428 4 : .status(),
3429 : 207
3430 : );
3431 4 : assert_eq!(
3432 4 : combine(
3433 4 : crate::Profile::default(),
3434 4 : Some("urn:ngsi-ld:V:1"),
3435 4 : vec![part(404), part(404)],
3436 4 : ok(),
3437 4 : &t
3438 4 : )
3439 4 : .status(),
3440 : 404
3441 : );
3442 4 : assert_eq!(
3443 4 : combine(
3444 4 : crate::Profile::default(),
3445 4 : Some("urn:ngsi-ld:V:1"),
3446 4 : vec![part(204), part(500)],
3447 4 : ok(),
3448 4 : &t
3449 4 : )
3450 4 : .status(),
3451 : 207
3452 : );
3453 4 : }
3454 :
3455 : /// ANTARES_FED_WRITE_SOURCE_404=ignore: a 404 beside a success
3456 : /// drops out (forge all-fixes D002_02_01_inc/D002_02_02_inc, Coraine
3457 : /// distop_delete_inclusive_source_404 step 03); 404 everywhere stays
3458 : /// one 404 and any other failure stays in the 207.
3459 : #[test]
3460 4 : fn ignore_profile_reads_a_404_beside_a_success_as_no_data() {
3461 4 : let t = antares_model::TenantId::new(TenantId::DEFAULT).expect("tenant");
3462 36 : let part = |status: u16| Part::local(status, String::new());
3463 16 : let ok = || StatusCode::NO_CONTENT.into_response();
3464 4 : let p = crate::Profile {
3465 4 : fed_write_ignore_404: true,
3466 4 : ..crate::Profile::default()
3467 4 : };
3468 4 : assert_eq!(
3469 4 : combine(
3470 4 : p,
3471 4 : Some("urn:ngsi-ld:V:1"),
3472 4 : vec![part(204), part(404)],
3473 4 : ok(),
3474 4 : &t
3475 4 : )
3476 4 : .status(),
3477 : 204
3478 : );
3479 4 : assert_eq!(
3480 4 : combine(
3481 4 : p,
3482 4 : Some("urn:ngsi-ld:V:1"),
3483 4 : vec![part(404), part(204)],
3484 4 : ok(),
3485 4 : &t
3486 4 : )
3487 4 : .status(),
3488 : 204
3489 : );
3490 4 : assert_eq!(
3491 4 : combine(
3492 4 : p,
3493 4 : Some("urn:ngsi-ld:V:1"),
3494 4 : vec![part(404), part(404)],
3495 4 : ok(),
3496 4 : &t
3497 4 : )
3498 4 : .status(),
3499 : 404
3500 : );
3501 4 : assert_eq!(
3502 4 : combine(
3503 4 : p,
3504 4 : Some("urn:ngsi-ld:V:1"),
3505 4 : vec![part(204), part(404), part(500)],
3506 4 : ok(),
3507 4 : &t
3508 4 : )
3509 4 : .status(),
3510 : 207
3511 : );
3512 4 : }
3513 :
3514 : /// 4.3.6.6: contextSourceInfo carries the headers a source needs to
3515 : /// answer at all (an API key, say). The minimal registration view built
3516 : /// for the subscription operations dropped them, so every forwarded
3517 : /// subscription create, update and delete went out unauthenticated.
3518 : #[test]
3519 4 : fn the_minimal_registration_view_keeps_context_source_info() {
3520 4 : let doc = json!({
3521 4 : "id": "urn:ngsi-ld:ContextSourceRegistration:csi",
3522 4 : "endpoint": "http://peer:9090",
3523 4 : "contextSourceInfo": [
3524 4 : {"key": "X-API-Key", "value": "s3cret"},
3525 4 : {"key": "jsonldContext", "value": "https://example.org/ctx.jsonld"}
3526 : ]
3527 : });
3528 4 : let reg = fed_reg_of("urn:ngsi-ld:ContextSourceRegistration:csi", &doc);
3529 4 : assert_eq!(
3530 : reg.csi,
3531 4 : vec![
3532 4 : ("X-API-Key".to_owned(), "s3cret".to_owned()),
3533 4 : (
3534 4 : "jsonldContext".to_owned(),
3535 4 : "https://example.org/ctx.jsonld".to_owned()
3536 4 : ),
3537 : ]
3538 : );
3539 4 : }
3540 :
3541 : /// 4.17: `type` is one Entity Type Selection. Splitting it on commas
3542 : /// destroys a conjunction, and the shared evaluator (which csource tests
3543 : /// against real registrations) never sees the expression the client sent.
3544 : #[test]
3545 4 : fn the_type_selection_reaches_registration_matching_whole() {
3546 4 : let ctx = antares_jsonld::Loader::new().core();
3547 4 : let mut params = HashMap::new();
3548 4 : params.insert("type".to_owned(), "(Home;Vehicle),Building".to_owned());
3549 4 : let spec = query_spec(&ctx, ¶ms);
3550 4 : assert_eq!(
3551 : spec.types,
3552 4 : Some(vec!["(Home;Vehicle),Building".to_owned()]),
3553 : "the selection travels as one expression"
3554 : );
3555 4 : }
3556 :
3557 : /// 6.6/6.7: which part of a forwarded URL names the Attribute. The
3558 : /// segment ends at the next `/`, so the `value` sub-resource and a 6.7
3559 : /// instanceId are outside it, and a query string is never part of a path.
3560 : #[test]
3561 4 : fn the_path_attribute_segment_is_the_one_between_attrs_and_the_next_slash() {
3562 36 : let name = |u: &str| path_attr_segment(u).map(|(_, n)| n);
3563 4 : assert_eq!(
3564 4 : name("/entities/urn:x/attrs/speed"),
3565 4 : Some("speed".to_owned())
3566 : );
3567 4 : assert_eq!(
3568 4 : name("/entities/urn:x/attrs/speed/value"),
3569 4 : Some("speed".to_owned())
3570 : );
3571 4 : assert_eq!(
3572 4 : name("/temporal/entities/urn:x/attrs/speed/urn:ngsi-ld:instance:1"),
3573 4 : Some("speed".to_owned())
3574 : );
3575 4 : assert_eq!(
3576 4 : name("/entities/urn:x/attrs/speed?type=Vehicle"),
3577 4 : Some("speed".to_owned())
3578 : );
3579 : // percent-decoded, because that is the form an @context expands
3580 4 : assert_eq!(
3581 4 : name("/entities/urn:x/attrs/http%3A%2F%2Fa.example%2Fs"),
3582 4 : Some("http://a.example/s".to_owned())
3583 : );
3584 : // resources that name no Attribute
3585 4 : assert_eq!(name("/entities/urn:x"), None);
3586 4 : assert_eq!(name("/entities/urn:x/attrs"), None);
3587 4 : assert_eq!(name("/entities/urn:x/attrs/"), None);
3588 4 : assert_eq!(name("/entities/urn:x/attrs/?type=V"), None);
3589 : // the range addresses the ENCODED segment, so writing a translated
3590 : // name back over it cannot corrupt the rest of the URL
3591 4 : let url = "/entities/urn:x/attrs/speed/value?q=1";
3592 4 : let (range, _) = path_attr_segment(url).expect("segment");
3593 4 : assert_eq!(&url[range], "speed");
3594 4 : }
3595 :
3596 : /// 4.3.6.6 recompaction where the request @context and the registered one
3597 : /// are the same document is the identity: the Context Source receives the
3598 : /// Attribute Fragment the client sent, sub-Attributes and all. Any other
3599 : /// answer would rewrite a payload the clause only asked to re-spell.
3600 : #[test]
3601 4 : fn an_attribute_fragment_round_trips_through_one_context_unchanged() {
3602 4 : let ctx = antares_jsonld::core_context();
3603 4 : let frag = serde_json::json!({
3604 4 : "type": "Property",
3605 4 : "value": 56,
3606 4 : "source": {"type": "Property", "value": "Speedometer"},
3607 : });
3608 4 : let exp = antares_jsonld::expand_attr_fragment(frag.as_object().expect("obj"), &ctx)
3609 4 : .expect("the fragment expands");
3610 4 : assert_eq!(
3611 4 : antares_jsonld::compact::compact_instance(&exp, &ctx),
3612 : frag,
3613 : "the same @context in and out must change nothing"
3614 : );
3615 4 : }
3616 :
3617 : /// RFC 3986 clause 3.3: `#`, `?` and `/` end a path segment, so a client
3618 : /// id carrying one would re-target the peer's resource. The characters an
3619 : /// NGSI-LD id legitimately uses (`urn:`, `:`, `-`) must survive unchanged
3620 : /// or every forward would address a different entity than the client did.
3621 : #[test]
3622 4 : fn path_segment_encodes_what_would_end_the_segment() {
3623 4 : assert_eq!(
3624 4 : path_segment("urn:ngsi-ld:Vehicle:A4567-W"),
3625 : "urn:ngsi-ld:Vehicle:A4567-W"
3626 : );
3627 4 : assert_eq!(path_segment("urn:x#"), "urn:x%23");
3628 4 : assert_eq!(path_segment("urn:x?q=1"), "urn:x%3Fq=1");
3629 4 : assert_eq!(path_segment("a/b"), "a%2Fb");
3630 : // an id already carrying a percent must not decode twice at the peer
3631 4 : assert_eq!(path_segment("a%2Fb"), "a%252Fb");
3632 : // `.` is unreserved, so dot segments pass through unchanged here —
3633 : // they are refused at the door instead (EntityId::new, check_attr_name)
3634 4 : assert_eq!(path_segment(".."), "..");
3635 : // non-ASCII is percent-encoded per its UTF-8 bytes
3636 4 : assert_eq!(path_segment("é"), "%C3%A9");
3637 4 : }
3638 :
3639 : /// RFC 7230 received-by is a TOKEN compared for equality:
3640 : /// `ends_with` made alias `b1` match pseudonym `sub-b1` (spurious loop)
3641 : /// and could never catch the converse.
3642 : #[test]
3643 4 : fn via_loop_compares_tokens_not_suffixes() {
3644 4 : assert!(via_loop(&hdrs(Some("1.1 b1")), "b1"));
3645 4 : assert!(via_loop(&hdrs(Some("1.1 b2, 1.1 b1")), "b1"));
3646 4 : assert!(via_loop(&hdrs(Some("HTTP/1.1 b1")), "b1"));
3647 4 : assert!(
3648 4 : !via_loop(&hdrs(Some("1.1 sub-b1")), "b1"),
3649 : "suffix must not match — a former suffix-match false positive"
3650 : );
3651 4 : assert!(!via_loop(&hdrs(Some("1.1 b10")), "b1"));
3652 4 : assert!(!via_loop(&hdrs(None), "b1"));
3653 : // malformed element carrying only a pseudonym still detects
3654 4 : assert!(via_loop(&hdrs(Some("b1")), "b1"));
3655 4 : }
3656 :
3657 52 : fn reg(mode: &str) -> FedReg {
3658 52 : FedReg {
3659 52 : reg_id: "urn:ngsi-ld:ContextSourceRegistration:test".into(),
3660 52 : endpoint: "http://peer:9090".into(),
3661 52 : mode: mode.into(),
3662 52 : ops: vec!["federationOps".into()],
3663 52 : attrs: None,
3664 52 : ent_ids: vec![],
3665 52 : ent_types: vec![],
3666 52 : ent_patterns: vec![],
3667 52 : ent_unrestricted: false,
3668 52 : tenant: None,
3669 52 : alias: None,
3670 52 : csi: vec![],
3671 52 : local_only: false,
3672 52 : timeout_ms: None,
3673 52 : cooldown_ms: None,
3674 52 : }
3675 52 : }
3676 :
3677 : /// 5.12 condition 5, response side: a query idPattern was forwarded to a
3678 : /// pattern-scoped registration BECAUSE the two patterns cannot be
3679 : /// compared — so the answer cannot be refused on the registration's
3680 : /// pattern when the CLIENT's own selection admits the id. An id outside
3681 : /// BOTH scopes stays refused (a peer must not inject unrelated entities
3682 : /// that would win on recency).
3683 : #[test]
3684 4 : fn a_response_id_the_query_selects_survives_the_registration_pattern() {
3685 4 : let mut r = reg("inclusive");
3686 4 : r.ent_patterns = vec!["^urn:ngsi-ld:V:sk_bb:.*$".into()];
3687 4 : let spec = crate::registry::CsrSpec {
3688 4 : id_pattern: Some("^urn:ngsi-ld:V:sk_zvolen:.*$".into()),
3689 4 : ..Default::default()
3690 4 : };
3691 4 : assert!(
3692 4 : admits_import(&r, &spec, "urn:ngsi-ld:V:sk_zvolen:7"),
3693 : "the client's own pattern admits the id the peer answered with"
3694 : );
3695 4 : assert!(
3696 4 : admits_import(&r, &spec, "urn:ngsi-ld:V:sk_bb:1"),
3697 : "the registration's own scope still admits"
3698 : );
3699 4 : assert!(
3700 4 : !admits_import(&r, &spec, "urn:ngsi-ld:V:sk_presov:9"),
3701 : "an id outside both the registration and the query is refused"
3702 : );
3703 : // no query selection at all: only the registration scope admits
3704 4 : let none = crate::registry::CsrSpec::default();
3705 4 : assert!(!admits_import(&r, &none, "urn:ngsi-ld:V:sk_zvolen:7"));
3706 : // exact query ids admit exactly themselves
3707 4 : let exact = crate::registry::CsrSpec {
3708 4 : ids: Some(vec!["urn:ngsi-ld:V:sk_zvolen:7".into()]),
3709 4 : ..Default::default()
3710 4 : };
3711 4 : assert!(admits_import(&r, &exact, "urn:ngsi-ld:V:sk_zvolen:7"));
3712 4 : assert!(!admits_import(&r, &exact, "urn:ngsi-ld:V:sk_zvolen:8"));
3713 4 : }
3714 :
3715 : /// 4.3.6.1/5.12: an idPattern-scoped registration gates payload items
3716 : /// exactly like an exact-id one — a foreign-razidlo item is not this
3717 : /// source's data; an id-less fragment cannot be disproven.
3718 : #[test]
3719 4 : fn covers_item_honours_entityinfo_id_patterns() {
3720 4 : let mut r = reg("redirect");
3721 4 : r.ent_patterns = vec!["^urn:ngsi-ld:V:sk_bb:.*$".into()];
3722 4 : let st = AppState::new("me".into());
3723 4 : let ctx = st.loader.core();
3724 12 : let item = |id: Option<&str>| {
3725 12 : let mut m = Map::new();
3726 12 : if let Some(id) = id {
3727 8 : m.insert("id".into(), Value::String(id.into()));
3728 8 : }
3729 12 : m
3730 12 : };
3731 4 : assert!(r.covers_item(&item(Some("urn:ngsi-ld:V:sk_bb:1")), &ctx));
3732 4 : assert!(
3733 4 : !r.covers_item(&item(Some("urn:ngsi-ld:V:sk_po:1")), &ctx),
3734 : "a foreign-razidlo item must not be covered"
3735 : );
3736 4 : assert!(
3737 4 : r.covers_item(&item(None), &ctx),
3738 : "id-less fragments stay covered"
3739 : );
3740 4 : }
3741 :
3742 : /// Table 5.2.40-1: "In the multi-tenancy use case (see clause 4.14), this
3743 : /// id shall be identifying a specific Tenant within a registered Context
3744 : /// Source." One alias for every tenant of a broker makes cross-tenant
3745 : /// federation inside that broker look like a loop.
3746 : #[test]
3747 4 : fn alias_identifies_the_tenant_not_just_the_broker() {
3748 20 : let tenant = |s: &str| antares_model::TenantId::new(s).expect("tenant");
3749 : // the default tenant keeps the bare alias — 6.3.14's own convention
3750 : // (the header is omitted, not sent as "default"), and the wire format
3751 : // every single-tenant peer already registered
3752 4 : assert_eq!(alias_for("antares1", &tenant("default")), "antares1");
3753 4 : assert_eq!(alias_for("antares1", &tenant("zvolen")), "antares1~zvolen");
3754 : // a chain naming this broker in ANOTHER tenant is not a loop: the
3755 : // registration points at a different (source, tenant) pair
3756 4 : let h = hdrs(Some("1.1 antares1~zvolen"));
3757 4 : assert!(via_loop(&h, &alias_for("antares1", &tenant("zvolen"))));
3758 4 : assert!(!via_loop(
3759 4 : &h,
3760 4 : &alias_for("antares1", &tenant("banskabystrica"))
3761 4 : ));
3762 4 : assert!(
3763 4 : !via_loop(&h, &alias_for("antares1", &tenant("default"))),
3764 : "the default tenant of this broker is its own Context Source"
3765 : );
3766 : // `~` cannot occur in a TenantId and is rejected in a configured
3767 : // alias, so the two halves can never blur into each other
3768 4 : assert!(antares_model::TenantId::new("a~b").is_err());
3769 4 : }
3770 :
3771 : /// Table 6.3.18-2: "the listing of previously encountered Context Sources
3772 : /// supplied is used when determining matching registrations", and 5.2.9
3773 : /// gives a registration the peer's `contextSourceAlias` "which is used to
3774 : /// identify loops". A source already in the chain is therefore not a
3775 : /// match — and the tenant-specific alias keeps that per (source, tenant).
3776 : #[tokio::test]
3777 4 : async fn registered_alias_in_the_via_chain_is_not_a_matching_registration() {
3778 4 : let st = AppState::new("me".into());
3779 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
3780 4 : let ctx = st.loader.core();
3781 12 : for (id, alias) in [
3782 4 : ("urn:ngsi-ld:ContextSourceRegistration:visited", "peer1"),
3783 4 : ("urn:ngsi-ld:ContextSourceRegistration:fresh", "peer2"),
3784 4 : ("urn:ngsi-ld:ContextSourceRegistration:anon", ""),
3785 4 : ] {
3786 12 : let mut doc = json!({
3787 12 : "id": id,
3788 12 : "type": "ContextSourceRegistration",
3789 12 : "endpoint": "http://peer:9090",
3790 12 : "information": [{"entities": [{"type": "https://uri.etsi.org/ngsi-ld/default-context/Vehicle"}]}],
3791 : });
3792 12 : if !alias.is_empty() {
3793 8 : doc["contextSourceAlias"] = json!(alias);
3794 8 : }
3795 12 : st.store
3796 12 : .create(&tenant, Kind::Registration, id, doc)
3797 12 : .await
3798 12 : .expect("seed registration");
3799 : }
3800 4 : let spec = crate::registry::CsrSpec {
3801 4 : types: Some(vec![
3802 4 : "https://uri.etsi.org/ngsi-ld/default-context/Vehicle".into()
3803 4 : ]),
3804 4 : ..Default::default()
3805 4 : };
3806 4 : let all = matching_regs(&st, &tenant, &spec, &ctx, &hdrs(None))
3807 4 : .await
3808 4 : .expect("registrations");
3809 4 : assert_eq!(all.len(), 3, "no Via ⇒ every registration matches");
3810 4 : let via = hdrs(Some("1.1 peer1"));
3811 4 : let left = matching_regs(&st, &tenant, &spec, &ctx, &via)
3812 4 : .await
3813 4 : .expect("registrations");
3814 8 : let ids: Vec<&str> = left.iter().map(|r| r.reg_id.as_str()).collect();
3815 4 : assert!(
3816 4 : !ids.contains(&"urn:ngsi-ld:ContextSourceRegistration:visited"),
3817 : "a source already in the Via chain must not match"
3818 : );
3819 4 : assert_eq!(
3820 4 : ids.len(),
3821 4 : 2,
3822 4 : "an unvisited peer and a registration with no alias still match"
3823 4 : );
3824 4 : }
3825 :
3826 : /// the registration mirror (hash order) and the store yield the
3827 : /// matching registrations in one order, by id, so a 207 names the same
3828 : /// registration whichever of the two answered.
3829 : #[tokio::test]
3830 4 : async fn matching_registrations_come_in_id_order_from_mirror_and_store() {
3831 4 : let mut st = AppState::new("me".into());
3832 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
3833 4 : let ctx = st.loader.core();
3834 4 : let mirror = crate::mirror::DocMirror::default();
3835 48 : for n in (0..12).rev() {
3836 48 : let id = format!("urn:ngsi-ld:ContextSourceRegistration:r{n:02}");
3837 48 : let doc = json!({
3838 48 : "id": id,
3839 48 : "type": "ContextSourceRegistration",
3840 48 : "endpoint": format!("http://peer{n}:9090"),
3841 48 : "information": [{"entities": [{"type": "https://uri.etsi.org/ngsi-ld/default-context/Vehicle"}]}],
3842 : });
3843 48 : mirror.apply("default", &id, Some(doc.clone()));
3844 48 : st.store
3845 48 : .create(&tenant, Kind::Registration, &id, doc)
3846 48 : .await
3847 48 : .expect("seed registration");
3848 : }
3849 4 : let spec = crate::registry::CsrSpec::default();
3850 4 : let want: Vec<String> = (0..12)
3851 48 : .map(|n| format!("urn:ngsi-ld:ContextSourceRegistration:r{n:02}"))
3852 4 : .collect();
3853 4 : let from_store: Vec<String> = matching_regs(&st, &tenant, &spec, &ctx, &hdrs(None))
3854 4 : .await
3855 4 : .expect("store")
3856 4 : .into_iter()
3857 4 : .map(|r| r.reg_id)
3858 4 : .collect();
3859 4 : assert_eq!(from_store, want, "the store answers by id");
3860 4 : st.reg_mirror = Some(std::sync::Arc::new(mirror));
3861 4 : let from_mirror: Vec<String> = matching_regs(&st, &tenant, &spec, &ctx, &hdrs(None))
3862 4 : .await
3863 4 : .expect("mirror")
3864 4 : .into_iter()
3865 4 : .map(|r| r.reg_id)
3866 4 : .collect();
3867 4 : assert_eq!(from_mirror, want, "the mirror answers in the store's order");
3868 4 : }
3869 :
3870 : /// TS 104 175 9.7: a write whose Via names this broker is a
3871 : /// detected loop whatever its registrations; nothing is forwarded.
3872 : #[test]
3873 4 : fn a_write_looping_back_is_508_whatever_its_registrations() {
3874 4 : let t = antares_model::TenantId::new("default").expect("tenant");
3875 4 : let h = hdrs(Some("1.1 me"));
3876 20 : for set in [
3877 4 : vec![reg("exclusive")],
3878 4 : vec![reg("redirect")],
3879 4 : vec![reg("inclusive")],
3880 4 : vec![reg("exclusive"), reg("inclusive")],
3881 4 : Vec::new(),
3882 4 : ] {
3883 20 : let mut regs = set;
3884 20 : let resp = handle_via_loop(&h, "me", &t, &mut regs).expect("a loop");
3885 20 : assert_eq!(resp.status(), StatusCode::LOOP_DETECTED);
3886 20 : assert!(regs.is_empty(), "looping forwards must be dropped");
3887 : }
3888 : // no loop → untouched
3889 4 : let mut regs = vec![reg("exclusive")];
3890 4 : assert!(handle_via_loop(&hdrs(None), "me", &t, &mut regs).is_none());
3891 4 : assert_eq!(regs.len(), 1);
3892 4 : }
3893 :
3894 : /// 4.3.6.4 / 5.2.9 localOnly: "distributed operations associated to this
3895 : /// Context Source Registration will act only on data held directly by
3896 : /// the registered Context Source itself" — the flag must survive
3897 : /// registration compilation so every forward can carry local=true.
3898 : #[tokio::test]
3899 4 : async fn local_only_survives_registration_compilation() {
3900 4 : let st = AppState::new("me".into());
3901 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
3902 4 : let ctx = st.loader.core();
3903 8 : for (id, local_only) in [
3904 4 : ("urn:ngsi-ld:ContextSourceRegistration:lo", true),
3905 4 : ("urn:ngsi-ld:ContextSourceRegistration:casc", false),
3906 4 : ] {
3907 8 : let mut doc = json!({
3908 8 : "id": id,
3909 8 : "type": "ContextSourceRegistration",
3910 8 : "endpoint": "http://peer:9090",
3911 8 : "information": [{"entities": [{"type": "https://uri.etsi.org/ngsi-ld/default-context/Vehicle"}]}],
3912 : });
3913 8 : if local_only {
3914 4 : doc["localOnly"] = json!(true);
3915 4 : }
3916 8 : st.store
3917 8 : .create(&tenant, Kind::Registration, id, doc)
3918 8 : .await
3919 8 : .expect("seed registration");
3920 : }
3921 4 : let spec = crate::registry::CsrSpec {
3922 4 : types: Some(vec![
3923 4 : "https://uri.etsi.org/ngsi-ld/default-context/Vehicle".into()
3924 4 : ]),
3925 4 : ..Default::default()
3926 4 : };
3927 4 : let regs = matching_regs(&st, &tenant, &spec, &ctx, &HeaderMap::new())
3928 4 : .await
3929 4 : .expect("registrations");
3930 8 : let lo = |id: &str| {
3931 8 : regs.iter()
3932 12 : .find(|r| r.reg_id == id)
3933 8 : .expect("compiled")
3934 : .local_only
3935 8 : };
3936 4 : assert!(lo("urn:ngsi-ld:ContextSourceRegistration:lo"));
3937 4 : assert!(!lo("urn:ngsi-ld:ContextSourceRegistration:casc"));
3938 4 : }
3939 :
3940 : /// Table 5.2.34-1: management.localOnly — "distributed operations
3941 : /// associated to this Context Source Registration will act only on data
3942 : /// held directly by the registered Context Source itself".
3943 : #[tokio::test]
3944 4 : async fn management_local_only_survives_registration_compilation() {
3945 4 : let st = AppState::new("me".into());
3946 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
3947 4 : let ctx = st.loader.core();
3948 4 : let id = "urn:ngsi-ld:ContextSourceRegistration:mgmt-lo";
3949 4 : let doc = json!({
3950 4 : "id": id,
3951 4 : "type": "ContextSourceRegistration",
3952 4 : "endpoint": "http://peer:9090",
3953 4 : "information": [{"entities": [{"type": "https://uri.etsi.org/ngsi-ld/default-context/Vehicle"}]}],
3954 4 : "management": {"localOnly": true}
3955 : });
3956 4 : st.store
3957 4 : .create(&tenant, Kind::Registration, id, doc)
3958 4 : .await
3959 4 : .expect("seed registration");
3960 4 : let spec = crate::registry::CsrSpec {
3961 4 : types: Some(vec![
3962 4 : "https://uri.etsi.org/ngsi-ld/default-context/Vehicle".into()
3963 4 : ]),
3964 4 : ..Default::default()
3965 4 : };
3966 4 : let regs = matching_regs(&st, &tenant, &spec, &ctx, &HeaderMap::new())
3967 4 : .await
3968 4 : .expect("registrations");
3969 4 : assert!(
3970 4 : regs.iter()
3971 4 : .find(|r| r.reg_id == id)
3972 4 : .expect("compiled")
3973 4 : .local_only,
3974 4 : "management.localOnly must compile into the forward flag"
3975 4 : );
3976 4 : }
3977 :
3978 : /// 4.3.6.2: "An auxiliary Context Source Registration never overrides
3979 : /// data held directly within a Context Broker" — supplementary attributes
3980 : /// are included, conflicting ones lose to the base regardless of recency.
3981 : #[test]
3982 4 : fn auxiliary_merge_supplements_but_never_overrides() {
3983 4 : let attr = "https://uri.etsi.org/ngsi-ld/default-context/speed";
3984 4 : let extra = "https://uri.etsi.org/ngsi-ld/default-context/color";
3985 4 : let mut base = json!({
3986 4 : "id": "urn:x", "type": ["T"],
3987 4 : attr: [{"type": "Property", "value": 1, "modifiedAt": "2020-01-01T00:00:00Z"}]
3988 : });
3989 : // aux instance is FRESHER and would win a 4.5.5.3 recency merge —
3990 : // auxiliary mode must still lose the conflict, yet supplement `color`
3991 4 : let add = json!({
3992 4 : "id": "urn:x", "type": ["T"],
3993 4 : attr: [{"type": "Property", "value": 2, "modifiedAt": "2026-01-01T00:00:00Z"}],
3994 4 : extra: [{"type": "Property", "value": "red"}]
3995 : });
3996 4 : merge_docs(&mut base, &add, true);
3997 4 : assert_eq!(base[attr][0]["value"], 1, "aux must not override local");
3998 4 : assert_eq!(base[extra][0]["value"], "red", "aux supplement is included");
3999 : // the same add as a non-aux inclusive source DOES win on recency
4000 4 : let mut base2 = json!({
4001 4 : "id": "urn:x", "type": ["T"],
4002 4 : attr: [{"type": "Property", "value": 1, "modifiedAt": "2020-01-01T00:00:00Z"}]
4003 : });
4004 4 : merge_docs(&mut base2, &add, false);
4005 4 : assert_eq!(base2[attr][0]["value"], 2);
4006 4 : }
4007 :
4008 : /// 4.20: retrieveEntity implements only 5.7.1 — a source offering it
4009 : /// alone is never a query target; queryEntity/queryBatch are.
4010 : #[test]
4011 4 : fn query_op_requires_query_support() {
4012 4 : let reg =
4013 20 : |ops: &[&str]| fed_reg_of("urn:r", &json!({"endpoint": "http://x", "operations": ops}));
4014 4 : assert_eq!(reg(&["retrieveEntity"]).query_op(), None);
4015 4 : assert_eq!(reg(&["queryEntity"]).query_op(), Some("queryEntity"));
4016 4 : assert_eq!(reg(&["queryBatch"]).query_op(), Some("queryBatch"));
4017 4 : assert_eq!(reg(&["federationOps"]).query_op(), Some("queryEntity"));
4018 4 : assert_eq!(reg(&["retrieveOps"]).query_op(), Some("queryEntity"));
4019 4 : }
4020 :
4021 12 : async fn seed_reg(
4022 12 : st: &AppState,
4023 12 : tenant: &antares_model::TenantId,
4024 12 : id: &str,
4025 12 : mode: &str,
4026 12 : entities: Value,
4027 12 : ) {
4028 12 : let doc = json!({
4029 12 : "id": id,
4030 12 : "type": "ContextSourceRegistration",
4031 12 : "mode": mode,
4032 12 : "operations": ["redirectionOps"],
4033 12 : "endpoint": "http://peer:9090",
4034 12 : "information": [{"entities": entities}],
4035 : });
4036 12 : st.store
4037 12 : .create(tenant, Kind::Registration, id, doc)
4038 12 : .await
4039 12 : .expect("seed registration");
4040 12 : }
4041 :
4042 : /// 5.2.9 / 5.12: an `idPattern` in the registration's EntityInfo matches
4043 : /// a write addressed by id, so the write forwards to that source; a
4044 : /// pattern the id does not match keeps the write local.
4045 : #[tokio::test]
4046 4 : async fn write_plan_forwards_a_write_by_id_to_an_id_pattern_registration() {
4047 4 : let st = AppState::new("me".into());
4048 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
4049 4 : let ctx = st.loader.core();
4050 4 : seed_reg(
4051 4 : &st,
4052 4 : &tenant,
4053 4 : "urn:ngsi-ld:ContextSourceRegistration:cars",
4054 4 : "exclusive",
4055 4 : json!([{"idPattern": "urn:ngsi-ld:Vehicle:.*"}]),
4056 4 : )
4057 4 : .await;
4058 4 : seed_reg(
4059 4 : &st,
4060 4 : &tenant,
4061 4 : "urn:ngsi-ld:ContextSourceRegistration:bikes",
4062 4 : "exclusive",
4063 4 : json!([{"idPattern": "urn:ngsi-ld:Bike:.*"}]),
4064 4 : )
4065 4 : .await;
4066 4 : let spec = crate::registry::CsrSpec {
4067 4 : ids: Some(vec!["urn:ngsi-ld:Vehicle:1".into()]),
4068 4 : ..Default::default()
4069 4 : };
4070 4 : let plan = write_plan(
4071 4 : &st,
4072 4 : &tenant,
4073 4 : &spec,
4074 4 : &ctx,
4075 4 : &HashMap::new(),
4076 4 : &HeaderMap::new(),
4077 4 : )
4078 4 : .await
4079 4 : .expect("plan");
4080 4 : let WritePlan::Forward(regs) = plan else {
4081 0 : panic!("no Via chain, so nothing answers early");
4082 : };
4083 4 : let ids: Vec<&str> = regs.iter().map(|r| r.reg_id.as_str()).collect();
4084 4 : assert_eq!(ids, vec!["urn:ngsi-ld:ContextSourceRegistration:cars"]);
4085 4 : }
4086 :
4087 : /// 6.3.18: a `Via` chain that already names this broker is answered by
4088 : /// the chain rule before any registration is used — with a single
4089 : /// exclusive source that is the 508 of Table 6.3.18-2 — and a chain
4090 : /// past the hop cap is 508 regardless of what matched.
4091 : #[tokio::test]
4092 4 : async fn write_plan_answers_a_via_loop_before_forwarding() {
4093 4 : let st = AppState::new("me".into());
4094 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
4095 4 : let ctx = st.loader.core();
4096 4 : seed_reg(
4097 4 : &st,
4098 4 : &tenant,
4099 4 : "urn:ngsi-ld:ContextSourceRegistration:cars",
4100 4 : "exclusive",
4101 4 : json!([{"idPattern": "urn:ngsi-ld:Vehicle:.*"}]),
4102 4 : )
4103 4 : .await;
4104 4 : let spec = crate::registry::CsrSpec {
4105 4 : ids: Some(vec!["urn:ngsi-ld:Vehicle:1".into()]),
4106 4 : ..Default::default()
4107 4 : };
4108 4 : let looped = write_plan(
4109 4 : &st,
4110 4 : &tenant,
4111 4 : &spec,
4112 4 : &ctx,
4113 4 : &HashMap::new(),
4114 4 : &hdrs(Some("1.1 me")),
4115 4 : )
4116 4 : .await
4117 4 : .expect("plan");
4118 4 : let WritePlan::Answered(resp) = looped else {
4119 0 : panic!("a Via chain naming this broker with one exclusive source is answered, not forwarded");
4120 : };
4121 4 : assert_eq!(resp.status(), 508);
4122 4 : let deep: String = (0..=MAX_VIA_HOPS)
4123 132 : .map(|i| format!("1.1 hop{i}"))
4124 4 : .collect::<Vec<_>>()
4125 4 : .join(", ");
4126 4 : let WritePlan::Answered(resp) = write_plan(
4127 4 : &st,
4128 4 : &tenant,
4129 4 : &spec,
4130 4 : &ctx,
4131 4 : &HashMap::new(),
4132 4 : &hdrs(Some(&deep)),
4133 4 : )
4134 4 : .await
4135 4 : .expect("plan") else {
4136 0 : panic!("a chain past the hop cap is refused outright");
4137 : };
4138 4 : assert_eq!(resp.status(), 508);
4139 : // no chain, no early answer: the same registration forwards
4140 4 : assert!(matches!(
4141 4 : write_plan(&st, &tenant, &spec, &ctx, &HashMap::new(), &HeaderMap::new()).await.expect("plan"),
4142 4 : WritePlan::Forward(regs) if regs.len() == 1
4143 4 : ));
4144 4 : }
4145 :
4146 : /// 4.3.6.2: "Auxiliary distributed operations are limited to context
4147 : /// information consumption operations" — write_regs must drop a matching
4148 : /// auxiliary registration while keeping an inclusive one.
4149 : #[tokio::test]
4150 4 : async fn write_regs_exclude_auxiliary_registrations() {
4151 4 : let st = AppState::new("me".into());
4152 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
4153 4 : let ctx = st.loader.core();
4154 8 : for (id, mode) in [
4155 4 : ("urn:ngsi-ld:ContextSourceRegistration:aux", "auxiliary"),
4156 4 : ("urn:ngsi-ld:ContextSourceRegistration:inc", "inclusive"),
4157 4 : ] {
4158 8 : let doc = json!({
4159 8 : "id": id,
4160 8 : "type": "ContextSourceRegistration",
4161 8 : "mode": mode,
4162 8 : "operations": ["redirectionOps"],
4163 8 : "endpoint": "http://peer:9090",
4164 8 : "information": [{"entities": [{"type": "https://uri.etsi.org/ngsi-ld/default-context/Vehicle"}]}],
4165 : });
4166 8 : st.store
4167 8 : .create(&tenant, Kind::Registration, id, doc)
4168 8 : .await
4169 8 : .expect("seed registration");
4170 : }
4171 4 : let spec = crate::registry::CsrSpec {
4172 4 : types: Some(vec![
4173 4 : "https://uri.etsi.org/ngsi-ld/default-context/Vehicle".into()
4174 4 : ]),
4175 4 : ..Default::default()
4176 4 : };
4177 4 : let regs = write_regs(
4178 4 : &st,
4179 4 : &tenant,
4180 4 : &spec,
4181 4 : &ctx,
4182 4 : &HashMap::new(),
4183 4 : &HeaderMap::new(),
4184 4 : )
4185 4 : .await
4186 4 : .expect("registrations");
4187 4 : let ids: Vec<&str> = regs.iter().map(|r| r.reg_id.as_str()).collect();
4188 4 : assert_eq!(ids, vec!["urn:ngsi-ld:ContextSourceRegistration:inc"]);
4189 4 : }
4190 :
4191 : /// 4.3.6.1: "Context Brokers shall respect" a Context Source's declared
4192 : /// operations subset — explicit operation names and operation groups
4193 : /// (5.2.9) both gate; anything outside the list must not be forwarded.
4194 : #[test]
4195 4 : fn operations_subset_gates_forwarding() {
4196 4 : let mut r = reg("inclusive"); // ops = ["federationOps"], the 5.2.9 default
4197 4 : assert!(r.supports("queryEntity"));
4198 4 : assert!(r.supports("createSubscription"));
4199 4 : assert!(
4200 4 : !r.supports("createEntity"),
4201 : "federationOps carries no provision operations"
4202 : );
4203 4 : r.ops = vec!["updateOps".into()];
4204 4 : assert!(r.supports("updateAttrs"));
4205 4 : assert!(!r.supports("queryEntity"));
4206 4 : r.ops = vec!["createEntity".into()];
4207 4 : assert!(r.supports("createEntity"), "explicit op name matches");
4208 4 : assert!(!r.supports("deleteEntity"));
4209 4 : r.ops = vec![];
4210 4 : assert!(!r.supports("queryEntity"), "empty subset forwards nothing");
4211 4 : }
4212 :
4213 : /// 4.3.6.1: "all constraints specified in the registration shall be
4214 : /// respected" — a forwarded fragment is reduced to the attributes the
4215 : /// RegistrationInfo covers; when nothing covered remains there is no
4216 : /// forward at all (None).
4217 : #[test]
4218 4 : fn forwarded_fragment_reduced_to_registration_scope() {
4219 4 : let st = AppState::new("me".into());
4220 4 : let ctx = st.loader.core();
4221 4 : let speed = "https://uri.etsi.org/ngsi-ld/default-context/speed";
4222 4 : let mut r = reg("inclusive");
4223 4 : r.attrs = Some(vec![speed.into()]);
4224 4 : let obj = json!({
4225 4 : "id": "urn:x", "type": "Vehicle",
4226 4 : "speed": {"type": "Property", "value": 3},
4227 4 : "color": {"type": "Property", "value": "red"},
4228 4 : "@context": "https://example.org/ctx.jsonld"
4229 : });
4230 4 : let out = reduce_to_scope(obj.as_object().expect("obj"), &r, &ctx).expect("covered");
4231 4 : let out = out.as_object().expect("out");
4232 4 : assert!(out.contains_key("speed"));
4233 4 : assert!(out.contains_key("id") && out.contains_key("type"));
4234 4 : assert!(
4235 4 : !out.contains_key("color"),
4236 : "uncovered attribute must be dropped"
4237 : );
4238 4 : assert!(!out.contains_key("@context"));
4239 : // nothing covered ⇒ no forwarded fragment
4240 4 : let only_color = json!({
4241 4 : "id": "urn:x", "type": "Vehicle",
4242 4 : "color": {"type": "Property", "value": "red"}
4243 : });
4244 4 : assert!(reduce_to_scope(only_color.as_object().expect("obj"), &r, &ctx).is_none());
4245 : // an unscoped registration (attrs: None) passes everything but @context
4246 4 : r.attrs = None;
4247 4 : let full = reduce_to_scope(obj.as_object().expect("obj"), &r, &ctx).expect("all");
4248 4 : let full = full.as_object().expect("full");
4249 4 : assert!(full.contains_key("color") && !full.contains_key("@context"));
4250 4 : }
4251 :
4252 : /// 4.5.5.3 p.60: "if an expiresAt DateTime is present on the
4253 : /// Attribute and the date lies in the past, it shall be discarded" —
4254 : /// BEFORE the observedAt/modifiedAt recency comparison.
4255 : #[test]
4256 4 : fn merge_discards_expired_instances_before_recency() {
4257 4 : let attr = "https://uri.etsi.org/ngsi-ld/default-context/speed";
4258 4 : let mut base = json!({
4259 4 : "id": "urn:x", "type": ["T"],
4260 4 : attr: [{"type": "Property", "value": 1, "modifiedAt": "2026-01-01T00:00:00Z"}]
4261 : });
4262 : // fresher instance, but expired → must NOT win
4263 4 : let add = json!({
4264 4 : "id": "urn:x", "type": ["T"],
4265 4 : attr: [{"type": "Property", "value": 2,
4266 4 : "modifiedAt": "2026-06-01T00:00:00Z",
4267 4 : "expiresAt": "2020-01-01T00:00:00Z"}]
4268 : });
4269 4 : merge_docs(&mut base, &add, false);
4270 4 : assert_eq!(base[attr][0]["value"], 1, "expired instance was discarded");
4271 : // an expired BASE instance loses to a live remote one even if fresher
4272 4 : let mut base2 = json!({
4273 4 : "id": "urn:x", "type": ["T"],
4274 4 : attr: [{"type": "Property", "value": 1,
4275 4 : "modifiedAt": "2026-06-01T00:00:00Z",
4276 4 : "expiresAt": "2020-01-01T00:00:00Z"}]
4277 : });
4278 4 : let add2 = json!({
4279 4 : "id": "urn:x", "type": ["T"],
4280 4 : attr: [{"type": "Property", "value": 2, "modifiedAt": "2026-01-01T00:00:00Z"}]
4281 : });
4282 4 : merge_docs(&mut base2, &add2, false);
4283 4 : assert_eq!(base2[attr][0]["value"], 2, "live instance replaces expired");
4284 4 : }
4285 :
4286 : /// TS 104 175 8.5.3: "any attribute data where an observedAt DateTime is
4287 : /// present shall take precedence ... If no observedAt is present, the
4288 : /// attribute data with the most recent modifiedAt DateTime shall be
4289 : /// provided": an old observation beats a newer write that observed
4290 : /// nothing, whichever source answered first.
4291 : #[test]
4292 4 : fn merge_prefers_an_observed_instance_over_a_newer_unobserved_one() {
4293 4 : let attr = "https://uri.etsi.org/ngsi-ld/default-context/speed";
4294 4 : let observed = json!({
4295 4 : "id": "urn:x", "type": ["T"],
4296 4 : attr: [{"type": "Property", "value": "observed",
4297 4 : "observedAt": "2020-01-01T00:00:00Z",
4298 4 : "modifiedAt": "2020-01-01T00:00:00Z"}]
4299 : });
4300 4 : let unobserved = json!({
4301 4 : "id": "urn:x", "type": ["T"],
4302 4 : attr: [{"type": "Property", "value": "unobserved",
4303 4 : "modifiedAt": "2026-01-01T00:00:00Z"}]
4304 : });
4305 8 : for (first, second) in [(&observed, &unobserved), (&unobserved, &observed)] {
4306 8 : let mut base = first.clone();
4307 8 : merge_docs(&mut base, second, false);
4308 8 : assert_eq!(base[attr][0]["value"], "observed", "{first} then {second}");
4309 : }
4310 : // between two observations the later one wins, between two
4311 : // unobserved instances the later write
4312 4 : let mut later = observed.clone();
4313 4 : later[attr][0]["observedAt"] = json!("2021-01-01T00:00:00Z");
4314 4 : later[attr][0]["value"] = json!("later");
4315 4 : let mut base = observed.clone();
4316 4 : merge_docs(&mut base, &later, false);
4317 4 : assert_eq!(base[attr][0]["value"], "later");
4318 4 : let mut newer = unobserved.clone();
4319 4 : newer[attr][0]["modifiedAt"] = json!("2027-01-01T00:00:00Z");
4320 4 : newer[attr][0]["value"] = json!("newer");
4321 4 : let mut base = unobserved.clone();
4322 4 : merge_docs(&mut base, &newer, false);
4323 4 : assert_eq!(base[attr][0]["value"], "newer");
4324 4 : }
4325 :
4326 : /// 4.5.5.3: entity-level expiresAt — missing from at least one received
4327 : /// version → removed; present in all versions → furthest in the future.
4328 : #[test]
4329 4 : fn merge_entity_expires_at_intersection_and_max() {
4330 4 : let mut base = json!({"id": "urn:x", "type": ["T"], "expiresAt": "2030-01-01T00:00:00Z"});
4331 4 : let add = json!({"id": "urn:x", "type": ["T"], "expiresAt": "2031-01-01T00:00:00Z"});
4332 4 : merge_docs(&mut base, &add, false);
4333 4 : assert_eq!(base["expiresAt"], "2031-01-01T00:00:00Z");
4334 : // one version without expiresAt → removed
4335 4 : let add2 = json!({"id": "urn:x", "type": ["T"]});
4336 4 : merge_docs(&mut base, &add2, false);
4337 4 : assert!(base.get("expiresAt").is_none(), "expiresAt must be removed");
4338 : // and never re-introduced by a later version that has one
4339 4 : let add3 = json!({"id": "urn:x", "type": ["T"], "expiresAt": "2032-01-01T00:00:00Z"});
4340 4 : merge_docs(&mut base, &add3, false);
4341 4 : assert!(base.get("expiresAt").is_none());
4342 4 : }
4343 :
4344 : /// 4.3.6.2: "An auxiliary Context Source Registration never overrides
4345 : /// data held directly within a Context Broker." The entity-level
4346 : /// expiresAt reconciliation is part of that data, so an auxiliary
4347 : /// version can neither remove it nor push it further out.
4348 : #[test]
4349 4 : fn auxiliary_merge_never_touches_entity_expires_at() {
4350 4 : let mut base = json!({"id": "urn:x", "type": ["T"], "expiresAt": "2030-01-01T00:00:00Z"});
4351 4 : let aux_without = json!({"id": "urn:x", "type": ["T"]});
4352 4 : merge_docs(&mut base, &aux_without, true);
4353 4 : assert_eq!(
4354 4 : base["expiresAt"], "2030-01-01T00:00:00Z",
4355 : "an auxiliary version lacking expiresAt must not remove the broker's own"
4356 : );
4357 4 : let aux_later = json!({"id": "urn:x", "type": ["T"], "expiresAt": "2031-01-01T00:00:00Z"});
4358 4 : merge_docs(&mut base, &aux_later, true);
4359 4 : assert_eq!(
4360 4 : base["expiresAt"], "2030-01-01T00:00:00Z",
4361 : "an auxiliary version must not extend the broker's own expiresAt"
4362 : );
4363 4 : }
4364 :
4365 : /// 4.5.5.3 arbitrates on the most recent DateTime, and 4.6.3 lets the
4366 : /// same instant be written with or without a fraction — so the winner
4367 : /// must be chosen on the instant, never on the spelling.
4368 : #[test]
4369 4 : fn recency_arbitrates_on_the_instant_not_the_spelling() {
4370 4 : let attr = "https://uri.etsi.org/ngsi-ld/default-context/speed";
4371 4 : let mut base = json!({
4372 4 : "id": "urn:x", "type": ["T"],
4373 4 : attr: [{"type": "Property", "value": 1, "observedAt": "2026-01-01T00:00:01Z"}]
4374 : });
4375 4 : let add = json!({
4376 4 : "id": "urn:x", "type": ["T"],
4377 4 : attr: [{"type": "Property", "value": 2, "observedAt": "2026-01-01T00:00:01.500Z"}]
4378 : });
4379 4 : merge_docs(&mut base, &add, false);
4380 4 : assert_eq!(
4381 4 : base[attr][0]["value"], 2,
4382 : "the later instant wins even though its string sorts lower"
4383 : );
4384 : // and the converse: a fraction that is EARLIER must not win
4385 4 : let older = json!({
4386 4 : "id": "urn:x", "type": ["T"],
4387 4 : attr: [{"type": "Property", "value": 3, "observedAt": "2026-01-01T00:00:01.250Z"}]
4388 : });
4389 4 : merge_docs(&mut base, &older, false);
4390 4 : assert_eq!(base[attr][0]["value"], 2, "an earlier instant must not win");
4391 4 : }
4392 :
4393 : /// 4.5.5.2/4.5.7: the registration-scope filter narrows ATTRIBUTES, so
4394 : /// the entity-level lifetime members must cross it — dropping expiresAt
4395 : /// here made the 4.5.5.3 reconciliation delete the local one.
4396 : #[test]
4397 4 : fn scoped_import_keeps_the_entity_level_lifetime_members() {
4398 4 : let speed = "https://uri.etsi.org/ngsi-ld/default-context/speed";
4399 4 : let reg = FedReg {
4400 4 : attrs: Some(vec![speed.to_owned()]),
4401 4 : ..FedReg::default()
4402 4 : };
4403 4 : let remote = json!({
4404 4 : "id": "urn:ngsi-ld:Vehicle:1",
4405 4 : "type": "Vehicle",
4406 4 : "expiresAt": "2030-01-01T00:00:00Z",
4407 4 : "speed": {"type": "Property", "value": 1},
4408 4 : "brandName": {"type": "Property", "value": "x"}
4409 : });
4410 4 : let ctx = antares_jsonld::Loader::new().core();
4411 4 : let imported = import_entity(&remote, ®, &ctx).expect("import");
4412 4 : assert_eq!(
4413 4 : imported["expiresAt"], "2030-01-01T00:00:00Z",
4414 : "entity-level expiresAt must survive the scope filter"
4415 : );
4416 4 : assert!(
4417 4 : imported
4418 4 : .get("https://uri.etsi.org/ngsi-ld/default-context/brandName")
4419 4 : .is_none(),
4420 : "an out-of-scope attribute must not be imported"
4421 : );
4422 4 : }
4423 :
4424 : /// 4.5.5.2: a received version's entity-level expiresAt is pushed onto
4425 : /// each Attribute instance — added where absent, capped where the
4426 : /// Attribute's own expiresAt lies further in the future.
4427 : #[test]
4428 4 : fn merge_pushes_entity_expires_at_onto_attributes() {
4429 4 : let attr = "https://uri.etsi.org/ngsi-ld/default-context/speed";
4430 4 : let mut base = json!({"id": "urn:x", "type": ["T"]});
4431 4 : let add = json!({
4432 4 : "id": "urn:x", "type": ["T"],
4433 4 : "expiresAt": "2030-01-01T00:00:00Z",
4434 4 : attr: [
4435 4 : {"type": "Property", "value": 1},
4436 4 : {"type": "Property", "value": 2, "datasetId": "urn:ngsi-ld:Dataset:1",
4437 4 : "expiresAt": "2035-01-01T00:00:00Z"},
4438 4 : {"type": "Property", "value": 3, "datasetId": "urn:ngsi-ld:Dataset:2",
4439 4 : "expiresAt": "2029-01-01T00:00:00Z"}
4440 : ]
4441 : });
4442 4 : merge_docs(&mut base, &add, false);
4443 4 : let inst = base[attr].as_array().expect("attr array");
4444 4 : assert_eq!(inst[0]["expiresAt"], "2030-01-01T00:00:00Z", "added");
4445 4 : assert_eq!(inst[1]["expiresAt"], "2030-01-01T00:00:00Z", "capped");
4446 4 : assert_eq!(inst[2]["expiresAt"], "2029-01-01T00:00:00Z", "earlier kept");
4447 4 : }
4448 :
4449 : /// 6.3.17: "the error response should be as informative as possible" —
4450 : /// informative about the OPERATION, not about the deployment. The part
4451 : /// that failed is identified by its Context Source Registration id
4452 : /// (5.2.9), which the client can retrieve from /csourceRegistrations;
4453 : /// the registration `endpoint` is not part of any client-facing payload,
4454 : /// and a client able to provoke a partial failure must not be able to
4455 : /// enumerate the address of every registered Context Source.
4456 : #[tokio::test]
4457 4 : async fn partial_failure_detail_omits_the_peer_endpoint() {
4458 : use std::io::{Read, Write};
4459 4 : crate::allow_private();
4460 : // a Context Source that refuses every forwarded write
4461 4 : let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
4462 4 : let port = listener.local_addr().expect("addr").port();
4463 4 : std::thread::spawn(move || {
4464 4 : for stream in listener.incoming() {
4465 4 : let Ok(mut s) = stream else { continue };
4466 4 : let mut buf = [0u8; 4096];
4467 4 : let _ = s.read(&mut buf);
4468 4 : let _ = s.write_all(
4469 4 : b"HTTP/1.1 400 Bad Request\r\nConnection: close\r\nContent-Length: 0\r\n\r\n",
4470 4 : );
4471 : }
4472 4 : });
4473 4 : let st = AppState::new("me".into());
4474 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
4475 4 : let mut r = reg("inclusive");
4476 4 : r.endpoint = format!("http://127.0.0.1:{port}");
4477 4 : let part = forward_part(
4478 4 : &st,
4479 4 : reqwest::Method::POST,
4480 4 : format!("{}/ngsi-ld/v1/entities", r.endpoint),
4481 4 : &[],
4482 4 : &HeaderMap::new(),
4483 4 : &tenant,
4484 4 : &r,
4485 4 : antares_jsonld::CORE_CONTEXT,
4486 4 : Some(json!({"id": "urn:ngsi-ld:V:1", "type": "Vehicle"})),
4487 4 : )
4488 4 : .await;
4489 4 : assert_eq!(part.status, 400, "the peer refused the write");
4490 : // one failed forward + one succeeded local part ⇒ 207 Multi-Status
4491 4 : let local = Part::local(204, "local write applied");
4492 4 : let resp = combine(
4493 4 : crate::Profile::default(),
4494 4 : Some("urn:ngsi-ld:V:1"),
4495 4 : vec![local, part],
4496 4 : StatusCode::NO_CONTENT.into_response(),
4497 4 : &tenant,
4498 : );
4499 4 : assert_eq!(resp.status(), StatusCode::MULTI_STATUS);
4500 4 : let bytes = axum::body::to_bytes(resp.into_body(), 1 << 20)
4501 4 : .await
4502 4 : .expect("body");
4503 4 : let body = String::from_utf8_lossy(&bytes).into_owned();
4504 4 : assert!(
4505 4 : !body.contains(&format!("127.0.0.1:{port}")),
4506 : "the peer host:port must never reach the client, got {body}"
4507 : );
4508 4 : assert!(
4509 4 : body.contains(&r.reg_id),
4510 4 : "the registration id is the client-safe identifier, got {body}"
4511 4 : );
4512 4 : }
4513 :
4514 : /// 6.3.18: the Via header exists "to avoid infinite loops", and Table
4515 : /// 6.3.18-2 makes its listing part of registration matching — every
4516 : /// element is compared against every candidate registration. A chain
4517 : /// longer than any real cascade is therefore both a loop symptom and a
4518 : /// work amplifier, and is refused before the registrations are read.
4519 : #[tokio::test]
4520 4 : async fn via_chain_beyond_the_hop_ceiling_is_refused() {
4521 8 : let chain = |n: usize| {
4522 8 : (0..n)
4523 260 : .map(|i| format!("1.1 hop{i}"))
4524 8 : .collect::<Vec<_>>()
4525 8 : .join(", ")
4526 8 : };
4527 4 : let t = antares_model::TenantId::new("default").expect("tenant");
4528 4 : let over = hdrs(Some(&chain(MAX_VIA_HOPS + 1)));
4529 4 : assert!(
4530 4 : via_loop(&over, "me"),
4531 : "a chain past the ceiling is treated as a loop"
4532 : );
4533 4 : let mut regs = vec![reg("inclusive")];
4534 4 : let resp = handle_via_loop(&over, "me", &t, &mut regs)
4535 4 : .expect("an over-long Via chain must be refused");
4536 4 : assert_eq!(resp.status(), StatusCode::LOOP_DETECTED);
4537 : // and no registration is consulted: the candidate set is empty
4538 : // without a single registration document being examined
4539 4 : let st = AppState::new("me".into());
4540 4 : let ctx = st.loader.core();
4541 4 : let id = "urn:ngsi-ld:ContextSourceRegistration:hops";
4542 4 : st.store
4543 4 : .create(
4544 4 : &t,
4545 4 : Kind::Registration,
4546 4 : id,
4547 4 : json!({
4548 4 : "id": id,
4549 4 : "type": "ContextSourceRegistration",
4550 4 : "endpoint": "http://peer:9090",
4551 4 : "information": [{"entities": [{"type": "https://uri.etsi.org/ngsi-ld/default-context/Vehicle"}]}],
4552 4 : }),
4553 4 : ).await
4554 4 : .expect("seed registration");
4555 4 : let spec = crate::registry::CsrSpec {
4556 4 : types: Some(vec![
4557 4 : "https://uri.etsi.org/ngsi-ld/default-context/Vehicle".into()
4558 4 : ]),
4559 4 : ..Default::default()
4560 4 : };
4561 4 : assert!(
4562 4 : matching_regs(&st, &t, &spec, &ctx, &over)
4563 4 : .await
4564 4 : .expect("registrations")
4565 4 : .is_empty(),
4566 : "no registration is matched past the hop ceiling"
4567 : );
4568 : // at the ceiling the chain is still processed normally
4569 4 : let at = hdrs(Some(&chain(MAX_VIA_HOPS)));
4570 4 : assert!(!via_loop(&at, "me"));
4571 4 : assert!(handle_via_loop(&at, "me", &t, &mut vec![reg("inclusive")]).is_none());
4572 4 : assert_eq!(
4573 4 : matching_regs(&st, &t, &spec, &ctx, &at)
4574 4 : .await
4575 4 : .expect("registrations")
4576 4 : .len(),
4577 4 : 1
4578 4 : );
4579 4 : }
4580 :
4581 : /// 5.7.2.4/4.23.1: `would_federate` only answers WHETHER a query leaves
4582 : /// the local scope, so it never compiles a forwarding set — but its
4583 : /// verdict must stay identical to the set's emptiness for every shape
4584 : /// that gates matching (type, id, the Via chain, local scope).
4585 : #[tokio::test]
4586 4 : async fn would_federate_agrees_with_the_compiled_forward_set() {
4587 4 : let st = AppState::new("me".into());
4588 4 : let t = antares_model::TenantId::new("default").expect("tenant");
4589 4 : let ctx = st.loader.core();
4590 4 : let id = "urn:ngsi-ld:ContextSourceRegistration:wf";
4591 4 : st.store
4592 4 : .create(
4593 4 : &t,
4594 4 : Kind::Registration,
4595 4 : id,
4596 4 : json!({
4597 4 : "id": id,
4598 4 : "type": "ContextSourceRegistration",
4599 4 : "endpoint": "http://peer:9090",
4600 4 : "contextSourceAlias": "peer1",
4601 4 : "information": [{"entities": [{"type": "https://uri.etsi.org/ngsi-ld/default-context/Vehicle"}]}],
4602 4 : }),
4603 4 : ).await
4604 4 : .expect("seed registration");
4605 20 : let params = |kv: &[(&str, &str)]| {
4606 20 : kv.iter()
4607 24 : .map(|(k, v)| ((*k).to_owned(), (*v).to_owned()))
4608 20 : .collect::<HashMap<String, String>>()
4609 20 : };
4610 20 : for (p, headers, expected) in [
4611 4 : (params(&[("type", "Vehicle")]), hdrs(None), true),
4612 4 : (params(&[("type", "Parking")]), hdrs(None), false),
4613 4 : (
4614 4 : params(&[("type", "Vehicle"), ("local", "true")]),
4615 4 : hdrs(None),
4616 4 : false,
4617 4 : ),
4618 4 : (
4619 4 : params(&[("type", "Vehicle")]),
4620 4 : hdrs(Some("1.1 peer1")),
4621 4 : false,
4622 4 : ),
4623 4 : (
4624 4 : params(&[("type", "Vehicle")]),
4625 4 : hdrs(Some("1.1 other")),
4626 4 : true,
4627 4 : ),
4628 4 : ] {
4629 20 : assert_eq!(
4630 20 : would_federate(&st, &t, &ctx, &p, &headers)
4631 20 : .await
4632 20 : .expect("registrations"),
4633 4 : expected,
4634 4 : "would_federate verdict for {p:?}"
4635 4 : );
4636 20 : assert_eq!(
4637 20 : active(&p)
4638 16 : && !matching_regs(&st, &t, &query_spec(&ctx, &p), &ctx, &headers)
4639 16 : .await
4640 16 : .expect("registrations")
4641 16 : .is_empty(),
4642 4 : expected,
4643 4 : "the compiled forward set must agree for {p:?}"
4644 4 : );
4645 4 : }
4646 4 : }
4647 : }
4648 :
4649 : #[cfg(test)]
4650 : mod clause_4_20 {
4651 : use super::*;
4652 :
4653 48 : fn reg(ops: &[&str]) -> FedReg {
4654 : FedReg {
4655 48 : ops: ops.iter().map(|s| (*s).to_owned()).collect(),
4656 48 : ..FedReg::default()
4657 : }
4658 48 : }
4659 :
4660 : /// Table 4.20-2: associationOps is federationOps WITHOUT the EntityMap
4661 : /// support operations (and without createEntityMapQueryTemporal, which is
4662 : /// in neither group).
4663 : #[test]
4664 4 : fn association_ops_exclude_the_entity_map_operations() {
4665 4 : let r = reg(&["associationOps"]);
4666 16 : for op in [
4667 4 : "retrieveEntity",
4668 4 : "queryEntity",
4669 4 : "deleteSubscription",
4670 4 : "retrieveContextSourceIdentity",
4671 4 : ] {
4672 16 : assert!(r.supports(op), "{op} is in associationOps");
4673 : }
4674 16 : for op in [
4675 4 : "retrieveEntityMap",
4676 4 : "updateEntityMap",
4677 4 : "deleteEntityMap",
4678 4 : "createEntityMapQueryEntity",
4679 4 : ] {
4680 16 : assert!(
4681 16 : !r.supports(op),
4682 : "{op} is NOT in associationOps (Table 4.20-2)"
4683 : );
4684 : }
4685 4 : }
4686 :
4687 : /// Table 4.20-1/2: individual names match themselves; groups match their
4688 : /// members; nothing matches createEntityMapQueryTemporal except itself.
4689 : #[test]
4690 4 : fn groups_and_individual_names() {
4691 4 : assert!(reg(&["federationOps"]).supports("retrieveEntityMap"));
4692 4 : assert!(reg(&["redirectionOps"]).supports("purgeEntity"));
4693 4 : assert!(!reg(&["redirectionOps"]).supports("createSubscription"));
4694 4 : assert!(reg(&["updateOps"]).supports("replaceAttrs"));
4695 4 : assert!(!reg(&["updateOps"]).supports("deleteEntity"));
4696 4 : assert!(reg(&["retrieveOps"]).supports("queryEntity"));
4697 4 : assert!(!reg(&["retrieveOps"]).supports("retrieveTemporal"));
4698 4 : assert!(reg(&["createEntityMapQueryTemporal"]).supports("createEntityMapQueryTemporal"));
4699 12 : for group in ["federationOps", "associationOps", "redirectionOps"] {
4700 12 : assert!(
4701 12 : !reg(&[group]).supports("createEntityMapQueryTemporal"),
4702 : "{group} does not include createEntityMapQueryTemporal"
4703 : );
4704 : }
4705 4 : }
4706 : }
4707 :
4708 : #[cfg(test)]
4709 : mod same_source_merge {
4710 : use super::*;
4711 :
4712 60 : fn reg(id: &str, endpoint: &str, attrs: Option<&[&str]>, types: &[&str]) -> FedReg {
4713 : FedReg {
4714 60 : reg_id: id.into(),
4715 60 : endpoint: endpoint.into(),
4716 60 : mode: "inclusive".into(),
4717 60 : ops: vec!["federationOps".into()],
4718 60 : attrs: attrs.map(|a| a.iter().map(|s| (*s).to_owned()).collect()),
4719 60 : ent_types: types.iter().map(|s| (*s).to_owned()).collect(),
4720 : // type-only EntityInfo, as the registration builder reads it
4721 : ent_unrestricted: true,
4722 60 : ..FedReg::default()
4723 : }
4724 60 : }
4725 :
4726 : /// Two registrations for one Context Source become one forward whose
4727 : /// scopes are the union; a different endpoint or mode stays separate.
4728 : #[test]
4729 4 : fn registrations_of_one_source_fold_into_one_forward() {
4730 4 : let merged = merge_same_source(vec![
4731 4 : reg("urn:r:1", "http://a", Some(&["speed"]), &["Vehicle"]),
4732 4 : reg(
4733 4 : "urn:r:2",
4734 4 : "http://a",
4735 4 : Some(&["heading", "speed"]),
4736 4 : &["Vehicle"],
4737 : ),
4738 4 : reg("urn:r:3", "http://b", Some(&["speed"]), &["Vehicle"]),
4739 : ]);
4740 4 : assert_eq!(merged.len(), 2);
4741 4 : assert_eq!(merged[0].reg_id, "urn:r:1");
4742 4 : assert_eq!(
4743 4 : merged[0].attrs.as_deref(),
4744 4 : Some(&["speed".to_owned(), "heading".to_owned()][..])
4745 : );
4746 4 : assert_eq!(merged[1].endpoint, "http://b");
4747 :
4748 : // equal attribute scopes: the types union
4749 4 : let merged = merge_same_source(vec![
4750 4 : reg("urn:r:1", "http://a", Some(&["speed"]), &["Vehicle"]),
4751 4 : reg("urn:r:2", "http://a", Some(&["speed"]), &["Bike"]),
4752 : ]);
4753 4 : assert_eq!(merged.len(), 1);
4754 4 : assert_eq!(merged[0].ent_types, ["Vehicle", "Bike"]);
4755 :
4756 : // 4.3.6.1, ETSI IOP_003_02_02: (Vehicle × all) and (Bike × speed)
4757 : // folded would be (both × all), a scope neither registration declared
4758 4 : let merged = merge_same_source(vec![
4759 4 : reg("urn:r:1", "http://a", None, &["Vehicle"]),
4760 4 : reg("urn:r:2", "http://a", Some(&["speed"]), &["Bike"]),
4761 : ]);
4762 4 : assert_eq!(merged.len(), 2, "an inexact union stays two forwards");
4763 :
4764 : // an unscoped registration widens the merged one to everything
4765 4 : let merged = merge_same_source(vec![
4766 4 : reg("urn:r:1", "http://a", Some(&["speed"]), &["Vehicle"]),
4767 4 : reg("urn:r:2", "http://a", None, &[]),
4768 : ]);
4769 4 : assert_eq!(merged.len(), 1);
4770 4 : assert!(merged[0].attrs.is_none());
4771 4 : assert!(merged[0].ent_types.is_empty(), "and to every type");
4772 :
4773 4 : let mut other_mode = reg("urn:r:2", "http://a", None, &[]);
4774 4 : other_mode.mode = "exclusive".into();
4775 4 : let merged = merge_same_source(vec![reg("urn:r:1", "http://a", None, &[]), other_mode]);
4776 4 : assert_eq!(merged.len(), 2, "a different mode is a different forward");
4777 4 : }
4778 :
4779 : /// 4.3.6.1: a source "may indicate that they are only willing to respond
4780 : /// to a limited subset of API operations. Context Brokers shall respect
4781 : /// this". The fold unions the entity and attribute scope, so folding two
4782 : /// registrations that declare different operations would send an
4783 : /// operation only one of them offered for the OTHER's Entities.
4784 : #[test]
4785 4 : fn registrations_of_one_source_that_offer_different_operations_do_not_fold() {
4786 4 : let mut reads = reg("urn:r:1", "http://a", None, &["Vehicle"]);
4787 4 : reads.ops = vec!["queryEntity".into()];
4788 4 : let mut writes = reg("urn:r:2", "http://a", None, &["Bike"]);
4789 4 : writes.ops = vec!["createEntity".into()];
4790 4 : let merged = merge_same_source(vec![reads, writes]);
4791 4 : assert_eq!(
4792 4 : merged.len(),
4793 : 2,
4794 : "a different operation subset is a different forward"
4795 : );
4796 4 : assert!(
4797 4 : merged[0].supports("queryEntity") && !merged[0].supports("createEntity"),
4798 : "the read registration must not gain the write registration's operation"
4799 : );
4800 4 : assert_eq!(
4801 4 : merged[0].ent_types,
4802 : ["Vehicle"],
4803 : "and must not gain its Entity scope either"
4804 : );
4805 :
4806 : // the same subset written two ways is still one source: 5.2.9 lets
4807 : // `operations` name a group, and the default IS a group
4808 4 : let mut spelled_out = reg("urn:r:2", "http://a", None, &["Bike"]);
4809 4 : spelled_out.ops = antares_model::operations::group_members("federationOps")
4810 4 : .expect("the default group")
4811 4 : .iter()
4812 76 : .map(|s| (*s).to_owned())
4813 4 : .collect();
4814 4 : let merged = merge_same_source(vec![
4815 4 : reg("urn:r:1", "http://a", None, &["Vehicle"]),
4816 4 : spelled_out,
4817 : ]);
4818 4 : assert_eq!(
4819 4 : merged.len(),
4820 : 1,
4821 : "federationOps and the names it stands for are one subset"
4822 : );
4823 4 : assert_eq!(merged[0].ent_types, ["Vehicle", "Bike"]);
4824 4 : }
4825 : }
|
