Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Retry policy and dead letters: settle a failed delivery, park it, replay it.
3 :
4 : use super::*;
5 :
6 : pub(super) static DEAD_LETTERS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
7 :
8 : /// Dead letters written by this process since start (`/q/health`
9 : /// deadLetters); the letters themselves live in the store.
10 94 : pub(crate) fn dead_letters_written() -> u64 {
11 94 : DEAD_LETTERS.load(std::sync::atomic::Ordering::Relaxed)
12 94 : }
13 :
14 : /// The retries the delivery policy grants after a failed first attempt,
15 : /// then the settlement: lastSuccess/status ok on success, a dead letter
16 : /// when the policy is exhausted.
17 : #[cfg(not(target_arch = "wasm32"))]
18 : #[allow(clippy::too_many_arguments)] // one param per piece of the attempt's state
19 16 : pub(super) async fn retry_and_settle(
20 16 : st: &AppState,
21 16 : tenant: &TenantId,
22 16 : kind: Kind,
23 16 : sub_id: &str,
24 16 : uri: &str,
25 16 : timeout_ms: u32,
26 16 : outbound: Outbound,
27 16 : first_err: String,
28 16 : ) {
29 16 : let policy = st.delivery;
30 16 : let started = std::time::Instant::now();
31 16 : let first_at = now_iso();
32 16 : let mut made = 1u32;
33 16 : let mut last_err = first_err.clone();
34 24 : while let Some(delay) = policy.next_delay(made, started.elapsed()) {
35 20 : tokio::time::sleep(delay).await;
36 : // the subscription may have gone, or its endpoint may have tripped
37 : // the breaker meanwhile — a retry is still one more attempt
38 16 : if st
39 16 : .store
40 16 : .get(tenant, kind, sub_id)
41 16 : .await
42 16 : .ok()
43 16 : .flatten()
44 16 : .is_none()
45 : {
46 4 : return;
47 12 : }
48 12 : made = made.saturating_add(1);
49 12 : match send_outbound(st, tenant, uri, timeout_ms, &outbound).await {
50 : Ok(()) => {
51 4 : st.egress.record_success(tenant.as_str(), uri);
52 4 : metrics::counter!("antares_notifications_retried_total", "outcome" => "ok")
53 4 : .increment(1);
54 4 : let ts = now_iso();
55 4 : let mut retried_doc: Option<Value> = None;
56 4 : st.store
57 4 : .mutate(tenant, kind, sub_id, |doc| {
58 4 : if let Some(o) = doc.as_object_mut() {
59 4 : o.remove("status");
60 4 : }
61 4 : if let Some(n) = doc
62 4 : .as_object_mut()
63 4 : .and_then(|o| o.get_mut("notification"))
64 4 : .and_then(Value::as_object_mut)
65 4 : {
66 4 : n.insert("lastSuccess".into(), Value::String(ts.clone()));
67 4 : n.insert("status".into(), Value::String("ok".into()));
68 4 : }
69 4 : retried_doc = Some(doc.clone());
70 4 : Ok::<(), antares_model::NgsiError>(())
71 4 : })
72 4 : .await
73 4 : .unwrap_or_else(|e| {
74 0 : tracing::warn!("retry bookkeeping writeback failed: {e}");
75 0 : None
76 0 : });
77 4 : mirror_bookkeeping(st, tenant, kind, sub_id, retried_doc);
78 4 : return;
79 : }
80 8 : Err((timed_out, e)) => {
81 8 : if timed_out {
82 0 : st.egress.record_failure(tenant.as_str(), uri);
83 8 : } else {
84 8 : st.egress.record_success(tenant.as_str(), uri);
85 8 : }
86 8 : last_err = e;
87 : }
88 : }
89 : }
90 4 : metrics::counter!("antares_notifications_retried_total", "outcome" => "dead").increment(1);
91 4 : let letter = dead_letter(
92 4 : sub_id, uri, timeout_ms, &outbound, made, &first_err, &last_err, &first_at,
93 : );
94 4 : let id = letter
95 4 : .get("id")
96 4 : .and_then(Value::as_str)
97 4 : .unwrap_or_default()
98 4 : .to_owned();
99 4 : match st.store.create(tenant, Kind::DeadLetter, &id, letter).await {
100 : Ok(_) => {
101 4 : DEAD_LETTERS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
102 4 : tracing::warn!(
103 : "notification for {sub_id} to {} dead-lettered after {made} attempts: {last_err}",
104 0 : redact_userinfo(uri)
105 : );
106 : }
107 0 : Err(e) => tracing::error!("dead letter for {sub_id} could not be stored: {e}"),
108 : }
109 12 : }
110 :
111 : /// The dead-letter document: everything a replay needs to send the very
112 : /// same request again, plus the attempt history. Written by the native
113 : /// retry loop (`retry_and_settle`) alone.
114 : #[cfg(not(target_arch = "wasm32"))]
115 : #[allow(clippy::too_many_arguments)] // one param per stored letter member
116 4 : pub(super) fn dead_letter(
117 4 : sub_id: &str,
118 4 : uri: &str,
119 4 : timeout_ms: u32,
120 4 : outbound: &Outbound,
121 4 : attempts: u32,
122 4 : first_err: &str,
123 4 : last_err: &str,
124 4 : first_at: &str,
125 4 : ) -> Value {
126 4 : json!({
127 4 : "id": format!("urn:ngsi-ld:DeadLetter:{}", uuid::Uuid::new_v4()),
128 4 : "type": "DeadLetter",
129 4 : "subscriptionId": sub_id,
130 4 : "uri": uri,
131 4 : "timeoutMs": timeout_ms,
132 4 : "attempts": attempts,
133 4 : "firstError": first_err,
134 4 : "lastError": last_err,
135 4 : "firstAt": first_at,
136 4 : "lastAt": now_iso(),
137 4 : "binding": antares_notifier::SinkRegistry::scheme_of(uri),
138 4 : "payload": outbound.body.clone(),
139 4 : "accept": outbound.accept,
140 4 : "link": outbound.link,
141 4 : "receiverInfo": outbound.receiver_info,
142 4 : "notifierInfo": outbound.notifier_info,
143 : })
144 4 : }
145 :
146 : /// The deadline a dead letter was written with. It is read back from
147 : /// storage, so a value past u32 caps at the 30 s ceiling of
148 : /// [`endpoint_timeout_ms`] instead of wrapping to a short deadline.
149 26 : pub(super) fn letter_timeout_ms(letter: &Value) -> u32 {
150 26 : letter["timeoutMs"]
151 26 : .as_u64()
152 26 : .map_or(5_000, |t| u32::try_from(t.min(30_000)).unwrap_or(30_000))
153 26 : }
154 :
155 : /// Replay one dead letter through the same binding, once. `Ok` = delivered
156 : /// (the caller deletes the letter); `Err` carries the failure text.
157 16 : pub(crate) async fn replay_dead_letter(
158 16 : st: &AppState,
159 16 : tenant: &TenantId,
160 16 : letter: &Value,
161 16 : ) -> Result<(), String> {
162 16 : let uri = letter["uri"].as_str().ok_or("dead letter without uri")?;
163 14 : let timeout_ms = letter_timeout_ms(letter);
164 14 : let outbound = Outbound::from_dead_letter(letter)?;
165 : // the egress policy of the moment applies, exactly as for a fresh send
166 12 : if in_process(st, uri).is_none() && st.sinks.sink_for_uri(uri).is_some_and(|s| s.network()) {
167 6 : st.egress
168 6 : .check_destination(uri)
169 6 : .await
170 6 : .map_err(|e| e.to_string())?;
171 6 : }
172 10 : send_outbound(st, tenant, uri, timeout_ms, &outbound)
173 10 : .await
174 10 : .map_err(|(_, e)| e)
175 16 : }
|
