Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! In-memory store — v0 storage backend.
3 : //!
4 : //! DELIBERATE DEVIATION (recorded in docs/adr/): the target backend is
5 : //! Postgres; the suite-green loop uses an in-memory backend first. The store
6 : //! API is shaped like the Postgres store traits (tenant first parameter
7 : //! everywhere) so the sqlx implementation can land behind the same seam.
8 : //!
9 : //! Documents are held in the *internal expanded form* produced by
10 : //! `antares_jsonld::expand` (IRI keys, instance arrays), with server-managed
11 : //! timestamps embedded (`createdAt`/`modifiedAt` at entity level and inside
12 : //! each attribute instance) — output layers strip them unless sysAttrs.
13 :
14 : use ::redb::{Database, Durability, ReadableDatabase, ReadableTableMetadata, TableDefinition};
15 : use antares_model::{NgsiError, TenantId};
16 : use antares_store::{context_row_owner, context_row_visible, filter, ChangeHook, Kind};
17 :
18 : mod redb;
19 : use self::redb::{
20 : key_bytes, split_key, table_for, Shadow, FORMAT_VERSION, T_ENTITIES, T_JSONLD_CONTEXTS, T_META,
21 : T_TEMPORAL_ENTITIES,
22 : };
23 : use serde_json::Value;
24 : use std::collections::{BTreeMap, HashMap};
25 : use std::path::Path;
26 : use std::sync::RwLock;
27 :
28 : /// How many `Cached` @context entries the broker keeps (5.13.1). The bound is
29 : /// on the CACHE only: it exists because one entry is stored per distinct
30 : /// external @context URL a request references, which is client-controlled,
31 : /// while the working set of real @contexts a deployment uses is small. Every
32 : /// entry holds a whole @context body, so the count is the memory bound too.
33 : pub const MAX_CACHED_CONTEXTS: usize = 1_000;
34 :
35 : /// The `Cached` entry ids to drop so at most `MAX_CACHED_CONTEXTS` remain,
36 : /// oldest `createdAt` first (5.13.1 stamps every stored entry with one; an
37 : /// entry without a stamp sorts oldest and goes first). Other kinds are never
38 : /// candidates.
39 4 : fn oldest_cached(contexts: &BTreeMap<String, Value>) -> Vec<String> {
40 4 : let mut cached: Vec<(&str, &str)> = contexts
41 4 : .iter()
42 4006 : .filter(|(_, d)| d.get("kind").and_then(Value::as_str) == Some("Cached"))
43 4002 : .map(|(id, d)| {
44 4002 : (
45 4002 : id.as_str(),
46 4002 : d.get("createdAt").and_then(Value::as_str).unwrap_or(""),
47 4002 : )
48 4002 : })
49 4 : .collect();
50 4 : if cached.len() <= MAX_CACHED_CONTEXTS {
51 2 : return Vec::new();
52 2 : }
53 20780 : cached.sort_unstable_by(|(a_id, a_ts), (b_id, b_ts)| (a_ts, a_id).cmp(&(b_ts, b_id)));
54 2 : cached[..cached.len() - MAX_CACHED_CONTEXTS]
55 2 : .iter()
56 2 : .map(|(id, _)| (*id).to_owned())
57 2 : .collect()
58 4 : }
59 :
60 : /// Drop attribute instances whose `expiresAt` passed (4.22) from an internal
61 : /// entity or temporal doc; an attribute whose last instance expired is
62 : /// removed entirely. Returns whether the doc changed.
63 1516 : fn prune_expired_instances(doc: &mut Value, now: &str) -> bool {
64 1516 : let Some(obj) = doc.as_object_mut() else {
65 0 : return false;
66 : };
67 1516 : let mut changed = false;
68 1516 : let attrs: Vec<String> = obj
69 1516 : .keys()
70 6771 : .filter(|k| !antares_model::ENTITY_META_KEYS.contains(&k.as_str()))
71 1516 : .cloned()
72 1516 : .collect();
73 1516 : for k in attrs {
74 721 : let Some(arr) = obj.get_mut(&k).and_then(Value::as_array_mut) else {
75 0 : continue;
76 : };
77 721 : let before = arr.len();
78 : // 4.6.3 leaves the seconds-fraction separator open, so the two
79 : // stamps have to be compared as instants and not as bytes — the rule
80 : // lives in `filter::expired_at`, which the read boundary uses too.
81 742 : arr.retain(|inst| !filter::expired_at(inst, now));
82 721 : if arr.len() != before {
83 12 : changed = true;
84 12 : if arr.is_empty() {
85 4 : obj.remove(&k);
86 8 : }
87 709 : }
88 : }
89 1516 : changed
90 1516 : }
91 :
92 : /// The 4.22 "now" the write paths judge `expiresAt` against — the same UTC-Z
93 : /// millisecond form the read boundary uses.
94 8818 : fn now_stamp() -> String {
95 8818 : chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
96 8818 : }
97 :
98 : const ALL_KINDS: [Kind; 9] = [
99 : Kind::Entity,
100 : Kind::Subscription,
101 : Kind::Registration,
102 : Kind::CSourceSubscription,
103 : Kind::Temporal,
104 : Kind::Snapshot,
105 : Kind::EntityMap,
106 : Kind::DistSub,
107 : Kind::DeadLetter,
108 : ];
109 :
110 : /// Run `f` off the tokio worker pool when called from a multi-thread runtime:
111 : /// a per-commit fsync must never stall an async worker. Outside a
112 : /// runtime (unit tests, startup) it just runs inline.
113 : ///
114 : /// `durable` is what decides. A per-document write blocks only where it
115 : /// commits, so it passes `shadow.is_some()`: in `memory` mode that write is
116 : /// a lock and a map insert, and handing a worker's queue to another thread —
117 : /// a handoff, and past the pool's live threads a spawn — costs more than the
118 : /// write and costs more the more cores there are to hand work between. The
119 : /// paths that hold the write section for a whole SCAN (the 4.22 sweep, the
120 : /// two purges) pass `true` whatever the mode: what makes them long there is
121 : /// the scan, not the commit.
122 68509 : fn on_blocking<T>(durable: bool, f: impl FnOnce() -> T) -> T {
123 : // wasm32: single-threaded, no tokio runtime — always inline.
124 : #[cfg(not(target_arch = "wasm32"))]
125 68509 : if durable {
126 3043 : if let Ok(h) = tokio::runtime::Handle::try_current() {
127 3037 : if h.runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread {
128 2775 : return tokio::task::block_in_place(f);
129 262 : }
130 6 : }
131 65466 : }
132 : #[cfg(target_arch = "wasm32")]
133 : let _ = durable;
134 65734 : f()
135 68509 : }
136 :
137 : #[derive(Default)]
138 : pub struct Store {
139 : inner: RwLock<Inner>,
140 : hook: RwLock<Option<ChangeHook>>,
141 : /// Set when this instance serves only the temporal seam: it never holds
142 : /// the entities, so the append guard must not look for them here.
143 : pub temporal_only: bool,
144 : /// `file` mode durability shadow; `None` = pure in-memory (`memory` mode).
145 : shadow: Option<Shadow>,
146 : /// Writers currently queued behind the single write-critical section
147 : /// (redb has ONE writer, so fsync commits serialize here). Exported via
148 : /// /q/health; the group-commit lever only gets built if a benchmark shows
149 : /// this depth sustained at the measured ~3.1k writes/s ceiling.
150 : write_waiters: std::sync::atomic::AtomicUsize,
151 : write_waiters_peak: std::sync::atomic::AtomicUsize,
152 : /// Change hooks fire in COMMIT order. The hook runs after the
153 : /// write-critical section (it may write other kinds through the store,
154 : /// so running it inside would deadlock), which lets a later commit's
155 : /// hook overtake an earlier one — the consumer would record stale state
156 : /// as newest. Entity writes therefore hold this from before the commit
157 : /// until the emit is done. Only the local single-process path needs it:
158 : /// across processes the transactional outbox is the ordered channel.
159 : emit_order: tokio::sync::Mutex<()>,
160 : }
161 :
162 : #[derive(Default)]
163 : struct Inner {
164 : /// tenant → id → internal entity doc (BTreeMap: deterministic list order).
165 : entities: HashMap<String, BTreeMap<String, Value>>,
166 : subscriptions: HashMap<String, BTreeMap<String, Value>>,
167 : registrations: HashMap<String, BTreeMap<String, Value>>,
168 : csource_subscriptions: HashMap<String, BTreeMap<String, Value>>,
169 : snapshots: HashMap<String, BTreeMap<String, Value>>,
170 : entity_map_docs: HashMap<String, BTreeMap<String, Value>>,
171 : dist_subs: HashMap<String, BTreeMap<String, Value>>,
172 : dead_letters: HashMap<String, BTreeMap<String, Value>>,
173 : /// tenant → entity id → temporal doc (attr IRI → instance array).
174 : temporal: HashMap<String, BTreeMap<String, Value>>,
175 : /// hosted/cached @context documents, shared across tenants by design.
176 : contexts: BTreeMap<String, Value>,
177 : }
178 :
179 : impl Store {
180 : /// Open (or create) the `file`-mode store: redb at `dir/antares.redb`,
181 : /// format-checked, in-memory maps rebuilt from the file.
182 : /// Any open/format/decode error refuses to start — never silently serve
183 : /// partial data.
184 56 : pub fn open_file(dir: &Path) -> Result<Self, String> {
185 56 : std::fs::create_dir_all(dir).map_err(|e| format!("create {}: {e}", dir.display()))?;
186 56 : let path = dir.join("antares.redb");
187 56 : let db = Database::create(&path).map_err(|e| format!("open {}: {e}", path.display()))?;
188 54 : Self::from_database(db, &path.display().to_string())
189 56 : }
190 :
191 : /// The target-independent half of `open_file`: format check +
192 : /// boot rebuild over an already-constructed redb `Database`. The
193 : /// browser build calls this with an OPFS-backed database — same shadow,
194 : /// same commit-before-ack, different `StorageBackend`.
195 58 : pub fn from_database(db: Database, label: &str) -> Result<Self, String> {
196 : // Format marker. Absent marker + existing data = a file this
197 : // binary cannot vouch for; refuse rather than guess.
198 58 : let stored_format = {
199 58 : let rt = db.begin_read().map_err(|e| e.to_string())?;
200 58 : match rt.open_table(T_META) {
201 28 : Ok(t) => t
202 28 : .get("format")
203 28 : .map_err(|e| e.to_string())?
204 28 : .map(|v| v.value().to_owned()),
205 30 : Err(::redb::TableError::TableDoesNotExist(_)) => None,
206 0 : Err(e) => return Err(e.to_string()),
207 : }
208 : };
209 58 : match stored_format.as_deref() {
210 28 : Some(FORMAT_VERSION) => {}
211 2 : Some(other) => {
212 2 : return Err(format!(
213 2 : "data file {label} has format {other}, this binary supports {FORMAT_VERSION} — \
214 2 : refusing to start"
215 2 : ));
216 : }
217 : None => {
218 30 : let rt = db.begin_read().map_err(|e| e.to_string())?;
219 238 : for kind in ALL_KINDS {
220 : // Absent = nothing stored under that kind. Any other
221 : // error is a table this binary cannot read, and reading
222 : // it as empty would let the guard stamp its marker onto
223 : // a file it is about to refuse.
224 238 : let t = match rt.open_table(table_for(kind)) {
225 2 : Ok(t) => t,
226 234 : Err(::redb::TableError::TableDoesNotExist(_)) => continue,
227 2 : Err(e) => return Err(e.to_string()),
228 : };
229 2 : if t.len().map_err(|e| e.to_string())? > 0 {
230 2 : return Err(format!(
231 2 : "data file {label} holds data but no format marker — refusing to start"
232 2 : ));
233 0 : }
234 : }
235 26 : drop(rt);
236 26 : let mut tx = db.begin_write().map_err(|e| e.to_string())?;
237 26 : tx.set_durability(Durability::Immediate)
238 26 : .map_err(|e| e.to_string())?;
239 : {
240 26 : let mut t = tx.open_table(T_META).map_err(|e| e.to_string())?;
241 26 : t.insert("format", FORMAT_VERSION)
242 26 : .map_err(|e| e.to_string())?;
243 : }
244 26 : tx.commit().map_err(|e| e.to_string())?;
245 : }
246 : }
247 :
248 : // Boot rebuild — scan every table into the in-memory maps.
249 52 : let mut inner = Inner::default();
250 52 : let rt = db.begin_read().map_err(|e| e.to_string())?;
251 468 : for kind in ALL_KINDS {
252 468 : let table = match rt.open_table(table_for(kind)) {
253 66 : Ok(t) => t,
254 402 : Err(::redb::TableError::TableDoesNotExist(_)) => continue,
255 0 : Err(e) => return Err(e.to_string()),
256 : };
257 66 : let map = match kind {
258 18 : Kind::Entity => &mut inner.entities,
259 6 : Kind::Subscription => &mut inner.subscriptions,
260 6 : Kind::Registration => &mut inner.registrations,
261 4 : Kind::CSourceSubscription => &mut inner.csource_subscriptions,
262 8 : Kind::Temporal => &mut inner.temporal,
263 6 : Kind::Snapshot => &mut inner.snapshots,
264 6 : Kind::EntityMap => &mut inner.entity_map_docs,
265 6 : Kind::DistSub => &mut inner.dist_subs,
266 6 : Kind::DeadLetter => &mut inner.dead_letters,
267 : };
268 80 : for row in ::redb::ReadableTable::iter(&table).map_err(|e| e.to_string())? {
269 80 : let (k, v) = row.map_err(|e| e.to_string())?;
270 80 : let (tenant, id) = split_key(k.value())
271 80 : .ok_or_else(|| format!("undecodable key in table {kind:?}"))?;
272 80 : let doc: Value = serde_json::from_slice(v.value())
273 80 : .map_err(|e| format!("undecodable value for {tenant}/{id}: {e}"))?;
274 80 : map.entry(tenant).or_default().insert(id, doc);
275 : }
276 : }
277 : // Same rule as the kind loop above: absent is empty, unreadable is a
278 : // refusal. Hosted and ImplicitlyCreated @contexts (5.13.1) are a
279 : // Tenant's own documents, so starting without this table is serving
280 : // partial data, which `open_file` promises never to do.
281 52 : let contexts = match rt.open_table(T_JSONLD_CONTEXTS) {
282 4 : Ok(t) => Some(t),
283 46 : Err(::redb::TableError::TableDoesNotExist(_)) => None,
284 2 : Err(e) => return Err(e.to_string()),
285 : };
286 50 : if let Some(t) = contexts {
287 4 : for row in ::redb::ReadableTable::iter(&t).map_err(|e| e.to_string())? {
288 2 : let (k, v) = row.map_err(|e| e.to_string())?;
289 2 : let id = String::from_utf8(k.value().to_vec())
290 2 : .map_err(|e| format!("undecodable context id: {e}"))?;
291 2 : let doc: Value = serde_json::from_slice(v.value())
292 2 : .map_err(|e| format!("undecodable context {id}: {e}"))?;
293 2 : inner.contexts.insert(id, doc);
294 : }
295 46 : }
296 50 : drop(rt);
297 :
298 50 : Ok(Self {
299 50 : inner: RwLock::new(inner),
300 50 : hook: RwLock::new(None),
301 50 : emit_order: tokio::sync::Mutex::new(()),
302 50 : shadow: Some(Shadow { db }),
303 50 : temporal_only: false,
304 50 : write_waiters: Default::default(),
305 50 : write_waiters_peak: Default::default(),
306 50 : })
307 58 : }
308 :
309 : /// Write-through for one doc (must be called inside the write-critical
310 : /// section so redb order equals memory order).
311 58495 : fn persist(&self, table: TableDefinition<&[u8], &[u8]>, key: &[u8], doc: Option<&Value>) {
312 58495 : if let Some(shadow) = &self.shadow {
313 : // None means DELETE the key, so a document that will not encode
314 : // must skip the write rather than collapse into one.
315 1241 : let bytes = match doc {
316 772 : Some(d) => match serde_json::to_vec(d) {
317 772 : Ok(b) => Some(b),
318 0 : Err(_) => return,
319 : },
320 469 : None => None,
321 : };
322 1241 : shadow.write(table, key, bytes.as_deref());
323 57254 : }
324 58495 : }
325 :
326 : /// Acquire the write-critical section, counting queued writers.
327 : ///
328 : /// Poison recovery (`into_inner`) is deliberate: a panic inside a caller's
329 : /// mutate closure unwinds one request; poisoning would turn it into a
330 : /// whole-process brick (every later store call panicking until restart).
331 : /// Consistency holds because mutations work on a clone and swap last.
332 68509 : fn write_inner(&self) -> std::sync::RwLockWriteGuard<'_, Inner> {
333 : use std::sync::atomic::Ordering;
334 68509 : let depth = self.write_waiters.fetch_add(1, Ordering::Relaxed) + 1;
335 68509 : self.write_waiters_peak.fetch_max(depth, Ordering::Relaxed);
336 68509 : let guard = self
337 68509 : .inner
338 68509 : .write()
339 68509 : .unwrap_or_else(std::sync::PoisonError::into_inner);
340 68509 : self.write_waiters.fetch_sub(1, Ordering::Relaxed);
341 68509 : guard
342 68509 : }
343 :
344 : /// Whether writes reach a durability shadow (`file` mode, and the
345 : /// browser build over OPFS) rather than living only in memory.
346 254 : pub fn shadowed(&self) -> bool {
347 254 : self.shadow.is_some()
348 254 : }
349 :
350 : /// (Currently queued writers, peak since start). The peak going
351 : /// nowhere near sustained depth is the evidence that the group-commit
352 : /// lever stays unbuilt.
353 8 : pub fn commit_queue(&self) -> (usize, usize) {
354 : use std::sync::atomic::Ordering;
355 8 : (
356 8 : self.write_waiters.load(Ordering::Relaxed),
357 8 : self.write_waiters_peak.load(Ordering::Relaxed),
358 8 : )
359 8 : }
360 :
361 566 : pub fn set_change_hook(&self, h: ChangeHook) {
362 566 : *self
363 566 : .hook
364 566 : .write()
365 566 : .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(h);
366 566 : }
367 :
368 : /// Held from before an entity commit until its emit returns (see
369 : /// `emit_order`). Not for other kinds: their writes emit nothing, and a
370 : /// hook that writes them re-enters the store.
371 39134 : async fn emit_ordered(&self) -> tokio::sync::MutexGuard<'_, ()> {
372 39134 : self.emit_order.lock().await
373 39134 : }
374 :
375 37858 : async fn emit(&self, tenant: &TenantId, before: Option<Value>, after: Option<Value>) {
376 : // cloned out of the lock first: a read guard cannot be held across
377 : // the await the hook now costs.
378 37858 : let hook = self
379 37858 : .hook
380 37858 : .read()
381 37858 : .unwrap_or_else(std::sync::PoisonError::into_inner)
382 37858 : .clone();
383 37858 : if let Some(h) = hook {
384 5900 : h(tenant, before, after).await;
385 31958 : }
386 37856 : }
387 :
388 : /// 4.22 garbage collection for the memory/file arm: remove entity docs
389 : /// whose `expiresAt` (byte-compared against the UTC-Z `now` stamp, same
390 : /// as the read filter) has passed, and prune expired ATTRIBUTE instances
391 : /// from current-state and temporal docs — the read filter hides them,
392 : /// but without physical removal a long-running store (the browser's OPFS
393 : /// file under ticking sensors) grows without bound. `file` mode persists
394 : /// each removal. Returns how many docs were reaped or pruned.
395 : ///
396 : /// Runs through `on_blocking` like every other mutating path: in `file`
397 : /// mode it holds the write-critical section for a full scan and issues
398 : /// one `Durability::Immediate` (fsync) commit per reaped doc, which must
399 : /// never happen on an async worker thread.
400 390 : pub fn sweep_expired(&self, now: &str) -> usize {
401 390 : on_blocking(true, || self.sweep_expired_locked(now))
402 390 : }
403 :
404 390 : fn sweep_expired_locked(&self, now: &str) -> usize {
405 390 : let mut inner = self.write_inner();
406 390 : let mut reaped = 0usize;
407 390 : let mut dead: Vec<(String, String)> = Vec::new();
408 1429 : for (tenant, docs) in &inner.entities {
409 1427 : for (id, doc) in docs {
410 1144 : if filter::expired_at(doc, now) {
411 8 : dead.push((tenant.clone(), id.clone()));
412 1136 : }
413 : }
414 : }
415 390 : for (tenant, id) in dead {
416 8 : if let Some(docs) = inner.entities.get_mut(&tenant) {
417 8 : self.persist(T_ENTITIES, &key_bytes(&tenant, &id), None);
418 8 : docs.remove(&id);
419 8 : reaped += 1;
420 8 : }
421 : }
422 : let Inner {
423 390 : entities, temporal, ..
424 390 : } = &mut *inner;
425 780 : for (table, map) in [
426 390 : (
427 390 : T_ENTITIES,
428 390 : entities as &mut HashMap<String, BTreeMap<String, Value>>,
429 390 : ),
430 390 : (T_TEMPORAL_ENTITIES, temporal),
431 390 : ] {
432 1803 : for (tenant, docs) in map.iter_mut() {
433 1785 : for (id, doc) in docs.iter_mut() {
434 1512 : if prune_expired_instances(doc, now) {
435 8 : reaped += 1;
436 8 : self.persist(table, &key_bytes(tenant, id), Some(doc));
437 1504 : }
438 : }
439 : }
440 : }
441 390 : reaped
442 390 : }
443 :
444 : /// Tenants that hold any subscriptions (interval-firing scan).
445 : /// 5.5.10: a Tenant exists once any create operation implicitly created
446 : /// it (any resource kind was ever written under it).
447 300 : pub fn tenant_exists(&self, tenant: &TenantId) -> bool {
448 300 : let inner = self
449 300 : .inner
450 300 : .read()
451 300 : .unwrap_or_else(std::sync::PoisonError::into_inner);
452 300 : ALL_KINDS
453 300 : .iter()
454 830 : .any(|k| Self::map(&inner, *k).contains_key(tenant.as_str()))
455 300 : }
456 :
457 : /// Every tenant name, sorted; the default tenant is always present, and
458 : /// the tenants the broker minted for itself are not — `/q/tenants` is the
459 : /// inventory of customer accounts, not of the broker's own bookkeeping.
460 414 : pub fn tenant_ids(&self) -> Vec<String> {
461 414 : let inner = self
462 414 : .inner
463 414 : .read()
464 414 : .unwrap_or_else(std::sync::PoisonError::into_inner);
465 414 : let mut names: std::collections::BTreeSet<String> =
466 414 : std::iter::once(TenantId::DEFAULT.to_string()).collect();
467 3726 : for kind in ALL_KINDS {
468 3726 : names.extend(
469 3726 : Self::map(&inner, kind)
470 3726 : .keys()
471 3922 : .filter(|t| !TenantId::is_reserved_str(t))
472 3726 : .cloned(),
473 : );
474 : }
475 414 : names.into_iter().collect()
476 414 : }
477 :
478 : /// What one tenant holds. Existence is the caller's question — an
479 : /// unknown tenant simply counts zero of everything.
480 24 : pub fn tenant_stats_one(&self, tenant: &TenantId) -> antares_store::TenantStats {
481 24 : let inner = self
482 24 : .inner
483 24 : .read()
484 24 : .unwrap_or_else(std::sync::PoisonError::into_inner);
485 168 : let n = |k: Kind| {
486 168 : Self::map(&inner, k)
487 168 : .get(tenant.as_str())
488 168 : .map_or(0, |m| m.len() as u64)
489 168 : };
490 24 : antares_store::TenantStats {
491 24 : entities: n(Kind::Entity),
492 24 : subscriptions: n(Kind::Subscription),
493 24 : registrations: n(Kind::Registration),
494 24 : csource_subscriptions: n(Kind::CSourceSubscription),
495 24 : snapshots: n(Kind::Snapshot),
496 24 : entity_maps: n(Kind::EntityMap),
497 24 : dist_subs: n(Kind::DistSub),
498 24 : created_at: None,
499 24 : tenant: tenant.as_str().to_owned(),
500 24 : }
501 24 : }
502 :
503 : /// Attribute instances held in the tenant's temporal documents.
504 20 : pub fn attr_instance_count(&self, tenant: &TenantId) -> u64 {
505 20 : let inner = self
506 20 : .inner
507 20 : .read()
508 20 : .unwrap_or_else(std::sync::PoisonError::into_inner);
509 20 : inner.temporal.get(tenant.as_str()).map_or(0, |docs| {
510 12 : docs.values()
511 12 : .filter_map(Value::as_object)
512 12 : .flat_map(|d| d.values().filter_map(Value::as_array))
513 24 : .map(|a| a.iter().filter(|i| i.is_object()).count() as u64)
514 12 : .sum()
515 12 : })
516 20 : }
517 :
518 : /// Drop every document of the given kinds for one tenant; `true` when
519 : /// the tenant held any of them. Persisted per key in `file` mode.
520 122 : pub fn purge_kinds(&self, tenant: &TenantId, kinds: &[Kind]) -> bool {
521 122 : on_blocking(true, || {
522 122 : let mut inner = self.write_inner();
523 122 : let mut hit = false;
524 634 : for kind in kinds {
525 634 : if let Some(docs) = Self::map_mut(&mut inner, *kind).remove(tenant.as_str()) {
526 124 : hit = true;
527 158 : for id in docs.keys() {
528 158 : self.persist(table_for(*kind), &key_bytes(tenant.as_str(), id), None);
529 158 : }
530 510 : }
531 : }
532 122 : hit
533 122 : })
534 122 : }
535 :
536 : /// Remove the tenant from every kind, history included.
537 64 : pub fn purge_tenant(&self, tenant: &TenantId) -> bool {
538 64 : let existed = self.purge_kinds(tenant, &ALL_KINDS);
539 : // ADR-0021: the Tenant's Hosted and ImplicitlyCreated @contexts are
540 : // its documents and go with it — a row that outlived its Tenant would
541 : // hand the next holder of that name a stranger's term mappings. A
542 : // Cached row belongs to no Tenant and stays.
543 64 : on_blocking(true, || {
544 64 : let mut inner = self.write_inner();
545 64 : let dead: Vec<String> = inner
546 64 : .contexts
547 64 : .iter()
548 64 : .filter(|(_, row)| context_row_owner(row) == Some(tenant.as_str()))
549 64 : .map(|(id, _)| id.clone())
550 64 : .collect();
551 64 : for id in dead {
552 4 : self.persist(T_JSONLD_CONTEXTS, id.as_bytes(), None);
553 4 : inner.contexts.remove(&id);
554 4 : }
555 64 : });
556 64 : existed
557 64 : }
558 :
559 666 : pub fn subscription_tenants(&self) -> Vec<String> {
560 666 : let inner = self
561 666 : .inner
562 666 : .read()
563 666 : .unwrap_or_else(std::sync::PoisonError::into_inner);
564 : // Registrations belong to this domain: one of the hydrations that
565 : // walks it fills the registration mirror, which the federation path
566 : // then reads alone. Without them a tenant holding registrations and
567 : // no subscription forwarded to no Context Source.
568 666 : let mut out: Vec<String> = inner
569 666 : .subscriptions
570 666 : .iter()
571 666 : .chain(inner.csource_subscriptions.iter())
572 666 : .chain(inner.registrations.iter())
573 666 : .filter(|(_, m)| !m.is_empty())
574 666 : .map(|(t, _)| t.clone())
575 666 : .collect();
576 666 : out.sort();
577 666 : out.dedup();
578 666 : out
579 666 : }
580 :
581 101370 : fn map(inner: &Inner, kind: Kind) -> &HashMap<String, BTreeMap<String, Value>> {
582 101370 : match kind {
583 57851 : Kind::Entity => &inner.entities,
584 997 : Kind::Subscription => &inner.subscriptions,
585 19031 : Kind::Registration => &inner.registrations,
586 1862 : Kind::CSourceSubscription => &inner.csource_subscriptions,
587 6440 : Kind::Temporal => &inner.temporal,
588 3018 : Kind::Snapshot => &inner.snapshots,
589 8971 : Kind::EntityMap => &inner.entity_map_docs,
590 2609 : Kind::DistSub => &inner.dist_subs,
591 591 : Kind::DeadLetter => &inner.dead_letters,
592 : }
593 101370 : }
594 :
595 65923 : fn map_mut(inner: &mut Inner, kind: Kind) -> &mut HashMap<String, BTreeMap<String, Value>> {
596 65923 : match kind {
597 39182 : Kind::Entity => &mut inner.entities,
598 1554 : Kind::Subscription => &mut inner.subscriptions,
599 2115 : Kind::Registration => &mut inner.registrations,
600 136 : Kind::CSourceSubscription => &mut inner.csource_subscriptions,
601 16068 : Kind::Temporal => &mut inner.temporal,
602 616 : Kind::Snapshot => &mut inner.snapshots,
603 4780 : Kind::EntityMap => &mut inner.entity_map_docs,
604 1324 : Kind::DistSub => &mut inner.dist_subs,
605 148 : Kind::DeadLetter => &mut inner.dead_letters,
606 : }
607 65923 : }
608 :
609 : /// 4.22: "expiresAt is defined as the system temporal Property at which a
610 : /// certain Entity, Property or Relationship shall become invalid." An
611 : /// entity past its expiry is invalid the moment the stamp passes, ahead
612 : /// of the sweep that physically reaps it, so every write path treats it
613 : /// as absent — otherwise the same id 404s on retrieve and 409s on create
614 : /// for a whole sweep interval. Only entities carry the entity-level
615 : /// stamp; subscriptions and registrations have their own expiry rules.
616 65305 : fn is_expired(&self, inner: &Inner, kind: Kind, tenant: &str, id: &str) -> bool {
617 65305 : kind == Kind::Entity
618 39134 : && Self::map(inner, kind)
619 39134 : .get(tenant)
620 39134 : .and_then(|m| m.get(id))
621 39134 : .is_some_and(|d| filter::expired_at(d, &now_stamp()))
622 65305 : }
623 :
624 : /// Insert a new resource; `false` if the id already exists.
625 27135 : pub async fn create(&self, tenant: &TenantId, kind: Kind, id: &str, doc: Value) -> bool {
626 27135 : let _order = match kind {
627 16516 : Kind::Entity => Some(self.emit_ordered().await),
628 10619 : _ => None,
629 : };
630 27135 : let created = on_blocking(self.shadow.is_some(), || {
631 27135 : let mut inner = self.write_inner();
632 27135 : let expired = self.is_expired(&inner, kind, tenant.as_str(), id);
633 27135 : let m = Self::map_mut(&mut inner, kind)
634 27135 : .entry(tenant.as_str().to_owned())
635 27135 : .or_default();
636 27135 : if m.contains_key(id) && !expired {
637 38 : false
638 : } else {
639 27097 : m.insert(id.to_owned(), doc.clone());
640 27097 : self.persist(table_for(kind), &key_bytes(tenant.as_str(), id), Some(&doc));
641 27097 : true
642 : }
643 27135 : });
644 27135 : if created && kind == Kind::Entity {
645 16498 : self.emit(tenant, None, Some(doc)).await;
646 10637 : }
647 27135 : created
648 27135 : }
649 :
650 : /// Insert or replace; returns `true` if it existed before. An expired
651 : /// entity did not (4.22), so the upsert reports CREATED and the caller
652 : /// answers 201 with a Location header instead of a silent 204.
653 12734 : pub async fn upsert(&self, tenant: &TenantId, kind: Kind, id: &str, doc: Value) -> bool {
654 12734 : let _order = match kind {
655 12676 : Kind::Entity => Some(self.emit_ordered().await),
656 58 : _ => None,
657 : };
658 12734 : let (prev, expired) = on_blocking(self.shadow.is_some(), || {
659 12734 : let mut inner = self.write_inner();
660 12734 : let expired = self.is_expired(&inner, kind, tenant.as_str(), id);
661 12734 : let prev = Self::map_mut(&mut inner, kind)
662 12734 : .entry(tenant.as_str().to_owned())
663 12734 : .or_default()
664 12734 : .insert(id.to_owned(), doc.clone());
665 12734 : self.persist(table_for(kind), &key_bytes(tenant.as_str(), id), Some(&doc));
666 12734 : (prev, expired)
667 12734 : });
668 12734 : let existed = prev.is_some() && !expired;
669 12734 : let prev = if expired { None } else { prev };
670 12734 : if kind == Kind::Entity {
671 12676 : self.emit(tenant, prev, Some(doc)).await;
672 58 : }
673 12732 : existed
674 12732 : }
675 :
676 29477 : pub fn get(&self, tenant: &TenantId, kind: Kind, id: &str) -> Option<Value> {
677 29477 : let inner = self
678 29477 : .inner
679 29477 : .read()
680 29477 : .unwrap_or_else(std::sync::PoisonError::into_inner);
681 29477 : Self::map(&inner, kind)
682 29477 : .get(tenant.as_str())
683 29477 : .and_then(|m| m.get(id))
684 29477 : .cloned()
685 29477 : }
686 :
687 : /// An expired entity is already invalid (4.22): deleting it is a 404,
688 : /// the same answer retrieving it gives, not a 204 for something the API
689 : /// stopped serving. The row itself still goes — the sweep would take it
690 : /// anyway, and leaving it would resurrect the 409.
691 10529 : pub async fn delete(&self, tenant: &TenantId, kind: Kind, id: &str) -> bool {
692 10529 : let _order = match kind {
693 4312 : Kind::Entity => Some(self.emit_ordered().await),
694 6217 : _ => None,
695 : };
696 10529 : let removed = on_blocking(self.shadow.is_some(), || {
697 10529 : let mut inner = self.write_inner();
698 : // an expired doc is left in place for the sweep to reap: removing
699 : // it here without persisting the removal would resurrect it on
700 : // the next boot of a `file`-mode store
701 10529 : if self.is_expired(&inner, kind, tenant.as_str(), id) {
702 10 : return None;
703 10519 : }
704 10519 : let removed = Self::map_mut(&mut inner, kind)
705 10519 : .get_mut(tenant.as_str())
706 10519 : .and_then(|m| m.remove(id));
707 10519 : if removed.is_some() {
708 4841 : self.persist(table_for(kind), &key_bytes(tenant.as_str(), id), None);
709 5819 : }
710 10519 : removed
711 10529 : });
712 10529 : let hit = removed.is_some();
713 10529 : if kind == Kind::Entity {
714 4312 : if let Some(old) = removed {
715 4130 : self.emit(tenant, Some(old), None).await;
716 182 : }
717 6217 : }
718 10529 : hit
719 10529 : }
720 :
721 : /// Delete one entity only if `keep` accepts the stored document. The
722 : /// read and the removal happen under one hold of the write lock, so
723 : /// nothing can replace the document between the decision and the delete.
724 : /// `None` = absent, expired (4.22, as in `delete`) or refused.
725 1164 : pub async fn delete_if(
726 1164 : &self,
727 1164 : tenant: &TenantId,
728 1164 : id: &str,
729 1164 : keep: &(dyn for<'v> Fn(&'v Value) -> bool + Sync),
730 1164 : ) -> Option<Value> {
731 1164 : let _order = self.emit_ordered().await;
732 1164 : let removed = on_blocking(self.shadow.is_some(), || {
733 1164 : let mut inner = self.write_inner();
734 1164 : if self.is_expired(&inner, Kind::Entity, tenant.as_str(), id) {
735 0 : return None;
736 1164 : }
737 1164 : let m = Self::map_mut(&mut inner, Kind::Entity).get_mut(tenant.as_str())?;
738 1126 : if !m.get(id).is_some_and(keep) {
739 886 : return None;
740 240 : }
741 240 : let removed = m.remove(id);
742 240 : if removed.is_some() {
743 240 : self.persist(
744 240 : table_for(Kind::Entity),
745 240 : &key_bytes(tenant.as_str(), id),
746 240 : None,
747 240 : );
748 240 : }
749 240 : removed
750 1164 : });
751 1164 : if let Some(old) = &removed {
752 240 : self.emit(tenant, Some(old.clone()), None).await;
753 924 : }
754 1164 : removed
755 1164 : }
756 :
757 : /// Snapshot of all docs of a kind for one tenant (id order).
758 21886 : pub fn list(&self, tenant: &TenantId, kind: Kind) -> Vec<Value> {
759 21886 : let inner = self
760 21886 : .inner
761 21886 : .read()
762 21886 : .unwrap_or_else(std::sync::PoisonError::into_inner);
763 21886 : Self::map(&inner, kind)
764 21886 : .get(tenant.as_str())
765 21886 : .map(|m| m.values().cloned().collect())
766 21886 : .unwrap_or_default()
767 21886 : }
768 :
769 : /// One id-ordered page of docs: ids strictly greater than `after`, at
770 : /// most `limit`. The per-tenant map is a `BTreeMap` keyed by id, so this
771 : /// is a range walk, not a scan-and-sort.
772 : ///
773 : /// 4.22 applies to entities inside the walk, BEFORE the page is cut: an
774 : /// expired document is not there to be paged, and dropping one after the
775 : /// cut would hand the caller a short page — which every walker reads as
776 : /// the end of the tenant.
777 6001 : pub fn list_page(
778 6001 : &self,
779 6001 : tenant: &TenantId,
780 6001 : kind: Kind,
781 6001 : after: Option<&str>,
782 6001 : limit: usize,
783 6001 : ) -> Vec<Value> {
784 : use std::ops::Bound;
785 6001 : let inner = self
786 6001 : .inner
787 6001 : .read()
788 6001 : .unwrap_or_else(std::sync::PoisonError::into_inner);
789 6001 : let lo = match after {
790 28 : Some(a) => Bound::Excluded(a.to_owned()),
791 5973 : None => Bound::Unbounded,
792 : };
793 6001 : let now = crate::store::any::now_utc();
794 6001 : Self::map(&inner, kind)
795 6001 : .get(tenant.as_str())
796 6001 : .map(|m| {
797 2288 : m.range((lo, Bound::Unbounded))
798 6944 : .filter_map(|(_, v)| {
799 6944 : let mut v = v.clone();
800 6944 : if kind == Kind::Entity && crate::store::filter::strip_expired(&mut v, &now)
801 : {
802 8 : return None;
803 6936 : }
804 6936 : Some(v)
805 6944 : })
806 2288 : .take(limit)
807 2288 : .collect()
808 2288 : })
809 6001 : .unwrap_or_default()
810 6001 : }
811 :
812 : /// One id-ordered window of docs and the size of the whole set. The
813 : /// per-tenant map is a `BTreeMap` keyed by id, so the window is a skip
814 : /// over an already-ordered iterator.
815 148 : pub fn list_slice(
816 148 : &self,
817 148 : tenant: &TenantId,
818 148 : kind: Kind,
819 148 : offset: usize,
820 148 : limit: usize,
821 148 : ) -> (Vec<Value>, usize) {
822 148 : let inner = self
823 148 : .inner
824 148 : .read()
825 148 : .unwrap_or_else(std::sync::PoisonError::into_inner);
826 148 : Self::map(&inner, kind)
827 148 : .get(tenant.as_str())
828 148 : .map(|m| {
829 80 : (
830 80 : m.values().skip(offset).take(limit).cloned().collect(),
831 80 : m.len(),
832 80 : )
833 80 : })
834 148 : .unwrap_or_default()
835 148 : }
836 :
837 : /// Read-modify-write on one document. Returns `None` when absent; the
838 : /// closure's error aborts without writing.
839 13743 : pub async fn mutate<T, E>(
840 13743 : &self,
841 13743 : tenant: &TenantId,
842 13743 : kind: Kind,
843 13743 : id: &str,
844 13743 : f: impl FnOnce(&mut Value) -> Result<T, E>,
845 13743 : ) -> Option<Result<T, E>> {
846 13743 : let _order = match kind {
847 4466 : Kind::Entity => Some(self.emit_ordered().await),
848 9277 : _ => None,
849 : };
850 13743 : let (result, change) = on_blocking(self.shadow.is_some(), || {
851 13743 : let mut inner = self.write_inner();
852 13743 : if self.is_expired(&inner, kind, tenant.as_str(), id) {
853 6 : return None; // 4.22: invalid, so absent
854 13737 : }
855 13737 : let doc = Self::map_mut(&mut inner, kind)
856 13737 : .get_mut(tenant.as_str())
857 13737 : .and_then(|m| m.get_mut(id))?;
858 10901 : let before = doc.clone();
859 10901 : let mut candidate = doc.clone();
860 10901 : Some(match f(&mut candidate) {
861 10847 : Ok(t) => {
862 10847 : if candidate != before {
863 10783 : self.persist(
864 10783 : table_for(kind),
865 10783 : &key_bytes(tenant.as_str(), id),
866 10783 : Some(&candidate),
867 10783 : );
868 10783 : }
869 10847 : let change = (kind == Kind::Entity && candidate != before)
870 10847 : .then(|| (before, candidate.clone()));
871 10847 : *doc = candidate;
872 10847 : (Ok(t), change)
873 : }
874 54 : Err(e) => (Err(e), None),
875 : })
876 13743 : })?;
877 10901 : if let Some((b, a)) = change {
878 4314 : self.emit(tenant, Some(b), Some(a)).await;
879 6587 : }
880 10901 : Some(result)
881 13743 : }
882 :
883 : // jsonldContexts: one keyspace for the whole process (key = context id, no
884 : // tenant prefix) — Cached rows are copies of public documents shared by
885 : // every tenant. Ownership of the tenant-authored kinds (Hosted,
886 : // ImplicitlyCreated, 5.13.1) travels in the document's "owner" member,
887 : // and every call below is answered through it (ADR-0021): the Postgres
888 : // arm has the same rule as a Row-Level Security policy over a generated
889 : // `tenant_id` column, and this arm, which has no policy engine under it,
890 : // applies it in the store.
891 2500 : pub fn context_put(
892 2500 : &self,
893 2500 : tenant: Option<&TenantId>,
894 2500 : id: &str,
895 2500 : doc: Value,
896 2500 : ) -> Result<(), NgsiError> {
897 2500 : on_blocking(self.shadow.is_some(), || {
898 2500 : let mut inner = self.write_inner();
899 : // The row that is there decides whether this call may replace it,
900 : // the row that arrives decides whether it may be stored: the two
901 : // halves of the policy the Postgres arm writes as USING and WITH
902 : // CHECK. Both are backstops — every caller mints its own id.
903 2500 : let held = inner.contexts.get(id);
904 2500 : if !held.is_none_or(|r| context_row_visible(r, tenant))
905 2498 : || !context_row_visible(&doc, tenant)
906 : {
907 2 : return Err(NgsiError::InternalError(
908 2 : "@context belongs to another tenant".into(),
909 2 : ));
910 2498 : }
911 2498 : self.persist(T_JSONLD_CONTEXTS, id.as_bytes(), Some(&doc));
912 2498 : let cached = doc.get("kind").and_then(Value::as_str) == Some("Cached");
913 2498 : inner.contexts.insert(id.to_owned(), doc);
914 : // 5.13.1: "Implementations shall periodically invalidate the
915 : // 'Cached' @contexts." One entry is stored per distinct external
916 : // URL a request references, so without a ceiling a client that
917 : // references fresh URLs grows this keyspace (and, in `file` mode,
918 : // the store on disk) forever. Oldest-first, and only for the
919 : // Cached kind — Hosted/ImplicitlyCreated entries are resources
920 : // the broker serves on demand, never a cache.
921 2498 : if cached && inner.contexts.len() > MAX_CACHED_CONTEXTS {
922 4 : let dead = oldest_cached(&inner.contexts);
923 4 : for id in dead {
924 2 : self.persist(T_JSONLD_CONTEXTS, id.as_bytes(), None);
925 2 : inner.contexts.remove(&id);
926 2 : }
927 2494 : }
928 2498 : Ok(())
929 2500 : })
930 2500 : }
931 :
932 884 : pub fn context_get(&self, tenant: Option<&TenantId>, id: &str) -> Option<Value> {
933 884 : self.inner
934 884 : .read()
935 884 : .unwrap_or_else(std::sync::PoisonError::into_inner)
936 884 : .contexts
937 884 : .get(id)
938 884 : .filter(|r| context_row_visible(r, tenant))
939 884 : .cloned()
940 884 : }
941 :
942 128 : pub fn context_delete(&self, tenant: Option<&TenantId>, id: &str) -> bool {
943 128 : on_blocking(self.shadow.is_some(), || {
944 128 : let mut inner = self.write_inner();
945 128 : let hit = inner
946 128 : .contexts
947 128 : .get(id)
948 128 : .is_some_and(|r| context_row_visible(r, tenant));
949 128 : if hit {
950 122 : inner.contexts.remove(id);
951 122 : self.persist(T_JSONLD_CONTEXTS, id.as_bytes(), None);
952 122 : }
953 128 : hit
954 128 : })
955 128 : }
956 :
957 : /// Every row without its `body` member (the `@context` document itself).
958 : /// A body may be 5 MiB and only the `Cached` rows are capped in number,
959 : /// so cloning whole rows here was a multi-gigabyte copy on the boot path.
960 158 : pub fn context_list_meta(&self, tenant: Option<&TenantId>) -> Vec<Value> {
961 158 : self.inner
962 158 : .read()
963 158 : .unwrap_or_else(std::sync::PoisonError::into_inner)
964 158 : .contexts
965 158 : .values()
966 4302 : .filter(|v| context_row_visible(v, tenant))
967 4220 : .map(|v| {
968 4220 : let mut row = v.clone();
969 4220 : if let Some(o) = row.as_object_mut() {
970 4220 : o.remove("body");
971 4220 : }
972 4220 : row
973 4220 : })
974 158 : .collect()
975 158 : }
976 : }
977 :
978 : #[cfg(test)]
979 : mod tests {
980 : use super::*;
981 : use serde_json::json;
982 :
983 : /// A change hook that PANICS must cost only its own caller: the panic
984 : /// unwinds through that writer, and every later writer finds the store
985 : /// usable — locks recover from the poisoning instead of turning one bad
986 : /// hook into a dead store.
987 : #[tokio::test(flavor = "multi_thread")]
988 2 : async fn a_panicking_hook_leaves_the_store_usable() {
989 : use std::sync::Arc;
990 2 : let s = Arc::new(Store::default());
991 2 : s.set_change_hook(Arc::new(
992 : |_t: &TenantId,
993 : _b: Option<Value>,
994 : after: Option<Value>|
995 4 : -> antares_store::HookFuture<'_> {
996 4 : Box::pin(async move {
997 4 : if after.as_ref().and_then(|a| a.get("boom")).is_some() {
998 2 : panic!("hook panic");
999 2 : }
1000 2 : })
1001 4 : },
1002 : ));
1003 2 : let t = TenantId::new("hook-panic").expect("tenant");
1004 2 : let s2 = s.clone();
1005 2 : let t2 = t.clone();
1006 2 : let poisoned = tokio::spawn(async move {
1007 2 : s2.upsert(
1008 2 : &t2,
1009 2 : Kind::Entity,
1010 2 : "urn:x:boom",
1011 2 : json!({"id": "urn:x:boom", "type": ["T"], "boom": true}),
1012 2 : )
1013 2 : .await;
1014 2 : })
1015 2 : .await;
1016 2 : assert!(poisoned.is_err(), "the hook's panic reaches its own caller");
1017 : // the NEXT writer and reader are untouched
1018 2 : assert!(
1019 2 : s.create(
1020 2 : &t,
1021 2 : Kind::Entity,
1022 2 : "urn:x:after",
1023 2 : json!({"id": "urn:x:after", "type": ["T"]})
1024 2 : )
1025 2 : .await
1026 : );
1027 2 : assert!(
1028 2 : s.get(&t, Kind::Entity, "urn:x:after").is_some(),
1029 : "one panicking hook must not take the store down"
1030 : );
1031 2 : assert!(
1032 2 : s.get(&t, Kind::Entity, "urn:x:boom").is_some(),
1033 2 : "the write that triggered the panic still committed — the hook runs after the commit"
1034 2 : );
1035 2 : }
1036 :
1037 : /// Two writers to the same entity: the change hook fires in COMMIT
1038 : /// order. The first writer's hook is gated open only after the second
1039 : /// writer has had every chance to overtake it — if the second commit may
1040 : /// emit before the first commit's emit, the notification pipeline sees
1041 : /// the versions reversed and records stale state as newest.
1042 : #[tokio::test(flavor = "multi_thread")]
1043 2 : async fn change_hook_fires_in_commit_order_per_entity() {
1044 : use std::sync::atomic::{AtomicBool, Ordering};
1045 : use std::sync::Arc;
1046 2 : let s = Arc::new(Store::default());
1047 2 : let seen: Arc<std::sync::Mutex<Vec<i64>>> = Arc::default();
1048 2 : let entered = Arc::new(AtomicBool::new(false));
1049 2 : let released = Arc::new(AtomicBool::new(false));
1050 : {
1051 2 : let (seen, entered, released) = (seen.clone(), entered.clone(), released.clone());
1052 2 : s.set_change_hook(Arc::new(
1053 : move |_t: &TenantId,
1054 : _b: Option<Value>,
1055 : after: Option<Value>|
1056 4 : -> antares_store::HookFuture<'_> {
1057 4 : let (seen, entered, released) =
1058 4 : (seen.clone(), entered.clone(), released.clone());
1059 4 : Box::pin(async move {
1060 4 : let v = after
1061 4 : .as_ref()
1062 4 : .and_then(|a| a.get("v"))
1063 4 : .and_then(Value::as_i64)
1064 4 : .unwrap_or(-1);
1065 4 : if v == 1 {
1066 2 : entered.store(true, Ordering::SeqCst);
1067 63 : while !released.load(Ordering::SeqCst) {
1068 61 : tokio::time::sleep(std::time::Duration::from_millis(1)).await;
1069 : }
1070 2 : }
1071 4 : seen.lock()
1072 4 : .unwrap_or_else(std::sync::PoisonError::into_inner)
1073 4 : .push(v);
1074 4 : })
1075 4 : },
1076 : ));
1077 : }
1078 2 : let t = TenantId::new("emit-order").expect("tenant");
1079 4 : let doc = |v: i64| json!({"id": "urn:x:1", "type": ["T"], "v": v});
1080 2 : let s1 = s.clone();
1081 2 : let t1c = t.clone();
1082 2 : let w1 = tokio::spawn(async move {
1083 2 : s1.upsert(&t1c, Kind::Entity, "urn:x:1", doc(1)).await;
1084 2 : });
1085 4 : while !entered.load(std::sync::atomic::Ordering::SeqCst) {
1086 2 : tokio::time::sleep(std::time::Duration::from_millis(1)).await;
1087 : }
1088 : // writer 1 has committed and sits inside its hook. Writer 2 now has
1089 : // every chance to commit AND emit before writer 1's emit finishes.
1090 2 : let s2 = s.clone();
1091 2 : let t2c = t.clone();
1092 2 : let w2 = tokio::spawn(async move {
1093 2 : s2.upsert(&t2c, Kind::Entity, "urn:x:1", doc(2)).await;
1094 2 : });
1095 2 : tokio::time::sleep(std::time::Duration::from_millis(60)).await;
1096 2 : released.store(true, std::sync::atomic::Ordering::SeqCst);
1097 2 : w1.await.expect("writer 1");
1098 2 : w2.await.expect("writer 2");
1099 2 : assert_eq!(
1100 2 : *seen
1101 2 : .lock()
1102 2 : .unwrap_or_else(std::sync::PoisonError::into_inner),
1103 2 : vec![1, 2],
1104 2 : "hooks must fire in commit order, or the consumer records stale state as newest"
1105 2 : );
1106 2 : }
1107 :
1108 : #[tokio::test]
1109 2 : async fn commit_queue_counts_writers() {
1110 : // Every write passes through the counted critical section.
1111 2 : let s = Store::default();
1112 2 : assert_eq!(s.commit_queue(), (0, 0));
1113 2 : let t = TenantId::new("t").unwrap();
1114 2 : s.create(&t, Kind::Entity, "urn:a", json!({"id": "urn:a"}))
1115 2 : .await;
1116 2 : let (depth, peak) = s.commit_queue();
1117 2 : assert_eq!(depth, 0, "no writer in flight after the call returns");
1118 2 : assert!(peak >= 1, "the write itself must register in the peak");
1119 2 : }
1120 :
1121 : /// What a store reports about itself, for an operator reading
1122 : /// `/q/health`: the memory and file modes are one backend with two
1123 : /// durability shapes, and the health body must not present them as the
1124 : /// same thing.
1125 : #[test]
1126 2 : fn a_store_reports_the_engine_it_actually_runs() {
1127 2 : let dir = tempdir("engine");
1128 2 : let mem = crate::store::any::AnyStore::Mem(Store::default());
1129 2 : let file = crate::store::any::AnyStore::Mem(Store::open_file(&dir).expect("open"));
1130 2 : assert_eq!(mem.version_info()["engine"], "memory");
1131 2 : assert_eq!(file.version_info()["engine"], "redb");
1132 2 : }
1133 :
1134 : /// The commit queue is a `file`-mode signal: it exists because redb has
1135 : /// one writer and commits fsync through it. A pure in-memory store has no
1136 : /// such committer, so it reports nothing rather than a number that would
1137 : /// read as the same thing and mean something else.
1138 : #[test]
1139 2 : fn only_a_shadowed_store_reports_a_commit_queue() {
1140 : use crate::store::any::AnyStore;
1141 2 : assert_eq!(AnyStore::Mem(Store::default()).commit_queue(), None);
1142 2 : let dir = tempdir("commit-queue");
1143 2 : let s = Store::open_file(&dir).expect("open");
1144 2 : assert!(
1145 2 : AnyStore::Mem(s).commit_queue().is_some(),
1146 : "a durable store reports the queue behind its single committer"
1147 : );
1148 2 : }
1149 :
1150 : #[tokio::test]
1151 2 : async fn tenant_isolation() {
1152 2 : let s = Store::default();
1153 2 : let t1 = TenantId::new("t1").unwrap();
1154 2 : let t2 = TenantId::new("t2").unwrap();
1155 2 : assert!(
1156 2 : s.create(&t1, Kind::Entity, "urn:a", json!({"id": "urn:a"}))
1157 2 : .await
1158 : );
1159 2 : assert!(s.get(&t2, Kind::Entity, "urn:a").is_none());
1160 2 : assert!(s.get(&t1, Kind::Entity, "urn:a").is_some());
1161 2 : assert!(!s.create(&t1, Kind::Entity, "urn:a", json!({})).await);
1162 2 : assert!(s.create(&t2, Kind::Entity, "urn:a", json!({})).await);
1163 2 : }
1164 :
1165 : /// 4.6.3 allows a comma as the seconds-fraction separator. Byte order
1166 : /// puts ',' (0x2C) before both '.' and 'Z', so a comma-form instance that
1167 : /// has NOT expired reads as expired against the point-form `now` this
1168 : /// store stamps — the write path would drop a live instance, and the
1169 : /// postgres path (which parses through `try_timestamptz`) would keep it.
1170 : #[tokio::test]
1171 2 : async fn a_comma_fraction_instance_is_not_pruned_before_it_expires() {
1172 2 : let now = "2026-09-01T12:00:00.000Z";
1173 2 : let mut live = json!({
1174 2 : "id": "urn:x", "type": ["T"],
1175 2 : "https://a/attr": [{"value": 1, "expiresAt": "2026-09-01T12:00:00,500Z"}]
1176 : });
1177 2 : assert!(
1178 2 : !prune_expired_instances(&mut live, now),
1179 : "live instance kept"
1180 : );
1181 2 : assert_eq!(live["https://a/attr"].as_array().map(Vec::len), Some(1));
1182 :
1183 : // one that HAS expired still goes, comma form included, and takes the
1184 : // attribute with it when it was the last instance
1185 2 : let mut gone = json!({
1186 2 : "id": "urn:x", "type": ["T"],
1187 2 : "https://a/attr": [{"value": 1, "expiresAt": "2026-09-01T11:59:59,500Z"}]
1188 : });
1189 2 : assert!(prune_expired_instances(&mut gone, now));
1190 2 : assert!(gone.get("https://a/attr").is_none());
1191 :
1192 : // the entity-level sweep reads the same stamp the same way
1193 2 : let s = Store::default();
1194 2 : let t = TenantId::new("t").expect("tenant");
1195 2 : assert!(
1196 2 : s.create(
1197 2 : &t,
1198 2 : Kind::Entity,
1199 2 : "urn:live",
1200 2 : json!({"id": "urn:live", "type": ["T"], "expiresAt": "2026-09-01T12:00:00,500Z"})
1201 2 : )
1202 2 : .await
1203 : );
1204 2 : assert!(
1205 2 : s.create(
1206 2 : &t,
1207 2 : Kind::Entity,
1208 2 : "urn:gone",
1209 2 : json!({"id": "urn:gone", "type": ["T"], "expiresAt": "2026-09-01T11:59:59,500Z"})
1210 2 : )
1211 2 : .await
1212 : );
1213 2 : assert_eq!(s.sweep_expired(now), 1, "only the expired one is reaped");
1214 2 : assert!(s.get(&t, Kind::Entity, "urn:live").is_some());
1215 2 : assert!(s.get(&t, Kind::Entity, "urn:gone").is_none());
1216 2 : }
1217 :
1218 : /// The same assertion over EVERY kind and the whole read/write surface,
1219 : /// with the SAME id on both sides — the shape a tenant-blind map gets
1220 : /// wrong, because it answers the neighbour's document instead of `None`.
1221 : #[tokio::test]
1222 2 : async fn no_kind_leaks_across_tenants() {
1223 2 : let s = Store::default();
1224 2 : let a = TenantId::new("t-a").expect("tenant");
1225 2 : let b = TenantId::new("t-b").expect("tenant");
1226 2 : let id = "urn:ngsi-ld:shared:1";
1227 18 : for kind in ALL_KINDS {
1228 18 : assert!(
1229 18 : s.create(&a, kind, id, json!({"id": id, "owner": "a"}))
1230 18 : .await
1231 2 : );
1232 2 :
1233 18 : assert!(s.get(&b, kind, id).is_none(), "{kind:?}: get");
1234 18 : assert!(s.list(&b, kind).is_empty(), "{kind:?}: list");
1235 18 : assert!(
1236 18 : s.list_page(&b, kind, None, 100).is_empty(),
1237 2 : "{kind:?}: list_page"
1238 2 : );
1239 18 : assert_eq!(
1240 18 : s.list_slice(&b, kind, 0, 100),
1241 18 : (Vec::new(), 0),
1242 2 : "{kind:?}: list_slice"
1243 2 : );
1244 2 :
1245 2 : // no write reaches the neighbour's document either
1246 18 : assert!(!s.delete(&b, kind, id).await, "{kind:?}: delete");
1247 18 : assert!(
1248 18 : s.mutate::<(), ()>(&b, kind, id, |d| {
1249 2 : d["owner"] = json!("b");
1250 2 : Ok(())
1251 2 : })
1252 18 : .await
1253 18 : .is_none(),
1254 2 : "{kind:?}: mutate"
1255 2 : );
1256 2 :
1257 2 : // the same id under another tenant is a create, not a conflict
1258 18 : assert!(
1259 18 : s.create(&b, kind, id, json!({"id": id, "owner": "b"}))
1260 18 : .await
1261 2 : );
1262 18 : assert_eq!(s.get(&a, kind, id).expect("a keeps its own")["owner"], "a");
1263 18 : assert_eq!(s.get(&b, kind, id).expect("b has its own")["owner"], "b");
1264 2 :
1265 2 : // and b deleting its own leaves a's untouched
1266 18 : assert!(s.delete(&b, kind, id).await, "{kind:?}: delete own");
1267 18 : assert!(s.get(&b, kind, id).is_none(), "{kind:?}: deleted");
1268 18 : assert_eq!(
1269 18 : s.get(&a, kind, id)
1270 18 : .expect("survives the neighbour's delete")["owner"],
1271 2 : "a"
1272 2 : );
1273 18 : assert!(s.delete(&a, kind, id).await, "{kind:?}: cleanup");
1274 2 : }
1275 2 : }
1276 :
1277 26 : fn tempdir(name: &str) -> std::path::PathBuf {
1278 26 : let dir = std::env::temp_dir().join(format!("antares-store-{name}-{}", std::process::id()));
1279 26 : let _ = std::fs::remove_dir_all(&dir);
1280 26 : std::fs::create_dir_all(&dir).expect("tempdir");
1281 26 : dir
1282 26 : }
1283 :
1284 : /// `open_file` promises "Any open/format/decode error refuses to start —
1285 : /// never silently serve partial data", and the boot rebuild's kind loop
1286 : /// keeps it: an absent table has nothing to load, any other error refuses.
1287 : /// The @context table is read as `if let Ok(...)` instead, so a table that
1288 : /// exists and cannot be opened counts as an empty one — and that is the
1289 : /// table whose absence costs a Tenant its own documents, since Hosted and
1290 : /// ImplicitlyCreated @contexts (5.13.1) hold term mappings authored
1291 : /// through its requests. The broker would come up and serve without them.
1292 : ///
1293 : /// A table of the same name under different key/value types is redb's
1294 : /// `TableTypeMismatch`: the shape a file written by another binary, or by
1295 : /// an older schema, actually has.
1296 : #[test]
1297 2 : fn an_unreadable_table_refuses_the_boot_instead_of_reading_as_empty() {
1298 2 : let dir = tempdir("unreadable-context-table");
1299 2 : let path = dir.join("antares.redb");
1300 2 : {
1301 2 : let db = Database::create(&path).expect("db");
1302 2 : let mut tx = db.begin_write().expect("tx");
1303 2 : tx.set_durability(Durability::Immediate).expect("dur");
1304 2 : {
1305 2 : let mut m = tx.open_table(T_META).expect("meta");
1306 2 : m.insert("format", FORMAT_VERSION).expect("insert");
1307 2 : }
1308 2 : {
1309 2 : let def = ::redb::TableDefinition::<&str, &str>::new("jsonld_contexts");
1310 2 : let mut t = tx.open_table(def).expect("mismatched contexts table");
1311 2 : t.insert("urn:ngsi-ld:ctx", "{}").expect("insert");
1312 2 : }
1313 2 : tx.commit().expect("commit");
1314 2 : }
1315 2 : let db = Database::create(&path).expect("reopen");
1316 2 : let Err(err) = Store::from_database(db, &path.display().to_string()) else {
1317 0 : panic!("a table that cannot be opened must refuse the boot, not start without it");
1318 : };
1319 2 : assert!(!err.is_empty(), "the refusal says what failed");
1320 2 : }
1321 :
1322 : /// The same read, in the guard that exists to refuse a file this binary
1323 : /// cannot vouch for: data present with no format marker. Skipping a table
1324 : /// it cannot open makes the guard conclude the file is empty and stamp
1325 : /// this binary's marker onto it — a write into a file it is about to
1326 : /// declare unreadable, and one that costs the operator the guard's own
1327 : /// diagnosis on every later start.
1328 : #[test]
1329 2 : fn the_marker_guard_does_not_stamp_a_file_it_could_not_read() {
1330 2 : let dir = tempdir("unstamped-unreadable");
1331 2 : let path = dir.join("antares.redb");
1332 2 : {
1333 2 : let db = Database::create(&path).expect("db");
1334 2 : let mut tx = db.begin_write().expect("tx");
1335 2 : tx.set_durability(Durability::Immediate).expect("dur");
1336 2 : let def = ::redb::TableDefinition::<&str, &str>::new("entities");
1337 2 : let mut t = tx.open_table(def).expect("mismatched entities table");
1338 2 : t.insert("k", "v").expect("insert");
1339 2 : drop(t);
1340 2 : tx.commit().expect("commit");
1341 2 : }
1342 2 : let db = Database::create(&path).expect("reopen");
1343 2 : let Err(_) = Store::from_database(db, &path.display().to_string()) else {
1344 0 : panic!("an unreadable table with no marker must refuse the boot");
1345 : };
1346 2 : let db = Database::create(&path).expect("reopen after the refusal");
1347 2 : let rt = db.begin_read().expect("read tx");
1348 2 : let stamped = match rt.open_table(T_META) {
1349 0 : Ok(m) => m.get("format").expect("get").is_some(),
1350 2 : Err(::redb::TableError::TableDoesNotExist(_)) => false,
1351 0 : Err(e) => panic!("{e}"),
1352 : };
1353 2 : assert!(
1354 2 : !stamped,
1355 : "a file the guard could not read must not carry this binary's format marker"
1356 : );
1357 2 : }
1358 :
1359 : /// Every kind + contexts round-trip through a close/reopen.
1360 : #[tokio::test]
1361 2 : async fn file_mode_survives_reopen() {
1362 2 : let dir = tempdir("reopen");
1363 2 : let t = TenantId::new("tenant_a-1").expect("tenant");
1364 : {
1365 2 : let s = Store::open_file(&dir).expect("open");
1366 18 : for kind in ALL_KINDS {
1367 18 : assert!(
1368 18 : s.create(&t, kind, "urn:x:1", json!({"kind": format!("{kind:?}")}))
1369 18 : .await
1370 : );
1371 : }
1372 : // a row the writers actually produce: the kind is what decides
1373 : // whose it is (ADR-0021), and a Cached copy belongs to no Tenant
1374 2 : s.context_put(
1375 2 : None,
1376 2 : "ctx1",
1377 2 : json!({"localId": "ctx1", "kind": "Cached", "body": {"@context": {}}}),
1378 : )
1379 2 : .expect("store");
1380 : }
1381 2 : let s = Store::open_file(&dir).expect("reopen");
1382 18 : for kind in ALL_KINDS {
1383 18 : assert_eq!(
1384 18 : s.get(&t, kind, "urn:x:1").expect("survives")["kind"],
1385 18 : format!("{kind:?}")
1386 : );
1387 : }
1388 2 : assert!(s.context_get(None, "ctx1").is_some());
1389 : // tenant isolation intact after rebuild
1390 2 : assert!(s
1391 2 : .get(&TenantId::default(), Kind::Entity, "urn:x:1")
1392 2 : .is_none());
1393 2 : let _ = std::fs::remove_dir_all(&dir);
1394 2 : }
1395 :
1396 : /// Deletes and mutations reach redb — no phantom state after restart.
1397 : #[tokio::test]
1398 2 : async fn file_mode_deletes_and_updates_persist() {
1399 2 : let dir = tempdir("delete");
1400 2 : let t = TenantId::default();
1401 : {
1402 2 : let s = Store::open_file(&dir).expect("open");
1403 2 : s.create(&t, Kind::Entity, "urn:gone", json!({"n": 1}))
1404 2 : .await;
1405 2 : s.create(&t, Kind::Entity, "urn:kept", json!({"n": 1}))
1406 2 : .await;
1407 2 : assert!(s.delete(&t, Kind::Entity, "urn:gone").await);
1408 2 : let r: Option<Result<(), ()>> = s
1409 2 : .mutate(&t, Kind::Entity, "urn:kept", |d| {
1410 2 : d["n"] = json!(2);
1411 2 : Ok(())
1412 2 : })
1413 2 : .await;
1414 2 : assert!(matches!(r, Some(Ok(()))));
1415 2 : s.context_put(None, "ctx", json!({"localId": "ctx", "kind": "Cached"}))
1416 2 : .expect("store");
1417 2 : assert!(s.context_delete(None, "ctx"));
1418 : }
1419 2 : let s = Store::open_file(&dir).expect("reopen");
1420 2 : assert!(
1421 2 : s.get(&t, Kind::Entity, "urn:gone").is_none(),
1422 : "phantom 409 trap"
1423 : );
1424 2 : assert_eq!(s.get(&t, Kind::Entity, "urn:kept").expect("kept")["n"], 2);
1425 2 : assert!(s.context_get(None, "ctx").is_none());
1426 2 : let _ = std::fs::remove_dir_all(&dir);
1427 2 : }
1428 :
1429 : /// A future-format file refuses to load with a clear message.
1430 : #[test]
1431 2 : fn file_mode_refuses_format_mismatch() {
1432 2 : let dir = tempdir("format");
1433 2 : {
1434 2 : let db = Database::create(dir.join("antares.redb")).expect("db");
1435 2 : let mut tx = db.begin_write().expect("tx");
1436 2 : tx.set_durability(Durability::Immediate).expect("dur");
1437 2 : tx.open_table(T_META)
1438 2 : .expect("meta")
1439 2 : .insert("format", "999")
1440 2 : .expect("insert");
1441 2 : tx.commit().expect("commit");
1442 2 : }
1443 2 : let err = match Store::open_file(&dir) {
1444 2 : Err(e) => e,
1445 0 : Ok(_) => panic!("must refuse a format-999 file"),
1446 : };
1447 2 : assert!(err.contains("format 999"), "err: {err}");
1448 2 : let _ = std::fs::remove_dir_all(&dir);
1449 2 : }
1450 :
1451 : /// A file that HOLDS rows but carries no format marker was written by a
1452 : /// binary whose key/value shape this one cannot vouch for. Refusing is
1453 : /// the whole point of the marker — serving it would answer requests from
1454 : /// data that may be misread. An empty file, by contrast, is just a fresh
1455 : /// one and gets the marker stamped.
1456 : #[test]
1457 2 : fn file_mode_refuses_data_without_a_format_marker() {
1458 2 : let dir = tempdir("nomarker");
1459 2 : {
1460 2 : let db = Database::create(dir.join("antares.redb")).expect("db");
1461 2 : let mut tx = db.begin_write().expect("tx");
1462 2 : tx.set_durability(Durability::Immediate).expect("dur");
1463 2 : {
1464 2 : let mut t = tx.open_table(T_ENTITIES).expect("entities");
1465 2 : let mut k = b"plain".to_vec();
1466 2 : k.push(0);
1467 2 : k.extend_from_slice(b"urn:e:1");
1468 2 : let bytes = serde_json::to_vec(&json!({"id": "urn:e:1"})).expect("serialize");
1469 2 : t.insert(k.as_slice(), bytes.as_slice()).expect("insert");
1470 2 : }
1471 2 : // deliberately NO meta table
1472 2 : tx.commit().expect("commit");
1473 2 : }
1474 2 : let err = match Store::open_file(&dir) {
1475 2 : Err(e) => e,
1476 0 : Ok(_) => panic!("must refuse data with no format marker"),
1477 : };
1478 2 : assert!(err.contains("no format marker"), "err: {err}");
1479 2 : assert!(err.contains("antares.redb"), "the file is named: {err}");
1480 2 : let _ = std::fs::remove_dir_all(&dir);
1481 :
1482 : // an EMPTY unmarked file is a fresh store, not a refusal
1483 2 : let fresh = tempdir("nomarker-empty");
1484 2 : {
1485 2 : let db = Database::create(fresh.join("antares.redb")).expect("db");
1486 2 : let tx = db.begin_write().expect("tx");
1487 2 : tx.commit().expect("commit");
1488 2 : }
1489 2 : Store::open_file(&fresh).expect("an empty file is a fresh store");
1490 2 : let _ = std::fs::remove_dir_all(&fresh);
1491 2 : }
1492 :
1493 : /// 4.22: "expiresAt is defined as the system temporal Property at which a
1494 : /// certain Entity, Property or Relationship shall become invalid." The
1495 : /// clause sanctions the DELETION lagging, not the invalidity — so the
1496 : /// write paths must agree with the read boundary the instant the stamp
1497 : /// passes. Before this, the same id was simultaneously a 404 on retrieve,
1498 : /// a 409 on create and a 204 on delete for a whole sweep interval.
1499 : #[tokio::test]
1500 2 : async fn an_expired_entity_is_absent_to_writes_too() {
1501 2 : let s = Store::default();
1502 2 : let t = TenantId::default();
1503 2 : let dead = json!({"id": "urn:e", "type": ["T"], "expiresAt": "2000-01-01T00:00:00Z"});
1504 2 : let live = json!({"id": "urn:l", "type": ["T"], "expiresAt": "2999-01-01T00:00:00Z"});
1505 2 : assert!(s.create(&t, Kind::Entity, "urn:e", dead.clone()).await);
1506 2 : assert!(s.create(&t, Kind::Entity, "urn:l", live.clone()).await);
1507 :
1508 : // patching or deleting something already invalid is a 404, not a 204
1509 2 : assert!(s
1510 2 : .mutate(&t, Kind::Entity, "urn:e", |_d| Ok::<(), ()>(()))
1511 2 : .await
1512 2 : .is_none());
1513 2 : assert!(
1514 2 : !s.delete(&t, Kind::Entity, "urn:e").await,
1515 : "expired delete is 404"
1516 : );
1517 : // …and creating over it succeeds instead of raising AlreadyExists
1518 2 : assert!(
1519 2 : s.create(&t, Kind::Entity, "urn:e", json!({"id": "urn:e", "n": 1}))
1520 2 : .await,
1521 : "an expired id must not 409 a create"
1522 : );
1523 2 : assert_eq!(
1524 2 : s.get(&t, Kind::Entity, "urn:e").expect("recreated")["n"],
1525 : 1,
1526 : "the create must have replaced the expired document"
1527 : );
1528 :
1529 : // an UNEXPIRED entity keeps every one of those answers
1530 2 : assert!(
1531 2 : !s.create(&t, Kind::Entity, "urn:l", live.clone()).await,
1532 : "409"
1533 : );
1534 2 : assert!(s
1535 2 : .mutate(&t, Kind::Entity, "urn:l", |_d| Ok::<(), ()>(()))
1536 2 : .await
1537 2 : .is_some());
1538 2 : assert!(s.delete(&t, Kind::Entity, "urn:l").await);
1539 :
1540 : // upsert over an expired id reports CREATED (201 + Location), not
1541 : // updated
1542 2 : s.create(&t, Kind::Entity, "urn:x", dead.clone()).await;
1543 2 : assert!(
1544 2 : !s.upsert(&t, Kind::Entity, "urn:x", json!({"id": "urn:x"}))
1545 2 : .await,
1546 : "an expired id must upsert as created"
1547 : );
1548 2 : assert!(
1549 2 : s.upsert(&t, Kind::Entity, "urn:x", json!({"id": "urn:x", "n": 2}))
1550 2 : .await,
1551 : "and the live one that replaced it as updated"
1552 : );
1553 :
1554 : // 4.22 is an ENTITY stamp: other kinds keep their own expiry rules
1555 2 : assert!(
1556 2 : s.create(&t, Kind::Subscription, "urn:s", dead.clone())
1557 2 : .await
1558 : );
1559 2 : assert!(
1560 2 : !s.create(&t, Kind::Subscription, "urn:s", dead).await,
1561 : "a subscription id still 409s"
1562 : );
1563 2 : assert!(s.delete(&t, Kind::Subscription, "urn:s").await);
1564 2 : }
1565 :
1566 : /// The change hook drives every notification and all temporal
1567 : /// auto-recording, so its contract is: create emits (None, Some), delete
1568 : /// emits (Some, None), a real mutate emits both images — and, just as
1569 : /// load-bearing, a NON-entity write and a no-op mutate emit NOTHING. A
1570 : /// subscription leaking into the hook would be mirrored into temporal
1571 : /// storage; a no-op emitting would re-notify every subscriber on every
1572 : /// idempotent PATCH.
1573 : #[tokio::test]
1574 2 : async fn the_change_hook_fires_for_entity_changes_only() {
1575 : use std::sync::{Arc, Mutex};
1576 : type Images = (Option<Value>, Option<Value>);
1577 2 : let seen: Arc<Mutex<Vec<Images>>> = Arc::default();
1578 2 : let s = Store::default();
1579 2 : let t = TenantId::default();
1580 2 : let rec = Arc::clone(&seen);
1581 2 : s.set_change_hook(Arc::new(
1582 : move |_t: &TenantId,
1583 : before: Option<Value>,
1584 : after: Option<Value>|
1585 6 : -> antares_store::HookFuture<'_> {
1586 6 : let rec = Arc::clone(&rec);
1587 6 : Box::pin(async move {
1588 6 : rec.lock().expect("record").push((before, after));
1589 6 : })
1590 6 : },
1591 : ));
1592 :
1593 2 : s.create(&t, Kind::Entity, "urn:e", json!({"id": "urn:e", "n": 1}))
1594 2 : .await;
1595 : // a write on another kind must not reach the hook at all
1596 2 : s.create(&t, Kind::Subscription, "urn:s", json!({"id": "urn:s"}))
1597 2 : .await;
1598 2 : s.upsert(
1599 2 : &t,
1600 2 : Kind::Subscription,
1601 2 : "urn:s",
1602 2 : json!({"id": "urn:s", "n": 9}),
1603 2 : )
1604 2 : .await;
1605 2 : s.delete(&t, Kind::Subscription, "urn:s").await;
1606 : // a mutate that changes nothing is not a change
1607 2 : let _ = s
1608 2 : .mutate(&t, Kind::Entity, "urn:e", |_d| Ok::<(), ()>(()))
1609 2 : .await;
1610 : // a real one is
1611 2 : let _ = s
1612 2 : .mutate(&t, Kind::Entity, "urn:e", |d| {
1613 2 : d["n"] = json!(2);
1614 2 : Ok::<(), ()>(())
1615 2 : })
1616 2 : .await;
1617 : // an aborted mutate writes nothing and emits nothing
1618 2 : let _ = s
1619 2 : .mutate(&t, Kind::Entity, "urn:e", |d| {
1620 2 : d["n"] = json!(3);
1621 2 : Err::<(), &str>("no")
1622 2 : })
1623 2 : .await;
1624 2 : s.delete(&t, Kind::Entity, "urn:e").await;
1625 :
1626 2 : let seen = seen.lock().expect("read");
1627 2 : assert_eq!(seen.len(), 3, "emitted: {seen:?}");
1628 2 : assert!(seen[0].0.is_none() && seen[0].1.is_some(), "create");
1629 2 : assert_eq!(seen[1].0.as_ref().expect("before")["n"], 1);
1630 2 : assert_eq!(seen[1].1.as_ref().expect("after")["n"], 2);
1631 2 : assert!(seen[2].0.is_some() && seen[2].1.is_none(), "delete");
1632 : // the aborted mutate left the document alone
1633 2 : assert!(s.get(&t, Kind::Entity, "urn:e").is_none());
1634 2 : }
1635 :
1636 : /// The supported backup route is stop-copy — close the broker, copy
1637 : /// the file, reopen the copy. This test IS that route.
1638 : #[tokio::test]
1639 2 : async fn file_mode_stop_copy_backup_restores() {
1640 2 : let dir = tempdir("backup");
1641 2 : let restore = tempdir("restore");
1642 2 : let t = TenantId::default();
1643 : {
1644 2 : let s = Store::open_file(&dir).expect("open");
1645 2 : s.create(&t, Kind::Entity, "urn:b", json!({"v": 42})).await;
1646 : } // broker stopped — file quiescent
1647 2 : std::fs::copy(dir.join("antares.redb"), restore.join("antares.redb")).expect("copy");
1648 2 : let s = Store::open_file(&restore).expect("open backup");
1649 2 : assert_eq!(s.get(&t, Kind::Entity, "urn:b").expect("restored")["v"], 42);
1650 2 : let _ = std::fs::remove_dir_all(&dir);
1651 2 : let _ = std::fs::remove_dir_all(&restore);
1652 2 : }
1653 :
1654 : /// 4.22: reaping an expired entity is a state change like any other, so
1655 : /// it must be durable — after a sweep the doc stays gone across a reopen,
1656 : /// whatever the tenant key on disk looks like, while a doc that has not
1657 : /// expired survives both.
1658 : #[test]
1659 2 : fn file_mode_sweep_removals_persist() {
1660 2 : let dir = tempdir("sweep-persist");
1661 : // Seed the file directly: one tenant key the boot rebuild accepts but
1662 : // `TenantId::new` rejects, alongside an ordinary one.
1663 : {
1664 2 : let db = Database::create(dir.join("antares.redb")).expect("db");
1665 2 : let mut tx = db.begin_write().expect("tx");
1666 2 : tx.set_durability(Durability::Immediate).expect("dur");
1667 2 : {
1668 2 : let mut m = tx.open_table(T_META).expect("meta");
1669 2 : m.insert("format", FORMAT_VERSION).expect("insert");
1670 2 : }
1671 : {
1672 2 : let mut t = tx.open_table(T_ENTITIES).expect("entities");
1673 6 : for (tenant, id, expires) in [
1674 2 : ("odd.tenant", "urn:e:1", "2000-01-01T00:00:00Z"),
1675 2 : ("odd.tenant", "urn:e:2", "2999-01-01T00:00:00Z"),
1676 2 : ("plain", "urn:e:3", "2000-01-01T00:00:00Z"),
1677 6 : ] {
1678 6 : let mut k = tenant.as_bytes().to_vec();
1679 6 : k.push(0);
1680 6 : k.extend_from_slice(id.as_bytes());
1681 6 : let doc = json!({"id": id, "type": ["T"], "expiresAt": expires});
1682 6 : let bytes = serde_json::to_vec(&doc).expect("serialize");
1683 6 : t.insert(k.as_slice(), bytes.as_slice()).expect("insert");
1684 6 : }
1685 : }
1686 2 : tx.commit().expect("commit");
1687 : }
1688 : {
1689 2 : let s = Store::open_file(&dir).expect("open");
1690 2 : assert_eq!(s.sweep_expired("2026-01-01T00:00:00Z"), 2, "both expired");
1691 : }
1692 2 : let s = Store::open_file(&dir).expect("reopen");
1693 2 : let inner = s.inner.read().expect("lock");
1694 2 : assert!(
1695 2 : !inner
1696 2 : .entities
1697 2 : .get("odd.tenant")
1698 2 : .is_some_and(|d| d.contains_key("urn:e:1")),
1699 : "swept entity resurrected on reopen"
1700 : );
1701 2 : assert!(
1702 2 : inner
1703 2 : .entities
1704 2 : .get("odd.tenant")
1705 2 : .is_some_and(|d| d.contains_key("urn:e:2")),
1706 : "unexpired entity must outlive the sweep"
1707 : );
1708 2 : assert!(
1709 2 : !inner
1710 2 : .entities
1711 2 : .get("plain")
1712 2 : .is_some_and(|d| d.contains_key("urn:e:3")),
1713 : "swept entity resurrected on reopen"
1714 : );
1715 2 : drop(inner);
1716 2 : let _ = std::fs::remove_dir_all(&dir);
1717 2 : }
1718 :
1719 : #[tokio::test]
1720 2 : async fn sweep_prunes_expired_attribute_instances() {
1721 2 : let s = Store::default();
1722 2 : let t = TenantId::default();
1723 2 : s.create(
1724 2 : &t,
1725 2 : Kind::Entity,
1726 2 : "urn:e",
1727 2 : json!({"id": "urn:e", "type": ["T"],
1728 2 : "attr": [
1729 2 : {"type": "Property", "value": 1, "expiresAt": "2000-01-01T00:00:00Z"},
1730 2 : {"type": "Property", "value": 2, "expiresAt": "2999-01-01T00:00:00Z"}],
1731 2 : "gone": [
1732 2 : {"type": "Property", "value": 3, "expiresAt": "2000-01-01T00:00:00Z"}]}),
1733 2 : )
1734 2 : .await;
1735 2 : s.create(
1736 2 : &t,
1737 2 : Kind::Temporal,
1738 2 : "urn:e",
1739 2 : json!({"id": "urn:e", "type": ["T"],
1740 2 : "attr": [
1741 2 : {"value": 1, "expiresAt": "2000-01-01T00:00:00Z"},
1742 2 : {"value": 2, "expiresAt": "2999-01-01T00:00:00Z"},
1743 2 : {"value": 3}]}),
1744 2 : )
1745 2 : .await;
1746 2 : assert_eq!(s.sweep_expired("2026-01-01T00:00:00Z"), 2);
1747 2 : let e = s.get(&t, Kind::Entity, "urn:e").expect("entity survives");
1748 2 : assert_eq!(e["attr"].as_array().expect("attr").len(), 1);
1749 2 : assert_eq!(e["attr"][0]["value"], 2);
1750 2 : assert!(e.get("gone").is_none(), "fully-expired attribute removed");
1751 2 : let tp = s
1752 2 : .get(&t, Kind::Temporal, "urn:e")
1753 2 : .expect("temporal survives");
1754 2 : let vals: Vec<i64> = tp["attr"]
1755 2 : .as_array()
1756 2 : .expect("instances")
1757 2 : .iter()
1758 4 : .map(|i| i["value"].as_i64().expect("value"))
1759 2 : .collect();
1760 2 : assert_eq!(vals, [2, 3], "expired pruned, no-expiry instance kept");
1761 2 : assert_eq!(s.sweep_expired("2026-01-01T00:00:00Z"), 0, "idempotent");
1762 2 : }
1763 :
1764 : /// ADR-0021 + 4.14: the memory/file arm has no Row-Level Security under
1765 : /// it, so the rule a Postgres deployment gets from the database is this
1766 : /// store's own job. A Hosted @context holds term mappings authored
1767 : /// through one Tenant's requests and belongs to that Tenant alone; a
1768 : /// Cached copy of a public document belongs to none and every Tenant
1769 : /// reaches it. A store that answered every caller would hand one
1770 : /// Tenant's term mappings to another, and 5.5.7 makes those mappings
1771 : /// decide what its payloads mean.
1772 : #[test]
1773 2 : fn clause_5_13_1_a_stored_context_answers_only_its_own_tenant() {
1774 2 : let s = Store::default();
1775 2 : let alpha = TenantId::new("alpha").expect("tenant");
1776 2 : let beta = TenantId::new("beta").expect("tenant");
1777 2 : let hosted = json!({"localId": "h", "kind": "Hosted", "owner": "alpha",
1778 2 : "body": {"@context": {"a": "https://alpha.example/a"}}});
1779 2 : s.context_put(Some(&alpha), "h", hosted.clone())
1780 2 : .expect("the owner stores its own row");
1781 2 : s.context_put(
1782 2 : None,
1783 2 : "c",
1784 2 : json!({"localId": "c", "kind": "Cached", "body": {"@context": {}}}),
1785 : )
1786 2 : .expect("a Cached copy belongs to no Tenant");
1787 :
1788 2 : assert!(
1789 2 : s.context_get(Some(&alpha), "h").is_some(),
1790 : "its owner reads it"
1791 : );
1792 2 : assert!(
1793 2 : s.context_get(Some(&beta), "h").is_none(),
1794 : "another Tenant's Hosted @context is as absent as one never stored"
1795 : );
1796 2 : assert!(
1797 2 : s.context_get(None, "h").is_none(),
1798 : "no Tenant in scope reaches no Tenant's documents"
1799 : );
1800 6 : for t in [Some(&alpha), Some(&beta), None] {
1801 6 : assert!(
1802 6 : s.context_get(t, "c").is_some(),
1803 : "a Cached copy is a public document and is shared"
1804 : );
1805 : }
1806 2 : assert!(
1807 2 : !s.context_delete(Some(&beta), "h"),
1808 : "a foreign delete takes nothing"
1809 : );
1810 2 : assert!(
1811 2 : s.context_put(Some(&beta), "h", hosted).is_err(),
1812 : "and a foreign write does not overwrite it either"
1813 : );
1814 2 : assert!(
1815 2 : s.context_list_meta(Some(&beta))
1816 2 : .iter()
1817 2 : .all(|r| r["localId"] != "h"),
1818 : "another Tenant's row is not listed"
1819 : );
1820 2 : assert!(
1821 2 : s.context_get(Some(&alpha), "h").is_some(),
1822 : "the owner's row survived every foreign attempt"
1823 : );
1824 :
1825 2 : s.purge_tenant(&alpha);
1826 2 : assert!(
1827 2 : s.context_get(Some(&alpha), "h").is_none(),
1828 : "purging the Tenant takes the @contexts it stored"
1829 : );
1830 2 : assert!(
1831 2 : s.context_get(None, "c").is_some(),
1832 : "and leaves the copy that belongs to no Tenant"
1833 : );
1834 2 : }
1835 :
1836 : /// 5.13.1: "Implementations shall periodically invalidate the 'Cached'
1837 : /// @contexts." The memory/file arm holds one entry per distinct external
1838 : /// URL a request referenced — client-controlled — so the same ceiling and
1839 : /// oldest-first eviction the Pg arm applies holds here, and it applies to
1840 : /// the Cached kind ONLY: a Hosted entry is client-owned data (5.13.2).
1841 : #[test]
1842 2 : fn clause_5_13_1_cached_contexts_are_capped_oldest_first() {
1843 2 : let s = Store::default();
1844 2004 : let entry = |kind: &str, created: &str| json!({"kind": kind, "createdAt": created, "body": {"@context": {}}});
1845 : // a Hosted entry older than every Cached one: age must not decide
1846 2 : s.context_put(
1847 2 : Some(&TenantId::default()),
1848 2 : "hosted",
1849 2 : entry("Hosted", "2000-01-01T00:00:00Z"),
1850 : )
1851 2 : .expect("store");
1852 2000 : for i in 0..MAX_CACHED_CONTEXTS {
1853 2000 : s.context_put(
1854 2000 : None,
1855 2000 : &format!("cached-{i:05}"),
1856 2000 : entry("Cached", &format!("2026-01-01T00:00:{:02}Z", i % 60)),
1857 2000 : )
1858 2000 : .expect("store");
1859 2000 : }
1860 4 : let cached = |s: &Store| {
1861 4 : s.context_list_meta(Some(&TenantId::default()))
1862 4 : .iter()
1863 4004 : .filter(|d| d["kind"] == "Cached")
1864 4 : .count()
1865 4 : };
1866 2 : assert_eq!(
1867 2 : cached(&s),
1868 : MAX_CACHED_CONTEXTS,
1869 : "at the ceiling, nothing lost"
1870 : );
1871 :
1872 : // one more Cached entry evicts exactly one — the oldest
1873 2 : s.context_put(None, "cached-new", entry("Cached", "2026-06-01T00:00:00Z"))
1874 2 : .expect("store");
1875 2 : assert_eq!(cached(&s), MAX_CACHED_CONTEXTS, "the ceiling holds");
1876 2 : assert!(
1877 2 : s.context_get(None, "cached-new").is_some(),
1878 : "the new entry stays"
1879 : );
1880 2 : assert!(
1881 2 : s.context_get(None, "cached-00000").is_none(),
1882 : "the oldest Cached entry is the one evicted"
1883 : );
1884 2 : assert!(
1885 2 : s.context_get(None, "cached-00001").is_some(),
1886 : "eviction stops at the ceiling"
1887 : );
1888 2 : assert!(
1889 2 : s.context_get(Some(&TenantId::default()), "hosted")
1890 2 : .is_some(),
1891 : "a Hosted entry is never a candidate, however old"
1892 : );
1893 2 : }
1894 :
1895 : #[tokio::test]
1896 2 : async fn mutate_aborts_on_error() {
1897 2 : let s = Store::default();
1898 2 : let t = TenantId::default();
1899 2 : s.create(&t, Kind::Entity, "urn:a", json!({"n": 1})).await;
1900 2 : let r: Option<Result<(), &str>> = s
1901 2 : .mutate(&t, Kind::Entity, "urn:a", |d| {
1902 2 : d["n"] = json!(2);
1903 2 : Err("nope")
1904 2 : })
1905 2 : .await;
1906 2 : assert!(matches!(r, Some(Err("nope"))));
1907 2 : assert_eq!(s.get(&t, Kind::Entity, "urn:a").unwrap()["n"], 1);
1908 2 : }
1909 :
1910 : /// A purged tenant is gone from every kind, the neighbour tenant is
1911 : /// untouched, and in `file` mode the removal survives a reopen.
1912 : #[tokio::test]
1913 2 : async fn purge_tenant_empties_every_kind_and_survives_reopen() {
1914 2 : let dir = tempdir("purge");
1915 2 : let a = TenantId::new("purge_a").expect("tenant");
1916 2 : let b = TenantId::new("purge_b").expect("tenant");
1917 : {
1918 2 : let s = Store::open_file(&dir).expect("open");
1919 4 : for t in [&a, &b] {
1920 36 : for kind in ALL_KINDS {
1921 36 : assert!(s.create(t, kind, "urn:x:1", json!({"id": "urn:x:1"})).await);
1922 : }
1923 : }
1924 4 : assert!(s.tenant_ids().iter().any(|t| t == "purge_a"), "listed");
1925 2 : let row = s.tenant_stats_one(&a);
1926 2 : assert_eq!((row.entities, row.subscriptions, row.dist_subs), (1, 1, 1));
1927 2 : assert!(s.purge_tenant(&a));
1928 2 : assert!(!s.purge_tenant(&a), "second purge finds nothing");
1929 2 : assert!(!s.tenant_exists(&a));
1930 2 : assert!(s.tenant_exists(&b));
1931 18 : for kind in ALL_KINDS {
1932 18 : assert!(s.get(&a, kind, "urn:x:1").is_none(), "{kind:?} row left");
1933 18 : assert!(s.get(&b, kind, "urn:x:1").is_some(), "{kind:?} lost for b");
1934 : }
1935 4 : assert!(s.tenant_ids().iter().all(|t| t != "purge_a"));
1936 2 : assert_eq!(
1937 2 : s.tenant_stats_one(&a).entities,
1938 : 0,
1939 : "a purged tenant counts nothing"
1940 : );
1941 : }
1942 2 : let s = Store::open_file(&dir).expect("reopen");
1943 2 : assert!(!s.tenant_exists(&a), "purge must be persisted");
1944 2 : assert!(s.tenant_exists(&b));
1945 2 : let _ = std::fs::remove_dir_all(&dir);
1946 2 : }
1947 : }
|
