Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! The tenant-keyed document mirrors the notification pipeline and the
3 : //! bus wiring share: a registration mirror, a subscription mirror with
4 : //! its inverted candidate index and sweep clocks, and the change event
5 : //! every write emits. A leaf: nothing here names another module, so the
6 : //! application state can hold these types without depending on the
7 : //! pipeline that consumes them.
8 :
9 : use serde_json::Value;
10 :
11 : /// One change queue event: tenant, before-image, after-image.
12 : pub type Change = (String, Option<Value>, Option<Value>);
13 :
14 : /// A per-instance tenant-keyed document mirror (bus=nats). One
15 : /// instance holds subscriptions (fed by the KV watcher), another holds
16 : /// registrations (fed by `ANTARES_REGISTRY` deltas). Postgres stays the
17 : /// system of record — this map is a cache with exactly one writer (the
18 : /// watcher task); readers only snapshot.
19 : ///
20 : /// A snapshot hands out `Arc`s, not copies. When this mirror is installed
21 : /// it IS the federation read path's registration source ([`Self::matching`]
22 : /// replaces the store's `matching_registrations`), and a broker at the
23 : /// 100 000-registration target that deep-copied every document per
24 : /// distributed request paid 84 ms and a hundred megabytes for a set it only
25 : /// ever reads. The documents are immutable once applied — a changed
26 : /// registration arrives as a whole new document — so sharing them costs a
27 : /// refcount and nothing else.
28 : #[derive(Default)]
29 : pub struct DocMirror {
30 : map: std::sync::RwLock<std::collections::HashMap<String, RegIndex>>,
31 : }
32 :
33 : /// One tenant's registrations, plus the two dimensions a distributed read
34 : /// narrows on before it evaluates 5.12 matching per registration.
35 : ///
36 : /// The buckets are the mirror's answer to the same question
37 : /// `matching_registrations` pushes into the store's `csource_index`, and
38 : /// they obey the same contract: a bucket may only ever drop a registration
39 : /// `reg_candidate` would reject anyway. So a registration lands in `any_*`
40 : /// — the set no key can exclude — whenever the dimension cannot decide it:
41 : /// a `RegistrationInfo` with no `entities` at all, an `EntityInfo` with no
42 : /// `type` (5.12 restricts such an entry by id alone), or an `EntityInfo`
43 : /// carrying an `idPattern` (5.12 condition 5 forwards on a pattern whatever
44 : /// ids were asked for).
45 : ///
46 : /// The membership is per REGISTRATION, not per `EntityInfo` — a
47 : /// registration whose one entry carries the asked-for type and whose other
48 : /// carries the asked-for id is kept, where the store's per-row `WHERE` drops
49 : /// it. Wider than the store, which is the safe direction for a prefilter.
50 : #[derive(Default)]
51 : struct RegIndex {
52 : docs: std::collections::HashMap<String, std::sync::Arc<Value>>,
53 : by_type: std::collections::HashMap<String, std::collections::HashSet<String>>,
54 : by_id: std::collections::HashMap<String, std::collections::HashSet<String>>,
55 : any_type: std::collections::HashSet<String>,
56 : any_id: std::collections::HashSet<String>,
57 : }
58 :
59 : /// The type and id keys one registration document occupies. `None` for a
60 : /// dimension means no key of it can exclude this registration.
61 334048 : fn reg_keys(doc: &Value) -> (Option<Vec<String>>, Option<Vec<String>>) {
62 334048 : let (mut types, mut ids) = (Vec::new(), Vec::new());
63 334048 : let (mut any_type, mut any_id) = (false, false);
64 : // No `information` at all is not a shape 5.2.9 allows, and nothing can
65 : // be read off it — the registration stays in both broad sets rather
66 : // than being narrowed out on a document this code does not understand.
67 334048 : let Some(infos) = doc.get("information").and_then(Value::as_array) else {
68 320040 : return (None, None);
69 : };
70 14008 : if infos.is_empty() {
71 0 : return (None, None);
72 14008 : }
73 14274 : for info in infos {
74 : // 5.2.9: a RegistrationInfo may name only propertyNames /
75 : // relationshipNames, which restricts attributes and no entity.
76 14274 : let Some(entities) = info.get("entities").and_then(Value::as_array) else {
77 98 : return (None, None);
78 : };
79 14176 : if entities.is_empty() {
80 0 : return (None, None);
81 14176 : }
82 14488 : for ei in entities {
83 14488 : match crate::registry::ei_types(ei) {
84 14488 : ts if ts.is_empty() => any_type = true,
85 14322 : ts => types.extend(ts.into_iter().map(str::to_owned)),
86 : }
87 14488 : if ei.get("idPattern").is_some() {
88 12050 : any_id = true;
89 12050 : } else {
90 2438 : match ei.get("id").and_then(Value::as_str) {
91 1252 : Some(id) => ids.push(id.to_owned()),
92 1186 : None => any_id = true,
93 : }
94 : }
95 : }
96 : }
97 13910 : ((!any_type).then_some(types), (!any_id).then_some(ids))
98 334048 : }
99 :
100 : /// The registrations a set of keys can reach: the ones filed under a named
101 : /// key, plus every one the dimension cannot decide. `None` in, `None` out —
102 : /// the caller asked nothing of this dimension, so it narrows nothing.
103 2268 : fn bucketed<'a>(
104 2268 : keys: Option<&[String]>,
105 2268 : by: &'a std::collections::HashMap<String, std::collections::HashSet<String>>,
106 2268 : any: &'a std::collections::HashSet<String>,
107 2268 : ) -> Option<std::collections::HashSet<&'a str>> {
108 2268 : let keys = keys?;
109 1442 : let mut out: std::collections::HashSet<&str> = any.iter().map(String::as_str).collect();
110 1710 : for k in keys {
111 1710 : if let Some(ids) = by.get(k) {
112 1182 : out.extend(ids.iter().map(String::as_str));
113 1182 : }
114 : }
115 1442 : Some(out)
116 2268 : }
117 :
118 : impl RegIndex {
119 : /// Drop a registration from the documents and from every bucket it is
120 : /// filed under. The keys come from the document being removed, so the
121 : /// index never needs a second copy of them.
122 334018 : fn remove(&mut self, id: &str) {
123 334018 : let Some(old) = self.docs.remove(id) else {
124 332992 : return;
125 : };
126 1026 : let (types, ids) = reg_keys(&old);
127 1026 : let unfile =
128 : |keys: Option<Vec<String>>,
129 : by: &mut std::collections::HashMap<String, std::collections::HashSet<String>>,
130 2052 : any: &mut std::collections::HashSet<String>| {
131 2052 : match keys {
132 656 : None => {
133 656 : any.remove(id);
134 656 : }
135 1396 : Some(ks) => {
136 1712 : for k in ks {
137 1712 : if let Some(set) = by.get_mut(&k) {
138 1556 : set.remove(id);
139 1556 : if set.is_empty() {
140 1380 : by.remove(&k);
141 1380 : }
142 156 : }
143 : }
144 : }
145 : }
146 2052 : };
147 1026 : unfile(types, &mut self.by_type, &mut self.any_type);
148 1026 : unfile(ids, &mut self.by_id, &mut self.any_id);
149 334018 : }
150 :
151 333022 : fn insert(&mut self, id: &str, doc: Value) {
152 333022 : self.remove(id);
153 333022 : let (types, ids) = reg_keys(&doc);
154 333022 : let file =
155 : |keys: Option<Vec<String>>,
156 : by: &mut std::collections::HashMap<String, std::collections::HashSet<String>>,
157 666044 : any: &mut std::collections::HashSet<String>| {
158 666044 : match keys {
159 652758 : None => {
160 652758 : any.insert(id.to_owned());
161 652758 : }
162 13286 : Some(ks) => {
163 13606 : for k in ks {
164 13606 : by.entry(k).or_default().insert(id.to_owned());
165 13606 : }
166 : }
167 : }
168 666044 : };
169 333022 : file(types, &mut self.by_type, &mut self.any_type);
170 333022 : file(ids, &mut self.by_id, &mut self.any_id);
171 333022 : self.docs.insert(id.to_owned(), std::sync::Arc::new(doc));
172 333022 : }
173 :
174 996 : fn is_empty(&self) -> bool {
175 996 : self.docs.is_empty()
176 996 : }
177 : }
178 :
179 : /// Both mirror flavours accept `{tenant, id, doc|null}` deltas — the seam
180 : /// wiring's hydrate/watch loops program against.
181 : pub trait Mirror: Send + Sync {
182 : fn apply(&self, tenant: &str, id: &str, doc: Option<Value>);
183 :
184 : /// A Context Source Registration Subscription was written somewhere.
185 : /// Mirrors that serve no interval sweep have nothing to do.
186 0 : fn csub_written(&self) {}
187 : }
188 :
189 : impl Mirror for DocMirror {
190 331324 : fn apply(&self, tenant: &str, id: &str, doc: Option<Value>) {
191 331324 : DocMirror::apply(self, tenant, id, doc);
192 331324 : }
193 : }
194 :
195 : /// The subscription mirror — docs plus the inverted candidate index.
196 : ///
197 : /// Bucketing per subscription (conservative, union-of-buckets = candidates):
198 : /// - has an `entities` selector whose every entry names a plain expanded
199 : /// type IRI → `by_type[iri]` (idPattern/watchedAttributes narrow FURTHER,
200 : /// so type is the widest exact key);
201 : /// - no selector but `watchedAttributes` → `by_attr[iri]` (such a sub can
202 : /// only fire when a watched attribute changed — 5.8.6);
203 : /// - anything else (4.17 selection expressions, shapes the index cannot
204 : /// prove) → `broad`, evaluated on every change.
205 : ///
206 : /// The mirror also carries the interval sweep's clocks: the earliest instant
207 : /// at which a periodic subscription it holds can be due (5.8.6 sends that
208 : /// Notification "when the time interval (in seconds) specified in such value
209 : /// field is reached"), so ticks that cannot fire anything never read the
210 : /// store. Per instance, not global — one broker process, one clock pair.
211 : #[derive(Default)]
212 : pub struct SubMirror {
213 : map: std::sync::RwLock<std::collections::HashMap<String, TenantIndex>>,
214 : /// Epoch millis; `0` = sweep at the next tick. Set by every sweep from
215 : /// the subscriptions it saw, and zeroed again by `apply` whenever a
216 : /// periodic subscription is written, so a new one is never waited out.
217 : pub(crate) next_sub_sweep_ms: std::sync::atomic::AtomicI64,
218 : /// The same clock for Context Source Registration Subscriptions
219 : /// (5.11.7). They are not mirrored as documents — the sweep reads them
220 : /// from the store — so a write signals this clock through
221 : /// [`SubMirror::csub_written`] instead, and a sweep falls back to
222 : /// `CSUB_SWEEP_BACKSTOP_MS` in case a signal was lost.
223 : pub(crate) next_csub_sweep_ms: std::sync::atomic::AtomicI64,
224 : }
225 :
226 : /// The longest a sweep parks the Context Source Registration Subscription
227 : /// half when nothing it saw is due.
228 : ///
229 : /// A write clears the clock, so this is not the path a newly created
230 : /// periodic subscription waits on: it is the repair time for a lost signal,
231 : /// which on the bus is a KV put that exhausted its retries. Between sweeps
232 : /// the half costs one `list` per tenant, and the tenant target is 10 000, so
233 : /// polling it at the tick rate is the broker's whole idle cost with nothing
234 : /// periodic configured — which is the state most deployments are in.
235 : ///
236 : /// Table 5.2.12-1 bounds `timeInterval` only by "greater than 0", so a
237 : /// sub-second interval is legal; the tick period bounds how closely any of
238 : /// them can be served, signal or no signal.
239 : // ponytail: one `list` per tenant per backstop, cross-tenant enumeration of
240 : // the periodic rows would replace it — that needs the RLS service escape
241 : // (`antares.service`) extended to the two subscription tables.
242 : pub(crate) const CSUB_SWEEP_BACKSTOP_MS: i64 = 60_000;
243 :
244 : #[derive(Default)]
245 : struct TenantIndex {
246 : // Shared, not owned: `candidates` runs once per change and hands back
247 : // every subscription that could fire. Cloning the documents there deep-
248 : // copied each one — every string and every nested map — on the hot path,
249 : // so the cost of a change grew with the size of the subscriptions it was
250 : // evaluated against rather than with the work it caused. An `Arc` clone
251 : // is a refcount bump, and the registration mirror above already holds
252 : // its documents this way.
253 : docs: std::collections::HashMap<String, std::sync::Arc<Value>>,
254 : by_type: std::collections::HashMap<String, std::collections::HashSet<String>>,
255 : by_attr: std::collections::HashMap<String, std::collections::HashSet<String>>,
256 : broad: std::collections::HashSet<String>,
257 : /// The [`tick`] each subscription first entered the mirror at; a later
258 : /// write keeps it, so a PATCH does not make a subscription new again.
259 : born: std::collections::HashMap<String, u64>,
260 : /// Periodic subscriptions whose last firing had nothing to send (5.8.6:
261 : /// "If there are no matching Entities, no Notification is sent"): epoch
262 : /// millis before which the sweep does not read them again. Such a firing
263 : /// leaves `lastNotification` alone, so without this the subscription
264 : /// stays due and re-runs its whole query on every tick. A write clears it.
265 : quiet_until: std::collections::HashMap<String, i64>,
266 : }
267 :
268 : /// One clock for both sides of a match: a change takes a tick when it is
269 : /// queued for matching, a subscription when it enters the mirror. A change
270 : /// queued before a subscription existed never reaches it, however late the
271 : /// matcher drains the queue (NGSI-LD notifies changes made after the
272 : /// subscription). Process-wide, which is the scope of the
273 : /// local bus.
274 : static CLOCK: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
275 :
276 : /// The next instant of [`CLOCK`]; never 0, never repeated.
277 143300 : pub(crate) fn tick() -> u64 {
278 143300 : CLOCK
279 143300 : .fetch_add(1, std::sync::atomic::Ordering::SeqCst)
280 143300 : .saturating_add(1)
281 143300 : }
282 :
283 : /// Which index bucket(s) one stored subscription doc belongs in.
284 : pub(crate) enum Keys {
285 : Types(Vec<String>),
286 : Attrs(Vec<String>),
287 : Broad,
288 : }
289 :
290 13388 : pub(crate) fn index_keys(doc: &Value) -> Keys {
291 13388 : if let Some(entities) = doc.get("entities").and_then(Value::as_array) {
292 13162 : let mut types = Vec::new();
293 13166 : for e in entities {
294 13166 : match e.get("type").and_then(Value::as_str) {
295 : // 4.17 selection expressions (and any wildcard) are evaluated
296 : // at match time — the index cannot prove them, so the whole
297 : // sub goes broad (entries are OR-ed: one opaque entry taints
298 : // the union).
299 13142 : Some(t) if !t.contains(['|', ',', ';', '(', ')', '*']) => {
300 12966 : types.push(t.to_owned());
301 12966 : }
302 200 : _ => return Keys::Broad,
303 : }
304 : }
305 12962 : if types.is_empty() {
306 0 : return Keys::Broad;
307 12962 : }
308 12962 : return Keys::Types(types);
309 226 : }
310 226 : if let Some(watched) = doc.get("watchedAttributes").and_then(Value::as_array) {
311 110 : let attrs: Vec<String> = watched
312 110 : .iter()
313 110 : .filter_map(Value::as_str)
314 110 : .map(str::to_owned)
315 110 : .collect();
316 110 : if attrs.len() == watched.len() && !attrs.is_empty() {
317 110 : return Keys::Attrs(attrs);
318 0 : }
319 116 : }
320 116 : Keys::Broad
321 13388 : }
322 :
323 : impl SubMirror {
324 : /// Apply one KV delta: `None` doc = deleted. Rekeys the index from the
325 : /// old doc before inserting the new one.
326 8399 : pub fn apply(&self, tenant: &str, id: &str, doc: Option<Value>) {
327 8399 : let mut map = self
328 8399 : .map
329 8399 : .write()
330 8399 : .unwrap_or_else(std::sync::PoisonError::into_inner);
331 8399 : let t = map.entry(tenant.to_owned()).or_default();
332 8399 : let born = t.born.remove(id);
333 8399 : t.quiet_until.remove(id);
334 8399 : if let Some(old) = t.docs.remove(id) {
335 5391 : match index_keys(&old) {
336 5365 : Keys::Types(ts) => {
337 5365 : for ty in ts {
338 5365 : if let Some(s) = t.by_type.get_mut(&ty) {
339 5365 : s.remove(id);
340 5365 : if s.is_empty() {
341 1296 : t.by_type.remove(&ty);
342 4069 : }
343 0 : }
344 : }
345 : }
346 0 : Keys::Attrs(ats) => {
347 0 : for a in ats {
348 0 : if let Some(s) = t.by_attr.get_mut(&a) {
349 0 : s.remove(id);
350 0 : if s.is_empty() {
351 0 : t.by_attr.remove(&a);
352 0 : }
353 0 : }
354 : }
355 : }
356 26 : Keys::Broad => {
357 26 : t.broad.remove(id);
358 26 : }
359 : }
360 3008 : }
361 8399 : if let Some(d) = doc {
362 7805 : if d.get("timeInterval").is_some() {
363 210 : // A periodic subscription just appeared or changed its
364 210 : // anchor: the sweep clock computed without it must not hold
365 210 : // it back (5.8.6).
366 210 : self.next_sub_sweep_ms
367 210 : .store(0, std::sync::atomic::Ordering::Relaxed);
368 7595 : }
369 7805 : match index_keys(&d) {
370 7449 : Keys::Types(ts) => {
371 7453 : for ty in ts {
372 7453 : t.by_type.entry(ty).or_default().insert(id.to_owned());
373 7453 : }
374 : }
375 106 : Keys::Attrs(ats) => {
376 106 : for a in ats {
377 106 : t.by_attr.entry(a).or_default().insert(id.to_owned());
378 106 : }
379 : }
380 250 : Keys::Broad => {
381 250 : t.broad.insert(id.to_owned());
382 250 : }
383 : }
384 7805 : t.docs.insert(id.to_owned(), std::sync::Arc::new(d));
385 7805 : t.born.insert(id.to_owned(), born.unwrap_or_else(tick));
386 594 : }
387 8399 : if map.get(tenant).is_some_and(|t| t.docs.is_empty()) {
388 554 : map.remove(tenant);
389 7845 : }
390 8399 : }
391 :
392 : /// 5.11.7: a Context Source Registration Subscription was written. Only
393 : /// the clock moves — these are matched against registrations rather than
394 : /// entities, so they are not carried in the candidate index, and the
395 : /// sweep reads the tenant's rows from the store once it wakes.
396 458 : pub(crate) fn csub_written(&self) {
397 458 : self.next_csub_sweep_ms
398 458 : .store(0, std::sync::atomic::Ordering::Relaxed);
399 458 : }
400 :
401 : /// The hot path: subscriptions that could possibly fire for a change
402 : /// touching these entity types and these changed attributes. Union of
403 : /// the type hits, the attr hits and the broad bucket — a superset of
404 : /// the firing set, never a subset.
405 142 : pub fn candidates(
406 142 : &self,
407 142 : tenant: &str,
408 142 : types: &[&str],
409 142 : changed_attrs: &[&str],
410 142 : ) -> Vec<std::sync::Arc<Value>> {
411 142 : self.candidates_before(tenant, types, changed_attrs, u64::MAX)
412 142 : }
413 :
414 : /// [`SubMirror::candidates`] of a change queued at `queued` (a `tick`):
415 : /// only the subscriptions that were in the mirror before it.
416 141061 : pub fn candidates_before(
417 141061 : &self,
418 141061 : tenant: &str,
419 141061 : types: &[&str],
420 141061 : changed_attrs: &[&str],
421 141061 : queued: u64,
422 141061 : ) -> Vec<std::sync::Arc<Value>> {
423 141061 : let map = self
424 141061 : .map
425 141061 : .read()
426 141061 : .unwrap_or_else(std::sync::PoisonError::into_inner);
427 141061 : let Some(t) = map.get(tenant) else {
428 136145 : return Vec::new();
429 : };
430 4916 : let mut ids: std::collections::HashSet<&str> = t.broad.iter().map(String::as_str).collect();
431 4920 : for ty in types {
432 4920 : if let Some(s) = t.by_type.get(*ty) {
433 4886 : ids.extend(s.iter().map(String::as_str));
434 4886 : }
435 : }
436 5054 : for a in changed_attrs {
437 5054 : if let Some(s) = t.by_attr.get(*a) {
438 106 : ids.extend(s.iter().map(String::as_str));
439 4948 : }
440 : }
441 4916 : ids.iter()
442 20832 : .filter(|id| t.born.get(**id).is_some_and(|b| *b < queued))
443 20820 : .filter_map(|id| t.docs.get(*id).cloned())
444 4916 : .collect()
445 141061 : }
446 :
447 : #[cfg(any(test, feature = "test-kit"))]
448 8 : pub fn docs(&self, tenant: &str) -> Vec<std::sync::Arc<Value>> {
449 8 : self.map
450 8 : .read()
451 8 : .unwrap_or_else(std::sync::PoisonError::into_inner)
452 8 : .get(tenant)
453 8 : .map(|t| t.docs.values().cloned().collect())
454 8 : .unwrap_or_default()
455 8 : }
456 :
457 : /// The interval sweep's whole input: the tenant's periodic (5.2.12
458 : /// `timeInterval`) subscriptions. The walk is over every subscription the
459 : /// tenant holds, but only the periodic ones are carried out of it — the
460 : /// sweep clocks keep a tick with nothing due from reaching this at all.
461 4991 : pub(crate) fn periodic_docs(&self, tenant: &str) -> Vec<std::sync::Arc<Value>> {
462 4991 : self.map
463 4991 : .read()
464 4991 : .unwrap_or_else(std::sync::PoisonError::into_inner)
465 4991 : .get(tenant)
466 4991 : .map(|t| {
467 194 : t.docs
468 194 : .values()
469 2497 : .filter(|d| d.get("timeInterval").is_some())
470 194 : .cloned()
471 194 : .collect()
472 194 : })
473 4991 : .unwrap_or_default()
474 4991 : }
475 :
476 : /// The instant an empty periodic firing parked this subscription until
477 : /// (see `TenantIndex::quiet_until`), if one did and no write came since.
478 68 : pub(crate) fn quiet_until(&self, tenant: &str, id: &str) -> Option<i64> {
479 68 : self.map
480 68 : .read()
481 68 : .unwrap_or_else(std::sync::PoisonError::into_inner)
482 68 : .get(tenant)
483 68 : .and_then(|t| t.quiet_until.get(id).copied())
484 68 : }
485 :
486 : /// Park a periodic subscription whose firing had nothing to send until
487 : /// `until`, its next interval. Only while the mirror still holds the
488 : /// document that firing evaluated: a write during the firing replaced
489 : /// it and cleared the park, and parking the new document on the old
490 : /// one's empty result would hold it back for a whole interval.
491 16 : pub(crate) fn park_quiet(
492 16 : &self,
493 16 : tenant: &str,
494 16 : id: &str,
495 16 : evaluated: &std::sync::Arc<Value>,
496 16 : until: i64,
497 16 : ) {
498 16 : let mut map = self
499 16 : .map
500 16 : .write()
501 16 : .unwrap_or_else(std::sync::PoisonError::into_inner);
502 16 : if let Some(t) = map.get_mut(tenant).filter(|t| {
503 16 : t.docs
504 16 : .get(id)
505 16 : .is_some_and(|d| std::sync::Arc::ptr_eq(d, evaluated))
506 16 : }) {
507 12 : t.quiet_until.insert(id.to_owned(), until);
508 12 : }
509 16 : }
510 :
511 : #[cfg(any(test, feature = "test-kit"))]
512 2 : pub fn tenants(&self) -> Vec<String> {
513 2 : self.map
514 2 : .read()
515 2 : .unwrap_or_else(std::sync::PoisonError::into_inner)
516 2 : .keys()
517 2 : .cloned()
518 2 : .collect()
519 2 : }
520 : }
521 :
522 : #[cfg(test)]
523 : impl SubMirror {
524 : /// Poison the index lock from a thread that panics while holding it:
525 : /// what a matcher that dies mid-write leaves behind, for the
526 : /// supervision tests.
527 4 : pub(crate) fn poison(self: &std::sync::Arc<Self>) {
528 4 : let m = std::sync::Arc::clone(self);
529 4 : let _ = std::thread::spawn(move || {
530 4 : let _held = m
531 4 : .map
532 4 : .write()
533 4 : .unwrap_or_else(std::sync::PoisonError::into_inner);
534 4 : panic!("poison the mirror");
535 : })
536 4 : .join();
537 4 : }
538 : }
539 :
540 : impl Mirror for SubMirror {
541 1612 : fn apply(&self, tenant: &str, id: &str, doc: Option<Value>) {
542 1612 : SubMirror::apply(self, tenant, id, doc);
543 1612 : }
544 :
545 2 : fn csub_written(&self) {
546 2 : SubMirror::csub_written(self);
547 2 : }
548 : }
549 :
550 : impl DocMirror {
551 : /// Apply one KV delta: `None` doc = deleted.
552 334018 : pub fn apply(&self, tenant: &str, id: &str, doc: Option<Value>) {
553 334018 : let mut map = self
554 334018 : .map
555 334018 : .write()
556 334018 : .unwrap_or_else(std::sync::PoisonError::into_inner);
557 334018 : match doc {
558 333022 : Some(d) => {
559 333022 : map.entry(tenant.to_owned()).or_default().insert(id, d);
560 333022 : }
561 : None => {
562 996 : if let Some(t) = map.get_mut(tenant) {
563 996 : t.remove(id);
564 996 : if t.is_empty() {
565 842 : map.remove(tenant);
566 842 : }
567 0 : }
568 : }
569 : }
570 334018 : }
571 :
572 : /// Take `fresh`'s contents in one step, dropping everything held before:
573 : /// a re-hydrate seeds a new mirror from the store and swaps it in, so a
574 : /// registration deleted while no delta could arrive leaves this one too.
575 2 : pub fn replace_with(&self, fresh: DocMirror) {
576 2 : let fresh = fresh
577 2 : .map
578 2 : .into_inner()
579 2 : .unwrap_or_else(std::sync::PoisonError::into_inner);
580 2 : *self
581 2 : .map
582 2 : .write()
583 2 : .unwrap_or_else(std::sync::PoisonError::into_inner) = fresh;
584 2 : }
585 :
586 : /// The registrations of one tenant that can match these ids and types,
587 : /// shared rather than copied.
588 : ///
589 : /// Both dimensions are optional and each narrows on its own; an absent
590 : /// one is a dimension the caller asked nothing about, never an empty
591 : /// answer. With both given a registration has to survive both, which is
592 : /// the store's rule too.
593 38386 : pub fn matching(
594 38386 : &self,
595 38386 : tenant: &str,
596 38386 : ids: Option<&[String]>,
597 38386 : types: Option<&[String]>,
598 38386 : ) -> Vec<std::sync::Arc<Value>> {
599 38386 : let map = self
600 38386 : .map
601 38386 : .read()
602 38386 : .unwrap_or_else(std::sync::PoisonError::into_inner);
603 38386 : let Some(t) = map.get(tenant) else {
604 37252 : return Vec::new();
605 : };
606 1134 : let by_type = bucketed(types, &t.by_type, &t.any_type);
607 1134 : let by_id = bucketed(ids, &t.by_id, &t.any_id);
608 1134 : let pick = |keys: std::collections::HashSet<&str>| -> Vec<std::sync::Arc<Value>> {
609 1116 : keys.into_iter()
610 8350 : .filter_map(|k| t.docs.get(k).map(std::sync::Arc::clone))
611 1116 : .collect()
612 1116 : };
613 1134 : match (by_type, by_id) {
614 18 : (None, None) => t.docs.values().map(std::sync::Arc::clone).collect(),
615 790 : (Some(k), None) | (None, Some(k)) => pick(k),
616 326 : (Some(a), Some(b)) => pick(a.intersection(&b).copied().collect()),
617 : }
618 38386 : }
619 :
620 20 : pub fn docs(&self, tenant: &str) -> Vec<Value> {
621 20 : self.matching(tenant, None, None)
622 20 : .into_iter()
623 20 : .map(|d| (*d).clone())
624 20 : .collect()
625 20 : }
626 :
627 : #[cfg(any(test, feature = "test-kit"))]
628 6 : pub fn tenants(&self) -> Vec<String> {
629 6 : self.map
630 6 : .read()
631 6 : .unwrap_or_else(std::sync::PoisonError::into_inner)
632 6 : .keys()
633 6 : .cloned()
634 6 : .collect()
635 6 : }
636 : }
637 :
638 : #[cfg(test)]
639 : mod born {
640 : use super::{tick, SubMirror};
641 : use serde_json::json;
642 :
643 16 : fn ids(m: &SubMirror, queued: u64) -> Vec<String> {
644 16 : let mut out: Vec<String> = m
645 16 : .candidates_before("t", &["Car"], &[], queued)
646 16 : .iter()
647 20 : .filter_map(|d| d["id"].as_str().map(str::to_owned))
648 16 : .collect();
649 16 : out.sort();
650 16 : out
651 16 : }
652 :
653 : /// a change queued before a subscription entered the mirror is
654 : /// not matched to it, however late it is drained; one queued after is.
655 : #[test]
656 4 : fn a_change_reaches_only_the_subscriptions_made_before_it() {
657 4 : let m = SubMirror::default();
658 16 : let sub = |id: &str| json!({"id": id, "entities": [{"type": "Car"}]});
659 4 : m.apply("t", "s:old", Some(sub("s:old")));
660 4 : let written = tick();
661 4 : m.apply("t", "s:new", Some(sub("s:new")));
662 4 : let later = tick();
663 4 : assert_eq!(ids(&m, written), ["s:old"]);
664 4 : assert_eq!(ids(&m, later), ["s:new", "s:old"]);
665 : // a PATCH of the old one does not make it new again
666 4 : m.apply("t", "s:old", Some(sub("s:old")));
667 4 : assert_eq!(ids(&m, written), ["s:old"]);
668 : // a deleted and recreated one is new
669 4 : m.apply("t", "s:old", None);
670 4 : m.apply("t", "s:old", Some(sub("s:old")));
671 4 : assert_eq!(ids(&m, later), ["s:new"]);
672 : // the unstamped path sees every subscription
673 4 : assert_eq!(m.candidates("t", &["Car"], &[]).len(), 2);
674 4 : }
675 :
676 : /// An empty periodic firing parks the document it evaluated, never one
677 : /// a write put in its place while it ran (the write cleared the park).
678 : #[test]
679 4 : fn an_empty_firing_does_not_park_a_document_written_during_it() {
680 4 : let m = SubMirror::default();
681 8 : let sub = |q: &str| json!({"id": "s:p", "timeInterval": 1, "q": q});
682 4 : m.apply("t", "s:p", Some(sub("a==1")));
683 4 : let evaluated = m.periodic_docs("t").pop().expect("doc");
684 4 : m.apply("t", "s:p", Some(sub("a==2")));
685 4 : m.park_quiet("t", "s:p", &evaluated, i64::MAX);
686 4 : assert_eq!(
687 4 : m.quiet_until("t", "s:p"),
688 : None,
689 : "the new document was parked"
690 : );
691 4 : let current = m.periodic_docs("t").pop().expect("doc");
692 4 : m.park_quiet("t", "s:p", ¤t, 7);
693 4 : assert_eq!(m.quiet_until("t", "s:p"), Some(7));
694 4 : }
695 : }
|
