Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! The storage seam: two driver traits every backend implements, plus the
3 : //! backend-neutral types they speak (resource kinds, filter shapes, the
4 : //! store-mode enum). This crate names no backend — redb, sqlx and the
5 : //! browser's OPFS all live behind these traits in their own crates.
6 : //!
7 : //! Current-state and temporal storage are SEPARATE interfaces on purpose:
8 : //! a deployment may run postgres current-state with no temporal store at
9 : //! all, or memory current-state with a database-backed history. A driver
10 : //! that does not support an operation answers with an error (or a benign
11 : //! no-op for internal bookkeeping), never a panic — `NoTemporal` is the
12 : //! canonical instance.
13 : #![cfg_attr(not(test), warn(clippy::expect_used))]
14 : #![deny(missing_docs)]
15 :
16 : #[cfg(feature = "test-kit")]
17 : /// The shared driver contract every backend runs as a test. It asserts, so
18 : /// it panics on a store that breaks an invariant — that is the point.
19 : #[allow(clippy::expect_used)]
20 : pub mod contract;
21 : pub mod filter;
22 :
23 : use antares_model::{NgsiError, TenantId};
24 : use serde_json::Value;
25 :
26 : /// [`CurrentStateDriver::list_digests`] of documents read out: each one's
27 : /// `id` and a digest of its JSON text, a document without an id left out.
28 : #[must_use]
29 0 : pub fn digests(docs: Vec<Value>) -> Vec<(String, u64)> {
30 : use std::hash::{DefaultHasher, Hash, Hasher};
31 0 : docs.into_iter()
32 0 : .filter_map(|doc| {
33 0 : let id = doc.get("id")?.as_str()?.to_owned();
34 0 : let mut h = DefaultHasher::new();
35 0 : doc.to_string().hash(&mut h);
36 0 : Some((id, h.finish()))
37 0 : })
38 0 : .collect()
39 0 : }
40 :
41 : /// A stored document as the object it is. Every document a driver holds is a
42 : /// JSON object — `contract` asserts the shape on every backend — so a value
43 : /// that is not one means the driver returned something the contract forbids
44 : /// underneath a live request. That fails the one request; it never takes the
45 : /// process down, which is what an unwrap here would do.
46 22404 : pub fn stored_object(doc: &mut Value) -> Result<&mut serde_json::Map<String, Value>, NgsiError> {
47 22404 : doc.as_object_mut()
48 22404 : .ok_or_else(|| NgsiError::InternalError("stored document is not a JSON object".into()))
49 22404 : }
50 :
51 : /// What a [`ChangeHook`] call returns: the hook records temporal history
52 : /// through the temporal driver, which is asynchronous, so the driver that
53 : /// fires the hook awaits it before its own write returns.
54 : pub type HookFuture<'a> = std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send + 'a>>;
55 :
56 : /// Called with (tenant, before, after, version) on every entity write — the
57 : /// local-mode change feed: create ⇒ (None, Some), delete ⇒ (Some, None).
58 : /// Shared rather than owned: a driver clones it out of its lock before
59 : /// awaiting, since a lock guard cannot be held across an await. Two writers
60 : /// of one Entity may run their hooks in either order; `version` is the order
61 : /// they committed in ([`ChangeVersion`]).
62 : pub type ChangeHook = std::sync::Arc<
63 : dyn for<'a> Fn(&'a TenantId, Option<Value>, Option<Value>, ChangeVersion) -> HookFuture<'a>
64 : + Send
65 : + Sync,
66 : >;
67 :
68 : /// Where one entity change stands among the changes of its Entity, read
69 : /// under the write's row lock: the incarnation (the stored row's creation
70 : /// stamp, which tells a recreated Entity from the one deleted before it, as
71 : /// its version starts again at 1) and the version the write left, one past
72 : /// the deleted row's for a delete. Ordered: of two changes of one (Tenant,
73 : /// id), the greater committed last. A driver without row versions counts
74 : /// its commits instead, under the order it emits in; one that cannot tell
75 : /// the order passes the default, and equal versions are all admitted.
76 : #[derive(Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord)]
77 : pub struct ChangeVersion {
78 : /// The stored row's creation stamp as the driver spells it; empty on a
79 : /// driver that counts commits.
80 : pub incarnation: String,
81 : /// The row version the write left, or the driver's commit count.
82 : pub version: i64,
83 : }
84 :
85 : /// The newest [`ChangeVersion`] a change-hook consumer has seen per (Tenant,
86 : /// Entity id), so it drops a change a later commit's hook overtook instead
87 : /// of ending with the older image. Bounded: two generations of at most `cap`
88 : /// entries, the older forgotten when the newer fills, so memory stays at
89 : /// `2 * cap` keys whatever the number of Entities. A key forgotten admits
90 : /// its next change: hooks overtake each other within the few writes in
91 : /// flight at once, far fewer than `cap`.
92 : pub struct NewestSeen {
93 : cap: usize,
94 : gens: std::sync::Mutex<[std::collections::HashMap<(String, String), Seen>; 2]>,
95 : }
96 :
97 : /// The newest version admitted per incarnation of one key, newest
98 : /// incarnation last; at most [`INCARNATIONS`] of them.
99 : type Seen = Vec<(String, i64)>;
100 :
101 : /// Incarnations remembered per key: a delete and recreate burst longer than
102 : /// this forgets the oldest, whose next change is then admitted.
103 : const INCARNATIONS: usize = 4;
104 :
105 : impl NewestSeen {
106 : /// A map of at most `2 * cap` keys.
107 1292 : pub fn new(cap: usize) -> Self {
108 1292 : Self {
109 1292 : cap: cap.max(1),
110 1292 : gens: std::sync::Mutex::default(),
111 1292 : }
112 1292 : }
113 :
114 : /// `false` when a change of `(tenant, id)` newer than `v` was admitted;
115 : /// otherwise `v` becomes the newest and `true`. An equal version is
116 : /// admitted: a change seen twice is delivered twice, never lost.
117 : ///
118 : /// Versions are compared only within one incarnation, never across
119 : /// them. The incarnation is a wall-clock creation stamp: one recreated in
120 : /// the same millisecond, or by a pod whose clock runs behind, does not
121 : /// sort after the deleted one's. Nor does arrival order rank them: with
122 : /// A, C and B arriving in that order from hooks that overtook each
123 : /// other, any rule that drops a "superseded" incarnation ends up muting
124 : /// the live one. So a change is dropped only when its own incarnation
125 : /// already delivered a newer version (the delete leaves one past the
126 : /// row's, so a late update of a deleted incarnation whose delete was seen
127 : /// is dropped). A late change of another incarnation is admitted: at
128 : /// worst a stale image, never a lost one.
129 140958 : pub fn admit(&self, tenant: &str, id: &str, v: &ChangeVersion) -> bool {
130 : // a poisoned map is a panic between two plain map operations: the
131 : // map itself is whole
132 140958 : let mut gens = self
133 140958 : .gens
134 140958 : .lock()
135 140958 : .unwrap_or_else(std::sync::PoisonError::into_inner);
136 140958 : let key = (tenant.to_owned(), id.to_owned());
137 140958 : let mut seen: Seen = gens[0]
138 140958 : .get(&key)
139 140958 : .or_else(|| gens[1].get(&key))
140 140958 : .cloned()
141 140958 : .unwrap_or_default();
142 140958 : if let Some(i) = seen.iter().position(|(inc, _)| *inc == v.incarnation) {
143 18020 : if seen[i].1 > v.version {
144 16 : return false;
145 18004 : }
146 18004 : seen.remove(i);
147 122938 : } else if seen.len() >= INCARNATIONS {
148 120 : seen.remove(0);
149 122818 : }
150 140942 : seen.push((v.incarnation.clone(), v.version));
151 140942 : if gens[0].len() >= self.cap && !gens[0].contains_key(&key) {
152 12 : gens[1] = std::mem::take(&mut gens[0]);
153 140930 : }
154 140942 : gens[0].insert(key, seen);
155 140942 : true
156 140958 : }
157 : }
158 :
159 : /// Which resource family an operation touches.
160 : #[derive(Clone, Copy, Debug, PartialEq, Eq)]
161 : pub enum Kind {
162 : /// 5.2.4 Entity current-state documents.
163 : Entity,
164 : /// 5.2.12 Subscription documents.
165 : Subscription,
166 : /// 5.2.9 Context Source Registration documents.
167 : Registration,
168 : /// Context Source Registration Subscription documents (5.11).
169 : CSourceSubscription,
170 : /// Temporal Representation documents (5.2.5).
171 : Temporal,
172 : /// 5.16 Snapshot status documents (+ the internal synth-tenant index).
173 : Snapshot,
174 : /// 5.14 EntityMap API documents.
175 : EntityMap,
176 : /// 5.8.1.4 distributed-subscription mappings (remote ids per CSR).
177 : DistSub,
178 : /// Notifications a delivery policy gave up on (dead letters), kept
179 : /// under the subscription's tenant for replay.
180 : DeadLetter,
181 : /// Stored view documents under the reserved tenant `antares-views`,
182 : /// and the policy engine's rule documents (ADR-0020 amendment) under `antares-policy`.
183 : /// No NGSI-LD route reads or writes this kind.
184 : View,
185 : }
186 :
187 : /// The four store backends, decided ONCE at startup and threaded as a value —
188 : /// never re-derived from strings or runtime probes, so a section gated on the
189 : /// wrong mode is unrepresentable.
190 : #[derive(Clone, Copy, Debug, PartialEq, Eq, Default)]
191 : pub enum StoreMode {
192 : /// In-process maps; nothing survives a restart.
193 : #[default]
194 : Memory,
195 : /// Embedded single-file store on local disk.
196 : File,
197 : /// PostgreSQL with PostGIS.
198 : Postgres,
199 : /// PostgreSQL with PostGIS and TimescaleDB hypertables for history.
200 : Timescale,
201 : }
202 :
203 : impl StoreMode {
204 : /// Every backend this workspace knows, in the order a listing shows
205 : /// them. One source of truth: `FromStr`, the unknown-mode message and
206 : /// the broker's built-with shelf all read it, so a backend added to the
207 : /// enum cannot go missing from any of them.
208 : pub const ALL: [StoreMode; 4] = [
209 : StoreMode::Memory,
210 : StoreMode::File,
211 : StoreMode::Postgres,
212 : StoreMode::Timescale,
213 : ];
214 :
215 : /// The accepted mode names, `memory|file|postgres|timescale`.
216 4 : pub fn names() -> String {
217 4 : StoreMode::ALL
218 4 : .iter()
219 16 : .map(|m| m.as_str())
220 4 : .collect::<Vec<_>>()
221 4 : .join("|")
222 4 : }
223 :
224 : /// The mode name as accepted by `FromStr` (`memory|file|postgres|timescale`).
225 510 : pub fn as_str(self) -> &'static str {
226 510 : match self {
227 193 : StoreMode::Memory => "memory",
228 124 : StoreMode::File => "file",
229 108 : StoreMode::Postgres => "postgres",
230 85 : StoreMode::Timescale => "timescale",
231 : }
232 510 : }
233 : /// Shared-database modes — the only ones that can back multiple instances.
234 41 : pub fn is_pg(self) -> bool {
235 41 : matches!(self, StoreMode::Postgres | StoreMode::Timescale)
236 41 : }
237 : }
238 :
239 : impl std::str::FromStr for StoreMode {
240 : type Err = String;
241 123 : fn from_str(s: &str) -> Result<Self, Self::Err> {
242 123 : StoreMode::ALL
243 123 : .into_iter()
244 224 : .find(|m| m.as_str() == s)
245 123 : .ok_or_else(|| format!("unknown store mode {s} ({})", StoreMode::names()))
246 123 : }
247 : }
248 :
249 : impl std::fmt::Display for StoreMode {
250 6 : fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
251 6 : f.write_str(self.as_str())
252 6 : }
253 : }
254 :
255 : /// What one tenant holds, per current-state document kind.
256 : #[derive(Debug, Clone, Default, PartialEq, Eq)]
257 : pub struct TenantStats {
258 : /// The tenant name.
259 : pub tenant: String,
260 : /// When the tenant row was created, if the backend records it.
261 : pub created_at: Option<String>,
262 : /// 5.2.4 entities.
263 : pub entities: u64,
264 : /// 5.2.12 subscriptions.
265 : pub subscriptions: u64,
266 : /// 5.2.9 context source registrations.
267 : pub registrations: u64,
268 : /// Context source registration subscriptions.
269 : pub csource_subscriptions: u64,
270 : /// 5.16 snapshots.
271 : pub snapshots: u64,
272 : /// Entity maps (5.14).
273 : pub entity_maps: u64,
274 : /// Distributed subscriptions this tenant placed at other brokers.
275 : pub dist_subs: u64,
276 : }
277 :
278 : /// How many bytes one tenant's Entities take in the store, measured
279 : /// there and never estimated: the stored row on Postgres, the serialized
280 : /// document in memory.
281 : #[derive(Debug, Clone, Default, PartialEq, Eq)]
282 : pub struct TenantBytes {
283 : /// Every Entity of the tenant.
284 : pub entities: u64,
285 : /// The same bytes by the Entity's first type, so the values sum to
286 : /// `entities`.
287 : pub by_type: std::collections::BTreeMap<String, u64>,
288 : }
289 :
290 : /// The read-modify-write closure as the object-safe seam carries it. The
291 : /// typed `Result<T, E>` travels through [`CurrentStateDriverExt::mutate`]'s
292 : /// side slot; the boxed closure only signals commit (`Ok`) vs reject
293 : /// (`Err`) to the driver.
294 : pub type MutateFn<'a> = Box<dyn FnOnce(&mut Value) -> Result<(), ()> + Send + 'a>;
295 : /// Per-id variant for batch mutation; the driver calls it once per PRESENT
296 : /// id, in input order — the ext trait's error slot depends on that order.
297 : pub type BatchMutateFn<'a> = Box<dyn FnMut(&str, &mut Value) -> Result<(), ()> + Send + 'a>;
298 : /// The delivery stamp expressed as a `mutate`. This is the rule itself —
299 : /// what `timesSent`, `lastNotification`, `lastSuccess` and `status` become
300 : /// after one attempt (5.2.14.2) — so a backend that reimplements
301 : /// [`CurrentStateDriver::record_delivery`] in its own query language is
302 : /// reimplementing THIS, and the two must agree.
303 4800 : pub async fn record_delivery_via_mutate(
304 4800 : d: &(impl CurrentStateDriver + ?Sized),
305 4800 : tenant: &TenantId,
306 4800 : kind: Kind,
307 4800 : id: &str,
308 4800 : now: &str,
309 4800 : ) -> Result<Option<Delivery>, NgsiError> {
310 4800 : let mut out: Option<Delivery> = None;
311 4800 : d.mutate::<(), ()>(tenant, kind, id, |doc| {
312 4780 : let mut prev_success = None;
313 4780 : if let Some(o) = doc.as_object_mut() {
314 4780 : o.remove("status");
315 4780 : }
316 4780 : if let Some(n) = doc
317 4780 : .as_object_mut()
318 4780 : .and_then(|o| o.get_mut("notification"))
319 4780 : .and_then(Value::as_object_mut)
320 4764 : {
321 4764 : let sent = n.get("timesSent").and_then(Value::as_i64).unwrap_or(0);
322 4764 : n.insert("timesSent".into(), serde_json::json!(sent + 1));
323 4764 : n.insert("lastNotification".into(), Value::String(now.to_owned()));
324 4764 : prev_success = n.insert("lastSuccess".into(), Value::String(now.to_owned()));
325 4764 : n.insert("status".into(), Value::String("ok".into()));
326 4764 : }
327 4780 : out = Some(Delivery {
328 4780 : doc: doc.clone(),
329 4780 : prev_success,
330 4780 : });
331 4780 : Ok(())
332 4780 : })
333 4800 : .await?;
334 4800 : Ok(out)
335 4800 : }
336 :
337 : /// The forward stamp expressed as a `mutate`. This is the Table 5.2.9-2
338 : /// rule itself — what `timesSent`, `timesFailed`, `lastSuccess`,
339 : /// `lastFailure` and `status` become after one distributed operation — so a
340 : /// backend that reimplements [`CurrentStateDriver::record_forward`] in its
341 : /// own query language is reimplementing THIS, and the two must agree.
342 : ///
343 : /// `timesSent` moves on EVERY attempt: the table defines it as the number of
344 : /// times the registration "triggered a distributed operation, including
345 : /// failed attempts". A member is written when it first has meaning, which is
346 : /// what cardinality 0..1 and "created on first successful operation" ask for:
347 : /// a registration that has only ever succeeded carries no `timesFailed` and
348 : /// no `lastFailure`, and one that has only ever failed carries no
349 : /// `lastSuccess`. `status` always names the LAST attempt.
350 1404 : pub async fn record_forward_via_mutate(
351 1404 : d: &(impl CurrentStateDriver + ?Sized),
352 1404 : tenant: &TenantId,
353 1404 : id: &str,
354 1404 : now: &str,
355 1404 : ok: bool,
356 1404 : ) -> Result<Option<Value>, NgsiError> {
357 1404 : let mut out: Option<Value> = None;
358 1404 : d.mutate::<(), ()>(tenant, Kind::Registration, id, |doc| {
359 1378 : if let Some(o) = doc.as_object_mut() {
360 1378 : let sent = o.get("timesSent").and_then(Value::as_i64).unwrap_or(0);
361 1378 : o.insert("timesSent".into(), serde_json::json!(sent + 1));
362 1378 : if ok {
363 1082 : o.insert("lastSuccess".into(), Value::String(now.to_owned()));
364 1082 : o.insert("status".into(), Value::String("ok".into()));
365 1082 : } else {
366 296 : let failed = o.get("timesFailed").and_then(Value::as_i64).unwrap_or(0);
367 296 : o.insert("timesFailed".into(), serde_json::json!(failed + 1));
368 296 : o.insert("lastFailure".into(), Value::String(now.to_owned()));
369 296 : o.insert("status".into(), Value::String("failed".into()));
370 296 : }
371 0 : }
372 1378 : out = Some(doc.clone());
373 1378 : Ok(())
374 1378 : })
375 1404 : .await?;
376 1404 : Ok(out)
377 1404 : }
378 :
379 : /// What one delivery attempt wrote: the stored subscription as it now
380 : /// stands (the mirror is fed from it) and the `lastSuccess` that was there
381 : /// before, which a failed attempt puts back.
382 : #[derive(Debug, Clone)]
383 : pub struct Delivery {
384 : /// The subscription as it now stands, the shape the mirror is fed.
385 : pub doc: Value,
386 : /// The `notification.lastSuccess` this attempt overwrote, absent when
387 : /// the subscription had never succeeded.
388 : pub prev_success: Option<Value>,
389 : }
390 :
391 : /// ADR-0021: the Tenant a stored `@context` row belongs to, or `None` when it
392 : /// belongs to none.
393 : ///
394 : /// 5.13.1 "Cached" is a copy of a public document the broker downloaded for
395 : /// whoever named its URL, so every Tenant reaches it. "Hosted" and
396 : /// "ImplicitlyCreated" are documents a client stored THROUGH a Tenant, and
397 : /// 5.5.7 makes their term mappings decide what that Tenant's payloads mean;
398 : /// the owner travels in the row. A row written before the member existed
399 : /// reads as the default Tenant's, and so does a row whose `kind` is missing
400 : /// altogether: no writer produces one, and an unrecognised row is safer read
401 : /// as somebody's than as everybody's.
402 41036 : pub fn context_row_owner(row: &Value) -> Option<&str> {
403 41036 : (row["kind"].as_str() != Some("Cached"))
404 41036 : .then(|| row["owner"].as_str().unwrap_or(TenantId::DEFAULT))
405 41036 : }
406 :
407 : /// Whether a call acting for `tenant` — `None` for no Tenant at all — reaches
408 : /// the row. The rule [`context_row_owner`] states, in the form the stores and
409 : /// the API both need.
410 40962 : pub fn context_row_visible(row: &Value, tenant: Option<&TenantId>) -> bool {
411 40962 : match context_row_owner(row) {
412 38464 : None => true,
413 2498 : Some(owner) => tenant.is_some_and(|t| t.as_str() == owner),
414 : }
415 40962 : }
416 :
417 : /// Current-state storage: everything except the temporal evolution.
418 : ///
419 : /// Contract carried over from the enum seam it replaces: every mutate is
420 : /// one transaction under the row lock, and a missing row is `None`, never
421 : /// an insert (a bookkeeping writeback racing a DELETE must not resurrect
422 : /// the row). Backends map their internal failures to
423 : /// `NgsiError::InternalError` with a GENERIC client-visible detail.
424 : #[async_trait::async_trait]
425 : pub trait CurrentStateDriver: Send + Sync {
426 : /// Readiness RIGHT NOW — a lost database flips /q/ready to 503.
427 : async fn ping(&self) -> Result<(), NgsiError>;
428 : /// (Queued writers, peak) of a single-writer commit section, if the
429 : /// backend has one.
430 0 : fn commit_queue(&self) -> Option<(usize, usize)> {
431 0 : None
432 0 : }
433 : /// Drain: finish in-flight work and disconnect cleanly.
434 : async fn close(&self);
435 : /// What this driver runs on, for `/q/health`: engine, server version,
436 : /// extensions — whatever an operator needs to tell two deployments of
437 : /// the same backend name apart. Read from state captured at startup,
438 : /// never by querying on the request: health is polled. An empty object
439 : /// (the default) means the driver has nothing to add to its name.
440 0 : fn version_info(&self) -> Value {
441 0 : Value::Object(serde_json::Map::new())
442 0 : }
443 : /// Installs the (tenant, before, after) hook called on every entity write.
444 : fn set_change_hook(&self, h: ChangeHook);
445 : /// Turn the same-transaction outbox producer on (bus=nats).
446 0 : fn set_outbox(&self, on: bool) {
447 0 : let _ = on;
448 0 : }
449 : /// Outbox drain: oldest-first page of pending rows `(seq, tenant, event)`.
450 0 : async fn outbox_peek(&self, limit: i64) -> Result<Vec<(i64, String, Value)>, NgsiError> {
451 : let _ = limit;
452 : Ok(Vec::new())
453 0 : }
454 : /// Outbox drain: delete EXACTLY the published rows.
455 0 : async fn outbox_ack(&self, seqs: &[i64]) -> Result<u64, NgsiError> {
456 0 : let _ = seqs;
457 : Ok(0)
458 0 : }
459 : /// Outbox drain: keep the rows of events the bus could not carry whole
460 : /// and take them out of the drain's page. The published message holds a
461 : /// claim-check reference; the row holds the bodies it stands for.
462 0 : async fn outbox_retain(&self, tenant: &TenantId, seqs: &[i64]) -> Result<u64, NgsiError> {
463 : let _ = (tenant, seqs);
464 : Ok(0)
465 0 : }
466 : /// The whole event behind a claim-check reference, or `None` once it has
467 : /// been reaped. Scoped to the tenant that wrote it.
468 0 : async fn outbox_event(&self, seq: i64, tenant: &TenantId) -> Result<Option<Value>, NgsiError> {
469 : let _ = (seq, tenant);
470 : Ok(None)
471 0 : }
472 :
473 : /// Insert a document; `false` if the id already exists (nothing written).
474 : async fn create(
475 : &self,
476 : tenant: &TenantId,
477 : kind: Kind,
478 : id: &str,
479 : doc: Value,
480 : ) -> Result<bool, NgsiError>;
481 : /// Batch create (entities only); created-flags in input order.
482 : async fn batch_create(
483 : &self,
484 : tenant: &TenantId,
485 : items: Vec<(String, Value)>,
486 : ) -> Result<Vec<bool>, NgsiError>;
487 : /// Batch delete (entities only); deleted-flags in input order, a
488 : /// duplicate id deletes once and reads absent the second time.
489 : async fn batch_delete(&self, tenant: &TenantId, ids: &[String])
490 : -> Result<Vec<bool>, NgsiError>;
491 : /// Batch upsert with REPLACE semantics (entities only); created-flags in
492 : /// input order.
493 : async fn batch_upsert(
494 : &self,
495 : tenant: &TenantId,
496 : items: Vec<(String, Value)>,
497 : ) -> Result<Vec<bool>, NgsiError>;
498 : /// Insert or replace a document. `true` means a document was ALREADY
499 : /// there and this call replaced it; `false` means this call created it.
500 : /// The polarity is the opposite of [`Self::batch_upsert`], which answers
501 : /// created-flags — the batch path needs them to split 201 from 204
502 : /// (5.6.8) while the single path ignores the value.
503 : async fn upsert(
504 : &self,
505 : tenant: &TenantId,
506 : kind: Kind,
507 : id: &str,
508 : doc: Value,
509 : ) -> Result<bool, NgsiError>;
510 : /// Read one document; `None` if absent.
511 : async fn get(
512 : &self,
513 : tenant: &TenantId,
514 : kind: Kind,
515 : id: &str,
516 : ) -> Result<Option<Value>, NgsiError>;
517 : /// Several Entities at once, keyed by id; an absent one has no entry. A
518 : /// backend that can read them together overrides this (Postgres: one
519 : /// transaction); the default reads them one by one.
520 : async fn get_entities(
521 : &self,
522 : tenant: &TenantId,
523 : ids: &[&str],
524 14 : ) -> Result<std::collections::HashMap<String, Value>, NgsiError> {
525 : let mut found = std::collections::HashMap::with_capacity(ids.len());
526 : for id in ids {
527 : if let Some(doc) = self.get(tenant, Kind::Entity, id).await? {
528 : found.insert((*id).to_owned(), doc);
529 : }
530 : }
531 : Ok(found)
532 14 : }
533 : /// Delete one document; `false` if it was absent.
534 : async fn delete(&self, tenant: &TenantId, kind: Kind, id: &str) -> Result<bool, NgsiError>;
535 : /// Delete one Entity, but only if `keep` accepts the stored document.
536 : /// `false` = absent OR refused; the caller cannot tell the two apart and
537 : /// does not need to, because 5.6.6.4 gives the same answer to both: an
538 : /// Entity the selector excludes "is not known" for this operation.
539 : ///
540 : /// The read and the delete happen under one lock. Required, deliberately:
541 : /// the obvious default is `get` then `delete`, and that pair is the race
542 : /// this method exists to remove — between the two the Entity can be
543 : /// deleted and recreated under the same id, and the delete then lands on
544 : /// a document the caller never inspected. A backend inheriting a default
545 : /// would keep the race with no compile error to say so.
546 : async fn delete_entity_if(
547 : &self,
548 : tenant: &TenantId,
549 : id: &str,
550 : keep: &(dyn for<'v> Fn(&'v Value) -> bool + Sync),
551 : ) -> Result<bool, NgsiError>;
552 : /// Every document of this kind in the tenant.
553 : ///
554 : /// A backend may refuse a tenant that holds too many to materialize
555 : /// (5.5.6 TooManyResults). That is right for a client query and wrong
556 : /// for a reader that must see ALL of them — use [`Self::list_page`].
557 : async fn list(&self, tenant: &TenantId, kind: Kind) -> Result<Vec<Value>, NgsiError>;
558 : /// Every document of this kind in the tenant as its id and a digest of
559 : /// its body, for a reader that holds the bodies already and fetches only
560 : /// the ones whose digest changed. Two equal bodies have one
561 : /// digest on one backend; nothing else is promised of it. Refused as
562 : /// [`Self::list`] is. A backend that digests without reading the
563 : /// documents out overrides this; the default reads them, which is
564 : /// correct and no cheaper.
565 : async fn list_digests(
566 : &self,
567 : tenant: &TenantId,
568 : kind: Kind,
569 0 : ) -> Result<Vec<(String, u64)>, NgsiError> {
570 : Ok(digests(self.list(tenant, kind).await?))
571 0 : }
572 : /// One id-ordered page of documents, for the internal readers that must
573 : /// see every one of them and so cannot be refused: ids strictly greater
574 : /// than `after`, at most `limit`. A short page means the end.
575 : ///
576 : /// Keyset, not offset: the caller walks a tenant that is being written
577 : /// to underneath it, where OFFSET skips and repeats rows. The peak cost
578 : /// is one page, not the whole tenant, which is what the row ceiling on
579 : /// `list` was protecting — so this carries no ceiling of its own, and a
580 : /// backend may not refuse it for volume.
581 : ///
582 : /// Required, deliberately: the obvious default is `list` sliced, and
583 : /// `list` is the read that may refuse. A backend inheriting that would
584 : /// silently reacquire the outage this method exists to prevent, with no
585 : /// compile error to say so.
586 : async fn list_page(
587 : &self,
588 : tenant: &TenantId,
589 : kind: Kind,
590 : after: Option<&str>,
591 : limit: usize,
592 : ) -> Result<Vec<Value>, NgsiError>;
593 : /// One id-ordered window of documents and the size of the whole set:
594 : /// elements `offset..offset + limit`, plus the count 6.3.10's `count`
595 : /// parameter reports. `limit` 0 is a legal request for the count alone.
596 : ///
597 : /// 5.5.9.1: "the query resolution mechanisms of the NGSI-LD System shall
598 : /// ensure that only up to a maximum of L NGSI-LD Elements are RETRIEVED
599 : /// and returned to the NGSI-LD client". Reading a whole tenant to serve
600 : /// one page of it is what that sentence rules out, and it is why this
601 : /// carries no row ceiling: the window bounds the result by construction,
602 : /// so the only thing a ceiling could refuse is a page the client is
603 : /// entitled to.
604 : ///
605 : /// Offset, not the keyset of [`Self::list_page`], because this serves
606 : /// 5.5.9.2's `limit`/`offset`, which lets a client "jump to a desired
607 : /// set of elements". A cursor cannot answer that; the two reads exist
608 : /// for different callers and neither replaces the other.
609 : ///
610 : /// Required for the same reason `list_page` is: the obvious default
611 : /// slices `list`, and `list` is the read that may refuse.
612 : async fn list_slice(
613 : &self,
614 : tenant: &TenantId,
615 : kind: Kind,
616 : offset: usize,
617 : limit: usize,
618 : ) -> Result<(Vec<Value>, usize), NgsiError>;
619 : /// Registrations that can match these ids/types (a backend may narrow;
620 : /// returning the full tenant list is always correct).
621 : async fn matching_registrations(
622 : &self,
623 : tenant: &TenantId,
624 : ids: Option<&[String]>,
625 : types: Option<&[String]>,
626 : ) -> Result<Vec<Value>, NgsiError>;
627 : /// Query Entities with the filter pushed down where the backend can
628 : /// take it; the caller re-checks unless the outcome says `decided`.
629 : async fn query_entities(
630 : &self,
631 : tenant: &TenantId,
632 : f: &filter::EntityFilter<'_>,
633 : ) -> Result<filter::QueryOutcome, NgsiError>;
634 : /// `query_entities` for a caller that re-decides every row and walks the
635 : /// match set in id-ordered pages: the backend may push `f.page` over the
636 : /// candidates it cannot decide (`paged` without `decided`), and never
637 : /// counts them. The default reads as `query_entities` does.
638 : async fn query_candidates(
639 : &self,
640 : tenant: &TenantId,
641 : f: &filter::EntityFilter<'_>,
642 0 : ) -> Result<filter::QueryOutcome, NgsiError> {
643 : self.query_entities(tenant, f).await
644 0 : }
645 : /// The Entity types the Tenant's live Entities carry (5.7.5), sorted,
646 : /// read without the documents: `/types` names them, and folding every
647 : /// Entity for the names read a whole Tenant per call. `None`
648 : /// when the backend has no such read, and the caller folds instead.
649 0 : async fn entity_type_names(&self, tenant: &TenantId) -> Result<Option<Vec<String>>, NgsiError> {
650 0 : let _ = tenant;
651 : Ok(None)
652 0 : }
653 : /// The types the Tenant's live Entities of two or more types carry when
654 : /// they also carry one of `any_of`, sorted: a view's rule names
655 : /// its types, and a type the rule does not name is shown only on an
656 : /// Entity of both. An Entity of `any_of` alone adds nothing. `None` when
657 : /// the backend has no such read, and the caller probes every type.
658 : async fn entity_types_beside(
659 : &self,
660 : tenant: &TenantId,
661 : any_of: &[String],
662 14 : ) -> Result<Option<Vec<String>>, NgsiError> {
663 : let _ = (tenant, any_of);
664 : Ok(None)
665 14 : }
666 : /// Each distinct `lead + s` the Tenant's ids start with, `s` being the
667 : /// text after `lead` up to the first `sep` (an id without one stands
668 : /// for itself), sorted by bytes: with
669 : /// [`filter::id_pattern_segments`] they turn an id pattern into exact
670 : /// prefixes. `None` when the backend has no such read or more than
671 : /// `max` exist.
672 : async fn entity_id_segments(
673 : &self,
674 : tenant: &TenantId,
675 : lead: &str,
676 : sep: char,
677 : max: usize,
678 0 : ) -> Result<Option<Vec<String>>, NgsiError> {
679 : let _ = (tenant, lead, sep, max);
680 : Ok(None)
681 0 : }
682 : /// The (type, member name) pairs the Tenant's Entities carry, sorted:
683 : /// `/types` with details and `/attributes` probe these instead
684 : /// of folding every Entity. The list may name a pair no live Entity
685 : /// carries any more, never miss one a live Entity carries; member names
686 : /// include the Entity frame's. `None` when the backend keeps no such
687 : /// list, and the caller folds instead.
688 : async fn entity_attr_pairs(
689 : &self,
690 : tenant: &TenantId,
691 2 : ) -> Result<Option<Vec<(String, String)>>, NgsiError> {
692 2 : let _ = tenant;
693 : Ok(None)
694 2 : }
695 : /// Read-modify-write under the row lock; `None` = absent (never an
696 : /// insert), `Some(Err)` = the closure rejected, nothing committed.
697 : async fn mutate_boxed<'a>(
698 : &self,
699 : tenant: &TenantId,
700 : kind: Kind,
701 : id: &str,
702 : f: MutateFn<'a>,
703 : ) -> Result<Option<Result<(), ()>>, NgsiError>;
704 : /// Batch read-modify-write (entities only); results align with `ids`.
705 : async fn batch_mutate_boxed<'a>(
706 : &self,
707 : tenant: &TenantId,
708 : ids: &[String],
709 : f: BatchMutateFn<'a>,
710 : ) -> Result<Vec<Option<Result<(), ()>>>, NgsiError>;
711 : /// 5.2.14.2 delivery bookkeeping: stamp one delivery attempt on a
712 : /// subscription and hand back the stored document. `timesSent` moves by
713 : /// one, `lastNotification` and `lastSuccess` take `now`, and the previous
714 : /// `lastSuccess` comes back so a failed attempt can roll it back.
715 : /// `None` means the row is gone — the subscription was deleted between
716 : /// matching and delivery, and nothing may be sent (5.8.6).
717 : ///
718 : /// The default expresses it as a `mutate`, which is correct everywhere.
719 : /// A backend whose `mutate` locks the row across a network round trip
720 : /// should override it with one statement: at fan-out every delivery on
721 : /// one subscription contends for that row, so the lock hold time — not
722 : /// the statement count — is what serializes delivery.
723 : async fn record_delivery(
724 : &self,
725 : tenant: &TenantId,
726 : kind: Kind,
727 : id: &str,
728 : now: &str,
729 0 : ) -> Result<Option<Delivery>, NgsiError> {
730 : record_delivery_via_mutate(self, tenant, kind, id, now).await
731 0 : }
732 : /// [`Self::record_delivery`] for several attempts at once, one answer per
733 : /// attempt in input order. The ids are distinct: an attempt books exactly
734 : /// one notification and its pre-image is the value it overwrote. The
735 : /// default books them one by one; a backend that can should write them
736 : /// as one statement, which is what keeps a fan-out of thousands of
737 : /// in-process deliveries at one round trip per batch.
738 : async fn record_deliveries(
739 : &self,
740 : tenant: &TenantId,
741 : kind: Kind,
742 : attempts: &[(&str, &str)],
743 4 : ) -> Result<Vec<Option<Delivery>>, NgsiError> {
744 : let mut out = Vec::with_capacity(attempts.len());
745 : for (id, now) in attempts {
746 : out.push(self.record_delivery(tenant, kind, id, now).await?);
747 : }
748 : Ok(out)
749 4 : }
750 : /// Reap expired docs/instances (backends with their own maintenance job
751 : /// return 0).
752 0 : async fn sweep_expired(&self) -> usize {
753 : 0
754 0 : }
755 : /// Retention of one Entity type in one Tenant (`/q/retention`):
756 : /// delete at most `batch` Entities whose types include `type_iri`, that
757 : /// carry no `expiresAt` of their own (a writer's 4.22 stamp wins) and
758 : /// were last modified before `before`, a UTC DateTime. Returns how many
759 : /// were deleted, or with `dry_run` would be, deleting nothing; a driver
760 : /// without current state has none.
761 : async fn retain_current(
762 : &self,
763 : tenant: &TenantId,
764 : type_iri: &str,
765 : before: &str,
766 : batch: usize,
767 : dry_run: bool,
768 0 : ) -> Result<u64, NgsiError> {
769 : let _ = (tenant, type_iri, before, batch, dry_run);
770 : Ok(0)
771 0 : }
772 : /// The Entities, across every Tenant, that carry a [`HISTORY_RETENTION`]
773 : /// Property: at most `limit` of (tenant, id, the Property as
774 : /// stored). Cross-tenant maintenance, like the 4.22 reap; a driver
775 : /// without current state has none.
776 : async fn history_overrides(
777 : &self,
778 : limit: usize,
779 0 : ) -> Result<Vec<(String, String, Value)>, NgsiError> {
780 : let _ = limit;
781 : Ok(Vec::new())
782 0 : }
783 : /// Table 5.2.9-2 forward bookkeeping: stamp one distributed operation on
784 : /// a registration and hand back the stored document. `None` means the row
785 : /// is gone — the registration was deleted while its forward was in
786 : /// flight, and a bookkeeping writeback must never resurrect it.
787 : ///
788 : /// The default expresses it as a `mutate`, which is correct everywhere.
789 : // ponytail: one read-modify-write per forward, so a registration every
790 : // request fans out to is one contended row; a backend whose `mutate`
791 : // locks that row across a network round trip can collapse it into one
792 : // statement the way `record_delivery` does, once a profile asks.
793 : async fn record_forward(
794 : &self,
795 : tenant: &TenantId,
796 : id: &str,
797 : now: &str,
798 : ok: bool,
799 1404 : ) -> Result<Option<Value>, NgsiError> {
800 : record_forward_via_mutate(self, tenant, id, now, ok).await
801 1404 : }
802 : /// 5.5.10: the default Tenant implicitly exists; others once created.
803 : async fn tenant_exists(&self, tenant: &TenantId) -> Result<bool, NgsiError>;
804 : /// The iteration domain of the interval sweep and of every mirror
805 : /// hydration: a tenant holding a Subscription, a Context Source
806 : /// Registration Subscription OR a Registration SHALL appear. A superset
807 : /// is allowed — the callers list per tenant afterwards and an empty list
808 : /// costs nothing — so a backend that cannot narrow the set cheaply may
809 : /// return every tenant it knows. A SUBSET is a silent outage: a tenant
810 : /// missing here never fires a periodic notification and never reaches
811 : /// the mirror.
812 : ///
813 : /// Registrations are in the domain because one of the hydrations fills
814 : /// the REGISTRATION mirror, and the federation path reads that mirror
815 : /// alone whenever it is installed. A domain that stopped at
816 : /// subscription-holding tenants left a tenant with registrations and no
817 : /// subscription forwarding to no Context Source at all.
818 : async fn subscription_tenants(&self) -> Result<Vec<String>, NgsiError>;
819 : /// Every tenant the backend knows, sorted. The default Tenant is listed
820 : /// even when empty (5.5.10: it always exists). Names only: at the
821 : /// 10 000-tenant target (ADR-0001) an inventory carrying per-kind counts
822 : /// would cost a count per kind per tenant, so the counts are paid per
823 : /// lookup in `tenant_stats_one`.
824 0 : async fn tenant_ids(&self) -> Result<Vec<String>, NgsiError> {
825 : Err(NgsiError::OperationNotSupported("tenant inventory".into()))
826 0 : }
827 : /// What one tenant holds; `None` when it does not exist.
828 0 : async fn tenant_stats_one(&self, tenant: &TenantId) -> Result<Option<TenantStats>, NgsiError> {
829 0 : let _ = tenant;
830 : Err(NgsiError::OperationNotSupported("tenant inventory".into()))
831 0 : }
832 : /// The bytes one tenant's Entities take; a scan of the tenant, so the
833 : /// caller caches it.
834 0 : async fn tenant_bytes(&self, tenant: &TenantId) -> Result<TenantBytes, NgsiError> {
835 0 : let _ = tenant;
836 : Err(NgsiError::OperationNotSupported("tenant size".into()))
837 0 : }
838 : /// Remove every current-state document of one tenant; `false` when the
839 : /// tenant did not exist. The default Tenant is emptied but keeps
840 : /// existing.
841 0 : async fn purge_tenant(&self, tenant: &TenantId) -> Result<bool, NgsiError> {
842 0 : let _ = tenant;
843 : Err(NgsiError::OperationNotSupported("tenant purge".into()))
844 0 : }
845 : /// 5.13 @context documents. `tenant` is the Tenant the call acts for, and
846 : /// `None` means no Tenant at all — the boot warm and the `Cached`
847 : /// write-through, which touch documents downloaded from public URLs.
848 : ///
849 : /// A backend enforces ADR-0021 on it: a `Cached` row belongs to no Tenant
850 : /// and is reachable with any `tenant`, and every other kind is reachable
851 : /// only by the Tenant named in its `owner` member. A backend that ignores
852 : /// the parameter fails the driver contract.
853 : async fn context_put(
854 : &self,
855 : tenant: Option<&TenantId>,
856 : id: &str,
857 : doc: Value,
858 : ) -> Result<(), NgsiError>;
859 : /// Read a stored @context document by id; `None` if absent or owned by
860 : /// another Tenant.
861 : async fn context_get(
862 : &self,
863 : tenant: Option<&TenantId>,
864 : id: &str,
865 : ) -> Result<Option<Value>, NgsiError>;
866 : /// Read-modify-write of one stored @context document under its row lock:
867 : /// `None` = absent or owned by another Tenant (never an insert),
868 : /// `Some(Err)` = the closure rejected and nothing was committed. A usage
869 : /// bump or a refetch racing a delete must leave the row deleted, and two
870 : /// racing bumps must both count (5.13.3.5).
871 : async fn context_mutate<'a>(
872 : &self,
873 : tenant: Option<&TenantId>,
874 : id: &str,
875 : f: MutateFn<'a>,
876 : ) -> Result<Option<Result<(), ()>>, NgsiError>;
877 : /// Delete a stored @context document; `false` if it was absent or owned
878 : /// by another Tenant.
879 : async fn context_delete(&self, tenant: Option<&TenantId>, id: &str) -> Result<bool, NgsiError>;
880 : /// Every stored @context document.
881 : /// Every stored `@context` row WITHOUT its `body` member — the url,
882 : /// localId, kind, owner and usage counters, and not the document.
883 : ///
884 : /// There is deliberately no read that returns every row WITH its body.
885 : /// A body is accepted up to `MAX_CONTEXT_BYTES` (5 MiB) and only the
886 : /// `Cached` rows are capped in number, so one such read materializes
887 : /// gigabytes — on the boot path, where it decides whether the broker
888 : /// starts at all. The callers that need one body ask for it by id
889 : /// ([`Self::context_get`]); nothing needs them all at once.
890 : async fn context_list_meta(&self, tenant: Option<&TenantId>) -> Result<Vec<Value>, NgsiError>;
891 : }
892 :
893 : /// The typed-slot lock, taken and released inside one closure call. A
894 : /// poisoned slot is read through: the panic that poisoned it is already on
895 : /// its way out of the driver, and the value behind it is whatever the
896 : /// closure managed to write.
897 22118 : fn lock<T>(m: &std::sync::Mutex<T>) -> std::sync::MutexGuard<'_, T> {
898 22118 : m.lock().unwrap_or_else(|p| p.into_inner())
899 22118 : }
900 :
901 : /// The same, for taking the slot's value once the driver has answered.
902 22332 : fn into_inner<T>(m: std::sync::Mutex<T>) -> T {
903 22332 : m.into_inner().unwrap_or_else(|p| p.into_inner())
904 22332 : }
905 :
906 : /// Typed sugar over the boxed mutate seam — call sites keep their
907 : /// `Result<T, E>` closures; the value crosses the object boundary in a
908 : /// side slot.
909 : pub trait CurrentStateDriverExt {
910 : /// Typed read-modify-write: `None` = absent, `Some(Err(e))` = the closure
911 : /// rejected and nothing was committed.
912 : fn mutate<T: Send, E: Send>(
913 : &self,
914 : tenant: &TenantId,
915 : kind: Kind,
916 : id: &str,
917 : f: impl FnOnce(&mut Value) -> Result<T, E> + Send,
918 : ) -> impl std::future::Future<Output = Result<Option<Result<T, E>>, NgsiError>> + Send;
919 : /// Typed batch read-modify-write (entities only); results align with `ids`.
920 : fn batch_mutate<E: Send>(
921 : &self,
922 : tenant: &TenantId,
923 : ids: &[String],
924 : f: impl FnMut(&str, &mut Value) -> Result<(), E> + Send,
925 : ) -> impl std::future::Future<Output = Result<Vec<Option<Result<(), E>>>, NgsiError>> + Send;
926 : }
927 :
928 : impl<S: CurrentStateDriver + ?Sized> CurrentStateDriverExt for S {
929 24982 : async fn mutate<T: Send, E: Send>(
930 24982 : &self,
931 24982 : tenant: &TenantId,
932 24982 : kind: Kind,
933 24982 : id: &str,
934 24982 : f: impl FnOnce(&mut Value) -> Result<T, E> + Send,
935 24982 : ) -> Result<Option<Result<T, E>>, NgsiError> {
936 24982 : let slot = std::sync::Mutex::new(None);
937 24982 : let r = self
938 24982 : .mutate_boxed(
939 24982 : tenant,
940 24982 : kind,
941 24982 : id,
942 24982 : Box::new(|v| {
943 21364 : let r = f(v);
944 21364 : let flag = if r.is_ok() { Ok(()) } else { Err(()) };
945 21364 : *lock(&slot) = Some(r);
946 21364 : flag
947 21364 : }),
948 : )
949 24982 : .await?;
950 24982 : Ok(r.and_then(|_| into_inner(slot)))
951 24982 : }
952 :
953 236 : async fn batch_mutate<E: Send>(
954 236 : &self,
955 236 : tenant: &TenantId,
956 236 : ids: &[String],
957 236 : mut f: impl FnMut(&str, &mut Value) -> Result<(), E> + Send,
958 236 : ) -> Result<Vec<Option<Result<(), E>>>, NgsiError> {
959 : // Errors land in the queue in closure-call order, which the trait
960 : // contract fixes to input order over present ids.
961 236 : let errs = std::sync::Mutex::new(std::collections::VecDeque::new());
962 236 : let r = self
963 236 : .batch_mutate_boxed(
964 236 : tenant,
965 236 : ids,
966 8324 : Box::new(|id, v| match f(id, v) {
967 8306 : Ok(()) => Ok(()),
968 18 : Err(e) => {
969 18 : lock(&errs).push_back(e);
970 18 : Err(())
971 : }
972 8324 : }),
973 : )
974 236 : .await?;
975 236 : let mut errs = into_inner(errs);
976 236 : let mut out = Vec::with_capacity(r.len());
977 8382 : for slot in r {
978 8382 : out.push(match slot {
979 58 : None => None,
980 8306 : Some(Ok(())) => Some(Ok(())),
981 : // one queued error per rejected id is the trait contract; a
982 : // driver that reports more rejections than the closure raised
983 : // has broken it, and the batch fails rather than inventing an
984 : // error for the caller to read.
985 18 : Some(Err(())) => Some(Err(errs.pop_front().ok_or_else(|| {
986 0 : NgsiError::InternalError(
987 0 : "batch driver reported more rejections than were raised".into(),
988 0 : )
989 0 : })?)),
990 : });
991 : }
992 236 : Ok(out)
993 236 : }
994 : }
995 :
996 : /// What a temporal event records.
997 : #[derive(Clone, Copy, Debug, PartialEq, Eq)]
998 : pub enum TemporalOp {
999 : /// An Attribute the entity did not carry before (one event per instance).
1000 : AttrCreated,
1001 : /// A changed instance of an existing Attribute (4.5.6 append).
1002 : AttrModified,
1003 : /// 4.5.6: the Scope changed through the Core API — recorded as a scope
1004 : /// instance whose observedAt copies modifiedAt.
1005 : ScopeChanged,
1006 : }
1007 :
1008 : /// One change the write path hands to the temporal seam. Events are
1009 : /// produced per attribute INSTANCE (the gate chain and a columnar writer
1010 : /// both work per instance) and drained per request, in order.
1011 : #[derive(Clone, Debug)]
1012 : pub struct TemporalEvent {
1013 : /// What the event records.
1014 : pub op: TemporalOp,
1015 : /// Owning tenant.
1016 : pub tenant: TenantId,
1017 : /// Id of the entity whose history this instance belongs to.
1018 : pub entity_id: String,
1019 : /// The entity's meta shell (id, type, createdAt, modifiedAt, scope) as
1020 : /// it stood after the write — what `temporal_append` creates on first
1021 : /// touch.
1022 : pub shell: Value,
1023 : /// Expanded Attribute name, or `scope`.
1024 : pub attr: String,
1025 : /// The instance snapshot: value, datasetId, observedAt, instanceId…
1026 : pub instance: Value,
1027 : }
1028 :
1029 : /// [`TemporalDriver::event_list`]'s default: consecutive events of one Entity
1030 : /// become one `temporal_append`; a scope change goes through `mutate`, in
1031 : /// order. A driver that overrides `event_list` falls back to it.
1032 : ///
1033 : /// An Entity run the driver refuses (a memory/file history at its cap)
1034 : /// loses its own events and no other's: the drain goes on, and answers the
1035 : /// first refusal once the whole request is through.
1036 121600 : pub async fn drain_events<D: TemporalDriver + ?Sized>(
1037 121600 : d: &D,
1038 121600 : evs: &[TemporalEvent],
1039 121600 : ) -> Result<(), NgsiError> {
1040 121600 : let mut first = None;
1041 121600 : let mut i = 0;
1042 243462 : while i < evs.len() {
1043 121862 : let (tenant, id) = (&evs[i].tenant, evs[i].entity_id.as_str());
1044 121862 : let end = i + evs[i..]
1045 121862 : .iter()
1046 127304 : .take_while(|e| e.tenant == *tenant && e.entity_id == id)
1047 121862 : .count();
1048 121862 : if let Err(e) = drain_run(d, &evs[i..end]).await {
1049 4 : first.get_or_insert(e);
1050 121858 : }
1051 121862 : i = end;
1052 : }
1053 121600 : first.map_or(Ok(()), Err)
1054 121600 : }
1055 :
1056 : /// One Entity's consecutive events of [`drain_events`].
1057 121862 : async fn drain_run<D: TemporalDriver + ?Sized>(
1058 121862 : d: &D,
1059 121862 : run: &[TemporalEvent],
1060 121862 : ) -> Result<(), NgsiError> {
1061 121862 : let Some(head) = run.first() else {
1062 0 : return Ok(());
1063 : };
1064 121862 : let (tenant, id) = (&head.tenant, head.entity_id.as_str());
1065 121862 : let mut additions = serde_json::Map::new();
1066 121862 : let mut shell = &head.shell;
1067 127042 : for ev in run {
1068 127042 : shell = &ev.shell;
1069 127042 : if ev.op == TemporalOp::ScopeChanged {
1070 : // the additions before it first: they may create the doc the
1071 : // scope instance goes on, and history keeps production order
1072 54 : if !additions.is_empty() {
1073 2 : let pending = Value::Object(std::mem::take(&mut additions));
1074 2 : d.temporal_append(tenant, id, shell, &pending).await?;
1075 52 : }
1076 54 : if push_scope(d, tenant, id, &ev.instance).await?.is_none() {
1077 : // no temporal doc yet: create the shell, then record
1078 26 : let empty = Value::Object(serde_json::Map::new());
1079 26 : d.temporal_append(tenant, id, shell, &empty).await?;
1080 26 : push_scope(d, tenant, id, &ev.instance).await?;
1081 28 : }
1082 126988 : } else if let Some(arr) = additions
1083 126988 : .entry(ev.attr.clone())
1084 126988 : .or_insert_with(|| Value::Array(Vec::new()))
1085 126988 : .as_array_mut()
1086 126988 : {
1087 126988 : arr.push(ev.instance.clone());
1088 126988 : }
1089 : }
1090 121862 : if !additions.is_empty() {
1091 121826 : d.temporal_append(tenant, id, shell, &Value::Object(additions))
1092 121826 : .await?;
1093 36 : }
1094 121858 : Ok(())
1095 121862 : }
1096 :
1097 : /// Appends a scope instance to the held temporal doc; `None` when there is
1098 : /// no doc.
1099 80 : async fn push_scope<D: TemporalDriver + ?Sized>(
1100 80 : d: &D,
1101 80 : tenant: &TenantId,
1102 80 : id: &str,
1103 80 : inst: &Value,
1104 80 : ) -> Result<Option<Result<(), ()>>, NgsiError> {
1105 80 : d.mutate(tenant, id, |doc| {
1106 54 : let target = doc.as_object_mut().ok_or(())?;
1107 54 : match target.get_mut("scope").and_then(Value::as_array_mut) {
1108 32 : Some(arr) if arr.first().is_some_and(Value::is_object) => {
1109 2 : arr.push(inst.clone());
1110 2 : }
1111 52 : _ => {
1112 52 : target.insert("scope".into(), Value::Array(vec![inst.clone()]));
1113 52 : }
1114 : }
1115 54 : Ok::<(), ()>(())
1116 54 : })
1117 80 : .await
1118 80 : }
1119 :
1120 : /// A drained request folded into one append per run of consecutive events of
1121 : /// one Entity, as [`TemporalDriver::event_list`]'s default folds it: tenant,
1122 : /// id, the last shell, the additions by Attribute. `None` when an event
1123 : /// changes scope, which goes through `mutate` in order with the appends
1124 : /// (a driver that writes several appends at once takes this).
1125 123512 : pub fn fold_appends(evs: &[TemporalEvent]) -> Option<Vec<(&TenantId, &str, &Value, Value)>> {
1126 131586 : if evs.iter().any(|ev| ev.op == TemporalOp::ScopeChanged) {
1127 38 : return None;
1128 123474 : }
1129 123474 : let mut out: Vec<(&TenantId, &str, &Value, Value)> = Vec::new();
1130 131546 : for ev in evs {
1131 131546 : let same = out
1132 131546 : .last()
1133 131546 : .is_some_and(|(t, id, _, _)| **t == ev.tenant && *id == ev.entity_id);
1134 131546 : if !same {
1135 123786 : out.push((
1136 123786 : &ev.tenant,
1137 123786 : &ev.entity_id,
1138 123786 : &ev.shell,
1139 123786 : Value::Object(serde_json::Map::new()),
1140 123786 : ));
1141 123786 : }
1142 131546 : if let Some((_, _, shell, Value::Object(additions))) = out.last_mut() {
1143 131546 : *shell = &ev.shell;
1144 131546 : if let Some(arr) = additions
1145 131546 : .entry(ev.attr.clone())
1146 131546 : .or_insert_with(|| Value::Array(Vec::new()))
1147 131546 : .as_array_mut()
1148 131546 : {
1149 131546 : arr.push(ev.instance.clone());
1150 131546 : }
1151 0 : }
1152 : }
1153 123474 : Some(out)
1154 123512 : }
1155 :
1156 : /// The member an attribute instance carries its write's provenance in from
1157 : /// the API to the temporal store: `channel`,
1158 : /// `actor`, `operation`, `reason`, `requestId`. A store takes it out of the
1159 : /// instance before it keeps the instance, whoever put it there, so no read
1160 : /// ever answers it as data; a JSON-LD keyword is never an attribute member.
1161 : pub const PROVENANCE_KEY: &str = "@provenance";
1162 :
1163 : /// Takes [`PROVENANCE_KEY`] out of an instance, returning what it held.
1164 146685 : pub fn take_provenance(instance: &mut Value) -> Option<Value> {
1165 146685 : instance.as_object_mut()?.remove(PROVENANCE_KEY)
1166 146685 : }
1167 :
1168 : /// Takes [`PROVENANCE_KEY`] out of every instance of a temporal document:
1169 : /// what a store that keeps no provenance does before it keeps the document.
1170 454 : pub fn strip_provenance(doc: &mut Value) {
1171 2888 : for instances in doc.as_object_mut().into_iter().flat_map(|d| d.values_mut()) {
1172 2888 : for instance in instances.as_array_mut().into_iter().flatten() {
1173 2532 : take_provenance(instance);
1174 2532 : }
1175 : }
1176 454 : }
1177 :
1178 : /// The expanded name of the Property an Entity carries its own history
1179 : /// retention in: a duration or `off`, the most
1180 : /// specific level, beside `expiresAt`, which ends the Entity itself.
1181 : pub const HISTORY_RETENTION: &str = "https://uri.etsi.org/ngsi-ld/default-context/historyRetention";
1182 :
1183 : /// The longest history retention whose rows a store may keep unlogged
1184 : /// (docs/src/storage.md "Short history"): a crash loses at most the
1185 : /// trail the sweep drops within this span anyway.
1186 : pub const SHORT_HISTORY: std::time::Duration = std::time::Duration::from_secs(3600);
1187 :
1188 : /// Which writes record history of [`SHORT_HISTORY`] or less, from the
1189 : /// Tenants' retention rules: a store keeps those rows where a crash
1190 : /// may lose them and writes them no WAL.
1191 : #[derive(Debug, Default, Clone, PartialEq, Eq)]
1192 : pub struct ShortHistory {
1193 : /// (Tenant, expanded type) of every type rule that says something of
1194 : /// history: `true` when it keeps [`SHORT_HISTORY`] or less.
1195 : pub types: std::collections::HashMap<(String, String), bool>,
1196 : /// Tenants whose `*` rule keeps [`SHORT_HISTORY`] or less.
1197 : pub tenants: std::collections::HashSet<String>,
1198 : }
1199 :
1200 : impl ShortHistory {
1201 : /// Whether a write of an Entity of `types` in `tenant` keeps short
1202 : /// history: some type's rule is short and none is longer, or no type has
1203 : /// a rule and the Tenant's `*` rule is short. An Entity with its own
1204 : /// [`HISTORY_RETENTION`] (`own`) never is: its level wins over both.
1205 : #[must_use]
1206 5029 : pub fn is_short(&self, tenant: &str, types: &[String], own: bool) -> bool {
1207 5029 : if own || (self.types.is_empty() && self.tenants.is_empty()) {
1208 4988 : return false;
1209 41 : }
1210 41 : let mut ruled = types
1211 41 : .iter()
1212 43 : .filter_map(|t| self.types.get(&(tenant.to_owned(), t.clone())))
1213 41 : .peekable();
1214 41 : if ruled.peek().is_none() {
1215 6 : return self.tenants.contains(tenant);
1216 35 : }
1217 35 : ruled.all(|short| *short)
1218 5029 : }
1219 : }
1220 :
1221 : /// The Entities of one Tenant a history retention statement reaches.
1222 : #[derive(Debug, Clone, Copy)]
1223 : pub enum HistoryScope<'a> {
1224 : /// One Entity, by its own `historyRetention`.
1225 : Entity(&'a str),
1226 : /// The Entities with this type, but not `except_ids`, which carry their
1227 : /// own retention.
1228 : Type {
1229 : /// The expanded type.
1230 : type_iri: &'a str,
1231 : /// Entities left out.
1232 : except_ids: &'a [String],
1233 : },
1234 : /// The Entities with none of `except_types` and not in `except_ids`:
1235 : /// what a Tenant's or the platform's default reaches.
1236 : Rest {
1237 : /// Types with a rule of their own.
1238 : except_types: &'a [String],
1239 : /// Entities left out.
1240 : except_ids: &'a [String],
1241 : },
1242 : }
1243 :
1244 : /// Temporal storage: the entity history (Temporal Evolution) plus the raw
1245 : /// temporal documents the 5.6.13-5.6.16 edit paths operate on.
1246 : ///
1247 : /// Internal bookkeeping (snapshot copies, delete cascades) degrades to
1248 : /// benign no-ops on a driver without temporal support; the CLIENT-facing
1249 : /// operations answer `OperationNotSupported` (422 per CIM 009
1250 : /// Table 6.3.2-1) instead.
1251 : #[async_trait::async_trait]
1252 : pub trait TemporalDriver: Send + Sync {
1253 : /// The drain: one call carries a whole request's events, in production
1254 : /// order. The default folds consecutive events of one entity into a
1255 : /// single `temporal_append` (scope changes go through `mutate`, as
1256 : /// 4.5.6 shapes them); a bulk writer overrides this and sees the batch.
1257 10 : async fn event_list(&self, evs: &[TemporalEvent]) -> Result<(), NgsiError> {
1258 : drain_events(self, evs).await
1259 10 : }
1260 : /// `false` = this deployment records no history (`NoTemporal`); the
1261 : /// write path skips recording entirely.
1262 157176 : fn supported(&self) -> bool {
1263 157176 : true
1264 157176 : }
1265 : /// Stored attribute instances of one tenant (inventory); a driver
1266 : /// without history reports 0.
1267 0 : async fn attr_instance_count(&self, tenant: &TenantId) -> Result<u64, NgsiError> {
1268 0 : let _ = tenant;
1269 : Ok(0)
1270 0 : }
1271 : /// [`Self::attr_instance_count`] for a read that answers without it when
1272 : /// the count is slow (`GET /q/tenants/{tenant}`): a store that
1273 : /// scans for the count gives up after `bound` with an error. The default
1274 : /// counts without a bound.
1275 : async fn attr_instance_count_within(
1276 : &self,
1277 : tenant: &TenantId,
1278 : bound: std::time::Duration,
1279 0 : ) -> Result<u64, NgsiError> {
1280 : let _ = bound;
1281 : self.attr_instance_count(tenant).await
1282 0 : }
1283 : /// The bytes those attribute instances take in the store; a driver
1284 : /// without history reports 0.
1285 2 : async fn attr_instance_bytes(&self, tenant: &TenantId) -> Result<u64, NgsiError> {
1286 2 : let _ = tenant;
1287 : Ok(0)
1288 2 : }
1289 : /// Retention of history in one Tenant (`/q/retention`):
1290 : /// delete at most `batch` attribute instances of the Entities `scope`
1291 : /// names whose instant (`observedAt`, else `modifiedAt`, `deletedAt`,
1292 : /// `createdAt`) is before `before`, a UTC DateTime. Returns how many were
1293 : /// deleted, or with `dry_run` would be, deleting nothing; a driver
1294 : /// without history has none.
1295 : async fn retain_history(
1296 : &self,
1297 : tenant: &TenantId,
1298 : scope: HistoryScope<'_>,
1299 : before: &str,
1300 : batch: usize,
1301 : dry_run: bool,
1302 0 : ) -> Result<u64, NgsiError> {
1303 : let _ = (tenant, scope, before, batch, dry_run);
1304 : Ok(0)
1305 0 : }
1306 : /// Retention by partition: drop every stored slice of history
1307 : /// whose rows all belong to Tenants in `floors` and lie before that
1308 : /// Tenant's floor, a UTC DateTime no rule of the Tenant keeps rows below.
1309 : /// A slice holding a row of any other Tenant stays. Returns one line per
1310 : /// slice dropped; a driver that stores history in no droppable slices
1311 : /// drops none, and the row sweep does the work.
1312 : async fn drop_retained_history(
1313 : &self,
1314 : floors: &[(String, String)],
1315 0 : ) -> Result<Vec<String>, NgsiError> {
1316 0 : let _ = floors;
1317 : Ok(Vec::new())
1318 0 : }
1319 : /// The retention rules' [`ShortHistory`], replaced whole on each reload
1320 : /// of the rules; a driver without unlogged storage ignores it.
1321 0 : fn set_short_history(&self, short: ShortHistory) {
1322 0 : let _ = short;
1323 0 : }
1324 : /// Every version of the rows of one Entity's history, by expanded Attribute name: each instance with its
1325 : /// `version` and `provenance`, a superseded one with `supersededAt`,
1326 : /// oldest first per measurement. A driver that keeps no versions has none.
1327 : async fn versions(
1328 : &self,
1329 : tenant: &TenantId,
1330 : entity_id: &str,
1331 0 : ) -> Result<serde_json::Map<String, Value>, NgsiError> {
1332 : let _ = (tenant, entity_id);
1333 : Ok(serde_json::Map::new())
1334 0 : }
1335 : /// Erasure: every row of `attrs` of one
1336 : /// Entity, the current version and every superseded one, observed at or
1337 : /// before `before` (a UTC DateTime; `None`: every row); `attrs` `None`
1338 : /// takes every Attribute, and the Entity's history itself once it holds
1339 : /// no row. Answers how many rows went; a driver without history has none.
1340 : async fn erase(
1341 : &self,
1342 : tenant: &TenantId,
1343 : entity_id: &str,
1344 : attrs: Option<&[String]>,
1345 : before: Option<&str>,
1346 0 : ) -> Result<u64, NgsiError> {
1347 : let _ = (tenant, entity_id, attrs, before);
1348 : Ok(0)
1349 0 : }
1350 : /// Remove the whole history of one tenant; nothing to do without history.
1351 0 : async fn purge_tenant(&self, tenant: &TenantId) -> Result<(), NgsiError> {
1352 0 : let _ = tenant;
1353 : Ok(())
1354 0 : }
1355 : /// Readiness of the temporal backend; the default is always ready.
1356 0 : async fn ping(&self) -> Result<(), NgsiError> {
1357 : Ok(())
1358 0 : }
1359 : /// What this temporal driver runs on, for `/q/health`; the same
1360 : /// contract as [`CurrentStateDriver::version_info`].
1361 6 : fn version_info(&self) -> Value {
1362 6 : Value::Object(serde_json::Map::new())
1363 6 : }
1364 : /// Drain: finish in-flight work and disconnect cleanly. A temporal
1365 : /// driver configured as a backend of its own holds its own pool, and the
1366 : /// shutdown path closes it here. May be called more than once — when one
1367 : /// instance serves both seams it is closed through each of them — so an
1368 : /// implementation makes it idempotent. The default has nothing to close.
1369 6 : async fn close(&self) {}
1370 : /// Auto-recording fast path: append instances, creating the meta shell
1371 : /// on first touch — and only for an entity that still exists (5.6.6
1372 : /// deletes history; an append overlapping the delete must not recreate
1373 : /// it).
1374 : async fn temporal_append(
1375 : &self,
1376 : tenant: &TenantId,
1377 : id: &str,
1378 : shell: &Value,
1379 : additions: &Value,
1380 : ) -> Result<(), NgsiError>;
1381 : /// Would `query_temporal` page EXACTLY under this `q` and range — the
1382 : /// driver's own entity verdict equal to the evaluator's on every row,
1383 : /// so a page it cuts is the page the caller would have cut? A driver
1384 : /// that pages after an in-process evaluation, or not at all, answers
1385 : /// `false`, and the caller pages the merged result itself (5.7.4.4).
1386 0 : fn q_pushdown_exact(
1387 0 : &self,
1388 0 : _q: &antares_ql::QNode,
1389 0 : _range: Option<&filter::InstanceRange<'_>>,
1390 0 : _expand: &dyn Fn(&str) -> String,
1391 0 : ) -> bool {
1392 0 : false
1393 0 : }
1394 : /// Query Temporal Evolution (5.7.4) with pushdown where possible.
1395 : async fn query_temporal(
1396 : &self,
1397 : tenant: &TenantId,
1398 : f: &filter::TemporalFilter<'_>,
1399 : ) -> Result<filter::TemporalOutcome, NgsiError>;
1400 : /// Retrieve Temporal Evolution (5.7.3) with instance pruning.
1401 : async fn get_temporal(
1402 : &self,
1403 : tenant: &TenantId,
1404 : id: &str,
1405 : f: &filter::TemporalFilter<'_>,
1406 : ) -> Result<Option<Value>, NgsiError>;
1407 : /// Raw temporal document access (the 5.6.11-5.6.16 edit/delete paths
1408 : /// and internal copies).
1409 : async fn get(&self, tenant: &TenantId, id: &str) -> Result<Option<Value>, NgsiError>;
1410 : /// Insert a temporal document; `false` if the id already exists.
1411 : async fn create(&self, tenant: &TenantId, id: &str, doc: Value) -> Result<bool, NgsiError>;
1412 : /// Insert or replace a temporal document. `true` means it was ALREADY
1413 : /// there and this call replaced it; `false` means this call created it,
1414 : /// the same polarity as the current-state seam.
1415 : async fn upsert(&self, tenant: &TenantId, id: &str, doc: Value) -> Result<bool, NgsiError>;
1416 : /// Delete an entity's whole history; `false` if it had none.
1417 : async fn delete(&self, tenant: &TenantId, id: &str) -> Result<bool, NgsiError>;
1418 : /// Every temporal document in the tenant.
1419 : async fn list(&self, tenant: &TenantId) -> Result<Vec<Value>, NgsiError>;
1420 : /// Read-modify-write of one temporal document under its row lock;
1421 : /// `None` = absent (never an insert), `Some(Err)` = rejected, not committed.
1422 : async fn mutate_boxed<'a>(
1423 : &self,
1424 : tenant: &TenantId,
1425 : id: &str,
1426 : f: MutateFn<'a>,
1427 : ) -> Result<Option<Result<(), ()>>, NgsiError>;
1428 : }
1429 :
1430 : /// Typed mutate sugar for the temporal seam, same slot trick as
1431 : /// [`CurrentStateDriverExt`].
1432 : pub trait TemporalDriverExt {
1433 : /// Typed read-modify-write of one temporal document: `None` = absent,
1434 : /// `Some(Err(e))` = the closure rejected and nothing was committed.
1435 : fn mutate<T: Send, E: Send>(
1436 : &self,
1437 : tenant: &TenantId,
1438 : id: &str,
1439 : f: impl FnOnce(&mut Value) -> Result<T, E> + Send,
1440 : ) -> impl std::future::Future<Output = Result<Option<Result<T, E>>, NgsiError>> + Send;
1441 : }
1442 :
1443 : impl<S: TemporalDriver + ?Sized> TemporalDriverExt for S {
1444 862 : async fn mutate<T: Send, E: Send>(
1445 862 : &self,
1446 862 : tenant: &TenantId,
1447 862 : id: &str,
1448 862 : f: impl FnOnce(&mut Value) -> Result<T, E> + Send,
1449 862 : ) -> Result<Option<Result<T, E>>, NgsiError> {
1450 862 : let slot = std::sync::Mutex::new(None);
1451 862 : let r = self
1452 862 : .mutate_boxed(
1453 862 : tenant,
1454 862 : id,
1455 862 : Box::new(|v| {
1456 736 : let r = f(v);
1457 736 : let flag = if r.is_ok() { Ok(()) } else { Err(()) };
1458 736 : *lock(&slot) = Some(r);
1459 736 : flag
1460 736 : }),
1461 : )
1462 862 : .await?;
1463 854 : Ok(r.and_then(|_| into_inner(slot)))
1464 862 : }
1465 : }
1466 :
1467 : /// The no-history driver: temporal OFF as a driver choice. Client-facing
1468 : /// reads answer `OperationNotSupported`; the recorder and the internal
1469 : /// bookkeeping paths degrade to no-ops.
1470 : pub struct NoTemporal;
1471 :
1472 6 : fn unsupported() -> NgsiError {
1473 6 : NgsiError::OperationNotSupported("no temporal store is configured".into())
1474 6 : }
1475 :
1476 : #[async_trait::async_trait]
1477 : impl TemporalDriver for NoTemporal {
1478 12 : fn supported(&self) -> bool {
1479 12 : false
1480 12 : }
1481 : async fn temporal_append(
1482 : &self,
1483 : _tenant: &TenantId,
1484 : _id: &str,
1485 : _shell: &Value,
1486 : _additions: &Value,
1487 2 : ) -> Result<(), NgsiError> {
1488 : Ok(())
1489 2 : }
1490 : async fn query_temporal(
1491 : &self,
1492 : _tenant: &TenantId,
1493 : _f: &filter::TemporalFilter<'_>,
1494 4 : ) -> Result<filter::TemporalOutcome, NgsiError> {
1495 : Err(unsupported())
1496 4 : }
1497 : async fn get_temporal(
1498 : &self,
1499 : _tenant: &TenantId,
1500 : _id: &str,
1501 : _f: &filter::TemporalFilter<'_>,
1502 2 : ) -> Result<Option<Value>, NgsiError> {
1503 : Err(unsupported())
1504 2 : }
1505 2 : async fn get(&self, _tenant: &TenantId, _id: &str) -> Result<Option<Value>, NgsiError> {
1506 : Ok(None)
1507 2 : }
1508 0 : async fn create(&self, _tenant: &TenantId, _id: &str, _doc: Value) -> Result<bool, NgsiError> {
1509 : Ok(false)
1510 0 : }
1511 0 : async fn upsert(&self, _tenant: &TenantId, _id: &str, _doc: Value) -> Result<bool, NgsiError> {
1512 : Ok(false)
1513 0 : }
1514 2 : async fn delete(&self, _tenant: &TenantId, _id: &str) -> Result<bool, NgsiError> {
1515 : Ok(false)
1516 2 : }
1517 2 : async fn list(&self, _tenant: &TenantId) -> Result<Vec<Value>, NgsiError> {
1518 : Ok(Vec::new())
1519 2 : }
1520 : async fn mutate_boxed<'a>(
1521 : &self,
1522 : _tenant: &TenantId,
1523 : _id: &str,
1524 : _f: MutateFn<'a>,
1525 2 : ) -> Result<Option<Result<(), ()>>, NgsiError> {
1526 : Ok(None)
1527 2 : }
1528 : }
1529 :
1530 : #[cfg(test)]
1531 : mod tests {
1532 : use super::*;
1533 :
1534 : /// A recreate whose creation stamp sorts at or below the deleted
1535 : /// incarnation's (one millisecond, a clock behind) is still admitted,
1536 : /// and so are its updates; within one incarnation an overtaken change
1537 : /// is still dropped.
1538 : #[test]
1539 2 : fn a_recreated_incarnation_is_admitted_whatever_its_stamp() {
1540 2 : let seen = NewestSeen::new(16);
1541 2 : let v = |incarnation: &str, version| ChangeVersion {
1542 12 : incarnation: incarnation.to_owned(),
1543 12 : version,
1544 12 : };
1545 2 : let old = "2026-01-01 00:00:00.500+00";
1546 2 : assert!(seen.admit("t", "urn:e", &v(old, 3)));
1547 : // the delete leaves one past the deleted row's version
1548 2 : assert!(seen.admit("t", "urn:e", &v(old, 4)));
1549 2 : let recreated = "2026-01-01 00:00:00.400+00";
1550 2 : assert!(
1551 2 : seen.admit("t", "urn:e", &v(recreated, 1)),
1552 : "the recreate was dropped"
1553 : );
1554 2 : assert!(
1555 2 : seen.admit("t", "urn:e", &v(recreated, 2)),
1556 : "its update was dropped"
1557 : );
1558 2 : assert!(
1559 2 : !seen.admit("t", "urn:e", &v(recreated, 1)),
1560 : "an overtaken change of one incarnation is still dropped"
1561 : );
1562 2 : assert!(
1563 2 : !seen.admit("t", "urn:e", &v(old, 3)),
1564 : "a late change of the superseded incarnation is still dropped"
1565 : );
1566 2 : }
1567 :
1568 : /// SECV-01: incarnations arriving out of order (A, C, then a late B)
1569 : /// never mute the live one: C's next update still goes out.
1570 : #[test]
1571 2 : fn a_late_third_incarnation_never_mutes_the_live_one() {
1572 2 : let v = |inc: &str, n: i64| ChangeVersion {
1573 24 : incarnation: inc.into(),
1574 24 : version: n,
1575 24 : };
1576 2 : let seen = NewestSeen::new(16);
1577 2 : assert!(seen.admit("t", "e", &v("A", 1)));
1578 2 : assert!(seen.admit("t", "e", &v("A", 2)), "A's delete");
1579 2 : assert!(seen.admit("t", "e", &v("C", 1)));
1580 2 : assert!(seen.admit("t", "e", &v("B", 1)), "B's change arrives late");
1581 2 : assert!(
1582 2 : seen.admit("t", "e", &v("C", 2)),
1583 : "the live Entity was muted"
1584 : );
1585 2 : assert!(
1586 2 : !seen.admit("t", "e", &v("C", 1)),
1587 : "C's own order still holds"
1588 : );
1589 2 : assert!(
1590 2 : !seen.admit("t", "e", &v("A", 1)),
1591 : "A's late update after its delete"
1592 : );
1593 8 : for i in 0..INCARNATIONS {
1594 8 : assert!(seen.admit("t", "e", &v(&format!("n{i}"), 1)));
1595 : }
1596 2 : assert!(
1597 2 : seen.admit("t", "e", &v("C", 1)),
1598 : "forgotten past the bound: admitted"
1599 : );
1600 2 : }
1601 :
1602 : /// S3-18: a change an overtaken hook delivers late is dropped, a newer
1603 : /// one, an equal one and a recreated Entity's first are admitted, and the
1604 : /// map forgets past two generations.
1605 : #[test]
1606 2 : fn newest_seen_drops_what_a_later_commit_overtook() {
1607 2 : let v = |inc: &str, n: i64| ChangeVersion {
1608 22 : incarnation: inc.into(),
1609 22 : version: n,
1610 22 : };
1611 2 : let seen = NewestSeen::new(2);
1612 2 : assert!(seen.admit("t", "e", &v("a", 2)));
1613 2 : assert!(
1614 2 : !seen.admit("t", "e", &v("a", 1)),
1615 : "the older image arrives last"
1616 : );
1617 2 : assert!(
1618 2 : seen.admit("t", "e", &v("a", 2)),
1619 : "seen twice, delivered twice"
1620 : );
1621 2 : assert!(seen.admit("u", "e", &v("a", 1)), "another Tenant's Entity");
1622 2 : assert!(seen.admit("t", "e", &v("b", 1)), "recreated after a delete");
1623 2 : assert!(
1624 2 : !seen.admit("t", "e", &v("a", 1)),
1625 : "a late change of the deleted incarnation older than one it delivered"
1626 : );
1627 8 : for i in 0..4 {
1628 8 : assert!(seen.admit("t", &format!("x{i}"), &v("a", 1)));
1629 : }
1630 2 : let held = seen
1631 2 : .gens
1632 2 : .lock()
1633 2 : .expect("map")
1634 2 : .iter()
1635 4 : .map(|g| g.len())
1636 2 : .sum::<usize>();
1637 2 : assert!(held <= 4, "two generations of two: {held}");
1638 2 : assert!(seen.admit("t", "e", &v("a", 1)), "forgotten past the bound");
1639 2 : }
1640 :
1641 : /// a short type rule wins over the Tenant's level, a longer type
1642 : /// rule among the types or the Entity's own retention keeps it logged.
1643 : #[test]
1644 2 : fn short_history_takes_the_most_specific_level() {
1645 2 : let mut s = ShortHistory::default();
1646 14 : let ty = |t: &str| vec![t.to_owned()];
1647 2 : assert!(!s.is_short("city", &ty("Bus"), false), "no rule, not short");
1648 2 : s.types.insert(("city".into(), "Bus".into()), true);
1649 2 : s.types.insert(("city".into(), "Stop".into()), false);
1650 2 : assert!(s.is_short("city", &ty("Bus"), false));
1651 2 : assert!(!s.is_short("city", &ty("Bus"), true), "own retention wins");
1652 2 : assert!(!s.is_short("town", &ty("Bus"), false), "another Tenant");
1653 2 : let both = vec!["Bus".to_owned(), "Stop".to_owned()];
1654 2 : assert!(!s.is_short("city", &both, false), "a longer type keeps it");
1655 2 : assert!(!s.is_short("city", &ty("Tram"), false), "no `*` rule");
1656 2 : s.tenants.insert("city".into());
1657 2 : assert!(s.is_short("city", &ty("Tram"), false), "the `*` rule");
1658 2 : assert!(!s.is_short("city", &ty("Stop"), false), "a type rule wins");
1659 2 : }
1660 :
1661 : /// The mode list is one source of truth: every variant round-trips
1662 : /// through its name, and the message for an unknown name lists all of
1663 : /// them — a backend added to the enum cannot be missing from either.
1664 : #[test]
1665 2 : fn every_store_mode_round_trips_and_the_error_lists_them_all() {
1666 8 : for m in StoreMode::ALL {
1667 8 : let back = m
1668 8 : .as_str()
1669 8 : .parse::<StoreMode>()
1670 8 : .unwrap_or_else(|e| panic!("{m} must parse back: {e}"));
1671 8 : assert_eq!(back, m);
1672 : }
1673 2 : let err = "mongo".parse::<StoreMode>().expect_err("unknown mode");
1674 2 : assert!(err.contains("mongo"), "{err}");
1675 8 : for m in StoreMode::ALL {
1676 8 : assert!(err.contains(m.as_str()), "the message must name {m}: {err}");
1677 : }
1678 2 : }
1679 :
1680 : /// A minimal driver proving the boxed seam round-trips typed results:
1681 : /// present row → the closure's T and E cross intact; absent → None.
1682 : struct OneDoc(std::sync::Mutex<Option<Value>>);
1683 : #[async_trait::async_trait]
1684 : impl CurrentStateDriver for OneDoc {
1685 0 : async fn ping(&self) -> Result<(), NgsiError> {
1686 : Ok(())
1687 0 : }
1688 0 : async fn close(&self) {}
1689 0 : fn set_change_hook(&self, _h: ChangeHook) {}
1690 : async fn create(
1691 : &self,
1692 : _t: &TenantId,
1693 : _k: Kind,
1694 : _id: &str,
1695 : doc: Value,
1696 2 : ) -> Result<bool, NgsiError> {
1697 : *self.0.lock().expect("lock") = Some(doc);
1698 : Ok(true)
1699 2 : }
1700 : async fn batch_create(
1701 : &self,
1702 : _t: &TenantId,
1703 : _items: Vec<(String, Value)>,
1704 0 : ) -> Result<Vec<bool>, NgsiError> {
1705 : unimplemented!()
1706 0 : }
1707 : async fn list_page(
1708 : &self,
1709 : _t: &TenantId,
1710 : _k: Kind,
1711 : _after: Option<&str>,
1712 : _limit: usize,
1713 0 : ) -> Result<Vec<Value>, NgsiError> {
1714 : unimplemented!()
1715 0 : }
1716 : async fn list_slice(
1717 : &self,
1718 : _t: &TenantId,
1719 : _k: Kind,
1720 : _offset: usize,
1721 : _limit: usize,
1722 0 : ) -> Result<(Vec<Value>, usize), NgsiError> {
1723 : unimplemented!()
1724 0 : }
1725 : async fn batch_delete(
1726 : &self,
1727 : _t: &TenantId,
1728 : _ids: &[String],
1729 0 : ) -> Result<Vec<bool>, NgsiError> {
1730 : unimplemented!()
1731 0 : }
1732 : async fn batch_upsert(
1733 : &self,
1734 : _t: &TenantId,
1735 : _items: Vec<(String, Value)>,
1736 0 : ) -> Result<Vec<bool>, NgsiError> {
1737 : unimplemented!()
1738 0 : }
1739 : async fn upsert(
1740 : &self,
1741 : _t: &TenantId,
1742 : _k: Kind,
1743 : _id: &str,
1744 : _doc: Value,
1745 0 : ) -> Result<bool, NgsiError> {
1746 : unimplemented!()
1747 0 : }
1748 : async fn get(
1749 : &self,
1750 : _t: &TenantId,
1751 : _k: Kind,
1752 : _id: &str,
1753 4 : ) -> Result<Option<Value>, NgsiError> {
1754 : Ok(self.0.lock().expect("lock").clone())
1755 4 : }
1756 0 : async fn delete(&self, _t: &TenantId, _k: Kind, _id: &str) -> Result<bool, NgsiError> {
1757 : unimplemented!()
1758 0 : }
1759 : async fn delete_entity_if(
1760 : &self,
1761 : _t: &TenantId,
1762 : _id: &str,
1763 : _keep: &(dyn for<'v> Fn(&'v Value) -> bool + Sync),
1764 0 : ) -> Result<bool, NgsiError> {
1765 : unimplemented!()
1766 0 : }
1767 0 : async fn list(&self, _t: &TenantId, _k: Kind) -> Result<Vec<Value>, NgsiError> {
1768 : unimplemented!()
1769 0 : }
1770 : async fn matching_registrations(
1771 : &self,
1772 : _t: &TenantId,
1773 : _ids: Option<&[String]>,
1774 : _types: Option<&[String]>,
1775 0 : ) -> Result<Vec<Value>, NgsiError> {
1776 : unimplemented!()
1777 0 : }
1778 : async fn query_entities(
1779 : &self,
1780 : _t: &TenantId,
1781 : _f: &filter::EntityFilter<'_>,
1782 0 : ) -> Result<filter::QueryOutcome, NgsiError> {
1783 : unimplemented!()
1784 0 : }
1785 : async fn mutate_boxed<'a>(
1786 : &self,
1787 : _t: &TenantId,
1788 : _k: Kind,
1789 : _id: &str,
1790 : f: MutateFn<'a>,
1791 6 : ) -> Result<Option<Result<(), ()>>, NgsiError> {
1792 : let mut guard = self.0.lock().expect("lock");
1793 : match guard.as_mut() {
1794 : None => Ok(None),
1795 : Some(v) => {
1796 : let mut copy = v.clone();
1797 : match f(&mut copy) {
1798 : Ok(()) => {
1799 : *v = copy;
1800 : Ok(Some(Ok(())))
1801 : }
1802 : Err(()) => Ok(Some(Err(()))),
1803 : }
1804 : }
1805 : }
1806 6 : }
1807 : async fn batch_mutate_boxed<'a>(
1808 : &self,
1809 : _t: &TenantId,
1810 : _ids: &[String],
1811 : _f: BatchMutateFn<'a>,
1812 0 : ) -> Result<Vec<Option<Result<(), ()>>>, NgsiError> {
1813 : unimplemented!()
1814 0 : }
1815 0 : async fn tenant_exists(&self, _t: &TenantId) -> Result<bool, NgsiError> {
1816 : Ok(true)
1817 0 : }
1818 0 : async fn subscription_tenants(&self) -> Result<Vec<String>, NgsiError> {
1819 : Ok(Vec::new())
1820 0 : }
1821 : async fn context_put(
1822 : &self,
1823 : _t: Option<&TenantId>,
1824 : _id: &str,
1825 : _doc: Value,
1826 0 : ) -> Result<(), NgsiError> {
1827 : Ok(())
1828 0 : }
1829 : async fn context_get(
1830 : &self,
1831 : _t: Option<&TenantId>,
1832 : _id: &str,
1833 0 : ) -> Result<Option<Value>, NgsiError> {
1834 : Ok(None)
1835 0 : }
1836 : async fn context_mutate<'a>(
1837 : &self,
1838 : _t: Option<&TenantId>,
1839 : _id: &str,
1840 : _f: MutateFn<'a>,
1841 0 : ) -> Result<Option<Result<(), ()>>, NgsiError> {
1842 : Ok(None)
1843 0 : }
1844 : async fn context_delete(
1845 : &self,
1846 : _t: Option<&TenantId>,
1847 : _id: &str,
1848 0 : ) -> Result<bool, NgsiError> {
1849 : Ok(false)
1850 0 : }
1851 0 : async fn context_list_meta(&self, _t: Option<&TenantId>) -> Result<Vec<Value>, NgsiError> {
1852 : Ok(Vec::new())
1853 0 : }
1854 : }
1855 :
1856 : #[tokio::test]
1857 2 : async fn typed_mutate_round_trips_through_the_boxed_seam() {
1858 2 : let t = TenantId::new("t").expect("tenant");
1859 2 : let d: std::sync::Arc<dyn CurrentStateDriver> =
1860 2 : std::sync::Arc::new(OneDoc(std::sync::Mutex::new(None)));
1861 : // absent → None, closure never runs
1862 2 : let r = d
1863 2 : .mutate::<u32, &str>(&t, Kind::Entity, "x", |_| panic!("must not run"))
1864 2 : .await
1865 2 : .expect("driver ok");
1866 2 : assert!(r.is_none());
1867 2 : d.create(&t, Kind::Entity, "x", serde_json::json!({"n": 1}))
1868 2 : .await
1869 2 : .expect("create");
1870 : // typed success crosses the boundary AND the write commits
1871 2 : let r = d
1872 2 : .mutate::<u32, &str>(&t, Kind::Entity, "x", |v| {
1873 2 : v["n"] = serde_json::json!(2);
1874 2 : Ok(7)
1875 2 : })
1876 2 : .await
1877 2 : .expect("driver ok");
1878 2 : assert_eq!(r, Some(Ok(7)));
1879 2 : assert_eq!(
1880 2 : d.get(&t, Kind::Entity, "x")
1881 2 : .await
1882 2 : .expect("get")
1883 2 : .expect("doc")["n"],
1884 : 2
1885 : );
1886 : // typed error crosses the boundary AND the write is discarded
1887 2 : let r = d
1888 2 : .mutate::<u32, &str>(&t, Kind::Entity, "x", |v| {
1889 2 : v["n"] = serde_json::json!(99);
1890 2 : Err("rejected")
1891 2 : })
1892 2 : .await
1893 2 : .expect("driver ok");
1894 2 : assert_eq!(r, Some(Err("rejected")));
1895 2 : assert_eq!(
1896 2 : d.get(&t, Kind::Entity, "x")
1897 2 : .await
1898 2 : .expect("get")
1899 2 : .expect("doc")["n"],
1900 2 : 2,
1901 2 : "a rejecting closure must not commit"
1902 2 : );
1903 2 : }
1904 :
1905 : /// A temporal driver that only records what the seam hands it: appends
1906 : /// as (id, additions) in call order, scope mutations on a held doc.
1907 : #[derive(Default)]
1908 : struct Recorder {
1909 : appends: std::sync::Mutex<Vec<(String, Value)>>,
1910 : doc: std::sync::Mutex<Option<Value>>,
1911 : /// the Entity whose appends this driver refuses (a full history)
1912 : refuse: Option<&'static str>,
1913 : }
1914 : #[async_trait::async_trait]
1915 : impl TemporalDriver for Recorder {
1916 : async fn temporal_append(
1917 : &self,
1918 : _t: &TenantId,
1919 : id: &str,
1920 : shell: &Value,
1921 : additions: &Value,
1922 18 : ) -> Result<(), NgsiError> {
1923 : if self.refuse == Some(id) {
1924 : return Err(NgsiError::BadRequestData("history full".into()));
1925 : }
1926 : self.appends
1927 : .lock()
1928 : .expect("lock")
1929 : .push((id.to_owned(), additions.clone()));
1930 : // a real driver creates the doc from the shell on first touch
1931 : self.doc
1932 : .lock()
1933 : .expect("lock")
1934 8 : .get_or_insert_with(|| shell.clone());
1935 : Ok(())
1936 18 : }
1937 : async fn query_temporal(
1938 : &self,
1939 : _t: &TenantId,
1940 : _f: &filter::TemporalFilter<'_>,
1941 0 : ) -> Result<filter::TemporalOutcome, NgsiError> {
1942 : unimplemented!()
1943 0 : }
1944 : async fn get_temporal(
1945 : &self,
1946 : _t: &TenantId,
1947 : _id: &str,
1948 : _f: &filter::TemporalFilter<'_>,
1949 0 : ) -> Result<Option<Value>, NgsiError> {
1950 : unimplemented!()
1951 0 : }
1952 6 : async fn get(&self, _t: &TenantId, _id: &str) -> Result<Option<Value>, NgsiError> {
1953 : Ok(self.doc.lock().expect("lock").clone())
1954 6 : }
1955 2 : async fn create(&self, _t: &TenantId, _id: &str, doc: Value) -> Result<bool, NgsiError> {
1956 : *self.doc.lock().expect("lock") = Some(doc);
1957 : Ok(true)
1958 2 : }
1959 0 : async fn upsert(&self, _t: &TenantId, _id: &str, _doc: Value) -> Result<bool, NgsiError> {
1960 : unimplemented!()
1961 0 : }
1962 0 : async fn delete(&self, _t: &TenantId, _id: &str) -> Result<bool, NgsiError> {
1963 : unimplemented!()
1964 0 : }
1965 0 : async fn list(&self, _t: &TenantId) -> Result<Vec<Value>, NgsiError> {
1966 : unimplemented!()
1967 0 : }
1968 : async fn mutate_boxed<'a>(
1969 : &self,
1970 : _t: &TenantId,
1971 : _id: &str,
1972 : f: MutateFn<'a>,
1973 8 : ) -> Result<Option<Result<(), ()>>, NgsiError> {
1974 : let mut guard = self.doc.lock().expect("lock");
1975 : match guard.as_mut() {
1976 : None => Ok(None),
1977 : Some(v) => Ok(Some(f(v))),
1978 : }
1979 8 : }
1980 : }
1981 :
1982 36 : fn ev(op: TemporalOp, id: &str, attr: &str, n: u32) -> TemporalEvent {
1983 36 : TemporalEvent {
1984 36 : op,
1985 36 : tenant: TenantId::new("t").expect("tenant"),
1986 36 : entity_id: id.into(),
1987 36 : shell: serde_json::json!({"id": id, "type": ["T"]}),
1988 36 : attr: attr.into(),
1989 36 : instance: serde_json::json!({"type": "Property", "value": n}),
1990 36 : }
1991 36 : }
1992 :
1993 : /// The drain folds one request's events into ONE append per entity run
1994 : /// — a 2-attribute entity is one call carrying both, not two — and
1995 : /// keeps production order across entities.
1996 : #[test]
1997 2 : fn fold_appends_is_one_append_per_entity_run_and_none_for_a_scope_change() {
1998 2 : let evs = [
1999 2 : ev(TemporalOp::AttrCreated, "urn:a", "speed", 1),
2000 2 : ev(TemporalOp::AttrCreated, "urn:a", "speed", 2),
2001 2 : ev(TemporalOp::AttrCreated, "urn:b", "speed", 3),
2002 2 : ev(TemporalOp::AttrCreated, "urn:a", "load", 4),
2003 2 : ];
2004 2 : let folded = fold_appends(&evs).expect("no scope change");
2005 2 : let ids: Vec<&str> = folded.iter().map(|(_, id, _, _)| *id).collect();
2006 2 : assert_eq!(ids, ["urn:a", "urn:b", "urn:a"]);
2007 2 : assert_eq!(folded[0].3["speed"].as_array().map(Vec::len), Some(2));
2008 2 : assert!(fold_appends(&[]).is_some_and(|f| f.is_empty()));
2009 2 : let scope = ev(TemporalOp::ScopeChanged, "urn:a", "scope", 5);
2010 2 : assert!(fold_appends(&[evs[0].clone(), scope]).is_none());
2011 2 : }
2012 :
2013 : #[tokio::test]
2014 2 : async fn event_list_folds_a_request_into_one_append_per_entity_run() {
2015 2 : let d = Recorder::default();
2016 2 : d.event_list(&[
2017 2 : ev(TemporalOp::AttrCreated, "urn:a", "speed", 1),
2018 2 : ev(TemporalOp::AttrCreated, "urn:a", "speed", 2),
2019 2 : ev(TemporalOp::AttrModified, "urn:a", "heading", 3),
2020 2 : ev(TemporalOp::AttrModified, "urn:b", "speed", 4),
2021 2 : ev(TemporalOp::AttrModified, "urn:a", "speed", 5),
2022 2 : ])
2023 2 : .await
2024 2 : .expect("drain ok");
2025 2 : let appends = d.appends.lock().expect("lock").clone();
2026 6 : let ids: Vec<&str> = appends.iter().map(|(id, _)| id.as_str()).collect();
2027 2 : assert_eq!(
2028 : ids,
2029 : ["urn:a", "urn:b", "urn:a"],
2030 : "one append per entity run, in order"
2031 : );
2032 2 : let first = &appends[0].1;
2033 2 : assert_eq!(first["speed"].as_array().map(Vec::len), Some(2), "{first}");
2034 2 : assert_eq!(
2035 2 : first["heading"].as_array().map(Vec::len),
2036 : Some(1),
2037 : "{first}"
2038 : );
2039 2 : assert!(
2040 2 : first.get("value").is_none(),
2041 : "instances live under their attribute: {first}"
2042 : );
2043 2 : assert_eq!(appends[2].1["speed"][0]["value"], 5);
2044 2 : }
2045 :
2046 : /// One Entity whose append the driver refuses (its history is full)
2047 : /// loses its own history of the request, never the other Entities':
2048 : /// the drain goes on and reports the refusal once it is done.
2049 : #[tokio::test]
2050 2 : async fn a_refused_append_drops_only_its_own_entity() {
2051 2 : let d = Recorder {
2052 2 : refuse: Some("urn:a"),
2053 2 : ..Recorder::default()
2054 2 : };
2055 2 : let r = d
2056 2 : .event_list(&[
2057 2 : ev(TemporalOp::AttrModified, "urn:a", "speed", 1),
2058 2 : ev(TemporalOp::AttrModified, "urn:b", "speed", 2),
2059 2 : ev(TemporalOp::AttrModified, "urn:c", "speed", 3),
2060 2 : ])
2061 2 : .await;
2062 2 : assert!(matches!(r, Err(NgsiError::BadRequestData(_))), "{r:?}");
2063 2 : let ids: Vec<String> = d
2064 2 : .appends
2065 2 : .lock()
2066 2 : .expect("lock")
2067 2 : .iter()
2068 4 : .map(|(id, _)| id.clone())
2069 2 : .collect();
2070 2 : assert_eq!(ids, ["urn:b", "urn:c"]);
2071 2 : }
2072 :
2073 : /// 4.5.6: a scope change becomes a scope instance on the held temporal
2074 : /// doc (array-of-instances form), not an attribute append.
2075 : #[tokio::test]
2076 2 : async fn event_list_records_scope_changes_as_scope_instances() {
2077 2 : let t = TenantId::new("t").expect("tenant");
2078 2 : let d = Recorder::default();
2079 2 : d.create(
2080 2 : &t,
2081 2 : "urn:a",
2082 2 : serde_json::json!({"id": "urn:a", "scope": "/old"}),
2083 2 : )
2084 2 : .await
2085 2 : .expect("create");
2086 2 : let mut scope = ev(TemporalOp::ScopeChanged, "urn:a", "scope", 0);
2087 2 : scope.instance = serde_json::json!({"type": "Property", "value": "/new",
2088 2 : "observedAt": "2026-01-01T00:00:00Z"});
2089 2 : d.event_list(&[scope]).await.expect("drain ok");
2090 2 : assert!(
2091 2 : d.appends.lock().expect("lock").is_empty(),
2092 : "no attribute append for a scope change"
2093 : );
2094 2 : let doc = d.get(&t, "urn:a").await.expect("get").expect("doc");
2095 2 : assert_eq!(doc["scope"][0]["value"], "/new", "{doc}");
2096 2 : assert_eq!(
2097 2 : doc["scope"].as_array().map(Vec::len),
2098 2 : Some(1),
2099 2 : "the plain scope became an instance array: {doc}"
2100 2 : );
2101 2 : }
2102 :
2103 : /// a scope change first in a run, with no temporal doc yet,
2104 : /// still lands: the shell is created rather than the instance dropped,
2105 : /// and additions queued before it are appended first.
2106 : #[tokio::test]
2107 2 : async fn event_list_keeps_a_scope_change_on_an_entity_without_history() {
2108 2 : let t = TenantId::new("t").expect("tenant");
2109 2 : let d = Recorder::default();
2110 2 : d.event_list(&[
2111 2 : ev(TemporalOp::ScopeChanged, "urn:a", "scope", 7),
2112 2 : ev(TemporalOp::AttrCreated, "urn:a", "speed", 1),
2113 2 : ])
2114 2 : .await
2115 2 : .expect("drain ok");
2116 2 : let doc = d.get(&t, "urn:a").await.expect("get").expect("doc");
2117 2 : assert_eq!(doc["scope"][0]["value"], 7, "{doc}");
2118 :
2119 2 : let d = Recorder::default();
2120 2 : d.event_list(&[
2121 2 : ev(TemporalOp::AttrCreated, "urn:a", "speed", 1),
2122 2 : ev(TemporalOp::ScopeChanged, "urn:a", "scope", 7),
2123 2 : ])
2124 2 : .await
2125 2 : .expect("drain ok");
2126 2 : let appends = d.appends.lock().expect("lock").clone();
2127 2 : assert_eq!(appends.len(), 1, "{appends:?}");
2128 2 : assert_eq!(appends[0].1["speed"][0]["value"], 1);
2129 2 : let doc = d.get(&t, "urn:a").await.expect("get").expect("doc");
2130 2 : assert_eq!(doc["scope"][0]["value"], 7, "{doc}");
2131 2 : }
2132 :
2133 : #[tokio::test]
2134 2 : async fn no_temporal_degrades_without_panicking() {
2135 2 : let t = TenantId::new("t").expect("tenant");
2136 2 : let d: std::sync::Arc<dyn TemporalDriver> = std::sync::Arc::new(NoTemporal);
2137 2 : assert!(!d.supported());
2138 : // client-facing reads: the spec error, not a panic
2139 2 : let e = match d
2140 2 : .query_temporal(&t, &filter::TemporalFilter::default())
2141 2 : .await
2142 : {
2143 2 : Err(e) => e,
2144 0 : Ok(_) => panic!("query_temporal on NoTemporal must be unsupported"),
2145 : };
2146 2 : assert_eq!(e.status(), 422);
2147 2 : assert_eq!(e.kind(), "OperationNotSupported");
2148 : // internal bookkeeping: benign no-ops
2149 2 : assert!(d
2150 2 : .temporal_append(&t, "x", &Value::Null, &Value::Null)
2151 2 : .await
2152 2 : .is_ok());
2153 2 : assert!(!d.delete(&t, "x").await.expect("ok"));
2154 2 : assert!(d.list(&t).await.expect("ok").is_empty());
2155 2 : assert!(d
2156 2 : .mutate::<(), ()>(&t, "x", |_| Ok(()))
2157 2 : .await
2158 2 : .expect("ok")
2159 2 : .is_none());
2160 2 : }
2161 : }
|
