Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Context source registration notifications (5.11, csourceNotification): triggers, the fan-out and the initial notification.
3 :
4 : use super::*;
5 :
6 : /// 5.11.7: which csource subs care about a registration change, and why.
7 178 : pub(super) fn csource_trigger(
8 178 : sub: &Value,
9 178 : before: Option<&Value>,
10 178 : after: Option<&Value>,
11 178 : ctx: &Context,
12 178 : ) -> Option<&'static str> {
13 356 : let m = |d: Option<&Value>| {
14 356 : d.is_some_and(|d| crate::registry::csr_matches_subscription(sub, d, ctx))
15 356 : };
16 178 : match (m(before), m(after)) {
17 62 : (false, true) => Some("newlyMatching"),
18 12 : (true, true) => Some("updated"),
19 60 : (true, false) => Some("noLongerMatching"),
20 44 : (false, false) => None,
21 : }
22 178 : }
23 :
24 : /// The notification validator reads EntityInfo.type as an array
25 : /// (`entities[0]["type"][0]`) — normalize to array form in notification data.
26 315 : pub(super) fn arrayify_entity_types(reg: &mut Value) {
27 315 : let Some(infos) = reg.get_mut("information").and_then(Value::as_array_mut) else {
28 0 : return;
29 : };
30 315 : for info in infos {
31 315 : let Some(es) = info.get_mut("entities").and_then(Value::as_array_mut) else {
32 0 : continue;
33 : };
34 315 : for e in es {
35 315 : if let Some(t) = e.get("type").filter(|t| t.is_string()).cloned() {
36 315 : if let Some(o) = e.as_object_mut() {
37 315 : o.insert("type".into(), Value::Array(vec![t]));
38 315 : }
39 0 : }
40 : }
41 : }
42 315 : }
43 :
44 : /// One prepared CSource notification, ready to send.
45 : pub struct CsourceJob {
46 : sub: Value,
47 : presented: Value,
48 : ctx: std::sync::Arc<antares_jsonld::Context>,
49 : reason: &'static str,
50 : }
51 :
52 : /// Registration create/update/delete → CSourceNotification fan-out (5.11.7),
53 : /// in two phases: `prepare_csource_jobs` runs IN the request path (store
54 : /// reads + matching + payload build — so job order is the handlers' commit
55 : /// order even on a slower store), and the caller spawns `send_csource_jobs`
56 : /// (network only — the ack must not block on the receiver: the ETSI mock
57 : /// replies only when the robot side wakes).
58 4256 : pub(crate) async fn prepare_csource_jobs(
59 4256 : st: &AppState,
60 4256 : tenant: &TenantId,
61 4256 : before: Option<Value>,
62 4256 : after: Option<Value>,
63 4256 : ) -> Vec<CsourceJob> {
64 4256 : let mut jobs = Vec::new();
65 : // 5.8.1.4: the Registration Subscriptions the distributed half owns are
66 : // not client resources, so they live under Kind::DistSub beside the
67 : // mapping documents — which carry no `type`. A document under the
68 : // client kind in the internal id namespace is a leftover of a release
69 : // that stored the internal ones there, and drives nothing.
70 4256 : let client = read_or_warn(
71 4256 : docs_where(st, tenant, Kind::CSourceSubscription, |d| {
72 152 : sub_str(d, "id")
73 152 : .is_some_and(|id| crate::registry::csr_kind(id) == Kind::CSourceSubscription)
74 152 : })
75 4256 : .await,
76 4256 : "the Context Source Registration Subscriptions of a changed registration",
77 : )
78 4256 : .into_iter();
79 4256 : let internal = read_or_warn(
80 4256 : docs_where(st, tenant, Kind::DistSub, |d| {
81 28 : d.get("type").and_then(Value::as_str) == Some("Subscription")
82 28 : })
83 4256 : .await,
84 4256 : "the internal Registration Subscriptions of a changed registration",
85 : )
86 4256 : .into_iter();
87 4256 : for sub in client.chain(internal) {
88 166 : if !is_active(&sub) || sub.get("timeInterval").is_some() {
89 12 : continue;
90 154 : }
91 154 : let ctx = sub_context(st, tenant, &sub).await;
92 154 : let Some(reason) = csource_trigger(&sub, before.as_ref(), after.as_ref(), &ctx) else {
93 40 : continue;
94 : };
95 114 : let spec = crate::registry::spec_for_subscription(&sub);
96 114 : let source = if reason == "noLongerMatching" {
97 52 : &before
98 : } else {
99 62 : &after
100 : };
101 114 : let Some(reg) = source.as_ref().or(before.as_ref()) else {
102 0 : continue;
103 : };
104 114 : let filtered = filter_csr(&spec, reg, &ctx);
105 114 : let mut presented = crate::registry::present_registration(&filtered, &ctx, false);
106 114 : arrayify_entity_types(&mut presented);
107 114 : jobs.push(CsourceJob {
108 114 : sub,
109 114 : presented,
110 114 : ctx,
111 114 : reason,
112 114 : });
113 : }
114 4256 : jobs
115 4256 : }
116 :
117 4119 : pub(crate) async fn send_csource_jobs(st: &AppState, tenant: &TenantId, jobs: Vec<CsourceJob>) {
118 4119 : for job in jobs {
119 : // 5.11.7: re-check the subscription still exists right
120 : // before the send — a deleted subscription must never notify.
121 110 : let sub_id = job
122 110 : .sub
123 110 : .get("id")
124 110 : .and_then(Value::as_str)
125 110 : .unwrap_or_default();
126 110 : let kind = crate::registry::csr_kind(sub_id);
127 110 : if !matches!(st.store.get(tenant, kind, sub_id).await, Ok(Some(_))) {
128 0 : continue;
129 110 : }
130 110 : deliver_as(
131 110 : st,
132 110 : tenant,
133 110 : kind,
134 110 : &job.sub,
135 110 : "ContextSourceNotification",
136 110 : vec![job.presented],
137 110 : &job.ctx,
138 110 : Some(job.reason),
139 110 : )
140 110 : .await;
141 : }
142 4119 : }
143 :
144 : /// Registration writes prepare one job per CSource subscription of the
145 : /// tenant (5.11.7) and every subscription with localOnly != true owns one,
146 : /// so a registration stream against many subscriptions queues
147 : /// subscriptions × registrations jobs faster than the sources drain them —
148 : /// at 10 000 × 100 the queued jobs were a 3 GB broker peak. The permit is
149 : /// taken in the request path: a write waits for a fan-out slot instead of
150 : /// stacking jobs, and the prepare order (the handlers' commit order) holds.
151 : pub(super) const CSOURCE_FANOUT_WIDTH: usize = 64;
152 : pub(super) static CSOURCE_FANOUT: tokio::sync::Semaphore =
153 : tokio::sync::Semaphore::const_new(CSOURCE_FANOUT_WIDTH);
154 :
155 : /// The fan-out slots one tenant may hold at once: a tenant
156 : /// whose CSource subscription endpoints hang keeps at most this many of the
157 : /// [`CSOURCE_FANOUT_WIDTH`] busy, so another tenant's registration writes
158 : /// still find a slot.
159 : pub(super) const CSOURCE_FANOUT_PER_TENANT: usize = CSOURCE_FANOUT_WIDTH / 8;
160 :
161 : type FanoutShares =
162 : std::sync::Mutex<std::collections::HashMap<String, std::sync::Arc<tokio::sync::Semaphore>>>;
163 : static FANOUT_SHARES: std::sync::LazyLock<FanoutShares> =
164 : std::sync::LazyLock::new(FanoutShares::default);
165 :
166 : /// Both permits one fan-out holds until its sends settle: the tenant's
167 : /// share first, as `delivery_permits` takes it, so a tenant at its share
168 : /// waits there and never inside the broker's slots. A share lives only
169 : /// while a fan-out holds it: creating one drops every entry nothing holds,
170 : /// so the map is bounded by the width, not by the tenants ever seen.
171 4296 : pub(super) async fn fanout_permits(tenant: &TenantId) -> impl Sized {
172 4296 : let shares = {
173 : // a poisoned map is a past panic; what it holds is rebuildable
174 4296 : let mut map = FANOUT_SHARES.lock().unwrap_or_else(|p| p.into_inner());
175 4296 : match map.get(tenant.as_str()) {
176 4164 : Some(s) => std::sync::Arc::clone(s),
177 : None => {
178 133 : map.retain(|_, s| std::sync::Arc::strong_count(s) > 1);
179 132 : let s = std::sync::Arc::new(tokio::sync::Semaphore::new(CSOURCE_FANOUT_PER_TENANT));
180 132 : map.insert(tenant.as_str().to_owned(), std::sync::Arc::clone(&s));
181 132 : s
182 : }
183 : }
184 : };
185 : // never closed, so acquire only fails if it were — then send unbounded
186 4296 : let share = shares.acquire_owned().await.ok();
187 4292 : (share, CSOURCE_FANOUT.acquire().await.ok())
188 4292 : }
189 :
190 : /// Registration create/update/delete → prepare in the request path, send
191 : /// spawned (the ack must not block on the receiver), bounded as above.
192 4256 : pub(crate) async fn csource_fanout(
193 4256 : st: &AppState,
194 4256 : tenant: &TenantId,
195 4256 : before: Option<Value>,
196 4256 : after: Option<Value>,
197 4256 : ) {
198 4256 : let permit = fanout_permits(tenant).await;
199 4256 : let jobs = prepare_csource_jobs(st, tenant, before, after).await;
200 4256 : let (st2, t2) = (st.clone(), tenant.clone());
201 4256 : crate::spawn(async move {
202 4119 : send_csource_jobs(&st2, &t2, jobs).await;
203 4119 : drop(permit);
204 4119 : });
205 4256 : }
206 :
207 : /// POST a CSourceNotification (5.3.2) under the same body bound as every
208 : /// other one: 5.11.2.4 sends "all matching Context Source Registrations",
209 : /// and a broker holding 100 000 of them must not turn that into one
210 : /// unbounded request. Over the cap the set leaves as several notifications,
211 : /// each carrying whole registrations — the trade [`deliver`] already makes
212 : /// for entity Notifications.
213 165 : pub(super) async fn deliver_csource(
214 165 : st: &AppState,
215 165 : tenant: &TenantId,
216 165 : sub: &Value,
217 165 : data: Vec<Value>,
218 165 : ctx: &Context,
219 165 : reason: &str,
220 165 : ) {
221 165 : let kind = crate::registry::csr_kind(sub_str(sub, "id").unwrap_or_default());
222 169 : for chunk in chunk_by_bytes(data, *crate::bounds::MAX_BODY_BYTES) {
223 169 : deliver_as(
224 169 : st,
225 169 : tenant,
226 169 : kind,
227 169 : sub,
228 169 : "ContextSourceNotification",
229 169 : chunk,
230 169 : ctx,
231 169 : Some(reason),
232 169 : )
233 169 : .await;
234 : }
235 129 : }
236 :
237 : /// Initial / post-update CSourceNotification with all currently matching
238 : /// registrations (5.11.2.4 / 5.11.3.4).
239 1186 : pub(crate) async fn csource_initial(st: &AppState, tenant: &TenantId, sub_id: &str) {
240 1186 : let Some(sub) = st
241 1186 : .store
242 1186 : .get(tenant, crate::registry::csr_kind(sub_id), sub_id)
243 1186 : .await
244 1186 : .ok()
245 1186 : .flatten()
246 : else {
247 0 : return;
248 : };
249 1186 : if !is_active(&sub) {
250 8 : return;
251 1178 : }
252 1178 : let ctx = sub_context(st, tenant, &sub).await;
253 1178 : let spec = crate::registry::spec_for_subscription(&sub);
254 1178 : let data: Vec<Value> = read_or_warn(
255 1178 : docs_where(st, tenant, Kind::Registration, |r| {
256 188 : crate::registry::csr_matches_subscription(&sub, r, &ctx)
257 188 : })
258 1178 : .await,
259 1178 : "the registrations an initial Context Source Notification carries",
260 : )
261 1178 : .into_iter()
262 1178 : .map(|r| {
263 165 : let mut p =
264 165 : crate::registry::present_registration(&filter_csr(&spec, &r, &ctx), &ctx, false);
265 165 : arrayify_entity_types(&mut p);
266 165 : p
267 165 : })
268 1178 : .collect();
269 1178 : if data.is_empty() {
270 1045 : return; // nothing currently matching ⇒ no initial notification
271 133 : }
272 133 : deliver_csource(st, tenant, &sub, data, &ctx, "newlyMatching").await;
273 1182 : }
274 :
275 : /// Registration copy reduced to the matching RegistrationInfo elements
276 : /// (5.10.2.5 / 5.11.7 "filtered Context Source Registrations").
277 315 : pub(super) fn filter_csr(spec: &crate::registry::CsrSpec, reg: &Value, ctx: &Context) -> Value {
278 315 : let mut out = reg.clone();
279 315 : let matching: Vec<Value> = crate::registry::matching_infos(spec, reg, ctx)
280 315 : .into_iter()
281 315 : .cloned()
282 315 : .collect();
283 315 : if !matching.is_empty() {
284 315 : if let Some(o) = out.as_object_mut() {
285 315 : o.insert("information".into(), Value::Array(matching));
286 315 : }
287 0 : }
288 315 : out
289 315 : }
|
