Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! 5.16 Snapshots (optional API group; resources 6.36 /snapshots,
3 : //! 6.37 /snapshots/{id}, 6.38 /snapshots/{id}/clone; scoping 6.3.22).
4 : //!
5 : //! A Snapshot freezes the results of a set of queries (5.2.41) into an
6 : //! isolated copy on which Core + Temporal API operations run implicitly
7 : //! local (5.5.15). Implementation shape: each snapshot owns a synthetic
8 : //! internal tenant ("snap-…"); the 6.3.22 NGSILD-Snapshot header is resolved
9 : //! by a middleware that rewrites the request's tenant, so every existing
10 : //! Core/Temporal handler serves snapshot content unchanged — and, because
11 : //! no registrations exist under the synthetic tenant, all operations are
12 : //! naturally local.
13 : //!
14 : //! Snapshot metadata lives in the store (Kind::Snapshot, ADR-0012) —
15 : //! persistent modes serve snapshots across restarts; 5.5.15 still allows
16 : //! dropping them under resource pressure (evict_over_cap).
17 : //! Fills follow the 5.7.2.4 distributed path and page past max_limit, up to
18 : //! the 5.5.6 result ceiling (fill_cap) and only for as long as the snapshot
19 : //! exists — a fill whose snapshot was deleted or evicted frees its copy
20 : //! (fill_cancelled), which nothing else could reach.
21 : //! Resource pressure evicts lowest-priority snapshots (evict_over_cap).
22 :
23 : use crate::negotiate::{
24 : check_params, created, no_content, parse_accept, parse_body, respond, single_header,
25 : tenant_from, ApiError, ApiResult, BodyKind, CleanParams,
26 : };
27 : use crate::state::{now_iso, AppState};
28 : use antares_model::{NgsiError, TenantId, API_ROOT};
29 : use antares_store::CurrentStateDriverExt;
30 : use antares_store::Kind;
31 : use axum::body::Bytes;
32 : use axum::extract::{Path, State};
33 : use axum::http::{HeaderMap, StatusCode};
34 : use axum::response::{IntoResponse, Response};
35 : use serde_json::{json, Map, Value};
36 : use std::collections::HashMap;
37 :
38 : const DEFAULT_LIFETIME_SECS: i64 = 86_400; // 1 day
39 : const MAX_LIFETIME_SECS: i64 = 604_800; // 7 days — the "configured limit"
40 :
41 106 : fn bad(m: String) -> NgsiError {
42 106 : NgsiError::BadRequestData(m)
43 106 : }
44 :
45 : /// 5.5.6: an unexpected failure while filling a snapshot surfaces as
46 : /// InternalError. Its detail reaches the client twice — in
47 : /// snapshotQueriesDetails.problemDetails (5.2.41) and in the 5.3.4
48 : /// SnapshotNotification body — so the internal error text stays in the
49 : /// server log and the client-visible detail is generic.
50 4 : fn opaque(what: &str, e: &dyn std::fmt::Debug) -> NgsiError {
51 4 : tracing::error!("snapshot {what} failed: {e:?}");
52 4 : NgsiError::InternalError(format!("{what} failed"))
53 4 : }
54 :
55 : /// 5.2.41: expiresAt from the suggested snapshotLifetime, bounded by the
56 : /// system limit (5.16.1.4 "applying the configured limit").
57 144 : fn expires_at(meta: &Map<String, Value>) -> Result<String, NgsiError> {
58 144 : let secs = match meta.get("snapshotLifetime").and_then(Value::as_str) {
59 6 : Some(d) => crate::entity_map::iso8601_secs(d)
60 6 : .ok_or_else(|| {
61 0 : bad(format!(
62 : "snapshotLifetime is not an ISO 8601 duration: {d:?}"
63 : ))
64 0 : })?
65 : // A zero or negative suggestion would answer 201 with a Location
66 : // that is already expired, so the broker floor applies as well as
67 : // the ceiling — the same reasoning the EntityMap lifetime carries.
68 6 : .clamp(1, MAX_LIFETIME_SECS),
69 138 : None => DEFAULT_LIFETIME_SECS,
70 : };
71 144 : Ok((chrono::Utc::now() + chrono::Duration::seconds(secs))
72 144 : .to_rfc3339_opts(chrono::SecondsFormat::Millis, true))
73 144 : }
74 :
75 1261 : fn expired(meta: &Value) -> bool {
76 1261 : meta.get("expiresAt")
77 1261 : .and_then(Value::as_str)
78 1261 : .and_then(|e| chrono::DateTime::parse_from_rfc3339(e).ok())
79 1261 : .is_some_and(|e| e < chrono::Utc::now())
80 1261 : }
81 :
82 : /// The internal tenant holding the synth-tenant -> (owner, snapshot id)
83 : /// index docs (durable reverse lookup for 6.3.22 notification stamping).
84 184 : fn snap_index_tenant() -> Option<TenantId> {
85 184 : TenantId::new_internal("snap-index").ok()
86 184 : }
87 :
88 : /// 5.2.41: a Snapshot's type is fixed to "Snapshot". The reverse-index
89 : /// marker docs share Kind::Snapshot storage but are not Snapshots, so every
90 : /// place that reads a document as one filters on this — a listing that does
91 : /// not would count and select the markers as snapshots.
92 1439 : fn is_snapshot(meta: &Value) -> bool {
93 1439 : meta.get("type").and_then(Value::as_str) == Some("Snapshot")
94 1439 : }
95 :
96 : /// Registry access with lazy expiry: an expired snapshot is gone to every
97 : /// caller. Reading is a read — the meta row, its reverse-index entry and
98 : /// the synthetic tenant's data are freed by `sweep_expired_snapshots` on
99 : /// the sweep tick, not by the request that found the expiry. Snapshot docs
100 : /// live in the store (Kind::Snapshot) so restarts keep them on persistent
101 : /// store modes.
102 922 : pub(crate) async fn snap_get(st: &AppState, tenant: &TenantId, id: &str) -> Option<Value> {
103 922 : st.store
104 922 : .get(tenant, Kind::Snapshot, id)
105 922 : .await
106 922 : .ok()
107 922 : .flatten()
108 922 : .filter(|meta| is_snapshot(meta) && !expired(meta))
109 922 : }
110 :
111 : /// 5.16.1.4: "If the NGSI-LD endpoint already knows about this Snapshot …
112 : /// an error of type AlreadyExists shall be raised." The store's create is
113 : /// the atomic check-and-insert, so two concurrent creates of the same id
114 : /// cannot both succeed and overwrite each other's synthetic tenant.
115 140 : async fn snap_insert(
116 140 : st: &AppState,
117 140 : tenant: &TenantId,
118 140 : id: &str,
119 140 : meta: &Value,
120 140 : ) -> Result<(), NgsiError> {
121 140 : if !st
122 140 : .store
123 140 : .create(tenant, Kind::Snapshot, id, meta.clone())
124 140 : .await?
125 : {
126 4 : return Err(NgsiError::AlreadyExists(format!(
127 4 : "snapshot {id} already exists"
128 4 : )));
129 136 : }
130 : // Durable reverse index for snapshot_of_synth. The marker carries the
131 : // synthetic tenant as its `id` as well as its key: it is the only record
132 : // that names that tenant once the Snapshot document is gone, and a
133 : // document a listing cannot identify is one nothing can walk.
134 136 : if let (Some(synth), Some(idx)) = (
135 136 : meta.get("__tenant").and_then(Value::as_str),
136 136 : snap_index_tenant(),
137 : ) {
138 136 : let _ = st
139 136 : .store
140 136 : .create(
141 136 : &idx,
142 136 : Kind::Snapshot,
143 136 : synth,
144 136 : json!({"id": synth, "tenant": tenant.as_str(), "snapshot": id}),
145 136 : )
146 136 : .await;
147 0 : }
148 136 : Ok(())
149 140 : }
150 :
151 : /// Write back the output-only members (5.2.41 Table 5.2.41-2) of a snapshot
152 : /// that already exists; a snapshot deleted meanwhile is not resurrected.
153 142 : async fn snap_put(st: &AppState, tenant: &TenantId, meta: Value) {
154 142 : let id = meta
155 142 : .get("id")
156 142 : .and_then(Value::as_str)
157 142 : .unwrap_or_default()
158 142 : .to_owned();
159 142 : let _ = st
160 142 : .store
161 142 : .mutate(tenant, Kind::Snapshot, &id, |d| {
162 142 : *d = meta.clone();
163 142 : Ok::<_, std::convert::Infallible>(())
164 142 : })
165 142 : .await;
166 142 : }
167 :
168 : /// Remove a snapshot everywhere: doc, synth-tenant index, data purge.
169 44 : pub(crate) async fn snap_remove(st: &AppState, tenant: &TenantId, id: &str, meta: &Value) {
170 44 : let _ = st.store.delete(tenant, Kind::Snapshot, id).await;
171 44 : purge_data_bg(st, meta);
172 44 : }
173 :
174 : /// 4.22 for snapshots: `snap_get` refuses an expired one, and this is what
175 : /// removes it — meta, reverse-index entry and the synthetic tenant's data,
176 : /// through the same `snap_remove` a read used to call, so an expiry
177 : /// collected on the tick frees exactly what an expiry collected on a read
178 : /// freed.
179 20062 : pub(crate) async fn sweep_expired_snapshots(st: &AppState, tenant: &TenantId) -> usize {
180 20062 : let mut dead: Vec<(String, Value)> = Vec::new();
181 20062 : if crate::csource::walk_docs(st, tenant, Kind::Snapshot, |doc| {
182 411 : if is_snapshot(&doc) && expired(&doc) {
183 8 : if let Some(id) = doc.get("id").and_then(Value::as_str) {
184 8 : dead.push((id.to_owned(), doc));
185 8 : }
186 403 : }
187 411 : Ok(())
188 411 : })
189 20062 : .await
190 20062 : .is_err()
191 : {
192 0 : return 0;
193 20062 : }
194 20062 : let n = dead.len();
195 20062 : for (id, meta) in dead {
196 8 : snap_remove(st, tenant, &id, &meta).await;
197 : }
198 20062 : n
199 20062 : }
200 :
201 276 : fn synth_tenant(meta: &Value) -> Option<TenantId> {
202 276 : meta.get("__tenant")
203 276 : .and_then(Value::as_str)
204 276 : .and_then(|t| TenantId::new_internal(t).ok())
205 276 : }
206 :
207 : /// Free the snapshot's isolated copy. 5.5.15 permits every Core and
208 : /// Temporal operation on a snapshot, so a snapshot-scoped request can put
209 : /// documents of ANY kind under the synthetic tenant — once the snapshot is
210 : /// gone none of them is reachable (no client may name the internal tenant),
211 : /// so every kind is dropped, not only the filled data.
212 : ///
213 : /// The tenant purge, not a per-document sweep: it covers every kind rather
214 : /// than a list that has to be kept in step with `Kind`, and it removes the
215 : /// tenant itself. A sweep leaves the empty tenant behind, and `/q/tenants`
216 : /// reports what the store holds a tenant entry for — so every snapshot ever
217 : /// deleted would leave a `snap-<uuid>` name in the inventory for the life of
218 : /// the deployment.
219 48 : async fn purge_synth(st: &AppState, synth: &TenantId) {
220 : // history lives behind its own driver seam
221 48 : let _ = st.temporal.purge_tenant(synth).await;
222 48 : let _ = st.store.purge_tenant(synth).await;
223 48 : }
224 :
225 : /// The teardown that outlives the request: the copy is dropped, and only
226 : /// then the reverse-index entry that named the synthetic tenant. That order
227 : /// is the recovery path — the Snapshot document is already gone, so the index
228 : /// entry is the last thing that says which tenant holds the copy, and a
229 : /// broker that stops mid-teardown leaves it pointing at data still there
230 : /// rather than data nothing names.
231 44 : fn purge_data_bg(st: &AppState, meta: &Value) {
232 44 : let Some(synth) = synth_tenant(meta) else {
233 0 : return;
234 : };
235 44 : let st = st.clone();
236 44 : crate::spawn(async move {
237 44 : purge_synth(&st, &synth).await;
238 44 : if let Some(idx) = snap_index_tenant() {
239 44 : let _ = st.store.delete(&idx, Kind::Snapshot, synth.as_str()).await;
240 0 : }
241 44 : });
242 44 : }
243 :
244 : /// The snapshot document as presented to clients (internal members hidden).
245 170 : fn present(meta: &Value) -> Value {
246 170 : let mut out = meta.clone();
247 170 : crate::policy::strip_internal(&mut out);
248 170 : out
249 170 : }
250 :
251 : enum Mode {
252 : Create,
253 : Clone,
254 : Update,
255 : }
256 :
257 : /// 5.2.41 Table 5.2.41-1/-2 validation. Output-only members are IGNORED
258 : /// (stripped); read-only members in the wrong mode are BadRequestData.
259 212 : fn validate(body: &Value, mode: Mode) -> Result<Map<String, Value>, NgsiError> {
260 212 : let mut o = body
261 212 : .as_object()
262 212 : .cloned()
263 212 : .ok_or_else(|| bad("snapshot must be a JSON object".into()))?;
264 212 : o.remove("@context");
265 : // the broker's own members (the synthetic tenant, the creating subject)
266 : // are not the client's to send — see policy::strip_internal
267 452 : o.retain(|k, _| !k.starts_with("__"));
268 : // output-only members (Table 5.2.41-2) "shall be ignored"
269 1484 : for k in [
270 212 : "snapshotStatus",
271 212 : "snapshotQueriesDetails",
272 212 : "snapshotTemporalQueriesDetails",
273 212 : "createdAt",
274 212 : "modifiedAt",
275 212 : "expiresAt",
276 212 : "lastUsedAt",
277 1484 : ] {
278 1484 : o.remove(k);
279 1484 : }
280 212 : match mode {
281 : Mode::Create => {
282 168 : if o.get("type").and_then(Value::as_str) != Some("Snapshot") {
283 2 : return Err(bad("type must be \"Snapshot\" (5.2.41)".into()));
284 166 : }
285 166 : if !o.contains_key("snapshotQueries") && !o.contains_key("snapshotTemporalQueries") {
286 12 : return Err(bad(
287 12 : "at least one of snapshotQueries or snapshotTemporalQueries \
288 12 : shall be present (5.2.41)"
289 12 : .into(),
290 12 : ));
291 154 : }
292 : }
293 : Mode::Clone | Mode::Update => {
294 : // "both shall be omitted when updating the Snapshot status or
295 : // cloning the Snapshot" — read-only after creation
296 44 : if o.contains_key("snapshotQueries") || o.contains_key("snapshotTemporalQueries") {
297 8 : return Err(bad(
298 8 : "snapshotQueries/snapshotTemporalQueries are read-only after \
299 8 : creation (5.2.41)"
300 8 : .into(),
301 8 : ));
302 36 : }
303 : // Table 5.2.41-1: the id "cannot be later modified in update
304 : // operations" — of whatever JSON type it is sent as
305 36 : if matches!(mode, Mode::Update) && o.contains_key("id") {
306 8 : return Err(bad("the snapshot id cannot be modified (5.2.41)".into()));
307 28 : }
308 : }
309 : }
310 358 : for (key, temporal) in [
311 182 : ("snapshotQueries", false),
312 182 : ("snapshotTemporalQueries", true),
313 182 : ] {
314 358 : if let Some(qs) = o.get(key) {
315 158 : let arr = qs
316 158 : .as_array()
317 158 : .filter(|a| !a.is_empty())
318 158 : .ok_or_else(|| bad(format!("{key} must be a non-empty array of Query (5.2.41)")))?;
319 158 : for q in arr {
320 158 : let qo = q
321 158 : .as_object()
322 158 : .ok_or_else(|| bad(format!("{key} entries must be Query objects (5.2.23)")))?;
323 158 : if qo.get("type").and_then(Value::as_str) != Some("Query") {
324 0 : return Err(bad(format!("{key} entries must have type Query (5.2.23)")));
325 158 : }
326 158 : if temporal != qo.contains_key("temporalQ") {
327 6 : return Err(bad(format!(
328 : "{key} entries must {} a temporalQ element (5.2.41)",
329 6 : if temporal { "carry" } else { "not carry" }
330 : )));
331 152 : }
332 : }
333 200 : }
334 : }
335 176 : if let Some(p) = o.get("snapshotPriority") {
336 46 : let ok = p.as_i64().is_some_and(|n| (1..=10).contains(&n));
337 46 : if !ok {
338 2 : return Err(bad(
339 2 : "snapshotPriority must be an integer between 1 and 10 (5.2.41)".into(),
340 2 : ));
341 44 : }
342 130 : }
343 174 : if let Some(l) = o.get("snapshotLifetime") {
344 6 : let ok = l
345 6 : .as_str()
346 6 : .and_then(crate::entity_map::iso8601_secs)
347 6 : .is_some();
348 6 : if !ok {
349 4 : return Err(bad(
350 4 : "snapshotLifetime must be an ISO 8601 duration (5.2.41)".into(),
351 4 : ));
352 2 : }
353 168 : }
354 : // Table 5.2.41-1: endpoint is a "Dereferenceable URI", so a string that is
355 : // no URI at all names nothing the 5.16.6 notification could be sent to.
356 170 : if let Some(e) = o.get("endpoint") {
357 12 : let uri = e
358 12 : .as_str()
359 12 : .ok_or_else(|| bad("endpoint must be a URI string (5.2.41)".into()))?;
360 12 : antares_model::EntityId::new(uri)
361 12 : .map_err(|_| bad(format!("endpoint is not a valid URI: {uri:?} (5.2.41)")))?;
362 158 : }
363 : // Table 5.2.41-1: receiverInfo is a KeyValuePair[] (5.2.22 — both members
364 : // Strings), and 5.16.6 renders each pair as one header on the
365 : // SnapshotNotification POST. 6.3.8 binds those to RFC 7230 for the same
366 : // reason it binds a Subscription's: a pair that cannot be a header leaves
367 : // a Snapshot whose notification can only ever fail, silently.
368 168 : if let Some(ri) = o.get("receiverInfo") {
369 8 : let pairs = ri
370 8 : .as_array()
371 8 : .ok_or_else(|| bad("receiverInfo must be a KeyValuePair array (5.2.41)".into()))?;
372 6 : for kv in pairs {
373 6 : let (k, v) = (kv["key"].as_str(), kv["value"].as_str());
374 6 : if !k.is_some_and(crate::negotiate::is_field_name)
375 4 : || !v.is_some_and(crate::negotiate::is_field_value)
376 : {
377 6 : return Err(bad(format!(
378 6 : "receiverInfo entry {kv} is not a valid HTTP header (RFC 7230, 6.3.8)"
379 6 : )));
380 0 : }
381 : }
382 160 : }
383 160 : Ok(o)
384 212 : }
385 :
386 : /// Fresh metadata for a new snapshot (create or clone) — 5.16.1.4/5.16.2.4:
387 : /// timestamps now, status "preparing", priority default 5, bounded expiresAt.
388 140 : fn new_meta(
389 140 : mut o: Map<String, Value>,
390 140 : tenant: &TenantId,
391 140 : headers: &HeaderMap,
392 140 : ) -> Result<(String, Value), NgsiError> {
393 140 : let id = match o.get("id").and_then(Value::as_str) {
394 14 : Some(id) => {
395 14 : antares_model::EntityId::new(id)
396 14 : .map_err(|_| bad(format!("snapshot id is not a valid URI: {id:?}")))?;
397 14 : id.to_owned()
398 : }
399 : None => {
400 126 : let id = format!("urn:ngsi-ld:Snapshot:{}", uuid::Uuid::new_v4());
401 126 : o.insert("id".into(), Value::String(id.clone()));
402 126 : id
403 : }
404 : };
405 140 : let ts = now_iso();
406 140 : o.insert("type".into(), Value::String("Snapshot".into()));
407 140 : o.insert("createdAt".into(), Value::String(ts.clone()));
408 140 : o.insert("modifiedAt".into(), Value::String(ts));
409 140 : o.insert("snapshotStatus".into(), Value::String("preparing".into()));
410 : // 5.2.41 Table 5.2.41-2: lastUsedAt "is initialized at creation time"
411 140 : o.insert("lastUsedAt".into(), Value::String(now_iso()));
412 140 : o.insert("expiresAt".into(), Value::String(expires_at(&o)?));
413 140 : o.entry("snapshotPriority".to_owned())
414 140 : .or_insert(Value::Number(5.into()));
415 140 : o.insert(
416 140 : "__tenant".into(),
417 140 : Value::String(format!("snap-{}", uuid::Uuid::new_v4().simple())),
418 : );
419 : // Whose snapshot this is. 5.16.1's fill runs after the request has been
420 : // answered, so the subject it copies under has to be recorded here or it
421 : // is gone; and 5.16.1/5.16.2/5.16.7 act on everything the tenant holds,
422 : // which is why a narrowing answer to any of them is a refusal
423 : // (policy::WHOLE_TENANT) rather than a smaller snapshot.
424 6 : if let Some(subject) =
425 140 : crate::policy::subject_member(&crate::policy::subject_of(tenant, headers))
426 6 : {
427 6 : o.insert(crate::policy::SUBJECT_MEMBER.into(), subject);
428 134 : }
429 140 : Ok((id, Value::Object(o)))
430 140 : }
431 :
432 : /// 5.2.41: lastUsedAt tracks "the point in time when the snapshot was most
433 : /// recently used" — refreshed on every snapshot-scoped operation.
434 76 : async fn snap_touch(st: &AppState, tenant: &TenantId, id: &str) {
435 76 : let _ = st
436 76 : .store
437 76 : .mutate(tenant, Kind::Snapshot, id, |meta| {
438 76 : if let Some(o) = meta.as_object_mut() {
439 76 : o.insert("lastUsedAt".into(), Value::String(now_iso()));
440 76 : }
441 76 : Ok::<_, std::convert::Infallible>(())
442 76 : })
443 76 : .await;
444 76 : }
445 :
446 : /// Reverse lookup: which (owner tenant, snapshot id) does a synthetic
447 : /// "snap-…" tenant belong to? Used to stamp NGSILD-Snapshot on
448 : /// notifications from snapshot-scoped subscriptions (6.3.22) without
449 : /// leaking the internal tenant.
450 1068 : pub(crate) async fn snapshot_of_synth(st: &AppState, synth: &str) -> Option<(TenantId, String)> {
451 : // only synthetic tenants are indexed; every other tenant skips the lookup
452 1068 : if !synth.starts_with("snap-") {
453 1064 : return None;
454 4 : }
455 4 : let idx = snap_index_tenant()?;
456 4 : let doc = st
457 4 : .store
458 4 : .get(&idx, Kind::Snapshot, synth)
459 4 : .await
460 4 : .ok()
461 4 : .flatten()?;
462 4 : let owner = TenantId::new(doc.get("tenant")?.as_str()?).ok()?;
463 4 : Some((owner, doc.get("snapshot")?.as_str()?.to_owned()))
464 1068 : }
465 :
466 : /// The tenant an operation is judged under. 6.3.22 scopes a request to a
467 : /// Snapshot by swapping the tenant for the snapshot's internal one, so the
468 : /// handlers below it serve the frozen copy; the subject asking is still the
469 : /// owner tenant's. An engine keyed by tenant — which is the shape every
470 : /// rule set has — would otherwise be asked about an id no deployment ever
471 : /// wrote a rule for, and every snapshot-scoped read would step out from
472 : /// under its rules. Only a synthetic tenant costs the lookup.
473 684 : pub(crate) async fn asking_tenant(st: &AppState, tenant: &TenantId) -> TenantId {
474 684 : match snapshot_of_synth(st, tenant.as_str()).await {
475 2 : Some((owner, _)) => owner,
476 682 : None => tenant.clone(),
477 : }
478 684 : }
479 :
480 : /// 5.5.15: "If an implementation determines that it is low on resources,
481 : /// it may delete one or more snapshots", considering snapshotPriority
482 : /// (lowest first; earliest expiresAt breaks ties). The resource signal is
483 : /// the per-tenant registry cap (AppState.snapshot_cap); the just-created
484 : /// snapshot is never the victim. Evicted snapshots with an endpoint are
485 : /// notified with expiresAt set before notifiedAt — the 5.3.4 deletion
486 : /// encoding.
487 136 : async fn evict_over_cap(st: &AppState, tenant: &TenantId, keep: &str) {
488 4 : let victims: Vec<Value> = {
489 136 : let mut metas: Vec<Value> = st
490 136 : .store
491 136 : .list(tenant, Kind::Snapshot)
492 136 : .await
493 136 : .unwrap_or_default()
494 136 : .into_iter()
495 136 : .filter(is_snapshot)
496 136 : .collect();
497 136 : if metas.len() <= st.snapshot_cap {
498 132 : return;
499 4 : }
500 4 : let over = metas.len() - st.snapshot_cap;
501 22 : metas.sort_by_key(|v| {
502 22 : (
503 22 : v.get("snapshotPriority")
504 22 : .and_then(Value::as_i64)
505 22 : .unwrap_or(5),
506 22 : v.get("expiresAt")
507 22 : .and_then(Value::as_str)
508 22 : .unwrap_or("")
509 22 : .to_owned(),
510 22 : )
511 22 : });
512 4 : metas
513 4 : .into_iter()
514 4 : .filter(|v| v.get("id").and_then(Value::as_str) != Some(keep))
515 4 : .take(over)
516 4 : .collect()
517 : };
518 4 : for mut meta in victims {
519 4 : if let Some(id) = meta.get("id").and_then(Value::as_str).map(str::to_owned) {
520 4 : snap_remove(st, tenant, &id, &meta).await;
521 0 : }
522 4 : if let Some(o) = meta.as_object_mut() {
523 4 : // deletion signal: expiresAt strictly before the notification
524 4 : o.insert("expiresAt".into(), Value::String(now_iso()));
525 4 : }
526 4 : let st2 = st.clone();
527 4 : crate::spawn(async move {
528 4 : send_notification(&st2, &meta).await;
529 4 : });
530 : }
531 136 : }
532 :
533 : // ---------- 5.16.1 Create Snapshot (POST /snapshots, 6.36.3.1) ----------
534 :
535 178 : pub async fn create_snapshot(
536 178 : State(st): State<AppState>,
537 178 : CleanParams(params): CleanParams,
538 178 : headers: HeaderMap,
539 178 : body: Bytes,
540 178 : ) -> Response {
541 178 : let go = async {
542 178 : let tenant = tenant_from(&headers)?;
543 178 : check_params(¶ms, &["local"])?;
544 178 : gate!(st, &tenant, &headers, "5.16.1").await?;
545 : // 6.3.5: the @context rules apply to the snapshot body like to any
546 : // other POST payload (media type, Link header, body @context)
547 176 : let v = parse_body(&st.loader, &headers, &body, BodyKind::Standard)
548 176 : .await?
549 : .value;
550 164 : let o = validate(&v, Mode::Create)?;
551 132 : let (id, meta) = new_meta(o, &tenant, &headers)?;
552 132 : snap_insert(&st, &tenant, &id, &meta).await?;
553 128 : evict_over_cap(&st, &tenant, &id).await;
554 128 : let (st2, t2, id2) = (st.clone(), tenant.clone(), id.clone());
555 128 : crate::spawn(async move {
556 128 : fill_snapshot(&st2, &t2, &id2).await;
557 128 : });
558 128 : Ok::<_, ApiError>(created(format!("/ngsi-ld/v1/snapshots/{id}"), &tenant))
559 178 : };
560 178 : go.await.unwrap_or_else(|e| e.into_response())
561 178 : }
562 :
563 : /// 5.16.1.4 background fill: execute every (temporal) query, store the
564 : /// results under the synthetic tenant, derive the status, notify.
565 128 : async fn fill_snapshot(st: &AppState, tenant: &TenantId, id: &str) {
566 128 : let Some(meta) = snap_get(st, tenant, id).await else {
567 0 : return;
568 : };
569 128 : let Some(synth) = synth_tenant(&meta) else {
570 0 : return;
571 : };
572 128 : let ctx = st.loader.core();
573 128 : let (mut n_fail, mut n_res, mut n_empty, mut copied) = (0usize, 0usize, 0usize, 0usize);
574 132 : let mut detail = |r: Result<usize, NgsiError>| -> Value {
575 132 : match r {
576 : Ok(0) => {
577 46 : n_empty += 1;
578 46 : json!({"resultStatus": "empty"})
579 : }
580 : Ok(_) => {
581 80 : n_res += 1;
582 80 : json!({"resultStatus": "success"})
583 : }
584 6 : Err(e) => {
585 6 : n_fail += 1;
586 6 : json!({"resultStatus": "failure",
587 6 : "problemDetails": crate::negotiate::problem_value(&e)})
588 : }
589 : }
590 132 : };
591 128 : let cap = fill_cap(st);
592 128 : let mut q_details = Vec::new();
593 128 : for q in meta
594 128 : .get("snapshotQueries")
595 128 : .and_then(Value::as_array)
596 128 : .into_iter()
597 128 : .flatten()
598 : {
599 122 : if fill_cancelled(st, tenant, id, &synth).await {
600 0 : return;
601 122 : }
602 122 : let r = run_query(st, tenant, &synth, q, &ctx, cap - copied).await;
603 122 : if let Ok(n) = &r {
604 116 : copied += n;
605 116 : }
606 122 : q_details.push(detail(r));
607 : }
608 128 : let mut tq_details = Vec::new();
609 128 : for q in meta
610 128 : .get("snapshotTemporalQueries")
611 128 : .and_then(Value::as_array)
612 128 : .into_iter()
613 128 : .flatten()
614 : {
615 10 : if fill_cancelled(st, tenant, id, &synth).await {
616 0 : return;
617 10 : }
618 10 : let r = run_temporal_query(st, tenant, &synth, id, q, cap - copied).await;
619 10 : if let Ok(n) = &r {
620 10 : copied += n;
621 10 : }
622 10 : tq_details.push(detail(r));
623 : }
624 128 : if fill_cancelled(st, tenant, id, &synth).await {
625 0 : return;
626 128 : }
627 128 : let status = if n_fail == 0 && n_empty == 0 && n_res > 0 {
628 76 : "success"
629 52 : } else if n_res > 0 {
630 0 : "partial"
631 52 : } else if n_empty > 0 {
632 46 : "empty"
633 : } else {
634 6 : "failure"
635 : };
636 128 : finish(st, tenant, id, status, Some((q_details, tq_details))).await;
637 128 : }
638 :
639 : /// One 5.2.23 Query executed per the DISTRIBUTED query behaviour
640 : /// (5.16.1.4 -> 5.7.2.4): local content plus every matching Context
641 : /// Source; results are copied into the snapshot's synthetic tenant. Every
642 : /// page is retrieved ("all pages are to be retrieved completely") within
643 : /// `budget` documents — beyond it nothing is copied (5.5.6).
644 122 : async fn run_query(
645 122 : st: &AppState,
646 122 : tenant: &TenantId,
647 122 : synth: &TenantId,
648 122 : q: &Value,
649 122 : ctx: &antares_jsonld::Context,
650 122 : budget: usize,
651 122 : ) -> Result<usize, NgsiError> {
652 122 : let qo = q
653 122 : .as_object()
654 122 : .ok_or_else(|| bad("Query must be an object".into()))?;
655 122 : let mut vp: HashMap<String, String> = HashMap::new();
656 122 : crate::paging::query_doc_params(qo, false, &mut vp)?;
657 366 : for k in ["limit", "offset", "count"] {
658 366 : vp.remove(k);
659 366 : }
660 122 : let headers = HeaderMap::new();
661 122 : let mut warnings = Vec::new();
662 122 : let fed = if crate::federation::active(&vp) {
663 122 : crate::federation::fed_query(st, tenant, &headers, ctx, &vp, &mut warnings).await?
664 : } else {
665 0 : Vec::new()
666 : };
667 : // 5.16.1.4: "If the size of the respective results require pagination,
668 : // all pages are to be retrieved completely." They are retrieved one at a
669 : // time. A query wider than the budget is refused, and it has to be
670 : // refused without first holding its whole match set in memory to measure
671 : // it: the store's pre-LIMIT count is that measurement. The count is the
672 : // match set's only while every predicate reached the store — an
673 : // idPattern is applied after it (5.2.33), so with one present the
674 : // documents that survive are counted instead.
675 122 : let exact_total = !vp.contains_key("idPattern");
676 : // Federated candidates are merged after the store answers (5.7.2.4), so
677 : // that round takes no page and is the whole match set; only a local
678 : // query walks on.
679 122 : let mut fed = fed;
680 122 : let mut n = 0usize;
681 122 : let mut offset = 0usize;
682 : loop {
683 : // Only the first round asks the store to count: the pre-LIMIT total
684 : // makes it visit the whole match set, and the budget needs that once.
685 : // Later rounds carry the next-page hint in the same member, which is
686 : // not a count and is not read as one.
687 128 : let counted = offset == 0;
688 128 : if counted {
689 122 : vp.insert("count".into(), "true".into());
690 122 : } else {
691 6 : vp.remove("count");
692 6 : }
693 128 : let batch = crate::entities::filter_entities_paged(
694 128 : st,
695 128 : tenant,
696 128 : &vp,
697 128 : ctx,
698 128 : std::mem::take(&mut fed),
699 128 : Some((offset, st.max_limit)),
700 128 : None,
701 128 : )
702 128 : .await
703 128 : .map_err(|e| match e {
704 0 : ApiError::Ngsi(n) => n,
705 0 : other => opaque("query execution", &other),
706 0 : })?;
707 128 : if exact_total && counted {
708 122 : if let Some(total) = batch.total.filter(|_| batch.paged) {
709 8 : if total > budget {
710 0 : return Err(too_many(total, budget));
711 8 : }
712 114 : }
713 6 : }
714 128 : let rows = batch.rows;
715 128 : let got = batch.docs.len();
716 128 : if n + got > budget {
717 4 : return Err(too_many(n + got, budget));
718 124 : }
719 : // A write the store refuses fails the query. What this returns is
720 : // what the caller adds to `copied`, and `copied` decides both the
721 : // budget left for the next query and whether the synthetic tenant is
722 : // materialized — an empty snapshot has to read as empty rather than
723 : // as a tenant that never existed. Counting a copy that never happened
724 : // costs the snapshot both, and 5.2.41 would publish the query as a
725 : // success besides.
726 124 : for doc in batch.docs {
727 112 : if let Some(id) = doc.get("id").and_then(Value::as_str) {
728 112 : st.store
729 112 : .create(synth, Kind::Entity, id, doc.clone())
730 112 : .await?;
731 0 : }
732 : }
733 122 : n += got;
734 122 : match crate::entities::next_scan_offset(offset, rows, 0, batch.paged, st.max_limit) {
735 6 : Some(next) => offset = next,
736 116 : None => break,
737 : }
738 : }
739 116 : Ok(n)
740 122 : }
741 :
742 : /// One temporal Query (temporalQ mandatory) — ids via the 5.7.4.4 path,
743 : /// full evolutions copied via the store, every page retrieved but never
744 : /// more than `budget` evolutions (5.5.6).
745 10 : async fn run_temporal_query(
746 10 : st: &AppState,
747 10 : tenant: &TenantId,
748 10 : synth: &TenantId,
749 10 : id: &str,
750 10 : q: &Value,
751 10 : budget: usize,
752 10 : ) -> Result<usize, NgsiError> {
753 10 : let qo = q
754 10 : .as_object()
755 10 : .ok_or_else(|| bad("Query must be an object".into()))?;
756 10 : let mut vp: HashMap<String, String> = HashMap::new();
757 10 : crate::paging::query_doc_params(qo, true, &mut vp)?;
758 30 : for k in ["limit", "offset", "count"] {
759 30 : vp.remove(k);
760 30 : }
761 10 : let mut headers = HeaderMap::new();
762 10 : if let Ok(v) = tenant.as_str().parse() {
763 10 : headers.insert("NGSILD-Tenant", v);
764 10 : }
765 : // 5.16.1.4: "If the size of the respective results require pagination,
766 : // all pages are to be retrieved completely."
767 10 : let mut n = 0usize;
768 10 : let mut offset = 0usize;
769 : loop {
770 : // a snapshot deleted (or evicted) mid-fill stops the paging; what is
771 : // already copied is reaped by the caller
772 12 : if snap_get(st, tenant, id).await.is_none() {
773 0 : break;
774 12 : }
775 12 : vp.insert("limit".into(), st.max_limit.to_string());
776 12 : vp.insert("offset".into(), offset.to_string());
777 : // The fill runs after the request that created the snapshot has
778 : // been answered, under a header map that is not the client's, so
779 : // the narrowing the creating subject was given is not in scope here
780 : // — a snapshot under a policy is P5's box (ADR-0020: a `Filter` on
781 : // a fill is a `Deny`, and the fill records whose it is).
782 : // The merged documents come back beside the response: an Evolution
783 : // this broker holds is copied from the store below, whole, and one a
784 : // registered Context Source holds exists only here.
785 12 : let mut merged: Vec<Value> = Vec::new();
786 12 : let resp = crate::temporal::query_temporal_collected(
787 12 : st,
788 12 : &vp,
789 12 : &headers,
790 12 : &crate::policy::Filter::default(),
791 12 : Some(&mut merged),
792 12 : )
793 12 : .await
794 12 : .map_err(|e| match e {
795 0 : ApiError::Ngsi(n) => n,
796 0 : other => opaque("temporal query execution", &other),
797 0 : })?;
798 12 : let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
799 12 : .await
800 12 : .map_err(|e| opaque("temporal result read", &e))?;
801 12 : let arr: Vec<Value> = serde_json::from_slice::<Value>(&bytes)
802 12 : .ok()
803 12 : .and_then(|v| v.as_array().cloned())
804 12 : .unwrap_or_default();
805 12 : let got = arr.len();
806 12 : if n + got > budget {
807 0 : return Err(too_many(n + got, budget));
808 12 : }
809 16 : for d in arr {
810 16 : let Some(eid) = d.get("id").and_then(Value::as_str) else {
811 0 : continue;
812 : };
813 : // `None` is an Entity with no Temporal Evolution to copy; an Err
814 : // is the store refusing the read, which is not the same thing and
815 : // is not this snapshot's to swallow.
816 : //
817 : // The copy REPLACES: a snapshot carrying both kinds of query runs
818 : // its Entity queries first, and copying an Entity into the
819 : // synthetic tenant fires the 5.6.11 auto-record hook, which leaves
820 : // a Temporal Evolution holding only the instant of the copy. An
821 : // insert-if-absent would find that stub and keep it, so the
822 : // snapshot would answer temporal reads with one instance of a
823 : // history the source has many of — and report the query a success.
824 16 : let held = st
825 16 : .temporal
826 16 : .get_temporal(
827 16 : tenant,
828 16 : eid,
829 16 : &antares_store::filter::TemporalFilter::default(),
830 16 : )
831 16 : .await?;
832 : // Nothing local means the Evolution came from a registered
833 : // Context Source (5.7.4.4). The query's merged answer is every
834 : // instance of it this broker was given, so it is what the
835 : // snapshot keeps — a local copy is the source's whole history,
836 : // a remote one is as much of it as the fan-out returned.
837 16 : let doc = match held {
838 14 : Some(doc) => Some(doc),
839 2 : None => merged
840 2 : .iter()
841 4 : .find(|m| m.get("id").and_then(Value::as_str) == Some(eid))
842 2 : .cloned(),
843 : };
844 16 : if let Some(doc) = doc {
845 16 : st.temporal.upsert(synth, eid, doc).await?;
846 16 : n += 1;
847 0 : }
848 : }
849 12 : if got < st.max_limit {
850 10 : break;
851 2 : }
852 2 : offset += st.max_limit;
853 : }
854 10 : Ok(n)
855 10 : }
856 :
857 136 : async fn finish(
858 136 : st: &AppState,
859 136 : tenant: &TenantId,
860 136 : id: &str,
861 136 : status: &str,
862 136 : details: Option<(Vec<Value>, Vec<Value>)>,
863 136 : ) {
864 136 : let Some(mut meta) = snap_get(st, tenant, id).await else {
865 0 : return;
866 : };
867 136 : if let Some(o) = meta.as_object_mut() {
868 136 : o.insert("snapshotStatus".into(), Value::String(status.into()));
869 136 : o.insert("modifiedAt".into(), Value::String(now_iso()));
870 136 : if let Some((q, tq)) = details {
871 128 : if !q.is_empty() {
872 122 : o.insert("snapshotQueriesDetails".into(), Value::Array(q));
873 122 : }
874 128 : if !tq.is_empty() {
875 10 : o.insert("snapshotTemporalQueriesDetails".into(), Value::Array(tq));
876 118 : }
877 8 : }
878 0 : }
879 136 : snap_put(st, tenant, meta.clone()).await;
880 136 : send_notification(st, &meta).await;
881 136 : }
882 :
883 : /// 5.16.6 / 5.3.4 SnapshotNotification (sent only when endpoint is set).
884 146 : async fn send_notification(st: &AppState, meta: &Value) {
885 146 : let Some(uri) = meta.get("endpoint").and_then(Value::as_str) else {
886 144 : return;
887 : };
888 2 : if st.egress.check_url(uri).await.is_err() {
889 0 : return;
890 2 : }
891 2 : let mut body = json!({
892 2 : "id": format!("urn:ngsi-ld:SnapshotNotification:{}", uuid::Uuid::new_v4()),
893 2 : "type": "SnapshotNotification",
894 2 : "notifiedAt": now_iso(),
895 2 : "snapshotId": meta.get("id").cloned().unwrap_or_default(),
896 2 : "snapshotStatus": meta.get("snapshotStatus").cloned().unwrap_or_default(),
897 2 : "snapshotPriority": meta.get("snapshotPriority").cloned().unwrap_or_default(),
898 2 : "expiresAt": meta.get("expiresAt").cloned().unwrap_or_default(),
899 : });
900 : // 5.3.4 Table 5.3.4-1 names the temporal list
901 : // "temporalSnapshotQueriesDetails" (unlike 5.2.41's
902 : // "snapshotTemporalQueriesDetails") — the notification datatype's own
903 : // table governs the notification payload.
904 4 : for (from, to) in [
905 2 : ("snapshotQueriesDetails", "snapshotQueriesDetails"),
906 2 : (
907 2 : "snapshotTemporalQueriesDetails",
908 2 : "temporalSnapshotQueriesDetails",
909 2 : ),
910 2 : ] {
911 4 : if let Some(v) = meta.get(from) {
912 4 : body[to] = v.clone();
913 4 : }
914 : }
915 2 : let mut req = st.http.post(uri).header("Content-Type", "application/json");
916 2 : if let Some(ri) = meta.get("receiverInfo").and_then(Value::as_array) {
917 0 : for kv in ri {
918 0 : if let (Some(k), Some(v)) = (
919 0 : kv.get("key").and_then(Value::as_str),
920 0 : kv.get("value").and_then(Value::as_str),
921 0 : ) {
922 0 : req = req.header(k, v);
923 0 : }
924 : }
925 2 : }
926 2 : let req = req.body(serde_json::to_vec(&body).unwrap_or_default());
927 2 : let _ = antares_jsonld::io_deadline(req.send(), 8_000).await;
928 146 : }
929 :
930 : // ---------- 5.16.7 Purge Snapshots (DELETE /snapshots, 6.36.3.2) ----------
931 :
932 46 : pub async fn purge_snapshots(
933 46 : State(st): State<AppState>,
934 46 : CleanParams(params): CleanParams,
935 46 : headers: HeaderMap,
936 46 : ) -> Response {
937 46 : let go = async {
938 46 : let tenant = tenant_from(&headers)?;
939 46 : check_params(¶ms, &["q", "local"])?;
940 46 : gate!(st, &tenant, &headers, "5.16.7").await?;
941 : // 5.16.7.4: the query is mandatory and restricted to Snapshot members
942 44 : let q = params
943 44 : .get("q")
944 44 : .ok_or_else(|| bad("purge requires a q over Snapshot members (5.16.7.4)".into()))?;
945 42 : let ast = antares_ql::parse_q(q)?;
946 : // 5.16.7.4: the query is "restricted to members of the Snapshot
947 : // data type" (Tables 5.2.41-1/-2)
948 : const MEMBERS: [&str; 15] = [
949 : "id",
950 : "type",
951 : "snapshotQueries",
952 : "snapshotTemporalQueries",
953 : "snapshotLifetime",
954 : "snapshotPriority",
955 : "endpoint",
956 : "receiverInfo",
957 : "snapshotStatus",
958 : "snapshotQueriesDetails",
959 : "snapshotTemporalQueriesDetails",
960 : "createdAt",
961 : "modifiedAt",
962 : "expiresAt",
963 : "lastUsedAt",
964 : ];
965 42 : if let Some(alien) = ast
966 42 : .attribute_paths()
967 42 : .into_iter()
968 42 : .find(|a| !MEMBERS.contains(a))
969 : {
970 12 : return Err(bad(format!(
971 12 : "purge q is restricted to Snapshot members (5.16.7.4): {alien:?}"
972 12 : ))
973 12 : .into());
974 30 : }
975 30 : let ctx = st.loader.core();
976 30 : let victims: Vec<Value> = st
977 30 : .store
978 30 : .list(&tenant, Kind::Snapshot)
979 30 : .await
980 30 : .unwrap_or_default()
981 30 : .into_iter()
982 30 : .filter(|meta| is_snapshot(meta) && crate::registry::csf_matches(&ast, meta, &ctx))
983 30 : .collect();
984 30 : for meta in victims {
985 8 : if let Some(id) = meta.get("id").and_then(Value::as_str) {
986 8 : snap_remove(&st, &tenant, id, &meta).await;
987 0 : }
988 : }
989 30 : Ok::<_, ApiError>(no_content(&tenant))
990 46 : };
991 46 : go.await.unwrap_or_else(|e| e.into_response())
992 46 : }
993 :
994 : /// 5.16.2.4 / 5.16.3.4 / 5.16.4.4 / 5.16.5.4: every /snapshots/{id} method
995 : /// opens the same way — the tenant it runs in, `local` (6.3.18) as the only
996 : /// parameter this resource takes, and an id that must be a valid URI before
997 : /// the store is touched at all.
998 306 : fn open_snapshot(
999 306 : params: &HashMap<String, String>,
1000 306 : headers: &HeaderMap,
1001 306 : id: &str,
1002 306 : ) -> ApiResult<TenantId> {
1003 306 : let tenant = tenant_from(headers)?;
1004 306 : check_params(params, &["local"])?;
1005 306 : antares_model::EntityId::new(id)
1006 306 : .map_err(|_| bad(format!("snapshot id is not a valid URI: {id:?}")))?;
1007 266 : Ok(tenant)
1008 306 : }
1009 :
1010 : // ---------- 5.16.3/5.16.4/5.16.5: /snapshots/{id} (6.37) ----------
1011 :
1012 200 : pub async fn retrieve_snapshot(
1013 200 : State(st): State<AppState>,
1014 200 : Path(id): Path<String>,
1015 200 : CleanParams(params): CleanParams,
1016 200 : headers: HeaderMap,
1017 200 : ) -> Response {
1018 200 : let go = async {
1019 200 : let tenant = open_snapshot(¶ms, &headers, &id)?;
1020 190 : gate!(st, &tenant, &headers, "5.16.3", ids: &[&id]).await?;
1021 190 : let accept = parse_accept(&headers)?;
1022 190 : let meta = snap_get(&st, &tenant, &id)
1023 190 : .await
1024 190 : .ok_or_else(|| NgsiError::ResourceNotFound(format!("snapshot {id} not found")))?;
1025 164 : let ctx = st.loader.core();
1026 164 : Ok::<_, ApiError>(respond(
1027 164 : StatusCode::OK,
1028 164 : present(&meta),
1029 164 : &ctx,
1030 164 : accept,
1031 164 : &tenant,
1032 164 : ))
1033 200 : };
1034 200 : go.await.unwrap_or_else(|e| e.into_response())
1035 200 : }
1036 :
1037 36 : pub async fn update_snapshot(
1038 36 : State(st): State<AppState>,
1039 36 : Path(id): Path<String>,
1040 36 : CleanParams(params): CleanParams,
1041 36 : headers: HeaderMap,
1042 36 : body: Bytes,
1043 36 : ) -> Response {
1044 36 : let go = async {
1045 36 : let tenant = open_snapshot(¶ms, &headers, &id)?;
1046 26 : gate!(st, &tenant, &headers, "5.16.4", ids: &[&id]).await?;
1047 26 : let accept = parse_accept(&headers)?;
1048 : // 5.16.4.4 binds this operation to 5.5.4, hence to the 6.3.5 rules
1049 26 : let v = parse_body(&st.loader, &headers, &body, BodyKind::Standard)
1050 26 : .await?
1051 : .value;
1052 22 : let frag = validate(&v, Mode::Update)?;
1053 16 : let mut meta = snap_get(&st, &tenant, &id)
1054 16 : .await
1055 16 : .ok_or_else(|| NgsiError::ResourceNotFound(format!("snapshot {id} not found")))?;
1056 6 : if let Some(o) = meta.as_object_mut() {
1057 : // 5.16.4.4 / 5.5.8 merge of the updatable members
1058 24 : for k in [
1059 6 : "snapshotLifetime",
1060 6 : "snapshotPriority",
1061 6 : "endpoint",
1062 6 : "receiverInfo",
1063 6 : ] {
1064 24 : match frag.get(k) {
1065 18 : None => {}
1066 0 : Some(Value::Null) => {
1067 0 : o.remove(k);
1068 0 : }
1069 6 : Some(v) => {
1070 6 : o.insert(k.into(), v.clone());
1071 6 : }
1072 : }
1073 : }
1074 6 : if frag.contains_key("snapshotLifetime") {
1075 : // "it is possible to indirectly update expiresAt" (5.2.41)
1076 0 : o.insert("expiresAt".into(), Value::String(expires_at(o)?));
1077 6 : }
1078 6 : o.insert("modifiedAt".into(), Value::String(now_iso()));
1079 0 : }
1080 6 : snap_put(&st, &tenant, meta.clone()).await;
1081 : // 5.16.6: notifications are also sent after any status update
1082 6 : let (st2, meta2) = (st.clone(), meta.clone());
1083 6 : crate::spawn(async move {
1084 6 : send_notification(&st2, &meta2).await;
1085 6 : });
1086 6 : let ctx = st.loader.core();
1087 6 : Ok::<_, ApiError>(respond(
1088 6 : StatusCode::OK,
1089 6 : present(&meta),
1090 6 : &ctx,
1091 6 : accept,
1092 6 : &tenant,
1093 6 : ))
1094 36 : };
1095 36 : go.await.unwrap_or_else(|e| e.into_response())
1096 36 : }
1097 :
1098 40 : pub async fn delete_snapshot(
1099 40 : State(st): State<AppState>,
1100 40 : Path(id): Path<String>,
1101 40 : CleanParams(params): CleanParams,
1102 40 : headers: HeaderMap,
1103 40 : ) -> Response {
1104 40 : let go = async {
1105 40 : let tenant = open_snapshot(¶ms, &headers, &id)?;
1106 30 : gate!(st, &tenant, &headers, "5.16.5", ids: &[&id]).await?;
1107 30 : let meta = snap_get(&st, &tenant, &id)
1108 30 : .await
1109 30 : .ok_or_else(|| NgsiError::ResourceNotFound(format!("snapshot {id} not found")))?;
1110 22 : snap_remove(&st, &tenant, &id, &meta).await;
1111 22 : Ok::<_, ApiError>(no_content(&tenant))
1112 40 : };
1113 40 : go.await.unwrap_or_else(|e| e.into_response())
1114 40 : }
1115 :
1116 : // ---------- 5.16.2 Clone Snapshot (POST /snapshots/{id}/clone, 6.38) ----------
1117 :
1118 30 : pub async fn clone_snapshot(
1119 30 : State(st): State<AppState>,
1120 30 : Path(id): Path<String>,
1121 30 : CleanParams(params): CleanParams,
1122 30 : headers: HeaderMap,
1123 30 : body: Bytes,
1124 30 : ) -> Response {
1125 30 : let go = async {
1126 30 : let tenant = open_snapshot(¶ms, &headers, &id)?;
1127 20 : gate!(st, &tenant, &headers, "5.16.2", ids: &[&id]).await?;
1128 18 : let src = snap_get(&st, &tenant, &id)
1129 18 : .await
1130 18 : .ok_or_else(|| NgsiError::ResourceNotFound(format!("snapshot {id} not found")))?;
1131 : // 5.16.2.3 makes the clone body optional; a body that IS sent obeys
1132 : // the 6.3.5 @context rules
1133 10 : let v: Value = if body.is_empty() {
1134 0 : json!({})
1135 : } else {
1136 10 : parse_body(&st.loader, &headers, &body, BodyKind::Standard)
1137 10 : .await?
1138 : .value
1139 : };
1140 10 : let mut o = validate(&v, Mode::Clone)?;
1141 : // the clone carries the source's (read-only) query lineage
1142 16 : for k in ["snapshotQueries", "snapshotTemporalQueries"] {
1143 16 : if let Some(qv) = src.get(k) {
1144 8 : o.insert(k.into(), qv.clone());
1145 8 : }
1146 : }
1147 8 : let (new_id, meta) = new_meta(o, &tenant, &headers)?;
1148 8 : snap_insert(&st, &tenant, &new_id, &meta).await?;
1149 : // a clone is a new snapshot: 5.5.15 resource pressure applies to it
1150 : // exactly as it does to a create, so cloning cannot grow the
1151 : // registry past the cap
1152 8 : evict_over_cap(&st, &tenant, &new_id).await;
1153 8 : let (st2, t2, sid, nid) = (st.clone(), tenant.clone(), id.clone(), new_id.clone());
1154 8 : crate::spawn(async move {
1155 8 : clone_fill(&st2, &t2, &sid, &nid).await;
1156 8 : });
1157 8 : Ok::<_, ApiError>(created(format!("/ngsi-ld/v1/snapshots/{new_id}"), &tenant))
1158 30 : };
1159 30 : go.await.unwrap_or_else(|e| e.into_response())
1160 30 : }
1161 :
1162 : /// 5.16.2.4 background copy: all Entity and Temporal data of the source.
1163 8 : async fn clone_fill(st: &AppState, tenant: &TenantId, src_id: &str, new_id: &str) {
1164 8 : let (Some(src), Some(new)) = (
1165 8 : snap_get(st, tenant, src_id).await,
1166 8 : snap_get(st, tenant, new_id).await,
1167 : ) else {
1168 0 : finish(st, tenant, new_id, "failure", None).await;
1169 0 : return;
1170 : };
1171 8 : let (Some(from), Some(to)) = (synth_tenant(&src), synth_tenant(&new)) else {
1172 0 : finish(st, tenant, new_id, "failure", None).await;
1173 0 : return;
1174 : };
1175 8 : let mut failed = false;
1176 8 : match st.store.list(&from, Kind::Entity).await {
1177 8 : Ok(docs) => {
1178 10 : for doc in docs {
1179 10 : if let Some(id) = doc.get("id").and_then(Value::as_str) {
1180 10 : if st
1181 10 : .store
1182 10 : .create(&to, Kind::Entity, id, doc.clone())
1183 10 : .await
1184 10 : .is_err()
1185 0 : {
1186 0 : failed = true;
1187 10 : }
1188 0 : }
1189 : }
1190 : }
1191 0 : Err(_) => failed = true,
1192 : }
1193 8 : match st.temporal.list(&from).await {
1194 8 : Ok(docs) => {
1195 10 : for doc in docs {
1196 10 : if let Some(id) = doc.get("id").and_then(Value::as_str) {
1197 10 : if st.temporal.create(&to, id, doc.clone()).await.is_err() {
1198 0 : failed = true;
1199 10 : }
1200 0 : }
1201 : }
1202 : }
1203 0 : Err(_) => failed = true,
1204 : }
1205 8 : if fill_cancelled(st, tenant, new_id, &to).await {
1206 0 : return;
1207 8 : }
1208 8 : finish(
1209 8 : st,
1210 8 : tenant,
1211 8 : new_id,
1212 8 : if failed { "failure" } else { "success" },
1213 8 : None,
1214 : )
1215 8 : .await;
1216 8 : }
1217 :
1218 : // ---------- 6.3.22: NGSILD-Snapshot scoping middleware ----------
1219 :
1220 : /// 6.3.22 / 5.5.15: resolve the NGSILD-Snapshot header to the snapshot's
1221 : /// synthetic tenant, so every Core/Temporal handler serves the frozen copy;
1222 : /// the header is echoed on the response (and the synthetic tenant never
1223 : /// leaks into NGSILD-Tenant).
1224 26078 : pub async fn snapshot_layer(
1225 26078 : State(st): State<AppState>,
1226 26078 : req: axum::extract::Request,
1227 26078 : next: axum::middleware::Next,
1228 26078 : ) -> Response {
1229 : // 6.3.22 gives the header one value, and it decides whether the request
1230 : // is answered from a frozen copy or from live data: an ambiguous one
1231 : // would serve the request against a dataset nobody named.
1232 26078 : let sid = match single_header(req.headers(), "NGSILD-Snapshot") {
1233 25978 : Ok(None) => return next.run(req).await,
1234 96 : Ok(Some(sid)) => sid,
1235 4 : Err(e) => return e.into_response(),
1236 : };
1237 : // 6.3.22: "If the HTTP header NGSILD-Snapshot is present in the HTTP
1238 : // request, it shall also be present in HTTP response" — every exit,
1239 : // the unscoped and the error ones included, leaves through here
1240 96 : let mut resp = scoped(&st, &sid, req, next).await;
1241 96 : if let Ok(v) = sid.parse() {
1242 96 : resp.headers_mut().insert("NGSILD-Snapshot", v);
1243 96 : }
1244 96 : resp
1245 26078 : }
1246 :
1247 96 : async fn scoped(
1248 96 : st: &AppState,
1249 96 : sid: &str,
1250 96 : mut req: axum::extract::Request,
1251 96 : next: axum::middleware::Next,
1252 96 : ) -> Response {
1253 : // the Snapshot API's own resources (6.36-6.38) are never
1254 : // snapshot-scoped — that is the /snapshots RESOURCE, not any path
1255 : // merely containing the word (an Attribute may be named "snapshots")
1256 96 : if req
1257 96 : .uri()
1258 96 : .path()
1259 96 : .trim_start_matches(API_ROOT)
1260 96 : .starts_with("/snapshots")
1261 : {
1262 4 : return next.run(req).await;
1263 92 : }
1264 92 : let tenant = match tenant_from(req.headers()) {
1265 92 : Ok(t) => t,
1266 0 : Err(e) => return e.into_response(),
1267 : };
1268 92 : let Some(meta) = snap_get(st, &tenant, sid).await else {
1269 16 : return ApiError::from(NgsiError::ResourceNotFound(format!(
1270 16 : "snapshot {sid} not found"
1271 16 : )))
1272 16 : .into_response();
1273 : };
1274 76 : let Some(synth) = synth_tenant(&meta) else {
1275 0 : return ApiError::from(NgsiError::InternalError("snapshot without tenant".into()))
1276 0 : .into_response();
1277 : };
1278 76 : snap_touch(st, &tenant, sid).await;
1279 76 : if let Ok(v) = synth.as_str().parse() {
1280 76 : req.headers_mut().insert("NGSILD-Tenant", v);
1281 76 : }
1282 76 : let mut resp = next.run(req).await;
1283 : // restore the caller's tenant view (the synthetic one is internal)
1284 76 : match tenant.as_str() {
1285 76 : "default" => {
1286 66 : resp.headers_mut().remove("NGSILD-Tenant");
1287 66 : }
1288 10 : t => {
1289 10 : if let Ok(v) = t.parse() {
1290 10 : resp.headers_mut().insert("NGSILD-Tenant", v);
1291 10 : }
1292 : }
1293 : }
1294 76 : resp
1295 96 : }
1296 :
1297 : /// 5.5.6: the fill copies its results into a second (isolated) tenant, so an
1298 : /// unbounded result set doubles the stored data set in one request. The
1299 : /// implementation threshold — "up to each implementation" — is a hundred
1300 : /// full pages per snapshot; beyond it the query reports TooManyResults.
1301 132 : fn fill_cap(st: &AppState) -> usize {
1302 132 : st.max_limit.saturating_mul(100)
1303 132 : }
1304 :
1305 : /// 5.16.1.4 + 5.5.15: the copied data is reachable only through its snapshot
1306 : /// document, so a fill whose snapshot was deleted (or evicted under resource
1307 : /// pressure) must stop and free what it already copied.
1308 276 : async fn fill_cancelled(st: &AppState, tenant: &TenantId, id: &str, synth: &TenantId) -> bool {
1309 276 : if snap_get(st, tenant, id).await.is_some() {
1310 272 : return false;
1311 4 : }
1312 4 : purge_synth(st, synth).await;
1313 4 : true
1314 276 : }
1315 :
1316 : /// 5.5.6: "When a query operation is producing so many results that can
1317 : /// potentially exhaust client or server resources … implementations shall
1318 : /// raise an error of type TooManyResults" — reported per query in the
1319 : /// 5.2.42 ExecutionResultDetails of the snapshot.
1320 4 : fn too_many(got: usize, budget: usize) -> NgsiError {
1321 4 : NgsiError::TooManyResults(format!(
1322 4 : "snapshot query yields {got} results, {budget} left of the snapshot ceiling"
1323 4 : ))
1324 4 : }
1325 :
1326 : #[cfg(test)]
1327 : mod clause_5_5_6 {
1328 : use super::*;
1329 :
1330 : /// 5.5.6 InternalError: the client-visible `detail` — served in
1331 : /// snapshotQueriesDetails.problemDetails and copied into the 5.3.4
1332 : /// SnapshotNotification body — must be generic. The Debug/IO text of
1333 : /// the underlying failure belongs in the server log only.
1334 : #[test]
1335 4 : fn snapshot_fill_error_detail_is_generic() {
1336 4 : let inner = ApiError::Bare(StatusCode::UNSUPPORTED_MEDIA_TYPE);
1337 4 : let pd = crate::negotiate::problem_value(&opaque("query execution", &inner));
1338 4 : let detail = pd["detail"].as_str().unwrap_or_default();
1339 4 : assert_eq!(detail, "query execution failed");
1340 4 : assert!(
1341 4 : !detail.contains("Bare") && !detail.contains("Unsupported"),
1342 : "internal error text leaked into the client-visible detail: {detail}"
1343 : );
1344 4 : }
1345 : }
1346 :
1347 : #[cfg(test)]
1348 : mod clause_5_16 {
1349 : use super::*;
1350 : use axum::body::Body;
1351 : use axum::http::Request;
1352 : use tower::ServiceExt;
1353 :
1354 40 : async fn state() -> AppState {
1355 40 : crate::wired_state("antares-snapshots").await
1356 40 : }
1357 :
1358 : /// 5.16 with RFC 9110 §9.2.1: reading an expired Snapshot must not
1359 : /// write. The meta row survives the read; the sweep is what frees it,
1360 : /// its reverse-index entry and the synthetic tenant's data.
1361 : #[tokio::test(flavor = "multi_thread")]
1362 4 : async fn an_expired_snapshot_is_refused_by_a_read_that_writes_nothing() {
1363 4 : let st = state().await;
1364 4 : let t = TenantId::default();
1365 4 : let id = "urn:ngsi-ld:Snapshot:stale";
1366 4 : let synth = "snap-default-stale";
1367 4 : st.store
1368 4 : .create(
1369 4 : &t,
1370 4 : Kind::Snapshot,
1371 4 : id,
1372 4 : json!({"id": id, "type": "Snapshot", "__tenant": synth,
1373 4 : "expiresAt": "2000-01-01T00:00:00Z"}),
1374 4 : )
1375 4 : .await
1376 4 : .expect("seed");
1377 4 : let idx = TenantId::new_internal("snap-index").expect("index tenant");
1378 4 : st.store
1379 4 : .create(&idx, Kind::Snapshot, synth, json!({"id": synth}))
1380 4 : .await
1381 4 : .expect("seed index");
1382 :
1383 4 : assert!(snap_get(&st, &t, id).await.is_none(), "5.16: not served");
1384 4 : assert!(
1385 4 : st.store
1386 4 : .get(&t, Kind::Snapshot, id)
1387 4 : .await
1388 4 : .expect("store")
1389 4 : .is_some(),
1390 : "the read deleted the meta: a GET must be safe (RFC 9110 9.2.1)"
1391 : );
1392 :
1393 4 : assert_eq!(sweep_expired_snapshots(&st, &t).await, 1, "the sweep reaps");
1394 4 : assert!(
1395 4 : st.store
1396 4 : .get(&t, Kind::Snapshot, id)
1397 4 : .await
1398 4 : .expect("store")
1399 4 : .is_none(),
1400 : "the sweep left the meta behind"
1401 : );
1402 : // The teardown runs behind the sweep: the copy is dropped first and
1403 : // the entry that names it last, so a broker that stops in the middle
1404 : // leaves a pointer to data that is still there.
1405 4 : for _ in 0..100 {
1406 6 : if st
1407 6 : .store
1408 6 : .get(&idx, Kind::Snapshot, synth)
1409 6 : .await
1410 6 : .expect("store")
1411 6 : .is_none()
1412 4 : {
1413 4 : return;
1414 4 : }
1415 4 : tokio::time::sleep(std::time::Duration::from_millis(20)).await;
1416 4 : }
1417 4 : panic!("the sweep left the reverse-index entry behind");
1418 4 : }
1419 :
1420 : /// The sweep takes the expiry and nothing else: a live Snapshot and the
1421 : /// reverse-index markers that share its storage both survive.
1422 : #[tokio::test]
1423 4 : async fn the_sweep_keeps_a_live_snapshot_and_the_index_markers() {
1424 4 : let st = state().await;
1425 4 : let t = TenantId::default();
1426 4 : let live = "urn:ngsi-ld:Snapshot:live";
1427 4 : st.store
1428 4 : .create(
1429 4 : &t,
1430 4 : Kind::Snapshot,
1431 4 : live,
1432 4 : json!({"id": live, "type": "Snapshot",
1433 4 : "expiresAt": "2999-01-01T00:00:00Z"}),
1434 4 : )
1435 4 : .await
1436 4 : .expect("seed");
1437 : // a marker doc: same Kind, not a Snapshot, and long past any expiry
1438 4 : st.store
1439 4 : .create(
1440 4 : &t,
1441 4 : Kind::Snapshot,
1442 4 : "snap-default-marker",
1443 4 : json!({"id": "snap-default-marker", "expiresAt": "2000-01-01T00:00:00Z"}),
1444 4 : )
1445 4 : .await
1446 4 : .expect("seed marker");
1447 :
1448 4 : assert_eq!(sweep_expired_snapshots(&st, &t).await, 0);
1449 4 : assert!(snap_get(&st, &t, live).await.is_some(), "live one survives");
1450 4 : assert!(
1451 4 : st.store
1452 4 : .get(&t, Kind::Snapshot, "snap-default-marker")
1453 4 : .await
1454 4 : .expect("store")
1455 4 : .is_some(),
1456 4 : "the sweep took a reverse-index marker for a Snapshot"
1457 4 : );
1458 4 : }
1459 :
1460 : /// One request through the full router — the 6.3.22 middleware included.
1461 528 : async fn send(
1462 528 : st: &AppState,
1463 528 : method: &str,
1464 528 : path: &str,
1465 528 : body: Option<(&str, String)>,
1466 528 : extra: &[(&str, &str)],
1467 528 : ) -> (StatusCode, HeaderMap, Value) {
1468 528 : let mut b = Request::builder().method(method).uri(path);
1469 528 : for (k, v) in extra {
1470 24 : b = b.header(*k, *v);
1471 24 : }
1472 528 : let req = match body {
1473 464 : Some((ct, payload)) => b
1474 464 : .header("Content-Type", ct)
1475 464 : .header("Content-Length", payload.len())
1476 464 : .body(Body::from(payload)),
1477 64 : None => b.body(Body::empty()),
1478 : }
1479 528 : .expect("request");
1480 528 : let res = crate::router(st.clone())
1481 528 : .oneshot(req)
1482 528 : .await
1483 528 : .expect("response");
1484 528 : let status = res.status();
1485 528 : let headers = res.headers().clone();
1486 528 : let bytes = axum::body::to_bytes(res.into_body(), usize::MAX)
1487 528 : .await
1488 528 : .expect("body");
1489 528 : let body = if bytes.is_empty() {
1490 456 : Value::Null
1491 : } else {
1492 72 : serde_json::from_slice(&bytes).unwrap_or(Value::Null)
1493 : };
1494 528 : (status, headers, body)
1495 528 : }
1496 :
1497 440 : async fn post_json(st: &AppState, path: &str, doc: &Value) -> (StatusCode, HeaderMap, Value) {
1498 440 : send(
1499 440 : st,
1500 440 : "POST",
1501 440 : path,
1502 440 : Some(("application/json", doc.to_string())),
1503 440 : &[],
1504 440 : )
1505 440 : .await
1506 440 : }
1507 :
1508 40 : fn header<'a>(h: &'a HeaderMap, name: &str) -> Option<&'a str> {
1509 40 : h.get(name).and_then(|v| v.to_str().ok())
1510 40 : }
1511 :
1512 : /// Create a snapshot over every Vehicle and wait for the background fill.
1513 24 : async fn snapshot_over_vehicles(st: &AppState) -> String {
1514 24 : let doc = json!({"type": "Snapshot",
1515 24 : "snapshotQueries": [{"type": "Query", "entities": [{"type": "Vehicle"}]}]});
1516 24 : let (status, headers, body) = post_json(st, "/ngsi-ld/v1/snapshots", &doc).await;
1517 24 : assert_eq!(status, StatusCode::CREATED, "{body}");
1518 24 : let loc = header(&headers, "Location").expect("Location").to_owned();
1519 24 : for _ in 0..200 {
1520 24 : let (_, _, b) = send(st, "GET", &loc, None, &[]).await;
1521 24 : if b["snapshotStatus"] != "preparing" {
1522 24 : return b["id"].as_str().expect("snapshot id").to_owned();
1523 0 : }
1524 0 : tokio::time::sleep(std::time::Duration::from_millis(20)).await;
1525 : }
1526 0 : panic!("snapshot never left preparing");
1527 24 : }
1528 :
1529 16 : async fn stored(st: &AppState, id: &str) -> Value {
1530 16 : st.store
1531 16 : .get(&TenantId::default(), Kind::Snapshot, id)
1532 16 : .await
1533 16 : .ok()
1534 16 : .flatten()
1535 16 : .expect("snapshot document")
1536 16 : }
1537 :
1538 : /// 6.3.22: the Snapshot API's own resources are never snapshot-scoped —
1539 : /// but that exemption is the /snapshots RESOURCE, not any path in which
1540 : /// the word occurs. An Entity attribute named "snapshots" stays scoped
1541 : /// to the frozen copy, so the live entity is untouched.
1542 : #[tokio::test(flavor = "multi_thread")]
1543 4 : async fn clause_6_3_22_scope_guard_matches_only_the_snapshot_resource() {
1544 4 : let st = state().await;
1545 4 : let ent = json!({"id": "urn:ngsi-ld:Vehicle:g1", "type": "Vehicle",
1546 4 : "snapshots": {"type": "Property", "value": 3}});
1547 4 : let (status, _, b) = post_json(&st, "/ngsi-ld/v1/entities", &ent).await;
1548 4 : assert_eq!(status, StatusCode::CREATED, "{b}");
1549 4 : let sid = snapshot_over_vehicles(&st).await;
1550 :
1551 4 : let (status, headers, b) = send(
1552 4 : &st,
1553 4 : "DELETE",
1554 4 : "/ngsi-ld/v1/entities/urn:ngsi-ld:Vehicle:g1/attrs/snapshots",
1555 4 : None,
1556 4 : &[("NGSILD-Snapshot", sid.as_str())],
1557 : )
1558 4 : .await;
1559 4 : assert_eq!(status, StatusCode::NO_CONTENT, "{b}");
1560 4 : assert_eq!(header(&headers, "NGSILD-Snapshot"), Some(sid.as_str()));
1561 :
1562 4 : let (status, _, live) = send(
1563 4 : &st,
1564 4 : "GET",
1565 4 : "/ngsi-ld/v1/entities/urn:ngsi-ld:Vehicle:g1",
1566 4 : None,
1567 4 : &[],
1568 : )
1569 4 : .await;
1570 4 : assert_eq!(status, StatusCode::OK, "{live}");
1571 4 : assert!(
1572 4 : live.get("snapshots").is_some(),
1573 : "a snapshot-scoped request deleted the LIVE attribute: {live}"
1574 : );
1575 4 : let (_, _, frozen) = send(
1576 4 : &st,
1577 4 : "GET",
1578 4 : "/ngsi-ld/v1/entities/urn:ngsi-ld:Vehicle:g1",
1579 4 : None,
1580 4 : &[("NGSILD-Snapshot", sid.as_str())],
1581 : )
1582 4 : .await;
1583 4 : assert!(
1584 4 : frozen.get("snapshots").is_none(),
1585 4 : "the frozen copy still carries the deleted attribute: {frozen}"
1586 4 : );
1587 4 : }
1588 :
1589 : /// 6.3.22: "If the HTTP header NGSILD-Snapshot is present in the HTTP
1590 : /// request, it shall also be present in HTTP response" — on the
1591 : /// unscoped Snapshot resources and on the error exits as well.
1592 : #[tokio::test(flavor = "multi_thread")]
1593 4 : async fn clause_6_3_22_header_is_echoed_on_every_exit() {
1594 4 : let st = state().await;
1595 4 : let sid = snapshot_over_vehicles(&st).await;
1596 4 : let (status, headers, b) = send(
1597 4 : &st,
1598 4 : "GET",
1599 4 : &format!("/ngsi-ld/v1/snapshots/{sid}"),
1600 4 : None,
1601 4 : &[("NGSILD-Snapshot", sid.as_str())],
1602 : )
1603 4 : .await;
1604 4 : assert_eq!(status, StatusCode::OK, "{b}");
1605 4 : assert_eq!(header(&headers, "NGSILD-Snapshot"), Some(sid.as_str()));
1606 :
1607 4 : let unknown = "urn:ngsi-ld:Snapshot:nope";
1608 4 : let (status, headers, b) = send(
1609 4 : &st,
1610 4 : "GET",
1611 4 : "/ngsi-ld/v1/entities?type=Vehicle",
1612 4 : None,
1613 4 : &[("NGSILD-Snapshot", unknown)],
1614 : )
1615 4 : .await;
1616 4 : assert_eq!(status, StatusCode::NOT_FOUND, "{b}");
1617 4 : assert_eq!(header(&headers, "NGSILD-Snapshot"), Some(unknown));
1618 4 : assert!(
1619 4 : !b.to_string().contains("snap-"),
1620 4 : "the synthetic tenant leaked to the client: {b}"
1621 4 : );
1622 4 : }
1623 :
1624 : /// 5.16.1.4: "If the NGSI-LD endpoint already knows about this Snapshot,
1625 : /// as there is an existing Snapshot whose id (URI) is equivalent, an
1626 : /// error of type AlreadyExists shall be raised" — and the rejected
1627 : /// create must not have replaced the existing snapshot.
1628 : #[tokio::test(flavor = "multi_thread")]
1629 4 : async fn clause_5_16_1_4_duplicate_id_is_already_exists() {
1630 4 : let st = state().await;
1631 4 : let mut doc = json!({"id": "urn:ngsi-ld:Snapshot:dup", "type": "Snapshot",
1632 4 : "snapshotPriority": 3,
1633 4 : "snapshotQueries": [{"type": "Query", "entities": [{"type": "Vehicle"}]}]});
1634 4 : let (status, _, b) = post_json(&st, "/ngsi-ld/v1/snapshots", &doc).await;
1635 4 : assert_eq!(status, StatusCode::CREATED, "{b}");
1636 4 : doc["snapshotPriority"] = json!(9);
1637 4 : let (status, _, b) = post_json(&st, "/ngsi-ld/v1/snapshots", &doc).await;
1638 4 : assert_eq!(status, StatusCode::CONFLICT, "{b}");
1639 4 : assert!(
1640 4 : b["type"]
1641 4 : .as_str()
1642 4 : .unwrap_or_default()
1643 4 : .ends_with("AlreadyExists"),
1644 : "{b}"
1645 : );
1646 4 : let (_, _, got) = send(
1647 4 : &st,
1648 4 : "GET",
1649 4 : "/ngsi-ld/v1/snapshots/urn:ngsi-ld:Snapshot:dup",
1650 4 : None,
1651 4 : &[],
1652 : )
1653 4 : .await;
1654 4 : assert_eq!(
1655 4 : got["snapshotPriority"], 3,
1656 : "the rejected create overwrote the existing snapshot: {got}"
1657 : );
1658 4 : assert!(
1659 4 : got.get("__tenant").is_none(),
1660 4 : "internal member served to the client: {got}"
1661 4 : );
1662 4 : }
1663 :
1664 : /// 6.3.5 (5.16.4.4 → 5.5.4): snapshot bodies obey the @context rules —
1665 : /// application/json with a body @context is BadRequestData,
1666 : /// application/ld+json without one is BadRequestData, any other media
1667 : /// type is a bare 415, and @context is never stored in the snapshot.
1668 : #[tokio::test(flavor = "multi_thread")]
1669 4 : async fn clause_6_3_5_snapshot_bodies_follow_the_context_rules() {
1670 4 : let st = state().await;
1671 4 : let core = "https://uri.etsi.org/ngsi-ld/v1/ngsi-ld-core-context-v1.9.jsonld";
1672 4 : let plain = json!({"type": "Snapshot",
1673 4 : "snapshotQueries": [{"type": "Query", "entities": [{"type": "Vehicle"}]}]});
1674 4 : let mut with_ctx = plain.clone();
1675 4 : with_ctx["@context"] = json!(core);
1676 :
1677 8 : for (ct, doc) in [
1678 4 : ("application/json", &with_ctx),
1679 4 : ("application/ld+json", &plain),
1680 4 : ] {
1681 8 : let (status, _, b) = send(
1682 8 : &st,
1683 8 : "POST",
1684 8 : "/ngsi-ld/v1/snapshots",
1685 8 : Some((ct, doc.to_string())),
1686 8 : &[],
1687 : )
1688 8 : .await;
1689 8 : assert_eq!(status, StatusCode::BAD_REQUEST, "{ct}: {b}");
1690 : }
1691 4 : let (status, _, _) = send(
1692 4 : &st,
1693 4 : "POST",
1694 4 : "/ngsi-ld/v1/snapshots",
1695 4 : Some(("text/plain", plain.to_string())),
1696 4 : &[],
1697 : )
1698 4 : .await;
1699 4 : assert_eq!(status, StatusCode::UNSUPPORTED_MEDIA_TYPE);
1700 :
1701 4 : let (status, headers, b) = send(
1702 4 : &st,
1703 4 : "POST",
1704 4 : "/ngsi-ld/v1/snapshots",
1705 4 : Some(("application/ld+json", with_ctx.to_string())),
1706 4 : &[],
1707 : )
1708 4 : .await;
1709 4 : assert_eq!(status, StatusCode::CREATED, "{b}");
1710 4 : let loc = header(&headers, "Location").expect("Location").to_owned();
1711 4 : let (_, _, got) = send(&st, "GET", &loc, None, &[]).await;
1712 4 : assert!(
1713 4 : got.get("@context").is_none(),
1714 : "@context stored as a snapshot member: {got}"
1715 : );
1716 4 : let patch = json!({"snapshotPriority": 7, "@context": core});
1717 4 : let (status, _, b) = send(
1718 4 : &st,
1719 4 : "PATCH",
1720 4 : &loc,
1721 4 : Some(("application/json", patch.to_string())),
1722 4 : &[],
1723 : )
1724 4 : .await;
1725 4 : assert_eq!(status, StatusCode::BAD_REQUEST, "{b}");
1726 4 : }
1727 :
1728 : /// 5.5.6: "When a query operation is producing so many results that can
1729 : /// potentially exhaust client or server resources … implementations
1730 : /// shall raise an error of type TooManyResults." A fill copies its
1731 : /// results, so the ceiling is enforced before anything is copied.
1732 : #[tokio::test(flavor = "multi_thread")]
1733 4 : async fn clause_5_5_6_fill_stops_at_the_result_ceiling() {
1734 4 : let mut st = state().await;
1735 4 : st.max_limit = 1;
1736 4 : st.default_limit = 1;
1737 404 : for i in 0..=fill_cap(&st) {
1738 404 : let ent = json!({"id": format!("urn:ngsi-ld:Vehicle:c{i}"), "type": "Vehicle"});
1739 404 : let (status, _, b) = post_json(&st, "/ngsi-ld/v1/entities", &ent).await;
1740 404 : assert_eq!(status, StatusCode::CREATED, "{b}");
1741 : }
1742 4 : let sid = snapshot_over_vehicles(&st).await;
1743 4 : let ready = stored(&st, &sid).await;
1744 4 : assert_eq!(ready["snapshotStatus"], "failure", "{ready}");
1745 4 : let detail = &ready["snapshotQueriesDetails"][0];
1746 4 : assert_eq!(detail["resultStatus"], "failure", "{ready}");
1747 4 : assert!(
1748 4 : detail["problemDetails"]["type"]
1749 4 : .as_str()
1750 4 : .unwrap_or_default()
1751 4 : .ends_with("TooManyResults"),
1752 : "{detail}"
1753 : );
1754 4 : let (status, _, list) = send(
1755 4 : &st,
1756 4 : "GET",
1757 4 : "/ngsi-ld/v1/entities?type=Vehicle&limit=1",
1758 4 : None,
1759 4 : &[("NGSILD-Snapshot", sid.as_str())],
1760 : )
1761 4 : .await;
1762 4 : assert_eq!(status, StatusCode::OK, "{list}");
1763 4 : assert_eq!(
1764 4 : list,
1765 4 : json!([]),
1766 4 : "the oversized result set was copied into the snapshot anyway"
1767 4 : );
1768 4 : }
1769 :
1770 : /// 5.16.1.4 + 5.5.15: the filled copy is reachable only through its
1771 : /// snapshot document, so a fill whose snapshot has been deleted (or
1772 : /// evicted) stops and frees what it already copied.
1773 : #[tokio::test(flavor = "multi_thread")]
1774 4 : async fn clause_5_16_1_4_fill_stops_and_reaps_when_its_snapshot_is_gone() {
1775 4 : let st = state().await;
1776 4 : let tenant = TenantId::default();
1777 4 : let orphan = TenantId::new_internal("snap-reap-test").expect("tenant");
1778 4 : let _ = st
1779 4 : .store
1780 4 : .create(
1781 4 : &orphan,
1782 4 : Kind::Entity,
1783 4 : "urn:ngsi-ld:Vehicle:r1",
1784 4 : json!({"id": "urn:ngsi-ld:Vehicle:r1"}),
1785 4 : )
1786 4 : .await;
1787 4 : assert!(
1788 4 : fill_cancelled(&st, &tenant, "urn:ngsi-ld:Snapshot:gone", &orphan).await,
1789 : "a fill kept running after its snapshot was deleted"
1790 : );
1791 4 : assert!(
1792 4 : st.store
1793 4 : .list(&orphan, Kind::Entity)
1794 4 : .await
1795 4 : .unwrap_or_default()
1796 4 : .is_empty(),
1797 : "the orphaned copy outlived its snapshot"
1798 : );
1799 :
1800 4 : let sid = snapshot_over_vehicles(&st).await;
1801 4 : let synth = synth_tenant(&stored(&st, &sid).await).expect("__tenant");
1802 4 : let _ = st
1803 4 : .store
1804 4 : .create(
1805 4 : &synth,
1806 4 : Kind::Entity,
1807 4 : "urn:ngsi-ld:Vehicle:r2",
1808 4 : json!({"id": "urn:ngsi-ld:Vehicle:r2"}),
1809 4 : )
1810 4 : .await;
1811 4 : assert!(
1812 4 : !fill_cancelled(&st, &tenant, &sid, &synth).await,
1813 : "a live snapshot must not cancel its own fill"
1814 : );
1815 4 : assert_eq!(
1816 4 : st.store
1817 4 : .list(&synth, Kind::Entity)
1818 4 : .await
1819 4 : .unwrap_or_default()
1820 4 : .len(),
1821 4 : 1,
1822 4 : "a live snapshot's copy was reaped"
1823 4 : );
1824 4 : }
1825 :
1826 : /// 5.16.5.4 + 5.5.15: deleting a snapshot frees the whole isolated copy,
1827 : /// including the subscriptions created through the 6.3.22 header — no
1828 : /// client can reach them afterwards, so nothing may keep firing.
1829 : #[tokio::test(flavor = "multi_thread")]
1830 4 : async fn clause_5_16_5_delete_frees_the_whole_snapshot_copy() {
1831 4 : let st = state().await;
1832 4 : let sid = snapshot_over_vehicles(&st).await;
1833 4 : let synth = synth_tenant(&stored(&st, &sid).await).expect("__tenant");
1834 4 : let sub = json!({"type": "Subscription", "entities": [{"type": "Vehicle"}],
1835 4 : "notification": {"endpoint": {"uri": "http://127.0.0.1:9/none"}}});
1836 4 : let (status, _, b) = send(
1837 4 : &st,
1838 4 : "POST",
1839 4 : "/ngsi-ld/v1/subscriptions",
1840 4 : Some(("application/json", sub.to_string())),
1841 4 : &[("NGSILD-Snapshot", sid.as_str())],
1842 : )
1843 4 : .await;
1844 4 : assert_eq!(status, StatusCode::CREATED, "{b}");
1845 4 : assert!(
1846 4 : !st.store
1847 4 : .list(&synth, Kind::Subscription)
1848 4 : .await
1849 4 : .unwrap_or_default()
1850 4 : .is_empty(),
1851 : "the scoped subscription did not land in the snapshot copy"
1852 : );
1853 :
1854 4 : let (status, _, _) = send(
1855 4 : &st,
1856 4 : "DELETE",
1857 4 : &format!("/ngsi-ld/v1/snapshots/{sid}"),
1858 4 : None,
1859 4 : &[],
1860 : )
1861 4 : .await;
1862 4 : assert_eq!(status, StatusCode::NO_CONTENT);
1863 4 : for _ in 0..100 {
1864 5 : let left = st
1865 5 : .store
1866 5 : .list(&synth, Kind::Subscription)
1867 5 : .await
1868 5 : .unwrap_or_default()
1869 5 : .len();
1870 5 : if left == 0 {
1871 4 : return;
1872 4 : }
1873 4 : tokio::time::sleep(std::time::Duration::from_millis(20)).await;
1874 4 : }
1875 4 : panic!("a snapshot-scoped subscription outlived its snapshot");
1876 4 : }
1877 :
1878 : /// 5.2.41 Table 5.2.41-1: the snapshot id "cannot be later modified in
1879 : /// update operations" — whatever JSON type the client sends it as.
1880 : #[test]
1881 4 : fn clause_5_16_4_4_id_cannot_be_modified() {
1882 4 : assert!(validate(&json!({"id": "urn:ngsi-ld:Snapshot:1"}), Mode::Update).is_err());
1883 4 : assert!(
1884 4 : validate(&json!({"id": 5}), Mode::Update).is_err(),
1885 : "a non-string id escaped the 5.2.41 restriction"
1886 : );
1887 4 : assert!(validate(&json!({"snapshotPriority": 7}), Mode::Update).is_ok());
1888 4 : }
1889 :
1890 : /// 5.16.1.4: expiresAt is set "taking into account the snapshotLifetime
1891 : /// requested, but applying the configured limit of the NGSI-LD system";
1892 : /// a malformed duration is BadRequestData, and a snapshot past its
1893 : /// expiresAt is gone together with its copy.
1894 : #[tokio::test(flavor = "multi_thread")]
1895 4 : async fn clause_5_16_1_4_lifetime_is_clamped_and_expiry_removes_the_snapshot() {
1896 4 : let mut o = Map::new();
1897 4 : o.insert("snapshotLifetime".into(), json!("P10Y"));
1898 4 : let exp = chrono::DateTime::parse_from_rfc3339(&expires_at(&o).expect("expiresAt"))
1899 4 : .expect("rfc3339")
1900 4 : .with_timezone(&chrono::Utc);
1901 4 : let limit = chrono::Utc::now() + chrono::Duration::seconds(MAX_LIFETIME_SECS);
1902 4 : assert!(exp <= limit, "the configured limit was not applied: {exp}");
1903 4 : assert!(
1904 4 : exp > chrono::Utc::now() + chrono::Duration::days(6),
1905 : "the suggested lifetime was ignored: {exp}"
1906 : );
1907 4 : assert!(validate(
1908 4 : &json!({"type": "Snapshot", "snapshotLifetime": "tomorrow",
1909 4 : "snapshotQueries": [{"type": "Query", "entities": [{"type": "Vehicle"}]}]}),
1910 4 : Mode::Create
1911 4 : )
1912 4 : .is_err());
1913 :
1914 4 : let st = state().await;
1915 4 : let sid = snapshot_over_vehicles(&st).await;
1916 4 : let synth = synth_tenant(&stored(&st, &sid).await).expect("__tenant");
1917 4 : let _ = st
1918 4 : .store
1919 4 : .create(
1920 4 : &synth,
1921 4 : Kind::Entity,
1922 4 : "urn:ngsi-ld:Vehicle:e1",
1923 4 : json!({"id": "urn:ngsi-ld:Vehicle:e1"}),
1924 4 : )
1925 4 : .await;
1926 4 : let _ = st
1927 4 : .store
1928 4 : .mutate(&TenantId::default(), Kind::Snapshot, &sid, |d| {
1929 4 : if let Some(o) = d.as_object_mut() {
1930 4 : o.insert("expiresAt".into(), json!("2000-01-01T00:00:00.000Z"));
1931 4 : }
1932 4 : Ok::<_, std::convert::Infallible>(())
1933 4 : })
1934 4 : .await;
1935 4 : let (status, _, b) = send(
1936 4 : &st,
1937 4 : "GET",
1938 4 : &format!("/ngsi-ld/v1/snapshots/{sid}"),
1939 4 : None,
1940 4 : &[],
1941 : )
1942 4 : .await;
1943 4 : assert_eq!(status, StatusCode::NOT_FOUND, "an expired snapshot: {b}");
1944 4 : assert!(
1945 4 : !st.store
1946 4 : .list(&synth, Kind::Entity)
1947 4 : .await
1948 4 : .unwrap_or_default()
1949 4 : .is_empty(),
1950 : "the GET freed the copy: reading must not write"
1951 : );
1952 : // The sweep is what frees it, and the purge it starts is background.
1953 4 : assert_eq!(
1954 4 : sweep_expired_snapshots(&st, &TenantId::default()).await,
1955 : 1,
1956 : "the sweep did not reap the expired snapshot"
1957 : );
1958 4 : for _ in 0..100 {
1959 6 : if st
1960 6 : .store
1961 6 : .list(&synth, Kind::Entity)
1962 6 : .await
1963 6 : .unwrap_or_default()
1964 6 : .is_empty()
1965 4 : {
1966 4 : return;
1967 4 : }
1968 4 : tokio::time::sleep(std::time::Duration::from_millis(20)).await;
1969 4 : }
1970 4 : panic!("the copy of an expired snapshot was not freed");
1971 4 : }
1972 : }
|
