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;
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 :
20 : static DRAIN_ERRORS: AtomicU64 = AtomicU64::new(0);
21 :
22 : /// Drains that failed in the driver (the events of that request are lost;
23 : /// the write itself stood).
24 94 : pub fn drain_errors() -> u64 {
25 94 : DRAIN_ERRORS.load(Ordering::Relaxed)
26 94 : }
27 :
28 : // The buffer rides the request's task; tokio task-locals need the `rt`
29 : // feature the single-threaded wasm build does not carry, so wasm records
30 : // immediately.
31 : #[cfg(not(target_arch = "wasm32"))]
32 : tokio::task_local! {
33 : static BUFFER: std::cell::RefCell<Vec<TemporalEvent>>;
34 : static CHANGES: std::cell::RefCell<Vec<crate::mirror::Change>>;
35 : }
36 :
37 : /// Buffer one entity change for the request in flight so the matcher
38 : /// receives the whole request at once. Handed back when no request is in
39 : /// flight — the caller then gives it to the matcher on the spot.
40 6462 : pub(crate) fn buffer_change(change: crate::mirror::Change) -> Option<crate::mirror::Change> {
41 : #[cfg(not(target_arch = "wasm32"))]
42 : {
43 6462 : let mut slot = Some(change);
44 6462 : let _ = CHANGES.try_with(|b| {
45 6264 : if let Some(c) = slot.take() {
46 6264 : b.borrow_mut().push(c);
47 6264 : }
48 6264 : });
49 6462 : slot
50 : }
51 : #[cfg(target_arch = "wasm32")]
52 : Some(change)
53 6462 : }
54 :
55 : /// Hand one event to the seam: buffered when a request is in flight,
56 : /// drained on the spot otherwise.
57 5654 : pub(crate) async fn push(st: &AppState, ev: TemporalEvent) {
58 : #[cfg(not(target_arch = "wasm32"))]
59 90 : let ev = {
60 5654 : let mut slot = Some(ev);
61 5654 : let buffered = BUFFER
62 5654 : .try_with(|b| {
63 5564 : if let Some(ev) = slot.take() {
64 5564 : b.borrow_mut().push(ev);
65 5564 : }
66 5564 : })
67 5654 : .is_ok();
68 5654 : if buffered {
69 5564 : return;
70 90 : }
71 90 : match slot {
72 90 : Some(ev) => ev,
73 0 : None => return,
74 : }
75 : };
76 90 : drain(st, vec![ev]).await;
77 5654 : }
78 :
79 : /// The gate chain: an event enters history only if every gate admits it.
80 : /// Gate 1 (value-change) runs in the producer — an unchanged instance never
81 : /// becomes an event (`changed_instances`). Adding a gate = one more entry
82 : /// here; producers and drivers stay untouched.
83 : const GATES: &[fn(&AppState, &TemporalEvent) -> bool] = &[observed_gate];
84 :
85 : /// Gate 2: ANTARES_TEMPORAL_RECORD. `all` admits everything; `observed`
86 : /// keeps only instances that carry `observedAt` — the spec's own
87 : /// measurement axis (4.5.7: observedAt is the default timeproperty), so
88 : /// metadata-shaped writes leave no history; `none` admits nothing.
89 5654 : fn observed_gate(st: &AppState, ev: &TemporalEvent) -> bool {
90 : use crate::state::TemporalRecord::*;
91 5654 : match st.temporal_record {
92 5592 : All => true,
93 48 : Observed => ev.instance.get("observedAt").is_some(),
94 14 : None => false,
95 : }
96 5654 : }
97 :
98 : /// The consumer side: one `event_list` call per drained batch.
99 26148 : pub(crate) async fn drain(st: &AppState, evs: Vec<TemporalEvent>) {
100 26148 : let evs: Vec<TemporalEvent> = evs
101 26148 : .into_iter()
102 26148 : .filter(|ev| GATES.iter().all(|gate| gate(st, ev)))
103 26148 : .collect();
104 26148 : if evs.is_empty() {
105 20816 : return;
106 5332 : }
107 5332 : if let Err(e) = st.temporal.event_list(&evs).await {
108 2 : DRAIN_ERRORS.fetch_add(1, Ordering::Relaxed);
109 2 : metrics::counter!("antares_temporal_drain_errors_total").increment(1);
110 2 : tracing::warn!(events = evs.len(), "temporal drain failed: {e}");
111 5330 : }
112 26148 : }
113 :
114 : /// Router layer: scopes the buffer over the handler and drains it once the
115 : /// response is built. Drained BEFORE the response is returned so
116 : /// read-your-writes holds at any store latency (the ETSI temporal suites
117 : /// read history straight after the write).
118 : #[cfg(not(target_arch = "wasm32"))]
119 26058 : pub(crate) async fn layer(
120 26058 : axum::extract::State(st): axum::extract::State<AppState>,
121 26058 : req: axum::extract::Request,
122 26058 : next: axum::middleware::Next,
123 26058 : ) -> axum::response::Response {
124 26058 : let (resp, evs, changes) = BUFFER
125 26058 : .scope(std::cell::RefCell::new(Vec::new()), async {
126 26058 : CHANGES
127 26058 : .scope(std::cell::RefCell::new(Vec::new()), async {
128 26058 : let resp = next.run(req).await;
129 26058 : (resp, BUFFER.with(|b| b.take()), CHANGES.with(|c| c.take()))
130 26058 : })
131 26058 : .await
132 26058 : })
133 26058 : .await;
134 26058 : drain(&st, evs).await;
135 26058 : if !changes.is_empty() {
136 6250 : if let Some(flush) = &st.change_flush {
137 6250 : flush(changes);
138 6250 : }
139 19808 : }
140 26058 : resp
141 26058 : }
142 :
143 : /// No buffer on wasm (pushes drain on the spot): the layer is a pass-through
144 : /// so the router composes identically on both targets.
145 : #[cfg(target_arch = "wasm32")]
146 : pub(crate) async fn layer(
147 : axum::extract::State(_st): axum::extract::State<AppState>,
148 : req: axum::extract::Request,
149 : next: axum::middleware::Next,
150 : ) -> axum::response::Response {
151 : next.run(req).await
152 : }
153 :
154 : /// delete_temporal_on_core_delete: entity deletion removes its temporal
155 : /// representation too (suite configuration parity). Skipped on bus=nats
156 : /// api pods — the recorder applies the entityDeleted fence instead.
157 5736 : pub(crate) async fn mirror_delete_entity(st: &AppState, tenant: &TenantId, id: &str) {
158 5736 : if !st.record_locally() {
159 0 : return;
160 5736 : }
161 5736 : if let Err(e) = st.temporal.delete(tenant, id).await {
162 0 : tracing::warn!("temporal mirror delete failed: {e}");
163 5736 : }
164 5736 : }
165 :
166 : /// 4.5.7/4.5.8: "In case the Property is deleted, an instance of the
167 : /// Property is recorded with its value set to the URI "urn:ngsi-ld:null"
168 : /// and the deletedAt Temporal Property set" (object for a Relationship;
169 : /// typed null shapes for the LanguageProperty/JsonProperty/Vocab/List
170 : /// subtypes). Each recorded instance carries an instanceId — the clause
171 : /// SHOULD that makes 5.6.14/5.6.15 selective modification possible.
172 52 : pub(crate) async fn mirror_delete_attr(
173 52 : st: &AppState,
174 52 : tenant: &TenantId,
175 52 : id: &str,
176 52 : attr_iri: &str,
177 52 : dataset_id: Option<&str>,
178 52 : ts: &str,
179 52 : ) -> bool {
180 52 : let mut had = false;
181 52 : let r = st
182 52 : .temporal
183 52 : .mutate(tenant, id, |doc| {
184 : // The mirror writes nothing into a document the temporal driver
185 : // handed back in a shape the contract forbids; `had` stays false and
186 : // the caller reports that nothing was mirrored.
187 26 : let Some(target) = doc.as_object_mut() else {
188 0 : return Ok::<(), std::convert::Infallible>(());
189 : };
190 26 : if attr_iri == "scope" {
191 : // scope deletion: temporal scope becomes an instance array with
192 : // value [] (the 020_19/020_20 shape)
193 4 : had = true;
194 4 : let inst = serde_json::json!({
195 4 : "type": "Property",
196 4 : "value": [],
197 4 : "instanceId": format!("urn:ngsi-ld:Instance:{}", uuid::Uuid::new_v4()),
198 4 : "deletedAt": ts,
199 : });
200 4 : match target.get_mut("scope").and_then(Value::as_array_mut) {
201 4 : Some(arr) if arr.first().is_some_and(|i| i.is_object()) => arr.push(inst),
202 4 : _ => {
203 4 : target.insert("scope".into(), Value::Array(vec![inst]));
204 4 : }
205 : }
206 4 : return Ok::<(), std::convert::Infallible>(());
207 22 : }
208 22 : if let Some(arr) = target.get_mut(attr_iri).and_then(Value::as_array_mut) {
209 20 : if arr.is_empty() {
210 0 : return Ok(());
211 20 : }
212 20 : had = true;
213 20 : let atype = arr
214 20 : .first()
215 20 : .and_then(|i| i.get("type"))
216 20 : .and_then(Value::as_str)
217 20 : .unwrap_or("Property")
218 20 : .to_owned();
219 20 : let mut inst = Map::new();
220 20 : inst.insert("type".into(), Value::String(atype.clone()));
221 20 : let null = Value::String("urn:ngsi-ld:null".into());
222 20 : match atype.as_str() {
223 20 : "Relationship" => {
224 2 : inst.insert("object".into(), null);
225 2 : }
226 18 : "LanguageProperty" => {
227 0 : inst.insert(
228 0 : "languageMap".into(),
229 0 : serde_json::json!({"@none": "urn:ngsi-ld:null"}),
230 0 : );
231 0 : }
232 18 : "JsonProperty" => {
233 0 : inst.insert("json".into(), null);
234 0 : }
235 18 : "VocabProperty" => {
236 0 : inst.insert("vocab".into(), null);
237 0 : }
238 18 : "ListProperty" => {
239 0 : inst.insert("valueList".into(), null);
240 0 : }
241 18 : "ListRelationship" => {
242 0 : inst.insert("objectList".into(), null);
243 0 : }
244 18 : _ => {
245 18 : inst.insert("value".into(), null);
246 18 : }
247 : }
248 20 : if let Some(ds) = dataset_id {
249 2 : inst.insert("datasetId".into(), Value::String(ds.to_owned()));
250 18 : }
251 20 : inst.insert(
252 20 : "instanceId".into(),
253 20 : Value::String(format!("urn:ngsi-ld:Instance:{}", uuid::Uuid::new_v4())),
254 : );
255 20 : inst.insert("deletedAt".into(), Value::String(ts.to_owned()));
256 20 : arr.push(Value::Object(inst));
257 2 : }
258 22 : Ok(())
259 26 : })
260 52 : .await;
261 52 : if let Err(e) = r {
262 0 : tracing::warn!("temporal attr mirror failed: {e}");
263 52 : }
264 52 : had
265 52 : }
|
