Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Notification delivery.
3 : //!
4 : //! A sink serves one family of `endpoint.uri` schemes (6.3.8, and 7.2 for
5 : //! the optional MQTT binding): it validates its own endpoints at
6 : //! subscription creation and delivers the prepared notification. The
7 : //! registry keys sinks by scheme and is the only way one is chosen — a
8 : //! scheme it does not hold is rejected at creation, never delivered through
9 : //! another binding. Sinks: http/reqwest, mqtt/rumqttc behind the `mqtt`
10 : //! feature.
11 : //!
12 : //! The egress policy (allowlist, private-range deny, per-destination
13 : //! breakers) runs in the caller before `deliver`, so a sink registered from
14 : //! outside this workspace cannot step around it.
15 : #![cfg_attr(not(test), warn(clippy::expect_used))]
16 :
17 : use antares_model::NgsiError;
18 : use serde_json::Value;
19 : use std::future::Future;
20 : use std::pin::Pin;
21 : use std::time::Duration;
22 :
23 : pub mod http;
24 : #[cfg(feature = "mqtt")]
25 : pub mod mqtt;
26 :
27 : pub use http::HttpSink;
28 :
29 : /// One prepared notification, deliverable any number of times: a retry or a
30 : /// dead-letter replay renders the identical message from the same parts.
31 : /// The parts are transport-neutral — 6.3.8 turns them into HTTP headers,
32 : /// Table 7.2-2 into the MQTT message's `metadata` object.
33 : #[derive(Clone, Debug, PartialEq)]
34 : pub struct Outbound {
35 : /// The Notification (5.3.1) to deliver.
36 : pub body: Value,
37 : /// `endpoint.accept` (Table 5.2.15-1): the MIME type of `body`.
38 : pub accept: String,
39 : /// The JSON-LD `@context` Link value belonging to `body` (6.3.8).
40 : pub link: String,
41 : /// `endpoint.receiverInfo` (Table 5.2.15-1) followed by the tenant and
42 : /// snapshot markers the binding has to convey (6.3.22).
43 : pub receiver_info: Vec<(String, String)>,
44 : /// `endpoint.notifierInfo` (Table 5.2.15-1): the parameters the binding
45 : /// needs to set up its channel, e.g. Table 7.2-1's MQTT-QoS. Opaque to
46 : /// every sink but the one whose scheme the endpoint names.
47 : pub notifier_info: Vec<(String, String)>,
48 : }
49 :
50 : impl Outbound {
51 : /// A dead letter back into deliverable form. Letters written before the
52 : /// bindings moved behind the registry carry a rendered HTTP header list, or
53 : /// an already-wrapped clause 7 message, instead of the endpoint members they
54 : /// were rendered from; both read back, so an upgrade does not strand the
55 : /// letters an operator has not replayed yet.
56 18 : pub fn from_dead_letter(letter: &Value) -> Result<Self, String> {
57 18 : let pairs = |v: &Value| -> Vec<(String, String)> {
58 4 : serde_json::from_value::<Vec<(String, String)>>(v.clone()).unwrap_or_default()
59 4 : };
60 18 : if letter.get("accept").and_then(Value::as_str).is_some() {
61 2 : return Ok(Self {
62 2 : body: letter["payload"].clone(),
63 2 : accept: letter["accept"].as_str().unwrap_or_default().to_owned(),
64 2 : link: letter["link"].as_str().unwrap_or_default().to_owned(),
65 2 : receiver_info: pairs(&letter["receiverInfo"]),
66 2 : notifier_info: pairs(&letter["notifierInfo"]),
67 2 : });
68 16 : }
69 : // Pre-registry HTTP: Content-Type and Link ARE the accept and link they
70 : // were rendered from; every other header came from receiverInfo.
71 16 : if let Some(headers) = letter.get("headers").filter(|h| !h.is_null()) {
72 12 : let headers = serde_json::from_value::<Vec<(String, String)>>(headers.clone())
73 12 : .map_err(|e| format!("dead letter headers unreadable: {e}"))?;
74 20 : let take = |name: &str| {
75 20 : headers
76 20 : .iter()
77 30 : .find(|(k, _)| k.eq_ignore_ascii_case(name))
78 20 : .map(|(_, v)| v.clone())
79 20 : .unwrap_or_default()
80 20 : };
81 : return Ok(Self {
82 10 : body: letter["payload"].clone(),
83 10 : accept: take("Content-Type"),
84 10 : link: take("Link"),
85 10 : receiver_info: headers
86 10 : .iter()
87 22 : .filter(|(k, _)| {
88 22 : !k.eq_ignore_ascii_case("Content-Type") && !k.eq_ignore_ascii_case("Link")
89 22 : })
90 10 : .cloned()
91 10 : .collect(),
92 10 : notifier_info: Vec::new(),
93 : });
94 4 : }
95 : // Pre-registry MQTT: the payload is the 7.2 message, so the notification
96 : // and its metadata come back out of the wrapper.
97 4 : let meta = &letter["payload"]["metadata"];
98 4 : if let Some(meta) = meta.as_object() {
99 2 : let mut notifier_info = Vec::new();
100 2 : if let Some(q) = letter["mqtt"]["qos"].as_u64() {
101 2 : notifier_info.push(("MQTT-QoS".to_owned(), q.to_string()));
102 2 : }
103 2 : if let Some(v5) = letter["mqtt"]["v5"].as_bool() {
104 2 : let v = if v5 { "mqtt5.0" } else { "mqtt3.1.1" };
105 2 : notifier_info.push(("MQTT-Version".to_owned(), v.to_owned()));
106 0 : }
107 : return Ok(Self {
108 2 : body: letter["payload"]["body"].clone(),
109 2 : accept: meta
110 2 : .get("Content-Type")
111 2 : .and_then(Value::as_str)
112 2 : .unwrap_or("application/json")
113 2 : .to_owned(),
114 2 : link: meta
115 2 : .get("Link")
116 2 : .and_then(Value::as_str)
117 2 : .unwrap_or_default()
118 2 : .to_owned(),
119 2 : receiver_info: meta
120 2 : .iter()
121 6 : .filter(|(k, _)| k.as_str() != "Content-Type" && k.as_str() != "Link")
122 2 : .filter_map(|(k, v)| Some((k.clone(), v.as_str()?.to_owned())))
123 2 : .collect(),
124 2 : notifier_info,
125 : });
126 2 : }
127 2 : Err("dead letter carries no deliverable notification".to_owned())
128 18 : }
129 :
130 : /// `notifier_info` in the borrowed pair form the sinks parse.
131 0 : pub fn notifier_pairs(&self) -> Vec<(&str, &str)> {
132 0 : self.notifier_info
133 0 : .iter()
134 0 : .map(|(k, v)| (k.as_str(), v.as_str()))
135 0 : .collect()
136 0 : }
137 : }
138 :
139 : /// Why one delivery attempt did not land. `timed_out` is the only class the
140 : /// caller's circuit breaker counts: an endpoint that answers — with any
141 : /// status — is alive.
142 : #[derive(Clone, Debug, PartialEq)]
143 : pub struct DeliveryError {
144 : /// The attempt ran out of time rather than being answered.
145 : pub timed_out: bool,
146 : /// Failure text for the log, the subscription status and the dead
147 : /// letter. Never carries endpoint credentials.
148 : pub message: String,
149 : }
150 :
151 : impl DeliveryError {
152 : /// The endpoint answered, or refused, within the deadline.
153 54 : pub fn failed(message: impl Into<String>) -> Self {
154 54 : Self {
155 54 : timed_out: false,
156 54 : message: message.into(),
157 54 : }
158 54 : }
159 :
160 : /// The deadline passed with no answer.
161 0 : pub fn timeout(message: impl Into<String>) -> Self {
162 0 : Self {
163 0 : timed_out: true,
164 0 : message: message.into(),
165 0 : }
166 0 : }
167 : }
168 :
169 : /// The future one `deliver` returns. `Send` on every target: the un-Send
170 : /// piece of a browser fetch is fenced inside `antares_jsonld::http_interaction`.
171 : pub type DeliveryFuture<'a> = Pin<Box<dyn Future<Output = Result<(), DeliveryError>> + Send + 'a>>;
172 :
173 : /// How one notification is delivered: how many attempts, how they are
174 : /// spaced, and how long after the first attempt the last one may still
175 : /// start. The default is a single attempt — exactly 5.8.6, which sends the
176 : /// notification once and books the outcome. Retries are an operator choice:
177 : /// they never move `timesSent` again (the notification is sent ONCE, the
178 : /// attempts are transport), a retry that succeeds books `lastSuccess` and
179 : /// `status` ok, an exhausted policy leaves a dead letter.
180 : #[derive(Clone, Copy, Debug, PartialEq)]
181 : pub struct DeliveryPolicy {
182 : /// Total attempts, first one included. 1 = never retry.
183 : pub attempts: u32,
184 : /// Delay before the first retry; doubles per retry up to `MAX_BACKOFF`.
185 : pub backoff: std::time::Duration,
186 : /// Fraction of the delay randomised in both directions (0.2 = ±20 %),
187 : /// so many subscriptions to one dead endpoint do not retry in lockstep.
188 : pub jitter: f32,
189 : /// A retry that would start later than this after the first attempt is
190 : /// not made.
191 : pub max_age: std::time::Duration,
192 : }
193 :
194 : impl Default for DeliveryPolicy {
195 3267 : fn default() -> Self {
196 3267 : Self {
197 3267 : attempts: 1,
198 3267 : backoff: std::time::Duration::from_secs(1),
199 3267 : jitter: 0.2,
200 3267 : max_age: std::time::Duration::from_secs(300),
201 3267 : }
202 3267 : }
203 : }
204 :
205 : impl DeliveryPolicy {
206 : /// Ceiling on one delay, whatever the doubling says.
207 : pub const MAX_BACKOFF: std::time::Duration = std::time::Duration::from_secs(60);
208 :
209 2 : pub fn with_max_age(self, max_age: std::time::Duration) -> Self {
210 2 : Self { max_age, ..self }
211 2 : }
212 :
213 : /// ANTARES_NOTIFY_ATTEMPTS / ANTARES_NOTIFY_BACKOFF_MS /
214 : /// ANTARES_NOTIFY_MAX_AGE_SECS, each optional; a value that is present
215 : /// but not a positive integer is a startup error, never a silent default.
216 59 : pub fn from_env() -> Result<Self, String> {
217 177 : let get = |k: &str| std::env::var(k).ok();
218 59 : Self::parse(
219 59 : get("ANTARES_NOTIFY_ATTEMPTS").as_deref(),
220 59 : get("ANTARES_NOTIFY_BACKOFF_MS").as_deref(),
221 59 : get("ANTARES_NOTIFY_MAX_AGE_SECS").as_deref(),
222 : )
223 59 : }
224 :
225 77 : pub fn parse(
226 77 : attempts: Option<&str>,
227 77 : backoff_ms: Option<&str>,
228 77 : max_age_secs: Option<&str>,
229 77 : ) -> Result<Self, String> {
230 213 : fn positive(name: &str, raw: Option<&str>) -> Result<Option<u64>, String> {
231 213 : match raw {
232 191 : None => Ok(None),
233 22 : Some(v) => v
234 22 : .trim()
235 22 : .parse::<u64>()
236 22 : .ok()
237 22 : .filter(|n| *n > 0)
238 22 : .map(Some)
239 22 : .ok_or_else(|| format!("{name} must be a positive integer, got {v:?}")),
240 : }
241 213 : }
242 77 : let d = Self::default();
243 : Ok(Self {
244 77 : attempts: positive("ANTARES_NOTIFY_ATTEMPTS", attempts)?
245 71 : .map_or(d.attempts, |n| u32::try_from(n).unwrap_or(u32::MAX)),
246 71 : backoff: positive("ANTARES_NOTIFY_BACKOFF_MS", backoff_ms)?
247 65 : .map_or(d.backoff, std::time::Duration::from_millis),
248 65 : jitter: d.jitter,
249 65 : max_age: positive("ANTARES_NOTIFY_MAX_AGE_SECS", max_age_secs)?
250 63 : .map_or(d.max_age, std::time::Duration::from_secs),
251 : })
252 77 : }
253 :
254 : /// The delay before the next attempt after `made` attempts, `elapsed`
255 : /// after the first one — `None` when the policy is exhausted or the
256 : /// retry would start past `max_age`.
257 152 : pub fn next_delay(
258 152 : &self,
259 152 : made: u32,
260 152 : elapsed: std::time::Duration,
261 152 : ) -> Option<std::time::Duration> {
262 152 : if made == 0 || made >= self.attempts {
263 12 : return None;
264 140 : }
265 140 : let doubled = self
266 140 : .backoff
267 140 : .checked_mul(1u32.checked_shl(made - 1).unwrap_or(u32::MAX))
268 140 : .unwrap_or(Self::MAX_BACKOFF)
269 140 : .min(Self::MAX_BACKOFF);
270 : // ±jitter, seeded from the clock's sub-second noise: cheap and
271 : // uncorrelated enough to spread retries; no RNG dependency.
272 140 : let noise = (std::time::SystemTime::now()
273 140 : .duration_since(std::time::UNIX_EPOCH)
274 140 : .map(|d| d.subsec_nanos())
275 140 : .unwrap_or(0)
276 : % 1000) as f32
277 : / 1000.0;
278 140 : let factor = 1.0 + self.jitter.clamp(0.0, 1.0) * (2.0 * noise - 1.0);
279 140 : let delay = doubled.mul_f32(factor.max(0.0));
280 140 : (elapsed + delay <= self.max_age).then_some(delay)
281 152 : }
282 : }
283 :
284 : /// Strip the authority's userinfo from an endpoint URI. 7.2 allows
285 : /// credentials there (`mqtt[s]://[<username>][:<password>]@<host>…`), and a
286 : /// rejected or failed endpoint travels back to the client as the `detail`
287 : /// member of the ProblemDetails body (5.5.3) and into the delivery logs —
288 : /// neither may carry the subscription's password. Everything after the
289 : /// authority's last `@` is kept; an `@` in the path or topic is data.
290 492 : pub fn redact_userinfo(uri: &str) -> String {
291 492 : if let Some(scheme_end) = uri.find("//") {
292 488 : let rest = &uri[scheme_end + 2..];
293 488 : let authority_end = rest.find(['/', '?', '#']).unwrap_or(rest.len());
294 488 : if let Some(at) = rest[..authority_end].rfind('@') {
295 52 : return format!("{}{}", &uri[..scheme_end + 2], &rest[at + 1..]);
296 436 : }
297 4 : }
298 440 : uri.to_owned()
299 492 : }
300 :
301 : /// A delivery binding for one URI scheme family.
302 : pub trait NotificationSink: Send + Sync {
303 : /// Schemes this sink serves, e.g. `["http", "https"]`.
304 : fn schemes(&self) -> &'static [&'static str];
305 :
306 : /// 5.8.1.4: the endpoint's own syntax and parameters, checked at
307 : /// subscription creation rather than at first delivery. `notifier_info`
308 : /// is `endpoint.notifierInfo` (Table 5.2.15-1) as key/value pairs. An
309 : /// endpoint that does not meet the sink's requirements is
310 : /// BadRequestData; the message names the URI with any userinfo
311 : /// credentials stripped, since it travels back in `detail` (5.5.3).
312 : fn parse_endpoint(&self, uri: &str, notifier_info: &[(&str, &str)]) -> Result<(), NgsiError>;
313 :
314 : /// Does an endpoint of this binding name a network destination? The
315 : /// caller runs the egress guard — host and port policy, private-range
316 : /// and metadata-address deny, per-destination circuit breaker — against
317 : /// every endpoint that does, before `deliver` and never inside it, so
318 : /// one guard covers every binding whatever scheme it serves. The
319 : /// default is the safe answer: a sink is policed unless it declares
320 : /// that it opens no socket, and a release binary registers no sink that
321 : /// declares otherwise.
322 402 : fn network(&self) -> bool {
323 402 : true
324 402 : }
325 :
326 : /// One attempt on the wire. `timeout` is `endpoint.timeout`
327 : /// (Table 5.2.15-1) already clamped by the caller.
328 : fn deliver<'a>(
329 : &'a self,
330 : uri: &'a str,
331 : out: &'a Outbound,
332 : timeout: Duration,
333 : ) -> DeliveryFuture<'a>;
334 : }
335 :
336 : /// Scheme → sink registry; populated by the composition root. Choosing a
337 : /// binding goes through here and nowhere else, so an endpoint scheme with no
338 : /// sink can never fall through to the HTTP binding.
339 : #[derive(Default)]
340 : pub struct SinkRegistry {
341 : sinks: Vec<Box<dyn NotificationSink>>,
342 : }
343 :
344 : impl SinkRegistry {
345 : /// Add a binding. A scheme already served keeps its first sink.
346 6384 : pub fn register(&mut self, sink: Box<dyn NotificationSink>) {
347 6384 : self.sinks.push(sink);
348 6384 : }
349 :
350 : /// The scheme of an endpoint URI, lowercased per IETF RFC 3986 §3.1.
351 1854 : pub fn scheme_of(uri: &str) -> String {
352 1854 : uri.split(':').next().unwrap_or("").to_ascii_lowercase()
353 1854 : }
354 :
355 : /// The sink serving `scheme`, if any.
356 1850 : pub fn sink_for(&self, scheme: &str) -> Option<&dyn NotificationSink> {
357 : // Linear scan is fine at <5 sinks; switch to a map when sinks multiply.
358 1850 : self.sinks
359 1850 : .iter()
360 1904 : .find(|s| s.schemes().contains(&scheme))
361 1850 : .map(AsRef::as_ref)
362 1850 : }
363 :
364 : /// The sink serving an endpoint URI, by its scheme.
365 1504 : pub fn sink_for_uri(&self, uri: &str) -> Option<&dyn NotificationSink> {
366 1504 : self.sink_for(&Self::scheme_of(uri))
367 1504 : }
368 :
369 : /// 5.8.1.4 reject-at-creation: the sink for this endpoint, having
370 : /// accepted the endpoint's own syntax. An endpoint whose scheme this
371 : /// deployment cannot deliver to is input data that does not meet the
372 : /// requirements of the operation — BadRequestData (Table 5.5.2-1, 400
373 : /// per Table 6.3.2-1). Not OperationNotSupported: Create Subscription
374 : /// is supported, this endpoint value is not.
375 344 : pub fn require(&self, uri: &str, notifier_info: &[(&str, &str)]) -> Result<(), NgsiError> {
376 344 : let scheme = Self::scheme_of(uri);
377 344 : let sink = self.sink_for(&scheme).ok_or_else(|| {
378 6 : NgsiError::BadRequestData(format!(
379 6 : "no notification binding registered for endpoint scheme {scheme:?} (6.3.8)"
380 6 : ))
381 6 : })?;
382 338 : sink.parse_endpoint(uri, notifier_info)
383 344 : }
384 :
385 : /// Schemes this deployment can deliver to, for `/q/health` and the
386 : /// startup banner.
387 92 : pub fn schemes(&self) -> Vec<&'static str> {
388 92 : let mut v: Vec<&'static str> = self
389 92 : .sinks
390 92 : .iter()
391 182 : .flat_map(|s| s.schemes())
392 92 : .copied()
393 92 : .collect();
394 92 : v.sort_unstable();
395 92 : v.dedup();
396 92 : v
397 92 : }
398 : }
399 :
400 : #[cfg(test)]
401 : mod tests {
402 : use super::*;
403 : use serde_json::json;
404 :
405 : struct FakeHttp;
406 : impl NotificationSink for FakeHttp {
407 18 : fn schemes(&self) -> &'static [&'static str] {
408 18 : &["http", "https"]
409 18 : }
410 8 : fn parse_endpoint(&self, uri: &str, _ni: &[(&str, &str)]) -> Result<(), NgsiError> {
411 8 : uri.contains("://")
412 8 : .then_some(())
413 8 : .ok_or_else(|| NgsiError::BadRequestData(format!("no authority in {uri:?}")))
414 8 : }
415 0 : fn deliver<'a>(
416 0 : &'a self,
417 0 : _uri: &'a str,
418 0 : _o: &'a Outbound,
419 0 : _t: Duration,
420 0 : ) -> DeliveryFuture<'a> {
421 0 : Box::pin(async { Ok(()) })
422 0 : }
423 : }
424 :
425 8 : fn registry() -> SinkRegistry {
426 8 : let mut reg = SinkRegistry::default();
427 8 : reg.register(Box::new(FakeHttp));
428 8 : reg
429 8 : }
430 :
431 : /// 5.8.1.4 + Table 5.5.2-1: an endpoint scheme no sink serves is input
432 : /// data that does not meet the operation's requirements — BadRequestData
433 : /// (400), not OperationNotSupported. Create Subscription is supported.
434 : #[test]
435 2 : fn unknown_scheme_is_bad_request_data() {
436 2 : let reg = registry();
437 2 : assert!(reg.require("http://h/n", &[]).is_ok());
438 2 : assert!(reg.require("https://h/n", &[]).is_ok());
439 2 : let err = reg
440 2 : .require("ws://h/n", &[])
441 2 : .expect_err("ws has no sink in v1");
442 2 : assert_eq!(err.status(), 400);
443 2 : assert!(matches!(err, NgsiError::BadRequestData(_)), "{err:?}");
444 2 : assert!(format!("{err}").contains("ws"), "{err}");
445 2 : }
446 :
447 : /// The scheme comparison is case-insensitive (IETF RFC 3986 §3.1),
448 : /// so `HTTP://…` is not silently unroutable.
449 : #[test]
450 2 : fn scheme_matching_ignores_case() {
451 2 : assert!(registry().require("HTTP://h/n", &[]).is_ok());
452 2 : assert_eq!(SinkRegistry::scheme_of("MQTTS://h/t"), "mqtts");
453 2 : }
454 :
455 : /// The sink's own endpoint check runs at creation, and its error is the
456 : /// one the client sees.
457 : #[test]
458 2 : fn the_sinks_own_rejection_is_returned() {
459 2 : let err = registry()
460 2 : .require("http:/malformed", &[])
461 2 : .expect_err("no authority");
462 2 : assert_eq!(err.status(), 400);
463 2 : assert!(format!("{err}").contains("no authority"), "{err}");
464 2 : }
465 :
466 : /// Every binding this workspace ships opens a socket, so every endpoint
467 : /// it delivers to passes the caller's egress policy. A sink that
468 : /// declares otherwise is in-process only and belongs to a test.
469 : #[test]
470 2 : fn every_shipped_sink_is_policed() {
471 2 : let mut reg = SinkRegistry::default();
472 2 : reg.register(Box::new(crate::HttpSink::new(
473 2 : antares_jsonld::client_builder(antares_jsonld::EgressPolicy {
474 2 : allow_private: true,
475 2 : })
476 2 : .build()
477 2 : .expect("client"),
478 2 : )));
479 : #[cfg(feature = "mqtt")]
480 2 : reg.register(Box::new(crate::mqtt::MqttSink::default()));
481 2 : assert!(!reg.sinks.is_empty());
482 4 : for s in ®.sinks {
483 4 : assert!(
484 4 : s.network(),
485 : "shipped sink {:?} skips the egress policy",
486 0 : s.schemes()
487 : );
488 : }
489 2 : }
490 :
491 : /// A dead letter reads back into the same notification whichever broker
492 : /// wrote it: the current shape, and the two shapes written before the
493 : /// bindings moved behind the registry.
494 : #[test]
495 2 : fn dead_letters_of_every_shape_read_back() {
496 2 : let body = json!({"type": "Notification", "subscriptionId": "urn:s:1"});
497 2 : let link = "<https://ctx>; rel=\"http://www.w3.org/ns/json-ld#context\"";
498 :
499 2 : let current = json!({"uri": "http://h/n", "payload": body,
500 2 : "accept": "application/ld+json", "link": link,
501 2 : "receiverInfo": [["Authorization", "Bearer t"]],
502 2 : "notifierInfo": []});
503 2 : let o = Outbound::from_dead_letter(¤t).expect("current shape");
504 2 : assert_eq!(o.body, body);
505 2 : assert_eq!(o.accept, "application/ld+json");
506 2 : assert_eq!(
507 : o.receiver_info,
508 2 : [("Authorization".into(), "Bearer t".into())]
509 : );
510 :
511 2 : let legacy_http = json!({"uri": "http://h/n", "binding": "http", "payload": body,
512 2 : "headers": [["Content-Type", "application/json"], ["Link", link],
513 : ["Authorization", "Bearer t"]]});
514 2 : let o = Outbound::from_dead_letter(&legacy_http).expect("legacy http shape");
515 2 : assert_eq!(o.body, body);
516 2 : assert_eq!(o.accept, "application/json");
517 2 : assert_eq!(o.link, link);
518 2 : assert_eq!(
519 : o.receiver_info,
520 2 : [("Authorization".into(), "Bearer t".into())],
521 : "Content-Type and Link are the accept and link, not receiverInfo"
522 : );
523 :
524 2 : let legacy_mqtt = json!({"uri": "mqtt://h/t", "binding": "mqtt",
525 2 : "mqtt": {"qos": 2, "v5": false},
526 2 : "payload": {"metadata": {"Content-Type": "application/json", "Link": link,
527 2 : "NGSILD-Tenant": "acme"},
528 2 : "body": body}});
529 2 : let o = Outbound::from_dead_letter(&legacy_mqtt).expect("legacy mqtt shape");
530 2 : assert_eq!(
531 : o.body, body,
532 : "the notification comes back out of the wrapper"
533 : );
534 2 : assert_eq!(o.accept, "application/json");
535 2 : assert_eq!(o.receiver_info, [("NGSILD-Tenant".into(), "acme".into())]);
536 2 : assert_eq!(
537 : o.notifier_info,
538 2 : [
539 2 : ("MQTT-QoS".to_owned(), "2".to_owned()),
540 2 : ("MQTT-Version".to_owned(), "mqtt3.1.1".to_owned())
541 2 : ]
542 : );
543 :
544 2 : assert!(Outbound::from_dead_letter(&json!({"uri": "http://h/n"})).is_err());
545 2 : }
546 :
547 : /// Endpoint URIs may carry credentials (mqtt[s]://user:pass@host, 7.1);
548 : /// log lines must never leak them.
549 : #[test]
550 2 : fn log_redaction_strips_uri_userinfo() {
551 2 : let red = redact_userinfo("mqtts://alice:s3cret@broker:8883/topic");
552 2 : assert_eq!(red, "mqtts://broker:8883/topic");
553 2 : assert!(!red.contains("s3cret"));
554 2 : assert!(!red.contains("alice"));
555 2 : assert_eq!(
556 2 : redact_userinfo("http://host:9090/notify"),
557 : "http://host:9090/notify"
558 : );
559 : // an '@' beyond the authority is path data, not userinfo
560 2 : assert_eq!(redact_userinfo("http://h/p@x"), "http://h/p@x");
561 2 : }
562 :
563 : /// An unregistered scheme resolves to no sink at delivery time either —
564 : /// there is no fall-through to the first registered binding.
565 : #[test]
566 2 : fn delivery_lookup_never_falls_through() {
567 2 : let reg = registry();
568 2 : assert!(reg.sink_for_uri("http://h/n").is_some());
569 2 : assert!(reg.sink_for_uri("memory://box").is_none());
570 2 : assert!(reg.sink_for("").is_none());
571 2 : assert_eq!(reg.schemes(), vec!["http", "https"]);
572 2 : }
573 : }
574 :
575 : #[cfg(test)]
576 : mod policy_tests {
577 : use super::DeliveryPolicy;
578 : use std::time::Duration;
579 :
580 10 : fn policy(attempts: u32, backoff_ms: u64, max_age_secs: u64) -> DeliveryPolicy {
581 10 : DeliveryPolicy {
582 10 : attempts,
583 10 : backoff: Duration::from_millis(backoff_ms),
584 10 : jitter: 0.0,
585 10 : max_age: Duration::from_secs(max_age_secs),
586 10 : }
587 10 : }
588 :
589 : /// Drive `op` the way the delivery loop does: one attempt, then a
590 : /// retry after every delay the policy grants. Returns the outcome and
591 : /// how many calls were made.
592 8 : fn run<E>(p: DeliveryPolicy, mut op: impl FnMut(u32) -> Result<(), E>) -> (Result<(), E>, u32) {
593 8 : let mut made = 1;
594 8 : let mut elapsed = Duration::ZERO;
595 8 : let mut last = op(made);
596 18 : while last.is_err() {
597 16 : let Some(d) = p.next_delay(made, elapsed) else {
598 6 : break;
599 : };
600 10 : elapsed += d;
601 10 : made += 1;
602 10 : last = op(made);
603 : }
604 8 : (last, made)
605 8 : }
606 :
607 : #[test]
608 2 : fn default_is_a_single_attempt() {
609 2 : let p = DeliveryPolicy::default();
610 2 : assert_eq!(p.attempts, 1);
611 2 : assert_eq!(p.next_delay(1, Duration::ZERO), None);
612 2 : let (res, made) = run(p, |_| Err::<(), _>("down"));
613 2 : assert_eq!(made, 1);
614 2 : assert_eq!(res, Err("down"));
615 2 : }
616 :
617 : #[test]
618 2 : fn fails_twice_then_succeeds_is_three_calls_and_one_ok() {
619 2 : let (res, made) = run(
620 2 : policy(3, 100, 60),
621 6 : |n| if n < 3 { Err("down") } else { Ok(()) },
622 : );
623 2 : assert_eq!(made, 3);
624 2 : assert_eq!(res, Ok(()));
625 2 : }
626 :
627 : #[test]
628 2 : fn always_failing_stops_after_attempts_with_the_last_error() {
629 2 : let (res, made) = run(policy(3, 100, 60), Err::<(), _>);
630 2 : assert_eq!(made, 3);
631 2 : assert_eq!(res, Err(3), "the LAST error is returned");
632 2 : }
633 :
634 : #[test]
635 2 : fn max_age_cuts_the_schedule_short() {
636 : // 100 ms, 200 ms, 400 ms … but only 250 ms of age allowed: the
637 : // second retry would land at 300 ms, so it is never made.
638 2 : let (res, made) = run(
639 2 : policy(10, 100, 0).with_max_age(Duration::from_millis(250)),
640 : |_| Err::<(), _>("down"),
641 : );
642 2 : assert_eq!(made, 2);
643 2 : assert!(res.is_err());
644 2 : }
645 :
646 : #[test]
647 2 : fn backoff_doubles_per_retry_and_jitter_stays_within_bounds() {
648 2 : let p = policy(5, 100, 60);
649 2 : assert_eq!(
650 2 : p.next_delay(1, Duration::ZERO),
651 2 : Some(Duration::from_millis(100))
652 : );
653 2 : assert_eq!(
654 2 : p.next_delay(2, Duration::ZERO),
655 2 : Some(Duration::from_millis(200))
656 : );
657 2 : assert_eq!(
658 2 : p.next_delay(3, Duration::ZERO),
659 2 : Some(Duration::from_millis(400))
660 : );
661 2 : let j = DeliveryPolicy { jitter: 0.5, ..p };
662 2 : for _ in 0..50 {
663 100 : let d = j.next_delay(1, Duration::ZERO).expect("granted");
664 100 : assert!(
665 100 : d >= Duration::from_millis(50) && d <= Duration::from_millis(150),
666 : "{d:?}"
667 : );
668 : }
669 2 : let d = j.next_delay(30, Duration::ZERO);
670 2 : assert_eq!(d, None, "attempt 30 of 5 is over");
671 2 : let big = policy(40, 100, 3600);
672 2 : let d = big.next_delay(35, Duration::ZERO).expect("granted");
673 2 : assert!(
674 2 : d <= DeliveryPolicy::MAX_BACKOFF,
675 : "2^34 backoff must not overflow: {d:?}"
676 : );
677 2 : }
678 :
679 : #[test]
680 2 : fn env_parsing_rejects_garbage_and_zero_attempts() {
681 2 : let parse =
682 18 : |a: Option<&str>, b: Option<&str>, m: Option<&str>| DeliveryPolicy::parse(a, b, m);
683 2 : assert_eq!(parse(None, None, None), Ok(DeliveryPolicy::default()));
684 2 : let p = parse(Some("3"), Some("250"), Some("30")).expect("valid");
685 2 : assert_eq!(p.attempts, 3);
686 2 : assert_eq!(p.backoff, Duration::from_millis(250));
687 2 : assert_eq!(p.max_age, Duration::from_secs(30));
688 14 : for bad in [
689 2 : parse(Some("0"), None, None),
690 2 : parse(Some("-1"), None, None),
691 2 : parse(Some("three"), None, None),
692 2 : parse(None, Some("0"), None),
693 2 : parse(None, Some("1e3"), None),
694 2 : parse(None, None, Some("never")),
695 2 : parse(Some("2"), Some(""), None),
696 2 : ] {
697 14 : assert!(bad.is_err(), "{bad:?}");
698 : }
699 2 : }
700 : }
|
