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};
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, Table, FORMAT_VERSION, T_ENTITIES, T_JSONLD_CONTEXTS,
21 : T_META, 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
55 2 : .iter()
56 2 : .take(cached.len().saturating_sub(MAX_CACHED_CONTEXTS))
57 2 : .map(|(id, _)| (*id).to_owned())
58 2 : .collect()
59 4 : }
60 :
61 : /// The attribute names of an internal entity doc: every member but the
62 : /// entity's own (`ENTITY_META_KEYS`), collected so the doc can be changed
63 : /// while they are walked.
64 6157 : fn attribute_names(obj: &serde_json::Map<String, Value>) -> Vec<String> {
65 6157 : obj.keys()
66 27159 : .filter(|k| !antares_model::ENTITY_META_KEYS.contains(&k.as_str()))
67 6157 : .cloned()
68 6157 : .collect()
69 6157 : }
70 :
71 : /// Drop attribute instances whose `expiresAt` passed (4.22) from an internal
72 : /// entity or temporal doc; an attribute whose last instance expired is
73 : /// removed entirely. Returns whether the doc changed.
74 6115 : pub(crate) fn prune_expired_instances(doc: &mut Value, now: &str) -> bool {
75 6115 : let Some(obj) = doc.as_object_mut() else {
76 0 : return false;
77 : };
78 6115 : let mut changed = false;
79 6125 : for k in attribute_names(obj) {
80 2527 : let Some(arr) = obj.get_mut(&k).and_then(Value::as_array_mut) else {
81 0 : continue;
82 : };
83 2527 : let before = arr.len();
84 : // 4.6.3 leaves the seconds-fraction separator open, so the two
85 : // stamps have to be compared as instants and not as bytes — the rule
86 : // lives in `filter::expired_at`, which the read boundary uses too.
87 2895 : arr.retain(|inst| !filter::expired_at(inst, now));
88 2527 : if arr.len() != before {
89 67 : changed = true;
90 67 : if arr.is_empty() {
91 44 : obj.remove(&k);
92 44 : }
93 2460 : }
94 : }
95 6115 : changed
96 6115 : }
97 :
98 : /// Whether an internal doc's types include `type_iri` (retention).
99 104 : fn typed(doc: &Value, type_iri: &str) -> bool {
100 104 : match doc.get("type") {
101 92 : Some(Value::Array(types)) => types.iter().any(|t| t.as_str() == Some(type_iri)),
102 12 : Some(Value::String(t)) => t == type_iri,
103 0 : _ => false,
104 : }
105 104 : }
106 :
107 : /// Whether `scope` reaches the temporal document `id`.
108 76 : fn in_scope(id: &str, doc: &Value, scope: antares_store::HistoryScope<'_>) -> bool {
109 : use antares_store::HistoryScope;
110 76 : match scope {
111 16 : HistoryScope::Entity(one) => id == one,
112 : HistoryScope::Type {
113 52 : type_iri,
114 52 : except_ids,
115 52 : } => typed(doc, type_iri) && !except_ids.iter().any(|e| e == id),
116 : HistoryScope::Rest {
117 8 : except_types,
118 8 : except_ids,
119 8 : } => !except_types.iter().any(|t| typed(doc, t)) && !except_ids.iter().any(|e| e == id),
120 : }
121 76 : }
122 :
123 : /// The instant an attribute instance is kept by, the same fallback the
124 : /// Postgres arm's `observed_at` column takes.
125 136 : fn instance_instant(inst: &Value) -> Option<&str> {
126 136 : ["observedAt", "modifiedAt", "deletedAt", "createdAt"]
127 136 : .iter()
128 156 : .find_map(|k| inst.get(*k).and_then(Value::as_str))
129 136 : }
130 :
131 : /// `a` before `b`, compared as instants (4.6.3 allows a comma fraction), by
132 : /// bytes when either does not parse.
133 9876 : pub(crate) fn earlier(a: &str, b: &str) -> bool {
134 9876 : let instant =
135 19752 : |s: &str| chrono::DateTime::parse_from_rfc3339(&filter::canonical_datetime(s)).ok();
136 9876 : match (instant(a), instant(b)) {
137 9876 : (Some(x), Some(y)) => x < y,
138 0 : _ => a < b,
139 : }
140 9876 : }
141 :
142 : /// The 4.22 "now" the write paths judge `expiresAt` against — the same UTC-Z
143 : /// millisecond form the read boundary uses.
144 31094 : fn now_stamp() -> String {
145 31094 : chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
146 31094 : }
147 :
148 : const ALL_KINDS: [Kind; 10] = [
149 : Kind::Entity,
150 : Kind::Subscription,
151 : Kind::Registration,
152 : Kind::CSourceSubscription,
153 : Kind::Temporal,
154 : Kind::Snapshot,
155 : Kind::EntityMap,
156 : Kind::DistSub,
157 : Kind::DeadLetter,
158 : Kind::View,
159 : ];
160 :
161 : /// Run `f` off the tokio worker pool when called from a multi-thread runtime:
162 : /// a per-commit fsync must never stall an async worker. Outside a
163 : /// runtime (unit tests, startup) it just runs inline.
164 : ///
165 : /// `durable` is what decides. A per-document write blocks only where it
166 : /// commits, so it passes `shadow.is_some()`: in `memory` mode that write is
167 : /// a lock and a map insert, and handing a worker's queue to another thread —
168 : /// a handoff, and past the pool's live threads a spawn — costs more than the
169 : /// write and costs more the more cores there are to hand work between. The
170 : /// paths that hold the write section for a whole SCAN (the 4.22 sweep, the
171 : /// two purges) pass `true` whatever the mode: what makes them long there is
172 : /// the scan, not the commit.
173 470863 : fn on_blocking<T>(durable: bool, f: impl FnOnce() -> T) -> T {
174 : // wasm32: single-threaded, no tokio runtime — always inline.
175 : #[cfg(not(target_arch = "wasm32"))]
176 470863 : if durable {
177 14874 : if let Ok(h) = tokio::runtime::Handle::try_current() {
178 14868 : if h.runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread {
179 14230 : return tokio::task::block_in_place(f);
180 638 : }
181 6 : }
182 455989 : }
183 : #[cfg(target_arch = "wasm32")]
184 : let _ = durable;
185 456633 : f()
186 470863 : }
187 :
188 : /// One write [`Store::stage`] queued: table, key, encoded doc (`None` =
189 : /// delete).
190 : type Staged = (Table, Vec<u8>, Option<Vec<u8>>);
191 :
192 : #[derive(Default)]
193 : pub struct Store {
194 : inner: RwLock<Inner>,
195 : hook: RwLock<Option<ChangeHook>>,
196 : /// Set when this instance serves only the temporal seam: it never holds
197 : /// the entities, so the append guard must not look for them here.
198 : pub temporal_only: bool,
199 : /// Most instances one Attribute's history may hold here; `0` = no cap
200 : /// (`ANTARES_MEM_MAX_INSTANCES`). An Entity's whole history is ONE
201 : /// document, cloned, compared and (in `file` mode) rewritten and fsynced
202 : /// whole on every append under the single write lock, so an unbounded
203 : /// one makes each write slower than the last, for every Tenant.
204 : pub max_instances: usize,
205 : /// `file` mode durability shadow; `None` = pure in-memory (`memory` mode).
206 : shadow: Option<Shadow>,
207 : /// Writers currently queued behind the single write-critical section
208 : /// (redb has ONE writer, so fsync commits serialize here). Exported via
209 : /// /q/health; the group-commit lever only gets built if a benchmark shows
210 : /// this depth sustained at the measured ~3.1k writes/s ceiling.
211 : write_waiters: std::sync::atomic::AtomicUsize,
212 : write_waiters_peak: std::sync::atomic::AtomicUsize,
213 : /// Change hooks fire in COMMIT order. The hook runs after the
214 : /// write-critical section (it may write other kinds through the store,
215 : /// so running it inside would deadlock), which lets a later commit's
216 : /// hook overtake an earlier one — the consumer would record stale state
217 : /// as newest. Entity writes therefore hold this from before the commit
218 : /// until the emit is done. Only the local single-process path needs it:
219 : /// across processes the transactional outbox is the ordered channel.
220 : emit_order: tokio::sync::Mutex<()>,
221 : /// Entity changes emitted, the version each carries to the hook: counted
222 : /// under `emit_order`, so in commit order.
223 : emitted: std::sync::atomic::AtomicI64,
224 : }
225 :
226 : #[derive(Default)]
227 : struct Inner {
228 : /// tenant → id → internal entity doc (BTreeMap: deterministic list order).
229 : entities: HashMap<String, BTreeMap<String, Value>>,
230 : subscriptions: HashMap<String, BTreeMap<String, Value>>,
231 : registrations: HashMap<String, BTreeMap<String, Value>>,
232 : csource_subscriptions: HashMap<String, BTreeMap<String, Value>>,
233 : snapshots: HashMap<String, BTreeMap<String, Value>>,
234 : entity_map_docs: HashMap<String, BTreeMap<String, Value>>,
235 : dist_subs: HashMap<String, BTreeMap<String, Value>>,
236 : dead_letters: HashMap<String, BTreeMap<String, Value>>,
237 : views: HashMap<String, BTreeMap<String, Value>>,
238 : /// tenant → entity id → temporal doc (attr IRI → instance array).
239 : temporal: HashMap<String, BTreeMap<String, Value>>,
240 : /// hosted/cached @context documents, shared across tenants by design.
241 : contexts: BTreeMap<String, Value>,
242 : }
243 :
244 : impl Store {
245 : /// Open (or create) the `file`-mode store: redb at `dir/antares.redb`,
246 : /// format-checked, in-memory maps rebuilt from the file.
247 : /// Any open/format/decode error refuses to start — never silently serve
248 : /// partial data.
249 70 : pub fn open_file(dir: &Path) -> Result<Self, String> {
250 70 : std::fs::create_dir_all(dir).map_err(|e| format!("create {}: {e}", dir.display()))?;
251 70 : let path = dir.join("antares.redb");
252 70 : let db = Database::create(&path).map_err(|e| format!("open {}: {e}", path.display()))?;
253 68 : Self::from_database(db, &path.display().to_string())
254 70 : }
255 :
256 : /// The target-independent half of `open_file`: format check +
257 : /// boot rebuild over an already-constructed redb `Database`. The
258 : /// browser build calls this with an OPFS-backed database — same shadow,
259 : /// same commit-before-ack, different `StorageBackend`.
260 72 : pub fn from_database(db: Database, label: &str) -> Result<Self, String> {
261 : // Format marker. Absent marker + existing data = a file this
262 : // binary cannot vouch for; refuse rather than guess.
263 72 : let stored_format = {
264 72 : let rt = db.begin_read().map_err(|e| e.to_string())?;
265 72 : match rt.open_table(T_META) {
266 36 : Ok(t) => t
267 36 : .get("format")
268 36 : .map_err(|e| e.to_string())?
269 36 : .map(|v| v.value().to_owned()),
270 36 : Err(::redb::TableError::TableDoesNotExist(_)) => None,
271 0 : Err(e) => return Err(e.to_string()),
272 : }
273 : };
274 72 : match stored_format.as_deref() {
275 36 : Some(FORMAT_VERSION) => {}
276 2 : Some(other) => {
277 2 : return Err(format!(
278 2 : "data file {label} has format {other}, this binary supports {FORMAT_VERSION} — \
279 2 : refusing to start"
280 2 : ));
281 : }
282 : None => {
283 36 : let rt = db.begin_read().map_err(|e| e.to_string())?;
284 324 : for kind in ALL_KINDS {
285 : // Absent = nothing stored under that kind. Any other
286 : // error is a table this binary cannot read, and reading
287 : // it as empty would let the guard stamp its marker onto
288 : // a file it is about to refuse.
289 324 : let t = match rt.open_table(table_for(kind)) {
290 2 : Ok(t) => t,
291 320 : Err(::redb::TableError::TableDoesNotExist(_)) => continue,
292 2 : Err(e) => return Err(e.to_string()),
293 : };
294 2 : if t.len().map_err(|e| e.to_string())? > 0 {
295 2 : return Err(format!(
296 2 : "data file {label} holds data but no format marker — refusing to start"
297 2 : ));
298 0 : }
299 : }
300 32 : drop(rt);
301 32 : let mut tx = db.begin_write().map_err(|e| e.to_string())?;
302 32 : tx.set_durability(Durability::Immediate)
303 32 : .map_err(|e| e.to_string())?;
304 : {
305 32 : let mut t = tx.open_table(T_META).map_err(|e| e.to_string())?;
306 32 : t.insert("format", FORMAT_VERSION)
307 32 : .map_err(|e| e.to_string())?;
308 : }
309 32 : tx.commit().map_err(|e| e.to_string())?;
310 : }
311 : }
312 :
313 : // Boot rebuild — scan every table into the in-memory maps.
314 66 : let mut inner = Inner::default();
315 66 : let rt = db.begin_read().map_err(|e| e.to_string())?;
316 660 : for kind in ALL_KINDS {
317 660 : let table = match rt.open_table(table_for(kind)) {
318 82 : Ok(t) => t,
319 578 : Err(::redb::TableError::TableDoesNotExist(_)) => continue,
320 0 : Err(e) => return Err(e.to_string()),
321 : };
322 82 : let map = match kind {
323 24 : Kind::Entity => &mut inner.entities,
324 6 : Kind::Subscription => &mut inner.subscriptions,
325 6 : Kind::Registration => &mut inner.registrations,
326 4 : Kind::CSourceSubscription => &mut inner.csource_subscriptions,
327 14 : Kind::Temporal => &mut inner.temporal,
328 6 : Kind::Snapshot => &mut inner.snapshots,
329 6 : Kind::EntityMap => &mut inner.entity_map_docs,
330 6 : Kind::DistSub => &mut inner.dist_subs,
331 6 : Kind::DeadLetter => &mut inner.dead_letters,
332 4 : Kind::View => &mut inner.views,
333 : };
334 1710 : for row in ::redb::ReadableTable::iter(&table).map_err(|e| e.to_string())? {
335 1710 : let (k, v) = row.map_err(|e| e.to_string())?;
336 1710 : let (tenant, id) = split_key(k.value())
337 1710 : .ok_or_else(|| format!("undecodable key in table {kind:?}"))?;
338 1710 : let doc: Value = serde_json::from_slice(v.value())
339 1710 : .map_err(|e| format!("undecodable value for {tenant}/{id}: {e}"))?;
340 1710 : map.entry(tenant).or_default().insert(id, doc);
341 : }
342 : }
343 : // Same rule as the kind loop above: absent is empty, unreadable is a
344 : // refusal. Hosted and ImplicitlyCreated @contexts (5.13.1) are a
345 : // Tenant's own documents, so starting without this table is serving
346 : // partial data, which `open_file` promises never to do.
347 66 : let contexts = match rt.open_table(T_JSONLD_CONTEXTS) {
348 4 : Ok(t) => Some(t),
349 60 : Err(::redb::TableError::TableDoesNotExist(_)) => None,
350 2 : Err(e) => return Err(e.to_string()),
351 : };
352 64 : if let Some(t) = contexts {
353 4 : for row in ::redb::ReadableTable::iter(&t).map_err(|e| e.to_string())? {
354 2 : let (k, v) = row.map_err(|e| e.to_string())?;
355 2 : let id = String::from_utf8(k.value().to_vec())
356 2 : .map_err(|e| format!("undecodable context id: {e}"))?;
357 2 : let doc: Value = serde_json::from_slice(v.value())
358 2 : .map_err(|e| format!("undecodable context {id}: {e}"))?;
359 2 : inner.contexts.insert(id, doc);
360 : }
361 60 : }
362 64 : drop(rt);
363 :
364 64 : Ok(Self {
365 64 : inner: RwLock::new(inner),
366 64 : hook: RwLock::new(None),
367 64 : emit_order: tokio::sync::Mutex::new(()),
368 64 : emitted: Default::default(),
369 64 : shadow: Some(Shadow { db }),
370 64 : temporal_only: false,
371 64 : max_instances: 0,
372 64 : write_waiters: Default::default(),
373 64 : write_waiters_peak: Default::default(),
374 64 : })
375 72 : }
376 :
377 : /// Write-through for one doc (must be called inside the write-critical
378 : /// section so redb order equals memory order).
379 450387 : fn persist(&self, table: Table, key: &[u8], doc: Option<&Value>) {
380 450387 : if let Some(shadow) = &self.shadow {
381 : // None means DELETE the key, so a document that will not encode
382 : // must skip the write rather than collapse into one.
383 12381 : let bytes = match doc {
384 9670 : Some(d) => match serde_json::to_vec(d) {
385 9670 : Ok(b) => Some(b),
386 0 : Err(_) => return,
387 : },
388 2711 : None => None,
389 : };
390 12381 : shadow.write(table, key, bytes.as_deref());
391 438006 : }
392 450387 : }
393 :
394 : /// Queue the write-through of one doc (`None` = delete) for
395 : /// [`Self::commit_staged`]; nothing in `memory` mode. A document that
396 : /// will not encode is skipped, as [`Self::persist`] skips it.
397 4442 : fn stage(&self, staged: &mut Vec<Staged>, table: Table, key: Vec<u8>, doc: Option<&Value>) {
398 4442 : if self.shadow.is_none() {
399 2626 : return;
400 1816 : }
401 1816 : let bytes = match doc {
402 122 : Some(d) => match serde_json::to_vec(d) {
403 122 : Ok(b) => Some(b),
404 0 : Err(_) => return,
405 : },
406 1694 : None => None,
407 : };
408 1816 : staged.push((table, key, bytes));
409 4442 : }
410 :
411 : /// Commit what [`Self::stage`] queued in one redb transaction (must be
412 : /// called inside the write-critical section, as `persist`): a
413 : /// multi-document change is durable whole, at the cost of one fsync.
414 3680 : fn commit_staged(&self, staged: &[Staged]) {
415 3680 : if let Some(shadow) = &self.shadow {
416 632 : shadow.write_all(
417 632 : staged
418 632 : .iter()
419 2258 : .map(|(t, k, v)| (*t, k.as_slice(), v.as_deref())),
420 : );
421 3048 : }
422 3680 : }
423 :
424 : /// Acquire the write-critical section, counting queued writers.
425 : ///
426 : /// Poison recovery (`into_inner`) is deliberate: a panic inside a caller's
427 : /// mutate closure unwinds one request; poisoning would turn it into a
428 : /// whole-process brick (every later store call panicking until restart).
429 : /// Consistency holds because mutations work on a clone and swap last.
430 470863 : fn write_inner(&self) -> std::sync::RwLockWriteGuard<'_, Inner> {
431 : use std::sync::atomic::Ordering;
432 470863 : let depth = self
433 470863 : .write_waiters
434 470863 : .fetch_add(1, Ordering::Relaxed)
435 470863 : .saturating_add(1);
436 470863 : self.write_waiters_peak.fetch_max(depth, Ordering::Relaxed);
437 470863 : let guard = self
438 470863 : .inner
439 470863 : .write()
440 470863 : .unwrap_or_else(std::sync::PoisonError::into_inner);
441 470863 : self.write_waiters.fetch_sub(1, Ordering::Relaxed);
442 470863 : guard
443 470863 : }
444 :
445 : /// Whether writes reach a durability shadow (`file` mode, and the
446 : /// browser build over OPFS) rather than living only in memory.
447 242 : pub fn shadowed(&self) -> bool {
448 242 : self.shadow.is_some()
449 242 : }
450 :
451 : /// (Currently queued writers, peak since start). The peak going
452 : /// nowhere near sustained depth is the evidence that the group-commit
453 : /// lever stays unbuilt.
454 8 : pub fn commit_queue(&self) -> (usize, usize) {
455 : use std::sync::atomic::Ordering;
456 8 : (
457 8 : self.write_waiters.load(Ordering::Relaxed),
458 8 : self.write_waiters_peak.load(Ordering::Relaxed),
459 8 : )
460 8 : }
461 :
462 1264 : pub fn set_change_hook(&self, h: ChangeHook) {
463 1264 : *self
464 1264 : .hook
465 1264 : .write()
466 1264 : .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(h);
467 1264 : }
468 :
469 : /// Held from before an entity commit until its emit returns (see
470 : /// `emit_order`). Not for other kinds: their writes emit nothing, and a
471 : /// hook that writes them re-enters the store.
472 182894 : async fn emit_ordered(&self) -> tokio::sync::MutexGuard<'_, ()> {
473 182894 : self.emit_order.lock().await
474 182894 : }
475 :
476 178306 : async fn emit(&self, tenant: &TenantId, before: Option<Value>, after: Option<Value>) {
477 : // cloned out of the lock first: a read guard cannot be held across
478 : // the await the hook now costs.
479 178306 : let hook = self
480 178306 : .hook
481 178306 : .read()
482 178306 : .unwrap_or_else(std::sync::PoisonError::into_inner)
483 178306 : .clone();
484 178306 : if let Some(h) = hook {
485 136790 : let version = antares_store::ChangeVersion {
486 136790 : incarnation: String::new(),
487 136790 : version: self
488 136790 : .emitted
489 136790 : .fetch_add(1, std::sync::atomic::Ordering::Relaxed)
490 136790 : .saturating_add(1),
491 136790 : };
492 136790 : h(tenant, before, after, version).await;
493 41516 : }
494 178304 : }
495 :
496 : /// 4.22 garbage collection for the memory/file arm: remove entity docs
497 : /// whose `expiresAt` (byte-compared against the UTC-Z `now` stamp, same
498 : /// as the read filter) has passed, and prune expired ATTRIBUTE instances
499 : /// from current-state and temporal docs — the read filter hides them,
500 : /// but without physical removal a long-running store (the browser's OPFS
501 : /// file under ticking sensors) grows without bound. `file` mode persists
502 : /// each removal. Returns how many docs were reaped or pruned.
503 : ///
504 : /// Runs through `on_blocking` like every other mutating path: in `file`
505 : /// mode it holds the write-critical section for a full scan and one
506 : /// `Durability::Immediate` (fsync) commit of every doc it reaped or
507 : /// pruned, which must never happen on an async worker thread.
508 972 : pub fn sweep_expired(&self, now: &str) -> usize {
509 972 : on_blocking(true, || self.sweep_expired_locked(now))
510 972 : }
511 :
512 972 : fn sweep_expired_locked(&self, now: &str) -> usize {
513 972 : let mut inner = self.write_inner();
514 972 : let mut reaped = 0usize;
515 972 : let mut dead: Vec<(String, String)> = Vec::new();
516 7958 : for (tenant, docs) in &inner.entities {
517 8758 : for (id, doc) in docs {
518 6445 : if filter::expired_at(doc, now) {
519 828 : dead.push((tenant.clone(), id.clone()));
520 6413 : }
521 : }
522 : }
523 : // One commit for the whole pass: an Entity and its history leave the
524 : // file together, and the section waits for one fsync, not one a key.
525 972 : let mut staged = Vec::new();
526 1760 : for (tenant, id) in dead {
527 828 : if let Some(docs) = inner.entities.get_mut(&tenant) {
528 828 : self.stage(&mut staged, T_ENTITIES, key_bytes(&tenant, &id), None);
529 828 : docs.remove(&id);
530 828 : reaped = reaped.saturating_add(1);
531 828 : }
532 : // Its history goes with it.
533 828 : if let Some(docs) = inner.temporal.get_mut(&tenant) {
534 822 : if docs.remove(&id).is_some() {
535 818 : self.stage(
536 818 : &mut staged,
537 818 : T_TEMPORAL_ENTITIES,
538 818 : key_bytes(&tenant, &id),
539 818 : None,
540 818 : );
541 818 : }
542 6 : }
543 : }
544 : let Inner {
545 972 : entities, temporal, ..
546 972 : } = &mut *inner;
547 1944 : for (table, map) in [
548 972 : (
549 972 : T_ENTITIES,
550 972 : entities as &mut HashMap<String, BTreeMap<String, Value>>,
551 972 : ),
552 972 : (T_TEMPORAL_ENTITIES, temporal),
553 972 : ] {
554 10778 : for (tenant, docs) in map.iter_mut() {
555 10760 : for (id, doc) in docs.iter_mut() {
556 6094 : if prune_expired_instances(doc, now) {
557 24 : reaped = reaped.saturating_add(1);
558 24 : self.stage(&mut staged, table, key_bytes(tenant, id), Some(doc));
559 6070 : }
560 : }
561 : }
562 : }
563 972 : self.commit_staged(&staged);
564 972 : reaped
565 972 : }
566 :
567 : /// Tenants that hold any subscriptions (interval-firing scan).
568 : /// 5.5.10: a Tenant exists once any create operation implicitly created
569 : /// it (any resource kind was ever written under it).
570 260 : pub fn tenant_exists(&self, tenant: &TenantId) -> bool {
571 260 : let inner = self
572 260 : .inner
573 260 : .read()
574 260 : .unwrap_or_else(std::sync::PoisonError::into_inner);
575 260 : ALL_KINDS
576 260 : .iter()
577 974 : .any(|k| Self::map(&inner, *k).contains_key(tenant.as_str()))
578 260 : }
579 :
580 : /// Every tenant name, sorted; the default tenant is always present, and
581 : /// the tenants the broker minted for itself are not — `/q/tenants` is the
582 : /// inventory of customer accounts, not of the broker's own bookkeeping.
583 1008 : pub fn tenant_ids(&self) -> Vec<String> {
584 1008 : let inner = self
585 1008 : .inner
586 1008 : .read()
587 1008 : .unwrap_or_else(std::sync::PoisonError::into_inner);
588 1008 : let mut names: std::collections::BTreeSet<String> =
589 1008 : std::iter::once(TenantId::DEFAULT.to_string()).collect();
590 10080 : for kind in ALL_KINDS {
591 10080 : names.extend(
592 10080 : Self::map(&inner, kind)
593 10080 : .keys()
594 17952 : .filter(|t| !TenantId::is_reserved_str(t))
595 10080 : .cloned(),
596 : );
597 : }
598 1008 : names.into_iter().collect()
599 1008 : }
600 :
601 : /// The types the Tenant's live Entities carry, sorted: one pass
602 : /// over the held documents, none of them cloned. 4.22: an expired Entity
603 : /// carries no type.
604 138 : pub fn entity_type_names(&self, tenant: &TenantId, now: &str) -> Vec<String> {
605 138 : self.entity_types_where(tenant, now, |_| true)
606 138 : }
607 :
608 : /// The types of the live Entities that carry one of `any_of`, sorted.
609 16 : pub fn entity_types_beside(
610 16 : &self,
611 16 : tenant: &TenantId,
612 16 : any_of: &[String],
613 16 : now: &str,
614 16 : ) -> Vec<String> {
615 544 : self.entity_types_where(tenant, now, |types| {
616 544 : types.len() > 1 && types.iter().any(|t| any_of.iter().any(|a| a == t))
617 544 : })
618 16 : }
619 :
620 : /// The (type, member name) pairs the Tenant's live Entities carry,
621 : /// sorted: one pass over the held documents.
622 96 : pub fn entity_attr_pairs(&self, tenant: &TenantId, now: &str) -> Vec<(String, String)> {
623 96 : let inner = self
624 96 : .inner
625 96 : .read()
626 96 : .unwrap_or_else(std::sync::PoisonError::into_inner);
627 96 : let mut pairs = std::collections::BTreeSet::new();
628 96 : if let Some(m) = Self::map(&inner, Kind::Entity).get(tenant.as_str()) {
629 880 : for doc in m.values().filter(|d| !filter::expired_at(d, now)) {
630 880 : let types: Vec<&str> = match &doc["type"] {
631 4 : Value::String(t) => vec![t.as_str()],
632 876 : Value::Array(a) => a.iter().filter_map(Value::as_str).collect(),
633 0 : _ => Vec::new(),
634 : };
635 3282 : for k in doc.as_object().into_iter().flat_map(|o| o.keys()) {
636 3340 : pairs.extend(types.iter().map(|t| (*t, k.as_str())));
637 : }
638 : }
639 20 : }
640 96 : pairs
641 96 : .into_iter()
642 664 : .map(|(t, k)| (t.to_owned(), k.to_owned()))
643 96 : .collect()
644 96 : }
645 :
646 154 : fn entity_types_where(
647 154 : &self,
648 154 : tenant: &TenantId,
649 154 : now: &str,
650 154 : keep: impl Fn(&[&str]) -> bool,
651 154 : ) -> Vec<String> {
652 154 : let inner = self
653 154 : .inner
654 154 : .read()
655 154 : .unwrap_or_else(std::sync::PoisonError::into_inner);
656 154 : let mut names = std::collections::BTreeSet::new();
657 154 : if let Some(m) = Self::map(&inner, Kind::Entity).get(tenant.as_str()) {
658 112738 : for doc in m.values().filter(|d| !filter::expired_at(d, now)) {
659 112738 : let types: Vec<&str> = match &doc["type"] {
660 4 : Value::String(t) => vec![t.as_str()],
661 112734 : Value::Array(a) => a.iter().filter_map(Value::as_str).collect(),
662 0 : _ => Vec::new(),
663 : };
664 112738 : if keep(&types) {
665 112202 : names.extend(types);
666 112296 : }
667 : }
668 36 : }
669 154 : names.into_iter().map(str::to_owned).collect()
670 154 : }
671 :
672 : /// What one tenant holds. Existence is the caller's question — an
673 : /// unknown tenant simply counts zero of everything.
674 36 : pub fn tenant_stats_one(&self, tenant: &TenantId) -> antares_store::TenantStats {
675 36 : let inner = self
676 36 : .inner
677 36 : .read()
678 36 : .unwrap_or_else(std::sync::PoisonError::into_inner);
679 252 : let n = |k: Kind| {
680 252 : Self::map(&inner, k)
681 252 : .get(tenant.as_str())
682 252 : .map_or(0, |m| m.len() as u64)
683 252 : };
684 36 : antares_store::TenantStats {
685 36 : entities: n(Kind::Entity),
686 36 : subscriptions: n(Kind::Subscription),
687 36 : registrations: n(Kind::Registration),
688 36 : csource_subscriptions: n(Kind::CSourceSubscription),
689 36 : snapshots: n(Kind::Snapshot),
690 36 : entity_maps: n(Kind::EntityMap),
691 36 : dist_subs: n(Kind::DistSub),
692 36 : created_at: None,
693 36 : tenant: tenant.as_str().to_owned(),
694 36 : }
695 36 : }
696 :
697 : /// Attribute instances held in the tenant's temporal documents.
698 32 : pub fn attr_instance_count(&self, tenant: &TenantId) -> u64 {
699 32 : let inner = self
700 32 : .inner
701 32 : .read()
702 32 : .unwrap_or_else(std::sync::PoisonError::into_inner);
703 32 : inner.temporal.get(tenant.as_str()).map_or(0, |docs| {
704 22 : docs.values()
705 22 : .filter_map(Value::as_object)
706 28 : .flat_map(|d| d.values().filter_map(Value::as_array))
707 1544 : .map(|a| a.iter().filter(|i| i.is_object()).count() as u64)
708 22 : .sum()
709 22 : })
710 32 : }
711 :
712 : /// The serialized size of the tenant's Entities, by first type.
713 30 : pub fn tenant_bytes(&self, tenant: &TenantId) -> antares_store::TenantBytes {
714 30 : let inner = self
715 30 : .inner
716 30 : .read()
717 30 : .unwrap_or_else(std::sync::PoisonError::into_inner);
718 30 : let mut out = antares_store::TenantBytes::default();
719 30 : for doc in inner
720 30 : .entities
721 30 : .get(tenant.as_str())
722 30 : .into_iter()
723 30 : .flat_map(BTreeMap::values)
724 : {
725 26 : let n = serde_json::to_vec(doc).map_or(0, |v| v.len() as u64);
726 26 : let ty = match doc.get("type") {
727 0 : Some(Value::String(t)) => t.as_str(),
728 26 : Some(Value::Array(a)) => a.first().and_then(Value::as_str).unwrap_or(""),
729 0 : _ => "",
730 : };
731 26 : out.entities = out.entities.saturating_add(n);
732 26 : let t = out.by_type.entry(ty.to_owned()).or_default();
733 26 : *t = t.saturating_add(n);
734 : }
735 30 : out
736 30 : }
737 :
738 : /// The serialized size of the tenant's temporal documents.
739 28 : pub fn attr_instance_bytes(&self, tenant: &TenantId) -> u64 {
740 28 : let inner = self
741 28 : .inner
742 28 : .read()
743 28 : .unwrap_or_else(std::sync::PoisonError::into_inner);
744 28 : inner.temporal.get(tenant.as_str()).map_or(0, |docs| {
745 18 : docs.values()
746 20 : .map(|d| serde_json::to_vec(d).map_or(0, |v| v.len() as u64))
747 18 : .sum()
748 18 : })
749 28 : }
750 :
751 : /// Retention of current state: drop at most `batch` Entities of
752 : /// `tenant` typed `type_iri` that carry no `expiresAt` and whose
753 : /// `modifiedAt` is before `before`. Persisted in one commit in `file`
754 : /// mode; `dry_run` counts them and drops none.
755 16 : pub fn retain_current(
756 16 : &self,
757 16 : tenant: &TenantId,
758 16 : type_iri: &str,
759 16 : before: &str,
760 16 : batch: usize,
761 16 : dry_run: bool,
762 16 : ) -> u64 {
763 16 : on_blocking(true, || {
764 16 : let mut inner = self.write_inner();
765 16 : let Some(docs) = inner.entities.get_mut(tenant.as_str()) else {
766 4 : return 0;
767 : };
768 12 : let dead: Vec<String> = docs
769 12 : .iter()
770 44 : .filter(|(_, doc)| {
771 44 : typed(doc, type_iri)
772 32 : && doc.get("expiresAt").is_none()
773 20 : && doc
774 20 : .get("modifiedAt")
775 20 : .and_then(Value::as_str)
776 20 : .is_some_and(|m| earlier(m, before))
777 44 : })
778 12 : .map(|(id, _)| id.clone())
779 12 : .take(batch)
780 12 : .collect();
781 12 : let mut staged = Vec::new();
782 12 : for id in dead.iter().filter(|_| !dry_run) {
783 4 : self.stage(
784 4 : &mut staged,
785 4 : T_ENTITIES,
786 4 : key_bytes(tenant.as_str(), id),
787 4 : None,
788 4 : );
789 4 : docs.remove(id);
790 4 : }
791 12 : self.commit_staged(&staged);
792 12 : dead.len() as u64
793 16 : })
794 16 : }
795 :
796 : /// The Entities of every Tenant carrying their own history retention,
797 : /// at most `limit`, with the Property as stored.
798 4 : pub fn history_overrides(&self, limit: usize) -> Vec<(String, String, Value)> {
799 4 : let inner = self
800 4 : .inner
801 4 : .read()
802 4 : .unwrap_or_else(std::sync::PoisonError::into_inner);
803 4 : inner
804 4 : .entities
805 4 : .iter()
806 6 : .flat_map(|(tenant, docs)| {
807 8 : docs.iter().filter_map(move |(id, doc)| {
808 8 : doc.get(antares_store::HISTORY_RETENTION)
809 8 : .map(|own| (tenant.clone(), id.clone(), own.clone()))
810 8 : })
811 6 : })
812 4 : .take(limit)
813 4 : .collect()
814 4 : }
815 :
816 : /// Retention of history: drop at most `batch` attribute
817 : /// instances of the `tenant`'s temporal documents `scope` names whose
818 : /// instant is before `before`; an attribute left without instances goes.
819 : /// `dry_run` counts them and drops none.
820 32 : pub fn retain_history(
821 32 : &self,
822 32 : tenant: &TenantId,
823 32 : scope: antares_store::HistoryScope<'_>,
824 32 : before: &str,
825 32 : batch: usize,
826 32 : dry_run: bool,
827 32 : ) -> u64 {
828 32 : on_blocking(true, || {
829 32 : let mut inner = self.write_inner();
830 32 : let Some(docs) = inner.temporal.get_mut(tenant.as_str()) else {
831 4 : return 0;
832 : };
833 28 : let mut left = batch;
834 28 : let mut dropped = 0u64;
835 28 : let mut staged = Vec::new();
836 84 : for (id, doc) in docs.iter_mut() {
837 84 : if left == 0 {
838 8 : break;
839 76 : }
840 76 : if !in_scope(id, doc, scope) {
841 34 : continue;
842 42 : }
843 42 : let Some(obj) = doc.as_object_mut() else {
844 0 : continue;
845 : };
846 42 : let before_count = dropped;
847 42 : for k in attribute_names(obj) {
848 40 : let Some(arr) = obj.get_mut(&k).and_then(Value::as_array_mut) else {
849 0 : continue;
850 : };
851 216 : arr.retain(|inst| {
852 216 : let old =
853 216 : left > 0 && instance_instant(inst).is_some_and(|t| earlier(t, before));
854 216 : if old {
855 116 : left = left.saturating_sub(1);
856 116 : dropped = dropped.saturating_add(1);
857 116 : }
858 216 : dry_run || !old
859 216 : });
860 40 : if arr.is_empty() && !dry_run {
861 10 : obj.remove(&k);
862 30 : }
863 : }
864 42 : if dropped != before_count && !dry_run {
865 22 : self.stage(
866 22 : &mut staged,
867 22 : T_TEMPORAL_ENTITIES,
868 22 : key_bytes(tenant.as_str(), id),
869 22 : Some(doc),
870 22 : );
871 22 : }
872 : }
873 28 : self.commit_staged(&staged);
874 28 : dropped
875 32 : })
876 32 : }
877 :
878 : /// Drop every document of the given kinds for one tenant; `true` when
879 : /// the tenant held any of them. Persisted in one commit in `file` mode.
880 86 : pub fn purge_kinds(&self, tenant: &TenantId, kinds: &[Kind]) -> bool {
881 86 : on_blocking(true, || {
882 86 : let mut inner = self.write_inner();
883 86 : let mut staged = Vec::new();
884 86 : let hit = self.purge_kinds_locked(&mut inner, tenant, kinds, &mut staged);
885 86 : self.commit_staged(&staged);
886 86 : hit
887 86 : })
888 86 : }
889 :
890 178 : fn purge_kinds_locked(
891 178 : &self,
892 178 : inner: &mut Inner,
893 178 : tenant: &TenantId,
894 178 : kinds: &[Kind],
895 178 : staged: &mut Vec<Staged>,
896 178 : ) -> bool {
897 178 : let mut hit = false;
898 1006 : for kind in kinds {
899 1006 : if let Some(docs) = Self::map_mut(inner, *kind).remove(tenant.as_str()) {
900 206 : hit = true;
901 252 : for id in docs.keys() {
902 252 : self.stage(
903 252 : staged,
904 252 : table_for(*kind),
905 252 : key_bytes(tenant.as_str(), id),
906 252 : None,
907 252 : );
908 252 : }
909 800 : }
910 : }
911 178 : hit
912 178 : }
913 :
914 : /// Remove the tenant from every kind, history included, and its own
915 : /// @contexts, in one commit in `file` mode: the operator's erase path
916 : /// never leaves part of a Tenant on disk.
917 92 : pub fn purge_tenant(&self, tenant: &TenantId) -> bool {
918 92 : on_blocking(true, || {
919 92 : let mut inner = self.write_inner();
920 92 : let mut staged = Vec::new();
921 92 : let existed = self.purge_kinds_locked(&mut inner, tenant, &ALL_KINDS, &mut staged);
922 : // ADR-0021: the Tenant's Hosted and ImplicitlyCreated @contexts are
923 : // its documents and go with it — a row that outlived its Tenant would
924 : // hand the next holder of that name a stranger's term mappings. A
925 : // Cached row belongs to no Tenant and stays.
926 92 : let dead: Vec<String> = inner
927 92 : .contexts
928 92 : .iter()
929 110 : .filter(|(_, row)| context_row_owner(row) == Some(tenant.as_str()))
930 92 : .map(|(id, _)| id.clone())
931 92 : .collect();
932 92 : for id in dead {
933 2 : self.stage(&mut staged, T_JSONLD_CONTEXTS, id.as_bytes().to_vec(), None);
934 2 : inner.contexts.remove(&id);
935 2 : }
936 92 : self.commit_staged(&staged);
937 92 : existed
938 92 : })
939 92 : }
940 :
941 2958 : pub fn subscription_tenants(&self) -> Vec<String> {
942 2958 : let inner = self
943 2958 : .inner
944 2958 : .read()
945 2958 : .unwrap_or_else(std::sync::PoisonError::into_inner);
946 : // Registrations belong to this domain: one of the hydrations that
947 : // walks it fills the registration mirror, which the federation path
948 : // then reads alone. Without them a tenant holding registrations and
949 : // no subscription forwarded to no Context Source.
950 2958 : let mut out: Vec<String> = inner
951 2958 : .subscriptions
952 2958 : .iter()
953 2958 : .chain(inner.csource_subscriptions.iter())
954 2958 : .chain(inner.registrations.iter())
955 3048 : .filter(|(_, m)| !m.is_empty())
956 2958 : .map(|(t, _)| t.clone())
957 2958 : .collect();
958 2958 : out.sort();
959 2958 : out.dedup();
960 2958 : out
961 2958 : }
962 :
963 542602 : fn map(inner: &Inner, kind: Kind) -> &HashMap<String, BTreeMap<String, Value>> {
964 542602 : match kind {
965 347168 : Kind::Entity => &inner.entities,
966 10665 : Kind::Subscription => &inner.subscriptions,
967 20063 : Kind::Registration => &inner.registrations,
968 4065 : Kind::CSourceSubscription => &inner.csource_subscriptions,
969 125618 : Kind::Temporal => &inner.temporal,
970 10639 : Kind::Snapshot => &inner.snapshots,
971 16236 : Kind::EntityMap => &inner.entity_map_docs,
972 5814 : Kind::DistSub => &inner.dist_subs,
973 1228 : Kind::DeadLetter => &inner.dead_letters,
974 1106 : Kind::View => &inner.views,
975 : }
976 542602 : }
977 :
978 457199 : fn map_mut(inner: &mut Inner, kind: Kind) -> &mut HashMap<String, BTreeMap<String, Value>> {
979 457199 : match kind {
980 182966 : Kind::Entity => &mut inner.entities,
981 6673 : Kind::Subscription => &mut inner.subscriptions,
982 3826 : Kind::Registration => &mut inner.registrations,
983 809 : Kind::CSourceSubscription => &mut inner.csource_subscriptions,
984 253708 : Kind::Temporal => &mut inner.temporal,
985 807 : Kind::Snapshot => &mut inner.snapshots,
986 4912 : Kind::EntityMap => &mut inner.entity_map_docs,
987 3200 : Kind::DistSub => &mut inner.dist_subs,
988 188 : Kind::DeadLetter => &mut inner.dead_letters,
989 110 : Kind::View => &mut inner.views,
990 : }
991 457199 : }
992 :
993 : /// 4.22: "expiresAt is defined as the system temporal Property at which a
994 : /// certain Entity, Property or Relationship shall become invalid." An
995 : /// entity past its expiry is invalid the moment the stamp passes, ahead
996 : /// of the sweep that physically reaps it, so every write path treats it
997 : /// as absent — otherwise the same id 404s on retrieve and 409s on create
998 : /// for a whole sweep interval. Only entities carry the entity-level
999 : /// stamp; subscriptions and registrations have their own expiry rules.
1000 456213 : fn is_expired(&self, inner: &Inner, kind: Kind, tenant: &str, id: &str) -> bool {
1001 456213 : kind == Kind::Entity
1002 182894 : && Self::map(inner, kind)
1003 182894 : .get(tenant)
1004 182894 : .and_then(|m| m.get(id))
1005 182894 : .is_some_and(|d| filter::expired_at(d, &now_stamp()))
1006 456213 : }
1007 :
1008 : /// Insert a new resource; `false` if the id already exists.
1009 261970 : pub async fn create(&self, tenant: &TenantId, kind: Kind, id: &str, doc: Value) -> bool {
1010 261970 : let _order = match kind {
1011 133196 : Kind::Entity => Some(self.emit_ordered().await),
1012 128774 : _ => None,
1013 : };
1014 261970 : let created = on_blocking(self.shadow.is_some(), || {
1015 261970 : let mut inner = self.write_inner();
1016 261970 : let expired = self.is_expired(&inner, kind, tenant.as_str(), id);
1017 261970 : let m = Self::map_mut(&mut inner, kind)
1018 261970 : .entry(tenant.as_str().to_owned())
1019 261970 : .or_default();
1020 261970 : if m.contains_key(id) && !expired {
1021 70 : false
1022 : } else {
1023 261900 : m.insert(id.to_owned(), doc.clone());
1024 261900 : self.persist(table_for(kind), &key_bytes(tenant.as_str(), id), Some(&doc));
1025 261900 : true
1026 : }
1027 261970 : });
1028 261970 : if created && kind == Kind::Entity {
1029 133146 : self.emit(tenant, None, Some(doc)).await;
1030 128824 : }
1031 261970 : created
1032 261970 : }
1033 :
1034 : /// Insert or replace; returns `true` if it existed before. An expired
1035 : /// entity did not (4.22), so the upsert reports CREATED and the caller
1036 : /// answers 201 with a Location header instead of a silent 204.
1037 18546 : pub async fn upsert(&self, tenant: &TenantId, kind: Kind, id: &str, doc: Value) -> bool {
1038 18546 : let _order = match kind {
1039 18452 : Kind::Entity => Some(self.emit_ordered().await),
1040 94 : _ => None,
1041 : };
1042 18546 : let (prev, expired) = on_blocking(self.shadow.is_some(), || {
1043 18546 : let mut inner = self.write_inner();
1044 18546 : let expired = self.is_expired(&inner, kind, tenant.as_str(), id);
1045 18546 : let prev = Self::map_mut(&mut inner, kind)
1046 18546 : .entry(tenant.as_str().to_owned())
1047 18546 : .or_default()
1048 18546 : .insert(id.to_owned(), doc.clone());
1049 18546 : self.persist(table_for(kind), &key_bytes(tenant.as_str(), id), Some(&doc));
1050 18546 : (prev, expired)
1051 18546 : });
1052 18546 : let existed = prev.is_some() && !expired;
1053 18546 : let prev = if expired { None } else { prev };
1054 18546 : if kind == Kind::Entity {
1055 18452 : self.emit(tenant, prev, Some(doc)).await;
1056 94 : }
1057 18544 : existed
1058 18544 : }
1059 :
1060 294573 : pub fn get(&self, tenant: &TenantId, kind: Kind, id: &str) -> Option<Value> {
1061 294573 : let inner = self
1062 294573 : .inner
1063 294573 : .read()
1064 294573 : .unwrap_or_else(std::sync::PoisonError::into_inner);
1065 294573 : Self::map(&inner, kind)
1066 294573 : .get(tenant.as_str())
1067 294573 : .and_then(|m| m.get(id))
1068 294573 : .cloned()
1069 294573 : }
1070 :
1071 : /// An expired entity is already invalid (4.22): deleting it is a 404,
1072 : /// the same answer retrieving it gives, not a 204 for something the API
1073 : /// stopped serving. The row itself still goes — the sweep would take it
1074 : /// anyway, and leaving it would resurrect the 409.
1075 19738 : pub async fn delete(&self, tenant: &TenantId, kind: Kind, id: &str) -> bool {
1076 19738 : let _order = match kind {
1077 7390 : Kind::Entity => Some(self.emit_ordered().await),
1078 12348 : _ => None,
1079 : };
1080 19738 : let removed = on_blocking(self.shadow.is_some(), || {
1081 19738 : let mut inner = self.write_inner();
1082 : // an expired doc is left in place for the sweep to reap: removing
1083 : // it here without persisting the removal would resurrect it on
1084 : // the next boot of a `file`-mode store
1085 19738 : if self.is_expired(&inner, kind, tenant.as_str(), id) {
1086 10 : return None;
1087 19728 : }
1088 19728 : let removed = Self::map_mut(&mut inner, kind)
1089 19728 : .get_mut(tenant.as_str())
1090 19728 : .and_then(|m| m.remove(id));
1091 19728 : if removed.is_some() {
1092 11084 : self.persist(table_for(kind), &key_bytes(tenant.as_str(), id), None);
1093 11084 : }
1094 19728 : removed
1095 19738 : });
1096 19738 : let hit = removed.is_some();
1097 19738 : if kind == Kind::Entity {
1098 7390 : if let Some(old) = removed {
1099 7328 : self.emit(tenant, Some(old), None).await;
1100 62 : }
1101 12348 : }
1102 19738 : hit
1103 19738 : }
1104 :
1105 : /// Delete one entity only if `keep` accepts the stored document. The
1106 : /// read and the removal happen under one hold of the write lock, so
1107 : /// nothing can replace the document between the decision and the delete.
1108 : /// `None` = absent, expired (4.22, as in `delete`) or refused.
1109 1620 : pub async fn delete_if(
1110 1620 : &self,
1111 1620 : tenant: &TenantId,
1112 1620 : id: &str,
1113 1620 : keep: &(dyn for<'v> Fn(&'v Value) -> bool + Sync),
1114 1620 : ) -> Option<Value> {
1115 1620 : let _order = self.emit_ordered().await;
1116 1620 : let removed = on_blocking(self.shadow.is_some(), || {
1117 1620 : let mut inner = self.write_inner();
1118 1620 : if self.is_expired(&inner, Kind::Entity, tenant.as_str(), id) {
1119 4 : return None;
1120 1616 : }
1121 1616 : let m = Self::map_mut(&mut inner, Kind::Entity).get_mut(tenant.as_str())?;
1122 1578 : if !m.get(id).is_some_and(keep) {
1123 122 : return None;
1124 1456 : }
1125 1456 : let removed = m.remove(id);
1126 1456 : if removed.is_some() {
1127 1456 : self.persist(
1128 1456 : table_for(Kind::Entity),
1129 1456 : &key_bytes(tenant.as_str(), id),
1130 1456 : None,
1131 1456 : );
1132 1456 : }
1133 1456 : removed
1134 1620 : });
1135 1620 : if let Some(old) = &removed {
1136 1456 : self.emit(tenant, Some(old.clone()), None).await;
1137 164 : }
1138 1620 : removed
1139 1620 : }
1140 :
1141 : /// Snapshot of all docs of a kind for one tenant (id order).
1142 10143 : pub fn list(&self, tenant: &TenantId, kind: Kind) -> Vec<Value> {
1143 10143 : let inner = self
1144 10143 : .inner
1145 10143 : .read()
1146 10143 : .unwrap_or_else(std::sync::PoisonError::into_inner);
1147 10143 : Self::map(&inner, kind)
1148 10143 : .get(tenant.as_str())
1149 10143 : .map(|m| m.values().cloned().collect())
1150 10143 : .unwrap_or_default()
1151 10143 : }
1152 :
1153 : /// One id-ordered page of docs: ids strictly greater than `after`, at
1154 : /// most `limit`. The per-tenant map is a `BTreeMap` keyed by id, so this
1155 : /// is a range walk, not a scan-and-sort.
1156 : ///
1157 : /// 4.22 applies to entities inside the walk, BEFORE the page is cut: an
1158 : /// expired document is not there to be paged, and dropping one after the
1159 : /// cut would hand the caller a short page — which every walker reads as
1160 : /// the end of the tenant.
1161 40233 : pub fn list_page(
1162 40233 : &self,
1163 40233 : tenant: &TenantId,
1164 40233 : kind: Kind,
1165 40233 : after: Option<&str>,
1166 40233 : limit: usize,
1167 40233 : ) -> Vec<Value> {
1168 : use std::ops::Bound;
1169 40233 : let inner = self
1170 40233 : .inner
1171 40233 : .read()
1172 40233 : .unwrap_or_else(std::sync::PoisonError::into_inner);
1173 40233 : let lo = match after {
1174 428 : Some(a) => Bound::Excluded(a.to_owned()),
1175 39805 : None => Bound::Unbounded,
1176 : };
1177 40233 : let now = crate::store::any::now_utc();
1178 40233 : Self::map(&inner, kind)
1179 40233 : .get(tenant.as_str())
1180 40233 : .map(|m| {
1181 8951 : m.range((lo, Bound::Unbounded))
1182 12169 : .filter_map(|(_, v)| {
1183 8798 : let mut v = v.clone();
1184 8798 : if kind == Kind::Entity && crate::store::filter::strip_expired(&mut v, &now)
1185 : {
1186 8 : return None;
1187 8790 : }
1188 8790 : Some(v)
1189 8798 : })
1190 8951 : .take(limit)
1191 8951 : .collect()
1192 8951 : })
1193 40233 : .unwrap_or_default()
1194 40233 : }
1195 :
1196 : /// Query Entities on the stored documents, under the read
1197 : /// lock: ids, the idPattern literal, types and attrs are tested on the
1198 : /// borrowed documents and only what the answer holds is copied, where a
1199 : /// snapshot of the tenant used to be. ids, types and attrs are exact
1200 : /// here, and so is a `q` without linked-entity terms when the request
1201 : /// @context comes along: it runs through the API's own evaluator.
1202 : /// Without anything else (`scopeQ`, a geo query, a linked
1203 : /// term) the outcome is `decided`: the projection applies and a page is
1204 : /// cut in id order (the map's order), counted when asked. Otherwise the
1205 : /// caller stays the arbiter over the narrowed rows. 4.22: an expired
1206 : /// Entity is not a match, before any cut; its expired instances are the
1207 : /// caller's.
1208 2989 : pub fn query_entities(
1209 2989 : &self,
1210 2989 : tenant: &TenantId,
1211 2989 : f: &filter::EntityFilter<'_>,
1212 2989 : ) -> filter::QueryOutcome {
1213 2989 : let now = crate::store::any::now_utc();
1214 : // a linked term reads other Entities under the caller's narrowing
1215 2989 : let q = match (f.q, f.ctx) {
1216 430 : (Some(q), Some(ctx)) if !q.has_linked_paths() => Some((q, ctx)),
1217 2579 : _ => None,
1218 : };
1219 2989 : let decided = (f.q.is_none() || q.is_some()) && f.scope_q.is_none() && f.geo.is_empty();
1220 2989 : let inner = self
1221 2989 : .inner
1222 2989 : .read()
1223 2989 : .unwrap_or_else(std::sync::PoisonError::into_inner);
1224 2989 : let Some(m) = Self::map(&inner, Kind::Entity).get(tenant.as_str()) else {
1225 : return filter::QueryOutcome {
1226 537 : rows: Vec::new(),
1227 537 : decided,
1228 537 : paged: decided && f.page.is_some(),
1229 537 : total: f.page.as_ref().filter(|_| decided).map(|_| 0),
1230 : };
1231 : };
1232 : // a pattern that does not compile matches nothing (`Rule::id_patterns`)
1233 2452 : let id_res: Option<Vec<_>> = f.id_patterns.map(|ps| {
1234 4 : ps.iter()
1235 6 : .filter_map(|p| antares_ql::regex::compile(&format!("^(?:{p})$")).ok())
1236 4 : .collect()
1237 4 : });
1238 758970 : let matches = |id: &str, doc: &Value| {
1239 758970 : !filter::expired_at(doc, &now)
1240 758946 : && id_res
1241 758946 : .as_ref()
1242 758946 : .is_none_or(|res| res.iter().any(|re| re.is_match(id)))
1243 758940 : && f.id_literal.is_none_or(|l| {
1244 208 : if l.anchored {
1245 128 : id.starts_with(l.text)
1246 : } else {
1247 80 : id.contains(l.text)
1248 : }
1249 208 : })
1250 758858 : && f.types.is_none_or(|sel| {
1251 756136 : let held: Vec<&str> = doc["type"]
1252 756136 : .as_array()
1253 756136 : .map(|a| a.iter().filter_map(Value::as_str).collect())
1254 756136 : .unwrap_or_default();
1255 756136 : sel.iter()
1256 878900 : .any(|group| group.iter().all(|w| held.contains(&w.as_str())))
1257 756136 : })
1258 61926 : && f.attrs
1259 61926 : .is_none_or(|attrs| attrs.iter().any(|a| has_live_attr(doc, a, &now)))
1260 60838 : && q.is_none_or(|(q, ctx)| {
1261 : // 4.22: `q` judges what a read may see, so an instance
1262 : // past its expiresAt never makes the Entity match;
1263 : // only a doc carrying one pays the copy
1264 23580 : if holds_expired_instance(doc, &now) {
1265 12 : let mut live = doc.clone();
1266 12 : filter::strip_expired(&mut live, &now);
1267 12 : antares_ql::eval::eval_q(q, &live, ctx, &|_| None)
1268 : } else {
1269 23568 : antares_ql::eval::eval_q(q, doc, ctx, &|_| None)
1270 : }
1271 23580 : })
1272 758970 : };
1273 : // ids name at most what the client sent: look each up, in id order
1274 2452 : let candidates: Box<dyn Iterator<Item = (&String, &Value)>> = match (f.ids, f.after_id) {
1275 318 : (Some(ids), after) => {
1276 318 : let mut want: Vec<&str> = ids.to_vec();
1277 318 : want.sort_unstable();
1278 318 : want.dedup();
1279 766 : want.retain(|id| after.is_none_or(|a| *id > a));
1280 762 : Box::new(want.into_iter().filter_map(|id| m.get_key_value(id)))
1281 : }
1282 76 : (None, Some(a)) => Box::new(
1283 76 : m.range::<str, _>((std::ops::Bound::Excluded(a), std::ops::Bound::Unbounded)),
1284 76 : ),
1285 2058 : (None, None) => Box::new(m.iter()),
1286 : };
1287 2452 : let hits = candidates
1288 758970 : .filter(|(id, doc)| matches(id, doc))
1289 2452 : .map(|(_, d)| d);
1290 30704 : let project = |doc: &Value| {
1291 30704 : let mut v = doc.clone();
1292 30704 : if decided {
1293 22966 : if let Some(o) = v.as_object_mut() {
1294 22966 : if let Some(keep) = f.keep_attrs {
1295 7640 : o.retain(|k, _| !k.starts_with("http") || keep.contains(k));
1296 21462 : } else if let Some(drop) = f.drop_attrs {
1297 116 : o.retain(|k, _| !drop.contains(k));
1298 21446 : }
1299 0 : }
1300 7738 : }
1301 30704 : v
1302 30704 : };
1303 2452 : let Some(p) = f.page.as_ref().filter(|_| decided) else {
1304 634 : return filter::QueryOutcome {
1305 634 : rows: hits.map(project).collect(),
1306 634 : decided,
1307 634 : paged: false,
1308 634 : total: None,
1309 634 : };
1310 : };
1311 1818 : let offset = usize::try_from(p.offset).unwrap_or(0);
1312 1818 : let limit = usize::try_from(p.limit).unwrap_or(usize::MAX);
1313 1818 : let mut rows: Vec<Value> = Vec::new();
1314 1818 : let (mut matched, mut more) = (0usize, false);
1315 29064 : for doc in hits {
1316 29064 : if matched >= offset && rows.len() < limit {
1317 20164 : rows.push(project(doc));
1318 20164 : } else if rows.len() == limit && !p.count {
1319 190 : more = true;
1320 190 : break;
1321 8710 : }
1322 : // the rest is counted, never copied
1323 28874 : matched = matched.saturating_add(1);
1324 : }
1325 1818 : let total = if p.count {
1326 145 : matched
1327 : } else {
1328 : // the smallest total consistent with "a next page exists"
1329 1673 : offset
1330 1673 : .saturating_add(rows.len())
1331 1673 : .saturating_add(usize::from(more))
1332 : };
1333 1818 : filter::QueryOutcome {
1334 1818 : rows,
1335 1818 : decided,
1336 1818 : paged: true,
1337 1818 : total: Some(i64::try_from(total).unwrap_or(i64::MAX)),
1338 1818 : }
1339 2989 : }
1340 :
1341 : /// One id-ordered window of docs and the size of the whole set. The
1342 : /// per-tenant map is a `BTreeMap` keyed by id, so the window is a skip
1343 : /// over an already-ordered iterator.
1344 214 : pub fn list_slice(
1345 214 : &self,
1346 214 : tenant: &TenantId,
1347 214 : kind: Kind,
1348 214 : offset: usize,
1349 214 : limit: usize,
1350 214 : ) -> (Vec<Value>, usize) {
1351 214 : let inner = self
1352 214 : .inner
1353 214 : .read()
1354 214 : .unwrap_or_else(std::sync::PoisonError::into_inner);
1355 214 : Self::map(&inner, kind)
1356 214 : .get(tenant.as_str())
1357 214 : .map(|m| {
1358 138 : (
1359 138 : m.values().skip(offset).take(limit).cloned().collect(),
1360 138 : m.len(),
1361 138 : )
1362 138 : })
1363 214 : .unwrap_or_default()
1364 214 : }
1365 :
1366 : /// Read-modify-write on one document. Returns `None` when absent; the
1367 : /// closure's error aborts without writing.
1368 154339 : pub async fn mutate<T, E>(
1369 154339 : &self,
1370 154339 : tenant: &TenantId,
1371 154339 : kind: Kind,
1372 154339 : id: &str,
1373 154339 : f: impl FnOnce(&mut Value) -> Result<T, E>,
1374 154339 : ) -> Option<Result<T, E>> {
1375 154339 : let _order = match kind {
1376 22236 : Kind::Entity => Some(self.emit_ordered().await),
1377 132103 : _ => None,
1378 : };
1379 154339 : let (result, change) = on_blocking(self.shadow.is_some(), || {
1380 154339 : let mut inner = self.write_inner();
1381 154339 : if self.is_expired(&inner, kind, tenant.as_str(), id) {
1382 6 : return None; // 4.22: invalid, so absent
1383 154333 : }
1384 154333 : let doc = Self::map_mut(&mut inner, kind)
1385 154333 : .get_mut(tenant.as_str())
1386 154333 : .and_then(|m| m.get_mut(id))?;
1387 150951 : let before = doc.clone();
1388 150951 : let mut candidate = doc.clone();
1389 150951 : Some(match f(&mut candidate) {
1390 150797 : Ok(t) => {
1391 150797 : if candidate != before {
1392 146693 : self.persist(
1393 146693 : table_for(kind),
1394 146693 : &key_bytes(tenant.as_str(), id),
1395 146693 : Some(&candidate),
1396 146693 : );
1397 146693 : }
1398 150797 : let change = (kind == Kind::Entity && candidate != before)
1399 150797 : .then(|| (before, candidate.clone()));
1400 150797 : *doc = candidate;
1401 150797 : (Ok(t), change)
1402 : }
1403 154 : Err(e) => (Err(e), None),
1404 : })
1405 154339 : })?;
1406 150951 : if let Some((b, a)) = change {
1407 17924 : self.emit(tenant, Some(b), Some(a)).await;
1408 133027 : }
1409 150951 : Some(result)
1410 154339 : }
1411 :
1412 : // jsonldContexts: one keyspace for the whole process (key = context id, no
1413 : // tenant prefix) — Cached rows are copies of public documents shared by
1414 : // every tenant. Ownership of the tenant-authored kinds (Hosted,
1415 : // ImplicitlyCreated, 5.13.1) travels in the document's "owner" member,
1416 : // and every call below is answered through it (ADR-0021): the Postgres
1417 : // arm has the same rule as a Row-Level Security policy over a generated
1418 : // `tenant_id` column, and this arm, which has no policy engine under it,
1419 : // applies it in the store.
1420 2492 : pub fn context_put(
1421 2492 : &self,
1422 2492 : tenant: Option<&TenantId>,
1423 2492 : id: &str,
1424 2492 : doc: Value,
1425 2492 : ) -> Result<(), NgsiError> {
1426 2492 : on_blocking(self.shadow.is_some(), || {
1427 2492 : let mut inner = self.write_inner();
1428 : // The row that is there decides whether this call may replace it,
1429 : // the row that arrives decides whether it may be stored: the two
1430 : // halves of the policy the Postgres arm writes as USING and WITH
1431 : // CHECK. Both are backstops — every caller mints its own id.
1432 2492 : let held = inner.contexts.get(id);
1433 2492 : if !held.is_none_or(|r| context_row_visible(r, tenant))
1434 2490 : || !context_row_visible(&doc, tenant)
1435 : {
1436 2 : return Err(NgsiError::InternalError(
1437 2 : "@context belongs to another tenant".into(),
1438 2 : ));
1439 2490 : }
1440 2490 : let mut staged = Vec::new();
1441 2490 : self.stage(
1442 2490 : &mut staged,
1443 : T_JSONLD_CONTEXTS,
1444 2490 : id.as_bytes().to_vec(),
1445 2490 : Some(&doc),
1446 : );
1447 2490 : let cached = doc.get("kind").and_then(Value::as_str) == Some("Cached");
1448 2490 : inner.contexts.insert(id.to_owned(), doc);
1449 : // 5.13.1: "Implementations shall periodically invalidate the
1450 : // 'Cached' @contexts." One entry is stored per distinct external
1451 : // URL a request references, so without a ceiling a client that
1452 : // references fresh URLs grows this keyspace (and, in `file` mode,
1453 : // the store on disk) forever. Oldest-first, and only for the
1454 : // Cached kind — Hosted/ImplicitlyCreated entries are resources
1455 : // the broker serves on demand, never a cache.
1456 2490 : if cached && inner.contexts.len() > MAX_CACHED_CONTEXTS {
1457 4 : let dead = oldest_cached(&inner.contexts);
1458 4 : for id in dead {
1459 2 : self.stage(&mut staged, T_JSONLD_CONTEXTS, id.as_bytes().to_vec(), None);
1460 2 : inner.contexts.remove(&id);
1461 2 : }
1462 2486 : }
1463 : // the entry and the evictions it caused, in one commit
1464 2490 : self.commit_staged(&staged);
1465 2490 : Ok(())
1466 2492 : })
1467 2492 : }
1468 :
1469 932 : pub fn context_get(&self, tenant: Option<&TenantId>, id: &str) -> Option<Value> {
1470 932 : self.inner
1471 932 : .read()
1472 932 : .unwrap_or_else(std::sync::PoisonError::into_inner)
1473 932 : .contexts
1474 932 : .get(id)
1475 932 : .filter(|r| context_row_visible(r, tenant))
1476 932 : .cloned()
1477 932 : }
1478 :
1479 : /// Read-modify-write under the write lock; `None` when the row is absent
1480 : /// or another Tenant's, `Some(Err)` when `f` refused (nothing written).
1481 10730 : pub fn context_mutate(
1482 10730 : &self,
1483 10730 : tenant: Option<&TenantId>,
1484 10730 : id: &str,
1485 10730 : f: impl FnOnce(&mut Value) -> Result<(), ()>,
1486 10730 : ) -> Option<Result<(), ()>> {
1487 10730 : on_blocking(self.shadow.is_some(), || {
1488 10730 : let mut inner = self.write_inner();
1489 10730 : let row = inner
1490 10730 : .contexts
1491 10730 : .get_mut(id)
1492 10730 : .filter(|r| context_row_visible(r, tenant))?;
1493 10474 : let mut doc = row.clone();
1494 10474 : if f(&mut doc).is_err() {
1495 0 : return Some(Err(()));
1496 10474 : }
1497 10474 : self.persist(T_JSONLD_CONTEXTS, id.as_bytes(), Some(&doc));
1498 10474 : *row = doc;
1499 10474 : Some(Ok(()))
1500 10730 : })
1501 10730 : }
1502 :
1503 230 : pub fn context_delete(&self, tenant: Option<&TenantId>, id: &str) -> bool {
1504 230 : on_blocking(self.shadow.is_some(), || {
1505 230 : let mut inner = self.write_inner();
1506 230 : let hit = inner
1507 230 : .contexts
1508 230 : .get(id)
1509 230 : .is_some_and(|r| context_row_visible(r, tenant));
1510 230 : if hit {
1511 224 : inner.contexts.remove(id);
1512 224 : self.persist(T_JSONLD_CONTEXTS, id.as_bytes(), None);
1513 224 : }
1514 230 : hit
1515 230 : })
1516 230 : }
1517 :
1518 : /// Every row without its `body` member (the `@context` document itself).
1519 : /// A body may be 5 MiB and only the `Cached` rows are capped in number,
1520 : /// so cloning whole rows here was a multi-gigabyte copy on the boot path.
1521 204 : pub fn context_list_meta(&self, tenant: Option<&TenantId>) -> Vec<Value> {
1522 204 : self.inner
1523 204 : .read()
1524 204 : .unwrap_or_else(std::sync::PoisonError::into_inner)
1525 204 : .contexts
1526 204 : .values()
1527 4680 : .filter(|v| context_row_visible(v, tenant))
1528 4564 : .map(|v| {
1529 4564 : let mut row = v.clone();
1530 4564 : if let Some(o) = row.as_object_mut() {
1531 4564 : o.remove("body");
1532 4564 : }
1533 4564 : row
1534 4564 : })
1535 204 : .collect()
1536 204 : }
1537 : }
1538 :
1539 : /// Whether `doc` holds Attribute `iri` once 4.22 has stripped its expired
1540 : /// instances: an instance array keeps it while one instance is live, as
1541 : /// `filter::strip_expired` decides.
1542 1182 : fn has_live_attr(doc: &Value, iri: &str, now: &str) -> bool {
1543 1182 : match doc.get(iri) {
1544 94 : Some(Value::Array(arr)) if iri.contains(':') => {
1545 94 : arr.is_empty() || arr.iter().any(|i| !filter::expired_at(i, now))
1546 : }
1547 0 : Some(_) => true,
1548 1088 : None => false,
1549 : }
1550 1182 : }
1551 :
1552 : /// Whether an Attribute instance of `doc`, a sub-Attribute's included, is
1553 : /// past its `expiresAt`: what `filter::strip_expired` would remove.
1554 53056 : fn holds_expired_instance(doc: &Value, now: &str) -> bool {
1555 53056 : doc.as_object().is_some_and(|o| {
1556 53056 : o.iter()
1557 138272 : .filter(|(k, _)| k.contains(':'))
1558 53056 : .filter_map(|(_, v)| v.as_array())
1559 53056 : .flatten()
1560 53056 : .any(|i| filter::expired_at(i, now) || holds_expired_instance(i, now))
1561 53056 : })
1562 53056 : }
1563 :
1564 : #[cfg(test)]
1565 : mod tests {
1566 : use super::*;
1567 : use serde_json::json;
1568 :
1569 : /// A change hook that PANICS must cost only its own caller: the panic
1570 : /// unwinds through that writer, and every later writer finds the store
1571 : /// usable — locks recover from the poisoning instead of turning one bad
1572 : /// hook into a dead store.
1573 : #[tokio::test(flavor = "multi_thread")]
1574 2 : async fn a_panicking_hook_leaves_the_store_usable() {
1575 : use std::sync::Arc;
1576 2 : let s = Arc::new(Store::default());
1577 2 : s.set_change_hook(Arc::new(
1578 : |_t: &TenantId,
1579 : _b: Option<Value>,
1580 : after: Option<Value>,
1581 : _v: antares_store::ChangeVersion|
1582 4 : -> antares_store::HookFuture<'_> {
1583 4 : Box::pin(async move {
1584 4 : if after.as_ref().and_then(|a| a.get("boom")).is_some() {
1585 2 : panic!("hook panic");
1586 2 : }
1587 2 : })
1588 4 : },
1589 : ));
1590 2 : let t = TenantId::new("hook-panic").expect("tenant");
1591 2 : let s2 = s.clone();
1592 2 : let t2 = t.clone();
1593 2 : let poisoned = tokio::spawn(async move {
1594 2 : s2.upsert(
1595 2 : &t2,
1596 2 : Kind::Entity,
1597 2 : "urn:x:boom",
1598 2 : json!({"id": "urn:x:boom", "type": ["T"], "boom": true}),
1599 2 : )
1600 2 : .await;
1601 2 : })
1602 2 : .await;
1603 2 : assert!(poisoned.is_err(), "the hook's panic reaches its own caller");
1604 : // the NEXT writer and reader are untouched
1605 2 : assert!(
1606 2 : s.create(
1607 2 : &t,
1608 2 : Kind::Entity,
1609 2 : "urn:x:after",
1610 2 : json!({"id": "urn:x:after", "type": ["T"]})
1611 2 : )
1612 2 : .await
1613 : );
1614 2 : assert!(
1615 2 : s.get(&t, Kind::Entity, "urn:x:after").is_some(),
1616 : "one panicking hook must not take the store down"
1617 : );
1618 2 : assert!(
1619 2 : s.get(&t, Kind::Entity, "urn:x:boom").is_some(),
1620 2 : "the write that triggered the panic still committed — the hook runs after the commit"
1621 2 : );
1622 2 : }
1623 :
1624 : /// Two writers to the same entity: the change hook fires in COMMIT
1625 : /// order. The first writer's hook is gated open only after the second
1626 : /// writer has had every chance to overtake it — if the second commit may
1627 : /// emit before the first commit's emit, the notification pipeline sees
1628 : /// the versions reversed and records stale state as newest.
1629 : #[tokio::test(flavor = "multi_thread")]
1630 2 : async fn change_hook_fires_in_commit_order_per_entity() {
1631 : use std::sync::atomic::{AtomicBool, Ordering};
1632 : use std::sync::Arc;
1633 2 : let s = Arc::new(Store::default());
1634 2 : let seen: Arc<std::sync::Mutex<Vec<i64>>> = Arc::default();
1635 2 : let entered = Arc::new(AtomicBool::new(false));
1636 2 : let released = Arc::new(AtomicBool::new(false));
1637 : {
1638 2 : let (seen, entered, released) = (seen.clone(), entered.clone(), released.clone());
1639 2 : s.set_change_hook(Arc::new(
1640 : move |_t: &TenantId,
1641 : _b: Option<Value>,
1642 : after: Option<Value>,
1643 : _v: antares_store::ChangeVersion|
1644 4 : -> antares_store::HookFuture<'_> {
1645 4 : let (seen, entered, released) =
1646 4 : (seen.clone(), entered.clone(), released.clone());
1647 4 : Box::pin(async move {
1648 4 : let v = after
1649 4 : .as_ref()
1650 4 : .and_then(|a| a.get("v"))
1651 4 : .and_then(Value::as_i64)
1652 4 : .unwrap_or(-1);
1653 4 : if v == 1 {
1654 2 : entered.store(true, Ordering::SeqCst);
1655 63 : while !released.load(Ordering::SeqCst) {
1656 61 : tokio::time::sleep(std::time::Duration::from_millis(1)).await;
1657 : }
1658 2 : }
1659 4 : seen.lock()
1660 4 : .unwrap_or_else(std::sync::PoisonError::into_inner)
1661 4 : .push(v);
1662 4 : })
1663 4 : },
1664 : ));
1665 : }
1666 2 : let t = TenantId::new("emit-order").expect("tenant");
1667 4 : let doc = |v: i64| json!({"id": "urn:x:1", "type": ["T"], "v": v});
1668 2 : let s1 = s.clone();
1669 2 : let t1c = t.clone();
1670 2 : let w1 = tokio::spawn(async move {
1671 2 : s1.upsert(&t1c, Kind::Entity, "urn:x:1", doc(1)).await;
1672 2 : });
1673 4 : while !entered.load(std::sync::atomic::Ordering::SeqCst) {
1674 2 : tokio::time::sleep(std::time::Duration::from_millis(1)).await;
1675 : }
1676 : // writer 1 has committed and sits inside its hook. Writer 2 now has
1677 : // every chance to commit AND emit before writer 1's emit finishes.
1678 2 : let s2 = s.clone();
1679 2 : let t2c = t.clone();
1680 2 : let w2 = tokio::spawn(async move {
1681 2 : s2.upsert(&t2c, Kind::Entity, "urn:x:1", doc(2)).await;
1682 2 : });
1683 2 : tokio::time::sleep(std::time::Duration::from_millis(60)).await;
1684 2 : released.store(true, std::sync::atomic::Ordering::SeqCst);
1685 2 : w1.await.expect("writer 1");
1686 2 : w2.await.expect("writer 2");
1687 2 : assert_eq!(
1688 2 : *seen
1689 2 : .lock()
1690 2 : .unwrap_or_else(std::sync::PoisonError::into_inner),
1691 2 : vec![1, 2],
1692 2 : "hooks must fire in commit order, or the consumer records stale state as newest"
1693 2 : );
1694 2 : }
1695 :
1696 : #[tokio::test]
1697 2 : async fn commit_queue_counts_writers() {
1698 : // Every write passes through the counted critical section.
1699 2 : let s = Store::default();
1700 2 : assert_eq!(s.commit_queue(), (0, 0));
1701 2 : let t = TenantId::new("t").unwrap();
1702 2 : s.create(&t, Kind::Entity, "urn:a", json!({"id": "urn:a"}))
1703 2 : .await;
1704 2 : let (depth, peak) = s.commit_queue();
1705 2 : assert_eq!(depth, 0, "no writer in flight after the call returns");
1706 2 : assert!(peak >= 1, "the write itself must register in the peak");
1707 2 : }
1708 :
1709 : /// What a store reports about itself, for an operator reading
1710 : /// `/q/health`: the memory and file modes are one backend with two
1711 : /// durability shapes, and the health body must not present them as the
1712 : /// same thing.
1713 : #[test]
1714 2 : fn a_store_reports_the_engine_it_actually_runs() {
1715 2 : let dir = tempdir("engine");
1716 2 : let mem = crate::store::any::AnyStore::Mem(Store::default());
1717 2 : let file = crate::store::any::AnyStore::Mem(Store::open_file(&dir).expect("open"));
1718 2 : assert_eq!(mem.version_info()["engine"], "memory");
1719 2 : assert_eq!(file.version_info()["engine"], "redb");
1720 2 : }
1721 :
1722 : /// The commit queue is a `file`-mode signal: it exists because redb has
1723 : /// one writer and commits fsync through it. A pure in-memory store has no
1724 : /// such committer, so it reports nothing rather than a number that would
1725 : /// read as the same thing and mean something else.
1726 : #[test]
1727 2 : fn only_a_shadowed_store_reports_a_commit_queue() {
1728 : use crate::store::any::AnyStore;
1729 2 : assert_eq!(AnyStore::Mem(Store::default()).commit_queue(), None);
1730 2 : let dir = tempdir("commit-queue");
1731 2 : let s = Store::open_file(&dir).expect("open");
1732 2 : assert!(
1733 2 : AnyStore::Mem(s).commit_queue().is_some(),
1734 : "a durable store reports the queue behind its single committer"
1735 : );
1736 2 : }
1737 :
1738 : #[tokio::test]
1739 2 : async fn tenant_isolation() {
1740 2 : let s = Store::default();
1741 2 : let t1 = TenantId::new("t1").unwrap();
1742 2 : let t2 = TenantId::new("t2").unwrap();
1743 2 : assert!(
1744 2 : s.create(&t1, Kind::Entity, "urn:a", json!({"id": "urn:a"}))
1745 2 : .await
1746 : );
1747 2 : assert!(s.get(&t2, Kind::Entity, "urn:a").is_none());
1748 2 : assert!(s.get(&t1, Kind::Entity, "urn:a").is_some());
1749 2 : assert!(!s.create(&t1, Kind::Entity, "urn:a", json!({})).await);
1750 2 : assert!(s.create(&t2, Kind::Entity, "urn:a", json!({})).await);
1751 2 : }
1752 :
1753 : /// 4.6.3 allows a comma as the seconds-fraction separator. Byte order
1754 : /// puts ',' (0x2C) before both '.' and 'Z', so a comma-form instance that
1755 : /// has NOT expired reads as expired against the point-form `now` this
1756 : /// store stamps — the write path would drop a live instance, and the
1757 : /// postgres path (which parses through `try_timestamptz`) would keep it.
1758 : #[tokio::test]
1759 2 : async fn a_comma_fraction_instance_is_not_pruned_before_it_expires() {
1760 2 : let now = "2026-09-01T12:00:00.000Z";
1761 2 : let mut live = json!({
1762 2 : "id": "urn:x", "type": ["T"],
1763 2 : "https://a/attr": [{"value": 1, "expiresAt": "2026-09-01T12:00:00,500Z"}]
1764 : });
1765 2 : assert!(
1766 2 : !prune_expired_instances(&mut live, now),
1767 : "live instance kept"
1768 : );
1769 2 : assert_eq!(live["https://a/attr"].as_array().map(Vec::len), Some(1));
1770 :
1771 : // one that HAS expired still goes, comma form included, and takes the
1772 : // attribute with it when it was the last instance
1773 2 : let mut gone = json!({
1774 2 : "id": "urn:x", "type": ["T"],
1775 2 : "https://a/attr": [{"value": 1, "expiresAt": "2026-09-01T11:59:59,500Z"}]
1776 : });
1777 2 : assert!(prune_expired_instances(&mut gone, now));
1778 2 : assert!(gone.get("https://a/attr").is_none());
1779 :
1780 : // the entity-level sweep reads the same stamp the same way
1781 2 : let s = Store::default();
1782 2 : let t = TenantId::new("t").expect("tenant");
1783 2 : assert!(
1784 2 : s.create(
1785 2 : &t,
1786 2 : Kind::Entity,
1787 2 : "urn:live",
1788 2 : json!({"id": "urn:live", "type": ["T"], "expiresAt": "2026-09-01T12:00:00,500Z"})
1789 2 : )
1790 2 : .await
1791 : );
1792 2 : assert!(
1793 2 : s.create(
1794 2 : &t,
1795 2 : Kind::Entity,
1796 2 : "urn:gone",
1797 2 : json!({"id": "urn:gone", "type": ["T"], "expiresAt": "2026-09-01T11:59:59,500Z"})
1798 2 : )
1799 2 : .await
1800 : );
1801 2 : assert_eq!(s.sweep_expired(now), 1, "only the expired one is reaped");
1802 2 : assert!(s.get(&t, Kind::Entity, "urn:live").is_some());
1803 2 : assert!(s.get(&t, Kind::Entity, "urn:gone").is_none());
1804 2 : }
1805 :
1806 : /// The same assertion over EVERY kind and the whole read/write surface,
1807 : /// with the SAME id on both sides — the shape a tenant-blind map gets
1808 : /// wrong, because it answers the neighbour's document instead of `None`.
1809 : #[tokio::test]
1810 2 : async fn no_kind_leaks_across_tenants() {
1811 2 : let s = Store::default();
1812 2 : let a = TenantId::new("t-a").expect("tenant");
1813 2 : let b = TenantId::new("t-b").expect("tenant");
1814 2 : let id = "urn:ngsi-ld:shared:1";
1815 20 : for kind in ALL_KINDS {
1816 20 : assert!(
1817 20 : s.create(&a, kind, id, json!({"id": id, "owner": "a"}))
1818 20 : .await
1819 2 : );
1820 2 :
1821 20 : assert!(s.get(&b, kind, id).is_none(), "{kind:?}: get");
1822 20 : assert!(s.list(&b, kind).is_empty(), "{kind:?}: list");
1823 20 : assert!(
1824 20 : s.list_page(&b, kind, None, 100).is_empty(),
1825 2 : "{kind:?}: list_page"
1826 2 : );
1827 20 : assert_eq!(
1828 20 : s.list_slice(&b, kind, 0, 100),
1829 20 : (Vec::new(), 0),
1830 2 : "{kind:?}: list_slice"
1831 2 : );
1832 2 :
1833 2 : // no write reaches the neighbour's document either
1834 20 : assert!(!s.delete(&b, kind, id).await, "{kind:?}: delete");
1835 20 : assert!(
1836 20 : s.mutate::<(), ()>(&b, kind, id, |d| {
1837 2 : d["owner"] = json!("b");
1838 2 : Ok(())
1839 2 : })
1840 20 : .await
1841 20 : .is_none(),
1842 2 : "{kind:?}: mutate"
1843 2 : );
1844 2 :
1845 2 : // the same id under another tenant is a create, not a conflict
1846 20 : assert!(
1847 20 : s.create(&b, kind, id, json!({"id": id, "owner": "b"}))
1848 20 : .await
1849 2 : );
1850 20 : assert_eq!(s.get(&a, kind, id).expect("a keeps its own")["owner"], "a");
1851 20 : assert_eq!(s.get(&b, kind, id).expect("b has its own")["owner"], "b");
1852 2 :
1853 2 : // and b deleting its own leaves a's untouched
1854 20 : assert!(s.delete(&b, kind, id).await, "{kind:?}: delete own");
1855 20 : assert!(s.get(&b, kind, id).is_none(), "{kind:?}: deleted");
1856 20 : assert_eq!(
1857 20 : s.get(&a, kind, id)
1858 20 : .expect("survives the neighbour's delete")["owner"],
1859 2 : "a"
1860 2 : );
1861 20 : assert!(s.delete(&a, kind, id).await, "{kind:?}: cleanup");
1862 2 : }
1863 2 : }
1864 :
1865 28 : fn tempdir(name: &str) -> std::path::PathBuf {
1866 28 : let dir = std::env::temp_dir().join(format!("antares-store-{name}-{}", std::process::id()));
1867 28 : let _ = std::fs::remove_dir_all(&dir);
1868 28 : std::fs::create_dir_all(&dir).expect("tempdir");
1869 28 : dir
1870 28 : }
1871 :
1872 : /// `open_file` promises "Any open/format/decode error refuses to start —
1873 : /// never silently serve partial data", and the boot rebuild's kind loop
1874 : /// keeps it: an absent table has nothing to load, any other error refuses.
1875 : /// The @context table is read as `if let Ok(...)` instead, so a table that
1876 : /// exists and cannot be opened counts as an empty one — and that is the
1877 : /// table whose absence costs a Tenant its own documents, since Hosted and
1878 : /// ImplicitlyCreated @contexts (5.13.1) hold term mappings authored
1879 : /// through its requests. The broker would come up and serve without them.
1880 : ///
1881 : /// A table of the same name under different key/value types is redb's
1882 : /// `TableTypeMismatch`: the shape a file written by another binary, or by
1883 : /// an older schema, actually has.
1884 : #[test]
1885 2 : fn an_unreadable_table_refuses_the_boot_instead_of_reading_as_empty() {
1886 2 : let dir = tempdir("unreadable-context-table");
1887 2 : let path = dir.join("antares.redb");
1888 2 : {
1889 2 : let db = Database::create(&path).expect("db");
1890 2 : let mut tx = db.begin_write().expect("tx");
1891 2 : tx.set_durability(Durability::Immediate).expect("dur");
1892 2 : {
1893 2 : let mut m = tx.open_table(T_META).expect("meta");
1894 2 : m.insert("format", FORMAT_VERSION).expect("insert");
1895 2 : }
1896 2 : {
1897 2 : let def = ::redb::TableDefinition::<&str, &str>::new("jsonld_contexts");
1898 2 : let mut t = tx.open_table(def).expect("mismatched contexts table");
1899 2 : t.insert("urn:ngsi-ld:ctx", "{}").expect("insert");
1900 2 : }
1901 2 : tx.commit().expect("commit");
1902 2 : }
1903 2 : let db = Database::create(&path).expect("reopen");
1904 2 : let Err(err) = Store::from_database(db, &path.display().to_string()) else {
1905 0 : panic!("a table that cannot be opened must refuse the boot, not start without it");
1906 : };
1907 2 : assert!(!err.is_empty(), "the refusal says what failed");
1908 2 : }
1909 :
1910 : /// The same read, in the guard that exists to refuse a file this binary
1911 : /// cannot vouch for: data present with no format marker. Skipping a table
1912 : /// it cannot open makes the guard conclude the file is empty and stamp
1913 : /// this binary's marker onto it — a write into a file it is about to
1914 : /// declare unreadable, and one that costs the operator the guard's own
1915 : /// diagnosis on every later start.
1916 : #[test]
1917 2 : fn the_marker_guard_does_not_stamp_a_file_it_could_not_read() {
1918 2 : let dir = tempdir("unstamped-unreadable");
1919 2 : let path = dir.join("antares.redb");
1920 2 : {
1921 2 : let db = Database::create(&path).expect("db");
1922 2 : let mut tx = db.begin_write().expect("tx");
1923 2 : tx.set_durability(Durability::Immediate).expect("dur");
1924 2 : let def = ::redb::TableDefinition::<&str, &str>::new("entities");
1925 2 : let mut t = tx.open_table(def).expect("mismatched entities table");
1926 2 : t.insert("k", "v").expect("insert");
1927 2 : drop(t);
1928 2 : tx.commit().expect("commit");
1929 2 : }
1930 2 : let db = Database::create(&path).expect("reopen");
1931 2 : let Err(_) = Store::from_database(db, &path.display().to_string()) else {
1932 0 : panic!("an unreadable table with no marker must refuse the boot");
1933 : };
1934 2 : let db = Database::create(&path).expect("reopen after the refusal");
1935 2 : let rt = db.begin_read().expect("read tx");
1936 2 : let stamped = match rt.open_table(T_META) {
1937 0 : Ok(m) => m.get("format").expect("get").is_some(),
1938 2 : Err(::redb::TableError::TableDoesNotExist(_)) => false,
1939 0 : Err(e) => panic!("{e}"),
1940 : };
1941 2 : assert!(
1942 2 : !stamped,
1943 : "a file the guard could not read must not carry this binary's format marker"
1944 : );
1945 2 : }
1946 :
1947 : /// Every kind + contexts round-trip through a close/reopen.
1948 : #[tokio::test]
1949 2 : async fn file_mode_survives_reopen() {
1950 2 : let dir = tempdir("reopen");
1951 2 : let t = TenantId::new("tenant_a-1").expect("tenant");
1952 : {
1953 2 : let s = Store::open_file(&dir).expect("open");
1954 20 : for kind in ALL_KINDS {
1955 20 : assert!(
1956 20 : s.create(&t, kind, "urn:x:1", json!({"kind": format!("{kind:?}")}))
1957 20 : .await
1958 : );
1959 : }
1960 : // a row the writers actually produce: the kind is what decides
1961 : // whose it is (ADR-0021), and a Cached copy belongs to no Tenant
1962 2 : s.context_put(
1963 2 : None,
1964 2 : "ctx1",
1965 2 : json!({"localId": "ctx1", "kind": "Cached", "body": {"@context": {}}}),
1966 : )
1967 2 : .expect("store");
1968 : }
1969 2 : let s = Store::open_file(&dir).expect("reopen");
1970 20 : for kind in ALL_KINDS {
1971 20 : assert_eq!(
1972 20 : s.get(&t, kind, "urn:x:1").expect("survives")["kind"],
1973 20 : format!("{kind:?}")
1974 : );
1975 : }
1976 2 : assert!(s.context_get(None, "ctx1").is_some());
1977 : // tenant isolation intact after rebuild
1978 2 : assert!(s
1979 2 : .get(&TenantId::default(), Kind::Entity, "urn:x:1")
1980 2 : .is_none());
1981 2 : let _ = std::fs::remove_dir_all(&dir);
1982 2 : }
1983 :
1984 : /// Deletes and mutations reach redb — no phantom state after restart.
1985 : #[tokio::test]
1986 2 : async fn file_mode_deletes_and_updates_persist() {
1987 2 : let dir = tempdir("delete");
1988 2 : let t = TenantId::default();
1989 : {
1990 2 : let s = Store::open_file(&dir).expect("open");
1991 2 : s.create(&t, Kind::Entity, "urn:gone", json!({"n": 1}))
1992 2 : .await;
1993 2 : s.create(&t, Kind::Entity, "urn:kept", json!({"n": 1}))
1994 2 : .await;
1995 2 : assert!(s.delete(&t, Kind::Entity, "urn:gone").await);
1996 2 : let r: Option<Result<(), ()>> = s
1997 2 : .mutate(&t, Kind::Entity, "urn:kept", |d| {
1998 2 : d["n"] = json!(2);
1999 2 : Ok(())
2000 2 : })
2001 2 : .await;
2002 2 : assert!(matches!(r, Some(Ok(()))));
2003 2 : s.context_put(None, "ctx", json!({"localId": "ctx", "kind": "Cached"}))
2004 2 : .expect("store");
2005 2 : assert!(s.context_delete(None, "ctx"));
2006 : }
2007 2 : let s = Store::open_file(&dir).expect("reopen");
2008 2 : assert!(
2009 2 : s.get(&t, Kind::Entity, "urn:gone").is_none(),
2010 : "phantom 409 trap"
2011 : );
2012 2 : assert_eq!(s.get(&t, Kind::Entity, "urn:kept").expect("kept")["n"], 2);
2013 2 : assert!(s.context_get(None, "ctx").is_none());
2014 2 : let _ = std::fs::remove_dir_all(&dir);
2015 2 : }
2016 :
2017 : /// A future-format file refuses to load with a clear message.
2018 : #[test]
2019 2 : fn file_mode_refuses_format_mismatch() {
2020 2 : let dir = tempdir("format");
2021 2 : {
2022 2 : let db = Database::create(dir.join("antares.redb")).expect("db");
2023 2 : let mut tx = db.begin_write().expect("tx");
2024 2 : tx.set_durability(Durability::Immediate).expect("dur");
2025 2 : tx.open_table(T_META)
2026 2 : .expect("meta")
2027 2 : .insert("format", "999")
2028 2 : .expect("insert");
2029 2 : tx.commit().expect("commit");
2030 2 : }
2031 2 : let err = match Store::open_file(&dir) {
2032 2 : Err(e) => e,
2033 0 : Ok(_) => panic!("must refuse a format-999 file"),
2034 : };
2035 2 : assert!(err.contains("format 999"), "err: {err}");
2036 2 : let _ = std::fs::remove_dir_all(&dir);
2037 2 : }
2038 :
2039 : /// A file that HOLDS rows but carries no format marker was written by a
2040 : /// binary whose key/value shape this one cannot vouch for. Refusing is
2041 : /// the whole point of the marker — serving it would answer requests from
2042 : /// data that may be misread. An empty file, by contrast, is just a fresh
2043 : /// one and gets the marker stamped.
2044 : #[test]
2045 2 : fn file_mode_refuses_data_without_a_format_marker() {
2046 2 : let dir = tempdir("nomarker");
2047 2 : {
2048 2 : let db = Database::create(dir.join("antares.redb")).expect("db");
2049 2 : let mut tx = db.begin_write().expect("tx");
2050 2 : tx.set_durability(Durability::Immediate).expect("dur");
2051 2 : {
2052 2 : let mut t = tx.open_table(T_ENTITIES).expect("entities");
2053 2 : let mut k = b"plain".to_vec();
2054 2 : k.push(0);
2055 2 : k.extend_from_slice(b"urn:e:1");
2056 2 : let bytes = serde_json::to_vec(&json!({"id": "urn:e:1"})).expect("serialize");
2057 2 : t.insert(k.as_slice(), bytes.as_slice()).expect("insert");
2058 2 : }
2059 2 : // deliberately NO meta table
2060 2 : tx.commit().expect("commit");
2061 2 : }
2062 2 : let err = match Store::open_file(&dir) {
2063 2 : Err(e) => e,
2064 0 : Ok(_) => panic!("must refuse data with no format marker"),
2065 : };
2066 2 : assert!(err.contains("no format marker"), "err: {err}");
2067 2 : assert!(err.contains("antares.redb"), "the file is named: {err}");
2068 2 : let _ = std::fs::remove_dir_all(&dir);
2069 :
2070 : // an EMPTY unmarked file is a fresh store, not a refusal
2071 2 : let fresh = tempdir("nomarker-empty");
2072 2 : {
2073 2 : let db = Database::create(fresh.join("antares.redb")).expect("db");
2074 2 : let tx = db.begin_write().expect("tx");
2075 2 : tx.commit().expect("commit");
2076 2 : }
2077 2 : Store::open_file(&fresh).expect("an empty file is a fresh store");
2078 2 : let _ = std::fs::remove_dir_all(&fresh);
2079 2 : }
2080 :
2081 : /// 4.22: "expiresAt is defined as the system temporal Property at which a
2082 : /// certain Entity, Property or Relationship shall become invalid." The
2083 : /// clause sanctions the DELETION lagging, not the invalidity — so the
2084 : /// write paths must agree with the read boundary the instant the stamp
2085 : /// passes. Before this, the same id was simultaneously a 404 on retrieve,
2086 : /// a 409 on create and a 204 on delete for a whole sweep interval.
2087 : #[tokio::test]
2088 2 : async fn an_expired_entity_is_absent_to_writes_too() {
2089 2 : let s = Store::default();
2090 2 : let t = TenantId::default();
2091 2 : let dead = json!({"id": "urn:e", "type": ["T"], "expiresAt": "2000-01-01T00:00:00Z"});
2092 2 : let live = json!({"id": "urn:l", "type": ["T"], "expiresAt": "2999-01-01T00:00:00Z"});
2093 2 : assert!(s.create(&t, Kind::Entity, "urn:e", dead.clone()).await);
2094 2 : assert!(s.create(&t, Kind::Entity, "urn:l", live.clone()).await);
2095 :
2096 : // patching or deleting something already invalid is a 404, not a 204
2097 2 : assert!(s
2098 2 : .mutate(&t, Kind::Entity, "urn:e", |_d| Ok::<(), ()>(()))
2099 2 : .await
2100 2 : .is_none());
2101 2 : assert!(
2102 2 : !s.delete(&t, Kind::Entity, "urn:e").await,
2103 : "expired delete is 404"
2104 : );
2105 : // …and creating over it succeeds instead of raising AlreadyExists
2106 2 : assert!(
2107 2 : s.create(&t, Kind::Entity, "urn:e", json!({"id": "urn:e", "n": 1}))
2108 2 : .await,
2109 : "an expired id must not 409 a create"
2110 : );
2111 2 : assert_eq!(
2112 2 : s.get(&t, Kind::Entity, "urn:e").expect("recreated")["n"],
2113 : 1,
2114 : "the create must have replaced the expired document"
2115 : );
2116 :
2117 : // an UNEXPIRED entity keeps every one of those answers
2118 2 : assert!(
2119 2 : !s.create(&t, Kind::Entity, "urn:l", live.clone()).await,
2120 : "409"
2121 : );
2122 2 : assert!(s
2123 2 : .mutate(&t, Kind::Entity, "urn:l", |_d| Ok::<(), ()>(()))
2124 2 : .await
2125 2 : .is_some());
2126 2 : assert!(s.delete(&t, Kind::Entity, "urn:l").await);
2127 :
2128 : // upsert over an expired id reports CREATED (201 + Location), not
2129 : // updated
2130 2 : s.create(&t, Kind::Entity, "urn:x", dead.clone()).await;
2131 2 : assert!(
2132 2 : !s.upsert(&t, Kind::Entity, "urn:x", json!({"id": "urn:x"}))
2133 2 : .await,
2134 : "an expired id must upsert as created"
2135 : );
2136 2 : assert!(
2137 2 : s.upsert(&t, Kind::Entity, "urn:x", json!({"id": "urn:x", "n": 2}))
2138 2 : .await,
2139 : "and the live one that replaced it as updated"
2140 : );
2141 :
2142 : // 4.22 is an ENTITY stamp: other kinds keep their own expiry rules
2143 2 : assert!(
2144 2 : s.create(&t, Kind::Subscription, "urn:s", dead.clone())
2145 2 : .await
2146 : );
2147 2 : assert!(
2148 2 : !s.create(&t, Kind::Subscription, "urn:s", dead).await,
2149 : "a subscription id still 409s"
2150 : );
2151 2 : assert!(s.delete(&t, Kind::Subscription, "urn:s").await);
2152 2 : }
2153 :
2154 : /// The change hook drives every notification and all temporal
2155 : /// auto-recording, so its contract is: create emits (None, Some), delete
2156 : /// emits (Some, None), a real mutate emits both images — and, just as
2157 : /// load-bearing, a NON-entity write and a no-op mutate emit NOTHING. A
2158 : /// subscription leaking into the hook would be mirrored into temporal
2159 : /// storage; a no-op emitting would re-notify every subscriber on every
2160 : /// idempotent PATCH.
2161 : #[tokio::test]
2162 2 : async fn the_change_hook_fires_for_entity_changes_only() {
2163 : use std::sync::{Arc, Mutex};
2164 : type Images = (Option<Value>, Option<Value>);
2165 2 : let seen: Arc<Mutex<Vec<Images>>> = Arc::default();
2166 2 : let s = Store::default();
2167 2 : let t = TenantId::default();
2168 2 : let rec = Arc::clone(&seen);
2169 2 : s.set_change_hook(Arc::new(
2170 : move |_t: &TenantId,
2171 : before: Option<Value>,
2172 : after: Option<Value>,
2173 : _v: antares_store::ChangeVersion|
2174 6 : -> antares_store::HookFuture<'_> {
2175 6 : let rec = Arc::clone(&rec);
2176 6 : Box::pin(async move {
2177 6 : rec.lock().expect("record").push((before, after));
2178 6 : })
2179 6 : },
2180 : ));
2181 :
2182 2 : s.create(&t, Kind::Entity, "urn:e", json!({"id": "urn:e", "n": 1}))
2183 2 : .await;
2184 : // a write on another kind must not reach the hook at all
2185 2 : s.create(&t, Kind::Subscription, "urn:s", json!({"id": "urn:s"}))
2186 2 : .await;
2187 2 : s.upsert(
2188 2 : &t,
2189 2 : Kind::Subscription,
2190 2 : "urn:s",
2191 2 : json!({"id": "urn:s", "n": 9}),
2192 2 : )
2193 2 : .await;
2194 2 : s.delete(&t, Kind::Subscription, "urn:s").await;
2195 : // a mutate that changes nothing is not a change
2196 2 : let _ = s
2197 2 : .mutate(&t, Kind::Entity, "urn:e", |_d| Ok::<(), ()>(()))
2198 2 : .await;
2199 : // a real one is
2200 2 : let _ = s
2201 2 : .mutate(&t, Kind::Entity, "urn:e", |d| {
2202 2 : d["n"] = json!(2);
2203 2 : Ok::<(), ()>(())
2204 2 : })
2205 2 : .await;
2206 : // an aborted mutate writes nothing and emits nothing
2207 2 : let _ = s
2208 2 : .mutate(&t, Kind::Entity, "urn:e", |d| {
2209 2 : d["n"] = json!(3);
2210 2 : Err::<(), &str>("no")
2211 2 : })
2212 2 : .await;
2213 2 : s.delete(&t, Kind::Entity, "urn:e").await;
2214 :
2215 2 : let seen = seen.lock().expect("read");
2216 2 : assert_eq!(seen.len(), 3, "emitted: {seen:?}");
2217 2 : assert!(seen[0].0.is_none() && seen[0].1.is_some(), "create");
2218 2 : assert_eq!(seen[1].0.as_ref().expect("before")["n"], 1);
2219 2 : assert_eq!(seen[1].1.as_ref().expect("after")["n"], 2);
2220 2 : assert!(seen[2].0.is_some() && seen[2].1.is_none(), "delete");
2221 : // the aborted mutate left the document alone
2222 2 : assert!(s.get(&t, Kind::Entity, "urn:e").is_none());
2223 2 : }
2224 :
2225 : /// The supported backup route is stop-copy — close the broker, copy
2226 : /// the file, reopen the copy. This test IS that route.
2227 : #[tokio::test]
2228 2 : async fn file_mode_stop_copy_backup_restores() {
2229 2 : let dir = tempdir("backup");
2230 2 : let restore = tempdir("restore");
2231 2 : let t = TenantId::default();
2232 : {
2233 2 : let s = Store::open_file(&dir).expect("open");
2234 2 : s.create(&t, Kind::Entity, "urn:b", json!({"v": 42})).await;
2235 : } // broker stopped — file quiescent
2236 2 : std::fs::copy(dir.join("antares.redb"), restore.join("antares.redb")).expect("copy");
2237 2 : let s = Store::open_file(&restore).expect("open backup");
2238 2 : assert_eq!(s.get(&t, Kind::Entity, "urn:b").expect("restored")["v"], 42);
2239 2 : let _ = std::fs::remove_dir_all(&dir);
2240 2 : let _ = std::fs::remove_dir_all(&restore);
2241 2 : }
2242 :
2243 : /// 4.22: reaping an expired entity is a state change like any other, so
2244 : /// it must be durable — after a sweep the doc stays gone across a reopen,
2245 : /// whatever the tenant key on disk looks like, while a doc that has not
2246 : /// expired survives both.
2247 : #[test]
2248 2 : fn file_mode_sweep_removals_persist() {
2249 2 : let dir = tempdir("sweep-persist");
2250 : // Seed the file directly: one tenant key the boot rebuild accepts but
2251 : // `TenantId::new` rejects, alongside an ordinary one.
2252 : {
2253 2 : let db = Database::create(dir.join("antares.redb")).expect("db");
2254 2 : let mut tx = db.begin_write().expect("tx");
2255 2 : tx.set_durability(Durability::Immediate).expect("dur");
2256 2 : {
2257 2 : let mut m = tx.open_table(T_META).expect("meta");
2258 2 : m.insert("format", FORMAT_VERSION).expect("insert");
2259 2 : }
2260 : {
2261 2 : let mut t = tx.open_table(T_ENTITIES).expect("entities");
2262 6 : for (tenant, id, expires) in [
2263 2 : ("odd.tenant", "urn:e:1", "2000-01-01T00:00:00Z"),
2264 2 : ("odd.tenant", "urn:e:2", "2999-01-01T00:00:00Z"),
2265 2 : ("plain", "urn:e:3", "2000-01-01T00:00:00Z"),
2266 6 : ] {
2267 6 : let mut k = tenant.as_bytes().to_vec();
2268 6 : k.push(0);
2269 6 : k.extend_from_slice(id.as_bytes());
2270 6 : let doc = json!({"id": id, "type": ["T"], "expiresAt": expires});
2271 6 : let bytes = serde_json::to_vec(&doc).expect("serialize");
2272 6 : t.insert(k.as_slice(), bytes.as_slice()).expect("insert");
2273 6 : }
2274 : }
2275 2 : tx.commit().expect("commit");
2276 : }
2277 : {
2278 2 : let s = Store::open_file(&dir).expect("open");
2279 2 : assert_eq!(s.sweep_expired("2026-01-01T00:00:00Z"), 2, "both expired");
2280 : }
2281 2 : let s = Store::open_file(&dir).expect("reopen");
2282 2 : let inner = s.inner.read().expect("lock");
2283 2 : assert!(
2284 2 : !inner
2285 2 : .entities
2286 2 : .get("odd.tenant")
2287 2 : .is_some_and(|d| d.contains_key("urn:e:1")),
2288 : "swept entity resurrected on reopen"
2289 : );
2290 2 : assert!(
2291 2 : inner
2292 2 : .entities
2293 2 : .get("odd.tenant")
2294 2 : .is_some_and(|d| d.contains_key("urn:e:2")),
2295 : "unexpired entity must outlive the sweep"
2296 : );
2297 2 : assert!(
2298 2 : !inner
2299 2 : .entities
2300 2 : .get("plain")
2301 2 : .is_some_and(|d| d.contains_key("urn:e:3")),
2302 : "swept entity resurrected on reopen"
2303 : );
2304 2 : drop(inner);
2305 2 : let _ = std::fs::remove_dir_all(&dir);
2306 2 : }
2307 :
2308 : /// S3-08, S3-09: in `file` mode a sweep that reaps many documents (an
2309 : /// expired Entity and its history each) commits them in ONE redb
2310 : /// transaction. One commit per key made the reap non-atomic on disk (a
2311 : /// crash between the Entity and its history left an orphan history no
2312 : /// sweep removes) and held the write section, which every reader waits
2313 : /// on, for one fsync per document. Timed against a single durable
2314 : /// commit on the same file: the sweep may cost a few, never one per key.
2315 : #[test]
2316 2 : fn a_file_mode_sweep_commits_once_whatever_it_reaps() {
2317 : const N: usize = 400;
2318 2 : let dir = tempdir("sweep-one-commit");
2319 : {
2320 2 : let db = Database::create(dir.join("antares.redb")).expect("db");
2321 2 : let tx = db.begin_write().expect("tx");
2322 2 : {
2323 2 : let mut m = tx.open_table(T_META).expect("meta");
2324 2 : m.insert("format", FORMAT_VERSION).expect("insert");
2325 2 : }
2326 4 : for table in [T_ENTITIES, T_TEMPORAL_ENTITIES] {
2327 4 : let mut t = tx.open_table(table).expect("table");
2328 1600 : for n in 0..N {
2329 1600 : let id = format!("urn:e:{n}");
2330 1600 : let doc = json!({"id": id, "type": ["T"], "expiresAt": "2000-01-01T00:00:00Z"});
2331 1600 : let bytes = serde_json::to_vec(&doc).expect("serialize");
2332 1600 : t.insert(key_bytes("t", &id).as_slice(), bytes.as_slice())
2333 1600 : .expect("insert");
2334 1600 : }
2335 4 : let keep = serde_json::to_vec(&json!({"id": "urn:keep", "type": ["T"]}))
2336 4 : .expect("serialize");
2337 4 : t.insert(key_bytes("t", "urn:keep").as_slice(), keep.as_slice())
2338 4 : .expect("insert");
2339 : }
2340 2 : tx.commit().expect("commit");
2341 : }
2342 2 : let s = std::sync::Arc::new(Store::open_file(&dir).expect("open"));
2343 : // one durable commit on this file, the unit the sweep is held to
2344 : const T_META_PROBE: Table = ::redb::TableDefinition::new("probe");
2345 2 : let started = std::time::Instant::now();
2346 10 : for n in 0..5 {
2347 10 : s.persist(
2348 10 : T_META_PROBE,
2349 10 : format!("probe{n}").as_bytes(),
2350 10 : Some(&json!(n)),
2351 10 : );
2352 10 : }
2353 2 : let commit = started.elapsed() / 5;
2354 2 : let t = TenantId::new("t").expect("tenant");
2355 2 : let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
2356 2 : let reader = {
2357 2 : let (s, t, stop) = (s.clone(), t.clone(), stop.clone());
2358 2 : std::thread::spawn(move || {
2359 2 : let mut worst = std::time::Duration::ZERO;
2360 23332 : while !stop.load(std::sync::atomic::Ordering::Relaxed) {
2361 23330 : let at = std::time::Instant::now();
2362 23330 : assert!(s.get(&t, Kind::Entity, "urn:keep").is_some());
2363 23330 : worst = worst.max(at.elapsed());
2364 : }
2365 2 : worst
2366 2 : })
2367 : };
2368 2 : std::thread::sleep(std::time::Duration::from_millis(20));
2369 2 : let started = std::time::Instant::now();
2370 2 : assert_eq!(s.sweep_expired("2026-01-01T00:00:00Z"), N);
2371 2 : let took = started.elapsed();
2372 2 : stop.store(true, std::sync::atomic::Ordering::Relaxed);
2373 2 : let worst = reader.join().expect("reader");
2374 2 : eprintln!("one commit {commit:?}; sweep of {N} {took:?}; worst read {worst:?}");
2375 2 : assert!(
2376 2 : took < commit * 20 + std::time::Duration::from_millis(200),
2377 : "a sweep of {N} took {took:?}, a commit {commit:?}: one commit per key"
2378 : );
2379 2 : drop(s);
2380 2 : let s = Store::open_file(&dir).expect("reopen");
2381 2 : let inner = s.inner.read().expect("lock");
2382 2 : assert_eq!(inner.entities.get("t").map(BTreeMap::len), Some(1));
2383 2 : assert_eq!(inner.temporal.get("t").map(BTreeMap::len), Some(1));
2384 2 : drop(inner);
2385 2 : let _ = std::fs::remove_dir_all(&dir);
2386 2 : }
2387 :
2388 : #[tokio::test]
2389 2 : async fn sweep_prunes_expired_attribute_instances() {
2390 2 : let s = Store::default();
2391 2 : let t = TenantId::default();
2392 2 : s.create(
2393 2 : &t,
2394 2 : Kind::Entity,
2395 2 : "urn:e",
2396 2 : json!({"id": "urn:e", "type": ["T"],
2397 2 : "attr": [
2398 2 : {"type": "Property", "value": 1, "expiresAt": "2000-01-01T00:00:00Z"},
2399 2 : {"type": "Property", "value": 2, "expiresAt": "2999-01-01T00:00:00Z"}],
2400 2 : "gone": [
2401 2 : {"type": "Property", "value": 3, "expiresAt": "2000-01-01T00:00:00Z"}]}),
2402 2 : )
2403 2 : .await;
2404 2 : s.create(
2405 2 : &t,
2406 2 : Kind::Temporal,
2407 2 : "urn:e",
2408 2 : json!({"id": "urn:e", "type": ["T"],
2409 2 : "attr": [
2410 2 : {"value": 1, "expiresAt": "2000-01-01T00:00:00Z"},
2411 2 : {"value": 2, "expiresAt": "2999-01-01T00:00:00Z"},
2412 2 : {"value": 3}]}),
2413 2 : )
2414 2 : .await;
2415 2 : assert_eq!(s.sweep_expired("2026-01-01T00:00:00Z"), 2);
2416 2 : let e = s.get(&t, Kind::Entity, "urn:e").expect("entity survives");
2417 2 : assert_eq!(e["attr"].as_array().expect("attr").len(), 1);
2418 2 : assert_eq!(e["attr"][0]["value"], 2);
2419 2 : assert!(e.get("gone").is_none(), "fully-expired attribute removed");
2420 2 : let tp = s
2421 2 : .get(&t, Kind::Temporal, "urn:e")
2422 2 : .expect("temporal survives");
2423 2 : let vals: Vec<i64> = tp["attr"]
2424 2 : .as_array()
2425 2 : .expect("instances")
2426 2 : .iter()
2427 4 : .map(|i| i["value"].as_i64().expect("value"))
2428 2 : .collect();
2429 2 : assert_eq!(vals, [2, 3], "expired pruned, no-expiry instance kept");
2430 2 : assert_eq!(s.sweep_expired("2026-01-01T00:00:00Z"), 0, "idempotent");
2431 2 : }
2432 :
2433 : /// ADR-0021 + 4.14: the memory/file arm has no Row-Level Security under
2434 : /// it, so the rule a Postgres deployment gets from the database is this
2435 : /// store's own job. A Hosted @context holds term mappings authored
2436 : /// through one Tenant's requests and belongs to that Tenant alone; a
2437 : /// Cached copy of a public document belongs to none and every Tenant
2438 : /// reaches it. A store that answered every caller would hand one
2439 : /// Tenant's term mappings to another, and 5.5.7 makes those mappings
2440 : /// decide what its payloads mean.
2441 : #[test]
2442 2 : fn clause_5_13_1_a_stored_context_answers_only_its_own_tenant() {
2443 2 : let s = Store::default();
2444 2 : let alpha = TenantId::new("alpha").expect("tenant");
2445 2 : let beta = TenantId::new("beta").expect("tenant");
2446 2 : let hosted = json!({"localId": "h", "kind": "Hosted", "owner": "alpha",
2447 2 : "body": {"@context": {"a": "https://alpha.example/a"}}});
2448 2 : s.context_put(Some(&alpha), "h", hosted.clone())
2449 2 : .expect("the owner stores its own row");
2450 2 : s.context_put(
2451 2 : None,
2452 2 : "c",
2453 2 : json!({"localId": "c", "kind": "Cached", "body": {"@context": {}}}),
2454 : )
2455 2 : .expect("a Cached copy belongs to no Tenant");
2456 :
2457 2 : assert!(
2458 2 : s.context_get(Some(&alpha), "h").is_some(),
2459 : "its owner reads it"
2460 : );
2461 2 : assert!(
2462 2 : s.context_get(Some(&beta), "h").is_none(),
2463 : "another Tenant's Hosted @context is as absent as one never stored"
2464 : );
2465 2 : assert!(
2466 2 : s.context_get(None, "h").is_none(),
2467 : "no Tenant in scope reaches no Tenant's documents"
2468 : );
2469 6 : for t in [Some(&alpha), Some(&beta), None] {
2470 6 : assert!(
2471 6 : s.context_get(t, "c").is_some(),
2472 : "a Cached copy is a public document and is shared"
2473 : );
2474 : }
2475 2 : assert!(
2476 2 : !s.context_delete(Some(&beta), "h"),
2477 : "a foreign delete takes nothing"
2478 : );
2479 2 : assert!(
2480 2 : s.context_put(Some(&beta), "h", hosted).is_err(),
2481 : "and a foreign write does not overwrite it either"
2482 : );
2483 2 : assert!(
2484 2 : s.context_list_meta(Some(&beta))
2485 2 : .iter()
2486 2 : .all(|r| r["localId"] != "h"),
2487 : "another Tenant's row is not listed"
2488 : );
2489 2 : assert!(
2490 2 : s.context_get(Some(&alpha), "h").is_some(),
2491 : "the owner's row survived every foreign attempt"
2492 : );
2493 :
2494 2 : s.purge_tenant(&alpha);
2495 2 : assert!(
2496 2 : s.context_get(Some(&alpha), "h").is_none(),
2497 : "purging the Tenant takes the @contexts it stored"
2498 : );
2499 2 : assert!(
2500 2 : s.context_get(None, "c").is_some(),
2501 : "and leaves the copy that belongs to no Tenant"
2502 : );
2503 2 : }
2504 :
2505 : /// 5.13.1: "Implementations shall periodically invalidate the 'Cached'
2506 : /// @contexts." The memory/file arm holds one entry per distinct external
2507 : /// URL a request referenced — client-controlled — so the same ceiling and
2508 : /// oldest-first eviction the Pg arm applies holds here, and it applies to
2509 : /// the Cached kind ONLY: a Hosted entry is client-owned data (5.13.2).
2510 : #[test]
2511 2 : fn clause_5_13_1_cached_contexts_are_capped_oldest_first() {
2512 2 : let s = Store::default();
2513 2004 : let entry = |kind: &str, created: &str| json!({"kind": kind, "createdAt": created, "body": {"@context": {}}});
2514 : // a Hosted entry older than every Cached one: age must not decide
2515 2 : s.context_put(
2516 2 : Some(&TenantId::default()),
2517 2 : "hosted",
2518 2 : entry("Hosted", "2000-01-01T00:00:00Z"),
2519 : )
2520 2 : .expect("store");
2521 2000 : for i in 0..MAX_CACHED_CONTEXTS {
2522 2000 : s.context_put(
2523 2000 : None,
2524 2000 : &format!("cached-{i:05}"),
2525 2000 : entry("Cached", &format!("2026-01-01T00:00:{:02}Z", i % 60)),
2526 2000 : )
2527 2000 : .expect("store");
2528 2000 : }
2529 4 : let cached = |s: &Store| {
2530 4 : s.context_list_meta(Some(&TenantId::default()))
2531 4 : .iter()
2532 4004 : .filter(|d| d["kind"] == "Cached")
2533 4 : .count()
2534 4 : };
2535 2 : assert_eq!(
2536 2 : cached(&s),
2537 : MAX_CACHED_CONTEXTS,
2538 : "at the ceiling, nothing lost"
2539 : );
2540 :
2541 : // one more Cached entry evicts exactly one — the oldest
2542 2 : s.context_put(None, "cached-new", entry("Cached", "2026-06-01T00:00:00Z"))
2543 2 : .expect("store");
2544 2 : assert_eq!(cached(&s), MAX_CACHED_CONTEXTS, "the ceiling holds");
2545 2 : assert!(
2546 2 : s.context_get(None, "cached-new").is_some(),
2547 : "the new entry stays"
2548 : );
2549 2 : assert!(
2550 2 : s.context_get(None, "cached-00000").is_none(),
2551 : "the oldest Cached entry is the one evicted"
2552 : );
2553 2 : assert!(
2554 2 : s.context_get(None, "cached-00001").is_some(),
2555 : "eviction stops at the ceiling"
2556 : );
2557 2 : assert!(
2558 2 : s.context_get(Some(&TenantId::default()), "hosted")
2559 2 : .is_some(),
2560 : "a Hosted entry is never a candidate, however old"
2561 : );
2562 2 : }
2563 :
2564 : #[tokio::test]
2565 2 : async fn mutate_aborts_on_error() {
2566 2 : let s = Store::default();
2567 2 : let t = TenantId::default();
2568 2 : s.create(&t, Kind::Entity, "urn:a", json!({"n": 1})).await;
2569 2 : let r: Option<Result<(), &str>> = s
2570 2 : .mutate(&t, Kind::Entity, "urn:a", |d| {
2571 2 : d["n"] = json!(2);
2572 2 : Err("nope")
2573 2 : })
2574 2 : .await;
2575 2 : assert!(matches!(r, Some(Err("nope"))));
2576 2 : assert_eq!(s.get(&t, Kind::Entity, "urn:a").unwrap()["n"], 1);
2577 2 : }
2578 :
2579 : /// A purged tenant is gone from every kind, the neighbour tenant is
2580 : /// untouched, and in `file` mode the removal survives a reopen.
2581 : #[tokio::test]
2582 2 : async fn purge_tenant_empties_every_kind_and_survives_reopen() {
2583 2 : let dir = tempdir("purge");
2584 2 : let a = TenantId::new("purge_a").expect("tenant");
2585 2 : let b = TenantId::new("purge_b").expect("tenant");
2586 : {
2587 2 : let s = Store::open_file(&dir).expect("open");
2588 4 : for t in [&a, &b] {
2589 40 : for kind in ALL_KINDS {
2590 40 : assert!(s.create(t, kind, "urn:x:1", json!({"id": "urn:x:1"})).await);
2591 : }
2592 : }
2593 4 : assert!(s.tenant_ids().iter().any(|t| t == "purge_a"), "listed");
2594 2 : let row = s.tenant_stats_one(&a);
2595 2 : assert_eq!((row.entities, row.subscriptions, row.dist_subs), (1, 1, 1));
2596 2 : assert!(s.purge_tenant(&a));
2597 2 : assert!(!s.purge_tenant(&a), "second purge finds nothing");
2598 2 : assert!(!s.tenant_exists(&a));
2599 2 : assert!(s.tenant_exists(&b));
2600 20 : for kind in ALL_KINDS {
2601 20 : assert!(s.get(&a, kind, "urn:x:1").is_none(), "{kind:?} row left");
2602 20 : assert!(s.get(&b, kind, "urn:x:1").is_some(), "{kind:?} lost for b");
2603 : }
2604 4 : assert!(s.tenant_ids().iter().all(|t| t != "purge_a"));
2605 2 : assert_eq!(
2606 2 : s.tenant_stats_one(&a).entities,
2607 : 0,
2608 : "a purged tenant counts nothing"
2609 : );
2610 : }
2611 2 : let s = Store::open_file(&dir).expect("reopen");
2612 2 : assert!(!s.tenant_exists(&a), "purge must be persisted");
2613 2 : assert!(s.tenant_exists(&b));
2614 2 : let _ = std::fs::remove_dir_all(&dir);
2615 2 : }
2616 : }
|
