Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Shared application state.
3 :
4 : use antares_jsonld::Loader;
5 : #[cfg(feature = "test-kit")]
6 : use antares_sql::store::{any::AnyStore, Store};
7 : use antares_store::{CurrentStateDriver, TemporalDriver};
8 : use std::sync::Arc;
9 : // Clock rule: std Instant panics on wasm32.
10 : #[cfg(not(target_arch = "wasm32"))]
11 : use std::time::Instant;
12 : #[cfg(target_arch = "wasm32")]
13 : use web_time::Instant;
14 :
15 : /// ANTARES_TEMPORAL_RECORD — the auto-recording gate on the write path.
16 : /// Direct temporal-API writes (POST /temporal/entities…) are never gated:
17 : /// this decides what the ENTITY endpoints leave behind as history.
18 : #[derive(Clone, Copy, Debug, PartialEq, Eq)]
19 : pub enum TemporalRecord {
20 : /// `all` (default): every changed attribute instance.
21 : All,
22 : /// `observed`: only instances carrying `observedAt` — the spec's own
23 : /// measurement axis (4.5.7); metadata-shaped writes leave no history.
24 : Observed,
25 : /// `none`: nothing is auto-recorded; the temporal endpoints still serve
26 : /// what the temporal API was given directly (unlike ANTARES_TEMPORAL=none,
27 : /// which turns the temporal seam off entirely).
28 : None,
29 : /// `ruled`: history is opt-in. An Entity records only
30 : /// when its own `historyRetention`, the retention rule of one of its
31 : /// types or its Tenant's `*` rule gives a duration or `forever`; a type
32 : /// no rule names keeps its current state only.
33 : Ruled,
34 : }
35 :
36 : impl std::str::FromStr for TemporalRecord {
37 : type Err = String;
38 20 : fn from_str(s: &str) -> Result<Self, String> {
39 20 : match s {
40 20 : "all" => Ok(Self::All),
41 6 : "observed" => Ok(Self::Observed),
42 4 : "none" => Ok(Self::None),
43 2 : "ruled" => Ok(Self::Ruled),
44 2 : other => Err(format!(
45 2 : "ANTARES_TEMPORAL_RECORD: unknown mode {other} (all|observed|none|ruled)"
46 2 : )),
47 : }
48 20 : }
49 : }
50 :
51 : /// Behaviour profile (docs/conformance/profiles.md): where the two
52 : /// spec versions or the conformance suites accept two answers, the one this
53 : /// deployment gives. Startup configuration only, never per request; the
54 : /// default of every member is the behaviour the pinned ETSI suite asserts.
55 : /// No member relaxes authorization, tenant isolation or an input limit.
56 : #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
57 : pub struct Profile {
58 : /// `ANTARES_NON_OBJECT_BODY=bad-request-data`: a JSON body that is no
59 : /// object (`[]`, `[1,2,3]`) where an object is expected answers
60 : /// BadRequestData (TS 104 175 8.2.3 second bullet, forge all-fixes,
61 : /// Coraine) instead of InvalidRequest (ETSI 001_02_02).
62 : pub non_object_bad_request_data: bool,
63 : /// `ANTARES_FED_WRITE_SOURCE_404=ignore`: on a distributed write, a
64 : /// source's 404 beside a success counts as "no data" (204, forge
65 : /// all-fixes, Coraine) instead of a part of the 207 (ETSI D002_02_*_inc).
66 : pub fed_write_ignore_404: bool,
67 : /// `ANTARES_IMPLICIT_TENANTS=listed`: only the operations TS 104 175
68 : /// 7.8.2 lists create a Tenant; Batch Upsert on an unknown Tenant
69 : /// answers NonexistentTenant (Coraine tenant_implicit_create).
70 : pub implicit_tenants_listed_only: bool,
71 : }
72 :
73 : impl Profile {
74 : /// Reads the profile keys; an unknown value is a startup error.
75 12 : pub fn from_env() -> Result<Self, String> {
76 36 : let get = |k: &str| std::env::var(k).ok();
77 12 : Self::from_vars(
78 12 : get("ANTARES_NON_OBJECT_BODY"),
79 12 : get("ANTARES_FED_WRITE_SOURCE_404"),
80 12 : get("ANTARES_IMPLICIT_TENANTS"),
81 : )
82 12 : }
83 :
84 : /// [`Profile::from_env`] over given values (unset = `None`).
85 20 : pub fn from_vars(
86 20 : non_object_body: Option<String>,
87 20 : fed_write_source_404: Option<String>,
88 20 : implicit_tenants: Option<String>,
89 20 : ) -> Result<Self, String> {
90 : Ok(Self {
91 20 : non_object_bad_request_data: profile_choice(
92 20 : "ANTARES_NON_OBJECT_BODY",
93 20 : non_object_body,
94 20 : ["invalid-request", "bad-request-data"],
95 0 : )?,
96 20 : fed_write_ignore_404: profile_choice(
97 20 : "ANTARES_FED_WRITE_SOURCE_404",
98 20 : fed_write_source_404,
99 20 : ["report", "ignore"],
100 2 : )?,
101 18 : implicit_tenants_listed_only: profile_choice(
102 18 : "ANTARES_IMPLICIT_TENANTS",
103 18 : implicit_tenants,
104 18 : ["with-upsert", "listed"],
105 0 : )?,
106 : })
107 20 : }
108 : }
109 :
110 : /// `false` for the default value (or unset/empty), `true` for the other.
111 58 : fn profile_choice(
112 58 : key: &str,
113 58 : value: Option<String>,
114 58 : [default, other]: [&str; 2],
115 58 : ) -> Result<bool, String> {
116 58 : match value.as_deref().map(str::trim) {
117 44 : None | Some("") => Ok(false),
118 14 : Some(v) if v == default => Ok(false),
119 8 : Some(v) if v == other => Ok(true),
120 2 : Some(v) => Err(format!("{key}: unknown value {v:?} ({default}|{other})")),
121 : }
122 58 : }
123 :
124 : /// History gate 3's set: the (Tenant, expanded type) pairs that
125 : /// record no history, shared by every clone of the state.
126 : pub type HistoryOff = Arc<std::sync::RwLock<HistoryRules>>;
127 :
128 : /// When an attribute recorded `onChange` writes a row: when its value moved past `deadband` (a number) or
129 : /// `min_distance` metres (a Point), or differs at all (anything else), or
130 : /// `heartbeat` passed since its last row.
131 : #[derive(Debug, Clone, Copy, PartialEq)]
132 : pub struct OnChange {
133 : pub heartbeat: std::time::Duration,
134 : pub deadband: Option<f64>,
135 : pub min_distance: Option<f64>,
136 : }
137 :
138 : /// How an attribute records history. Unset, it is `every` (a row per
139 : /// changed instance), and `observed` under `ANTARES_TEMPORAL_RECORD=ruled`.
140 : #[derive(Debug, Clone, Copy, PartialEq)]
141 : pub enum RecordMode {
142 : /// `every`: a row per changed instance.
143 : Every,
144 : /// `never`: no row, for a bookkeeping attribute such as
145 : /// `dateModified`.
146 : Never,
147 : /// `onChange`: a row when the value moved or the heartbeat passed.
148 : OnChange(OnChange),
149 : /// `observed`: a row only for an instance whose `observedAt` is
150 : /// newer than the attribute's last row, so a value re-sent with its old
151 : /// `observedAt`, or sent with none, adds nothing.
152 : Observed,
153 : }
154 :
155 : /// How one type records history, for the type and per expanded attribute:
156 : /// `None` takes the broker's default ([`RecordMode`]).
157 : #[derive(Debug, Default, Clone, PartialEq)]
158 : pub struct Recording {
159 : pub default: Option<RecordMode>,
160 : pub attributes: std::collections::HashMap<String, Option<RecordMode>>,
161 : }
162 :
163 : impl Recording {
164 : /// How `attr` records: its own setting, else the type's.
165 : #[must_use]
166 52920 : pub fn of(&self, attr: &str) -> Option<RecordMode> {
167 52920 : self.attributes.get(attr).copied().unwrap_or(self.default)
168 52920 : }
169 :
170 : /// Whether some attribute records `onChange`.
171 : #[must_use]
172 128 : pub fn has_on_change(&self) -> bool {
173 128 : std::iter::once(&self.default)
174 128 : .chain(self.attributes.values())
175 132 : .any(|m| matches!(m, Some(RecordMode::OnChange(_))))
176 128 : }
177 : }
178 :
179 : /// What the write path needs of the Tenants' history retention: which writes record none.
180 : #[derive(Debug, Default, Clone, PartialEq)]
181 : pub struct HistoryRules {
182 : /// (Tenant, type) pairs whose rule says `history: "off"`.
183 : pub off: std::collections::HashSet<(String, String)>,
184 : /// (Tenant, type) pairs with a history rule of their own (a duration,
185 : /// `off` or `forever`), which their Tenant's `*` rule does not reach.
186 : pub ruled: std::collections::HashSet<(String, String)>,
187 : /// Tenants whose `*` rule says `history: "off"`.
188 : pub space_off: std::collections::HashSet<String>,
189 : /// Tenants whose `*` rule gives a duration or `forever`: under
190 : /// `ANTARES_TEMPORAL_RECORD=ruled` every type without a rule of its own
191 : /// records.
192 : pub space_on: std::collections::HashSet<String>,
193 : /// (Tenant, type) pairs whose rule records some attribute other than
194 : /// `every`.
195 : pub recording: std::collections::HashMap<(String, String), Recording>,
196 : }
197 :
198 : impl HistoryRules {
199 : /// No rule turns any history off.
200 : #[must_use]
201 137584 : pub fn is_empty(&self) -> bool {
202 137584 : self.off.is_empty() && self.space_off.is_empty()
203 137584 : }
204 : }
205 :
206 : /// The Tenants the store recently answered "exists" for. The 5.5.10
207 : /// wall asks about the request's Tenant ahead of every operation that
208 : /// names one, and nearly every answer is "exists", so a name it holds skips
209 : /// the store for [`KnownTenants::TTL`]. It keeps positive answers only: a
210 : /// write can create a Tenant at any moment, so a "does not exist" is never
211 : /// kept. A purge on this process forgets the name at once; a purge on
212 : /// another process sharing the store is seen once the entry ages out.
213 : #[derive(Debug, Default)]
214 : pub struct KnownTenants(std::sync::Mutex<std::collections::HashMap<String, Instant>>);
215 :
216 : impl KnownTenants {
217 : /// How long a remembered "exists" stands in for the store's answer.
218 : pub const TTL: std::time::Duration = std::time::Duration::from_secs(30);
219 : /// The most names held at once (ADR-0001's 10 000-tenant target): a
220 : /// full map takes no new name until its old entries age out.
221 : pub const CAP: usize = 10_000;
222 :
223 982 : fn map(&self) -> std::sync::MutexGuard<'_, std::collections::HashMap<String, Instant>> {
224 982 : self.0
225 982 : .lock()
226 982 : .unwrap_or_else(std::sync::PoisonError::into_inner)
227 982 : }
228 :
229 : /// Did the store say this Tenant exists within the last [`Self::TTL`]?
230 : #[must_use]
231 756 : pub fn holds(&self, tenant: &antares_model::TenantId) -> bool {
232 756 : self.map()
233 756 : .get(tenant.as_str())
234 756 : .is_some_and(|at| at.elapsed() < Self::TTL)
235 756 : }
236 :
237 : /// The store just said this Tenant exists.
238 198 : pub fn remember(&self, tenant: &antares_model::TenantId) {
239 198 : let mut map = self.map();
240 198 : if map.len() >= Self::CAP && !map.contains_key(tenant.as_str()) {
241 0 : map.retain(|_, at| at.elapsed() < Self::TTL);
242 0 : if map.len() >= Self::CAP {
243 0 : return;
244 0 : }
245 198 : }
246 198 : map.insert(tenant.as_str().to_owned(), Instant::now());
247 198 : }
248 :
249 : /// The Tenant was purged: the next request asks the store again.
250 28 : pub fn forget(&self, tenant: &antares_model::TenantId) {
251 28 : self.map().remove(tenant.as_str());
252 28 : }
253 : }
254 :
255 : /// The handler `wire` installs for a notification that arrives on the
256 : /// internal distributed-subscription endpoint (5.8.1.4).
257 : pub type CsourceNotification = Arc<
258 : dyn for<'a> Fn(
259 : &'a AppState,
260 : &'a antares_model::TenantId,
261 : &'a str,
262 : Option<&'a str>,
263 : &'a [serde_json::Value],
264 : ) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send + 'a>>
265 : + Send
266 : + Sync,
267 : >;
268 :
269 : /// The request headers an in-process call carries from its caller
270 : /// ([`AppState::call`]): the two that select WHICH data the operation runs
271 : /// against, and the one that says what its terms mean. Everything else is
272 : /// the façade's own to set — an inner request is not the outer one.
273 : static PROPAGATED: [axum::http::HeaderName; 3] = [
274 : axum::http::HeaderName::from_static("ngsild-tenant"),
275 : axum::http::HeaderName::from_static("ngsild-snapshot"),
276 : axum::http::HeaderName::from_static("link"),
277 : ];
278 :
279 : /// How many [`AppState::call`] frames the running task is inside: zero
280 : /// outside an in-process call. A caller that must not reach the router past
281 : /// [`crate::bounds::MAX_IN_PROCESS_CALL_DEPTH`] reads it to answer for
282 : /// itself rather than take the router's 500.
283 : #[must_use]
284 112 : pub fn in_process_depth() -> usize {
285 112 : INPROCESS_DEPTH.try_with(|d| *d).unwrap_or(0)
286 112 : }
287 :
288 : tokio::task_local! {
289 : /// How many [`AppState::call`] frames this task is already inside, held
290 : /// against [`crate::bounds::MAX_IN_PROCESS_CALL_DEPTH`]. Absent outside a
291 : /// façade call, which reads as zero.
292 : pub(crate) static INPROCESS_DEPTH: usize;
293 : }
294 :
295 : /// The broker's state as every handler and layer receives it. Cloned at
296 : /// every middleware layer and extractor of every request, so a clone is
297 : /// three reference counts: the per-process part sits behind one `Arc`
298 : /// ([`Shared`], read through `Deref`), and only what a narrowed state
299 : /// changes, the view and its in-process router memo, is held per state.
300 : /// Writing a shared field (`DerefMut`) copies the shared part when another
301 : /// state holds it, so a write made before serving is never seen by a state
302 : /// cloned earlier, as it was when every field was copied.
303 : #[derive(Clone)]
304 : pub struct AppState {
305 : shared: Arc<Shared>,
306 : /// The router [`AppState::call`] serves in-process requests through,
307 : /// built on first use. Empty until a façade actually calls: building it
308 : /// costs about 1.5 ms, which is worth memoizing per state and not worth
309 : /// paying in a host that never makes an in-process call.
310 : pub(crate) inbound: Arc<std::sync::OnceLock<axum::Router>>,
311 : }
312 :
313 : /// The per-process part of [`AppState`]: every handle and setting a state
314 : /// and its narrowed copies share.
315 : #[derive(Clone)]
316 : pub struct Shared {
317 : pub store: Arc<dyn CurrentStateDriver>,
318 : /// The temporal driver — by default the same backend instance as
319 : /// `store`; a deployment may load a different one (or none).
320 : pub temporal: Arc<dyn TemporalDriver>,
321 : /// What the current-state driver is CALLED, reported by `/q/health` (NOT
322 : /// by `/info/sourceIdentity`, which is a spec resource) — a name, not an
323 : /// enumeration: a driver from outside this workspace mounts the same way
324 : /// as one from inside, and nothing here branches on it.
325 : pub store_name: String,
326 : /// The temporal backend `/q/health` names: the store's own mode when one
327 : /// instance serves both seams, `None` when history is off (`NoTemporal`);
328 : /// a second backend overwrites it after construction.
329 : pub temporal_name: Option<String>,
330 : pub loader: Arc<Loader>,
331 : pub started: Instant,
332 : /// Startup timestamp (createdAt of the built-in core @context entry).
333 : pub started_at: String,
334 : pub host_alias: String,
335 : /// Default page size for queries without an explicit limit.
336 : pub default_limit: usize,
337 : /// Hard ceiling on limit (TooManyResults guard).
338 : pub max_limit: usize,
339 : /// 6.3.10: the most instances of one Attribute a temporal response
340 : /// serves ([`crate::bounds::temporal_instance_limit`]); the ETSI suite's
341 : /// brokers and its unit cases run with 9.
342 : pub temporal_instance_limit: usize,
343 : /// The instances of one Attribute served when no lastN/firstN is asked
344 : /// ([`crate::bounds::temporal_default_instances`]), at most the ceiling.
345 : pub temporal_default_instances: usize,
346 : /// The most bytes of one in-process answer a handler buffers
347 : /// ([`crate::bounds::MAX_ANSWER_BYTES`]).
348 : pub max_answer_bytes: usize,
349 : /// One shared outbound client, timeouts set at construction.
350 : pub http: antares_jsonld::HttpClient,
351 : /// Federation-forwarding client — longer deadline: the ETSI mock replies
352 : /// to unstubbed forwards only when the robot side wakes (up to ~5 s).
353 : pub fed_http: antares_jsonld::HttpClient,
354 : /// Notification bindings by `endpoint.uri` scheme (6.3.8, 7.2). The
355 : /// only way a delivery transport is chosen; `with_sink` adds one.
356 : pub sinks: Arc<antares_notifier::SinkRegistry>,
357 : /// HTTP surfaces mounted outside the NGSI-LD API root, each under its
358 : /// own reserved prefix; `with_surface` adds one. The admin surface is
359 : /// here by default.
360 : pub surfaces: Arc<Vec<Box<dyn crate::ApiSurface>>>,
361 : /// The transport every federation forward asks before the socket:
362 : /// shared by every clone, so a state made before the
363 : /// transport was registered (the one an `Ops` holds) forwards through it
364 : /// too. Empty in vanilla: every forward is HTTP.
365 : pub(crate) forward_transport: Arc<std::sync::OnceLock<Arc<dyn crate::ForwardTransport>>>,
366 : /// Bounds-wall rejection counters (exported by /q/health).
367 : pub limits: Arc<crate::bounds::LimitStats>,
368 : /// `GET /q/tenants/{tenant}` sizes by tenant with the instant they were
369 : /// measured: a size is a scan of the tenant, so it is reused for
370 : /// [`crate::TENANT_BYTES_TTL`]. One entry per existing tenant
371 : /// asked about; a purge drops its entry.
372 : pub tenant_bytes: Arc<std::sync::Mutex<std::collections::HashMap<String, crate::BytesEntry>>>,
373 : /// The 5.5.10 wall's memory of Tenants that exist.
374 : pub known_tenants: Arc<KnownTenants>,
375 : /// Allocator stats provider (set by the broker; None in tests/wasm).
376 : pub mem_stats: Option<Arc<dyn Fn() -> serde_json::Value + Send + Sync>>,
377 : /// Bus state provider for /q/health (`bus: {mode, connected,
378 : /// reconnects}`) — installed by the nats wiring only, so the member is
379 : /// absent for bus=local.
380 : pub bus_stats: Option<Arc<dyn Fn() -> serde_json::Value + Send + Sync>>,
381 : /// One egress policy for notifications and federation forwards
382 : /// (scheme allowlist, private-range deny, per-destination breakers).
383 : pub egress: Arc<crate::egress::Egress>,
384 : /// The policy engine every operation is asked about (ADR-0020), or
385 : /// `None` when a deployment attached none with
386 : /// [`AppState::with_policy`] — the default, where the broker behaves
387 : /// exactly as it did before the seam existed. `None` is not an instance
388 : /// of `AllowAll`: the gate takes its answer without building a subject,
389 : /// boxing a future or arming a timer, so a broker with no engine pays
390 : /// nothing for the seam.
391 : pub policy: Option<Arc<dyn crate::policy::PolicyEngine>>,
392 : /// The `/q` admin credential ([`AppState::with_admin_auth`]); `None`
393 : /// serves `/q` without one, as the broker did.
394 : pub admin_auth: Option<Arc<crate::admin_auth::AdminAuth>>,
395 : /// Set the moment SIGTERM arrives, BEFORE the listener stops
396 : /// accepting — `/q/health` then answers 503 DRAINING so the load balancer
397 : /// takes this instance out while its socket still works.
398 : pub draining: Arc<std::sync::atomic::AtomicBool>,
399 : /// Change batches the matcher queue accepted and has not finished
400 : /// delivering. A shutdown drain waits for zero before the pool closes.
401 : pub pending_changes: Arc<std::sync::atomic::AtomicUsize>,
402 : /// Called after every Subscription and Context Source Registration
403 : /// Subscription CUD, so the mirror the matcher reads follows the store.
404 : /// On the bus the wiring pushes the change into the KV mirror bucket;
405 : /// in local mode it applies to this process's own mirror.
406 : #[allow(clippy::type_complexity)] // a boxed callback, named where it is installed
407 : pub sub_sync: Option<
408 : Arc<
409 : dyn Fn(&antares_model::TenantId, antares_store::Kind, &str, Option<&serde_json::Value>)
410 : + Send
411 : + Sync,
412 : >,
413 : >,
414 : /// bus=nats: the KV-watched compiled-subscription mirror the matcher
415 : /// reads, so the hot path never touches Postgres. `None` in local
416 : /// mode (the matcher reads the store directly).
417 : pub sub_mirror: Option<Arc<crate::mirror::SubMirror>>,
418 : /// bus=local: takes the entity changes one request buffered (the history
419 : /// layer hands them over after the handler) so the matcher sees a batch
420 : /// request as one unit. `None` until the local pipeline is wired.
421 : pub change_flush: Option<Arc<dyn Fn(Vec<crate::mirror::Change>) + Send + Sync>>,
422 : /// Called after every Registration CUD: bus=nats publishes the delta on
423 : /// `ANTARES_REGISTRY`, bus=local applies it to `reg_mirror` in process.
424 : /// `None` while no mirror is wired.
425 : #[allow(clippy::type_complexity)] // a boxed callback, named where it is installed
426 : pub reg_sync: Option<
427 : Arc<dyn Fn(&antares_model::TenantId, &str, Option<&serde_json::Value>) + Send + Sync>,
428 : >,
429 : /// The ONE per-process compiled registration mirror, delta-fed from
430 : /// `ANTARES_REGISTRY` (bus=nats) or by `reg_sync` (bus=local); expiry
431 : /// stays filtered at the single yield point (`federation::matching_regs`).
432 : /// `None` until wired or when its hydrate failed: the store is read then.
433 : pub reg_mirror: Option<Arc<crate::mirror::DocMirror>>,
434 : /// 5.2.34 (bus=nats): shares a cooldown stamp with the other api pods —
435 : /// a per-process stamp re-dials a failed source from every pod behind
436 : /// the LB. Seconds-scale state: broadcast on the
437 : /// registry stream, deliberately not persisted. `None` in local mode.
438 : #[allow(clippy::type_complexity)] // a boxed callback, named where it is installed
439 : pub reg_fail_sync: Option<Arc<dyn Fn(&str, bool) + Send + Sync>>,
440 : /// Renders the Prometheus text format for /q/metrics. Installed by
441 : /// the broker (the only crate that knows an exporter exists);
442 : /// `None` = 404, the facade calls elsewhere stay no-ops.
443 : pub metrics_render: Option<Arc<dyn Fn() -> String + Send + Sync>>,
444 : /// 5.8.1.4 consumer half: a notification addressed to the internal
445 : /// `urn:antares:distsub:` endpoint never leaves the broker, it re-enters
446 : /// it. The delivery path hands it here instead of naming the module that
447 : /// owns distributed subscriptions; `wire` installs the handler. `None`
448 : /// drops such a notification, which is what a broker that never created
449 : /// an internal subscription should do with one.
450 : pub csource_notification: Option<CsourceNotification>,
451 : /// True only under bus=nats (set by the broker's wiring): multiple
452 : /// processes share the store, so interval-subscription firings must be
453 : /// claimed single-winner. bus=local keeps the direct path — a
454 : /// claim there would disturb the 046_12 bookkeeping ordering for
455 : /// nothing.
456 : pub nats: bool,
457 : /// History gate 2 (ANTARES_TEMPORAL_RECORD): which changed attribute
458 : /// instances the write path records into history. Default `All`, which
459 : /// the ETSI temporal suites assume.
460 : pub temporal_record: TemporalRecord,
461 : /// Behaviour profile keys ([`Profile`]); the broker sets it from the env.
462 : pub profile: Profile,
463 : /// The mesh identity of the gateway whose provenance headers the broker
464 : /// believes, set by an embedding binary; `None` believes nobody.
465 : pub provenance_from: Option<String>,
466 : /// A request to the NGSI-LD API that carries no subject header is served
467 : /// only with the admin bearer token
468 : /// ([`crate::admin_auth::subject_or_admin`]). `false` (the default)
469 : /// serves it as before; a deployment sets it when it names subject
470 : /// headers.
471 : pub subject_required: bool,
472 : /// History gate 3: the (Tenant, expanded type) pairs whose
473 : /// retention rule says `history: "off"`. The views reload replaces the
474 : /// set; an Entity with any of these types records no attribute instance.
475 : pub history_off: HistoryOff,
476 : /// The base URL remote Context Sources reach THIS broker at — used as
477 : /// the notification endpoint of forwarded subscription copies
478 : /// (5.8.1.4). ANTARES_PUBLIC_URL, defaulting to
479 : /// http://{host_alias}:{ANTARES_HTTP_PORT} (portless when 80/unset).
480 : pub public_url: String,
481 : /// 5.5.15 resource-pressure signal: max snapshots per tenant — above it
482 : /// the lowest-snapshotPriority snapshots are evicted. Snapshot documents
483 : /// themselves live in the store (Kind::Snapshot) so persistent modes
484 : /// survive restarts.
485 : pub snapshot_cap: usize,
486 : /// How a notification is delivered: attempts, backoff, age ceiling.
487 : /// Default = one attempt (5.8.6 as written).
488 : pub delivery: antares_notifier::DeliveryPolicy,
489 : }
490 :
491 : impl std::ops::Deref for AppState {
492 : type Target = Shared;
493 :
494 2701409 : fn deref(&self) -> &Shared {
495 2701409 : &self.shared
496 2701409 : }
497 : }
498 :
499 : impl std::ops::DerefMut for AppState {
500 9440 : fn deref_mut(&mut self) -> &mut Shared {
501 : // the memoised in-process router captured the settings about to
502 : // change: a later call must build one over the new ones
503 9440 : self.inbound = Arc::new(std::sync::OnceLock::new());
504 9440 : Arc::make_mut(&mut self.shared)
505 9440 : }
506 : }
507 :
508 : impl AppState {
509 : /// This state's broker, told apart from the other brokers of the
510 : /// process: its narrowed copies share it, and no live broker shares it
511 : /// with another.
512 9385 : pub(crate) fn broker(&self) -> usize {
513 9385 : Arc::as_ptr(&self.shared).addr()
514 9385 : }
515 :
516 : /// Whether `other` is this state: the same broker.
517 : #[must_use]
518 177 : pub fn same(&self, other: &Self) -> bool {
519 177 : self.broker() == other.broker()
520 177 : }
521 :
522 : /// The built-in store: in-memory, or — under the reserved harness
523 : /// variable `ANTARES_TEST_STORE=file` — a fresh on-disk redb store per
524 : /// state, so the same test binary proves the durable backend without a
525 : /// second copy of every test. The broker's own boot path never calls
526 : /// this; it composes from `ANTARES_STORE` in `with_drivers`.
527 : #[cfg(feature = "test-kit")]
528 4638 : pub fn new(host_alias: String) -> Self {
529 : #[cfg(not(target_arch = "wasm32"))]
530 4638 : if std::env::var("ANTARES_TEST_STORE").as_deref() == Ok("file") {
531 0 : let dir = std::env::temp_dir()
532 0 : .join("antares-test-store")
533 0 : .join(uuid::Uuid::new_v4().simple().to_string());
534 : // reachable only through the harness variable read above; a
535 : // harness that cannot open its own store has nothing to run
536 : #[allow(clippy::expect_used)]
537 0 : let store = Store::open_file(&dir).expect("open the redb test store");
538 0 : return Self::with_store(host_alias, Arc::new(AnyStore::Mem(store)), "file");
539 4638 : }
540 4638 : Self::with_store(
541 4638 : host_alias,
542 4638 : Arc::new(AnyStore::Mem(Store::default())),
543 4638 : "memory",
544 : )
545 4638 : }
546 :
547 : /// Convenience over the built-in backends: one `AnyStore` serves as
548 : /// both drivers.
549 : #[cfg(feature = "test-kit")]
550 4732 : pub fn with_store(host_alias: String, store: Arc<AnyStore>, store_name: &str) -> Self {
551 4732 : let temporal: Arc<dyn TemporalDriver> = store.clone();
552 4732 : Self::with_drivers(host_alias, store, temporal, store_name)
553 4732 : }
554 :
555 4814 : pub fn with_drivers(
556 4814 : host_alias: String,
557 4814 : store: Arc<dyn CurrentStateDriver>,
558 4814 : temporal: Arc<dyn TemporalDriver>,
559 4814 : store_name: &str,
560 4814 : ) -> Self {
561 : // One policy value, read once, shared by every outbound path —
562 : // the gate (scheme/breakers) and the clients (DNS pinning, redirect
563 : // cap) can never disagree about what is allowed.
564 4814 : let egress_policy = antares_jsonld::EgressPolicy::from_env();
565 : // Cached-@context rows are the ONE source of truth for 5.13
566 : // existence — wire the write-through HERE so every composition
567 : // (native binary, wasm, tests) gets it; a composition that forgets
568 : // the writer is the "expiry checked in some paths" disease with
569 : // @contexts instead of expiry.
570 4814 : let loader = Arc::new(Loader::new());
571 : {
572 4814 : let store = store.clone();
573 4814 : loader.set_cache_writer(Arc::new(move |tenant, url, ctx_value| {
574 320 : let store = store.clone();
575 320 : Box::pin(async move {
576 320 : if hosted_row_id(&*store, tenant, url).await.is_some() {
577 0 : return; // broker-local (Hosted/Implicit) URLs are not Cached entries
578 320 : }
579 320 : let id = uuid::Uuid::new_v5(&uuid::Uuid::NAMESPACE_URL, url.as_bytes());
580 : // A refetch (staleness, delete+reload) must keep the row's
581 : // identity and hit counters — only the body is new — and
582 : // it swaps the body under the row lock, so a bump racing
583 : // it keeps its count. Only a row that is not there is
584 : // written whole.
585 : // The row this writes is `Cached` and belongs to no Tenant
586 : // (ADR-0021), but the call still acts for the Tenant whose
587 : // resolution triggered the fetch: that is what lets the guard
588 : // above see this Tenant's own Hosted rows.
589 320 : let id = id.to_string();
590 320 : let body = serde_json::json!({"@context": ctx_value});
591 320 : let refetched = store
592 320 : .context_mutate(
593 320 : tenant,
594 320 : &id,
595 320 : Box::new(|row| {
596 0 : row.as_object_mut()
597 0 : .ok_or(())?
598 0 : .insert("body".into(), body.clone());
599 0 : Ok(())
600 0 : }),
601 : )
602 320 : .await;
603 320 : if matches!(refetched, Ok(Some(Ok(())))) {
604 0 : return;
605 320 : }
606 320 : let doc = serde_json::json!({
607 320 : "url": url,
608 320 : "localId": id,
609 320 : "kind": "Cached",
610 320 : "createdAt": now_iso(),
611 320 : "numberOfHits": 0,
612 320 : "lastUsage": null,
613 320 : "body": body,
614 : });
615 320 : if let Err(e) = store.context_put(tenant, &id, doc).await {
616 : // a client-named @context URL may carry userinfo
617 0 : tracing::warn!(
618 : "@context write-through failed for {}: {e}",
619 0 : antares_notifier::redact_userinfo(url)
620 : );
621 320 : }
622 320 : })
623 320 : }));
624 : }
625 : // 5.13.3.5: hit counters live in the SHARED row, not per instance
626 : // — behind a load balancer per-instance counters split-brain.
627 : // A bump that finds the row gone reports
628 : // a cross-instance delete; the loader then drops its warm copies so
629 : // the delete is honoured everywhere (5.13.5.4). Pinned core contexts
630 : // have no row and are never evicted.
631 : {
632 4814 : let store = store.clone();
633 32136 : loader.set_usage_bump(Arc::new(move |tenant, url| {
634 32136 : let store = store.clone();
635 32136 : Box::pin(async move {
636 32136 : if Loader::is_pinned_core(url) {
637 11428 : return true;
638 20708 : }
639 20708 : let id = match hosted_row_id(&*store, tenant, url).await {
640 122 : Some(id) => id,
641 20586 : None => uuid::Uuid::new_v5(&uuid::Uuid::NAMESPACE_URL, url.as_bytes())
642 20586 : .to_string(),
643 : };
644 : // under the row lock: a bump racing a delete leaves the row
645 : // deleted, and two racing bumps both count
646 20708 : let bumped = store
647 20708 : .context_mutate(
648 20708 : tenant,
649 20708 : &id,
650 20708 : Box::new(|row| {
651 20704 : let o = row.as_object_mut().ok_or(())?;
652 20704 : let hits = o
653 20704 : .get("numberOfHits")
654 20704 : .and_then(serde_json::Value::as_u64)
655 20704 : .unwrap_or(0)
656 20704 : .saturating_add(1);
657 20704 : o.insert("numberOfHits".into(), serde_json::json!(hits));
658 20704 : o.insert("lastUsage".into(), serde_json::json!(now_iso()));
659 20704 : Ok(())
660 20704 : }),
661 : )
662 20708 : .await;
663 20704 : match bumped {
664 20704 : Ok(Some(Ok(()))) => true,
665 : // a row that cannot carry the counters is left
666 : // uncounted, and still exists: it must not be evicted
667 0 : Ok(Some(Err(()))) => true,
668 4 : Ok(None) => false,
669 : // a store hiccup must never evict a healthy cache entry
670 0 : Err(e) => {
671 0 : tracing::warn!(
672 : "@context hit bump failed for {}: {e}",
673 0 : antares_notifier::redact_userinfo(url)
674 : );
675 0 : true
676 : }
677 : }
678 32136 : })
679 32136 : }));
680 : }
681 : // 5.13.1: an @context this broker HOSTS is served from its row, not
682 : // fetched. Its warm copy in the loader is only a cache — a restart
683 : // reloads Cached rows alone, and the bounded cache may evict it at
684 : // any moment — and the URL under it is minted from the request Host
685 : // header, so a miss would otherwise send the broker to whatever
686 : // address a client put there for the term mappings of its own
687 : // Tenant's payloads. The row carries the owner, so 5.5.10 still
688 : // decides who may resolve it.
689 : {
690 4814 : let store = store.clone();
691 4814 : loader.set_local_lookup(Arc::new(move |tenant, url| {
692 404 : let store = store.clone();
693 404 : Box::pin(async move {
694 : // A miss for a named Tenant still asks the DEFAULT one: a
695 : // fetch of this URL would reach the serve endpoint with no
696 : // NGSILD-Tenant, get the default Tenant's row and cache it
697 : // for everyone. Found here, it keeps its owner.
698 404 : let default = antares_model::TenantId::default();
699 404 : let (_, row) = match hosted_row(&*store, tenant, url).await {
700 20 : Some(found) => found,
701 384 : None if tenant.is_some_and(|t| *t != default) => {
702 14 : hosted_row(&*store, Some(&default), url).await?
703 : }
704 370 : None => return None,
705 : };
706 : // Ownership is one rule for the whole broker (ADR-0021): a
707 : // Cached row is a copy of a public document and belongs to no
708 : // Tenant, everything else belongs to its `owner` — with the
709 : // DEFAULT Tenant for a row written before that member
710 : // existed, which is the Tenant that lists, serves and deletes
711 : // it.
712 28 : let owner = antares_store::context_row_owner(&row)
713 28 : .and_then(|o| antares_model::TenantId::new_internal(o).ok());
714 28 : Some((
715 28 : owner,
716 28 : row.pointer("/body/@context").cloned().unwrap_or_default(),
717 28 : ))
718 404 : })
719 404 : }));
720 : }
721 : // 5.8.1.4: this URL is handed to peer brokers as the notification
722 : // endpoint for distributed subscriptions — the default must carry
723 : // the HTTP port or peers dial port 80 (ETSI-matrix ADV_02 shape).
724 4814 : let public_url =
725 4814 : std::env::var("ANTARES_PUBLIC_URL").unwrap_or_else(|_| {
726 4672 : match std::env::var("ANTARES_HTTP_PORT") {
727 12 : Ok(p) if p != "80" => format!("http://{host_alias}:{p}"),
728 4662 : _ => format!("http://{host_alias}"),
729 : }
730 4672 : });
731 4814 : let temporal_name = temporal.supported().then(|| store_name.to_owned());
732 : // 6.3.8 is the mandatory binding; clause 7's MQTT one is optional and
733 : // registers only when compiled in. Both go through the registry, so
734 : // notification delivery never names a transport.
735 4814 : let http = outbound_client(
736 4814 : egress_policy,
737 4814 : std::time::Duration::from_secs(slow_factor().saturating_mul(5)),
738 : );
739 4814 : let mut sinks = antares_notifier::SinkRegistry::default();
740 4814 : sinks.register(Box::new(antares_notifier::HttpSink::new(http.clone())));
741 : #[cfg(feature = "mqtt")]
742 4814 : sinks.register(Box::new(antares_notifier::mqtt::MqttSink::default()));
743 4814 : Self {
744 4814 : inbound: Arc::new(std::sync::OnceLock::new()),
745 4814 : shared: Arc::new(Shared {
746 4814 : store,
747 4814 : temporal,
748 4814 : store_name: store_name.to_owned(),
749 4814 : temporal_name,
750 4814 : loader,
751 4814 : started: Instant::now(),
752 4814 : started_at: now_iso(),
753 4814 : host_alias,
754 4814 : default_limit: *crate::bounds::DEFAULT_LIMIT,
755 4814 : max_limit: *crate::bounds::MAX_LIMIT,
756 4814 : temporal_instance_limit: crate::bounds::temporal_instance_limit(),
757 4814 : temporal_default_instances: crate::bounds::temporal_default_instances(),
758 4814 : max_answer_bytes: crate::bounds::MAX_ANSWER_BYTES,
759 4814 : http: http.clone(),
760 4814 : fed_http: outbound_client(
761 4814 : egress_policy,
762 4814 : std::time::Duration::from_secs(slow_factor().saturating_mul(8)),
763 4814 : ),
764 4814 : sinks: Arc::new(sinks),
765 4814 : forward_transport: Arc::new(std::sync::OnceLock::new()),
766 4814 : surfaces: Arc::new(vec![Box::new(crate::Admin)]),
767 4814 : limits: Arc::new(crate::bounds::LimitStats::default()),
768 4814 : tenant_bytes: Arc::default(),
769 4814 : known_tenants: Arc::default(),
770 4814 : mem_stats: None,
771 4814 : bus_stats: None,
772 4814 : egress: Arc::new(crate::egress::Egress::new(egress_policy)),
773 4814 : policy: None,
774 4814 : admin_auth: None,
775 4814 : draining: Arc::new(std::sync::atomic::AtomicBool::new(false)),
776 4814 : pending_changes: Arc::default(),
777 4814 : sub_sync: None,
778 4814 : sub_mirror: None,
779 4814 : change_flush: None,
780 4814 : reg_sync: None,
781 4814 : reg_fail_sync: None,
782 4814 : reg_mirror: None,
783 4814 : metrics_render: None,
784 4814 : csource_notification: None,
785 4814 : nats: false,
786 4814 : temporal_record: TemporalRecord::All,
787 4814 : profile: Profile::default(),
788 4814 : provenance_from: None,
789 4814 : subject_required: false,
790 4814 : history_off: HistoryOff::default(),
791 4814 : public_url,
792 4814 : snapshot_cap: 1024,
793 4814 : delivery: antares_notifier::DeliveryPolicy::default(),
794 4814 : }),
795 4814 : }
796 4814 : }
797 :
798 : /// Attach a policy engine (ADR-0020). One engine per broker, chosen at
799 : /// startup: an engine that could be swapped per request would make what
800 : /// a caller may see a function of when they asked. Call before the
801 : /// state is shared.
802 : #[must_use]
803 966 : pub fn with_policy(mut self, engine: Arc<dyn crate::policy::PolicyEngine>) -> Self {
804 966 : self.policy = Some(engine);
805 966 : self
806 966 : }
807 :
808 : /// Puts every `/q` route except health and readiness (and metrics, when
809 : /// `auth` leaves them open) behind `auth`'s bearer secret. Set before
810 : /// the router is built: the admin surface reads it when it mounts.
811 : #[must_use]
812 18 : pub fn with_admin_auth(mut self, auth: crate::admin_auth::AdminAuth) -> Self {
813 18 : self.admin_auth = Some(Arc::new(auth));
814 18 : self
815 18 : }
816 :
817 : /// The engine `/q/health` reports. No engine reports the built-in name:
818 : /// the decision an operator reads is the one the broker takes.
819 86 : pub fn policy_name(&self) -> &str {
820 86 : self.policy
821 86 : .as_ref()
822 86 : .map_or(crate::policy::BUILT_IN_NAME, |e| e.name())
823 86 : }
824 :
825 : /// Register one more notification binding (6.3.8). A deployment adds a
826 : /// sink for a scheme this workspace does not ship; endpoints naming that
827 : /// scheme then validate and deliver through it. Call before the state is
828 : /// shared — a clone already handed out keeps the registry it was made
829 : /// with.
830 : ///
831 : /// Not behind `test-kit`, unlike the mirror accessors beside it: this is
832 : /// the sink seam itself (ADR-0016), the call `docs/src/extending.md`
833 : /// tells a deployment to make, and `examples/plugin-example` is the host
834 : /// that proves it from outside. A release library without it has no
835 : /// notification binding seam at all.
836 : #[must_use]
837 16 : pub fn with_sink(mut self, sink: Box<dyn antares_notifier::NotificationSink>) -> Self {
838 16 : match Arc::get_mut(&mut self.sinks) {
839 16 : Some(reg) => reg.register(sink),
840 0 : None => tracing::warn!("sink registry already shared; binding not registered"),
841 : }
842 16 : self
843 16 : }
844 :
845 : /// Register the transport every federation forward asks before the
846 : /// socket. A
847 : /// forward it answers `None` for goes over HTTP as before. Every clone
848 : /// of this state shares the one slot, before or after this call. The
849 : /// transport usually holds such a clone, so it lives as long as the
850 : /// state's last clone does: register it in a host that keeps its state
851 : /// for the life of the process, as the broker does.
852 : ///
853 : /// # Errors
854 : /// When a transport is already registered: a second one would leave
855 : /// the choice between them to registration order.
856 40 : pub fn with_forward_transport(
857 40 : self,
858 40 : transport: Arc<dyn crate::ForwardTransport>,
859 40 : ) -> Result<Self, String> {
860 40 : self.forward_transport
861 40 : .set(transport)
862 40 : .map_err(|_| "a forward transport is already registered".to_owned())?;
863 38 : Ok(self)
864 40 : }
865 :
866 : /// Mount one more HTTP surface (`/q`, `/x`, or below `/x`). A prefix
867 : /// outside those, or one that overlaps a surface already mounted, is an
868 : /// error the caller is expected to make fatal: a surface that could
869 : /// shadow a spec resource would make conformance a function of
870 : /// deployment configuration, and two surfaces on one prefix would leave
871 : /// the winner to route-matching order. Call before the state is shared.
872 76 : pub fn with_surface(mut self, s: Box<dyn crate::ApiSurface>) -> Result<Self, String> {
873 76 : crate::surface::check_prefix(s.prefix())?;
874 64 : if let Some(clash) = self
875 64 : .surfaces
876 64 : .iter()
877 64 : .find(|m| crate::surface::overlaps(m.prefix(), s.prefix()))
878 : {
879 8 : return Err(format!(
880 8 : "api surface {:?} claims {:?}, already served by {:?} at {:?}",
881 8 : s.name(),
882 8 : s.prefix(),
883 8 : clash.name(),
884 8 : clash.prefix()
885 8 : ));
886 56 : }
887 56 : match Arc::get_mut(&mut self.surfaces) {
888 56 : Some(v) => v.push(s),
889 0 : None => return Err("api surfaces already shared; register before serving".into()),
890 : }
891 56 : Ok(self)
892 76 : }
893 :
894 : /// Replace the mounted surfaces with a deployment's own selection —
895 : /// what a binary reads out of its configuration, rather than what the
896 : /// default mounting put there. Same prefix rules as `with_surface`, and
897 : /// a selection may leave admin out: `/q` is then not served at all.
898 18 : pub fn with_surfaces(mut self, list: Vec<Box<dyn crate::ApiSurface>>) -> Result<Self, String> {
899 18 : self.surfaces = Arc::new(Vec::new());
900 20 : for s in list {
901 20 : self = self.with_surface(s)?;
902 : }
903 16 : Ok(self)
904 18 : }
905 :
906 : /// Serve one request through this broker's own router, in process.
907 : ///
908 : /// This is the seam a façade for another standard (SensorThings, OGC
909 : /// API, WFS, OData) is built on: the façade is an [`crate::ApiSurface`]
910 : /// under `/x/<standard>` that translates its own request into an NGSI-LD
911 : /// one and calls this. There is no second data path — the inner request
912 : /// takes the same route as one off the socket, so negotiation, the
913 : /// bounds wall, tenancy, the policy seam, history and notifications all
914 : /// apply exactly once and exactly as they do for an NGSI-LD client.
915 : ///
916 : /// `caller` is the outer request's headers, and the ones that decide
917 : /// WHICH data an operation runs against are copied into the inner
918 : /// request when it does not set them itself:
919 : ///
920 : /// - `NGSILD-Tenant` (6.3.14) — a façade that forgot it would answer
921 : /// every caller out of the default tenant, so it is not left to the
922 : /// façade to remember;
923 : /// - `NGSILD-Snapshot` (6.3.22) — for the same reason: a façade called
924 : /// inside a snapshot request must not quietly serve live data;
925 : /// - `Link` (6.3.5) — the `@context` the caller supplied, so a term
926 : /// means the same thing on both sides of the translation;
927 : /// - every header `ANTARES_POLICY_SUBJECT_HEADERS` names, so the policy
928 : /// engine is asked about the caller rather than about the façade.
929 : ///
930 : /// All values of a copied header are carried, never just the first: a
931 : /// repeated `NGSILD-Tenant` is `BadRequestData` (6.3.14), and a façade
932 : /// must not be the place where a repeat is laundered into a single valid
933 : /// value.
934 : ///
935 : /// `&self` on purpose — an inner call runs while the outer handler is
936 : /// suspended on it, and the router clone is an `Arc` bump over the same
937 : /// state. The router itself is built once per state, on the first call:
938 : /// building one costs about 1.5 ms, which no façade should pay per
939 : /// request. After that first call the state counts as shared, so
940 : /// `with_surface` and the other builders refuse — which is the rule
941 : /// they already state.
942 100 : pub async fn call(
943 100 : &self,
944 100 : caller: &axum::http::HeaderMap,
945 100 : req: axum::http::Request<axum::body::Body>,
946 100 : ) -> axum::response::Response {
947 : use axum::response::IntoResponse as _;
948 : // A façade route that translates into a request its own surface
949 : // serves would call itself for as long as the stack lasts, and
950 : // every frame builds a router. The chain is counted per task, so a
951 : // façade over a façade is served and a loop is not.
952 100 : let depth = in_process_depth();
953 100 : if depth >= crate::bounds::MAX_IN_PROCESS_CALL_DEPTH {
954 4 : return crate::negotiate::ApiError::from(antares_model::NgsiError::InternalError(
955 4 : format!(
956 4 : "in-process call depth exceeded {}",
957 4 : crate::bounds::MAX_IN_PROCESS_CALL_DEPTH
958 4 : ),
959 4 : ))
960 4 : .into_response();
961 96 : }
962 96 : let router = self.inbound.get_or_init(|| {
963 : // Every layer of a built router captured an `AppState` clone, so
964 : // the memo owns states. The clone it is built from must NOT be
965 : // able to reach THIS cell: a state that could would hold itself
966 : // alive for the life of the process and pin its store handle
967 : // with it — a file store would keep its lock after the host
968 : // dropped everything. A fresh cell means the memo of a nested
969 : // façade call hangs off this one and dies with it.
970 60 : let mut inner = self.clone();
971 60 : inner.inbound = Arc::new(std::sync::OnceLock::new());
972 60 : crate::router(inner)
973 60 : });
974 96 : let (mut parts, body) = req.into_parts();
975 288 : for name in &PROPAGATED {
976 288 : if parts.headers.contains_key(name) {
977 0 : continue;
978 288 : }
979 288 : for v in caller.get_all(name) {
980 36 : parts.headers.append(name.clone(), v.clone());
981 36 : }
982 : }
983 96 : for name in crate::policy::SUBJECT_HEADERS.iter() {
984 88 : let Ok(name) = axum::http::HeaderName::from_bytes(name.as_bytes()) else {
985 0 : continue;
986 : };
987 88 : if parts.headers.contains_key(&name) {
988 0 : continue;
989 88 : }
990 88 : for v in caller.get_all(&name) {
991 8 : parts.headers.append(name.clone(), v.clone());
992 8 : }
993 : }
994 96 : let req = axum::http::Request::from_parts(parts, body);
995 96 : match INPROCESS_DEPTH
996 96 : .scope(
997 96 : depth.saturating_add(1),
998 96 : tower::ServiceExt::oneshot(router.clone(), req),
999 : )
1000 96 : .await
1001 : {
1002 96 : Ok(r) => r,
1003 : // `Router`'s error is `Infallible`; the arm stays honest rather
1004 : // than unwrapping (the workspace denies unwrap outside tests).
1005 0 : Err(_) => crate::negotiate::ApiError::from(antares_model::NgsiError::InternalError(
1006 0 : "the in-process router failed".into(),
1007 0 : ))
1008 0 : .into_response(),
1009 : }
1010 100 : }
1011 :
1012 : /// [`AppState::call`] on a task of its own, at the caller's in-process
1013 : /// depth. A chain of calls each awaiting the next polls every frame
1014 : /// inside the one before it, and a few federation handlers deep that
1015 : /// is more stack than a worker thread has; on its own task each frame
1016 : /// is polled from the scheduler, as a request over HTTP is. Dropping
1017 : /// the returned future (a forward's timeout) aborts the task.
1018 : ///
1019 : /// The new task inherits no view: the call runs through `view` when one
1020 : /// is given ([`crate::policy::through_view`]), so the engine decides a
1021 : /// forward under a grant's view as a request through that view,
1022 : /// and never by the view of the request that triggered the forward.
1023 : #[cfg(not(target_arch = "wasm32"))]
1024 0 : pub async fn call_on_task(
1025 0 : &self,
1026 0 : view: Option<&str>,
1027 0 : caller: axum::http::HeaderMap,
1028 0 : req: axum::http::Request<axum::body::Body>,
1029 0 : ) -> axum::response::Response {
1030 : use axum::response::IntoResponse as _;
1031 : /// Aborts the task when the forward waiting on it is dropped.
1032 : struct AbortOnDrop(tokio::task::AbortHandle);
1033 : impl Drop for AbortOnDrop {
1034 0 : fn drop(&mut self) {
1035 0 : self.0.abort();
1036 0 : }
1037 : }
1038 0 : let st = self.clone();
1039 0 : let view = view.map(str::to_owned);
1040 0 : let task = tokio::spawn(INPROCESS_DEPTH.scope(in_process_depth(), async move {
1041 0 : match view {
1042 0 : Some(id) => crate::policy::through_view(&id, st.call(&caller, req)).await,
1043 0 : None => st.call(&caller, req).await,
1044 : }
1045 0 : }));
1046 0 : let _abort = AbortOnDrop(task.abort_handle());
1047 0 : task.await.unwrap_or_else(|_| {
1048 0 : crate::negotiate::ApiError::from(antares_model::NgsiError::InternalError(
1049 0 : "the in-process call failed".into(),
1050 0 : ))
1051 0 : .into_response()
1052 0 : })
1053 0 : }
1054 :
1055 : /// Temporal auto-recording happens synchronously in the write path in
1056 : /// EVERY bus mode (read-your-writes) — but only when the loaded
1057 : /// temporal driver actually records anything.
1058 152318 : pub fn record_locally(&self) -> bool {
1059 152318 : self.temporal.supported()
1060 152318 : }
1061 :
1062 : /// Fire the subscription-sync hook for one written row. `kind` decides
1063 : /// what the mirror does with it: a Subscription is indexed as a document,
1064 : /// a Context Source Registration Subscription only wakes the interval
1065 : /// sweep (5.11.7) — it is matched against registrations, not entities.
1066 2066 : pub(crate) fn sub_changed(
1067 2066 : &self,
1068 2066 : tenant: &antares_model::TenantId,
1069 2066 : kind: antares_store::Kind,
1070 2066 : id: &str,
1071 2066 : doc: Option<&serde_json::Value>,
1072 2066 : ) {
1073 2066 : if let Some(h) = &self.sub_sync {
1074 1956 : h(tenant, kind, id, doc);
1075 1956 : }
1076 2066 : }
1077 :
1078 : /// 5.2.34: stamp the per-registration cooldown locally AND on the other
1079 : /// api pods (no-op half in local mode). Keyed per tenant — the id alone
1080 : /// is client-chosen per tenant (5.5.10) and must not gate a neighbour.
1081 2 : pub(crate) fn reg_cooldown_stamp(
1082 2 : &self,
1083 2 : tenant: &antares_model::TenantId,
1084 2 : reg_id: &str,
1085 2 : ok: bool,
1086 2 : ) {
1087 2 : let key = crate::egress::reg_key(tenant.as_str(), reg_id);
1088 2 : self.egress.reg_record(&key, ok);
1089 2 : if let Some(h) = &self.reg_fail_sync {
1090 0 : // the broadcast carries the COMPOSED key; receiving pods stamp it
1091 0 : // verbatim, so their lookups agree without re-deriving anything
1092 0 : h(&key, ok);
1093 2 : }
1094 2 : }
1095 :
1096 : /// Fire the registration-delta hook (no-op in local mode).
1097 4270 : pub(crate) fn reg_changed(
1098 4270 : &self,
1099 4270 : tenant: &antares_model::TenantId,
1100 4270 : id: &str,
1101 4270 : doc: Option<&serde_json::Value>,
1102 4270 : ) {
1103 4270 : if let Some(h) = &self.reg_sync {
1104 2206 : h(tenant, id, doc);
1105 2742 : }
1106 4270 : }
1107 : }
1108 :
1109 : /// The ONE outbound-client construction for this crate (timeouts at
1110 : /// construction). Title-case headers are an http1 knob and timeouts are
1111 : /// client-level knobs — both native-only; the browser's fetch supplies its
1112 : /// own transport on wasm32.
1113 : /// Outbound timeouts stretch 10× when the test binary runs under a
1114 : /// sanitizer (ANTARES_TEST_SANITIZER, set by the strict workflow):
1115 : /// ThreadSanitizer slows every thread, and 371 tests sharing one runner
1116 : /// pushed loopback forwards past the 2 s connect / 5 s total limits —
1117 : /// each run failing a different test with a 504. Production is unchanged.
1118 21006 : pub fn slow_factor() -> u64 {
1119 21006 : antares_jsonld::slow_factor()
1120 21006 : }
1121 :
1122 9628 : fn outbound_client(
1123 9628 : policy: antares_jsonld::EgressPolicy,
1124 9628 : total: std::time::Duration,
1125 9628 : ) -> antares_jsonld::HttpClient {
1126 9628 : let b = antares_jsonld::with_timeouts(
1127 9628 : antares_jsonld::client_builder(policy),
1128 9628 : std::time::Duration::from_secs(slow_factor().saturating_mul(2)),
1129 9628 : total,
1130 : );
1131 : // the suite's notification receiver asserts header names
1132 : // case-sensitively ("Link", "X-Additional-Key")
1133 : #[cfg(not(target_arch = "wasm32"))]
1134 9628 : let b = b.http1_title_case_headers();
1135 : // reqwest fails to build only when the process cannot initialise its TLS
1136 : // backend: no outbound request of any kind can be made after that
1137 : #[allow(clippy::expect_used)]
1138 9628 : let c = b.build().expect("http client");
1139 9628 : antares_jsonld::wrap_client(c)
1140 9628 : }
1141 :
1142 : /// 5.13.1: an @context this broker HOSTS (Hosted or ImplicitlyCreated) is
1143 : /// identified by its stored row, never by the URL's shape — a peer broker or
1144 : /// an attacker can serve a document under the same resource path, and such a
1145 : /// URL is external to us (a Cached entry). The row records the URL it was
1146 : /// minted under, so that is what decides: a URL whose trailing segment names
1147 : /// a stored row but whose origin is somebody else's names a document this
1148 : /// broker does not host, and the stored row — another Tenant's, as often as
1149 : /// not — must not be read, counted or rewritten for it. Returns the local
1150 : /// row id when the URL is the one the row was minted under.
1151 21446 : async fn hosted_row(
1152 21446 : store: &dyn CurrentStateDriver,
1153 21446 : tenant: Option<&antares_model::TenantId>,
1154 21446 : url: &str,
1155 21446 : ) -> Option<(String, serde_json::Value)> {
1156 21446 : let (_, seg) = url.rsplit_once("/ngsi-ld/v1/jsonldContexts/")?;
1157 190 : let seg = seg.split(['?', '#']).next().unwrap_or(seg);
1158 190 : if seg.is_empty() || seg.contains('/') {
1159 0 : return None;
1160 190 : }
1161 190 : let row = store
1162 190 : .context_get(tenant, seg)
1163 190 : .await
1164 190 : .ok()
1165 190 : .flatten()
1166 190 : .filter(|row| row["url"].as_str() == Some(url))?;
1167 150 : Some((seg.to_owned(), row))
1168 21446 : }
1169 :
1170 21028 : async fn hosted_row_id(
1171 21028 : store: &dyn CurrentStateDriver,
1172 21028 : tenant: Option<&antares_model::TenantId>,
1173 21028 : url: &str,
1174 21028 : ) -> Option<String> {
1175 21028 : hosted_row(store, tenant, url).await.map(|(id, _)| id)
1176 21028 : }
1177 :
1178 : /// Server-managed timestamp, ISO 8601 UTC with milliseconds.
1179 209845 : pub fn now_iso() -> String {
1180 209845 : chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
1181 209845 : }
1182 :
1183 : #[cfg(test)]
1184 : mod jsonld_context_locality_5_13 {
1185 : use super::*;
1186 :
1187 : /// Serve one @context document under `path` on a loopback port, counting
1188 : /// fetches (the negative half: a warm copy must NOT be refetched).
1189 24 : fn context_server(path: &str) -> (String, Arc<std::sync::atomic::AtomicUsize>) {
1190 24 : let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
1191 24 : let port = listener.local_addr().expect("addr").port();
1192 24 : let fetches = Arc::new(std::sync::atomic::AtomicUsize::new(0));
1193 24 : let n = fetches.clone();
1194 24 : std::thread::spawn(move || {
1195 24 : for stream in listener.incoming() {
1196 12 : let Ok(mut s) = stream else { continue };
1197 12 : n.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
1198 12 : let body = r#"{"@context":{"peerTemp":"http://example.org/peerTemp"}}"#;
1199 12 : let resp = format!(
1200 : "HTTP/1.1 200 OK\r\nContent-Type: application/ld+json\r\n\
1201 : Content-Length: {}\r\nConnection: close\r\n\r\n{body}",
1202 12 : body.len()
1203 : );
1204 : use std::io::{Read, Write};
1205 12 : let mut buf = [0u8; 2048];
1206 12 : let _ = s.read(&mut buf);
1207 12 : let _ = s.write_all(resp.as_bytes());
1208 : }
1209 24 : });
1210 24 : (format!("http://127.0.0.1:{port}{path}"), fetches)
1211 24 : }
1212 :
1213 16 : fn fetch_count(c: &Arc<std::sync::atomic::AtomicUsize>) -> usize {
1214 16 : c.load(std::sync::atomic::Ordering::SeqCst)
1215 16 : }
1216 :
1217 : /// 5.13.1: a Cached @context is one this broker fetched from elsewhere.
1218 : /// Which URLs this broker HOSTS is a property of the stored rows, not of
1219 : /// the URL path, so a peer's @context served under the same resource path
1220 : /// is persisted as Cached and never mistaken for a row deleted through
1221 : /// another instance (5.13.5.4).
1222 : #[tokio::test]
1223 4 : async fn a_peer_context_url_under_the_local_path_is_cached() {
1224 4 : let st = AppState::new("me".into());
1225 4 : let (url, fetches) = context_server("/ngsi-ld/v1/jsonldContexts/peer-ctx");
1226 4 : let user = serde_json::json!(url);
1227 4 : st.loader.resolve(&user).await.expect("resolve");
1228 4 : let id = uuid::Uuid::new_v5(&uuid::Uuid::NAMESPACE_URL, url.as_bytes()).to_string();
1229 4 : let row = st
1230 4 : .store
1231 4 : .context_get(None, &id)
1232 4 : .await
1233 4 : .expect("store")
1234 4 : .expect("the fetched @context is persisted as a Cached row");
1235 4 : assert_eq!(row["kind"], "Cached");
1236 4 : assert_eq!(row["url"], serde_json::json!(url));
1237 4 : st.loader.resolve(&user).await.expect("resolve again");
1238 4 : assert_eq!(
1239 4 : fetch_count(&fetches),
1240 : 1,
1241 : "a warm, still-existing @context must not be refetched"
1242 : );
1243 4 : assert_eq!(
1244 4 : st.store
1245 4 : .context_get(None, &id)
1246 4 : .await
1247 4 : .expect("store")
1248 4 : .expect("row")["numberOfHits"],
1249 4 : serde_json::json!(2),
1250 4 : "both counted uses land on the shared row (5.13.3.5)"
1251 4 : );
1252 4 : }
1253 :
1254 : /// the default Tenant's Hosted @context is not fetched for
1255 : /// another Tenant through the broker's own (tenant-less) serve URL.
1256 : #[tokio::test]
1257 4 : async fn a_default_tenant_context_is_not_fetched_for_another_tenant() {
1258 4 : let st = AppState::new("me".into());
1259 4 : let owner = antares_model::TenantId::default();
1260 4 : let (url, fetches) = context_server("/ngsi-ld/v1/jsonldContexts/local-d");
1261 4 : st.store
1262 4 : .context_put(
1263 4 : Some(&owner),
1264 4 : "local-d",
1265 4 : serde_json::json!({
1266 4 : "url": url, "localId": "local-d", "kind": "Hosted",
1267 4 : "owner": owner.as_str(), "createdAt": now_iso(),
1268 4 : "body": {"@context": {"peerTemp": "http://example.org/peerTemp"}},
1269 4 : }),
1270 4 : )
1271 4 : .await
1272 4 : .expect("seed hosted row");
1273 4 : let beta = antares_model::TenantId::new("beta").expect("tenant");
1274 4 : let user = serde_json::json!(url);
1275 4 : for _ in 0..2 {
1276 8 : let e = st.loader.resolve_for(&beta, &user).await;
1277 8 : assert!(
1278 8 : matches!(e, Err(antares_model::NgsiError::LdContextNotAvailable(_))),
1279 : "{e:?}"
1280 : );
1281 : }
1282 4 : assert_eq!(fetch_count(&fetches), 0, "never fetched over HTTP");
1283 4 : st.loader
1284 4 : .resolve_for(&owner, &user)
1285 4 : .await
1286 4 : .expect("owner resolves");
1287 4 : }
1288 :
1289 : /// a Cached @context serves every Tenant, so a named Tenant
1290 : /// may neither delete nor reload it; the default Tenant may.
1291 : #[tokio::test]
1292 4 : async fn only_the_default_tenant_deletes_a_cached_context() {
1293 4 : let st = AppState::new("me".into());
1294 4 : let (url, _) = context_server("/shared.jsonld");
1295 4 : let alpha = antares_model::TenantId::new("alpha").expect("tenant");
1296 4 : st.loader
1297 4 : .resolve_for(&alpha, &serde_json::json!(url))
1298 4 : .await
1299 4 : .expect("cached");
1300 4 : let path = format!(
1301 : "/ngsi-ld/v1/jsonldContexts/{}",
1302 4 : url.replace(':', "%3A").replace('/', "%2F")
1303 : );
1304 16 : let send = |method: &str, uri: String, tenant: Option<&str>| {
1305 16 : let mut b = axum::http::Request::builder().method(method).uri(uri);
1306 16 : if let Some(t) = tenant {
1307 8 : b = b.header("NGSILD-Tenant", t);
1308 8 : }
1309 16 : let req = b.body(axum::body::Body::empty()).expect("request");
1310 16 : tower::ServiceExt::oneshot(crate::router(st.clone()), req)
1311 16 : };
1312 8 : for uri in [path.clone(), format!("{path}?reload=true")] {
1313 8 : let refused = send("DELETE", uri, Some("beta")).await.expect("response");
1314 8 : assert_eq!(refused.status(), axum::http::StatusCode::FORBIDDEN);
1315 : }
1316 4 : let details = send("GET", format!("{path}?details=true"), None)
1317 4 : .await
1318 4 : .expect("response");
1319 4 : assert_eq!(
1320 4 : details.status(),
1321 : axum::http::StatusCode::OK,
1322 : "the row is kept"
1323 : );
1324 4 : let deleted = send("DELETE", path, None).await.expect("response");
1325 4 : assert_eq!(deleted.status(), axum::http::StatusCode::NO_CONTENT);
1326 4 : }
1327 :
1328 : /// a Tenant purge drops its Hosted @context from the loader
1329 : /// too, so the name recreated is not served the old term mappings.
1330 : #[tokio::test]
1331 4 : async fn a_purged_tenants_hosted_context_resolves_no_more() {
1332 4 : let st = AppState::new("me".into());
1333 4 : let t = antares_model::TenantId::new("purgectx").expect("tenant");
1334 4 : let (url, fetches) = context_server("/ngsi-ld/v1/jsonldContexts/local-p");
1335 4 : st.store
1336 4 : .context_put(
1337 4 : Some(&t),
1338 4 : "local-p",
1339 4 : serde_json::json!({
1340 4 : "url": url, "localId": "local-p", "kind": "Hosted",
1341 4 : "owner": t.as_str(), "createdAt": now_iso(),
1342 4 : "body": {"@context": {"peerTemp": "http://example.org/peerTemp"}},
1343 4 : }),
1344 4 : )
1345 4 : .await
1346 4 : .expect("seed hosted row");
1347 4 : let user = serde_json::json!(url);
1348 4 : st.loader
1349 4 : .resolve_for(&t, &user)
1350 4 : .await
1351 4 : .expect("owner resolves");
1352 8 : let req = |m: &str, uri: &str, body: &str| {
1353 8 : axum::http::Request::builder()
1354 8 : .method(m)
1355 8 : .uri(uri)
1356 8 : .header("NGSILD-Tenant", "purgectx")
1357 8 : .header("Content-Type", "application/json")
1358 8 : .header("Content-Length", body.len())
1359 8 : .body(axum::body::Body::from(body.to_owned()))
1360 8 : .expect("request")
1361 8 : };
1362 4 : let entity = r#"{"id": "urn:ngsi-ld:Room:p", "type": "Room"}"#;
1363 4 : let made = tower::ServiceExt::oneshot(
1364 4 : crate::router(st.clone()),
1365 4 : req("POST", "/ngsi-ld/v1/entities", entity),
1366 4 : )
1367 4 : .await
1368 4 : .expect("response");
1369 4 : assert_eq!(made.status(), axum::http::StatusCode::CREATED);
1370 4 : let purged = tower::ServiceExt::oneshot(
1371 4 : crate::router(st.clone()),
1372 4 : req("DELETE", "/q/tenants/purgectx", ""),
1373 4 : )
1374 4 : .await
1375 4 : .expect("response");
1376 4 : assert_eq!(purged.status(), axum::http::StatusCode::NO_CONTENT);
1377 : // the purge released the loader's entry for the row while it still
1378 : // listed it; run after the store purge it listed nothing
1379 4 : assert!(st.loader.usage_get(&url).await.is_none());
1380 4 : assert_eq!(fetch_count(&fetches), 0);
1381 4 : }
1382 :
1383 : /// 5.13.3.5 counts uses of an @context this broker hosts on its own
1384 : /// stored row — no second, Cached copy of the same document, and no
1385 : /// fetch of its URL: the row holds the document (5.13.1 "Hosted").
1386 : #[tokio::test]
1387 4 : async fn a_hosted_context_is_counted_on_its_own_row() {
1388 4 : let st = AppState::new("me".into());
1389 4 : let owner = antares_model::TenantId::default();
1390 4 : let (url, fetches) = context_server("/ngsi-ld/v1/jsonldContexts/local-1");
1391 : // no `owner` member: a row written before it existed belongs to the
1392 : // default Tenant, which is the one resolving below
1393 4 : st.store
1394 4 : .context_put(
1395 4 : Some(&owner),
1396 4 : "local-1",
1397 4 : serde_json::json!({
1398 4 : "url": url,
1399 4 : "localId": "local-1",
1400 4 : "kind": "Hosted",
1401 4 : "createdAt": now_iso(),
1402 4 : "body": {"@context": {"peerTemp": "http://example.org/peerTemp"}},
1403 4 : }),
1404 4 : )
1405 4 : .await
1406 4 : .expect("seed hosted row");
1407 4 : let user = serde_json::json!(url);
1408 4 : st.loader.resolve_for(&owner, &user).await.expect("resolve");
1409 4 : st.loader
1410 4 : .resolve_for(&owner, &user)
1411 4 : .await
1412 4 : .expect("resolve again");
1413 4 : let cached = uuid::Uuid::new_v5(&uuid::Uuid::NAMESPACE_URL, url.as_bytes()).to_string();
1414 4 : assert!(
1415 4 : st.store
1416 4 : .context_get(None, &cached)
1417 4 : .await
1418 4 : .expect("store")
1419 4 : .is_none(),
1420 : "a hosted @context must not be duplicated as a Cached row"
1421 : );
1422 4 : assert_eq!(
1423 4 : st.store
1424 4 : .context_get(Some(&owner), "local-1")
1425 4 : .await
1426 4 : .expect("store")
1427 4 : .expect("row")["numberOfHits"],
1428 4 : serde_json::json!(2)
1429 : );
1430 4 : assert_eq!(
1431 4 : fetch_count(&fetches),
1432 4 : 0,
1433 4 : "a hosted @context is read from its row, never fetched over the network"
1434 4 : );
1435 4 : }
1436 :
1437 : /// 5.13.1: what this broker HOSTS comes from its own store, not from the
1438 : /// network. The in-process copy is a cache, and a miss — a restart (only
1439 : /// `Cached` rows are preloaded) or an eviction from the bounded document
1440 : /// cache — must not turn into an outbound GET of a URL a client chose:
1441 : /// the URL is minted from the request's `Host` header when
1442 : /// `ANTARES_PUBLIC_URL` is unset, so a spoofed one would send the broker
1443 : /// to the spoofer for the term mappings that expand that Tenant's
1444 : /// payloads. 5.5.10 still holds through the store: the row's owner is
1445 : /// the only Tenant it resolves for.
1446 : #[tokio::test]
1447 4 : async fn a_hosted_context_resolves_from_the_store_after_its_copy_is_gone() {
1448 4 : let st = AppState::new("me".into());
1449 4 : let alpha = antares_model::TenantId::new("alpha").expect("tenant");
1450 4 : let beta = antares_model::TenantId::new("beta").expect("tenant");
1451 : // a dead port: anything resolved can only have come from the row
1452 4 : let url = "http://127.0.0.1:9/ngsi-ld/v1/jsonldContexts/hosted-1";
1453 4 : st.store
1454 4 : .context_put(
1455 4 : Some(&alpha),
1456 4 : "hosted-1",
1457 4 : serde_json::json!({
1458 4 : "url": url,
1459 4 : "localId": "hosted-1",
1460 4 : "kind": "Hosted",
1461 4 : "createdAt": now_iso(),
1462 4 : "owner": "alpha",
1463 4 : "body": {"@context": {"secret": "https://alpha.example/secret"}},
1464 4 : }),
1465 4 : )
1466 4 : .await
1467 4 : .expect("seed hosted row");
1468 4 : let user = serde_json::json!(url);
1469 4 : let ctx = st
1470 4 : .loader
1471 4 : .resolve_for(&alpha, &user)
1472 4 : .await
1473 4 : .expect("the owning Tenant resolves its stored @context");
1474 4 : assert_eq!(ctx.expand_key("secret"), "https://alpha.example/secret");
1475 4 : let err = st
1476 4 : .loader
1477 4 : .resolve_for(&beta, &user)
1478 4 : .await
1479 4 : .expect_err("another Tenant may not resolve it (5.5.10)");
1480 4 : assert!(
1481 4 : matches!(err, antares_model::NgsiError::LdContextNotAvailable(_)),
1482 4 : "got {err:?}"
1483 4 : );
1484 4 : }
1485 :
1486 : /// 5.13.1 + 5.5.10: a stored @context with no `owner` member is not
1487 : /// ownerless. `contexts.rs` `row_visible` reads a row written before that
1488 : /// member existed as the DEFAULT Tenant's — it is listed, served and
1489 : /// deleted through that Tenant alone — so resolving one must answer the
1490 : /// same way. Read as "belongs to no Tenant" it would expand every other
1491 : /// Tenant's payloads with the default Tenant's private term mappings.
1492 : #[tokio::test]
1493 4 : async fn a_stored_context_without_an_owner_belongs_to_the_default_tenant() {
1494 4 : let st = AppState::new("me".into());
1495 4 : let default = antares_model::TenantId::new("default").expect("tenant");
1496 4 : let other = antares_model::TenantId::new("beta").expect("tenant");
1497 4 : let url = "http://127.0.0.1:9/ngsi-ld/v1/jsonldContexts/legacy-1";
1498 4 : st.store
1499 4 : .context_put(
1500 4 : Some(&default),
1501 4 : "legacy-1",
1502 4 : serde_json::json!({
1503 4 : "url": url,
1504 4 : "localId": "legacy-1",
1505 4 : "kind": "Hosted",
1506 4 : "createdAt": now_iso(),
1507 4 : "body": {"@context": {"legacy": "https://legacy.example/term"}},
1508 4 : }),
1509 4 : )
1510 4 : .await
1511 4 : .expect("seed a row from before the owner member");
1512 4 : let user = serde_json::json!(url);
1513 4 : let ctx = st
1514 4 : .loader
1515 4 : .resolve_for(&default, &user)
1516 4 : .await
1517 4 : .expect("the Tenant the row belongs to resolves it");
1518 4 : assert_eq!(ctx.expand_key("legacy"), "https://legacy.example/term");
1519 4 : let err = st
1520 4 : .loader
1521 4 : .resolve_for(&other, &user)
1522 4 : .await
1523 4 : .expect_err("no other Tenant may resolve it (5.5.10)");
1524 4 : assert!(
1525 4 : matches!(err, antares_model::NgsiError::LdContextNotAvailable(_)),
1526 4 : "got {err:?}"
1527 4 : );
1528 4 : }
1529 :
1530 : /// 5.13.1 again, with the local id of a row that exists: what this broker
1531 : /// hosts is the row MINTED under that URL, so a document served from
1532 : /// somewhere else under the same resource path — with the local id of a
1533 : /// Hosted @context another Tenant added — is external. It becomes a
1534 : /// Cached row of its own, and the Tenant's row is neither read for its
1535 : /// mappings nor counted against.
1536 : #[tokio::test]
1537 4 : async fn a_peer_url_reusing_a_stored_local_id_leaves_that_row_alone() {
1538 4 : let st = AppState::new("me".into());
1539 4 : let (url, _) = context_server("/ngsi-ld/v1/jsonldContexts/alpha-1");
1540 4 : let row = serde_json::json!({
1541 4 : "url": "https://broker.example/ngsi-ld/v1/jsonldContexts/alpha-1",
1542 4 : "localId": "alpha-1",
1543 4 : "kind": "Hosted",
1544 4 : "createdAt": now_iso(),
1545 4 : "owner": "alpha",
1546 4 : "body": {"@context": {"a": "http://example.org/a"}},
1547 : });
1548 4 : let alpha = antares_model::TenantId::new("alpha").expect("tenant");
1549 4 : st.store
1550 4 : .context_put(Some(&alpha), "alpha-1", row.clone())
1551 4 : .await
1552 4 : .expect("seed hosted row");
1553 4 : st.loader
1554 4 : .resolve_for(&antares_model::TenantId::default(), &serde_json::json!(url))
1555 4 : .await
1556 4 : .expect("resolve");
1557 4 : assert_eq!(
1558 4 : st.store
1559 4 : .context_get(Some(&alpha), "alpha-1")
1560 4 : .await
1561 4 : .expect("store")
1562 4 : .expect("row"),
1563 : row,
1564 : "another Tenant's row must be untouched by a peer URL"
1565 : );
1566 4 : let cached = uuid::Uuid::new_v5(&uuid::Uuid::NAMESPACE_URL, url.as_bytes()).to_string();
1567 4 : assert_eq!(
1568 4 : st.store
1569 4 : .context_get(None, &cached)
1570 4 : .await
1571 4 : .expect("store")
1572 4 : .expect("row")["kind"],
1573 4 : serde_json::json!("Cached"),
1574 4 : "the fetched document is this broker's Cached copy (5.13.1)"
1575 4 : );
1576 4 : }
1577 : }
1578 :
1579 : #[cfg(all(test, feature = "test-kit"))]
1580 : mod clone_cost {
1581 : use super::*;
1582 :
1583 : /// every middleware layer and extractor of a request clones the
1584 : /// state; a clone copies no String, no client and no handle.
1585 : #[test]
1586 4 : fn a_clone_shares_the_per_process_part() {
1587 4 : let a = AppState::new("antares-test".into());
1588 4 : let b = a.clone();
1589 4 : assert!(std::ptr::eq(&a.http, &b.http), "the HTTP client is copied");
1590 4 : assert!(std::ptr::eq(&a.fed_http, &b.fed_http));
1591 4 : assert!(std::ptr::eq(a.public_url.as_ptr(), b.public_url.as_ptr()));
1592 4 : assert!(std::ptr::eq(a.store_name.as_ptr(), b.store_name.as_ptr()));
1593 4 : }
1594 :
1595 : /// A setting written before serving is the writer's own: a state cloned
1596 : /// earlier keeps what it had, as it did when every field was copied.
1597 : #[test]
1598 4 : fn a_write_after_a_clone_is_not_seen_by_the_clone() {
1599 4 : let a = AppState::new("antares-test".into());
1600 4 : let mut c = a.clone();
1601 4 : c.max_limit = 7;
1602 4 : assert_eq!((a.max_limit, c.max_limit), (1000, 7));
1603 4 : assert!(!std::ptr::eq(&a.http, &c.http));
1604 4 : }
1605 :
1606 : /// a setting written after an in-process call reaches the
1607 : /// next call; the memoised router of the clone was built over the old
1608 : /// settings and answered `limit=5` as if `max_limit` were still 1000.
1609 : #[tokio::test]
1610 4 : async fn a_write_after_a_call_reaches_the_next_call() {
1611 8 : let call = |st: &AppState| {
1612 8 : let req = axum::http::Request::get("/ngsi-ld/v1/entities?type=T&limit=5")
1613 8 : .body(axum::body::Body::empty())
1614 8 : .expect("request");
1615 8 : let st = st.clone();
1616 8 : async move { st.call(&axum::http::HeaderMap::new(), req).await.status() }
1617 8 : };
1618 4 : let a = AppState::new("antares-test".into());
1619 4 : assert_eq!(call(&a).await, axum::http::StatusCode::OK);
1620 4 : let mut c = a.clone();
1621 4 : c.max_limit = 1;
1622 4 : assert_eq!(call(&c).await, axum::http::StatusCode::FORBIDDEN);
1623 4 : }
1624 : }
|
