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 490 : fn reg_keys(doc: &Value) -> (Option<Vec<String>>, Option<Vec<String>>) {
62 490 : let (mut types, mut ids) = (Vec::new(), Vec::new());
63 490 : 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 490 : let Some(infos) = doc.get("information").and_then(Value::as_array) else {
68 8 : return (None, None);
69 : };
70 482 : if infos.is_empty() {
71 0 : return (None, None);
72 482 : }
73 484 : for info in infos {
74 : // 5.2.9: a RegistrationInfo may name only propertyNames /
75 : // relationshipNames, which restricts attributes and no entity.
76 484 : let Some(entities) = info.get("entities").and_then(Value::as_array) else {
77 90 : return (None, None);
78 : };
79 394 : if entities.is_empty() {
80 0 : return (None, None);
81 394 : }
82 394 : for ei in entities {
83 394 : match crate::registry::ei_types(ei) {
84 394 : ts if ts.is_empty() => any_type = true,
85 228 : ts => types.extend(ts.into_iter().map(str::to_owned)),
86 : }
87 394 : if ei.get("idPattern").is_some() {
88 90 : any_id = true;
89 90 : } else {
90 304 : match ei.get("id").and_then(Value::as_str) {
91 184 : Some(id) => ids.push(id.to_owned()),
92 120 : None => any_id = true,
93 : }
94 : }
95 : }
96 : }
97 392 : ((!any_type).then_some(types), (!any_id).then_some(ids))
98 490 : }
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 272 : fn bucketed<'a>(
104 272 : keys: Option<&[String]>,
105 272 : by: &'a std::collections::HashMap<String, std::collections::HashSet<String>>,
106 272 : any: &'a std::collections::HashSet<String>,
107 272 : ) -> Option<std::collections::HashSet<&'a str>> {
108 272 : let keys = keys?;
109 212 : let mut out: std::collections::HashSet<&str> = any.iter().map(String::as_str).collect();
110 212 : for k in keys {
111 212 : if let Some(ids) = by.get(k) {
112 198 : out.extend(ids.iter().map(String::as_str));
113 198 : }
114 : }
115 212 : Some(out)
116 272 : }
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 488 : fn remove(&mut self, id: &str) {
123 488 : let Some(old) = self.docs.remove(id) else {
124 480 : return;
125 : };
126 8 : let (types, ids) = reg_keys(&old);
127 8 : let unfile =
128 : |keys: Option<Vec<String>>,
129 : by: &mut std::collections::HashMap<String, std::collections::HashSet<String>>,
130 16 : any: &mut std::collections::HashSet<String>| {
131 16 : match keys {
132 10 : None => {
133 10 : any.remove(id);
134 10 : }
135 6 : Some(ks) => {
136 6 : for k in ks {
137 6 : if let Some(set) = by.get_mut(&k) {
138 6 : set.remove(id);
139 6 : if set.is_empty() {
140 6 : by.remove(&k);
141 6 : }
142 0 : }
143 : }
144 : }
145 : }
146 16 : };
147 8 : unfile(types, &mut self.by_type, &mut self.any_type);
148 8 : unfile(ids, &mut self.by_id, &mut self.any_id);
149 488 : }
150 :
151 482 : fn insert(&mut self, id: &str, doc: Value) {
152 482 : self.remove(id);
153 482 : let (types, ids) = reg_keys(&doc);
154 482 : let file =
155 : |keys: Option<Vec<String>>,
156 : by: &mut std::collections::HashMap<String, std::collections::HashSet<String>>,
157 964 : any: &mut std::collections::HashSet<String>| {
158 964 : match keys {
159 562 : None => {
160 562 : any.insert(id.to_owned());
161 562 : }
162 402 : Some(ks) => {
163 406 : for k in ks {
164 406 : by.entry(k).or_default().insert(id.to_owned());
165 406 : }
166 : }
167 : }
168 964 : };
169 482 : file(types, &mut self.by_type, &mut self.any_type);
170 482 : file(ids, &mut self.by_id, &mut self.any_id);
171 482 : self.docs.insert(id.to_owned(), std::sync::Arc::new(doc));
172 482 : }
173 :
174 6 : fn is_empty(&self) -> bool {
175 6 : self.docs.is_empty()
176 6 : }
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 52 : fn apply(&self, tenant: &str, id: &str, doc: Option<Value>) {
191 52 : DocMirror::apply(self, tenant, id, doc);
192 52 : }
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 : }
258 :
259 : /// Which index bucket(s) one stored subscription doc belongs in.
260 : pub(crate) enum Keys {
261 : Types(Vec<String>),
262 : Attrs(Vec<String>),
263 : Broad,
264 : }
265 :
266 1868 : pub(crate) fn index_keys(doc: &Value) -> Keys {
267 1868 : if let Some(entities) = doc.get("entities").and_then(Value::as_array) {
268 1658 : let mut types = Vec::new();
269 1662 : for e in entities {
270 1662 : match e.get("type").and_then(Value::as_str) {
271 : // 4.17 selection expressions (and any wildcard) are evaluated
272 : // at match time — the index cannot prove them, so the whole
273 : // sub goes broad (entries are OR-ed: one opaque entry taints
274 : // the union).
275 1638 : Some(t) if !t.contains(['|', ',', ';', '(', ')', '*']) => {
276 1502 : types.push(t.to_owned());
277 1502 : }
278 160 : _ => return Keys::Broad,
279 : }
280 : }
281 1498 : if types.is_empty() {
282 0 : return Keys::Broad;
283 1498 : }
284 1498 : return Keys::Types(types);
285 210 : }
286 210 : if let Some(watched) = doc.get("watchedAttributes").and_then(Value::as_array) {
287 110 : let attrs: Vec<String> = watched
288 110 : .iter()
289 110 : .filter_map(Value::as_str)
290 110 : .map(str::to_owned)
291 110 : .collect();
292 110 : if attrs.len() == watched.len() && !attrs.is_empty() {
293 110 : return Keys::Attrs(attrs);
294 0 : }
295 100 : }
296 100 : Keys::Broad
297 1868 : }
298 :
299 : impl SubMirror {
300 : /// Apply one KV delta: `None` doc = deleted. Rekeys the index from the
301 : /// old doc before inserting the new one.
302 1146 : pub fn apply(&self, tenant: &str, id: &str, doc: Option<Value>) {
303 1146 : let mut map = self
304 1146 : .map
305 1146 : .write()
306 1146 : .unwrap_or_else(std::sync::PoisonError::into_inner);
307 1146 : let t = map.entry(tenant.to_owned()).or_default();
308 1146 : if let Some(old) = t.docs.remove(id) {
309 402 : match index_keys(&old) {
310 400 : Keys::Types(ts) => {
311 400 : for ty in ts {
312 400 : if let Some(s) = t.by_type.get_mut(&ty) {
313 400 : s.remove(id);
314 400 : if s.is_empty() {
315 348 : t.by_type.remove(&ty);
316 348 : }
317 0 : }
318 : }
319 : }
320 0 : Keys::Attrs(ats) => {
321 0 : for a in ats {
322 0 : if let Some(s) = t.by_attr.get_mut(&a) {
323 0 : s.remove(id);
324 0 : if s.is_empty() {
325 0 : t.by_attr.remove(&a);
326 0 : }
327 0 : }
328 : }
329 : }
330 2 : Keys::Broad => {
331 2 : t.broad.remove(id);
332 2 : }
333 : }
334 744 : }
335 1146 : if let Some(d) = doc {
336 1086 : if d.get("timeInterval").is_some() {
337 24 : // A periodic subscription just appeared or changed its
338 24 : // anchor: the sweep clock computed without it must not hold
339 24 : // it back (5.8.6).
340 24 : self.next_sub_sweep_ms
341 24 : .store(0, std::sync::atomic::Ordering::Relaxed);
342 1062 : }
343 1086 : match index_keys(&d) {
344 762 : Keys::Types(ts) => {
345 766 : for ty in ts {
346 766 : t.by_type.entry(ty).or_default().insert(id.to_owned());
347 766 : }
348 : }
349 106 : Keys::Attrs(ats) => {
350 106 : for a in ats {
351 106 : t.by_attr.entry(a).or_default().insert(id.to_owned());
352 106 : }
353 : }
354 218 : Keys::Broad => {
355 218 : t.broad.insert(id.to_owned());
356 218 : }
357 : }
358 1086 : t.docs.insert(id.to_owned(), std::sync::Arc::new(d));
359 60 : }
360 1146 : if map.get(tenant).is_some_and(|t| t.docs.is_empty()) {
361 56 : map.remove(tenant);
362 1090 : }
363 1146 : }
364 :
365 : /// 5.11.7: a Context Source Registration Subscription was written. Only
366 : /// the clock moves — these are matched against registrations rather than
367 : /// entities, so they are not carried in the candidate index, and the
368 : /// sweep reads the tenant's rows from the store once it wakes.
369 8 : pub(crate) fn csub_written(&self) {
370 8 : self.next_csub_sweep_ms
371 8 : .store(0, std::sync::atomic::Ordering::Relaxed);
372 8 : }
373 :
374 : /// The hot path: subscriptions that could possibly fire for a change
375 : /// touching these entity types and these changed attributes. Union of
376 : /// the type hits, the attr hits and the broad bucket — a superset of
377 : /// the firing set, never a subset.
378 2511 : pub fn candidates(
379 2511 : &self,
380 2511 : tenant: &str,
381 2511 : types: &[&str],
382 2511 : changed_attrs: &[&str],
383 2511 : ) -> Vec<std::sync::Arc<Value>> {
384 2511 : let map = self
385 2511 : .map
386 2511 : .read()
387 2511 : .unwrap_or_else(std::sync::PoisonError::into_inner);
388 2511 : let Some(t) = map.get(tenant) else {
389 2081 : return Vec::new();
390 : };
391 430 : let mut ids: std::collections::HashSet<&str> = t.broad.iter().map(String::as_str).collect();
392 432 : for ty in types {
393 432 : if let Some(s) = t.by_type.get(*ty) {
394 416 : ids.extend(s.iter().map(String::as_str));
395 416 : }
396 : }
397 452 : for a in changed_attrs {
398 452 : if let Some(s) = t.by_attr.get(*a) {
399 106 : ids.extend(s.iter().map(String::as_str));
400 346 : }
401 : }
402 430 : ids.iter()
403 12230 : .filter_map(|id| t.docs.get(*id).cloned())
404 430 : .collect()
405 2511 : }
406 :
407 : #[cfg(any(test, feature = "test-kit"))]
408 8 : pub fn docs(&self, tenant: &str) -> Vec<std::sync::Arc<Value>> {
409 8 : self.map
410 8 : .read()
411 8 : .unwrap_or_else(std::sync::PoisonError::into_inner)
412 8 : .get(tenant)
413 8 : .map(|t| t.docs.values().cloned().collect())
414 8 : .unwrap_or_default()
415 8 : }
416 :
417 : /// The interval sweep's whole input: the tenant's periodic (5.2.12
418 : /// `timeInterval`) subscriptions. The walk is over every subscription the
419 : /// tenant holds, but only the periodic ones are carried out of it — the
420 : /// sweep clocks keep a tick with nothing due from reaching this at all.
421 131 : pub(crate) fn periodic_docs(&self, tenant: &str) -> Vec<std::sync::Arc<Value>> {
422 131 : self.map
423 131 : .read()
424 131 : .unwrap_or_else(std::sync::PoisonError::into_inner)
425 131 : .get(tenant)
426 131 : .map(|t| {
427 40 : t.docs
428 40 : .values()
429 48 : .filter(|d| d.get("timeInterval").is_some())
430 40 : .cloned()
431 40 : .collect()
432 40 : })
433 131 : .unwrap_or_default()
434 131 : }
435 :
436 : #[cfg(any(test, feature = "test-kit"))]
437 2 : pub fn tenants(&self) -> Vec<String> {
438 2 : self.map
439 2 : .read()
440 2 : .unwrap_or_else(std::sync::PoisonError::into_inner)
441 2 : .keys()
442 2 : .cloned()
443 2 : .collect()
444 2 : }
445 : }
446 :
447 : #[cfg(test)]
448 : impl SubMirror {
449 : /// Poison the index lock from a thread that panics while holding it:
450 : /// what a matcher that dies mid-write leaves behind, for the
451 : /// supervision tests.
452 4 : pub(crate) fn poison(self: &std::sync::Arc<Self>) {
453 4 : let m = std::sync::Arc::clone(self);
454 4 : let _ = std::thread::spawn(move || {
455 4 : let _held = m
456 4 : .map
457 4 : .write()
458 4 : .unwrap_or_else(std::sync::PoisonError::into_inner);
459 4 : panic!("poison the mirror");
460 : })
461 4 : .join();
462 4 : }
463 : }
464 :
465 : impl Mirror for SubMirror {
466 8 : fn apply(&self, tenant: &str, id: &str, doc: Option<Value>) {
467 8 : SubMirror::apply(self, tenant, id, doc);
468 8 : }
469 :
470 2 : fn csub_written(&self) {
471 2 : SubMirror::csub_written(self);
472 2 : }
473 : }
474 :
475 : impl DocMirror {
476 : /// Apply one KV delta: `None` doc = deleted.
477 488 : pub fn apply(&self, tenant: &str, id: &str, doc: Option<Value>) {
478 488 : let mut map = self
479 488 : .map
480 488 : .write()
481 488 : .unwrap_or_else(std::sync::PoisonError::into_inner);
482 488 : match doc {
483 482 : Some(d) => {
484 482 : map.entry(tenant.to_owned()).or_default().insert(id, d);
485 482 : }
486 : None => {
487 6 : if let Some(t) = map.get_mut(tenant) {
488 6 : t.remove(id);
489 6 : if t.is_empty() {
490 4 : map.remove(tenant);
491 4 : }
492 0 : }
493 : }
494 : }
495 488 : }
496 :
497 : /// The registrations of one tenant that can match these ids and types,
498 : /// shared rather than copied.
499 : ///
500 : /// Both dimensions are optional and each narrows on its own; an absent
501 : /// one is a dimension the caller asked nothing about, never an empty
502 : /// answer. With both given a registration has to survive both, which is
503 : /// the store's rule too.
504 152 : pub fn matching(
505 152 : &self,
506 152 : tenant: &str,
507 152 : ids: Option<&[String]>,
508 152 : types: Option<&[String]>,
509 152 : ) -> Vec<std::sync::Arc<Value>> {
510 152 : let map = self
511 152 : .map
512 152 : .read()
513 152 : .unwrap_or_else(std::sync::PoisonError::into_inner);
514 152 : let Some(t) = map.get(tenant) else {
515 16 : return Vec::new();
516 : };
517 136 : let by_type = bucketed(types, &t.by_type, &t.any_type);
518 136 : let by_id = bucketed(ids, &t.by_id, &t.any_id);
519 136 : let pick = |keys: std::collections::HashSet<&str>| -> Vec<std::sync::Arc<Value>> {
520 124 : keys.into_iter()
521 7268 : .filter_map(|k| t.docs.get(k).map(std::sync::Arc::clone))
522 124 : .collect()
523 124 : };
524 136 : match (by_type, by_id) {
525 12 : (None, None) => t.docs.values().map(std::sync::Arc::clone).collect(),
526 36 : (Some(k), None) | (None, Some(k)) => pick(k),
527 88 : (Some(a), Some(b)) => pick(a.intersection(&b).copied().collect()),
528 : }
529 152 : }
530 :
531 18 : pub fn docs(&self, tenant: &str) -> Vec<Value> {
532 18 : self.matching(tenant, None, None)
533 18 : .into_iter()
534 18 : .map(|d| (*d).clone())
535 18 : .collect()
536 18 : }
537 :
538 : #[cfg(any(test, feature = "test-kit"))]
539 6 : pub fn tenants(&self) -> Vec<String> {
540 6 : self.map
541 6 : .read()
542 6 : .unwrap_or_else(std::sync::PoisonError::into_inner)
543 6 : .keys()
544 6 : .cloned()
545 6 : .collect()
546 6 : }
547 : }
|
