Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! The temporal seam's producer side (ADR-0013): the write path pushes
3 : //! `TemporalEvent`s into a per-request buffer; the buffer is drained ONCE
4 : //! per request — after the handler, before the response leaves — so the
5 : //! driver sees the whole request in one `event_list` call and a client
6 : //! reading its own history right after the write always finds it.
7 : //!
8 : //! Outside a request (background jobs, tests without the router) there is
9 : //! no buffer, and a push drains immediately: the seam degrades to today's
10 : //! per-change recording rather than losing events. A driver error in the
11 : //! drain is logged and counted (`/q/health` temporalDrainErrors) — it never
12 : //! changes the response of a write that already committed.
13 :
14 : use crate::state::{AppState, RecordMode};
15 : use antares_model::TenantId;
16 : use antares_store::{TemporalDriverExt as _, TemporalEvent};
17 : use serde_json::{Map, Value};
18 : use std::sync::atomic::{AtomicU64, Ordering};
19 : use std::time::Duration;
20 :
21 : static DRAIN_ERRORS: AtomicU64 = AtomicU64::new(0);
22 :
23 : /// Drains that failed in the driver (the events of that request are lost;
24 : /// the write itself stood).
25 102 : pub fn drain_errors() -> u64 {
26 102 : DRAIN_ERRORS.load(Ordering::Relaxed)
27 102 : }
28 :
29 : // The buffer rides the request's task; tokio task-locals need the `rt`
30 : // feature the single-threaded wasm build does not carry, so wasm records
31 : // immediately.
32 : #[cfg(not(target_arch = "wasm32"))]
33 : tokio::task_local! {
34 : static BUFFER: std::cell::RefCell<Vec<TemporalEvent>>;
35 : static STAGED: std::cell::RefCell<Vec<TemporalEvent>>;
36 : static CHANGES: std::cell::RefCell<Vec<crate::mirror::Change>>;
37 : static PROVENANCE: Value;
38 : static TENANT: Option<TenantId>;
39 : }
40 :
41 : /// `ANTARES_HISTORY_BACKFILL=on`: a Core API write of
42 : /// an instance older than the one the Entity holds goes to history only.
43 : /// Off (the default) keeps NGSI-LD's replace, which the ETSI suites assume.
44 51572 : fn backfill_on() -> bool {
45 : static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
46 51572 : *ON.get_or_init(|| std::env::var("ANTARES_HISTORY_BACKFILL").as_deref() == Ok("on"))
47 51572 : }
48 :
49 : /// Whether `incoming`, about to replace `current` of `attr` on the Entity
50 : /// `doc` (expanded, as stored), is a backfill: both observed and `incoming`
51 : /// the older. A backfill is recorded into the request's history here, and
52 : /// the caller keeps `current`. Outside a request, or with backfill off,
53 : /// nothing is one.
54 51572 : pub(crate) fn backfilled(doc: &Value, attr: &str, current: &Value, incoming: &Value) -> bool {
55 51572 : if !backfill_on() {
56 51562 : return false;
57 10 : }
58 20 : let at = |i: &Value| {
59 20 : i.get("observedAt").and_then(Value::as_str).and_then(|s| {
60 20 : chrono::DateTime::parse_from_rfc3339(&antares_store::filter::canonical_datetime(s)).ok()
61 20 : })
62 20 : };
63 10 : let (Some(was), Some(now)) = (at(current), at(incoming)) else {
64 0 : return false;
65 : };
66 10 : let Some(id) = doc.get("id").and_then(Value::as_str) else {
67 0 : return false;
68 : };
69 10 : if now >= was {
70 2 : return false;
71 8 : }
72 : #[cfg(target_arch = "wasm32")]
73 : let _ = (attr, id);
74 : #[cfg(not(target_arch = "wasm32"))]
75 : {
76 8 : let Ok(Some(tenant)) = TENANT.try_with(Clone::clone) else {
77 0 : return false;
78 : };
79 8 : let mut instance = incoming.clone();
80 8 : if let Some(o) = instance.as_object_mut() {
81 8 : let iid = crate::notify::instance_id(id, attr, o);
82 8 : o.entry("instanceId".to_owned())
83 8 : .or_insert(Value::String(iid));
84 8 : o.insert(
85 8 : antares_store::PROVENANCE_KEY.to_owned(),
86 8 : provenance("backfill"),
87 8 : );
88 8 : }
89 8 : let mut shell = Map::new();
90 40 : for k in ["id", "type", "createdAt", "modifiedAt", "scope"] {
91 40 : if let Some(v) = doc.get(k) {
92 16 : shell.insert(k.to_owned(), v.clone());
93 24 : }
94 : }
95 8 : let event = TemporalEvent {
96 8 : op: antares_store::TemporalOp::AttrModified,
97 8 : tenant,
98 8 : entity_id: id.to_owned(),
99 8 : shell: Value::Object(shell),
100 8 : attr: attr.to_owned(),
101 8 : instance,
102 8 : };
103 8 : stage(event)
104 : }
105 : #[cfg(target_arch = "wasm32")]
106 : false
107 51572 : }
108 :
109 : /// Holds a backfill's event until its write is known to commit: the
110 : /// closure that found it may still be refused or rerun, and
111 : /// [`settle`] buffers it only then. `false` outside a request.
112 : #[cfg(not(target_arch = "wasm32"))]
113 24 : fn stage(event: TemporalEvent) -> bool {
114 24 : STAGED.try_with(|b| b.borrow_mut().push(event)).is_ok()
115 24 : }
116 :
117 : /// A read-modify-write of Entity `id` (re)starts: the backfills an earlier
118 : /// run of its closure staged are void. Called first thing in
119 : /// every closure that can reach [`backfilled`].
120 13952 : pub(crate) fn restage(id: &str) {
121 : #[cfg(not(target_arch = "wasm32"))]
122 13952 : let _ = STAGED.try_with(|b| b.borrow_mut().retain(|e| e.entity_id != id));
123 : #[cfg(target_arch = "wasm32")]
124 : let _ = id;
125 13952 : }
126 :
127 : /// The read-modify-write of Entity `id` returned: its staged backfills go
128 : /// to the request's history when the write committed, and are dropped when
129 : /// it was refused or found nothing.
130 14050 : pub(crate) fn settle(id: &str, committed: bool) {
131 : #[cfg(not(target_arch = "wasm32"))]
132 : {
133 14050 : let mine = STAGED
134 14050 : .try_with(|b| {
135 14050 : let mut b = b.borrow_mut();
136 14050 : let (mine, rest) = std::mem::take(&mut *b)
137 14050 : .into_iter()
138 14050 : .partition::<Vec<_>, _>(|e| e.entity_id == id);
139 14050 : *b = rest;
140 14050 : mine
141 14050 : })
142 14050 : .unwrap_or_default();
143 14050 : if committed && !mine.is_empty() {
144 12 : let _ = BUFFER.try_with(|b| b.borrow_mut().extend(mine));
145 14038 : }
146 : }
147 : #[cfg(target_arch = "wasm32")]
148 : let _ = (id, committed);
149 14050 : }
150 :
151 : /// [`settle`] for each id of a batch read-modify-write, by its slot.
152 172 : pub(crate) fn settle_all<T, E>(ids: &[String], res: &[Option<Result<T, E>>]) {
153 288 : for (id, r) in ids.iter().zip(res) {
154 288 : settle(id, matches!(r, Some(Ok(_))));
155 : }
156 172 : }
157 :
158 : /// The provenance of the request in flight with its `operation` and
159 : /// `writtenAt`, when the broker stored it; outside a
160 : /// request, a write of the broker's own: `api`.
161 132004 : pub(crate) fn provenance(operation: &str) -> Value {
162 : #[cfg(not(target_arch = "wasm32"))]
163 132004 : let request = PROVENANCE.try_with(Clone::clone).ok();
164 : #[cfg(target_arch = "wasm32")]
165 : let request: Option<Value> = None;
166 132004 : let base = request.unwrap_or_else(|| serde_json::json!({"channel": "api"}));
167 132004 : let mut out = crate::provenance::with_operation(&base, operation);
168 132004 : if let Some(o) = out.as_object_mut() {
169 132004 : let now = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true);
170 132004 : o.insert("writtenAt".into(), Value::String(now));
171 132004 : }
172 132004 : out
173 132004 : }
174 :
175 : /// Whether the request in flight is audited: each change is then kept as a version
176 : /// whatever the type's history rule says.
177 319478 : fn audited() -> bool {
178 : #[cfg(not(target_arch = "wasm32"))]
179 319478 : return PROVENANCE
180 319478 : .try_with(crate::provenance::audited)
181 319478 : .unwrap_or(false);
182 : #[cfg(target_arch = "wasm32")]
183 : false
184 319478 : }
185 :
186 : /// Buffer one entity change for the request in flight so the matcher
187 : /// receives the whole request at once. Handed back when no request is in
188 : /// flight — the caller then gives it to the matcher on the spot.
189 140898 : pub(crate) fn buffer_change(change: crate::mirror::Change) -> Option<crate::mirror::Change> {
190 : #[cfg(not(target_arch = "wasm32"))]
191 : {
192 140898 : let mut slot = Some(change);
193 140898 : let _ = CHANGES.try_with(|b| {
194 27868 : if let Some(c) = slot.take() {
195 27868 : b.borrow_mut().push(c);
196 27868 : }
197 27868 : });
198 140898 : slot
199 : }
200 : #[cfg(target_arch = "wasm32")]
201 : Some(change)
202 140898 : }
203 :
204 : /// Hand one event to the seam: buffered when a request is in flight,
205 : /// drained on the spot otherwise.
206 131624 : pub(crate) async fn push(st: &AppState, ev: TemporalEvent) {
207 : #[cfg(not(target_arch = "wasm32"))]
208 112394 : let ev = {
209 131624 : let mut slot = Some(ev);
210 131624 : let buffered = BUFFER
211 131624 : .try_with(|b| {
212 19230 : if let Some(ev) = slot.take() {
213 19230 : b.borrow_mut().push(ev);
214 19230 : }
215 19230 : })
216 131624 : .is_ok();
217 131624 : if buffered {
218 19230 : return;
219 112394 : }
220 112394 : match slot {
221 112394 : Some(ev) => ev,
222 0 : None => return,
223 : }
224 : };
225 112394 : drain(st, vec![ev]).await;
226 131624 : }
227 :
228 : /// The gate chain: an event enters history only if every gate admits it.
229 : /// Gate 1 (value-change) runs in the producer — an unchanged instance never
230 : /// becomes an event (`changed_instances`) — and so does gate 3, retention
231 : /// ([`records_history`]), which needs the whole Entity. Adding a gate = one more entry
232 : /// here; producers and drivers stay untouched.
233 : const GATES: &[fn(&AppState, &TemporalEvent) -> bool] = &[observed_gate];
234 :
235 : /// Gate 2: ANTARES_TEMPORAL_RECORD. `all` admits everything; `observed`
236 : /// keeps only instances that carry `observedAt` — the spec's own
237 : /// measurement axis (4.5.7: observedAt is the default timeproperty), so
238 : /// metadata-shaped writes leave no history; `none` admits nothing.
239 131632 : fn observed_gate(st: &AppState, ev: &TemporalEvent) -> bool {
240 : use crate::state::TemporalRecord::*;
241 : // An audited write is kept without `observedAt` too.
242 131632 : let audited = || {
243 20 : ev.instance
244 20 : .get(antares_store::PROVENANCE_KEY)
245 20 : .is_some_and(crate::provenance::audited)
246 20 : };
247 131632 : match st.temporal_record {
248 131570 : All | Ruled => true,
249 48 : Observed => ev.instance.get("observedAt").is_some() || audited(),
250 14 : None => false,
251 : }
252 131632 : }
253 :
254 : /// An Entity's own history retention, read from its
255 : /// [`antares_store::HISTORY_RETENTION`] Property.
256 : #[derive(Debug, Clone, Copy, PartialEq, Eq)]
257 : pub enum OwnRetention {
258 : /// `off`: the Entity keeps its current state only.
259 : Off,
260 : /// A duration: its history is kept that long, whatever its type says.
261 : Keep(Duration),
262 : }
263 :
264 : /// What the Entity document `doc` (expanded, as stored) says of its own
265 : /// history: `None` when it carries no `historyRetention`, `Some(Err)` with
266 : /// the reason when the value is neither a duration nor `off` (a typo,
267 : /// `forever`), which the rules of its type then override.
268 137612 : pub fn own_retention(doc: &Value) -> Option<Result<OwnRetention, String>> {
269 137612 : doc.get(antares_store::HISTORY_RETENTION).map(retention_of)
270 137612 : }
271 :
272 : /// What one stored `historyRetention` Property (one instance, or an array
273 : /// whose first instance counts) says.
274 : ///
275 : /// # Errors
276 : /// Why it is neither a duration nor `off`.
277 36 : pub fn retention_of(own: &Value) -> Result<OwnRetention, String> {
278 36 : let first = match own {
279 8 : Value::Array(all) => all.first(),
280 28 : one => Some(one),
281 : };
282 36 : let Some(raw) = first.and_then(|p| p.get("value")).and_then(Value::as_str) else {
283 8 : return Err("historyRetention is not a Property with a string value".into());
284 : };
285 28 : if raw == "off" {
286 8 : Ok(OwnRetention::Off)
287 : } else {
288 20 : parse_duration(raw).map(OwnRetention::Keep)
289 : }
290 36 : }
291 :
292 : /// Gate 3, run in the producer where the whole Entity is at
293 : /// hand: whether a write of `doc` in `tenant` records history. The most
294 : /// specific level wins: the Entity's own `historyRetention`, then the rule
295 : /// of one of its types (any `off` among them records nothing), then the
296 : /// Tenant's `*` rule. Below them all, `ANTARES_TEMPORAL_RECORD=ruled`
297 : /// records nothing (history is opt-in), any other mode everything.
298 137596 : pub(crate) fn records_history(st: &AppState, tenant: &TenantId, doc: &Value) -> bool {
299 137596 : if audited() {
300 0 : return true;
301 137596 : }
302 137596 : match own_retention(doc) {
303 4 : Some(Ok(OwnRetention::Off)) => return false,
304 8 : Some(Ok(OwnRetention::Keep(_))) => return true,
305 137584 : Some(Err(_)) | None => {}
306 : }
307 137584 : let rules = st
308 137584 : .history_off
309 137584 : .read()
310 137584 : .unwrap_or_else(std::sync::PoisonError::into_inner);
311 137584 : let opt_in = st.temporal_record == crate::state::TemporalRecord::Ruled;
312 137584 : if rules.is_empty() && !opt_in {
313 137522 : return true;
314 62 : }
315 62 : let tenant = tenant.as_str();
316 62 : let types: Vec<&str> = match doc.get("type") {
317 40 : Some(Value::String(t)) => vec![t.as_str()],
318 18 : Some(Value::Array(ts)) => ts.iter().filter_map(Value::as_str).collect(),
319 4 : _ => Vec::new(),
320 : };
321 112 : let has = |set: &std::collections::HashSet<(String, String)>| {
322 112 : types
323 112 : .iter()
324 112 : .any(|t| set.contains(&(tenant.to_owned(), (*t).to_owned())))
325 112 : };
326 62 : if has(&rules.off) {
327 12 : return false;
328 50 : }
329 50 : if opt_in {
330 30 : return has(&rules.ruled) || rules.space_on.contains(tenant);
331 20 : }
332 20 : has(&rules.ruled) || !rules.space_off.contains(tenant)
333 137596 : }
334 :
335 : /// The last row each attribute recorded `onChange` wrote, by (Tenant,
336 : /// Entity, attribute, datasetId).
337 : type LastRows = std::collections::HashMap<(String, String, String, String), LastRow>;
338 :
339 : /// An attribute's newest row, and the last late row written below it.
340 : struct LastRow {
341 : payload: Value,
342 : at: chrono::DateTime<chrono::Utc>,
343 : /// A late instance (older than `at`) is judged against the last late row,
344 : /// or against `at` before there is one, so a replayed or retried batch
345 : /// of older seconds records one row per heartbeat, not one per second
346 : /// (a Vehicle wrote one row in 4.5 s under a 30 s heartbeat).
347 : late: Option<(Value, chrono::DateTime<chrono::Utc>)>,
348 : }
349 : static LAST_ROWS: std::sync::LazyLock<std::sync::Mutex<LastRows>> =
350 : std::sync::LazyLock::new(Default::default);
351 : /// The most attributes [`LAST_ROWS`] remembers.
352 : // One map per process, cleared when full or restarted; the next
353 : // write of each attribute then records a row, which costs a row, never one
354 : // lost. A shared store if replicas ever record far more than one row per
355 : // heartbeat between them.
356 : const LAST_ROWS_MAX: usize = 500_000;
357 :
358 : /// The Attributes of the temporal `doc` its type records `onChange`:
359 : /// a query carries their level into a window and weighs
360 : /// their values by the time each held.
361 2054 : pub(crate) fn held_attrs(
362 2054 : st: &AppState,
363 2054 : tenant: &TenantId,
364 2054 : doc: &Value,
365 2054 : ) -> std::collections::HashSet<String> {
366 2054 : let rules = st
367 2054 : .history_off
368 2054 : .read()
369 2054 : .unwrap_or_else(std::sync::PoisonError::into_inner);
370 2054 : let (false, Some(obj)) = (rules.recording.is_empty(), doc.as_object()) else {
371 1902 : return Default::default();
372 : };
373 152 : let types: Vec<&str> = match obj.get("type") {
374 8 : Some(Value::String(t)) => vec![t.as_str()],
375 144 : Some(Value::Array(ts)) => ts.iter().filter_map(Value::as_str).collect(),
376 0 : _ => Vec::new(),
377 : };
378 152 : let recordings: Vec<_> = types
379 152 : .iter()
380 156 : .filter_map(|t| {
381 156 : rules
382 156 : .recording
383 156 : .get(&(tenant.as_str().to_owned(), (*t).to_owned()))
384 156 : })
385 152 : .collect();
386 152 : obj.keys()
387 1058 : .filter(|k| {
388 1058 : recordings
389 1058 : .iter()
390 1058 : .any(|r| matches!(r.of(k), Some(RecordMode::OnChange(_))))
391 1058 : })
392 152 : .cloned()
393 152 : .collect()
394 2054 : }
395 :
396 : /// Does some type of `tenant` record an Attribute `onChange`? Its
397 : /// aggregations are then computed here, never pushed into the store.
398 1970 : pub(crate) fn records_on_change(st: &AppState, tenant: &TenantId) -> bool {
399 1970 : st.history_off
400 1970 : .read()
401 1970 : .unwrap_or_else(std::sync::PoisonError::into_inner)
402 1970 : .recording
403 1970 : .iter()
404 1970 : .any(|((t, _), r)| t == tenant.as_str() && r.has_on_change())
405 1970 : }
406 :
407 : /// Gate 4: whether an instance of `attr`, changed
408 : /// on the Entity `entity` with `types` in `tenant`, writes a row. With no
409 : /// rule the mode is `every`, or `observed` under `ruled`. `every`
410 : /// always does, `never` never; `onChange` when its value moved past the
411 : /// deadband or minimum distance since the attribute's last row, when it
412 : /// differs at all from a row of the same `observedAt` (a correction), or
413 : /// once the heartbeat passed. An instance older than that row (a backfill,
414 : /// a retried batch) is judged the same way against the nearest row this
415 : /// process knows below it.
416 : /// `observed` when its `observedAt` is newer than that of the instance it
417 : /// replaces (`before`, the attribute as the Entity held it).
418 181882 : pub(crate) fn admits_change(
419 181882 : st: &AppState,
420 181882 : tenant: &TenantId,
421 181882 : types: &[&str],
422 181882 : entity: &str,
423 181882 : attr: &str,
424 181882 : before: Option<&Value>,
425 181882 : instance: &Value,
426 181882 : ) -> bool {
427 : // An audited write keeps every change; gate 1 already dropped
428 : // an exact repeat.
429 181882 : if audited() {
430 0 : return true;
431 181882 : }
432 181882 : let default = if st.temporal_record == crate::state::TemporalRecord::Ruled {
433 22 : RecordMode::Observed
434 : } else {
435 181860 : RecordMode::Every
436 : };
437 181882 : let mode = {
438 181882 : let rules = st
439 181882 : .history_off
440 181882 : .read()
441 181882 : .unwrap_or_else(std::sync::PoisonError::into_inner);
442 182690 : types.iter().find_map(|t| {
443 182690 : rules
444 182690 : .recording
445 182690 : .get(&(tenant.as_str().to_owned(), (*t).to_owned()))
446 182690 : .and_then(|r| r.of(attr))
447 182690 : })
448 : };
449 181882 : let mode = match mode.unwrap_or(default) {
450 130050 : RecordMode::Every => return true,
451 4 : RecordMode::Never => return false,
452 46 : RecordMode::Observed => return newly_observed(before, instance),
453 51782 : RecordMode::OnChange(mode) => mode,
454 : };
455 51782 : let at = observed_at(instance).unwrap_or_else(chrono::Utc::now);
456 51782 : let payload = payload_of(instance);
457 51782 : let key = (
458 51782 : tenant.as_str().to_owned(),
459 51782 : entity.to_owned(),
460 51782 : attr.to_owned(),
461 51782 : instance
462 51782 : .get("datasetId")
463 51782 : .and_then(Value::as_str)
464 51782 : .unwrap_or_default()
465 51782 : .to_owned(),
466 51782 : );
467 51782 : let mut last = LAST_ROWS
468 51782 : .lock()
469 51782 : .unwrap_or_else(std::sync::PoisonError::into_inner);
470 51782 : let admits = |was: &Value, when: chrono::DateTime<chrono::Utc>| {
471 51744 : if at == when {
472 10 : return payload != *was;
473 51734 : }
474 51734 : let apart = at.signed_duration_since(when).abs();
475 51734 : apart.to_std().is_ok_and(|d| d >= mode.heartbeat) || moved(&mode, was, &payload)
476 51744 : };
477 51782 : match last.get_mut(&key) {
478 51744 : Some(row) if at < row.at => {
479 328 : let (was, when) = row
480 328 : .late
481 328 : .as_ref()
482 328 : .map_or((&row.payload, row.at), |(p, t)| (p, *t));
483 328 : let writes = admits(was, when);
484 328 : if writes {
485 16 : row.late = Some((payload, at));
486 312 : }
487 328 : writes
488 : }
489 51416 : Some(row) if !admits(&row.payload, row.at) => false,
490 : _ => {
491 1580 : if last.len() >= LAST_ROWS_MAX {
492 0 : last.clear();
493 1580 : }
494 1580 : last.insert(
495 1580 : key,
496 1580 : LastRow {
497 1580 : payload,
498 1580 : at,
499 1580 : late: None,
500 1580 : },
501 1580 : );
502 1580 : true
503 : }
504 : }
505 181882 : }
506 :
507 : /// The instant `instance` was observed at, if it says.
508 51858 : fn observed_at(instance: &Value) -> Option<chrono::DateTime<chrono::Utc>> {
509 51858 : let s = instance.get("observedAt")?.as_str()?;
510 51854 : chrono::DateTime::parse_from_rfc3339(&antares_store::filter::canonical_datetime(s))
511 51854 : .ok()
512 51854 : .map(|t| t.with_timezone(&chrono::Utc))
513 51858 : }
514 :
515 : /// `record: observed`: whether `instance` carries an `observedAt` newer than
516 : /// the instance of its datasetId in `before` (an array or one instance).
517 46 : fn newly_observed(before: Option<&Value>, instance: &Value) -> bool {
518 46 : let Some(now) = observed_at(instance) else {
519 4 : return false;
520 : };
521 68 : let ds = |i: &Value| i.get("datasetId").cloned();
522 42 : let was = match before {
523 30 : Some(Value::Array(all)) => all.iter().find(|b| ds(b) == ds(instance)),
524 4 : Some(one) if ds(one) == ds(instance) => Some(one),
525 8 : _ => None,
526 : };
527 42 : was.and_then(observed_at).is_none_or(|was| now > was)
528 46 : }
529 :
530 : /// An instance without its stamps and provenance: what a row records.
531 51782 : fn payload_of(instance: &Value) -> Value {
532 51782 : let mut p = instance.clone();
533 51782 : if let Some(o) = p.as_object_mut() {
534 258910 : for k in [
535 51782 : "instanceId",
536 51782 : "observedAt",
537 51782 : "createdAt",
538 51782 : "modifiedAt",
539 51782 : antares_store::PROVENANCE_KEY,
540 258910 : ] {
541 258910 : o.remove(k);
542 258910 : }
543 0 : }
544 51782 : p
545 51782 : }
546 :
547 : /// Whether `now` moved from `was` enough to be a row under `mode`.
548 50710 : fn moved(mode: &crate::state::OnChange, was: &Value, now: &Value) -> bool {
549 101420 : let number = |v: &Value| v.get("value").and_then(Value::as_f64);
550 50710 : if let (Some(band), Some(a), Some(b)) = (mode.deadband, number(was), number(now)) {
551 968 : return (a - b).abs() > band;
552 49742 : }
553 99484 : let point = |v: &Value| {
554 99484 : let g = v.get("value")?;
555 99484 : let c = g.get("coordinates")?.as_array()?;
556 16 : (g.get("type")?.as_str()? == "Point").then_some((c.first()?.as_f64()?, c.get(1)?.as_f64()?))
557 99484 : };
558 49742 : if let (Some(metres), Some(a), Some(b)) = (mode.min_distance, point(was), point(now)) {
559 8 : return haversine(a, b) > metres;
560 49734 : }
561 49734 : was != now
562 50710 : }
563 :
564 : /// Metres between two (longitude, latitude) points on a sphere of the
565 : /// Earth's mean radius: within a fraction of a percent at a city's scale.
566 8 : fn haversine((lon1, lat1): (f64, f64), (lon2, lat2): (f64, f64)) -> f64 {
567 8 : let (p1, p2) = (lat1.to_radians(), lat2.to_radians());
568 8 : let (dp, dl) = ((lat2 - lat1).to_radians(), (lon2 - lon1).to_radians());
569 8 : let a = (dp / 2.0).sin().powi(2) + p1.cos() * p2.cos() * (dl / 2.0).sin().powi(2);
570 8 : 2.0 * 6_371_008.8 * a.sqrt().asin()
571 8 : }
572 :
573 : /// The bounds a duration has to sit in: one minute to ten years.
574 : const SHORTEST: Duration = Duration::from_secs(60);
575 : const LONGEST: Duration = Duration::from_hours(87_600);
576 :
577 : /// A duration as ISO 8601 (`PT5M`, `PT8H`, `P30D`, `P1DT12H`) or in short
578 : /// form (`90s`, `5m`, `8h`, `14d`), between one minute and ten years.
579 : ///
580 : /// # Errors
581 : /// Why `raw` is not one.
582 100 : pub fn parse_duration(raw: &str) -> Result<Duration, String> {
583 100 : parse_duration_within(raw, SHORTEST, LONGEST)
584 100 : }
585 :
586 : /// A duration as [`parse_duration`] reads it, between `shortest` and
587 : /// `longest` (the quality sweep's `maxAge` starts at one second).
588 : ///
589 : /// # Errors
590 : /// Why `raw` is not one.
591 100 : pub fn parse_duration_within(
592 100 : raw: &str,
593 100 : shortest: Duration,
594 100 : longest: Duration,
595 100 : ) -> Result<Duration, String> {
596 100 : let bad = || format!("{raw:?} is not a duration such as PT5M, PT8H, P30D, 5m, 8h or 14d");
597 100 : let secs = if let Some(iso) = raw.strip_prefix('P') {
598 60 : let (days, time) = match iso.split_once('T') {
599 36 : Some((d, t)) if !t.is_empty() => (d, Some(t)),
600 8 : Some(_) => return Err(bad()),
601 24 : None => (iso, None),
602 : };
603 52 : let mut total = 0u64;
604 52 : if !days.is_empty() {
605 24 : total = days
606 24 : .strip_suffix('D')
607 24 : .and_then(|n| n.parse::<u64>().ok())
608 24 : .ok_or_else(bad)?
609 20 : .checked_mul(86_400)
610 20 : .ok_or_else(bad)?;
611 28 : } else if time.is_none() {
612 4 : return Err(bad());
613 24 : }
614 44 : let mut rest = time.unwrap_or_default();
615 128 : for (unit, scale) in [('H', 3600u64), ('M', 60), ('S', 1)] {
616 128 : if let Some((n, tail)) = rest.split_once(unit) {
617 32 : let n = n.parse::<u64>().map_err(|_| bad())?;
618 28 : total = n
619 28 : .checked_mul(scale)
620 28 : .and_then(|n| total.checked_add(n))
621 28 : .ok_or_else(bad)?;
622 28 : rest = tail;
623 96 : }
624 : }
625 40 : if !rest.is_empty() {
626 4 : return Err(bad());
627 36 : }
628 36 : total
629 : } else {
630 40 : let split = raw.len().checked_sub(1).ok_or_else(bad)?;
631 36 : let (n, unit) = (
632 36 : raw.get(..split).ok_or_else(bad)?,
633 36 : raw.get(split..).ok_or_else(bad)?,
634 : );
635 36 : let scale = match unit {
636 36 : "s" => 1u64,
637 32 : "m" => 60,
638 24 : "h" => 3600,
639 20 : "d" => 86_400,
640 16 : _ => return Err(bad()),
641 : };
642 20 : n.parse::<u64>()
643 20 : .ok()
644 20 : .and_then(|n| n.checked_mul(scale))
645 20 : .ok_or_else(bad)?
646 : };
647 52 : let d = Duration::from_secs(secs);
648 52 : if d < shortest || d > longest {
649 8 : return Err(format!(
650 8 : "{raw:?} is outside {}s to {}s",
651 8 : shortest.as_secs(),
652 8 : longest.as_secs()
653 8 : ));
654 44 : }
655 44 : Ok(d)
656 100 : }
657 :
658 : /// The consumer side: one `event_list` call per drained batch.
659 139660 : pub(crate) async fn drain(st: &AppState, evs: Vec<TemporalEvent>) {
660 139660 : let evs: Vec<TemporalEvent> = evs
661 139660 : .into_iter()
662 140148 : .filter(|ev| GATES.iter().all(|gate| gate(st, ev)))
663 139660 : .collect();
664 139660 : if evs.is_empty() {
665 16176 : return;
666 123484 : }
667 123484 : if let Err(e) = st.temporal.event_list(&evs).await {
668 4 : DRAIN_ERRORS.fetch_add(1, Ordering::Relaxed);
669 4 : metrics::counter!("antares_temporal_drain_errors_total").increment(1);
670 4 : tracing::warn!(events = evs.len(), "temporal drain failed: {e}");
671 123480 : }
672 139660 : }
673 :
674 : /// What one request buffered: its history events and its changes for the
675 : /// matcher. They stand for writes that already committed, so they are
676 : /// handed on however the request ends — a handler that panicked, or a
677 : /// request whose future was dropped (a client that went away) still leaves
678 : /// its history and its notifications, which dropping the buffers with the
679 : /// task-locals lost.
680 : #[cfg(not(target_arch = "wasm32"))]
681 : struct Buffered {
682 : st: AppState,
683 : evs: Vec<TemporalEvent>,
684 : changes: Vec<crate::mirror::Change>,
685 : }
686 :
687 : #[cfg(not(target_arch = "wasm32"))]
688 : impl Buffered {
689 : /// Takes what the request's buffers hold; called inside their scope.
690 110068 : fn collect(&mut self) {
691 110068 : if let Ok(evs) = BUFFER.try_with(std::cell::RefCell::take) {
692 55038 : self.evs.extend(evs);
693 55038 : }
694 110068 : if let Ok(changes) = CHANGES.try_with(std::cell::RefCell::take) {
695 55038 : self.changes.extend(changes);
696 55038 : }
697 110068 : }
698 :
699 : /// The drain and the hand-over to the matcher as a task of their own:
700 : /// once started they finish whether or not anybody still waits for them,
701 : /// so a request dropped in the middle of the drain loses neither half.
702 : /// `None` when there is nothing to hand on.
703 110068 : fn hand_on(&mut self) -> Option<tokio::task::JoinHandle<()>> {
704 110068 : if self.evs.is_empty() && self.changes.is_empty() {
705 82802 : return None;
706 27266 : }
707 27266 : let Ok(rt) = tokio::runtime::Handle::try_current() else {
708 0 : tracing::warn!(
709 0 : events = self.evs.len(),
710 0 : changes = self.changes.len(),
711 : "no runtime left to record a request's history"
712 : );
713 0 : return None;
714 : };
715 27266 : let (st, evs, changes) = (
716 27266 : self.st.clone(),
717 27266 : std::mem::take(&mut self.evs),
718 27266 : std::mem::take(&mut self.changes),
719 27266 : );
720 27266 : Some(rt.spawn(async move {
721 27266 : drain(&st, evs).await;
722 27266 : if !changes.is_empty() {
723 27264 : if let Some(flush) = &st.change_flush {
724 27264 : flush(changes);
725 27264 : }
726 2 : }
727 27266 : }))
728 110068 : }
729 : }
730 :
731 : /// Dropped before [`layer`] took it out of the buffers' scope: the request
732 : /// ended without an answer. Tokio drops a scoped future with its
733 : /// task-locals set, so the buffers are still there to collect.
734 : #[cfg(not(target_arch = "wasm32"))]
735 : impl Drop for Buffered {
736 55038 : fn drop(&mut self) {
737 55038 : self.collect();
738 55038 : let _ = self.hand_on();
739 55038 : }
740 : }
741 :
742 : /// Router layer: scopes the buffer over the handler and drains it once the
743 : /// response is built. Drained BEFORE the response is returned so
744 : /// read-your-writes holds at any store latency (the ETSI temporal suites
745 : /// read history straight after the write).
746 : #[cfg(not(target_arch = "wasm32"))]
747 55038 : pub(crate) async fn layer(
748 55038 : axum::extract::State(st): axum::extract::State<AppState>,
749 55038 : req: axum::extract::Request,
750 55038 : next: axum::middleware::Next,
751 55038 : ) -> axum::response::Response {
752 55038 : let provenance = crate::provenance::of_request(req.headers(), st.provenance_from.as_deref());
753 55038 : let tenant = crate::negotiate::tenant_from(req.headers()).ok();
754 55038 : let (resp, mut buffered) = BUFFER
755 55038 : .scope(std::cell::RefCell::new(Vec::new()), async {
756 55038 : STAGED
757 55038 : .scope(
758 55038 : std::cell::RefCell::new(Vec::new()),
759 55038 : CHANGES.scope(std::cell::RefCell::new(Vec::new()), async {
760 55038 : PROVENANCE
761 55038 : .scope(
762 55038 : provenance,
763 55038 : TENANT.scope(tenant, async {
764 55038 : let mut buffered = Buffered {
765 55038 : st: st.clone(),
766 55038 : evs: Vec::new(),
767 55038 : changes: Vec::new(),
768 55038 : };
769 55038 : let resp = next.run(req).await;
770 55034 : buffered.collect();
771 55034 : (resp, buffered)
772 55034 : }),
773 : )
774 55038 : .await
775 55034 : }),
776 : )
777 55038 : .await
778 55034 : })
779 55038 : .await;
780 55030 : if let Some(task) = buffered.hand_on() {
781 27258 : if let Err(e) = task.await {
782 0 : tracing::warn!("recording a request's history failed: {e}");
783 27258 : }
784 27772 : }
785 55030 : resp
786 55030 : }
787 :
788 : /// No buffer on wasm (pushes drain on the spot): the layer is a pass-through
789 : /// so the router composes identically on both targets.
790 : #[cfg(target_arch = "wasm32")]
791 : pub(crate) async fn layer(
792 : axum::extract::State(_st): axum::extract::State<AppState>,
793 : req: axum::extract::Request,
794 : next: axum::middleware::Next,
795 : ) -> axum::response::Response {
796 : next.run(req).await
797 : }
798 :
799 : /// delete_temporal_on_core_delete: entity deletion removes its temporal
800 : /// representation too (suite configuration parity). Skipped on bus=nats
801 : /// api pods — the recorder applies the entityDeleted fence instead.
802 11424 : pub(crate) async fn mirror_delete_entity(st: &AppState, tenant: &TenantId, id: &str) {
803 11424 : if !st.record_locally() {
804 0 : return;
805 11424 : }
806 11424 : if let Err(e) = st.temporal.delete(tenant, id).await {
807 4 : note_mirror_error();
808 4 : tracing::warn!("temporal mirror delete failed: {e}");
809 11420 : }
810 11424 : }
811 :
812 : /// A history write of the delete mirror failed: counted with the drain's,
813 : /// because the history it should have recorded is lost the same way.
814 8 : fn note_mirror_error() {
815 8 : DRAIN_ERRORS.fetch_add(1, Ordering::Relaxed);
816 8 : metrics::counter!("antares_temporal_drain_errors_total").increment(1);
817 8 : }
818 :
819 : /// 4.5.7/4.5.8: "In case the Property is deleted, an instance of the
820 : /// Property is recorded with its value set to the URI "urn:ngsi-ld:null"
821 : /// and the deletedAt Temporal Property set" (object for a Relationship;
822 : /// typed null shapes for the LanguageProperty/JsonProperty/Vocab/List
823 : /// subtypes). Each recorded instance carries an instanceId — the clause
824 : /// SHOULD that makes 5.6.14/5.6.15 selective modification possible.
825 464 : pub(crate) async fn mirror_delete_attr(
826 464 : st: &AppState,
827 464 : tenant: &TenantId,
828 464 : id: &str,
829 464 : attr_iri: &str,
830 464 : dataset_id: Option<&str>,
831 464 : ts: &str,
832 464 : ) {
833 464 : let r = st
834 464 : .temporal
835 464 : .mutate(tenant, id, |doc| {
836 : // The mirror writes nothing into a document the temporal driver
837 : // handed back in a shape the contract forbids.
838 452 : let Some(target) = doc.as_object_mut() else {
839 0 : return Ok::<(), std::convert::Infallible>(());
840 : };
841 452 : if attr_iri == "scope" {
842 : // scope deletion: a deleted scope is the NGSI-LD Null (4.18:
843 : // urn:ngsi-ld:null appears only for deleted scopes; 4.5.0)
844 24 : let inst = serde_json::json!({
845 24 : "type": "Property",
846 24 : "value": "urn:ngsi-ld:null",
847 24 : "instanceId": format!("urn:ngsi-ld:Instance:{}", uuid::Uuid::new_v4()),
848 24 : "deletedAt": ts, "createdAt": ts, "modifiedAt": ts,
849 : });
850 24 : match target.get_mut("scope").and_then(Value::as_array_mut) {
851 24 : Some(arr) if arr.first().is_some_and(|i| i.is_object()) => arr.push(inst),
852 24 : _ => {
853 24 : target.insert("scope".into(), Value::Array(vec![inst]));
854 24 : }
855 : }
856 24 : return Ok::<(), std::convert::Infallible>(());
857 428 : }
858 428 : if let Some(arr) = target.get_mut(attr_iri).and_then(Value::as_array_mut) {
859 428 : if arr.is_empty() {
860 0 : return Ok(());
861 428 : }
862 428 : let atype = arr
863 428 : .first()
864 428 : .and_then(|i| i.get("type"))
865 428 : .and_then(Value::as_str)
866 428 : .unwrap_or("Property")
867 428 : .to_owned();
868 428 : let mut inst = Map::new();
869 428 : inst.insert("type".into(), Value::String(atype.clone()));
870 428 : let null = Value::String("urn:ngsi-ld:null".into());
871 428 : match atype.as_str() {
872 428 : "Relationship" => {
873 84 : inst.insert("object".into(), null);
874 84 : }
875 344 : "LanguageProperty" => {
876 60 : inst.insert(
877 60 : "languageMap".into(),
878 60 : serde_json::json!({"@none": "urn:ngsi-ld:null"}),
879 60 : );
880 60 : }
881 284 : "JsonProperty" => {
882 4 : inst.insert("json".into(), null);
883 4 : }
884 280 : "VocabProperty" => {
885 8 : inst.insert("vocab".into(), null);
886 8 : }
887 272 : "ListProperty" => {
888 0 : inst.insert("valueList".into(), null);
889 0 : }
890 272 : "ListRelationship" => {
891 0 : inst.insert("objectList".into(), null);
892 0 : }
893 272 : _ => {
894 272 : inst.insert("value".into(), null);
895 272 : }
896 : }
897 428 : if let Some(ds) = dataset_id {
898 84 : inst.insert("datasetId".into(), Value::String(ds.to_owned()));
899 344 : }
900 428 : inst.insert(
901 428 : "instanceId".into(),
902 428 : Value::String(format!("urn:ngsi-ld:Instance:{}", uuid::Uuid::new_v4())),
903 : );
904 428 : inst.insert("deletedAt".into(), Value::String(ts.to_owned()));
905 : // the tombstone instance is written at the deletion: a
906 : // timeproperty=createdAt|modifiedAt window finds it there
907 428 : inst.insert("createdAt".into(), Value::String(ts.to_owned()));
908 428 : inst.insert("modifiedAt".into(), Value::String(ts.to_owned()));
909 428 : inst.insert(antares_store::PROVENANCE_KEY.into(), provenance("delete"));
910 428 : arr.push(Value::Object(inst));
911 0 : }
912 428 : Ok(())
913 452 : })
914 464 : .await;
915 464 : if let Err(e) = r {
916 4 : note_mirror_error();
917 4 : tracing::warn!("temporal attr mirror failed: {e}");
918 460 : }
919 464 : }
920 :
921 : #[cfg(all(test, not(target_arch = "wasm32")))]
922 : mod tests {
923 : use super::*;
924 :
925 : /// A history store whose writes all fail.
926 : struct Down;
927 :
928 : #[async_trait::async_trait]
929 : impl antares_store::TemporalDriver for Down {
930 0 : async fn event_list(&self, _evs: &[TemporalEvent]) -> Result<(), antares_model::NgsiError> {
931 : Err(antares_model::NgsiError::InternalError(
932 : "history store down".into(),
933 : ))
934 0 : }
935 : async fn temporal_append(
936 : &self,
937 : _t: &TenantId,
938 : _id: &str,
939 : _shell: &Value,
940 : _additions: &Value,
941 0 : ) -> Result<(), antares_model::NgsiError> {
942 : Err(antares_model::NgsiError::InternalError(
943 : "history store down".into(),
944 : ))
945 0 : }
946 : async fn query_temporal(
947 : &self,
948 : _t: &TenantId,
949 : _f: &antares_store::filter::TemporalFilter<'_>,
950 0 : ) -> Result<antares_store::filter::TemporalOutcome, antares_model::NgsiError> {
951 : Err(antares_model::NgsiError::InternalError(
952 : "history store down".into(),
953 : ))
954 0 : }
955 : async fn get_temporal(
956 : &self,
957 : _t: &TenantId,
958 : _id: &str,
959 : _f: &antares_store::filter::TemporalFilter<'_>,
960 0 : ) -> Result<Option<Value>, antares_model::NgsiError> {
961 : Err(antares_model::NgsiError::InternalError(
962 : "history store down".into(),
963 : ))
964 0 : }
965 : async fn get(
966 : &self,
967 : _t: &TenantId,
968 : _id: &str,
969 0 : ) -> Result<Option<Value>, antares_model::NgsiError> {
970 : Err(antares_model::NgsiError::InternalError(
971 : "history store down".into(),
972 : ))
973 0 : }
974 : async fn create(
975 : &self,
976 : _t: &TenantId,
977 : _id: &str,
978 : _doc: Value,
979 0 : ) -> Result<bool, antares_model::NgsiError> {
980 : Err(antares_model::NgsiError::InternalError(
981 : "history store down".into(),
982 : ))
983 0 : }
984 : async fn upsert(
985 : &self,
986 : _t: &TenantId,
987 : _id: &str,
988 : _doc: Value,
989 0 : ) -> Result<bool, antares_model::NgsiError> {
990 : Err(antares_model::NgsiError::InternalError(
991 : "history store down".into(),
992 : ))
993 0 : }
994 4 : async fn delete(&self, _t: &TenantId, _id: &str) -> Result<bool, antares_model::NgsiError> {
995 : Err(antares_model::NgsiError::InternalError(
996 : "history store down".into(),
997 : ))
998 4 : }
999 0 : async fn list(&self, _t: &TenantId) -> Result<Vec<Value>, antares_model::NgsiError> {
1000 : Err(antares_model::NgsiError::InternalError(
1001 : "history store down".into(),
1002 : ))
1003 0 : }
1004 : async fn mutate_boxed<'a>(
1005 : &self,
1006 : _t: &TenantId,
1007 : _id: &str,
1008 : _f: antares_store::MutateFn<'a>,
1009 4 : ) -> Result<Option<Result<(), ()>>, antares_model::NgsiError> {
1010 : Err(antares_model::NgsiError::InternalError(
1011 : "history store down".into(),
1012 : ))
1013 4 : }
1014 : }
1015 :
1016 : /// The delete mirrors lose history exactly as a failed drain does, so
1017 : /// they are counted in the same `temporalDrainErrors`.
1018 : #[tokio::test]
1019 4 : async fn a_failed_delete_mirror_is_counted_as_a_drain_error() {
1020 4 : let base = AppState::new("antares-mirror-errors".into());
1021 4 : let st = AppState::with_drivers(
1022 4 : "antares-mirror-errors".into(),
1023 4 : base.store.clone(),
1024 4 : std::sync::Arc::new(Down),
1025 4 : "memory",
1026 : );
1027 4 : let t = TenantId::new("mirrorerrors").expect("tenant");
1028 4 : let before = drain_errors();
1029 4 : mirror_delete_entity(&st, &t, "urn:ngsi-ld:T:1").await;
1030 4 : let after_entity = drain_errors();
1031 4 : assert!(
1032 4 : after_entity > before,
1033 : "the entity delete mirror was not counted"
1034 : );
1035 4 : mirror_delete_attr(
1036 4 : &st,
1037 4 : &t,
1038 4 : "urn:ngsi-ld:T:1",
1039 4 : "https://x/a",
1040 4 : None,
1041 4 : "2026-01-01T00:00:00Z",
1042 4 : )
1043 4 : .await;
1044 4 : assert!(
1045 4 : drain_errors() > after_entity,
1046 4 : "the attribute delete mirror was not counted"
1047 4 : );
1048 4 : }
1049 :
1050 24 : fn ev(id: &str) -> TemporalEvent {
1051 24 : TemporalEvent {
1052 24 : op: antares_store::TemporalOp::AttrModified,
1053 24 : tenant: TenantId::new("t").expect("tenant"),
1054 24 : entity_id: id.to_owned(),
1055 24 : shell: serde_json::json!({"id": id}),
1056 24 : attr: "https://example.org/level".to_owned(),
1057 24 : instance: serde_json::json!({"type": "Property", "value": 1}),
1058 24 : }
1059 24 : }
1060 :
1061 : /// a backfill found inside a read-modify-write reaches the
1062 : /// request's history only when that write commits; a refused write
1063 : /// leaves none, and a rerun closure records its backfill once.
1064 : #[tokio::test]
1065 4 : async fn a_backfill_is_recorded_only_for_a_committed_write() {
1066 4 : let buffered = BUFFER
1067 4 : .scope(std::cell::RefCell::new(Vec::new()), async {
1068 4 : STAGED
1069 4 : .scope(std::cell::RefCell::new(Vec::new()), async {
1070 : // refused: nothing reaches history
1071 4 : restage("urn:a");
1072 4 : assert!(stage(ev("urn:a")));
1073 4 : settle("urn:a", false);
1074 4 : assert_eq!(BUFFER.with(|b| b.borrow().len()), 0);
1075 : // rerun twice, then committed: one event
1076 4 : restage("urn:b");
1077 4 : assert!(stage(ev("urn:b")));
1078 4 : restage("urn:b");
1079 4 : assert!(stage(ev("urn:b")));
1080 : // another Entity's staged event stays staged
1081 4 : assert!(stage(ev("urn:c")));
1082 4 : settle_all::<(), ()>(&["urn:b".to_owned()], &[Some(Ok(()))]);
1083 4 : assert_eq!(STAGED.with(|b| b.borrow().len()), 1);
1084 4 : BUFFER.with(|b| b.take())
1085 4 : })
1086 4 : .await
1087 4 : })
1088 4 : .await;
1089 4 : let ids: Vec<&str> = buffered.iter().map(|e| e.entity_id.as_str()).collect();
1090 4 : assert_eq!(ids, ["urn:b"]);
1091 4 : }
1092 :
1093 : /// A handler behind [`layer`] that commits a write, buffers its history
1094 : /// event and its change, and then ends as `end` does; the matcher's
1095 : /// hand-over is counted.
1096 8 : fn committed_then(
1097 8 : end: fn() -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>>,
1098 8 : ) -> (AppState, axum::Router, std::sync::Arc<AtomicU64>) {
1099 8 : let mut st = AppState::new("t".into());
1100 8 : let flushed = std::sync::Arc::new(AtomicU64::new(0));
1101 8 : let counter = flushed.clone();
1102 8 : st.change_flush = Some(std::sync::Arc::new(move |changes| {
1103 8 : counter.fetch_add(changes.len() as u64, Ordering::SeqCst);
1104 8 : }));
1105 8 : let handler = move |axum::extract::State(st): axum::extract::State<AppState>| async move {
1106 8 : let tenant = TenantId::new("t").expect("tenant");
1107 8 : let doc =
1108 8 : serde_json::json!({"id": "urn:ngsi-ld:T:1", "type": ["https://example.org/T"]});
1109 8 : st.store
1110 8 : .upsert(&tenant, antares_store::Kind::Entity, "urn:ngsi-ld:T:1", doc)
1111 8 : .await
1112 8 : .expect("commit");
1113 8 : push(&st, ev("urn:ngsi-ld:T:1")).await;
1114 8 : assert!(buffer_change(("urn:ngsi-ld:T:1".to_owned(), None, None)).is_none());
1115 8 : end().await;
1116 12 : };
1117 8 : let app = axum::Router::new()
1118 8 : .route("/w", axum::routing::post(handler))
1119 8 : .layer(axum::middleware::from_fn_with_state(st.clone(), layer))
1120 8 : .layer(tower_http::catch_panic::CatchPanicLayer::new())
1121 8 : .with_state(st.clone());
1122 8 : (st, app, flushed)
1123 8 : }
1124 :
1125 : /// Whether the event `committed_then` buffered reached history and its
1126 : /// change the matcher, waiting a moment for the task that hands them on.
1127 8 : async fn handed_on(st: &AppState, flushed: &AtomicU64) -> bool {
1128 8 : let tenant = TenantId::new("t").expect("tenant");
1129 8 : for _ in 0..200 {
1130 16 : let recorded = st
1131 16 : .temporal
1132 16 : .get(&tenant, "urn:ngsi-ld:T:1")
1133 16 : .await
1134 16 : .expect("read")
1135 16 : .is_some();
1136 16 : if recorded && flushed.load(Ordering::SeqCst) == 1 {
1137 8 : return true;
1138 8 : }
1139 8 : tokio::time::sleep(Duration::from_millis(10)).await;
1140 : }
1141 0 : false
1142 8 : }
1143 :
1144 8 : fn post() -> axum::extract::Request {
1145 8 : axum::http::Request::post("/w")
1146 8 : .body(axum::body::Body::empty())
1147 8 : .expect("request")
1148 8 : }
1149 :
1150 : /// A client that goes away drops its request's future: what the
1151 : /// handler already committed is still recorded and still notified.
1152 : #[tokio::test]
1153 4 : async fn a_dropped_request_still_records_and_notifies_what_it_committed() {
1154 : use tower::ServiceExt as _;
1155 4 : let (st, app, flushed) = committed_then(|| Box::pin(std::future::pending()));
1156 4 : let answer = tokio::time::timeout(Duration::from_millis(200), app.oneshot(post())).await;
1157 4 : assert!(answer.is_err(), "the handler never answers");
1158 4 : assert!(
1159 4 : handed_on(&st, &flushed).await,
1160 4 : "the committed write was lost"
1161 4 : );
1162 4 : }
1163 :
1164 : /// So is a handler that panics after its write committed (the 500
1165 : /// comes from a layer outside this one).
1166 : #[tokio::test]
1167 4 : async fn a_panicking_handler_still_records_and_notifies_what_it_committed() {
1168 : use tower::ServiceExt as _;
1169 4 : let (st, app, flushed) = committed_then(|| Box::pin(async { panic!("after the commit") }));
1170 4 : let resp = app.oneshot(post()).await.expect("response");
1171 4 : assert_eq!(resp.status(), axum::http::StatusCode::INTERNAL_SERVER_ERROR);
1172 4 : assert!(
1173 4 : handed_on(&st, &flushed).await,
1174 4 : "the committed write was lost"
1175 4 : );
1176 4 : }
1177 : }
|
