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 122 : fn bad(m: String) -> NgsiError {
42 122 : NgsiError::BadRequestData(m)
43 122 : }
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 196 : fn expires_at(meta: &Map<String, Value>) -> Result<String, NgsiError> {
58 196 : let secs = match meta.get("snapshotLifetime").and_then(Value::as_str) {
59 14 : Some(d) => crate::entity_map::iso8601_secs(d)
60 14 : .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 14 : .clamp(1, MAX_LIFETIME_SECS),
69 182 : None => DEFAULT_LIFETIME_SECS,
70 : };
71 : // `secs` is clamped to MAX_LIFETIME_SECS above, far inside chrono's range.
72 196 : let now = chrono::Utc::now();
73 196 : Ok(now
74 196 : .checked_add_signed(chrono::Duration::seconds(secs))
75 196 : .unwrap_or(now)
76 196 : .to_rfc3339_opts(chrono::SecondsFormat::Millis, true))
77 196 : }
78 :
79 2192 : fn expired(meta: &Value) -> bool {
80 2192 : meta.get("expiresAt")
81 2192 : .and_then(Value::as_str)
82 2192 : .and_then(|e| chrono::DateTime::parse_from_rfc3339(e).ok())
83 2192 : .is_some_and(|e| e < chrono::Utc::now())
84 2192 : }
85 :
86 : /// The internal tenant holding the synth-tenant -> (owner, snapshot id)
87 : /// index docs (durable reverse lookup for 6.3.22 notification stamping).
88 230 : fn snap_index_tenant() -> Option<TenantId> {
89 230 : TenantId::new_internal("snap-index").ok()
90 230 : }
91 :
92 : /// 5.2.41: a Snapshot's type is fixed to "Snapshot". The reverse-index
93 : /// marker docs share Kind::Snapshot storage but are not Snapshots, so every
94 : /// place that reads a document as one filters on this — a listing that does
95 : /// not would count and select the markers as snapshots.
96 2436 : fn is_snapshot(meta: &Value) -> bool {
97 2436 : meta.get("type").and_then(Value::as_str) == Some("Snapshot")
98 2436 : }
99 :
100 : /// Registry access with lazy expiry: an expired snapshot is gone to every
101 : /// caller. Reading is a read — the meta row, its reverse-index entry and
102 : /// the synthetic tenant's data are freed by `sweep_expired_snapshots` on
103 : /// the sweep tick, not by the request that found the expiry. Snapshot docs
104 : /// live in the store (Kind::Snapshot) so restarts keep them on persistent
105 : /// store modes.
106 1262 : pub(crate) async fn snap_get(st: &AppState, tenant: &TenantId, id: &str) -> Option<Value> {
107 1262 : st.store
108 1262 : .get(tenant, Kind::Snapshot, id)
109 1262 : .await
110 1262 : .ok()
111 1262 : .flatten()
112 1262 : .filter(|meta| is_snapshot(meta) && !expired(meta))
113 1262 : }
114 :
115 : /// 5.16.1.4: "If the NGSI-LD endpoint already knows about this Snapshot …
116 : /// an error of type AlreadyExists shall be raised." The store's create is
117 : /// the atomic check-and-insert, so two concurrent creates of the same id
118 : /// cannot both succeed and overwrite each other's synthetic tenant.
119 184 : async fn snap_insert(
120 184 : st: &AppState,
121 184 : tenant: &TenantId,
122 184 : id: &str,
123 184 : meta: &Value,
124 184 : ) -> Result<(), NgsiError> {
125 184 : if !st
126 184 : .store
127 184 : .create(tenant, Kind::Snapshot, id, meta.clone())
128 184 : .await?
129 : {
130 4 : return Err(NgsiError::AlreadyExists(format!(
131 4 : "snapshot {id} already exists"
132 4 : )));
133 180 : }
134 : // Durable reverse index for snapshot_of_synth. The marker carries the
135 : // synthetic tenant as its `id` as well as its key: it is the only record
136 : // that names that tenant once the Snapshot document is gone, and a
137 : // document a listing cannot identify is one nothing can walk.
138 180 : if let (Some(synth), Some(idx)) = (
139 180 : meta.get("__tenant").and_then(Value::as_str),
140 180 : snap_index_tenant(),
141 : ) {
142 180 : let _ = st
143 180 : .store
144 180 : .create(
145 180 : &idx,
146 180 : Kind::Snapshot,
147 180 : synth,
148 180 : json!({"id": synth, "tenant": tenant.as_str(), "snapshot": id}),
149 180 : )
150 180 : .await;
151 0 : }
152 180 : Ok(())
153 184 : }
154 :
155 : /// Members `finish` owns (5.2.41 Table 5.2.41-2, output only).
156 : const FINISH_MEMBERS: &[&str] = &[
157 : "snapshotStatus",
158 : "modifiedAt",
159 : "snapshotQueriesDetails",
160 : "snapshotTemporalQueriesDetails",
161 : ];
162 :
163 : /// Members an Update Snapshot (5.16.4.4) writes.
164 : const UPDATE_MEMBERS: &[&str] = &[
165 : "snapshotLifetime",
166 : "snapshotPriority",
167 : "endpoint",
168 : "receiverInfo",
169 : "expiresAt",
170 : "modifiedAt",
171 : ];
172 :
173 : /// Write `members` of `meta` back onto a snapshot that already exists (a
174 : /// member `meta` lacks is removed); a snapshot deleted meanwhile is not
175 : /// resurrected. Only the named members are written, so a fill finishing
176 : /// beside an Update Snapshot keeps the other's members: each writer read
177 : /// the document before the other wrote it. Returns the stored document.
178 193 : async fn snap_put(
179 193 : st: &AppState,
180 193 : tenant: &TenantId,
181 193 : meta: &Value,
182 193 : members: &[&str],
183 193 : ) -> Option<Value> {
184 193 : let id = meta.get("id").and_then(Value::as_str)?;
185 193 : st.store
186 193 : .mutate(tenant, Kind::Snapshot, id, |d| {
187 193 : if let Some(o) = d.as_object_mut() {
188 808 : for k in members {
189 808 : match meta.get(*k) {
190 561 : Some(v) => o.insert((*k).to_owned(), v.clone()),
191 247 : None => o.remove(*k),
192 : };
193 : }
194 0 : }
195 193 : Ok::<_, std::convert::Infallible>(d.clone())
196 193 : })
197 193 : .await
198 193 : .ok()
199 193 : .flatten()
200 193 : .and_then(Result::ok)
201 193 : }
202 :
203 : /// Remove a snapshot everywhere: doc, synth-tenant index, data purge.
204 46 : pub(crate) async fn snap_remove(st: &AppState, tenant: &TenantId, id: &str, meta: &Value) {
205 46 : let _ = st.store.delete(tenant, Kind::Snapshot, id).await;
206 46 : purge_data_bg(st, meta);
207 46 : }
208 :
209 : /// 4.22 for snapshots: `snap_get` refuses an expired one, and this is what
210 : /// removes it — meta, reverse-index entry and the synthetic tenant's data,
211 : /// through the same `snap_remove` a read used to call, so an expiry
212 : /// collected on the tick frees exactly what an expiry collected on a read
213 : /// freed.
214 145070 : pub(crate) async fn sweep_expired_snapshots(st: &AppState, tenant: &TenantId) -> usize {
215 145070 : let mut dead: Vec<(String, Value)> = Vec::new();
216 145070 : if crate::csource::walk_docs(st, tenant, Kind::Snapshot, |doc| {
217 1012 : if is_snapshot(&doc) && expired(&doc) {
218 8 : if let Some(id) = doc.get("id").and_then(Value::as_str) {
219 8 : dead.push((id.to_owned(), doc));
220 8 : }
221 1004 : }
222 1012 : Ok(())
223 1012 : })
224 145070 : .await
225 145070 : .is_err()
226 : {
227 : // nothing is freed this round, and that is not silent: the tenant
228 : // only, no error text (a value that would not open is a credential)
229 12 : tracing::error!(
230 0 : tenant = tenant.as_str(),
231 : "the expired-snapshot sweep could not list this tenant's snapshots"
232 : );
233 12 : return 0;
234 145058 : }
235 145058 : let n = dead.len();
236 145058 : for (id, meta) in dead {
237 8 : snap_remove(st, tenant, &id, &meta).await;
238 : }
239 145058 : n
240 145070 : }
241 :
242 365 : fn synth_tenant(meta: &Value) -> Option<TenantId> {
243 365 : meta.get("__tenant")
244 365 : .and_then(Value::as_str)
245 365 : .and_then(|t| TenantId::new_internal(t).ok())
246 365 : }
247 :
248 : /// Free the snapshot's isolated copy. 5.5.15 permits every Core and
249 : /// Temporal operation on a snapshot, so a snapshot-scoped request can put
250 : /// documents of ANY kind under the synthetic tenant — once the snapshot is
251 : /// gone none of them is reachable (no client may name the internal tenant),
252 : /// so every kind is dropped, not only the filled data.
253 : ///
254 : /// The tenant purge, not a per-document sweep: it covers every kind rather
255 : /// than a list that has to be kept in step with `Kind`, and it removes the
256 : /// tenant itself. A sweep leaves the empty tenant behind, and `/q/tenants`
257 : /// reports what the store holds a tenant entry for — so every snapshot ever
258 : /// deleted would leave a `snap-<uuid>` name in the inventory for the life of
259 : /// the deployment.
260 52 : async fn purge_synth(st: &AppState, synth: &TenantId) {
261 : // history lives behind its own driver seam
262 52 : let _ = st.temporal.purge_tenant(synth).await;
263 52 : let _ = st.store.purge_tenant(synth).await;
264 52 : }
265 :
266 : /// The teardown that outlives the request: the copy is dropped, and only
267 : /// then the reverse-index entry that named the synthetic tenant. That order
268 : /// is the recovery path — the Snapshot document is already gone, so the index
269 : /// entry is the last thing that says which tenant holds the copy, and a
270 : /// broker that stops mid-teardown leaves it pointing at data still there
271 : /// rather than data nothing names.
272 46 : fn purge_data_bg(st: &AppState, meta: &Value) {
273 46 : let Some(synth) = synth_tenant(meta) else {
274 0 : return;
275 : };
276 46 : let st = st.clone();
277 46 : crate::spawn(async move {
278 46 : purge_synth(&st, &synth).await;
279 46 : if let Some(idx) = snap_index_tenant() {
280 46 : let _ = st.store.delete(&idx, Kind::Snapshot, synth.as_str()).await;
281 0 : }
282 46 : });
283 46 : }
284 :
285 : /// The snapshot document as presented to clients (internal members hidden).
286 : /// `receiverInfo` and userinfo in `endpoint` are the credentials the 5.16.6
287 : /// notification is sent with, so a read answers them masked, exactly as a
288 : /// Subscription's (`subscriptions::mask_receiver_info`).
289 276 : fn present(meta: &Value) -> Value {
290 276 : let mut out = meta.clone();
291 276 : crate::policy::strip_internal(&mut out);
292 276 : if let Some(o) = out.as_object_mut() {
293 276 : let mut n = as_notification(o);
294 276 : crate::subscriptions::mask_receiver_info(&mut n);
295 276 : from_notification(o, n);
296 276 : }
297 276 : out
298 276 : }
299 :
300 : /// A Snapshot's `endpoint` and `receiverInfo` (Table 5.2.41-1) in the
301 : /// `notification.endpoint` shape a Subscription keeps them in (5.2.15), so
302 : /// both go through the one masking rule.
303 296 : fn as_notification(o: &Map<String, Value>) -> Map<String, Value> {
304 296 : let mut ep = Map::new();
305 296 : if let Some(u) = o.get("endpoint") {
306 46 : ep.insert("uri".into(), u.clone());
307 250 : }
308 296 : if let Some(r) = o.get("receiverInfo") {
309 38 : ep.insert("receiverInfo".into(), r.clone());
310 258 : }
311 296 : Map::from_iter([("endpoint".to_owned(), Value::Object(ep))])
312 296 : }
313 :
314 : /// [`as_notification`] back: the members it carried, as they now stand.
315 284 : fn from_notification(o: &mut Map<String, Value>, n: Map<String, Value>) {
316 284 : let Some(Value::Object(ep)) = n.get("endpoint") else {
317 0 : return;
318 : };
319 568 : for (from, to) in [("uri", "endpoint"), ("receiverInfo", "receiverInfo")] {
320 568 : if let (Some(v), true) = (ep.get(from), o.contains_key(to)) {
321 60 : o.insert(to.into(), v.clone());
322 508 : }
323 : }
324 284 : }
325 :
326 : /// 5.16.4 update of the notification members. A client that read the
327 : /// Snapshot and writes it back sends the masked values it was answered:
328 : /// those keep the stored credential, and only beside the stored endpoint
329 : /// (`subscriptions::keep_masked_receiver_info`). The credential stays with
330 : /// its endpoint the other way too: an update that moves the endpoint and
331 : /// sends no `receiverInfo` drops the stored one, or the next notification
332 : /// would carry it to whatever host the update named.
333 20 : fn keep_receiver_info(
334 20 : frag: &mut Map<String, Value>,
335 20 : stored: &Map<String, Value>,
336 20 : ) -> Result<(), NgsiError> {
337 20 : if !frag.contains_key("endpoint") && !frag.contains_key("receiverInfo") {
338 10 : return Ok(());
339 10 : }
340 10 : let mut f = Map::from_iter([(
341 10 : "notification".to_owned(),
342 10 : Value::Object(as_notification(frag)),
343 10 : )]);
344 10 : let held = Map::from_iter([(
345 10 : "notification".to_owned(),
346 10 : Value::Object(as_notification(stored)),
347 10 : )]);
348 10 : crate::subscriptions::keep_masked_receiver_info(&mut f, &held)?;
349 8 : if let Some(Value::Object(n)) = f.remove("notification") {
350 8 : from_notification(frag, n);
351 8 : }
352 8 : let moved = frag
353 8 : .get("endpoint")
354 8 : .is_some_and(|e| Some(e) != stored.get("endpoint"));
355 8 : if moved && !frag.contains_key("receiverInfo") && stored.contains_key("receiverInfo") {
356 2 : frag.insert("receiverInfo".into(), Value::Null);
357 6 : }
358 8 : Ok(())
359 20 : }
360 :
361 : enum Mode {
362 : Create,
363 : Clone,
364 : Update,
365 : }
366 :
367 : /// 5.2.41 Table 5.2.41-1/-2 validation. Output-only members are IGNORED
368 : /// (stripped); read-only members in the wrong mode are BadRequestData.
369 282 : fn validate(body: &Value, mode: Mode) -> Result<Map<String, Value>, NgsiError> {
370 282 : let mut o = body
371 282 : .as_object()
372 282 : .cloned()
373 282 : .ok_or_else(|| bad("snapshot must be a JSON object".into()))?;
374 282 : o.remove("@context");
375 : // the broker's own members (the synthetic tenant, the creating subject)
376 : // are not the client's to send — see policy::strip_internal
377 614 : o.retain(|k, _| !k.starts_with("__"));
378 : // output-only members (Table 5.2.41-2) "shall be ignored"
379 1974 : for k in [
380 282 : "snapshotStatus",
381 282 : "snapshotQueriesDetails",
382 282 : "snapshotTemporalQueriesDetails",
383 282 : "createdAt",
384 282 : "modifiedAt",
385 282 : "expiresAt",
386 282 : "lastUsedAt",
387 1974 : ] {
388 1974 : o.remove(k);
389 1974 : }
390 282 : match mode {
391 : Mode::Create => {
392 204 : if o.get("type").and_then(Value::as_str) != Some("Snapshot") {
393 2 : return Err(bad("type must be \"Snapshot\" (5.2.41)".into()));
394 202 : }
395 202 : if !o.contains_key("snapshotQueries") && !o.contains_key("snapshotTemporalQueries") {
396 12 : return Err(bad(
397 12 : "at least one of snapshotQueries or snapshotTemporalQueries \
398 12 : shall be present (5.2.41)"
399 12 : .into(),
400 12 : ));
401 190 : }
402 : }
403 : Mode::Clone | Mode::Update => {
404 : // "both shall be omitted when updating the Snapshot status or
405 : // cloning the Snapshot" — read-only after creation
406 78 : if o.contains_key("snapshotQueries") || o.contains_key("snapshotTemporalQueries") {
407 8 : return Err(bad(
408 8 : "snapshotQueries/snapshotTemporalQueries are read-only after \
409 8 : creation (5.2.41)"
410 8 : .into(),
411 8 : ));
412 70 : }
413 : // Table 5.2.41-1: the id "cannot be later modified in update
414 : // operations" — of whatever JSON type it is sent as
415 70 : if matches!(mode, Mode::Update) && o.contains_key("id") {
416 8 : return Err(bad("the snapshot id cannot be modified (5.2.41)".into()));
417 62 : }
418 : // 5.16.4.4 modifies the Snapshot "as mandated by clause 5.5.8":
419 : // a member whose value is an NGSI-LD Null, in either form, is
420 : // removed. It reaches the merge as JSON null, which the type
421 : // checks below pass by.
422 62 : if matches!(mode, Mode::Update) {
423 228 : for k in UPDATE_MEMBERS {
424 228 : if o.get(*k)
425 228 : .is_some_and(|v| v.is_null() || v.as_str() == Some("urn:ngsi-ld:null"))
426 16 : {
427 16 : o.insert((*k).to_owned(), Value::Null);
428 212 : }
429 : }
430 24 : }
431 : }
432 : }
433 : // an update's removal (above); anywhere else a null is a wrong type
434 252 : let removed = |v: &Value| matches!(mode, Mode::Update) && v.is_null();
435 498 : for (key, temporal) in [
436 252 : ("snapshotQueries", false),
437 252 : ("snapshotTemporalQueries", true),
438 252 : ] {
439 498 : if let Some(qs) = o.get(key) {
440 194 : let arr = qs
441 194 : .as_array()
442 194 : .filter(|a| !a.is_empty())
443 194 : .ok_or_else(|| bad(format!("{key} must be a non-empty array of Query (5.2.41)")))?;
444 194 : for q in arr {
445 194 : let qo = q
446 194 : .as_object()
447 194 : .ok_or_else(|| bad(format!("{key} entries must be Query objects (5.2.23)")))?;
448 194 : if qo.get("type").and_then(Value::as_str) != Some("Query") {
449 0 : return Err(bad(format!("{key} entries must have type Query (5.2.23)")));
450 194 : }
451 194 : if temporal != qo.contains_key("temporalQ") {
452 6 : return Err(bad(format!(
453 : "{key} entries must {} a temporalQ element (5.2.41)",
454 6 : if temporal { "carry" } else { "not carry" }
455 : )));
456 188 : }
457 : }
458 304 : }
459 : }
460 246 : if let Some(p) = o.get("snapshotPriority").filter(|v| !removed(v)) {
461 56 : let ok = p.as_i64().is_some_and(|n| (1..=10).contains(&n));
462 56 : if !ok {
463 2 : return Err(bad(
464 2 : "snapshotPriority must be an integer between 1 and 10 (5.2.41)".into(),
465 2 : ));
466 54 : }
467 190 : }
468 244 : if let Some(l) = o.get("snapshotLifetime").filter(|v| !removed(v)) {
469 18 : let ok = l
470 18 : .as_str()
471 18 : .and_then(crate::entity_map::iso8601_secs)
472 18 : .is_some();
473 18 : if !ok {
474 4 : return Err(bad(
475 4 : "snapshotLifetime must be an ISO 8601 duration (5.2.41)".into(),
476 4 : ));
477 14 : }
478 226 : }
479 : // Table 5.2.41-1: endpoint is a "Dereferenceable URI", so a string that is
480 : // no URI at all names nothing the 5.16.6 notification could be sent to.
481 240 : if let Some(e) = o.get("endpoint").filter(|v| !removed(v)) {
482 42 : let uri = e
483 42 : .as_str()
484 42 : .ok_or_else(|| bad("endpoint must be a URI string (5.2.41)".into()))?;
485 38 : antares_model::EntityId::new(uri).map_err(|_| {
486 2 : bad(format!(
487 : "endpoint is not a valid URI: {:?} (5.2.41)",
488 2 : antares_notifier::redact_userinfo(uri)
489 : ))
490 2 : })?;
491 : // 5.16.6 sends the SnapshotNotification by HTTP POST: an endpoint
492 : // that binding cannot send to (one written `http:user:pass@host`,
493 : // without the `//`, included) names nothing the notification could
494 : // reach, and its userinfo would escape the masking and the sealing,
495 : // which read the credential after a `//`
496 36 : antares_notifier::http::check_http_endpoint(uri)?;
497 198 : }
498 : // Table 5.2.41-1: receiverInfo is a KeyValuePair[] (5.2.22 — both members
499 : // Strings), and 5.16.6 renders each pair as one header on the
500 : // SnapshotNotification POST. 6.3.8 binds those to RFC 7230 for the same
501 : // reason it binds a Subscription's: a pair that cannot be a header leaves
502 : // a Snapshot whose notification can only ever fail, silently.
503 226 : if let Some(ri) = o.get("receiverInfo").filter(|v| !removed(v)) {
504 24 : let pairs = ri
505 24 : .as_array()
506 24 : .ok_or_else(|| bad("receiverInfo must be a KeyValuePair array (5.2.41)".into()))?;
507 22 : for kv in pairs {
508 22 : let (k, v) = (kv["key"].as_str(), kv["value"].as_str());
509 22 : if !k.is_some_and(crate::negotiate::is_field_name)
510 20 : || !v.is_some_and(crate::negotiate::is_field_value)
511 : {
512 6 : return Err(bad(format!(
513 6 : "receiverInfo entry {kv} is not a valid HTTP header (RFC 7230, 6.3.8)"
514 6 : )));
515 16 : }
516 : }
517 202 : }
518 218 : Ok(o)
519 282 : }
520 :
521 : /// Fresh metadata for a new snapshot (create or clone) — 5.16.1.4/5.16.2.4:
522 : /// timestamps now, status "preparing", priority default 5, bounded expiresAt.
523 184 : fn new_meta(
524 184 : mut o: Map<String, Value>,
525 184 : tenant: &TenantId,
526 184 : headers: &HeaderMap,
527 184 : ) -> Result<(String, Value), NgsiError> {
528 184 : let id = match o.get("id").and_then(Value::as_str) {
529 16 : Some(id) => {
530 16 : antares_model::EntityId::new(id)
531 16 : .map_err(|_| bad(format!("snapshot id is not a valid URI: {id:?}")))?;
532 16 : id.to_owned()
533 : }
534 : None => {
535 168 : let id = format!("urn:ngsi-ld:Snapshot:{}", uuid::Uuid::new_v4());
536 168 : o.insert("id".into(), Value::String(id.clone()));
537 168 : id
538 : }
539 : };
540 184 : let ts = now_iso();
541 184 : o.insert("type".into(), Value::String("Snapshot".into()));
542 184 : o.insert("createdAt".into(), Value::String(ts.clone()));
543 184 : o.insert("modifiedAt".into(), Value::String(ts));
544 184 : o.insert("snapshotStatus".into(), Value::String("preparing".into()));
545 : // 5.2.41 Table 5.2.41-2: lastUsedAt "is initialized at creation time"
546 184 : o.insert("lastUsedAt".into(), Value::String(now_iso()));
547 184 : o.insert("expiresAt".into(), Value::String(expires_at(&o)?));
548 184 : o.entry("snapshotPriority".to_owned())
549 184 : .or_insert(Value::Number(5.into()));
550 184 : o.insert(
551 184 : "__tenant".into(),
552 184 : Value::String(format!("snap-{}", uuid::Uuid::new_v4().simple())),
553 : );
554 : // Whose snapshot this is. 5.16.1's fill runs after the request has been
555 : // answered, so the subject it copies under has to be recorded here or it
556 : // is gone; and 5.16.1/5.16.2/5.16.7 act on everything the tenant holds,
557 : // which is why a narrowing answer to any of them is a refusal
558 : // (policy::WHOLE_TENANT) rather than a smaller snapshot.
559 6 : if let Some(subject) =
560 184 : crate::policy::subject_member(&crate::policy::subject_of(tenant, headers))
561 6 : {
562 6 : o.insert(crate::policy::SUBJECT_MEMBER.into(), subject);
563 178 : }
564 184 : Ok((id, Value::Object(o)))
565 184 : }
566 :
567 : /// 5.2.41: lastUsedAt tracks "the point in time when the snapshot was most
568 : /// recently used" — refreshed on every snapshot-scoped operation.
569 78 : async fn snap_touch(st: &AppState, tenant: &TenantId, id: &str) {
570 78 : let _ = st
571 78 : .store
572 78 : .mutate(tenant, Kind::Snapshot, id, |meta| {
573 78 : if let Some(o) = meta.as_object_mut() {
574 78 : o.insert("lastUsedAt".into(), Value::String(now_iso()));
575 78 : }
576 78 : Ok::<_, std::convert::Infallible>(())
577 78 : })
578 78 : .await;
579 78 : }
580 :
581 : /// Reverse lookup: which (owner tenant, snapshot id) does a synthetic
582 : /// "snap-…" tenant belong to? Used to stamp NGSILD-Snapshot on
583 : /// notifications from snapshot-scoped subscriptions (6.3.22) without
584 : /// leaking the internal tenant.
585 8522 : pub(crate) async fn snapshot_of_synth(st: &AppState, synth: &str) -> Option<(TenantId, String)> {
586 : // only synthetic tenants are indexed; every other tenant skips the lookup
587 8522 : if !synth.starts_with("snap-") {
588 8518 : return None;
589 4 : }
590 4 : let idx = snap_index_tenant()?;
591 4 : let doc = st
592 4 : .store
593 4 : .get(&idx, Kind::Snapshot, synth)
594 4 : .await
595 4 : .ok()
596 4 : .flatten()?;
597 4 : let owner = TenantId::new(doc.get("tenant")?.as_str()?).ok()?;
598 4 : Some((owner, doc.get("snapshot")?.as_str()?.to_owned()))
599 8522 : }
600 :
601 : /// The tenant an operation is judged under. 6.3.22 scopes a request to a
602 : /// Snapshot by swapping the tenant for the snapshot's internal one, so the
603 : /// handlers below it serve the frozen copy; the subject asking is still the
604 : /// owner tenant's. An engine keyed by tenant — which is the shape every
605 : /// rule set has — would otherwise be asked about an id no deployment ever
606 : /// wrote a rule for, and every snapshot-scoped read would step out from
607 : /// under its rules. Only a synthetic tenant costs the lookup.
608 3448 : pub(crate) async fn asking_tenant(st: &AppState, tenant: &TenantId) -> TenantId {
609 3448 : match snapshot_of_synth(st, tenant.as_str()).await {
610 2 : Some((owner, _)) => owner,
611 3446 : None => tenant.clone(),
612 : }
613 3448 : }
614 :
615 : /// 5.5.15: "If an implementation determines that it is low on resources,
616 : /// it may delete one or more snapshots", considering snapshotPriority
617 : /// (lowest first; earliest expiresAt breaks ties). The resource signal is
618 : /// the per-tenant registry cap (AppState.snapshot_cap); the just-created
619 : /// snapshot is never the victim. Evicted snapshots with an endpoint are
620 : /// notified with expiresAt set before notifiedAt — the 5.3.4 deletion
621 : /// encoding.
622 : ///
623 : /// A registry that cannot be listed is an error, not an empty one: read as
624 : /// empty, the cap was never enforced while the listing failed.
625 180 : async fn evict_over_cap(st: &AppState, tenant: &TenantId, keep: &str) -> Result<(), NgsiError> {
626 4 : let victims: Vec<Value> = {
627 180 : let mut metas: Vec<Value> = st
628 180 : .store
629 180 : .list(tenant, Kind::Snapshot)
630 180 : .await?
631 178 : .into_iter()
632 178 : .filter(is_snapshot)
633 178 : .collect();
634 178 : if metas.len() <= st.snapshot_cap {
635 174 : return Ok(());
636 4 : }
637 4 : let over = metas.len().saturating_sub(st.snapshot_cap);
638 22 : metas.sort_by_key(|v| {
639 22 : (
640 22 : v.get("snapshotPriority")
641 22 : .and_then(Value::as_i64)
642 22 : .unwrap_or(5),
643 22 : v.get("expiresAt")
644 22 : .and_then(Value::as_str)
645 22 : .unwrap_or("")
646 22 : .to_owned(),
647 22 : )
648 22 : });
649 4 : metas
650 4 : .into_iter()
651 4 : .filter(|v| v.get("id").and_then(Value::as_str) != Some(keep))
652 4 : .take(over)
653 4 : .collect()
654 : };
655 4 : for mut meta in victims {
656 4 : if let Some(id) = meta.get("id").and_then(Value::as_str).map(str::to_owned) {
657 4 : snap_remove(st, tenant, &id, &meta).await;
658 0 : }
659 4 : if let Some(o) = meta.as_object_mut() {
660 4 : // deletion signal: expiresAt strictly before the notification
661 4 : o.insert("expiresAt".into(), Value::String(now_iso()));
662 4 : }
663 4 : let st2 = st.clone();
664 4 : crate::spawn(async move {
665 4 : send_notification(&st2, &meta).await;
666 4 : });
667 : }
668 4 : Ok(())
669 180 : }
670 :
671 : // ---------- 5.16.1 Create Snapshot (POST /snapshots, 6.36.3.1) ----------
672 :
673 214 : pub async fn create_snapshot(
674 214 : State(st): State<AppState>,
675 214 : CleanParams(params): CleanParams,
676 214 : headers: HeaderMap,
677 214 : body: Bytes,
678 214 : ) -> Response {
679 214 : let go = async {
680 214 : let tenant = tenant_from(&headers)?;
681 214 : check_params(¶ms, &["local"])?;
682 214 : gate!(st, &tenant, &headers, "5.16.1").await?;
683 : // 6.3.5: the @context rules apply to the snapshot body like to any
684 : // other POST payload (media type, Link header, body @context)
685 212 : let parsed = parse_body(&st, &headers, &body, BodyKind::Standard).await?;
686 200 : let mut o = validate(&parsed.value, Mode::Create)?;
687 : // The queries are written in the terms of this request's @context
688 : // (6.3.5) and run after it has been answered, so the context is
689 : // recorded with them, as a Subscription records its own.
690 160 : o.insert("__context".into(), parsed.ctx.source.clone());
691 160 : let (id, meta) = new_meta(o, &tenant, &headers)?;
692 160 : snap_insert(&st, &tenant, &id, &meta).await?;
693 : // the cap cannot be checked: the new snapshot does not stay
694 156 : if let Err(e) = evict_over_cap(&st, &tenant, &id).await {
695 2 : snap_remove(&st, &tenant, &id, &meta).await;
696 2 : return Err(e.into());
697 154 : }
698 154 : let (st2, t2, id2) = (st.clone(), tenant.clone(), id.clone());
699 154 : crate::spawn(async move {
700 149 : fill_snapshot(&st2, &t2, &id2).await;
701 146 : });
702 154 : Ok::<_, ApiError>(created(format!("/ngsi-ld/v1/snapshots/{id}"), &tenant))
703 214 : };
704 214 : go.await.unwrap_or_else(|e| e.into_response())
705 214 : }
706 :
707 : /// 5.16.1.4 background fill: execute every (temporal) query, store the
708 : /// results under the synthetic tenant, derive the status, notify.
709 149 : async fn fill_snapshot(st: &AppState, tenant: &TenantId, id: &str) {
710 149 : let Some(meta) = snap_get(st, tenant, id).await else {
711 0 : return;
712 : };
713 149 : let Some(synth) = synth_tenant(&meta) else {
714 0 : return;
715 : };
716 149 : let ctx = snapshot_context(st, tenant, &meta).await;
717 149 : let (mut n_fail, mut n_res, mut n_empty, mut copied) = (0usize, 0usize, 0usize, 0usize);
718 151 : let mut detail = |r: Result<usize, NgsiError>| -> Value {
719 151 : match r {
720 : Ok(0) => {
721 54 : n_empty = n_empty.saturating_add(1);
722 54 : json!({"resultStatus": "empty"})
723 : }
724 : Ok(_) => {
725 87 : n_res = n_res.saturating_add(1);
726 87 : json!({"resultStatus": "success"})
727 : }
728 10 : Err(e) => {
729 10 : n_fail = n_fail.saturating_add(1);
730 10 : json!({"resultStatus": "failure",
731 10 : "problemDetails": crate::negotiate::problem_value(&e)})
732 : }
733 : }
734 151 : };
735 149 : let cap = fill_cap(st);
736 149 : let mut q_details = Vec::new();
737 149 : for q in meta
738 149 : .get("snapshotQueries")
739 149 : .and_then(Value::as_array)
740 149 : .into_iter()
741 149 : .flatten()
742 : {
743 133 : if fill_cancelled(st, tenant, id, &synth).await {
744 2 : return;
745 131 : }
746 131 : let r = run_query(st, tenant, &synth, q, &ctx, cap.saturating_sub(copied)).await;
747 131 : if let Ok(n) = &r {
748 125 : copied = copied.saturating_add(*n);
749 125 : }
750 131 : q_details.push(detail(r));
751 : }
752 147 : let mut tq_details = Vec::new();
753 147 : for q in meta
754 147 : .get("snapshotTemporalQueries")
755 147 : .and_then(Value::as_array)
756 147 : .into_iter()
757 147 : .flatten()
758 : {
759 20 : if fill_cancelled(st, tenant, id, &synth).await {
760 0 : return;
761 20 : }
762 20 : let r =
763 20 : run_temporal_query(st, tenant, &synth, id, q, &ctx, cap.saturating_sub(copied)).await;
764 20 : if let Ok(n) = &r {
765 16 : copied = copied.saturating_add(*n);
766 16 : }
767 20 : tq_details.push(detail(r));
768 : }
769 147 : if fill_cancelled(st, tenant, id, &synth).await {
770 0 : return;
771 147 : }
772 147 : let status = if n_fail == 0 && n_empty == 0 && n_res > 0 {
773 83 : "success"
774 64 : } else if n_res > 0 {
775 0 : "partial"
776 64 : } else if n_empty > 0 {
777 54 : "empty"
778 : } else {
779 10 : "failure"
780 : };
781 147 : finish(st, tenant, id, status, Some((q_details, tq_details))).await;
782 146 : }
783 :
784 : /// The @context the creating request was expanded with (6.3.5), resolved
785 : /// within the Snapshot's Tenant (5.5.10); the core context for a Snapshot
786 : /// stored without one, or one whose context no longer loads.
787 149 : async fn snapshot_context(
788 149 : st: &AppState,
789 149 : tenant: &TenantId,
790 149 : meta: &Value,
791 149 : ) -> std::sync::Arc<antares_jsonld::Context> {
792 149 : match meta.get("__context").filter(|v| !v.is_null()) {
793 149 : Some(v) => st
794 149 : .loader
795 149 : .resolve_quiet_for(tenant, v)
796 149 : .await
797 149 : .unwrap_or_else(|_| st.loader.core()),
798 0 : None => st.loader.core(),
799 : }
800 149 : }
801 :
802 : /// One 5.2.23 Query executed per the DISTRIBUTED query behaviour
803 : /// (5.16.1.4 -> 5.7.2.4): local content plus every matching Context
804 : /// Source; results are copied into the snapshot's synthetic tenant. Every
805 : /// page is retrieved ("all pages are to be retrieved completely") within
806 : /// `budget` documents — beyond it nothing is copied (5.5.6).
807 131 : async fn run_query(
808 131 : st: &AppState,
809 131 : tenant: &TenantId,
810 131 : synth: &TenantId,
811 131 : q: &Value,
812 131 : ctx: &antares_jsonld::Context,
813 131 : budget: usize,
814 131 : ) -> Result<usize, NgsiError> {
815 131 : let qo = q
816 131 : .as_object()
817 131 : .ok_or_else(|| bad("Query must be an object".into()))?;
818 131 : let mut vp: HashMap<String, String> = HashMap::new();
819 131 : crate::paging::query_doc_params(qo, false, &mut vp)?;
820 : // the caller's own `q`, held to the caller's caps before the paged
821 : // filter reads it with the narrowed ones (threat V23)
822 131 : if let Some(q) = vp.get("q") {
823 12 : antares_ql::parse_q(q)?;
824 119 : }
825 393 : for k in ["limit", "offset", "count"] {
826 393 : vp.remove(k);
827 393 : }
828 131 : let headers = HeaderMap::new();
829 131 : let mut warnings = Vec::new();
830 131 : let fed = if crate::federation::active(&vp) {
831 131 : crate::federation::fed_query(st, tenant, &headers, ctx, &vp, &mut warnings).await?
832 : } else {
833 0 : Vec::new()
834 : };
835 : // 5.16.1.4: "If the size of the respective results require pagination,
836 : // all pages are to be retrieved completely." They are retrieved one at a
837 : // time. A query wider than the budget is refused, and it has to be
838 : // refused without first holding its whole match set in memory to measure
839 : // it: the store's pre-LIMIT count is that measurement. The count is the
840 : // match set's only while every predicate reached the store — an
841 : // idPattern is applied after it (5.2.33), so with one present the
842 : // documents that survive are counted instead.
843 131 : let exact_total = !vp.contains_key("idPattern");
844 : // Federated candidates are merged after the store answers (5.7.2.4), so
845 : // that round takes no page and is the whole match set; only a local
846 : // query walks on.
847 131 : let mut fed = fed;
848 131 : let mut n = 0usize;
849 131 : let mut offset = 0usize;
850 : loop {
851 : // Only the first round asks the store to count: the pre-LIMIT total
852 : // makes it visit the whole match set, and the budget needs that once.
853 : // Later rounds carry the next-page hint in the same member, which is
854 : // not a count and is not read as one.
855 137 : let counted = offset == 0;
856 137 : if counted {
857 131 : vp.insert("count".into(), "true".into());
858 131 : } else {
859 6 : vp.remove("count");
860 6 : }
861 137 : let batch = crate::entities::filter_entities_paged(
862 137 : st,
863 137 : tenant,
864 137 : &vp,
865 137 : ctx,
866 137 : std::mem::take(&mut fed),
867 137 : crate::entities::Pushdown {
868 137 : page: Some((offset, st.max_limit)),
869 137 : proj: None,
870 137 : order_keys: None,
871 137 : },
872 137 : // a snapshot under a narrowing is refused at its gate
873 137 : None,
874 137 : )
875 137 : .await
876 137 : .map_err(|e| match e {
877 0 : ApiError::Ngsi(n) => n,
878 0 : other => opaque("query execution", &other),
879 0 : })?;
880 137 : if exact_total && counted {
881 131 : if let Some(total) = batch.total.filter(|_| batch.paged) {
882 125 : if total > budget {
883 4 : return Err(too_many(total, budget));
884 121 : }
885 6 : }
886 6 : }
887 133 : let rows = batch.rows;
888 133 : let got = batch.docs.len();
889 133 : if n.saturating_add(got) > budget {
890 0 : return Err(too_many(n.saturating_add(got), budget));
891 133 : }
892 : // A write the store refuses fails the query. What this returns is
893 : // what the caller adds to `copied`, and `copied` decides both the
894 : // budget left for the next query and whether the synthetic tenant is
895 : // materialized — an empty snapshot has to read as empty rather than
896 : // as a tenant that never existed. Counting a copy that never happened
897 : // costs the snapshot both, and 5.2.41 would publish the query as a
898 : // success besides.
899 133 : for doc in batch.docs {
900 110 : if let Some(id) = doc.get("id").and_then(Value::as_str) {
901 110 : st.store
902 110 : .create(synth, Kind::Entity, id, doc.clone())
903 110 : .await?;
904 0 : }
905 : }
906 131 : n = n.saturating_add(got);
907 131 : match crate::entities::next_scan_offset(offset, rows, 0, batch.paged, st.max_limit) {
908 6 : Some(next) => offset = next,
909 125 : None => break,
910 : }
911 : }
912 125 : Ok(n)
913 131 : }
914 :
915 : /// One temporal Query (temporalQ mandatory) — ids via the 5.7.4.4 path,
916 : /// full evolutions copied via the store, every page retrieved but never
917 : /// more than `budget` evolutions (5.5.6).
918 20 : async fn run_temporal_query(
919 20 : st: &AppState,
920 20 : tenant: &TenantId,
921 20 : synth: &TenantId,
922 20 : id: &str,
923 20 : q: &Value,
924 20 : ctx: &antares_jsonld::Context,
925 20 : budget: usize,
926 20 : ) -> Result<usize, NgsiError> {
927 20 : let qo = q
928 20 : .as_object()
929 20 : .ok_or_else(|| bad("Query must be an object".into()))?;
930 20 : let mut vp: HashMap<String, String> = HashMap::new();
931 20 : crate::paging::query_doc_params(qo, true, &mut vp)?;
932 60 : for k in ["limit", "offset", "count"] {
933 60 : vp.remove(k);
934 60 : }
935 : // The copy holds what the query selects: its attrs and its window. A
936 : // whole Evolution would hand a snapshot reader history the query never
937 : // asked for.
938 20 : let tq = crate::temporalq::TemporalQ::from_params(&vp, true)?;
939 20 : let attrs: Option<Vec<String>> = vp
940 20 : .get("attrs")
941 20 : .map(|a| a.split(',').map(|t| ctx.expand_key(t.trim())).collect());
942 20 : let tf = antares_store::filter::TemporalFilter {
943 20 : attrs: attrs.as_deref(),
944 20 : range: tq.as_ref().map(|t| antares_store::filter::InstanceRange {
945 20 : timerel: &t.timerel,
946 20 : time_at: &t.time_at,
947 20 : end_time_at: t.end_time_at.as_deref(),
948 20 : timeproperty: &t.timeproperty,
949 20 : }),
950 20 : timeproperty: tq
951 20 : .as_ref()
952 20 : .map_or("observedAt", |t| t.timeproperty.as_str()),
953 20 : ..Default::default()
954 : };
955 20 : let mut headers = HeaderMap::new();
956 20 : if let Ok(v) = tenant.as_str().parse() {
957 20 : headers.insert("NGSILD-Tenant", v);
958 20 : }
959 : // The query expands its terms with `ctx`, passed below; a URL context
960 : // rides as the Link a request carries (6.3.5) as well, which a Context
961 : // Source the query reaches is handed.
962 20 : if let Some(url) = ctx.source.as_str() {
963 18 : let link = format!(
964 : "<{url}>; rel=\"http://www.w3.org/ns/json-ld#context\"; type=\"application/ld+json\""
965 : );
966 18 : if let Ok(v) = link.parse() {
967 18 : headers.insert("Link", v);
968 18 : }
969 2 : }
970 : // 5.16.1.4: "If the size of the respective results require pagination,
971 : // all pages are to be retrieved completely."
972 20 : let mut n = 0usize;
973 20 : let mut offset = 0usize;
974 : loop {
975 : // a snapshot deleted (or evicted) mid-fill stops the paging; what is
976 : // already copied is reaped by the caller
977 22 : if snap_get(st, tenant, id).await.is_none() {
978 0 : break;
979 22 : }
980 22 : vp.insert("limit".into(), st.max_limit.to_string());
981 22 : vp.insert("offset".into(), offset.to_string());
982 : // The fill runs after the request that created the snapshot has
983 : // been answered, under a header map that is not the client's, so
984 : // the narrowing the creating subject was given is not in scope here
985 : // — a snapshot under a policy is P5's box (ADR-0020: a `Filter` on
986 : // a fill is a `Deny`, and the fill records whose it is).
987 : // The merged documents come back beside the response: an Evolution
988 : // this broker holds is copied from the store below, whole, and one a
989 : // registered Context Source holds exists only here.
990 22 : let mut merged: Vec<Value> = Vec::new();
991 22 : let resp = crate::temporal::query_temporal_collected(
992 22 : st,
993 22 : &vp,
994 22 : &headers,
995 22 : &crate::policy::Filter::default(),
996 22 : Some(&mut merged),
997 22 : Some(std::sync::Arc::new(ctx.clone())),
998 22 : )
999 22 : .await
1000 22 : .map_err(|e| match e {
1001 0 : ApiError::Ngsi(n) => n,
1002 0 : other => opaque("temporal query execution", &other),
1003 0 : })?;
1004 22 : let bytes = axum::body::to_bytes(resp.into_body(), st.max_answer_bytes)
1005 22 : .await
1006 22 : .map_err(|_| NgsiError::TooManyResults(crate::bounds::ANSWER_TOO_LARGE.into()))?;
1007 18 : let arr: Vec<Value> = serde_json::from_slice::<Value>(&bytes)
1008 18 : .ok()
1009 18 : .and_then(|v| v.as_array().cloned())
1010 18 : .unwrap_or_default();
1011 18 : let got = arr.len();
1012 18 : if n.saturating_add(got) > budget {
1013 0 : return Err(too_many(n.saturating_add(got), budget));
1014 18 : }
1015 22 : for d in arr {
1016 22 : let Some(eid) = d.get("id").and_then(Value::as_str) else {
1017 0 : continue;
1018 : };
1019 : // `None` is an Entity with no Temporal Evolution to copy; an Err
1020 : // is the store refusing the read, which is not the same thing and
1021 : // is not this snapshot's to swallow.
1022 : //
1023 : // The copy REPLACES: a snapshot carrying both kinds of query runs
1024 : // its Entity queries first, and copying an Entity into the
1025 : // synthetic tenant fires the 5.6.11 auto-record hook, which leaves
1026 : // a Temporal Evolution holding only the instant of the copy. An
1027 : // insert-if-absent would find that stub and keep it, so the
1028 : // snapshot would answer temporal reads with one instance of a
1029 : // history the source has many of — and report the query a success.
1030 22 : let held = st
1031 22 : .temporal
1032 22 : .get_temporal(tenant, eid, &tf)
1033 22 : .await?
1034 22 : .map(|doc| selected(doc, attrs.as_deref(), tq.as_ref()));
1035 : // Nothing local means the Evolution came from a registered
1036 : // Context Source (5.7.4.4). The query's merged answer is every
1037 : // instance of it this broker was given, so it is what the
1038 : // snapshot keeps — a local copy is the source's whole history,
1039 : // a remote one is as much of it as the fan-out returned.
1040 22 : let doc = match held {
1041 20 : Some(doc) => Some(doc),
1042 2 : None => merged
1043 2 : .iter()
1044 4 : .find(|m| m.get("id").and_then(Value::as_str) == Some(eid))
1045 2 : .cloned(),
1046 : };
1047 22 : if let Some(doc) = doc {
1048 22 : st.temporal.upsert(synth, eid, doc).await?;
1049 22 : n = n.saturating_add(1);
1050 0 : }
1051 : }
1052 18 : if got < st.max_limit {
1053 16 : break;
1054 2 : }
1055 2 : offset = offset.saturating_add(st.max_limit);
1056 : }
1057 16 : Ok(n)
1058 20 : }
1059 :
1060 : /// An Evolution cut to the Attributes `attrs` names and the instances in
1061 : /// `tq`'s window; meta members stay. The store's window prunes too, but only
1062 : /// as a narrowing a backend may skip.
1063 20 : fn selected(
1064 20 : mut doc: Value,
1065 20 : attrs: Option<&[String]>,
1066 20 : tq: Option<&crate::temporalq::TemporalQ>,
1067 20 : ) -> Value {
1068 20 : if let Some(o) = doc.as_object_mut() {
1069 104 : o.retain(|k, v| {
1070 104 : if antares_model::is_meta(k) {
1071 80 : return true;
1072 24 : }
1073 24 : if attrs.is_some_and(|want| !want.iter().any(|w| w == k)) {
1074 4 : return false;
1075 20 : }
1076 20 : if let (Some(tq), Some(insts)) = (tq, v.as_array_mut()) {
1077 30 : insts.retain(|i| tq.instance_matches(i));
1078 0 : }
1079 20 : true
1080 104 : });
1081 0 : }
1082 20 : doc
1083 20 : }
1084 :
1085 171 : async fn finish(
1086 171 : st: &AppState,
1087 171 : tenant: &TenantId,
1088 171 : id: &str,
1089 171 : status: &str,
1090 171 : details: Option<(Vec<Value>, Vec<Value>)>,
1091 171 : ) {
1092 171 : let Some(mut meta) = snap_get(st, tenant, id).await else {
1093 0 : return;
1094 : };
1095 171 : if let Some(o) = meta.as_object_mut() {
1096 171 : o.insert("snapshotStatus".into(), Value::String(status.into()));
1097 171 : o.insert("modifiedAt".into(), Value::String(now_iso()));
1098 171 : if let Some((q, tq)) = details {
1099 147 : if !q.is_empty() {
1100 131 : o.insert("snapshotQueriesDetails".into(), Value::Array(q));
1101 131 : }
1102 147 : if !tq.is_empty() {
1103 20 : o.insert("snapshotTemporalQueriesDetails".into(), Value::Array(tq));
1104 127 : }
1105 24 : }
1106 0 : }
1107 171 : if let Some(meta) = snap_put(st, tenant, &meta, FINISH_MEMBERS).await {
1108 171 : send_notification(st, &meta).await;
1109 0 : }
1110 168 : }
1111 :
1112 : /// 5.16.6 / 5.3.4 SnapshotNotification (sent only when endpoint is set).
1113 191 : async fn send_notification(st: &AppState, meta: &Value) {
1114 191 : let Some(uri) = meta.get("endpoint").and_then(Value::as_str) else {
1115 177 : return;
1116 : };
1117 14 : if st.egress.check_url(uri).await.is_err() {
1118 0 : return;
1119 14 : }
1120 14 : let mut body = json!({
1121 14 : "id": format!("urn:ngsi-ld:SnapshotNotification:{}", uuid::Uuid::new_v4()),
1122 14 : "type": "SnapshotNotification",
1123 14 : "notifiedAt": now_iso(),
1124 14 : "snapshotId": meta.get("id").cloned().unwrap_or_default(),
1125 14 : "snapshotStatus": meta.get("snapshotStatus").cloned().unwrap_or_default(),
1126 14 : "snapshotPriority": meta.get("snapshotPriority").cloned().unwrap_or_default(),
1127 14 : "expiresAt": meta.get("expiresAt").cloned().unwrap_or_default(),
1128 : });
1129 : // 5.3.4 Table 5.3.4-1 names the temporal list
1130 : // "temporalSnapshotQueriesDetails" (unlike 5.2.41's
1131 : // "snapshotTemporalQueriesDetails") — the notification datatype's own
1132 : // table governs the notification payload.
1133 28 : for (from, to) in [
1134 14 : ("snapshotQueriesDetails", "snapshotQueriesDetails"),
1135 14 : (
1136 14 : "snapshotTemporalQueriesDetails",
1137 14 : "temporalSnapshotQueriesDetails",
1138 14 : ),
1139 14 : ] {
1140 28 : if let (Some(v), Some(body)) = (meta.get(from), body.as_object_mut()) {
1141 15 : body.insert(to.into(), v.clone());
1142 15 : }
1143 : }
1144 14 : let mut req = st.http.post(uri).header("Content-Type", "application/json");
1145 14 : if let Some(ri) = meta.get("receiverInfo").and_then(Value::as_array) {
1146 10 : for kv in ri {
1147 10 : if let (Some(k), Some(v)) = (
1148 10 : kv.get("key").and_then(Value::as_str),
1149 10 : kv.get("value").and_then(Value::as_str),
1150 10 : ) {
1151 10 : req = req.header(k, v);
1152 10 : }
1153 : }
1154 4 : }
1155 14 : let req = req.body(serde_json::to_vec(&body).unwrap_or_default());
1156 14 : let _ = antares_jsonld::io_deadline(req.send(), 8_000).await;
1157 185 : }
1158 :
1159 : // ---------- 5.16.7 Purge Snapshots (DELETE /snapshots, 6.36.3.2) ----------
1160 :
1161 60 : pub async fn purge_snapshots(
1162 60 : State(st): State<AppState>,
1163 60 : CleanParams(params): CleanParams,
1164 60 : headers: HeaderMap,
1165 60 : ) -> Response {
1166 60 : let go = async {
1167 60 : let tenant = tenant_from(&headers)?;
1168 60 : check_params(¶ms, &["q", "local"])?;
1169 60 : gate!(st, &tenant, &headers, "5.16.7").await?;
1170 : // 5.16.7.4: the query is mandatory and restricted to Snapshot members
1171 58 : let q = params
1172 58 : .get("q")
1173 58 : .ok_or_else(|| bad("purge requires a q over Snapshot members (5.16.7.4)".into()))?;
1174 56 : let ast = antares_ql::parse_q(q)?;
1175 : // 5.16.7.4: the query is "restricted to members of the Snapshot
1176 : // data type" (Tables 5.2.41-1/-2)
1177 : const MEMBERS: [&str; 15] = [
1178 : "id",
1179 : "type",
1180 : "snapshotQueries",
1181 : "snapshotTemporalQueries",
1182 : "snapshotLifetime",
1183 : "snapshotPriority",
1184 : "endpoint",
1185 : "receiverInfo",
1186 : "snapshotStatus",
1187 : "snapshotQueriesDetails",
1188 : "snapshotTemporalQueriesDetails",
1189 : "createdAt",
1190 : "modifiedAt",
1191 : "expiresAt",
1192 : "lastUsedAt",
1193 : ];
1194 56 : if let Some(alien) = ast
1195 56 : .attribute_paths()
1196 56 : .into_iter()
1197 58 : .find(|a| !MEMBERS.contains(a))
1198 : {
1199 12 : return Err(bad(format!(
1200 12 : "purge q is restricted to Snapshot members (5.16.7.4): {alien:?}"
1201 12 : ))
1202 12 : .into());
1203 44 : }
1204 : // `endpoint` (its userinfo) and `receiverInfo` are the credentials
1205 : // the 5.16.6 notification is sent with, which every read answers
1206 : // masked (`present`). A purge matching on them answers 204 either
1207 : // way and shows by which snapshots are gone whether a guessed
1208 : // token matched, so a q may not name them.
1209 44 : if let Some(secret) = ast
1210 44 : .attribute_paths()
1211 44 : .into_iter()
1212 46 : .find(|a| ["endpoint", "receiverInfo"].contains(a))
1213 : {
1214 12 : return Err(bad(format!(
1215 12 : "purge q may not filter on {secret:?}: it holds the notification \
1216 12 : credentials (5.16.7.4)"
1217 12 : ))
1218 12 : .into());
1219 32 : }
1220 32 : let ctx = st.loader.core();
1221 : // a listing that fails purges nothing, and says so: a 204 here
1222 : // would report a purge that never ran
1223 32 : let victims: Vec<Value> = st
1224 32 : .store
1225 32 : .list(&tenant, Kind::Snapshot)
1226 32 : .await?
1227 30 : .into_iter()
1228 30 : .filter(|meta| is_snapshot(meta) && crate::registry::csf_matches(&ast, meta, &ctx))
1229 30 : .collect();
1230 30 : for meta in victims {
1231 8 : if let Some(id) = meta.get("id").and_then(Value::as_str) {
1232 8 : snap_remove(&st, &tenant, id, &meta).await;
1233 0 : }
1234 : }
1235 30 : Ok::<_, ApiError>(no_content(&tenant))
1236 60 : };
1237 60 : go.await.unwrap_or_else(|e| e.into_response())
1238 60 : }
1239 :
1240 : /// 5.16.2.4 / 5.16.3.4 / 5.16.4.4 / 5.16.5.4: every /snapshots/{id} method
1241 : /// opens the same way — the tenant it runs in, `local` (6.3.18) as the only
1242 : /// parameter this resource takes, and an id that must be a valid URI before
1243 : /// the store is touched at all.
1244 442 : fn open_snapshot(
1245 442 : params: &HashMap<String, String>,
1246 442 : headers: &HeaderMap,
1247 442 : id: &str,
1248 442 : ) -> ApiResult<TenantId> {
1249 442 : let tenant = tenant_from(headers)?;
1250 442 : check_params(params, &["local"])?;
1251 442 : antares_model::EntityId::new(id)
1252 442 : .map_err(|_| bad(format!("snapshot id is not a valid URI: {id:?}")))?;
1253 402 : Ok(tenant)
1254 442 : }
1255 :
1256 : // ---------- 5.16.3/5.16.4/5.16.5: /snapshots/{id} (6.37) ----------
1257 :
1258 302 : pub async fn retrieve_snapshot(
1259 302 : State(st): State<AppState>,
1260 302 : Path(id): Path<String>,
1261 302 : CleanParams(params): CleanParams,
1262 302 : headers: HeaderMap,
1263 302 : ) -> Response {
1264 302 : let go = async {
1265 302 : let tenant = open_snapshot(¶ms, &headers, &id)?;
1266 292 : gate!(st, &tenant, &headers, "5.16.3", ids: &[&id]).await?;
1267 292 : let accept = parse_accept(&headers)?;
1268 292 : let meta = snap_get(&st, &tenant, &id)
1269 292 : .await
1270 292 : .ok_or_else(|| NgsiError::ResourceNotFound(format!("snapshot {id} not found")))?;
1271 258 : let ctx = st.loader.core();
1272 258 : Ok::<_, ApiError>(respond(
1273 258 : StatusCode::OK,
1274 258 : present(&meta),
1275 258 : &ctx,
1276 258 : accept,
1277 258 : &tenant,
1278 258 : ))
1279 302 : };
1280 302 : go.await.unwrap_or_else(|e| e.into_response())
1281 302 : }
1282 :
1283 54 : pub async fn update_snapshot(
1284 54 : State(st): State<AppState>,
1285 54 : Path(id): Path<String>,
1286 54 : CleanParams(params): CleanParams,
1287 54 : headers: HeaderMap,
1288 54 : body: Bytes,
1289 54 : ) -> Response {
1290 54 : let go = async {
1291 54 : let tenant = open_snapshot(¶ms, &headers, &id)?;
1292 44 : gate!(st, &tenant, &headers, "5.16.4", ids: &[&id]).await?;
1293 44 : let accept = parse_accept(&headers)?;
1294 : // 5.16.4.4 binds this operation to 5.5.4, hence to the 6.3.5 rules;
1295 : // a PATCH may name its body application/merge-patch+json (6.3.4)
1296 44 : let v = parse_body(&st, &headers, &body, BodyKind::MergePatch)
1297 44 : .await?
1298 : .value;
1299 40 : let mut frag = validate(&v, Mode::Update)?;
1300 30 : let mut meta = snap_get(&st, &tenant, &id)
1301 30 : .await
1302 30 : .ok_or_else(|| NgsiError::ResourceNotFound(format!("snapshot {id} not found")))?;
1303 20 : if let Some(o) = meta.as_object_mut() {
1304 20 : keep_receiver_info(&mut frag, o)?;
1305 : // 5.16.4.4 / 5.5.8 merge of the updatable members
1306 72 : for k in [
1307 18 : "snapshotLifetime",
1308 18 : "snapshotPriority",
1309 18 : "endpoint",
1310 18 : "receiverInfo",
1311 18 : ] {
1312 72 : match frag.get(k) {
1313 36 : None => {}
1314 18 : Some(Value::Null) => {
1315 18 : o.remove(k);
1316 18 : }
1317 18 : Some(v) => {
1318 18 : o.insert(k.into(), v.clone());
1319 18 : }
1320 : }
1321 : }
1322 18 : if frag.contains_key("snapshotLifetime") {
1323 : // "it is possible to indirectly update expiresAt" (5.2.41)
1324 8 : o.insert("expiresAt".into(), Value::String(expires_at(o)?));
1325 10 : }
1326 : // a removed priority is the default one again (Table 5.2.41-1)
1327 18 : o.entry("snapshotPriority".to_owned())
1328 18 : .or_insert(Value::Number(5.into()));
1329 18 : o.insert("modifiedAt".into(), Value::String(now_iso()));
1330 0 : }
1331 18 : let meta = snap_put(&st, &tenant, &meta, UPDATE_MEMBERS)
1332 18 : .await
1333 18 : .ok_or_else(|| NgsiError::ResourceNotFound(format!("snapshot {id} not found")))?;
1334 : // 5.16.6: notifications are also sent after any status update
1335 18 : let (st2, meta2) = (st.clone(), meta.clone());
1336 18 : crate::spawn(async move {
1337 16 : send_notification(&st2, &meta2).await;
1338 13 : });
1339 18 : let ctx = st.loader.core();
1340 18 : Ok::<_, ApiError>(respond(
1341 18 : StatusCode::OK,
1342 18 : present(&meta),
1343 18 : &ctx,
1344 18 : accept,
1345 18 : &tenant,
1346 18 : ))
1347 54 : };
1348 54 : go.await.unwrap_or_else(|e| e.into_response())
1349 54 : }
1350 :
1351 40 : pub async fn delete_snapshot(
1352 40 : State(st): State<AppState>,
1353 40 : Path(id): Path<String>,
1354 40 : CleanParams(params): CleanParams,
1355 40 : headers: HeaderMap,
1356 40 : ) -> Response {
1357 40 : let go = async {
1358 40 : let tenant = open_snapshot(¶ms, &headers, &id)?;
1359 30 : gate!(st, &tenant, &headers, "5.16.5", ids: &[&id]).await?;
1360 30 : let meta = snap_get(&st, &tenant, &id)
1361 30 : .await
1362 30 : .ok_or_else(|| NgsiError::ResourceNotFound(format!("snapshot {id} not found")))?;
1363 22 : snap_remove(&st, &tenant, &id, &meta).await;
1364 22 : Ok::<_, ApiError>(no_content(&tenant))
1365 40 : };
1366 40 : go.await.unwrap_or_else(|e| e.into_response())
1367 40 : }
1368 :
1369 : // ---------- 5.16.2 Clone Snapshot (POST /snapshots/{id}/clone, 6.38) ----------
1370 :
1371 46 : pub async fn clone_snapshot(
1372 46 : State(st): State<AppState>,
1373 46 : Path(id): Path<String>,
1374 46 : CleanParams(params): CleanParams,
1375 46 : headers: HeaderMap,
1376 46 : body: Bytes,
1377 46 : ) -> Response {
1378 46 : let go = async {
1379 46 : let tenant = open_snapshot(¶ms, &headers, &id)?;
1380 36 : gate!(st, &tenant, &headers, "5.16.2", ids: &[&id]).await?;
1381 34 : let src = snap_get(&st, &tenant, &id)
1382 34 : .await
1383 34 : .ok_or_else(|| NgsiError::ResourceNotFound(format!("snapshot {id} not found")))?;
1384 : // 5.16.2.3 makes the clone body optional; a body that IS sent obeys
1385 : // the 6.3.5 @context rules
1386 26 : let v: Value = if body.is_empty() {
1387 16 : json!({})
1388 : } else {
1389 10 : parse_body(&st, &headers, &body, BodyKind::Standard)
1390 10 : .await?
1391 : .value
1392 : };
1393 26 : let mut o = validate(&v, Mode::Clone)?;
1394 : // the clone carries the source's (read-only) query lineage
1395 48 : for k in ["snapshotQueries", "snapshotTemporalQueries"] {
1396 48 : if let Some(qv) = src.get(k) {
1397 8 : o.insert(k.into(), qv.clone());
1398 40 : }
1399 : }
1400 24 : let (new_id, meta) = new_meta(o, &tenant, &headers)?;
1401 24 : snap_insert(&st, &tenant, &new_id, &meta).await?;
1402 : // a clone is a new snapshot: 5.5.15 resource pressure applies to it
1403 : // exactly as it does to a create, so cloning cannot grow the
1404 : // registry past the cap
1405 24 : if let Err(e) = evict_over_cap(&st, &tenant, &new_id).await {
1406 0 : snap_remove(&st, &tenant, &new_id, &meta).await;
1407 0 : return Err(e.into());
1408 24 : }
1409 24 : let (st2, t2, sid, nid) = (st.clone(), tenant.clone(), id.clone(), new_id.clone());
1410 24 : crate::spawn(async move {
1411 24 : clone_fill(&st2, &t2, &sid, &nid).await;
1412 24 : });
1413 24 : Ok::<_, ApiError>(created(format!("/ngsi-ld/v1/snapshots/{new_id}"), &tenant))
1414 46 : };
1415 46 : go.await.unwrap_or_else(|e| e.into_response())
1416 46 : }
1417 :
1418 : /// 5.16.2.4 background copy: all Entity and Temporal data of the source.
1419 : /// A source still `preparing` is waited for (its copy is not complete
1420 : /// until its fill finishes), for as long as both snapshots exist; a source
1421 : /// that ends in `failure` or disappears makes the clone a `failure`. The
1422 : /// copy stops at the same ceiling a fill does (5.5.6).
1423 24 : async fn clone_fill(st: &AppState, tenant: &TenantId, src_id: &str, new_id: &str) {
1424 24 : let src = loop {
1425 48 : let Some(src) = snap_get(st, tenant, src_id).await else {
1426 0 : finish(st, tenant, new_id, "failure", None).await;
1427 0 : return;
1428 : };
1429 48 : if src.get("snapshotStatus").and_then(Value::as_str) != Some("preparing") {
1430 24 : break src;
1431 24 : }
1432 24 : if snap_get(st, tenant, new_id).await.is_none() {
1433 0 : return;
1434 24 : }
1435 : // the browser's own timer on wasm32 (tokio time has no reactor there)
1436 : #[cfg(not(target_arch = "wasm32"))]
1437 24 : tokio::time::sleep(std::time::Duration::from_millis(CLONE_POLL_MS.into())).await;
1438 : #[cfg(target_arch = "wasm32")]
1439 : gloo_timers::future::TimeoutFuture::new(CLONE_POLL_MS).await;
1440 : };
1441 24 : let (Some(from), Some(to)) = (
1442 24 : synth_tenant(&src),
1443 24 : snap_get(st, tenant, new_id)
1444 24 : .await
1445 24 : .as_ref()
1446 24 : .and_then(synth_tenant),
1447 : ) else {
1448 0 : finish(st, tenant, new_id, "failure", None).await;
1449 0 : return;
1450 : };
1451 24 : let failed = src.get("snapshotStatus").and_then(Value::as_str) == Some("failure")
1452 20 : || copy_snapshot_data(st, &from, &to, fill_cap(st))
1453 20 : .await
1454 20 : .is_err();
1455 24 : if fill_cancelled(st, tenant, new_id, &to).await {
1456 0 : return;
1457 24 : }
1458 24 : finish(
1459 24 : st,
1460 24 : tenant,
1461 24 : new_id,
1462 24 : if failed { "failure" } else { "success" },
1463 24 : None,
1464 : )
1465 24 : .await;
1466 24 : }
1467 :
1468 : /// How often a clone looks whether its source finished preparing.
1469 : const CLONE_POLL_MS: u32 = 50;
1470 :
1471 : /// Copy every Entity (one `list_page` at a time) and every Temporal
1472 : /// Evolution of `from` into `to`, at most `cap` documents of each. An
1473 : /// Evolution REPLACES what `to` holds: copying an Entity fires the 5.6.11
1474 : /// auto-record hook, which leaves a stub Evolution of one instance.
1475 24 : async fn copy_snapshot_data(
1476 24 : st: &AppState,
1477 24 : from: &TenantId,
1478 24 : to: &TenantId,
1479 24 : cap: usize,
1480 24 : ) -> Result<(), NgsiError> {
1481 24 : let mut n = 0usize;
1482 24 : let mut after: Option<String> = None;
1483 : loop {
1484 424 : let page = st
1485 424 : .store
1486 424 : .list_page(from, Kind::Entity, after.as_deref(), st.max_limit)
1487 424 : .await?;
1488 424 : let Some(last) = page
1489 424 : .last()
1490 424 : .and_then(|d| d.get("id"))
1491 424 : .and_then(Value::as_str)
1492 : else {
1493 2 : break;
1494 : };
1495 422 : after = Some(last.to_owned());
1496 422 : n = n.saturating_add(page.len());
1497 422 : if n > cap {
1498 4 : return Err(too_many(n, cap));
1499 418 : }
1500 418 : let full = page.len() >= st.max_limit;
1501 426 : for doc in page {
1502 426 : if let Some(id) = doc.get("id").and_then(Value::as_str) {
1503 426 : st.store.create(to, Kind::Entity, id, doc.clone()).await?;
1504 0 : }
1505 : }
1506 418 : if !full {
1507 18 : break;
1508 400 : }
1509 : }
1510 20 : let docs = st.temporal.list(from).await?;
1511 20 : if docs.len() > cap {
1512 0 : return Err(too_many(docs.len(), cap));
1513 20 : }
1514 20 : for doc in docs {
1515 18 : if let Some(id) = doc.get("id").and_then(Value::as_str) {
1516 18 : st.temporal.upsert(to, id, doc.clone()).await?;
1517 0 : }
1518 : }
1519 20 : Ok(())
1520 24 : }
1521 :
1522 : // ---------- 6.3.22: NGSILD-Snapshot scoping middleware ----------
1523 :
1524 : /// 6.3.22 / 5.5.15: resolve the NGSILD-Snapshot header to the snapshot's
1525 : /// synthetic tenant, so every Core/Temporal handler serves the frozen copy;
1526 : /// the header is echoed on the response (and the synthetic tenant never
1527 : /// leaks into NGSILD-Tenant).
1528 55050 : pub async fn snapshot_layer(
1529 55050 : State(st): State<AppState>,
1530 55050 : req: axum::extract::Request,
1531 55050 : next: axum::middleware::Next,
1532 55050 : ) -> Response {
1533 : // 6.3.22 gives the header one value, and it decides whether the request
1534 : // is answered from a frozen copy or from live data: an ambiguous one
1535 : // would serve the request against a dataset nobody named.
1536 55050 : let sid = match single_header(req.headers(), "NGSILD-Snapshot") {
1537 54948 : Ok(None) => return next.run(req).await,
1538 98 : Ok(Some(sid)) => sid,
1539 4 : Err(e) => return e.into_response(),
1540 : };
1541 : // 6.3.22: "If the HTTP header NGSILD-Snapshot is present in the HTTP
1542 : // request, it shall also be present in HTTP response" — every exit,
1543 : // the unscoped and the error ones included, leaves through here
1544 98 : let mut resp = scoped(&st, &sid, req, next).await;
1545 98 : if let Ok(v) = sid.parse() {
1546 98 : resp.headers_mut().insert("NGSILD-Snapshot", v);
1547 98 : }
1548 98 : resp
1549 55050 : }
1550 :
1551 98 : async fn scoped(
1552 98 : st: &AppState,
1553 98 : sid: &str,
1554 98 : mut req: axum::extract::Request,
1555 98 : next: axum::middleware::Next,
1556 98 : ) -> Response {
1557 : // the Snapshot API's own resources (6.36-6.38) are never
1558 : // snapshot-scoped — that is the /snapshots RESOURCE, not any path
1559 : // merely containing the word (an Attribute may be named "snapshots")
1560 98 : if req
1561 98 : .uri()
1562 98 : .path()
1563 98 : .trim_start_matches(API_ROOT)
1564 98 : .starts_with("/snapshots")
1565 : {
1566 4 : return next.run(req).await;
1567 94 : }
1568 94 : let tenant = match tenant_from(req.headers()) {
1569 94 : Ok(t) => t,
1570 0 : Err(e) => return e.into_response(),
1571 : };
1572 94 : let Some(meta) = snap_get(st, &tenant, sid).await else {
1573 16 : return ApiError::from(NgsiError::ResourceNotFound(format!(
1574 16 : "snapshot {sid} not found"
1575 16 : )))
1576 16 : .into_response();
1577 : };
1578 78 : let Some(synth) = synth_tenant(&meta) else {
1579 0 : return ApiError::from(NgsiError::InternalError("snapshot without tenant".into()))
1580 0 : .into_response();
1581 : };
1582 78 : snap_touch(st, &tenant, sid).await;
1583 78 : if let Ok(v) = synth.as_str().parse() {
1584 78 : req.headers_mut().insert("NGSILD-Tenant", v);
1585 78 : }
1586 78 : let mut resp = next.run(req).await;
1587 : // restore the caller's tenant view (the synthetic one is internal)
1588 78 : match tenant.as_str() {
1589 78 : "default" => {
1590 68 : resp.headers_mut().remove("NGSILD-Tenant");
1591 68 : }
1592 10 : t => {
1593 10 : if let Ok(v) = t.parse() {
1594 10 : resp.headers_mut().insert("NGSILD-Tenant", v);
1595 10 : }
1596 : }
1597 : }
1598 78 : resp
1599 98 : }
1600 :
1601 : /// 5.5.6: the fill copies its results into a second (isolated) tenant, so an
1602 : /// unbounded result set doubles the stored data set in one request. The
1603 : /// implementation threshold — "up to each implementation" — is a hundred
1604 : /// full pages per snapshot; beyond it the query reports TooManyResults.
1605 181 : fn fill_cap(st: &AppState) -> usize {
1606 181 : st.max_limit.saturating_mul(100)
1607 181 : }
1608 :
1609 : /// 5.16.1.4 + 5.5.15: the copied data is reachable only through its snapshot
1610 : /// document, so a fill whose snapshot was deleted (or evicted under resource
1611 : /// pressure) must stop and free what it already copied.
1612 332 : async fn fill_cancelled(st: &AppState, tenant: &TenantId, id: &str, synth: &TenantId) -> bool {
1613 332 : if snap_get(st, tenant, id).await.is_some() {
1614 326 : return false;
1615 6 : }
1616 6 : purge_synth(st, synth).await;
1617 6 : true
1618 332 : }
1619 :
1620 : /// 5.5.6: "When a query operation is producing so many results that can
1621 : /// potentially exhaust client or server resources … implementations shall
1622 : /// raise an error of type TooManyResults" — reported per query in the
1623 : /// 5.2.42 ExecutionResultDetails of the snapshot.
1624 8 : fn too_many(got: usize, budget: usize) -> NgsiError {
1625 8 : NgsiError::TooManyResults(format!(
1626 8 : "snapshot query yields {got} results, {budget} left of the snapshot ceiling"
1627 8 : ))
1628 8 : }
1629 :
1630 : #[cfg(test)]
1631 : mod clause_5_5_6 {
1632 : use super::*;
1633 :
1634 : /// 5.5.6 InternalError: the client-visible `detail` — served in
1635 : /// snapshotQueriesDetails.problemDetails and copied into the 5.3.4
1636 : /// SnapshotNotification body — must be generic. The Debug/IO text of
1637 : /// the underlying failure belongs in the server log only.
1638 : #[test]
1639 4 : fn snapshot_fill_error_detail_is_generic() {
1640 4 : let inner = ApiError::Bare(StatusCode::UNSUPPORTED_MEDIA_TYPE);
1641 4 : let pd = crate::negotiate::problem_value(&opaque("query execution", &inner));
1642 4 : let detail = pd["detail"].as_str().unwrap_or_default();
1643 4 : assert_eq!(detail, "query execution failed");
1644 4 : assert!(
1645 4 : !detail.contains("Bare") && !detail.contains("Unsupported"),
1646 : "internal error text leaked into the client-visible detail: {detail}"
1647 : );
1648 4 : }
1649 : }
1650 :
1651 : #[cfg(test)]
1652 : mod clause_5_16 {
1653 : use super::*;
1654 : use axum::body::Body;
1655 : use axum::http::Request;
1656 : use tower::ServiceExt;
1657 :
1658 72 : async fn state() -> AppState {
1659 72 : crate::wired_state("antares-snapshots").await
1660 72 : }
1661 :
1662 : /// 5.16 with RFC 9110 §9.2.1: reading an expired Snapshot must not
1663 : /// write. The meta row survives the read; the sweep is what frees it,
1664 : /// its reverse-index entry and the synthetic tenant's data.
1665 : #[tokio::test(flavor = "multi_thread")]
1666 4 : async fn an_expired_snapshot_is_refused_by_a_read_that_writes_nothing() {
1667 4 : let st = state().await;
1668 4 : let t = TenantId::default();
1669 4 : let id = "urn:ngsi-ld:Snapshot:stale";
1670 4 : let synth = "snap-default-stale";
1671 4 : st.store
1672 4 : .create(
1673 4 : &t,
1674 4 : Kind::Snapshot,
1675 4 : id,
1676 4 : json!({"id": id, "type": "Snapshot", "__tenant": synth,
1677 4 : "expiresAt": "2000-01-01T00:00:00Z"}),
1678 4 : )
1679 4 : .await
1680 4 : .expect("seed");
1681 4 : let idx = TenantId::new_internal("snap-index").expect("index tenant");
1682 4 : st.store
1683 4 : .create(&idx, Kind::Snapshot, synth, json!({"id": synth}))
1684 4 : .await
1685 4 : .expect("seed index");
1686 :
1687 4 : assert!(snap_get(&st, &t, id).await.is_none(), "5.16: not served");
1688 4 : assert!(
1689 4 : st.store
1690 4 : .get(&t, Kind::Snapshot, id)
1691 4 : .await
1692 4 : .expect("store")
1693 4 : .is_some(),
1694 : "the read deleted the meta: a GET must be safe (RFC 9110 9.2.1)"
1695 : );
1696 :
1697 4 : assert_eq!(sweep_expired_snapshots(&st, &t).await, 1, "the sweep reaps");
1698 4 : assert!(
1699 4 : st.store
1700 4 : .get(&t, Kind::Snapshot, id)
1701 4 : .await
1702 4 : .expect("store")
1703 4 : .is_none(),
1704 : "the sweep left the meta behind"
1705 : );
1706 : // The teardown runs behind the sweep: the copy is dropped first and
1707 : // the entry that names it last, so a broker that stops in the middle
1708 : // leaves a pointer to data that is still there.
1709 4 : for _ in 0..100 {
1710 7 : if st
1711 7 : .store
1712 7 : .get(&idx, Kind::Snapshot, synth)
1713 7 : .await
1714 7 : .expect("store")
1715 7 : .is_none()
1716 4 : {
1717 4 : return;
1718 4 : }
1719 4 : tokio::time::sleep(std::time::Duration::from_millis(20)).await;
1720 4 : }
1721 4 : panic!("the sweep left the reverse-index entry behind");
1722 4 : }
1723 :
1724 : /// The sweep takes the expiry and nothing else: a live Snapshot and the
1725 : /// reverse-index markers that share its storage both survive.
1726 : #[tokio::test]
1727 4 : async fn the_sweep_keeps_a_live_snapshot_and_the_index_markers() {
1728 4 : let st = state().await;
1729 4 : let t = TenantId::default();
1730 4 : let live = "urn:ngsi-ld:Snapshot:live";
1731 4 : st.store
1732 4 : .create(
1733 4 : &t,
1734 4 : Kind::Snapshot,
1735 4 : live,
1736 4 : json!({"id": live, "type": "Snapshot",
1737 4 : "expiresAt": "2999-01-01T00:00:00Z"}),
1738 4 : )
1739 4 : .await
1740 4 : .expect("seed");
1741 : // a marker doc: same Kind, not a Snapshot, and long past any expiry
1742 4 : st.store
1743 4 : .create(
1744 4 : &t,
1745 4 : Kind::Snapshot,
1746 4 : "snap-default-marker",
1747 4 : json!({"id": "snap-default-marker", "expiresAt": "2000-01-01T00:00:00Z"}),
1748 4 : )
1749 4 : .await
1750 4 : .expect("seed marker");
1751 :
1752 4 : assert_eq!(sweep_expired_snapshots(&st, &t).await, 0);
1753 4 : assert!(snap_get(&st, &t, live).await.is_some(), "live one survives");
1754 4 : assert!(
1755 4 : st.store
1756 4 : .get(&t, Kind::Snapshot, "snap-default-marker")
1757 4 : .await
1758 4 : .expect("store")
1759 4 : .is_some(),
1760 4 : "the sweep took a reverse-index marker for a Snapshot"
1761 4 : );
1762 4 : }
1763 :
1764 : /// One request through the full router — the 6.3.22 middleware included.
1765 617 : async fn send(
1766 617 : st: &AppState,
1767 617 : method: &str,
1768 617 : path: &str,
1769 617 : body: Option<(&str, String)>,
1770 617 : extra: &[(&str, &str)],
1771 617 : ) -> (StatusCode, HeaderMap, Value) {
1772 617 : let mut b = Request::builder().method(method).uri(path);
1773 617 : for (k, v) in extra {
1774 24 : b = b.header(*k, *v);
1775 24 : }
1776 617 : let req = match body {
1777 524 : Some((ct, payload)) => b
1778 524 : .header("Content-Type", ct)
1779 524 : .header("Content-Length", payload.len())
1780 524 : .body(Body::from(payload)),
1781 93 : None => b.body(Body::empty()),
1782 : }
1783 617 : .expect("request");
1784 617 : let res = crate::router(st.clone())
1785 617 : .oneshot(req)
1786 617 : .await
1787 617 : .expect("response");
1788 617 : let status = res.status();
1789 617 : let headers = res.headers().clone();
1790 617 : let bytes = axum::body::to_bytes(res.into_body(), usize::MAX)
1791 617 : .await
1792 617 : .expect("body");
1793 617 : let body = if bytes.is_empty() {
1794 496 : Value::Null
1795 : } else {
1796 121 : serde_json::from_slice(&bytes).unwrap_or(Value::Null)
1797 : };
1798 617 : (status, headers, body)
1799 617 : }
1800 :
1801 : /// 5.16.4.4 modifies the Snapshot "as mandated by clause 5.5.8": a
1802 : /// member whose value is an NGSI-LD Null (JSON null or
1803 : /// "urn:ngsi-ld:null") is removed, and 6.3.4 lets a PATCH name its body
1804 : /// application/merge-patch+json. The type checks refused the null with
1805 : /// 400, the literal "urn:ngsi-ld:null" was stored as the endpoint, and a
1806 : /// merge-patch body was 415.
1807 : #[tokio::test(flavor = "multi_thread")]
1808 4 : async fn clause_5_16_4_4_an_ngsi_ld_null_removes_a_member() {
1809 4 : let st = state().await;
1810 4 : let snap = json!({"type": "Snapshot", "endpoint": "http://127.0.0.1:9/n",
1811 4 : "snapshotPriority": 3, "snapshotLifetime": "PT1H",
1812 4 : "receiverInfo": [{"key": "X-A", "value": "1"}],
1813 4 : "snapshotQueries": [{"type": "Query", "entities": [{"type": "Vehicle"}]}]});
1814 4 : let (status, headers, b) = post_json(&st, "/ngsi-ld/v1/snapshots", &snap).await;
1815 4 : assert_eq!(status, StatusCode::CREATED, "{b}");
1816 4 : let loc = header(&headers, "Location").expect("Location").to_owned();
1817 4 : let patch = json!({"endpoint": null, "receiverInfo": "urn:ngsi-ld:null",
1818 4 : "snapshotPriority": null, "snapshotLifetime": "urn:ngsi-ld:null"});
1819 4 : let (status, _, body) = send(
1820 4 : &st,
1821 4 : "PATCH",
1822 4 : &loc,
1823 4 : Some(("application/merge-patch+json", patch.to_string())),
1824 4 : &[],
1825 : )
1826 4 : .await;
1827 4 : assert_eq!(status, StatusCode::OK, "{body}");
1828 12 : for k in ["endpoint", "receiverInfo", "snapshotLifetime"] {
1829 12 : assert!(body.get(k).is_none(), "{k} not removed: {body}");
1830 : }
1831 4 : assert_eq!(body["snapshotPriority"], 5, "the default priority: {body}");
1832 : // a null where no update removes it stays a wrong type
1833 4 : let mut bad = snap.clone();
1834 4 : bad["endpoint"] = Value::Null;
1835 4 : let (status, _, b) = post_json(&st, "/ngsi-ld/v1/snapshots", &bad).await;
1836 4 : assert_eq!(status, StatusCode::BAD_REQUEST, "{b}");
1837 4 : }
1838 :
1839 : /// 5.16.6 notifies by HTTP POST: an endpoint written `http:user:pass@host`
1840 : /// (no `//`) is refused on create and on update, without the credential
1841 : /// in the answer; stored, its userinfo would be neither masked nor sealed
1842 : /// and reqwest would still send it.
1843 : #[tokio::test]
1844 4 : async fn an_endpoint_without_the_double_slash_is_refused_unechoed() {
1845 4 : let st = state().await;
1846 4 : let uri = "http:hook:pa55word@localhost:1/n";
1847 4 : let snap = json!({"type": "Snapshot", "endpoint": uri,
1848 4 : "snapshotQueries": [{"type": "Query", "entities": [{"type": "Vehicle"}]}]});
1849 4 : let (status, _, b) = post_json(&st, "/ngsi-ld/v1/snapshots", &snap).await;
1850 4 : assert_eq!(status, StatusCode::BAD_REQUEST, "{b}");
1851 4 : assert!(
1852 4 : !b.to_string().contains("pa55word"),
1853 : "the credential was echoed"
1854 : );
1855 4 : let ok = json!({"type": "Snapshot",
1856 4 : "snapshotQueries": [{"type": "Query", "entities": [{"type": "Vehicle"}]}]});
1857 4 : let (status, headers, b) = post_json(&st, "/ngsi-ld/v1/snapshots", &ok).await;
1858 4 : assert_eq!(status, StatusCode::CREATED, "{b}");
1859 4 : let loc = header(&headers, "Location").expect("Location").to_owned();
1860 4 : let (status, _, b) = send(
1861 4 : &st,
1862 4 : "PATCH",
1863 4 : &loc,
1864 4 : Some(("application/json", json!({"endpoint": uri}).to_string())),
1865 4 : &[],
1866 : )
1867 4 : .await;
1868 4 : assert_eq!(status, StatusCode::BAD_REQUEST, "{b}");
1869 4 : assert!(
1870 4 : !b.to_string().contains("pa55word"),
1871 4 : "the credential was echoed"
1872 4 : );
1873 4 : }
1874 :
1875 472 : async fn post_json(st: &AppState, path: &str, doc: &Value) -> (StatusCode, HeaderMap, Value) {
1876 472 : send(
1877 472 : st,
1878 472 : "POST",
1879 472 : path,
1880 472 : Some(("application/json", doc.to_string())),
1881 472 : &[],
1882 472 : )
1883 472 : .await
1884 472 : }
1885 :
1886 72 : fn header<'a>(h: &'a HeaderMap, name: &str) -> Option<&'a str> {
1887 72 : h.get(name).and_then(|v| v.to_str().ok())
1888 72 : }
1889 :
1890 : /// Create a snapshot over every Vehicle and wait for the background fill.
1891 24 : async fn snapshot_over_vehicles(st: &AppState) -> String {
1892 24 : let doc = json!({"type": "Snapshot",
1893 24 : "snapshotQueries": [{"type": "Query", "entities": [{"type": "Vehicle"}]}]});
1894 24 : let (status, headers, body) = post_json(st, "/ngsi-ld/v1/snapshots", &doc).await;
1895 24 : assert_eq!(status, StatusCode::CREATED, "{body}");
1896 24 : let loc = header(&headers, "Location").expect("Location").to_owned();
1897 24 : for _ in 0..200 {
1898 45 : let (_, _, b) = send(st, "GET", &loc, None, &[]).await;
1899 45 : if b["snapshotStatus"] != "preparing" {
1900 24 : return b["id"].as_str().expect("snapshot id").to_owned();
1901 21 : }
1902 21 : tokio::time::sleep(std::time::Duration::from_millis(20)).await;
1903 : }
1904 0 : panic!("snapshot never left preparing");
1905 24 : }
1906 :
1907 95 : async fn stored(st: &AppState, id: &str) -> Value {
1908 95 : st.store
1909 95 : .get(&TenantId::default(), Kind::Snapshot, id)
1910 95 : .await
1911 95 : .ok()
1912 95 : .flatten()
1913 95 : .expect("snapshot document")
1914 95 : }
1915 :
1916 : /// 6.3.22: the Snapshot API's own resources are never snapshot-scoped —
1917 : /// but that exemption is the /snapshots RESOURCE, not any path in which
1918 : /// the word occurs. An Entity attribute named "snapshots" stays scoped
1919 : /// to the frozen copy, so the live entity is untouched.
1920 : #[tokio::test(flavor = "multi_thread")]
1921 4 : async fn clause_6_3_22_scope_guard_matches_only_the_snapshot_resource() {
1922 4 : let st = state().await;
1923 4 : let ent = json!({"id": "urn:ngsi-ld:Vehicle:g1", "type": "Vehicle",
1924 4 : "snapshots": {"type": "Property", "value": 3}});
1925 4 : let (status, _, b) = post_json(&st, "/ngsi-ld/v1/entities", &ent).await;
1926 4 : assert_eq!(status, StatusCode::CREATED, "{b}");
1927 4 : let sid = snapshot_over_vehicles(&st).await;
1928 :
1929 4 : let (status, headers, b) = send(
1930 4 : &st,
1931 4 : "DELETE",
1932 4 : "/ngsi-ld/v1/entities/urn:ngsi-ld:Vehicle:g1/attrs/snapshots",
1933 4 : None,
1934 4 : &[("NGSILD-Snapshot", sid.as_str())],
1935 : )
1936 4 : .await;
1937 4 : assert_eq!(status, StatusCode::NO_CONTENT, "{b}");
1938 4 : assert_eq!(header(&headers, "NGSILD-Snapshot"), Some(sid.as_str()));
1939 :
1940 4 : let (status, _, live) = send(
1941 4 : &st,
1942 4 : "GET",
1943 4 : "/ngsi-ld/v1/entities/urn:ngsi-ld:Vehicle:g1",
1944 4 : None,
1945 4 : &[],
1946 : )
1947 4 : .await;
1948 4 : assert_eq!(status, StatusCode::OK, "{live}");
1949 4 : assert!(
1950 4 : live.get("snapshots").is_some(),
1951 : "a snapshot-scoped request deleted the LIVE attribute: {live}"
1952 : );
1953 4 : let (_, _, frozen) = send(
1954 4 : &st,
1955 4 : "GET",
1956 4 : "/ngsi-ld/v1/entities/urn:ngsi-ld:Vehicle:g1",
1957 4 : None,
1958 4 : &[("NGSILD-Snapshot", sid.as_str())],
1959 : )
1960 4 : .await;
1961 4 : assert!(
1962 4 : frozen.get("snapshots").is_none(),
1963 4 : "the frozen copy still carries the deleted attribute: {frozen}"
1964 4 : );
1965 4 : }
1966 :
1967 : /// 6.3.22: "If the HTTP header NGSILD-Snapshot is present in the HTTP
1968 : /// request, it shall also be present in HTTP response" — on the
1969 : /// unscoped Snapshot resources and on the error exits as well.
1970 : #[tokio::test(flavor = "multi_thread")]
1971 4 : async fn clause_6_3_22_header_is_echoed_on_every_exit() {
1972 4 : let st = state().await;
1973 4 : let sid = snapshot_over_vehicles(&st).await;
1974 4 : let (status, headers, b) = send(
1975 4 : &st,
1976 4 : "GET",
1977 4 : &format!("/ngsi-ld/v1/snapshots/{sid}"),
1978 4 : None,
1979 4 : &[("NGSILD-Snapshot", sid.as_str())],
1980 : )
1981 4 : .await;
1982 4 : assert_eq!(status, StatusCode::OK, "{b}");
1983 4 : assert_eq!(header(&headers, "NGSILD-Snapshot"), Some(sid.as_str()));
1984 :
1985 4 : let unknown = "urn:ngsi-ld:Snapshot:nope";
1986 4 : let (status, headers, b) = send(
1987 4 : &st,
1988 4 : "GET",
1989 4 : "/ngsi-ld/v1/entities?type=Vehicle",
1990 4 : None,
1991 4 : &[("NGSILD-Snapshot", unknown)],
1992 : )
1993 4 : .await;
1994 4 : assert_eq!(status, StatusCode::NOT_FOUND, "{b}");
1995 4 : assert_eq!(header(&headers, "NGSILD-Snapshot"), Some(unknown));
1996 4 : assert!(
1997 4 : !b.to_string().contains("snap-"),
1998 4 : "the synthetic tenant leaked to the client: {b}"
1999 4 : );
2000 4 : }
2001 :
2002 : /// 5.16.1.4: "If the NGSI-LD endpoint already knows about this Snapshot,
2003 : /// as there is an existing Snapshot whose id (URI) is equivalent, an
2004 : /// error of type AlreadyExists shall be raised" — and the rejected
2005 : /// create must not have replaced the existing snapshot.
2006 : #[tokio::test(flavor = "multi_thread")]
2007 4 : async fn clause_5_16_1_4_duplicate_id_is_already_exists() {
2008 4 : let st = state().await;
2009 4 : let mut doc = json!({"id": "urn:ngsi-ld:Snapshot:dup", "type": "Snapshot",
2010 4 : "snapshotPriority": 3,
2011 4 : "snapshotQueries": [{"type": "Query", "entities": [{"type": "Vehicle"}]}]});
2012 4 : let (status, _, b) = post_json(&st, "/ngsi-ld/v1/snapshots", &doc).await;
2013 4 : assert_eq!(status, StatusCode::CREATED, "{b}");
2014 4 : doc["snapshotPriority"] = json!(9);
2015 4 : let (status, _, b) = post_json(&st, "/ngsi-ld/v1/snapshots", &doc).await;
2016 4 : assert_eq!(status, StatusCode::CONFLICT, "{b}");
2017 4 : assert!(
2018 4 : b["type"]
2019 4 : .as_str()
2020 4 : .unwrap_or_default()
2021 4 : .ends_with("AlreadyExists"),
2022 : "{b}"
2023 : );
2024 4 : let (_, _, got) = send(
2025 4 : &st,
2026 4 : "GET",
2027 4 : "/ngsi-ld/v1/snapshots/urn:ngsi-ld:Snapshot:dup",
2028 4 : None,
2029 4 : &[],
2030 : )
2031 4 : .await;
2032 4 : assert_eq!(
2033 4 : got["snapshotPriority"], 3,
2034 : "the rejected create overwrote the existing snapshot: {got}"
2035 : );
2036 4 : assert!(
2037 4 : got.get("__tenant").is_none(),
2038 4 : "internal member served to the client: {got}"
2039 4 : );
2040 4 : }
2041 :
2042 : /// 6.3.5 (5.16.4.4 → 5.5.4): snapshot bodies obey the @context rules —
2043 : /// application/json with a body @context is BadRequestData,
2044 : /// application/ld+json without one is BadRequestData, any other media
2045 : /// type is a bare 415, and @context is never stored in the snapshot.
2046 : #[tokio::test(flavor = "multi_thread")]
2047 4 : async fn clause_6_3_5_snapshot_bodies_follow_the_context_rules() {
2048 4 : let st = state().await;
2049 4 : let core = "https://uri.etsi.org/ngsi-ld/v1/ngsi-ld-core-context-v1.9.jsonld";
2050 4 : let plain = json!({"type": "Snapshot",
2051 4 : "snapshotQueries": [{"type": "Query", "entities": [{"type": "Vehicle"}]}]});
2052 4 : let mut with_ctx = plain.clone();
2053 4 : with_ctx["@context"] = json!(core);
2054 :
2055 8 : for (ct, doc) in [
2056 4 : ("application/json", &with_ctx),
2057 4 : ("application/ld+json", &plain),
2058 4 : ] {
2059 8 : let (status, _, b) = send(
2060 8 : &st,
2061 8 : "POST",
2062 8 : "/ngsi-ld/v1/snapshots",
2063 8 : Some((ct, doc.to_string())),
2064 8 : &[],
2065 : )
2066 8 : .await;
2067 8 : assert_eq!(status, StatusCode::BAD_REQUEST, "{ct}: {b}");
2068 : }
2069 4 : let (status, _, _) = send(
2070 4 : &st,
2071 4 : "POST",
2072 4 : "/ngsi-ld/v1/snapshots",
2073 4 : Some(("text/plain", plain.to_string())),
2074 4 : &[],
2075 : )
2076 4 : .await;
2077 4 : assert_eq!(status, StatusCode::UNSUPPORTED_MEDIA_TYPE);
2078 :
2079 4 : let (status, headers, b) = send(
2080 4 : &st,
2081 4 : "POST",
2082 4 : "/ngsi-ld/v1/snapshots",
2083 4 : Some(("application/ld+json", with_ctx.to_string())),
2084 4 : &[],
2085 : )
2086 4 : .await;
2087 4 : assert_eq!(status, StatusCode::CREATED, "{b}");
2088 4 : let loc = header(&headers, "Location").expect("Location").to_owned();
2089 4 : let (_, _, got) = send(&st, "GET", &loc, None, &[]).await;
2090 4 : assert!(
2091 4 : got.get("@context").is_none(),
2092 : "@context stored as a snapshot member: {got}"
2093 : );
2094 4 : let patch = json!({"snapshotPriority": 7, "@context": core});
2095 4 : let (status, _, b) = send(
2096 4 : &st,
2097 4 : "PATCH",
2098 4 : &loc,
2099 4 : Some(("application/json", patch.to_string())),
2100 4 : &[],
2101 : )
2102 4 : .await;
2103 4 : assert_eq!(status, StatusCode::BAD_REQUEST, "{b}");
2104 4 : }
2105 :
2106 : /// 5.5.6: "When a query operation is producing so many results that can
2107 : /// potentially exhaust client or server resources … implementations
2108 : /// shall raise an error of type TooManyResults." A fill copies its
2109 : /// results, so the ceiling is enforced before anything is copied.
2110 : /// 5.5.6 and a temporal fill page larger than the broker buffers
2111 : /// for one request is TooManyResults, not held whole in memory.
2112 : #[tokio::test(flavor = "multi_thread")]
2113 4 : async fn clause_5_5_6_fill_stops_at_the_answer_byte_bound() {
2114 4 : let mut st = state().await;
2115 4 : let ent = json!({"id": "urn:ngsi-ld:Vehicle:big", "type": "Vehicle",
2116 4 : "speed": {"type": "Property", "value": 80}});
2117 4 : let (status, _, b) = post_json(&st, "/ngsi-ld/v1/entities", &ent).await;
2118 4 : assert_eq!(status, StatusCode::CREATED, "{b}");
2119 4 : st.max_answer_bytes = 16;
2120 4 : let doc = json!({"type": "Snapshot",
2121 4 : "snapshotTemporalQueries": [{"type": "Query", "entities": [{"type": "Vehicle"}],
2122 4 : "temporalQ": {"timerel": "after", "timeAt": "2000-01-01T00:00:00Z",
2123 4 : "timeproperty": "createdAt"}}]});
2124 4 : let (status, headers, body) = post_json(&st, "/ngsi-ld/v1/snapshots", &doc).await;
2125 4 : assert_eq!(status, StatusCode::CREATED, "{body}");
2126 4 : let loc = header(&headers, "Location").expect("Location").to_owned();
2127 4 : let mut ready = Value::Null;
2128 4 : for _ in 0..200 {
2129 8 : let (_, _, b) = send(&st, "GET", &loc, None, &[]).await;
2130 8 : if b["snapshotStatus"] != "preparing" {
2131 4 : ready = b;
2132 4 : break;
2133 4 : }
2134 4 : tokio::time::sleep(std::time::Duration::from_millis(20)).await;
2135 : }
2136 4 : assert_eq!(ready["snapshotStatus"], "failure", "{ready}");
2137 4 : let detail = ready["snapshotTemporalQueriesDetails"][0]["problemDetails"].clone();
2138 4 : assert!(
2139 4 : detail["type"]
2140 4 : .as_str()
2141 4 : .unwrap_or_default()
2142 4 : .ends_with("TooManyResults"),
2143 : "{ready}"
2144 : );
2145 4 : assert_eq!(detail["detail"], crate::bounds::ANSWER_TOO_LARGE, "{ready}");
2146 4 : }
2147 :
2148 : #[tokio::test(flavor = "multi_thread")]
2149 4 : async fn clause_5_5_6_fill_stops_at_the_result_ceiling() {
2150 4 : let mut st = state().await;
2151 4 : st.max_limit = 1;
2152 4 : st.default_limit = 1;
2153 404 : for i in 0..=fill_cap(&st) {
2154 404 : let ent = json!({"id": format!("urn:ngsi-ld:Vehicle:c{i}"), "type": "Vehicle"});
2155 404 : let (status, _, b) = post_json(&st, "/ngsi-ld/v1/entities", &ent).await;
2156 404 : assert_eq!(status, StatusCode::CREATED, "{b}");
2157 : }
2158 4 : let sid = snapshot_over_vehicles(&st).await;
2159 4 : let ready = stored(&st, &sid).await;
2160 4 : assert_eq!(ready["snapshotStatus"], "failure", "{ready}");
2161 4 : let detail = &ready["snapshotQueriesDetails"][0];
2162 4 : assert_eq!(detail["resultStatus"], "failure", "{ready}");
2163 4 : assert!(
2164 4 : detail["problemDetails"]["type"]
2165 4 : .as_str()
2166 4 : .unwrap_or_default()
2167 4 : .ends_with("TooManyResults"),
2168 : "{detail}"
2169 : );
2170 4 : let (status, _, list) = send(
2171 4 : &st,
2172 4 : "GET",
2173 4 : "/ngsi-ld/v1/entities?type=Vehicle&limit=1",
2174 4 : None,
2175 4 : &[("NGSILD-Snapshot", sid.as_str())],
2176 : )
2177 4 : .await;
2178 4 : assert_eq!(status, StatusCode::OK, "{list}");
2179 4 : assert_eq!(
2180 4 : list,
2181 4 : json!([]),
2182 4 : "the oversized result set was copied into the snapshot anyway"
2183 4 : );
2184 4 : }
2185 :
2186 : /// 5.16.1.4 + 5.5.15: the filled copy is reachable only through its
2187 : /// snapshot document, so a fill whose snapshot has been deleted (or
2188 : /// evicted) stops and frees what it already copied.
2189 : #[tokio::test(flavor = "multi_thread")]
2190 4 : async fn clause_5_16_1_4_fill_stops_and_reaps_when_its_snapshot_is_gone() {
2191 4 : let st = state().await;
2192 4 : let tenant = TenantId::default();
2193 4 : let orphan = TenantId::new_internal("snap-reap-test").expect("tenant");
2194 4 : let _ = st
2195 4 : .store
2196 4 : .create(
2197 4 : &orphan,
2198 4 : Kind::Entity,
2199 4 : "urn:ngsi-ld:Vehicle:r1",
2200 4 : json!({"id": "urn:ngsi-ld:Vehicle:r1"}),
2201 4 : )
2202 4 : .await;
2203 4 : assert!(
2204 4 : fill_cancelled(&st, &tenant, "urn:ngsi-ld:Snapshot:gone", &orphan).await,
2205 : "a fill kept running after its snapshot was deleted"
2206 : );
2207 4 : assert!(
2208 4 : st.store
2209 4 : .list(&orphan, Kind::Entity)
2210 4 : .await
2211 4 : .unwrap_or_default()
2212 4 : .is_empty(),
2213 : "the orphaned copy outlived its snapshot"
2214 : );
2215 :
2216 4 : let sid = snapshot_over_vehicles(&st).await;
2217 4 : let synth = synth_tenant(&stored(&st, &sid).await).expect("__tenant");
2218 4 : let _ = st
2219 4 : .store
2220 4 : .create(
2221 4 : &synth,
2222 4 : Kind::Entity,
2223 4 : "urn:ngsi-ld:Vehicle:r2",
2224 4 : json!({"id": "urn:ngsi-ld:Vehicle:r2"}),
2225 4 : )
2226 4 : .await;
2227 4 : assert!(
2228 4 : !fill_cancelled(&st, &tenant, &sid, &synth).await,
2229 : "a live snapshot must not cancel its own fill"
2230 : );
2231 4 : assert_eq!(
2232 4 : st.store
2233 4 : .list(&synth, Kind::Entity)
2234 4 : .await
2235 4 : .unwrap_or_default()
2236 4 : .len(),
2237 4 : 1,
2238 4 : "a live snapshot's copy was reaped"
2239 4 : );
2240 4 : }
2241 :
2242 : /// 5.16.5.4 + 5.5.15: deleting a snapshot frees the whole isolated copy,
2243 : /// including the subscriptions created through the 6.3.22 header — no
2244 : /// client can reach them afterwards, so nothing may keep firing.
2245 : #[tokio::test(flavor = "multi_thread")]
2246 4 : async fn clause_5_16_5_delete_frees_the_whole_snapshot_copy() {
2247 4 : let st = state().await;
2248 4 : let sid = snapshot_over_vehicles(&st).await;
2249 4 : let synth = synth_tenant(&stored(&st, &sid).await).expect("__tenant");
2250 4 : let sub = json!({"type": "Subscription", "entities": [{"type": "Vehicle"}],
2251 4 : "notification": {"endpoint": {"uri": "http://127.0.0.1:9/none"}}});
2252 4 : let (status, _, b) = send(
2253 4 : &st,
2254 4 : "POST",
2255 4 : "/ngsi-ld/v1/subscriptions",
2256 4 : Some(("application/json", sub.to_string())),
2257 4 : &[("NGSILD-Snapshot", sid.as_str())],
2258 : )
2259 4 : .await;
2260 4 : assert_eq!(status, StatusCode::CREATED, "{b}");
2261 4 : assert!(
2262 4 : !st.store
2263 4 : .list(&synth, Kind::Subscription)
2264 4 : .await
2265 4 : .unwrap_or_default()
2266 4 : .is_empty(),
2267 : "the scoped subscription did not land in the snapshot copy"
2268 : );
2269 :
2270 4 : let (status, _, _) = send(
2271 4 : &st,
2272 4 : "DELETE",
2273 4 : &format!("/ngsi-ld/v1/snapshots/{sid}"),
2274 4 : None,
2275 4 : &[],
2276 : )
2277 4 : .await;
2278 4 : assert_eq!(status, StatusCode::NO_CONTENT);
2279 4 : for _ in 0..100 {
2280 7 : let left = st
2281 7 : .store
2282 7 : .list(&synth, Kind::Subscription)
2283 7 : .await
2284 7 : .unwrap_or_default()
2285 7 : .len();
2286 7 : if left == 0 {
2287 4 : return;
2288 4 : }
2289 4 : tokio::time::sleep(std::time::Duration::from_millis(20)).await;
2290 4 : }
2291 4 : panic!("a snapshot-scoped subscription outlived its snapshot");
2292 4 : }
2293 :
2294 : /// 5.2.41 Table 5.2.41-1: the snapshot id "cannot be later modified in
2295 : /// update operations" — whatever JSON type the client sends it as.
2296 : #[test]
2297 4 : fn clause_5_16_4_4_id_cannot_be_modified() {
2298 4 : assert!(validate(&json!({"id": "urn:ngsi-ld:Snapshot:1"}), Mode::Update).is_err());
2299 4 : assert!(
2300 4 : validate(&json!({"id": 5}), Mode::Update).is_err(),
2301 : "a non-string id escaped the 5.2.41 restriction"
2302 : );
2303 4 : assert!(validate(&json!({"snapshotPriority": 7}), Mode::Update).is_ok());
2304 4 : }
2305 :
2306 : /// 5.16.1.4: expiresAt is set "taking into account the snapshotLifetime
2307 : /// requested, but applying the configured limit of the NGSI-LD system";
2308 : /// a malformed duration is BadRequestData, and a snapshot past its
2309 : /// expiresAt is gone together with its copy.
2310 : #[tokio::test(flavor = "multi_thread")]
2311 4 : async fn clause_5_16_1_4_lifetime_is_clamped_and_expiry_removes_the_snapshot() {
2312 4 : let mut o = Map::new();
2313 4 : o.insert("snapshotLifetime".into(), json!("P10Y"));
2314 4 : let exp = chrono::DateTime::parse_from_rfc3339(&expires_at(&o).expect("expiresAt"))
2315 4 : .expect("rfc3339")
2316 4 : .with_timezone(&chrono::Utc);
2317 4 : let limit = chrono::Utc::now() + chrono::Duration::seconds(MAX_LIFETIME_SECS);
2318 4 : assert!(exp <= limit, "the configured limit was not applied: {exp}");
2319 4 : assert!(
2320 4 : exp > chrono::Utc::now() + chrono::Duration::days(6),
2321 : "the suggested lifetime was ignored: {exp}"
2322 : );
2323 4 : assert!(validate(
2324 4 : &json!({"type": "Snapshot", "snapshotLifetime": "tomorrow",
2325 4 : "snapshotQueries": [{"type": "Query", "entities": [{"type": "Vehicle"}]}]}),
2326 4 : Mode::Create
2327 4 : )
2328 4 : .is_err());
2329 :
2330 4 : let st = state().await;
2331 4 : let sid = snapshot_over_vehicles(&st).await;
2332 4 : let synth = synth_tenant(&stored(&st, &sid).await).expect("__tenant");
2333 4 : let _ = st
2334 4 : .store
2335 4 : .create(
2336 4 : &synth,
2337 4 : Kind::Entity,
2338 4 : "urn:ngsi-ld:Vehicle:e1",
2339 4 : json!({"id": "urn:ngsi-ld:Vehicle:e1"}),
2340 4 : )
2341 4 : .await;
2342 4 : let _ = st
2343 4 : .store
2344 4 : .mutate(&TenantId::default(), Kind::Snapshot, &sid, |d| {
2345 4 : if let Some(o) = d.as_object_mut() {
2346 4 : o.insert("expiresAt".into(), json!("2000-01-01T00:00:00.000Z"));
2347 4 : }
2348 4 : Ok::<_, std::convert::Infallible>(())
2349 4 : })
2350 4 : .await;
2351 4 : let (status, _, b) = send(
2352 4 : &st,
2353 4 : "GET",
2354 4 : &format!("/ngsi-ld/v1/snapshots/{sid}"),
2355 4 : None,
2356 4 : &[],
2357 : )
2358 4 : .await;
2359 4 : assert_eq!(status, StatusCode::NOT_FOUND, "an expired snapshot: {b}");
2360 4 : assert!(
2361 4 : !st.store
2362 4 : .list(&synth, Kind::Entity)
2363 4 : .await
2364 4 : .unwrap_or_default()
2365 4 : .is_empty(),
2366 : "the GET freed the copy: reading must not write"
2367 : );
2368 : // The sweep is what frees it, and the purge it starts is background.
2369 4 : assert_eq!(
2370 4 : sweep_expired_snapshots(&st, &TenantId::default()).await,
2371 : 1,
2372 : "the sweep did not reap the expired snapshot"
2373 : );
2374 4 : for _ in 0..100 {
2375 7 : if st
2376 7 : .store
2377 7 : .list(&synth, Kind::Entity)
2378 7 : .await
2379 7 : .unwrap_or_default()
2380 7 : .is_empty()
2381 4 : {
2382 4 : return;
2383 4 : }
2384 4 : tokio::time::sleep(std::time::Duration::from_millis(20)).await;
2385 4 : }
2386 4 : panic!("the copy of an expired snapshot was not freed");
2387 4 : }
2388 :
2389 : const DEF: &str = "https://uri.etsi.org/ngsi-ld/default-context/";
2390 :
2391 : /// A Snapshot document held directly, with its synthetic tenant.
2392 20 : async fn seeded_snapshot(st: &AppState, id: &str, status: &str) -> TenantId {
2393 20 : let synth = format!("snap-default-{}", id.rsplit(':').next().unwrap_or("x"));
2394 20 : st.store
2395 20 : .create(
2396 20 : &TenantId::default(),
2397 20 : Kind::Snapshot,
2398 20 : id,
2399 20 : json!({"id": id, "type": "Snapshot", "__tenant": synth,
2400 20 : "snapshotStatus": status, "expiresAt": "2999-01-01T00:00:00Z"}),
2401 20 : )
2402 20 : .await
2403 20 : .expect("seed snapshot");
2404 20 : synth_tenant(&stored(st, id).await).expect("__tenant")
2405 20 : }
2406 :
2407 416 : async fn vehicle_in(st: &AppState, t: &TenantId, n: usize) {
2408 416 : let id = format!("urn:ngsi-ld:Vehicle:c{n}");
2409 416 : st.store
2410 416 : .create(
2411 416 : t,
2412 416 : Kind::Entity,
2413 416 : &id,
2414 416 : json!({"id": id, "type": format!("{DEF}Vehicle")}),
2415 416 : )
2416 416 : .await
2417 416 : .expect("seed entity");
2418 416 : }
2419 :
2420 : /// POST .../clone and the clone's id.
2421 16 : async fn clone_of(st: &AppState, sid: &str) -> String {
2422 16 : let (status, headers, body) = send(
2423 16 : st,
2424 16 : "POST",
2425 16 : &format!("/ngsi-ld/v1/snapshots/{sid}/clone"),
2426 16 : Some(("application/json", String::new())),
2427 16 : &[],
2428 : )
2429 16 : .await;
2430 16 : assert_eq!(status, StatusCode::CREATED, "{body}");
2431 16 : let loc = header(&headers, "Location").expect("Location");
2432 16 : loc.rsplit('/').next().expect("clone id").to_owned()
2433 16 : }
2434 :
2435 20 : async fn settled(st: &AppState, id: &str) -> Value {
2436 20 : for _ in 0..200 {
2437 35 : let meta = stored(st, id).await;
2438 35 : if meta["snapshotStatus"] != "preparing" {
2439 20 : return meta;
2440 15 : }
2441 15 : tokio::time::sleep(std::time::Duration::from_millis(20)).await;
2442 : }
2443 0 : panic!("snapshot {id} never left preparing");
2444 20 : }
2445 :
2446 : /// 5.16.2.4: the clone holds the source's whole Temporal
2447 : /// Evolution, never the one-instance stub its own Entity copy recorded.
2448 : #[tokio::test(flavor = "multi_thread")]
2449 4 : async fn clause_5_16_2_4_clone_copies_every_temporal_instance() {
2450 4 : let st = state().await;
2451 4 : let sid = "urn:ngsi-ld:Snapshot:tsrc";
2452 4 : let from = seeded_snapshot(&st, sid, "success").await;
2453 4 : let eid = "urn:ngsi-ld:Vehicle:c0";
2454 4 : let insts: Vec<Value> = (1..=5)
2455 20 : .map(|d| {
2456 20 : json!({"type": "Property", "value": d,
2457 20 : "observedAt": format!("2026-01-0{d}T00:00:00Z")})
2458 20 : })
2459 4 : .collect();
2460 4 : st.temporal
2461 4 : .upsert(
2462 4 : &from,
2463 4 : eid,
2464 4 : json!({"id": eid, "type": format!("{DEF}Vehicle"),
2465 4 : format!("{DEF}speed"): insts.clone()}),
2466 4 : )
2467 4 : .await
2468 4 : .expect("seed evolution");
2469 4 : vehicle_in(&st, &from, 0).await;
2470 4 : let cid = clone_of(&st, sid).await;
2471 4 : assert_eq!(settled(&st, &cid).await["snapshotStatus"], "success");
2472 4 : let to = synth_tenant(&stored(&st, &cid).await).expect("__tenant");
2473 4 : let doc = st
2474 4 : .temporal
2475 4 : .get(&to, eid)
2476 4 : .await
2477 4 : .expect("store")
2478 4 : .expect("evolution");
2479 4 : assert_eq!(
2480 4 : doc[format!("{DEF}speed")].as_array().map(Vec::len),
2481 : Some(5),
2482 : "the clone kept a stub: {doc}"
2483 : );
2484 : // the stub the 5.6.11 hook leaves when it records the Entity copy
2485 4 : let stubbed = TenantId::new_internal("snap-default-stubbed").expect("tenant");
2486 4 : st.temporal
2487 4 : .upsert(
2488 4 : &stubbed,
2489 4 : eid,
2490 4 : json!({"id": eid, format!("{DEF}speed"): [insts.first().cloned().expect("an instance")]}),
2491 4 : )
2492 4 : .await
2493 4 : .expect("seed stub");
2494 4 : copy_snapshot_data(&st, &from, &stubbed, fill_cap(&st))
2495 4 : .await
2496 4 : .expect("copy");
2497 4 : let doc = st
2498 4 : .temporal
2499 4 : .get(&stubbed, eid)
2500 4 : .await
2501 4 : .expect("store")
2502 4 : .expect("evolution");
2503 4 : assert_eq!(
2504 4 : doc[format!("{DEF}speed")].as_array().map(Vec::len),
2505 4 : Some(5),
2506 4 : "the copy kept the hook's stub: {doc}"
2507 4 : );
2508 4 : }
2509 :
2510 : /// 5.16.2.4: a clone of a source still preparing waits for
2511 : /// the source's fill and copies what it ends with; a source that fails
2512 : /// fails the clone.
2513 : #[tokio::test(flavor = "multi_thread")]
2514 4 : async fn clause_5_16_2_4_clone_waits_for_a_preparing_source() {
2515 4 : let st = state().await;
2516 4 : let sid = "urn:ngsi-ld:Snapshot:psrc";
2517 4 : let from = seeded_snapshot(&st, sid, "preparing").await;
2518 4 : vehicle_in(&st, &from, 1).await;
2519 4 : let cid = clone_of(&st, sid).await;
2520 4 : tokio::time::sleep(std::time::Duration::from_millis(300)).await;
2521 4 : assert_eq!(stored(&st, &cid).await["snapshotStatus"], "preparing");
2522 4 : vehicle_in(&st, &from, 2).await;
2523 4 : let _ = st
2524 4 : .store
2525 4 : .mutate(&TenantId::default(), Kind::Snapshot, sid, |d| {
2526 4 : d["snapshotStatus"] = json!("success");
2527 4 : Ok::<_, std::convert::Infallible>(())
2528 4 : })
2529 4 : .await;
2530 4 : assert_eq!(settled(&st, &cid).await["snapshotStatus"], "success");
2531 4 : let to = synth_tenant(&stored(&st, &cid).await).expect("__tenant");
2532 4 : assert_eq!(
2533 4 : st.store.list(&to, Kind::Entity).await.expect("list").len(),
2534 : 2
2535 : );
2536 :
2537 4 : let fid = "urn:ngsi-ld:Snapshot:fsrc";
2538 4 : seeded_snapshot(&st, fid, "failure").await;
2539 4 : let cid = clone_of(&st, fid).await;
2540 4 : assert_eq!(settled(&st, &cid).await["snapshotStatus"], "failure");
2541 4 : }
2542 :
2543 : /// 5.5.6: a clone stops at the fill ceiling of its own
2544 : /// broker, whatever the source was filled under.
2545 : #[tokio::test(flavor = "multi_thread")]
2546 4 : async fn clause_5_5_6_clone_stops_at_the_result_ceiling() {
2547 4 : let mut st = state().await;
2548 4 : st.max_limit = 1;
2549 4 : let sid = "urn:ngsi-ld:Snapshot:big";
2550 4 : let from = seeded_snapshot(&st, sid, "success").await;
2551 404 : for n in 0..=fill_cap(&st) {
2552 404 : vehicle_in(&st, &from, n).await;
2553 : }
2554 4 : let cid = clone_of(&st, sid).await;
2555 4 : assert_eq!(settled(&st, &cid).await["snapshotStatus"], "failure");
2556 4 : }
2557 :
2558 : /// 5.16.4.4: a fill finishing with a document it read
2559 : /// before an Update Snapshot keeps the update's expiresAt, and the
2560 : /// update keeps the fill's status.
2561 : #[tokio::test(flavor = "multi_thread")]
2562 4 : async fn clause_5_16_4_4_finish_and_update_keep_each_others_members() {
2563 4 : let st = state().await;
2564 4 : let t = TenantId::default();
2565 4 : let sid = "urn:ngsi-ld:Snapshot:race";
2566 4 : seeded_snapshot(&st, sid, "preparing").await;
2567 4 : let mut read_by_fill = snap_get(&st, &t, sid).await.expect("meta");
2568 4 : let (status, _, b) = send(
2569 4 : &st,
2570 4 : "PATCH",
2571 4 : &format!("/ngsi-ld/v1/snapshots/{sid}"),
2572 4 : Some((
2573 4 : "application/json",
2574 4 : json!({"snapshotLifetime": "PT1H"}).to_string(),
2575 4 : )),
2576 4 : &[],
2577 : )
2578 4 : .await;
2579 4 : assert!(status.is_success(), "{status} {b}");
2580 4 : let updated = stored(&st, sid).await["expiresAt"].clone();
2581 4 : assert_ne!(updated, json!("2999-01-01T00:00:00Z"));
2582 4 : read_by_fill["snapshotStatus"] = json!("success");
2583 4 : snap_put(&st, &t, &read_by_fill, FINISH_MEMBERS).await;
2584 4 : let meta = stored(&st, sid).await;
2585 4 : assert_eq!(
2586 4 : meta["expiresAt"], updated,
2587 : "the fill wrote back a stale expiresAt"
2588 : );
2589 4 : assert_eq!(meta["snapshotStatus"], "success");
2590 4 : }
2591 :
2592 : /// 5.16.1.4: a temporal snapshot query's attrs and window
2593 : /// bound what the snapshot holds.
2594 : #[tokio::test(flavor = "multi_thread")]
2595 4 : async fn clause_5_16_1_4_temporal_copy_keeps_the_query_attrs_and_window() {
2596 4 : let st = state().await;
2597 4 : let eid = "urn:ngsi-ld:Vehicle:tq";
2598 16 : let inst = |v: i64, d: u8| {
2599 16 : json!({"type": "Property", "value": v,
2600 16 : "observedAt": format!("2026-01-0{d}T00:00:00Z")})
2601 16 : };
2602 4 : let ent = json!({"id": eid, "type": "Vehicle",
2603 4 : "a": [inst(1, 1), inst(2, 3), inst(3, 5)], "b": [inst(9, 3)]});
2604 4 : let (status, _, b) = post_json(&st, "/ngsi-ld/v1/temporal/entities", &ent).await;
2605 4 : assert!(status.is_success(), "{status} {b}");
2606 4 : let doc = json!({"type": "Snapshot", "snapshotTemporalQueries": [{
2607 4 : "type": "Query", "entities": [{"type": "Vehicle"}], "attrs": ["a"],
2608 4 : "temporalQ": {"timerel": "between", "timeAt": "2026-01-02T00:00:00Z",
2609 4 : "endTimeAt": "2026-01-04T00:00:00Z"}}]});
2610 4 : let (status, headers, b) = post_json(&st, "/ngsi-ld/v1/snapshots", &doc).await;
2611 4 : assert_eq!(status, StatusCode::CREATED, "{b}");
2612 4 : let sid = header(&headers, "Location")
2613 4 : .and_then(|l| l.rsplit('/').next())
2614 4 : .expect("id")
2615 4 : .to_owned();
2616 4 : assert_eq!(settled(&st, &sid).await["snapshotStatus"], "success");
2617 4 : let synth = synth_tenant(&stored(&st, &sid).await).expect("__tenant");
2618 4 : let held = st
2619 4 : .temporal
2620 4 : .get(&synth, eid)
2621 4 : .await
2622 4 : .expect("store")
2623 4 : .expect("copy");
2624 4 : assert!(
2625 4 : held.get(format!("{DEF}b")).is_none(),
2626 : "b is outside attrs: {held}"
2627 : );
2628 4 : assert_eq!(
2629 4 : held[format!("{DEF}a")].as_array().map(Vec::len),
2630 4 : Some(1),
2631 4 : "instances outside the window: {held}"
2632 4 : );
2633 4 : }
2634 : }
|
