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 : }
30 :
31 : impl std::str::FromStr for TemporalRecord {
32 : type Err = String;
33 20 : fn from_str(s: &str) -> Result<Self, String> {
34 20 : match s {
35 20 : "all" => Ok(Self::All),
36 6 : "observed" => Ok(Self::Observed),
37 4 : "none" => Ok(Self::None),
38 2 : other => Err(format!(
39 2 : "ANTARES_TEMPORAL_RECORD: unknown mode {other} (all|observed|none)"
40 2 : )),
41 : }
42 20 : }
43 : }
44 :
45 : /// The handler `wire` installs for a notification that arrives on the
46 : /// internal distributed-subscription endpoint (5.8.1.4).
47 : pub type CsourceNotification = Arc<
48 : dyn for<'a> Fn(
49 : &'a AppState,
50 : &'a antares_model::TenantId,
51 : &'a str,
52 : Option<&'a str>,
53 : &'a [serde_json::Value],
54 : ) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send + 'a>>
55 : + Send
56 : + Sync,
57 : >;
58 :
59 : /// The request headers an in-process call carries from its caller
60 : /// ([`AppState::call`]): the two that select WHICH data the operation runs
61 : /// against, and the one that says what its terms mean. Everything else is
62 : /// the façade's own to set — an inner request is not the outer one.
63 : static PROPAGATED: [axum::http::HeaderName; 3] = [
64 : axum::http::HeaderName::from_static("ngsild-tenant"),
65 : axum::http::HeaderName::from_static("ngsild-snapshot"),
66 : axum::http::HeaderName::from_static("link"),
67 : ];
68 :
69 : tokio::task_local! {
70 : /// How many [`AppState::call`] frames this task is already inside, held
71 : /// against [`crate::bounds::MAX_IN_PROCESS_CALL_DEPTH`]. Absent outside a
72 : /// façade call, which reads as zero.
73 : static INPROCESS_DEPTH: usize;
74 : }
75 :
76 : #[derive(Clone)]
77 : pub struct AppState {
78 : pub store: Arc<dyn CurrentStateDriver>,
79 : /// The temporal driver — by default the same backend instance as
80 : /// `store`; a deployment may load a different one (or none).
81 : pub temporal: Arc<dyn TemporalDriver>,
82 : /// What the current-state driver is CALLED, reported by `/q/health` (NOT
83 : /// by `/info/sourceIdentity`, which is a spec resource) — a name, not an
84 : /// enumeration: a driver from outside this workspace mounts the same way
85 : /// as one from inside, and nothing here branches on it.
86 : pub store_name: String,
87 : /// The temporal backend `/q/health` names: the store's own mode when one
88 : /// instance serves both seams, `None` when history is off (`NoTemporal`);
89 : /// a second backend overwrites it after construction.
90 : pub temporal_name: Option<String>,
91 : pub loader: Arc<Loader>,
92 : pub started: Instant,
93 : /// Startup timestamp (createdAt of the built-in core @context entry).
94 : pub started_at: String,
95 : pub host_alias: String,
96 : /// Default page size for queries without an explicit limit.
97 : pub default_limit: usize,
98 : /// Hard ceiling on limit (TooManyResults guard).
99 : pub max_limit: usize,
100 : /// One shared outbound client, timeouts set at construction.
101 : pub http: antares_jsonld::HttpClient,
102 : /// Federation-forwarding client — longer deadline: the ETSI mock replies
103 : /// to unstubbed forwards only when the robot side wakes (up to ~5 s).
104 : pub fed_http: antares_jsonld::HttpClient,
105 : /// Notification bindings by `endpoint.uri` scheme (6.3.8, 7.2). The
106 : /// only way a delivery transport is chosen; `with_sink` adds one.
107 : pub sinks: Arc<antares_notifier::SinkRegistry>,
108 : /// HTTP surfaces mounted outside the NGSI-LD API root, each under its
109 : /// own reserved prefix; `with_surface` adds one. The admin surface is
110 : /// here by default.
111 : pub surfaces: Arc<Vec<Box<dyn crate::ApiSurface>>>,
112 : /// The router [`AppState::call`] serves in-process requests through,
113 : /// built on first use. Empty until a façade actually calls: building it
114 : /// costs about 1.5 ms, which is worth memoizing per state and not worth
115 : /// paying in a host that never makes an in-process call.
116 : pub(crate) inbound: Arc<std::sync::OnceLock<axum::Router>>,
117 : /// Bounds-wall rejection counters (exported by /q/health).
118 : pub limits: Arc<crate::bounds::LimitStats>,
119 : /// Allocator stats provider (set by the broker; None in tests/wasm).
120 : pub mem_stats: Option<Arc<dyn Fn() -> serde_json::Value + Send + Sync>>,
121 : /// Bus state provider for /q/health (`bus: {mode, connected,
122 : /// reconnects}`) — installed by the nats wiring only, so the member is
123 : /// absent for bus=local.
124 : pub bus_stats: Option<Arc<dyn Fn() -> serde_json::Value + Send + Sync>>,
125 : /// One egress policy for notifications and federation forwards
126 : /// (scheme allowlist, private-range deny, per-destination breakers).
127 : pub egress: Arc<crate::egress::Egress>,
128 : /// The policy engine every operation is asked about (ADR-0020), or
129 : /// `None` when a deployment attached none with
130 : /// [`AppState::with_policy`] — the default, where the broker behaves
131 : /// exactly as it did before the seam existed. `None` is not an instance
132 : /// of `AllowAll`: the gate takes its answer without building a subject,
133 : /// boxing a future or arming a timer, so a broker with no engine pays
134 : /// nothing for the seam.
135 : pub policy: Option<Arc<dyn crate::policy::PolicyEngine>>,
136 : /// Set the moment SIGTERM arrives, BEFORE the listener stops
137 : /// accepting — `/q/health` then answers 503 DRAINING so the load balancer
138 : /// takes this instance out while its socket still works.
139 : pub draining: Arc<std::sync::atomic::AtomicBool>,
140 : /// Change batches the matcher queue accepted and has not finished
141 : /// delivering. A shutdown drain waits for zero before the pool closes.
142 : pub pending_changes: Arc<std::sync::atomic::AtomicUsize>,
143 : /// Called after every Subscription and Context Source Registration
144 : /// Subscription CUD, so the mirror the matcher reads follows the store.
145 : /// On the bus the wiring pushes the change into the KV mirror bucket;
146 : /// in local mode it applies to this process's own mirror.
147 : #[allow(clippy::type_complexity)] // a boxed callback, named where it is installed
148 : pub sub_sync: Option<
149 : Arc<
150 : dyn Fn(&antares_model::TenantId, antares_store::Kind, &str, Option<&serde_json::Value>)
151 : + Send
152 : + Sync,
153 : >,
154 : >,
155 : /// bus=nats: the KV-watched compiled-subscription mirror the matcher
156 : /// reads, so the hot path never touches Postgres. `None` in local
157 : /// mode (the matcher reads the store directly).
158 : pub sub_mirror: Option<Arc<crate::mirror::SubMirror>>,
159 : /// bus=local: takes the entity changes one request buffered (the history
160 : /// layer hands them over after the handler) so the matcher sees a batch
161 : /// request as one unit. `None` until the local pipeline is wired.
162 : pub change_flush: Option<Arc<dyn Fn(Vec<crate::mirror::Change>) + Send + Sync>>,
163 : /// bus=nats: called after every Registration CUD so the wiring can
164 : /// publish the delta on `ANTARES_REGISTRY`. `None` in local mode.
165 : #[allow(clippy::type_complexity)] // a boxed callback, named where it is installed
166 : pub reg_sync: Option<
167 : Arc<dyn Fn(&antares_model::TenantId, &str, Option<&serde_json::Value>) + Send + Sync>,
168 : >,
169 : /// bus=nats: the ONE per-process compiled registration mirror,
170 : /// delta-fed from `ANTARES_REGISTRY`; expiry stays filtered at the single
171 : /// yield point (`federation::matching_regs`). `None` in local mode.
172 : pub reg_mirror: Option<Arc<crate::mirror::DocMirror>>,
173 : /// 5.2.34 (bus=nats): shares a cooldown stamp with the other api pods —
174 : /// a per-process stamp re-dials a failed source from every pod behind
175 : /// the LB. Seconds-scale state: broadcast on the
176 : /// registry stream, deliberately not persisted. `None` in local mode.
177 : #[allow(clippy::type_complexity)] // a boxed callback, named where it is installed
178 : pub reg_fail_sync: Option<Arc<dyn Fn(&str, bool) + Send + Sync>>,
179 : /// Renders the Prometheus text format for /q/metrics. Installed by
180 : /// the broker (the only crate that knows an exporter exists);
181 : /// `None` = 404, the facade calls elsewhere stay no-ops.
182 : pub metrics_render: Option<Arc<dyn Fn() -> String + Send + Sync>>,
183 : /// 5.8.1.4 consumer half: a notification addressed to the internal
184 : /// `urn:antares:distsub:` endpoint never leaves the broker, it re-enters
185 : /// it. The delivery path hands it here instead of naming the module that
186 : /// owns distributed subscriptions; `wire` installs the handler. `None`
187 : /// drops such a notification, which is what a broker that never created
188 : /// an internal subscription should do with one.
189 : pub csource_notification: Option<CsourceNotification>,
190 : /// True only under bus=nats (set by the broker's wiring): multiple
191 : /// processes share the store, so interval-subscription firings must be
192 : /// claimed single-winner. bus=local keeps the direct path — a
193 : /// claim there would disturb the 046_12 bookkeeping ordering for
194 : /// nothing.
195 : pub nats: bool,
196 : /// History gate 2 (ANTARES_TEMPORAL_RECORD): which changed attribute
197 : /// instances the write path records into history. Default `All`, which
198 : /// the ETSI temporal suites assume.
199 : pub temporal_record: TemporalRecord,
200 : /// The base URL remote Context Sources reach THIS broker at — used as
201 : /// the notification endpoint of forwarded subscription copies
202 : /// (5.8.1.4). ANTARES_PUBLIC_URL, defaulting to
203 : /// http://{host_alias}:{ANTARES_HTTP_PORT} (portless when 80/unset).
204 : pub public_url: String,
205 : /// 5.5.15 resource-pressure signal: max snapshots per tenant — above it
206 : /// the lowest-snapshotPriority snapshots are evicted. Snapshot documents
207 : /// themselves live in the store (Kind::Snapshot) so persistent modes
208 : /// survive restarts.
209 : pub snapshot_cap: usize,
210 : /// How a notification is delivered: attempts, backoff, age ceiling.
211 : /// Default = one attempt (5.8.6 as written).
212 : pub delivery: antares_notifier::DeliveryPolicy,
213 : }
214 :
215 : impl AppState {
216 : /// The built-in store: in-memory, or — under the reserved harness
217 : /// variable `ANTARES_TEST_STORE=file` — a fresh on-disk redb store per
218 : /// state, so the same test binary proves the durable backend without a
219 : /// second copy of every test. The broker's own boot path never calls
220 : /// this; it composes from `ANTARES_STORE` in `with_drivers`.
221 : #[cfg(feature = "test-kit")]
222 3092 : pub fn new(host_alias: String) -> Self {
223 : #[cfg(not(target_arch = "wasm32"))]
224 3092 : if std::env::var("ANTARES_TEST_STORE").as_deref() == Ok("file") {
225 0 : let dir = std::env::temp_dir()
226 0 : .join("antares-test-store")
227 0 : .join(uuid::Uuid::new_v4().simple().to_string());
228 : // reachable only through the harness variable read above; a
229 : // harness that cannot open its own store has nothing to run
230 : #[allow(clippy::expect_used)]
231 0 : let store = Store::open_file(&dir).expect("open the redb test store");
232 0 : return Self::with_store(host_alias, Arc::new(AnyStore::Mem(store)), "file");
233 3092 : }
234 3092 : Self::with_store(
235 3092 : host_alias,
236 3092 : Arc::new(AnyStore::Mem(Store::default())),
237 3092 : "memory",
238 : )
239 3092 : }
240 :
241 : /// Convenience over the built-in backends: one `AnyStore` serves as
242 : /// both drivers.
243 : #[cfg(feature = "test-kit")]
244 3140 : pub fn with_store(host_alias: String, store: Arc<AnyStore>, store_name: &str) -> Self {
245 3140 : let temporal: Arc<dyn TemporalDriver> = store.clone();
246 3140 : Self::with_drivers(host_alias, store, temporal, store_name)
247 3140 : }
248 :
249 3182 : pub fn with_drivers(
250 3182 : host_alias: String,
251 3182 : store: Arc<dyn CurrentStateDriver>,
252 3182 : temporal: Arc<dyn TemporalDriver>,
253 3182 : store_name: &str,
254 3182 : ) -> Self {
255 : // One policy value, read once, shared by every outbound path —
256 : // the gate (scheme/breakers) and the clients (DNS pinning, redirect
257 : // cap) can never disagree about what is allowed.
258 3182 : let egress_policy = antares_jsonld::EgressPolicy::from_env();
259 : // Cached-@context rows are the ONE source of truth for 5.13
260 : // existence — wire the write-through HERE so every composition
261 : // (native binary, wasm, tests) gets it; a composition that forgets
262 : // the writer is the "expiry checked in some paths" disease with
263 : // @contexts instead of expiry.
264 3182 : let loader = Arc::new(Loader::new());
265 : {
266 3182 : let store = store.clone();
267 3182 : loader.set_cache_writer(Arc::new(move |tenant, url, ctx_value| {
268 126 : let store = store.clone();
269 126 : Box::pin(async move {
270 126 : if hosted_row_id(&*store, tenant, url).await.is_some() {
271 0 : return; // broker-local (Hosted/Implicit) URLs are not Cached entries
272 126 : }
273 126 : let id = uuid::Uuid::new_v5(&uuid::Uuid::NAMESPACE_URL, url.as_bytes());
274 : // A refetch (staleness, delete+reload) must keep the row's
275 : // identity and hit counters — only the body is new.
276 : // The row this writes is `Cached` and belongs to no Tenant
277 : // (ADR-0021), but the call still acts for the Tenant whose
278 : // resolution triggered the fetch: that is what lets the guard
279 : // above see this Tenant's own Hosted rows.
280 126 : let prior = store
281 126 : .context_get(tenant, &id.to_string())
282 126 : .await
283 126 : .ok()
284 126 : .flatten();
285 252 : let field = |k: &str| {
286 252 : prior
287 252 : .as_ref()
288 252 : .and_then(|p| p[k].as_str())
289 252 : .map(str::to_owned)
290 252 : };
291 126 : let created = field("createdAt").unwrap_or_else(now_iso);
292 126 : let doc = serde_json::json!({
293 126 : "url": url,
294 126 : "localId": id.to_string(),
295 126 : "kind": "Cached",
296 126 : "createdAt": created,
297 126 : "numberOfHits": prior
298 126 : .as_ref()
299 126 : .and_then(|p| p["numberOfHits"].as_u64())
300 126 : .unwrap_or(0),
301 126 : "lastUsage": field("lastUsage"),
302 126 : "body": {"@context": ctx_value},
303 : });
304 126 : if let Err(e) = store.context_put(tenant, &id.to_string(), doc).await {
305 : // a client-named @context URL may carry userinfo
306 0 : tracing::warn!(
307 : "@context write-through failed for {}: {e}",
308 0 : antares_notifier::redact_userinfo(url)
309 : );
310 126 : }
311 126 : })
312 126 : }));
313 : }
314 : // 5.13.3.5: hit counters live in the SHARED row, not per instance
315 : // — behind a load balancer per-instance counters split-brain.
316 : // A bump that finds the row gone reports
317 : // a cross-instance delete; the loader then drops its warm copies so
318 : // the delete is honoured everywhere (5.13.5.4). Pinned core contexts
319 : // have no row and are never evicted.
320 : {
321 3182 : let store = store.clone();
322 3182 : loader.set_usage_bump(Arc::new(move |tenant, url| {
323 1542 : let store = store.clone();
324 1542 : Box::pin(async move {
325 1542 : if Loader::is_pinned_core(url) {
326 1284 : return true;
327 258 : }
328 : // One read, not two: a bump runs on every counted use of a
329 : // non-pinned @context, and the hosted probe already carries
330 : // the row it found.
331 258 : let held = hosted_row(&*store, tenant, url).await;
332 258 : let (id, row) = match held {
333 56 : Some((id, row)) => (id, Ok(Some(row))),
334 : None => {
335 202 : let id = uuid::Uuid::new_v5(&uuid::Uuid::NAMESPACE_URL, url.as_bytes())
336 202 : .to_string();
337 202 : let row = store.context_get(tenant, &id).await;
338 202 : (id, row)
339 : }
340 : };
341 254 : match row {
342 : // the row is what the store handed back, and `Value`'s
343 : // index panics on anything that is not an object; a row
344 : // that cannot carry the counters is left uncounted
345 254 : Ok(Some(mut doc)) if doc.is_object() => {
346 254 : let hits = doc["numberOfHits"].as_u64().unwrap_or(0) + 1;
347 254 : doc["numberOfHits"] = serde_json::json!(hits);
348 254 : doc["lastUsage"] = serde_json::json!(now_iso());
349 254 : if let Err(e) = store.context_put(tenant, &id, doc).await {
350 0 : tracing::warn!(
351 : "@context hit bump failed for {}: {e}",
352 0 : antares_notifier::redact_userinfo(url)
353 : );
354 254 : }
355 254 : true
356 : }
357 : // a row that cannot carry the counters is left
358 : // uncounted, and still exists: it must not be evicted
359 0 : Ok(Some(_)) => true,
360 4 : Ok(None) => false,
361 : // a store hiccup must never evict a healthy cache entry
362 0 : Err(_) => true,
363 : }
364 1542 : })
365 1542 : }));
366 : }
367 : // 5.13.1: an @context this broker HOSTS is served from its row, not
368 : // fetched. Its warm copy in the loader is only a cache — a restart
369 : // reloads Cached rows alone, and the bounded cache may evict it at
370 : // any moment — and the URL under it is minted from the request Host
371 : // header, so a miss would otherwise send the broker to whatever
372 : // address a client put there for the term mappings of its own
373 : // Tenant's payloads. The row carries the owner, so 5.5.10 still
374 : // decides who may resolve it.
375 : {
376 3182 : let store = store.clone();
377 4420 : loader.set_local_lookup(Arc::new(move |tenant, url| {
378 4238 : let store = store.clone();
379 4238 : Box::pin(async move {
380 4238 : let (_, row) = hosted_row(&*store, tenant, url).await?;
381 : // Ownership is one rule for the whole broker (ADR-0021): a
382 : // Cached row is a copy of a public document and belongs to no
383 : // Tenant, everything else belongs to its `owner` — with the
384 : // DEFAULT Tenant for a row written before that member
385 : // existed, which is the Tenant that lists, serves and deletes
386 : // it.
387 12 : let owner = antares_store::context_row_owner(&row)
388 12 : .and_then(|o| antares_model::TenantId::new_internal(o).ok());
389 12 : Some((owner, row["body"]["@context"].clone()))
390 4238 : })
391 4238 : }));
392 : }
393 : // 5.8.1.4: this URL is handed to peer brokers as the notification
394 : // endpoint for distributed subscriptions — the default must carry
395 : // the HTTP port or peers dial port 80 (ETSI-matrix ADV_02 shape).
396 3182 : let public_url =
397 3182 : std::env::var("ANTARES_PUBLIC_URL").unwrap_or_else(|_| {
398 3130 : match std::env::var("ANTARES_HTTP_PORT") {
399 12 : Ok(p) if p != "80" => format!("http://{host_alias}:{p}"),
400 3120 : _ => format!("http://{host_alias}"),
401 : }
402 3130 : });
403 3182 : let temporal_name = temporal.supported().then(|| store_name.to_owned());
404 : // 6.3.8 is the mandatory binding; clause 7's MQTT one is optional and
405 : // registers only when compiled in. Both go through the registry, so
406 : // notification delivery never names a transport.
407 3182 : let http = outbound_client(
408 3182 : egress_policy,
409 3182 : std::time::Duration::from_secs(5 * slow_factor()),
410 : );
411 3182 : let mut sinks = antares_notifier::SinkRegistry::default();
412 3182 : sinks.register(Box::new(antares_notifier::HttpSink::new(http.clone())));
413 : #[cfg(feature = "mqtt")]
414 3182 : sinks.register(Box::new(antares_notifier::mqtt::MqttSink::default()));
415 3182 : Self {
416 3182 : store,
417 3182 : temporal,
418 3182 : store_name: store_name.to_owned(),
419 3182 : temporal_name,
420 3182 : loader,
421 3182 : started: Instant::now(),
422 3182 : started_at: now_iso(),
423 3182 : host_alias,
424 3182 : default_limit: 1000,
425 3182 : max_limit: 1000,
426 3182 : http: http.clone(),
427 3182 : fed_http: outbound_client(
428 3182 : egress_policy,
429 3182 : std::time::Duration::from_secs(8 * slow_factor()),
430 3182 : ),
431 3182 : sinks: Arc::new(sinks),
432 3182 : inbound: Arc::new(std::sync::OnceLock::new()),
433 3182 : surfaces: Arc::new(vec![Box::new(crate::Admin)]),
434 3182 : limits: Arc::new(crate::bounds::LimitStats::default()),
435 3182 : mem_stats: None,
436 3182 : bus_stats: None,
437 3182 : egress: Arc::new(crate::egress::Egress::new(egress_policy)),
438 3182 : policy: None,
439 3182 : draining: Arc::new(std::sync::atomic::AtomicBool::new(false)),
440 3182 : pending_changes: Arc::default(),
441 3182 : sub_sync: None,
442 3182 : sub_mirror: None,
443 3182 : change_flush: None,
444 3182 : reg_sync: None,
445 3182 : reg_fail_sync: None,
446 3182 : reg_mirror: None,
447 3182 : metrics_render: None,
448 3182 : csource_notification: None,
449 3182 : nats: false,
450 3182 : temporal_record: TemporalRecord::All,
451 3182 : public_url,
452 3182 : snapshot_cap: 1024,
453 3182 : delivery: antares_notifier::DeliveryPolicy::default(),
454 3182 : }
455 3182 : }
456 :
457 : /// Attach a policy engine (ADR-0020). One engine per broker, chosen at
458 : /// startup: an engine that could be swapped per request would make what
459 : /// a caller may see a function of when they asked. Call before the
460 : /// state is shared.
461 : #[must_use]
462 446 : pub fn with_policy(mut self, engine: Arc<dyn crate::policy::PolicyEngine>) -> Self {
463 446 : self.policy = Some(engine);
464 446 : self
465 446 : }
466 :
467 : /// The engine `/q/health` reports. No engine reports the built-in name:
468 : /// the decision an operator reads is the one the broker takes.
469 90 : pub fn policy_name(&self) -> &str {
470 90 : self.policy
471 90 : .as_ref()
472 90 : .map_or(crate::policy::BUILT_IN_NAME, |e| e.name())
473 90 : }
474 :
475 : /// Register one more notification binding (6.3.8). A deployment adds a
476 : /// sink for a scheme this workspace does not ship; endpoints naming that
477 : /// scheme then validate and deliver through it. Call before the state is
478 : /// shared — a clone already handed out keeps the registry it was made
479 : /// with.
480 : ///
481 : /// Not behind `test-kit`, unlike the mirror accessors beside it: this is
482 : /// the sink seam itself (ADR-0016), the call `docs/src/extending.md`
483 : /// tells a deployment to make, and `examples/plugin-example` is the host
484 : /// that proves it from outside. A release library without it has no
485 : /// notification binding seam at all.
486 : #[must_use]
487 8 : pub fn with_sink(mut self, sink: Box<dyn antares_notifier::NotificationSink>) -> Self {
488 8 : match Arc::get_mut(&mut self.sinks) {
489 8 : Some(reg) => reg.register(sink),
490 0 : None => tracing::warn!("sink registry already shared; binding not registered"),
491 : }
492 8 : self
493 8 : }
494 :
495 : /// Mount one more HTTP surface (`/q`, `/x`, or below `/x`). A prefix
496 : /// outside those, or one that overlaps a surface already mounted, is an
497 : /// error the caller is expected to make fatal: a surface that could
498 : /// shadow a spec resource would make conformance a function of
499 : /// deployment configuration, and two surfaces on one prefix would leave
500 : /// the winner to route-matching order. Call before the state is shared.
501 96 : pub fn with_surface(mut self, s: Box<dyn crate::ApiSurface>) -> Result<Self, String> {
502 96 : crate::surface::check_prefix(s.prefix())?;
503 72 : if let Some(clash) = self
504 72 : .surfaces
505 72 : .iter()
506 72 : .find(|m| crate::surface::overlaps(m.prefix(), s.prefix()))
507 : {
508 16 : return Err(format!(
509 16 : "api surface {:?} claims {:?}, already served by {:?} at {:?}",
510 16 : s.name(),
511 16 : s.prefix(),
512 16 : clash.name(),
513 16 : clash.prefix()
514 16 : ));
515 56 : }
516 56 : match Arc::get_mut(&mut self.surfaces) {
517 56 : Some(v) => v.push(s),
518 0 : None => return Err("api surfaces already shared; register before serving".into()),
519 : }
520 56 : Ok(self)
521 96 : }
522 :
523 : /// Replace the mounted surfaces with a deployment's own selection —
524 : /// what a binary reads out of its configuration, rather than what the
525 : /// default mounting put there. Same prefix rules as `with_surface`, and
526 : /// a selection may leave admin out: `/q` is then not served at all.
527 20 : pub fn with_surfaces(mut self, list: Vec<Box<dyn crate::ApiSurface>>) -> Result<Self, String> {
528 20 : self.surfaces = Arc::new(Vec::new());
529 24 : for s in list {
530 24 : self = self.with_surface(s)?;
531 : }
532 16 : Ok(self)
533 20 : }
534 :
535 : /// Serve one request through this broker's own router, in process.
536 : ///
537 : /// This is the seam a façade for another standard (SensorThings, OGC
538 : /// API, WFS, OData) is built on: the façade is an [`crate::ApiSurface`]
539 : /// under `/x/<standard>` that translates its own request into an NGSI-LD
540 : /// one and calls this. There is no second data path — the inner request
541 : /// takes the same route as one off the socket, so negotiation, the
542 : /// bounds wall, tenancy, the policy seam, history and notifications all
543 : /// apply exactly once and exactly as they do for an NGSI-LD client.
544 : ///
545 : /// `caller` is the outer request's headers, and the ones that decide
546 : /// WHICH data an operation runs against are copied into the inner
547 : /// request when it does not set them itself:
548 : ///
549 : /// - `NGSILD-Tenant` (6.3.14) — a façade that forgot it would answer
550 : /// every caller out of the default tenant, so it is not left to the
551 : /// façade to remember;
552 : /// - `NGSILD-Snapshot` (6.3.22) — for the same reason: a façade called
553 : /// inside a snapshot request must not quietly serve live data;
554 : /// - `Link` (6.3.5) — the `@context` the caller supplied, so a term
555 : /// means the same thing on both sides of the translation;
556 : /// - every header `ANTARES_POLICY_SUBJECT_HEADERS` names, so the policy
557 : /// engine is asked about the caller rather than about the façade.
558 : ///
559 : /// All values of a copied header are carried, never just the first: a
560 : /// repeated `NGSILD-Tenant` is `BadRequestData` (6.3.14), and a façade
561 : /// must not be the place where a repeat is laundered into a single valid
562 : /// value.
563 : ///
564 : /// `&self` on purpose — an inner call runs while the outer handler is
565 : /// suspended on it, and the router clone is an `Arc` bump over the same
566 : /// state. The router itself is built once per state, on the first call:
567 : /// building one costs about 1.5 ms, which no façade should pay per
568 : /// request. After that first call the state counts as shared, so
569 : /// `with_surface` and the other builders refuse — which is the rule
570 : /// they already state.
571 46 : pub async fn call(
572 46 : &self,
573 46 : caller: &axum::http::HeaderMap,
574 46 : req: axum::http::Request<axum::body::Body>,
575 46 : ) -> axum::response::Response {
576 : use axum::response::IntoResponse as _;
577 : // A façade route that translates into a request its own surface
578 : // serves would call itself for as long as the stack lasts, and
579 : // every frame builds a router. The chain is counted per task, so a
580 : // façade over a façade is served and a loop is not.
581 46 : let depth = INPROCESS_DEPTH.try_with(|d| *d).unwrap_or(0);
582 46 : if depth >= crate::bounds::MAX_IN_PROCESS_CALL_DEPTH {
583 2 : return crate::negotiate::ApiError::from(antares_model::NgsiError::InternalError(
584 2 : format!(
585 2 : "in-process call depth exceeded {}",
586 2 : crate::bounds::MAX_IN_PROCESS_CALL_DEPTH
587 2 : ),
588 2 : ))
589 2 : .into_response();
590 44 : }
591 44 : let router = self.inbound.get_or_init(|| {
592 : // Every layer of a built router captured an `AppState` clone, so
593 : // the memo owns states. The clone it is built from must NOT be
594 : // able to reach THIS cell: a state that could would hold itself
595 : // alive for the life of the process and pin its store handle
596 : // with it — a file store would keep its lock after the host
597 : // dropped everything. A fresh cell means the memo of a nested
598 : // façade call hangs off this one and dies with it.
599 26 : let mut inner = self.clone();
600 26 : inner.inbound = Arc::new(std::sync::OnceLock::new());
601 26 : crate::router(inner)
602 26 : });
603 44 : let (mut parts, body) = req.into_parts();
604 132 : for name in &PROPAGATED {
605 132 : if parts.headers.contains_key(name) {
606 0 : continue;
607 132 : }
608 132 : for v in caller.get_all(name) {
609 18 : parts.headers.append(name.clone(), v.clone());
610 18 : }
611 : }
612 44 : for name in crate::policy::SUBJECT_HEADERS.iter() {
613 44 : let Ok(name) = axum::http::HeaderName::from_bytes(name.as_bytes()) else {
614 0 : continue;
615 : };
616 44 : if parts.headers.contains_key(&name) {
617 0 : continue;
618 44 : }
619 44 : for v in caller.get_all(&name) {
620 4 : parts.headers.append(name.clone(), v.clone());
621 4 : }
622 : }
623 44 : let req = axum::http::Request::from_parts(parts, body);
624 44 : match INPROCESS_DEPTH
625 44 : .scope(depth + 1, tower::ServiceExt::oneshot(router.clone(), req))
626 44 : .await
627 : {
628 44 : Ok(r) => r,
629 : // `Router`'s error is `Infallible`; the arm stays honest rather
630 : // than unwrapping (the workspace denies unwrap outside tests).
631 0 : Err(_) => crate::negotiate::ApiError::from(antares_model::NgsiError::InternalError(
632 0 : "the in-process router failed".into(),
633 0 : ))
634 0 : .into_response(),
635 : }
636 46 : }
637 :
638 : /// Temporal auto-recording happens synchronously in the write path in
639 : /// EVERY bus mode (read-your-writes) — but only when the loaded
640 : /// temporal driver actually records anything.
641 12200 : pub fn record_locally(&self) -> bool {
642 12200 : self.temporal.supported()
643 12200 : }
644 :
645 : /// Fire the subscription-sync hook for one written row. `kind` decides
646 : /// what the mirror does with it: a Subscription is indexed as a document,
647 : /// a Context Source Registration Subscription only wakes the interval
648 : /// sweep (5.11.7) — it is matched against registrations, not entities.
649 368 : pub(crate) fn sub_changed(
650 368 : &self,
651 368 : tenant: &antares_model::TenantId,
652 368 : kind: antares_store::Kind,
653 368 : id: &str,
654 368 : doc: Option<&serde_json::Value>,
655 368 : ) {
656 368 : if let Some(h) = &self.sub_sync {
657 296 : h(tenant, kind, id, doc);
658 296 : }
659 368 : }
660 :
661 : /// 5.2.34: stamp the per-registration cooldown locally AND on the other
662 : /// api pods (no-op half in local mode). Keyed per tenant — the id alone
663 : /// is client-chosen per tenant (5.5.10) and must not gate a neighbour.
664 0 : pub(crate) fn reg_cooldown_stamp(
665 0 : &self,
666 0 : tenant: &antares_model::TenantId,
667 0 : reg_id: &str,
668 0 : ok: bool,
669 0 : ) {
670 0 : let key = crate::egress::reg_key(tenant.as_str(), reg_id);
671 0 : self.egress.reg_record(&key, ok);
672 0 : if let Some(h) = &self.reg_fail_sync {
673 0 : // the broadcast carries the COMPOSED key; receiving pods stamp it
674 0 : // verbatim, so their lookups agree without re-deriving anything
675 0 : h(&key, ok);
676 0 : }
677 0 : }
678 :
679 : /// Fire the registration-delta hook (no-op in local mode).
680 2134 : pub(crate) fn reg_changed(
681 2134 : &self,
682 2134 : tenant: &antares_model::TenantId,
683 2134 : id: &str,
684 2134 : doc: Option<&serde_json::Value>,
685 2134 : ) {
686 2134 : if let Some(h) = &self.reg_sync {
687 0 : h(tenant, id, doc);
688 2134 : }
689 2134 : }
690 : }
691 :
692 : /// The ONE outbound-client construction for this crate (timeouts at
693 : /// construction). Title-case headers are an http1 knob and timeouts are
694 : /// client-level knobs — both native-only; the browser's fetch supplies its
695 : /// own transport on wasm32.
696 : /// Outbound timeouts stretch 10× when the test binary runs under a
697 : /// sanitizer (ANTARES_TEST_SANITIZER, set by the strict workflow):
698 : /// ThreadSanitizer slows every thread, and 371 tests sharing one runner
699 : /// pushed loopback forwards past the 2 s connect / 5 s total limits —
700 : /// each run failing a different test with a 504. Production is unchanged.
701 13359 : pub fn slow_factor() -> u64 {
702 13359 : antares_jsonld::slow_factor()
703 13359 : }
704 :
705 6364 : fn outbound_client(
706 6364 : policy: antares_jsonld::EgressPolicy,
707 6364 : total: std::time::Duration,
708 6364 : ) -> antares_jsonld::HttpClient {
709 6364 : let b = antares_jsonld::with_timeouts(
710 6364 : antares_jsonld::client_builder(policy),
711 6364 : std::time::Duration::from_secs(2 * slow_factor()),
712 6364 : total,
713 : );
714 : // the suite's notification receiver asserts header names
715 : // case-sensitively ("Link", "X-Additional-Key")
716 : #[cfg(not(target_arch = "wasm32"))]
717 6364 : let b = b.http1_title_case_headers();
718 : // reqwest fails to build only when the process cannot initialise its TLS
719 : // backend: no outbound request of any kind can be made after that
720 : #[allow(clippy::expect_used)]
721 6364 : let c = b.build().expect("http client");
722 6364 : antares_jsonld::wrap_client(c)
723 6364 : }
724 :
725 : /// 5.13.1: an @context this broker HOSTS (Hosted or ImplicitlyCreated) is
726 : /// identified by its stored row, never by the URL's shape — a peer broker or
727 : /// an attacker can serve a document under the same resource path, and such a
728 : /// URL is external to us (a Cached entry). The row records the URL it was
729 : /// minted under, so that is what decides: a URL whose trailing segment names
730 : /// a stored row but whose origin is somebody else's names a document this
731 : /// broker does not host, and the stored row — another Tenant's, as often as
732 : /// not — must not be read, counted or rewritten for it. Returns the local
733 : /// row id when the URL is the one the row was minted under.
734 4622 : async fn hosted_row(
735 4622 : store: &dyn CurrentStateDriver,
736 4622 : tenant: Option<&antares_model::TenantId>,
737 4622 : url: &str,
738 4622 : ) -> Option<(String, serde_json::Value)> {
739 4622 : let (_, seg) = url.rsplit_once("/ngsi-ld/v1/jsonldContexts/")?;
740 100 : let seg = seg.split(['?', '#']).next().unwrap_or(seg);
741 100 : if seg.is_empty() || seg.contains('/') {
742 0 : return None;
743 100 : }
744 100 : let row = store
745 100 : .context_get(tenant, seg)
746 100 : .await
747 100 : .ok()
748 100 : .flatten()
749 100 : .filter(|row| row["url"].as_str() == Some(url))?;
750 68 : Some((seg.to_owned(), row))
751 4622 : }
752 :
753 126 : async fn hosted_row_id(
754 126 : store: &dyn CurrentStateDriver,
755 126 : tenant: Option<&antares_model::TenantId>,
756 126 : url: &str,
757 126 : ) -> Option<String> {
758 126 : hosted_row(store, tenant, url).await.map(|(id, _)| id)
759 126 : }
760 :
761 : /// Server-managed timestamp, ISO 8601 UTC with milliseconds.
762 18163 : pub fn now_iso() -> String {
763 18163 : chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
764 18163 : }
765 :
766 : #[cfg(test)]
767 : mod jsonld_context_locality_5_13 {
768 : use super::*;
769 :
770 : /// Serve one @context document under `path` on a loopback port, counting
771 : /// fetches (the negative half: a warm copy must NOT be refetched).
772 12 : fn context_server(path: &str) -> (String, Arc<std::sync::atomic::AtomicUsize>) {
773 12 : let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
774 12 : let port = listener.local_addr().expect("addr").port();
775 12 : let fetches = Arc::new(std::sync::atomic::AtomicUsize::new(0));
776 12 : let n = fetches.clone();
777 12 : std::thread::spawn(move || {
778 12 : for stream in listener.incoming() {
779 8 : let Ok(mut s) = stream else { continue };
780 8 : n.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
781 8 : let body = r#"{"@context":{"peerTemp":"http://example.org/peerTemp"}}"#;
782 8 : let resp = format!(
783 : "HTTP/1.1 200 OK\r\nContent-Type: application/ld+json\r\n\
784 : Content-Length: {}\r\nConnection: close\r\n\r\n{body}",
785 8 : body.len()
786 : );
787 : use std::io::{Read, Write};
788 8 : let mut buf = [0u8; 2048];
789 8 : let _ = s.read(&mut buf);
790 8 : let _ = s.write_all(resp.as_bytes());
791 : }
792 12 : });
793 12 : (format!("http://127.0.0.1:{port}{path}"), fetches)
794 12 : }
795 :
796 8 : fn fetch_count(c: &Arc<std::sync::atomic::AtomicUsize>) -> usize {
797 8 : c.load(std::sync::atomic::Ordering::SeqCst)
798 8 : }
799 :
800 : /// 5.13.1: a Cached @context is one this broker fetched from elsewhere.
801 : /// Which URLs this broker HOSTS is a property of the stored rows, not of
802 : /// the URL path, so a peer's @context served under the same resource path
803 : /// is persisted as Cached and never mistaken for a row deleted through
804 : /// another instance (5.13.5.4).
805 : #[tokio::test]
806 4 : async fn a_peer_context_url_under_the_local_path_is_cached() {
807 4 : let st = AppState::new("me".into());
808 4 : let (url, fetches) = context_server("/ngsi-ld/v1/jsonldContexts/peer-ctx");
809 4 : let user = serde_json::json!(url);
810 4 : st.loader.resolve(&user).await.expect("resolve");
811 4 : let id = uuid::Uuid::new_v5(&uuid::Uuid::NAMESPACE_URL, url.as_bytes()).to_string();
812 4 : let row = st
813 4 : .store
814 4 : .context_get(None, &id)
815 4 : .await
816 4 : .expect("store")
817 4 : .expect("the fetched @context is persisted as a Cached row");
818 4 : assert_eq!(row["kind"], "Cached");
819 4 : assert_eq!(row["url"], serde_json::json!(url));
820 4 : st.loader.resolve(&user).await.expect("resolve again");
821 4 : assert_eq!(
822 4 : fetch_count(&fetches),
823 : 1,
824 : "a warm, still-existing @context must not be refetched"
825 : );
826 4 : assert_eq!(
827 4 : st.store
828 4 : .context_get(None, &id)
829 4 : .await
830 4 : .expect("store")
831 4 : .expect("row")["numberOfHits"],
832 4 : serde_json::json!(2),
833 4 : "both counted uses land on the shared row (5.13.3.5)"
834 4 : );
835 4 : }
836 :
837 : /// 5.13.3.5 counts uses of an @context this broker hosts on its own
838 : /// stored row — no second, Cached copy of the same document, and no
839 : /// fetch of its URL: the row holds the document (5.13.1 "Hosted").
840 : #[tokio::test]
841 4 : async fn a_hosted_context_is_counted_on_its_own_row() {
842 4 : let st = AppState::new("me".into());
843 4 : let owner = antares_model::TenantId::default();
844 4 : let (url, fetches) = context_server("/ngsi-ld/v1/jsonldContexts/local-1");
845 : // no `owner` member: a row written before it existed belongs to the
846 : // default Tenant, which is the one resolving below
847 4 : st.store
848 4 : .context_put(
849 4 : Some(&owner),
850 4 : "local-1",
851 4 : serde_json::json!({
852 4 : "url": url,
853 4 : "localId": "local-1",
854 4 : "kind": "Hosted",
855 4 : "createdAt": now_iso(),
856 4 : "body": {"@context": {"peerTemp": "http://example.org/peerTemp"}},
857 4 : }),
858 4 : )
859 4 : .await
860 4 : .expect("seed hosted row");
861 4 : let user = serde_json::json!(url);
862 4 : st.loader.resolve_for(&owner, &user).await.expect("resolve");
863 4 : st.loader
864 4 : .resolve_for(&owner, &user)
865 4 : .await
866 4 : .expect("resolve again");
867 4 : let cached = uuid::Uuid::new_v5(&uuid::Uuid::NAMESPACE_URL, url.as_bytes()).to_string();
868 4 : assert!(
869 4 : st.store
870 4 : .context_get(None, &cached)
871 4 : .await
872 4 : .expect("store")
873 4 : .is_none(),
874 : "a hosted @context must not be duplicated as a Cached row"
875 : );
876 4 : assert_eq!(
877 4 : st.store
878 4 : .context_get(Some(&owner), "local-1")
879 4 : .await
880 4 : .expect("store")
881 4 : .expect("row")["numberOfHits"],
882 4 : serde_json::json!(2)
883 : );
884 4 : assert_eq!(
885 4 : fetch_count(&fetches),
886 4 : 0,
887 4 : "a hosted @context is read from its row, never fetched over the network"
888 4 : );
889 4 : }
890 :
891 : /// 5.13.1: what this broker HOSTS comes from its own store, not from the
892 : /// network. The in-process copy is a cache, and a miss — a restart (only
893 : /// `Cached` rows are preloaded) or an eviction from the bounded document
894 : /// cache — must not turn into an outbound GET of a URL a client chose:
895 : /// the URL is minted from the request's `Host` header when
896 : /// `ANTARES_PUBLIC_URL` is unset, so a spoofed one would send the broker
897 : /// to the spoofer for the term mappings that expand that Tenant's
898 : /// payloads. 5.5.10 still holds through the store: the row's owner is
899 : /// the only Tenant it resolves for.
900 : #[tokio::test]
901 4 : async fn a_hosted_context_resolves_from_the_store_after_its_copy_is_gone() {
902 4 : let st = AppState::new("me".into());
903 4 : let alpha = antares_model::TenantId::new("alpha").expect("tenant");
904 4 : let beta = antares_model::TenantId::new("beta").expect("tenant");
905 : // a dead port: anything resolved can only have come from the row
906 4 : let url = "http://127.0.0.1:9/ngsi-ld/v1/jsonldContexts/hosted-1";
907 4 : st.store
908 4 : .context_put(
909 4 : Some(&alpha),
910 4 : "hosted-1",
911 4 : serde_json::json!({
912 4 : "url": url,
913 4 : "localId": "hosted-1",
914 4 : "kind": "Hosted",
915 4 : "createdAt": now_iso(),
916 4 : "owner": "alpha",
917 4 : "body": {"@context": {"secret": "https://alpha.example/secret"}},
918 4 : }),
919 4 : )
920 4 : .await
921 4 : .expect("seed hosted row");
922 4 : let user = serde_json::json!(url);
923 4 : let ctx = st
924 4 : .loader
925 4 : .resolve_for(&alpha, &user)
926 4 : .await
927 4 : .expect("the owning Tenant resolves its stored @context");
928 4 : assert_eq!(ctx.expand_key("secret"), "https://alpha.example/secret");
929 4 : let err = st
930 4 : .loader
931 4 : .resolve_for(&beta, &user)
932 4 : .await
933 4 : .expect_err("another Tenant may not resolve it (5.5.10)");
934 4 : assert!(
935 4 : matches!(err, antares_model::NgsiError::LdContextNotAvailable(_)),
936 4 : "got {err:?}"
937 4 : );
938 4 : }
939 :
940 : /// 5.13.1 + 5.5.10: a stored @context with no `owner` member is not
941 : /// ownerless. `contexts.rs` `row_visible` reads a row written before that
942 : /// member existed as the DEFAULT Tenant's — it is listed, served and
943 : /// deleted through that Tenant alone — so resolving one must answer the
944 : /// same way. Read as "belongs to no Tenant" it would expand every other
945 : /// Tenant's payloads with the default Tenant's private term mappings.
946 : #[tokio::test]
947 4 : async fn a_stored_context_without_an_owner_belongs_to_the_default_tenant() {
948 4 : let st = AppState::new("me".into());
949 4 : let default = antares_model::TenantId::new("default").expect("tenant");
950 4 : let other = antares_model::TenantId::new("beta").expect("tenant");
951 4 : let url = "http://127.0.0.1:9/ngsi-ld/v1/jsonldContexts/legacy-1";
952 4 : st.store
953 4 : .context_put(
954 4 : Some(&default),
955 4 : "legacy-1",
956 4 : serde_json::json!({
957 4 : "url": url,
958 4 : "localId": "legacy-1",
959 4 : "kind": "Hosted",
960 4 : "createdAt": now_iso(),
961 4 : "body": {"@context": {"legacy": "https://legacy.example/term"}},
962 4 : }),
963 4 : )
964 4 : .await
965 4 : .expect("seed a row from before the owner member");
966 4 : let user = serde_json::json!(url);
967 4 : let ctx = st
968 4 : .loader
969 4 : .resolve_for(&default, &user)
970 4 : .await
971 4 : .expect("the Tenant the row belongs to resolves it");
972 4 : assert_eq!(ctx.expand_key("legacy"), "https://legacy.example/term");
973 4 : let err = st
974 4 : .loader
975 4 : .resolve_for(&other, &user)
976 4 : .await
977 4 : .expect_err("no other Tenant may resolve it (5.5.10)");
978 4 : assert!(
979 4 : matches!(err, antares_model::NgsiError::LdContextNotAvailable(_)),
980 4 : "got {err:?}"
981 4 : );
982 4 : }
983 :
984 : /// 5.13.1 again, with the local id of a row that exists: what this broker
985 : /// hosts is the row MINTED under that URL, so a document served from
986 : /// somewhere else under the same resource path — with the local id of a
987 : /// Hosted @context another Tenant added — is external. It becomes a
988 : /// Cached row of its own, and the Tenant's row is neither read for its
989 : /// mappings nor counted against.
990 : #[tokio::test]
991 4 : async fn a_peer_url_reusing_a_stored_local_id_leaves_that_row_alone() {
992 4 : let st = AppState::new("me".into());
993 4 : let (url, _) = context_server("/ngsi-ld/v1/jsonldContexts/alpha-1");
994 4 : let row = serde_json::json!({
995 4 : "url": "https://broker.example/ngsi-ld/v1/jsonldContexts/alpha-1",
996 4 : "localId": "alpha-1",
997 4 : "kind": "Hosted",
998 4 : "createdAt": now_iso(),
999 4 : "owner": "alpha",
1000 4 : "body": {"@context": {"a": "http://example.org/a"}},
1001 : });
1002 4 : let alpha = antares_model::TenantId::new("alpha").expect("tenant");
1003 4 : st.store
1004 4 : .context_put(Some(&alpha), "alpha-1", row.clone())
1005 4 : .await
1006 4 : .expect("seed hosted row");
1007 4 : st.loader
1008 4 : .resolve_for(&antares_model::TenantId::default(), &serde_json::json!(url))
1009 4 : .await
1010 4 : .expect("resolve");
1011 4 : assert_eq!(
1012 4 : st.store
1013 4 : .context_get(Some(&alpha), "alpha-1")
1014 4 : .await
1015 4 : .expect("store")
1016 4 : .expect("row"),
1017 : row,
1018 : "another Tenant's row must be untouched by a peer URL"
1019 : );
1020 4 : let cached = uuid::Uuid::new_v5(&uuid::Uuid::NAMESPACE_URL, url.as_bytes()).to_string();
1021 4 : assert_eq!(
1022 4 : st.store
1023 4 : .context_get(None, &cached)
1024 4 : .await
1025 4 : .expect("store")
1026 4 : .expect("row")["kind"],
1027 4 : serde_json::json!("Cached"),
1028 4 : "the fetched document is this broker's Cached copy (5.13.1)"
1029 4 : );
1030 4 : }
1031 : }
|
