Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! One notification on the wire: cooldown, endpoint options, receiverInfo, in-process transports and the outbound send.
3 :
4 : use super::*;
5 :
6 : /// POST the Notification (5.3.1) and write the 5.2.14.2 bookkeeping back.
7 4865 : pub(crate) async fn deliver(
8 4865 : st: &AppState,
9 4865 : tenant: &TenantId,
10 4865 : sub: &Value,
11 4865 : data: Vec<Value>,
12 4865 : ctx: &Context,
13 4865 : ) {
14 : // a notification body is bounded the way an inbound body is (6.3.4 wall,
15 : // MAX_BODY_BYTES): a grouped delivery over the cap leaves as several
16 : // notifications, each whole entities, never one unbounded POST
17 4865 : for chunk in chunk_by_bytes(data, *crate::bounds::MAX_BODY_BYTES) {
18 4865 : deliver_as(
19 4865 : st,
20 4865 : tenant,
21 4865 : Kind::Subscription,
22 4865 : sub,
23 4865 : "Notification",
24 4865 : chunk,
25 4865 : ctx,
26 4865 : None,
27 4865 : )
28 4865 : .await;
29 : }
30 4748 : }
31 :
32 : /// Split `data` into runs whose serialized sizes stay under `cap`, cutting
33 : /// only at whole items; one item alone over the cap still travels alone.
34 5042 : pub(super) fn chunk_by_bytes(data: Vec<Value>, cap: usize) -> Vec<Vec<Value>> {
35 5042 : let mut out: Vec<Vec<Value>> = Vec::new();
36 5042 : let mut size = 0usize;
37 5433 : for item in data {
38 5433 : let n = serde_json::to_vec(&item).map(|b| b.len()).unwrap_or(0);
39 5433 : match out.last_mut() {
40 395 : Some(run) if size.saturating_add(n) <= cap => {
41 379 : run.push(item);
42 379 : size = size.saturating_add(n);
43 379 : }
44 5054 : _ => {
45 5054 : out.push(vec![item]);
46 5054 : size = n;
47 5054 : }
48 : }
49 : }
50 5042 : out
51 5042 : }
52 :
53 : /// Table 5.2.15-1 `timeout`: per-endpoint delivery deadline in milliseconds.
54 : /// "The NGSI-LD system can override this value" — clamped to [100 ms, 30 s]
55 : /// so one subscription cannot park a delivery task for minutes. Default 5 s
56 : /// (the previous hard-coded deadline). HTTP only: the clause scopes it to
57 : /// bindings that "always return a response".
58 5086 : pub(super) fn endpoint_timeout_ms(ep: &serde_json::Map<String, Value>) -> u32 {
59 5086 : ep.get("timeout")
60 5086 : .and_then(Value::as_f64)
61 5086 : .filter(|t| *t > 0.0)
62 5086 : .map(|t| (t as u32).clamp(100, 30_000))
63 5086 : .unwrap_or(5_000)
64 5086 : }
65 :
66 : /// Table 5.2.15-1 `cooldown`: "Once a failure has occurred, minimum period of
67 : /// time in milliseconds which shall elapse before attempting to make a
68 : /// subsequent notification to the same endpoint after failure. If requests
69 : /// are received before the cooldown period has expired, no notification is
70 : /// sent." — i.e. matches inside the window are DROPPED, not queued.
71 5106 : pub(super) fn in_cooldown(sub: &Value, now: chrono::DateTime<chrono::Utc>) -> bool {
72 5106 : let n = sub.get("notification");
73 5106 : let Some(cd) = n
74 5106 : .and_then(|n| n.get("endpoint"))
75 5106 : .and_then(|e| e.get("cooldown"))
76 5106 : .and_then(Value::as_f64)
77 5106 : .filter(|c| *c > 0.0)
78 : else {
79 5094 : return false;
80 : };
81 : // the gate exists only "once a failure has occurred" and only until a
82 : // success clears it — notification.status tracks exactly that (5.2.14.2)
83 12 : if n.and_then(|n| n.get("status")).and_then(Value::as_str) != Some("failed") {
84 4 : return false;
85 8 : }
86 8 : let Some(lf) = n.and_then(|n| n.get("lastFailure")).and_then(Value::as_str) else {
87 0 : return false;
88 : };
89 8 : let Ok(t) = chrono::DateTime::parse_from_rfc3339(lf) else {
90 0 : return false;
91 : };
92 8 : let elapsed = now
93 8 : .signed_duration_since(t.with_timezone(&chrono::Utc))
94 8 : .num_milliseconds();
95 8 : (elapsed as f64) < cd
96 5106 : }
97 :
98 : #[allow(clippy::too_many_arguments)] // one param per notification dimension
99 5196 : pub(super) async fn deliver_as(
100 5196 : st: &AppState,
101 5196 : tenant: &TenantId,
102 5196 : kind: Kind,
103 5196 : sub: &Value,
104 5196 : ntype: &str,
105 5196 : data: Vec<Value>,
106 5196 : ctx: &Context,
107 5196 : trigger_reason: Option<&str>,
108 5196 : ) {
109 : // 5.2.12: a paused subscription (isActive false) and an expired one send
110 : // nothing. Every caller that assembles data locally checks this first;
111 : // the ones that arrive with data already assembled — the 5.8.6 inbound
112 : // notification among them — did not, so the check belongs here too.
113 5196 : if !is_active(sub) {
114 0 : return;
115 5196 : }
116 5196 : let sub_id = sub_str(sub, "id").unwrap_or_default().to_owned();
117 5196 : let Some(ep) = sub
118 5196 : .get("notification")
119 5196 : .and_then(|n| n.get("endpoint"))
120 5196 : .and_then(Value::as_object)
121 : else {
122 0 : return;
123 : };
124 5196 : let Some(uri) = ep.get("uri").and_then(Value::as_str) else {
125 0 : return;
126 : };
127 : // 5.8.1.4 consumer half: the internal CSR subscription's notifications
128 : // are handled in-process (urn:antares:distsub:{tenant}\n{own sub id})
129 5196 : if let Some(own) = uri.strip_prefix("urn:antares:distsub:") {
130 104 : if let (Some((_, own_id)), Some(handler)) =
131 106 : (own.split_once('\n'), st.csource_notification.as_ref())
132 : {
133 104 : handler(st, tenant, own_id, trigger_reason, &data).await;
134 2 : }
135 102 : return;
136 5090 : }
137 : // 6.3.8: the binding comes from the registry and nowhere else. Creation
138 : // rejects an endpoint whose scheme no sink serves, so a stored row that
139 : // still names one was hand-edited — it is dropped, never delivered
140 : // through some other binding.
141 5090 : if st.sinks.sink_for_uri(uri).is_none() {
142 0 : tracing::warn!(
143 : "subscription {sub_id} endpoint {} has no registered binding",
144 0 : redact_userinfo(uri)
145 : );
146 0 : return;
147 5090 : }
148 : // endpoint.cooldown — drop (never queue) while the window is open.
149 : // Before any bookkeeping: a suppressed notification was never sent, so
150 : // timesSent/lastNotification must not move.
151 5090 : if in_cooldown(sub, chrono::Utc::now()) {
152 0 : tracing::debug!("subscription {sub_id} in cooldown; notification suppressed (5.2.15)");
153 0 : return;
154 5090 : }
155 : // An open circuit is the same class of self-inflicted suppression: no
156 : // request leaves the process, so Table 5.2.14.2-1 timesSent ("number of
157 : // times that the notification has been sent") and lastNotification ("the
158 : // instant when the last notification has been sent") must not move
159 : // either. `is_open` returning false IS the half-open probe, so the
160 : // check stays exactly once per attempt.
161 : // A binding that opens no socket has no destination for the policy or
162 : // the breaker to judge, nor has a delivery the forward transport makes
163 : // in process; every other network binding is policed below.
164 5090 : let policed =
165 5090 : in_process(st, uri).is_none() && st.sinks.sink_for_uri(uri).is_some_and(|s| s.network());
166 5090 : if policed && st.egress.is_open(tenant.as_str(), uri) {
167 8 : tracing::debug!(
168 : "notification to {} short-circuited (breaker open)",
169 0 : redact_userinfo(uri)
170 : );
171 8 : return;
172 5082 : }
173 5082 : let accept = ep
174 5082 : .get("accept")
175 5082 : .and_then(Value::as_str)
176 5082 : .unwrap_or("application/json");
177 5082 : let now = now_iso();
178 5082 : let mut body = json!({
179 5082 : "id": format!("urn:ngsi-ld:{ntype}:{}", uuid::Uuid::new_v4()),
180 5082 : "type": ntype,
181 5082 : "subscriptionId": sub_id,
182 5082 : "notifiedAt": now,
183 5082 : "data": data,
184 : });
185 5082 : if let (Some(r), Some(body)) = (trigger_reason, body.as_object_mut()) {
186 173 : body.insert("triggerReason".into(), Value::String(r.into()));
187 4909 : }
188 : // ADR-0020: the engine sees the notification the broker means to send,
189 : // before it is encoded — the data entities are still the 5.2.14.1
190 : // documents the subscription asked for, not a FeatureCollection or a
191 : // set of @context-injected ones, so a projection lands on entities the
192 : // engine can name. Before the bookkeeping below, because 5.8.6 moves
193 : // timesSent for a notification that "shall be sent": one that is
194 : // dropped here never was, exactly like a cooldown or an open breaker.
195 : // The subject is the subscriber's, stored when the subscription was
196 : // created — delivery is broker-initiated and there is no request here
197 : // to read one off.
198 : // With no engine attached the answer is Deliver and the subject is
199 : // never read out of the subscription — this runs once per notification
200 : // per subscription.
201 5082 : if let Some(engine) = &st.policy {
202 36 : match crate::policy::pre_notify(
203 36 : engine.as_ref(),
204 36 : &crate::policy::stored_subject(tenant, sub),
205 36 : sub,
206 36 : &mut body,
207 36 : ) {
208 14 : crate::policy::NotifyDecision::Deliver => {}
209 : crate::policy::NotifyDecision::Drop => {
210 2 : tracing::debug!("subscription {sub_id} notification dropped by the policy engine");
211 2 : return;
212 : }
213 20 : crate::policy::NotifyDecision::Filter(f) => {
214 : // a condition the notification path cannot re-run is not a
215 : // narrowing it can claim to have applied
216 20 : if f.q.is_some() || f.scope_q.is_some() || f.areas().next().is_some() {
217 4 : tracing::error!(
218 : "policy engine {} narrowed a notification with a query; dropping it",
219 0 : engine.name()
220 : );
221 4 : return;
222 16 : }
223 : // a linked condition already read the target unprojected:
224 : // the narrowing came too late to apply to it
225 16 : if (!f.omit.is_empty() || !f.pick.is_empty() || f.has_rules())
226 16 : && linked_condition(sub)
227 : {
228 2 : tracing::debug!(
229 : "subscription {sub_id} has a linked q under a projecting narrowing; dropping"
230 : );
231 2 : return;
232 14 : }
233 14 : let rules = crate::repr::type_rules(&f).filter(|r| !r.rules.is_empty());
234 14 : let f = crate::repr::compacted_filter(&f, ctx);
235 14 : let n = sub.get("notification");
236 14 : let key_values = matches!(
237 14 : n.and_then(|n| n.get("format")).and_then(Value::as_str),
238 0 : Some("keyValues" | "simplified")
239 : );
240 14 : if let Some(arr) = body.get_mut("data").and_then(Value::as_array_mut) {
241 14 : for e in arr.iter_mut() {
242 14 : project_joined(&f, e, key_values);
243 14 : }
244 0 : }
245 : // the rules read each Entity's stored state
246 : // now, so one that left zone A carries no `speed`, nor a
247 : // `previousValue` of it, and one no rule holds for is not
248 : // notified at all
249 14 : if let Some(rules) = rules {
250 4 : let shown = shown_now(
251 4 : st,
252 4 : tenant,
253 4 : Some(&rules),
254 4 : None,
255 4 : body.get("data").unwrap_or(&Value::Null),
256 4 : ctx,
257 4 : )
258 4 : .await;
259 4 : if let Some(arr) = body.get_mut("data").and_then(Value::as_array_mut) {
260 4 : arr.retain(|e| matches!(entity_shown(&shown, e), Some(Some(_))));
261 4 : for e in arr.iter_mut() {
262 4 : project_shown(&shown, e, key_values);
263 4 : }
264 4 : if arr.is_empty() {
265 0 : tracing::debug!(
266 : "subscription {sub_id}: no notified Entity is in the narrowing now; dropping"
267 : );
268 0 : return;
269 4 : }
270 0 : }
271 10 : }
272 : }
273 : }
274 5046 : }
275 : // 5.8.6: a subscription's ngsildConformance pins the notification format —
276 : // amend the data entities per the 4.3.6.8 fallbacks.
277 5074 : if let Some(ver) = sub_str(sub, "ngsildConformance").and_then(crate::conformance::parse_version)
278 : {
279 0 : if let Some(d) = body.get_mut("data") {
280 0 : crate::conformance::amend_payload(d, ver);
281 0 : }
282 5074 : }
283 5074 : if accept == "application/ld+json" {
284 : // JSON-LD notifications carry the @context inside each data entity
285 : // (046_14: data[0] must contain @context; no Link header) — same rule
286 : // over MQTT: with ld+json the @context travels in the body (7.2).
287 14 : if let Some(arr) = body.get_mut("data").and_then(Value::as_array_mut) {
288 14 : for e in arr.iter_mut() {
289 14 : *e = inject_context(e.clone(), ctx);
290 14 : }
291 0 : }
292 5060 : }
293 5074 : if accept == "application/geo+json" {
294 : // Table 5.3.1-1: with endpoint.accept application/geo+json, data is
295 : // a FeatureCollection (5.2.30); if receiverInfo carries
296 : // Prefer=body=json the FeatureCollection carries no @context.
297 4 : let prefer_body_json = ep
298 4 : .get("receiverInfo")
299 4 : .and_then(Value::as_array)
300 4 : .is_some_and(|ri| {
301 2 : ri.iter().any(|kv| {
302 2 : kv.get("key").and_then(Value::as_str) == Some("Prefer")
303 2 : && kv.get("value").and_then(Value::as_str) == Some("body=json")
304 2 : })
305 2 : });
306 4 : let entities = body
307 4 : .get("data")
308 4 : .and_then(Value::as_array)
309 4 : .cloned()
310 4 : .unwrap_or_default();
311 4 : let mut fc = crate::repr::to_geojson_collection(entities, None);
312 4 : if let (false, Some(fc)) = (prefer_body_json, fc.as_object_mut()) {
313 2 : fc.insert("@context".into(), crate::negotiate::served_context(ctx));
314 2 : }
315 4 : if let Some(body) = body.as_object_mut() {
316 4 : body.insert("data".into(), fc);
317 4 : }
318 5070 : }
319 5074 : let receiver_info = kv_pairs(ep.get("receiverInfo"));
320 :
321 : // 6.3.22: a subscription living under a snapshot's synthetic tenant
322 : // notifies with the NGSILD-Snapshot header and the OWNER tenant — the
323 : // internal "snap-…" tenant never leaks.
324 5074 : let (hdr_tenant, snapshot_id) =
325 5074 : match crate::snapshots::snapshot_of_synth(st, tenant.as_str()).await {
326 2 : Some((owner, sid)) => (owner, Some(sid)),
327 5072 : None => (tenant.clone(), None),
328 : };
329 :
330 : // Prepared BEFORE the bookkeeping writeback so the optimistic stamp
331 : // covers only the in-flight attempt (046_12_01 race). The parts are
332 : // transport-neutral: the sink registered for the endpoint's scheme turns
333 : // them into HTTP headers (6.3.8) or an MQTT metadata object (Table
334 : // 7.2-2).
335 5074 : let mut info = receiver_info;
336 5074 : strip_reserved_markers(&mut info);
337 5074 : if hdr_tenant.as_str() != "default" {
338 284 : info.push(("NGSILD-Tenant".into(), hdr_tenant.as_str().to_owned()));
339 4790 : }
340 5074 : if let Some(sid) = &snapshot_id {
341 2 : info.push(("NGSILD-Snapshot".into(), sid.clone()));
342 5072 : }
343 5074 : let outbound = Outbound {
344 5074 : body,
345 5074 : accept: accept.to_owned(),
346 5074 : link: link_header_value(ctx),
347 5074 : receiver_info: info,
348 5074 : notifier_info: kv_pairs(ep.get("notifierInfo")),
349 5074 : };
350 : // Bookkeeping BEFORE the send (5.8.6/5.2.14.2: lastNotification is the
351 : // instant the notification is sent). The ETSI mock unblocks the test the
352 : // moment the request ARRIVES, so a post-response-only writeback races the
353 : // test's immediate Retrieve Subscription (CI flake on 046_12_01).
354 : // Optimistic ok; a failed attempt is corrected right below — the transient
355 : // window is the in-flight attempt itself, and the failure TPs wait for the
356 : // attempt to resolve before asserting.
357 : // One store call: the stamp is a fixed mutation, so a backend can write
358 : // it as a single statement instead of locking the row across a round
359 : // trip. At fan-out that lock is what serializes delivery.
360 : // An in-process delivery (a socket, the forward transport) books with
361 : // every other one of its tenant in one statement; a network one
362 : // keeps its own stamp right before its own send (046_12_01).
363 5074 : let booked = if !policed && kind == Kind::Subscription {
364 116 : super::coalesce::book_in_process(st, tenant, &sub_id, &now).await
365 : } else {
366 4958 : st.store
367 4958 : .record_delivery(tenant, kind, &sub_id, &now)
368 4958 : .await
369 4958 : .unwrap_or_else(|e| {
370 0 : tracing::warn!("bookkeeping writeback failed: {e}");
371 0 : None
372 0 : })
373 : };
374 : // 5.8.6: notifications are sent for the subscriptions the broker holds.
375 : // No row to book against means the subscription was deleted (or the
376 : // store failed) between matching and delivery — nothing may be sent.
377 5074 : let Some(booked) = booked else {
378 8 : return;
379 : };
380 5066 : let mut prev_success = booked.prev_success;
381 5066 : mirror_bookkeeping(st, tenant, kind, &sub_id, Some(booked.doc));
382 : // The notification endpoint is an egress destination like any other
383 : // — policy check once, breaker consulted before the attempt.
384 : // A refusal is a delivery failure for bookkeeping (status "failed",
385 : // lastSuccess rolled back below) but never breaker state: the policy
386 : // verdict says nothing about the endpoint's health.
387 5066 : let refused = policed
388 4950 : && match st.egress.check_destination(uri).await {
389 4942 : Ok(()) => false,
390 8 : Err(e) => {
391 8 : tracing::warn!(
392 : "notification endpoint {} refused by egress policy: {e}",
393 0 : redact_userinfo(uri)
394 : );
395 8 : true
396 : }
397 : };
398 : // (delivered, timed_out): only a TIMEOUT-class failure feeds the breaker
399 : // — that protects against peers that eat the deadline. An endpoint
400 : // that ANSWERS (any status) is alive, costs only its own response time,
401 : // and 6.3.8 says the notification shall be sent — suppressing sends to a
402 : // responding host:port starves unrelated subscriptions sharing it.
403 5066 : let timeout_ms = endpoint_timeout_ms(ep);
404 5066 : let first = if refused {
405 8 : Err((false, "refused by egress policy".to_owned()))
406 : } else {
407 5058 : send_outbound(st, tenant, uri, timeout_ms, &outbound).await
408 : };
409 4917 : let (ok, timed_out) = match &first {
410 4764 : Ok(()) => (true, false),
411 153 : Err((t, _)) => (false, *t),
412 : };
413 4917 : if policed && !refused {
414 4795 : if ok {
415 4650 : st.egress.record_success(tenant.as_str(), uri);
416 4650 : } else if timed_out {
417 36 : st.egress.record_failure(tenant.as_str(), uri);
418 109 : } else {
419 109 : // the destination responded (or refused fast): alive — clear
420 109 : // any stale consecutive-timeout state
421 109 : st.egress.record_success(tenant.as_str(), uri);
422 109 : }
423 122 : }
424 : // Delivery counters by binding (facade — no-op without the broker's
425 : // recorder). The label is the sink's first scheme, so the two members of
426 : // a family share one series.
427 4917 : let scheme = st
428 4917 : .sinks
429 4917 : .sink_for_uri(uri)
430 4917 : .and_then(|s| s.schemes().first().copied())
431 4917 : .unwrap_or("unknown");
432 4917 : if ok {
433 4764 : metrics::counter!("antares_notifications_sent_total", "scheme" => scheme).increment(1);
434 4764 : } else {
435 153 : metrics::counter!("antares_notifications_failed_total", "scheme" => scheme).increment(1);
436 153 : }
437 4917 : if !ok {
438 : // 5.8.6 / 5.11.7: subscription status → "failed" on delivery failure;
439 : // roll back the optimistic lastSuccess stamp.
440 153 : let ts = now_iso();
441 : // An in-process failure (a socket queue that overflowed) waits in
442 : // its tenant's queue like its booking did: a fan-out of overflowing
443 : // sockets holds one write-back at a time, not one per socket.
444 153 : let failed_doc = if !policed && kind == Kind::Subscription {
445 0 : super::coalesce::fail_in_process(st, tenant, &sub_id, prev_success.take(), &ts).await
446 : } else {
447 153 : record_failure(st, tenant, kind, &sub_id, prev_success.take(), &ts).await
448 : };
449 153 : mirror_bookkeeping(st, tenant, kind, &sub_id, failed_doc);
450 : // Retries are transport, not new notifications: they run on their
451 : // own task (never on the request path, never delaying another
452 : // subscription's delivery) and book only the final outcome — a
453 : // success sets lastSuccess/status ok without touching timesSent;
454 : // an exhausted policy leaves a dead letter.
455 : #[cfg(not(target_arch = "wasm32"))]
456 153 : if !refused && st.delivery.attempts > 1 {
457 16 : let first_err = first.err().map(|(_, e)| e).unwrap_or_default();
458 16 : let (st, tenant, uri) = (st.clone(), tenant.clone(), uri.to_owned());
459 16 : crate::spawn(async move {
460 16 : retry_and_settle(
461 16 : &st, &tenant, kind, &sub_id, &uri, timeout_ms, outbound, first_err,
462 16 : )
463 16 : .await;
464 12 : });
465 137 : }
466 4764 : }
467 5043 : }
468 :
469 : /// 5.8.6 / 5.11.7 after a failed attempt: status "failed", the optimistic
470 : /// `lastSuccess` rolled back to `prev_success`, `lastFailure` and
471 : /// `timesFailed` moved. The stored document as it now stands, `None` when the
472 : /// row is gone or the write failed (logged).
473 153 : pub(super) async fn record_failure(
474 153 : st: &AppState,
475 153 : tenant: &TenantId,
476 153 : kind: Kind,
477 153 : sub_id: &str,
478 153 : mut prev_success: Option<Value>,
479 153 : ts: &str,
480 153 : ) -> Option<Value> {
481 153 : let mut failed_doc: Option<Value> = None;
482 153 : st.store
483 153 : .mutate(tenant, kind, sub_id, |doc| {
484 133 : if let Some(o) = doc.as_object_mut() {
485 133 : o.insert("status".into(), Value::String("failed".into()));
486 133 : }
487 133 : if let Some(n) = doc
488 133 : .as_object_mut()
489 133 : .and_then(|o| o.get_mut("notification"))
490 133 : .and_then(Value::as_object_mut)
491 : {
492 133 : match prev_success.take() {
493 20 : Some(v) => n.insert("lastSuccess".into(), v),
494 113 : None => n.remove("lastSuccess"),
495 : };
496 133 : n.insert("lastFailure".into(), Value::String(ts.to_owned()));
497 : // Table 5.2.14.2-1 timesFailed: "Number of times an
498 : // unsuccessful response (or timeout) has been received
499 : // when delivering the notification" — an output-only
500 : // member implementations shall generate.
501 133 : let failed = n.get("timesFailed").and_then(Value::as_i64).unwrap_or(0);
502 133 : n.insert("timesFailed".into(), json!(failed.saturating_add(1)));
503 133 : n.insert("status".into(), Value::String("failed".into()));
504 0 : }
505 133 : failed_doc = Some(doc.clone());
506 133 : Ok::<(), antares_model::NgsiError>(())
507 133 : })
508 153 : .await
509 153 : .unwrap_or_else(|e| {
510 0 : tracing::warn!("failure-status writeback failed: {e}");
511 0 : None
512 0 : });
513 153 : failed_doc
514 153 : }
515 :
516 : /// A `KeyValuePair[]` member of `endpoint` (Table 5.2.15-1) as owned pairs.
517 : /// A member that is not an array of well-formed pairs contributes nothing:
518 : /// 5.2.12 validation at creation already refused a malformed one.
519 12216 : pub(crate) fn kv_pairs(v: Option<&Value>) -> Vec<(String, String)> {
520 12216 : v.and_then(Value::as_array)
521 12216 : .map(|a| {
522 82 : a.iter()
523 106 : .filter_map(|kv| {
524 : Some((
525 106 : kv.get("key")?.as_str()?.to_owned(),
526 106 : kv.get("value")?.as_str()?.to_owned(),
527 : ))
528 106 : })
529 82 : .collect()
530 82 : })
531 12216 : .unwrap_or_default()
532 12216 : }
533 :
534 : /// 6.3.22 / 6.3.8: `NGSILD-Tenant` and `NGSILD-Snapshot` on a notification
535 : /// are the broker's own statement of where the data came from, appended to
536 : /// the `receiverInfo` pairs. The HTTP binding appends every pair it is
537 : /// handed, so a subscriber naming one of the two in `receiverInfo` would
538 : /// put a second value of it on the wire beside the broker's, and a receiver
539 : /// reading "the" tenant of a notification could not tell which one the
540 : /// broker meant. Ordinary custom headers are untouched.
541 5078 : pub(super) fn strip_reserved_markers(info: &mut Vec<(String, String)>) {
542 5078 : info.retain(|(k, _)| {
543 36 : !k.eq_ignore_ascii_case("NGSILD-Tenant") && !k.eq_ignore_ascii_case("NGSILD-Snapshot")
544 36 : });
545 5078 : }
546 :
547 : /// The forward transport, when it delivers the notifications sent to `uri`
548 : /// in process.
549 14843 : pub(super) fn in_process<'a>(
550 14843 : st: &'a AppState,
551 14843 : uri: &str,
552 14843 : ) -> Option<&'a Arc<dyn crate::ForwardTransport>> {
553 14843 : st.forward_transport.get().filter(|t| t.notifies(uri))
554 14843 : }
555 :
556 : /// One attempt on the wire, through the binding the registry holds for the
557 : /// endpoint's scheme (6.3.8), or in process when the forward transport
558 : /// takes the endpoint, on every attempt, retries and
559 : /// replays included. `Err((timed_out, why))`: only a timeout-class failure
560 : /// feeds the breaker — an endpoint that answers is alive.
561 5080 : pub(super) async fn send_outbound(
562 5080 : st: &AppState,
563 5080 : tenant: &TenantId,
564 5080 : uri: &str,
565 5080 : timeout_ms: u32,
566 5080 : outbound: &Outbound,
567 5080 : ) -> Result<(), (bool, String)> {
568 5080 : if let Some(transport) = in_process(st, uri) {
569 10 : return transport
570 10 : .notify(tenant, uri, &outbound.body)
571 10 : .await
572 10 : .map_err(|e| (false, e));
573 5070 : }
574 5070 : let Some(sink) = st.sinks.sink_for_uri(uri) else {
575 2 : return Err((
576 2 : false,
577 2 : format!(
578 2 : "no notification binding registered for {}",
579 2 : redact_userinfo(uri)
580 2 : ),
581 2 : ));
582 : };
583 5068 : sink.deliver(
584 5068 : uri,
585 5068 : outbound,
586 5068 : std::time::Duration::from_millis(u64::from(timeout_ms)),
587 5068 : )
588 5068 : .await
589 4919 : .map_err(|e| (e.timed_out, e.message))
590 4931 : }
591 :
592 : /// The matcher reads subscriptions from the
593 : /// SubMirror, so every notification bookkeeping writeback must be applied
594 : /// there too — otherwise the mirror copy never gains
595 : /// `notification.lastNotification` and 5.2.12 `throttling` suppresses
596 : /// nothing. In-process apply only: a KV write per notification would not
597 : /// scale to the 100k-sub target, so in bus=nats multi-pod deployments the
598 : /// throttling window is per-pod approximate.
599 : /// Known ceiling: exact distributed throttling = per-notification KV sync or a
600 : /// store read in `throttled()`; add if a deployment needs the strict window.
601 : /// The mirror learns the counters from the document the writeback just
602 : /// committed under the row lock; `None` (no row) leaves it untouched.
603 5223 : pub(super) fn mirror_bookkeeping(
604 5223 : st: &AppState,
605 5223 : tenant: &TenantId,
606 5223 : kind: Kind,
607 5223 : sub_id: &str,
608 5223 : doc: Option<Value>,
609 5223 : ) {
610 5223 : if kind != Kind::Subscription {
611 214 : return;
612 5009 : }
613 5009 : if let (Some(m), Some(doc)) = (&st.sub_mirror, doc) {
614 4721 : m.apply(tenant.as_str(), sub_id, Some(doc));
615 4721 : }
616 5223 : }
|
