Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Antares — NGSI-LD context broker (composition root).
3 : //!
4 : //! Config: ANTARES_* env vars only for v0 (antares.toml layering can land
5 : //! later via figment). Unknown ANTARES_* keys are fatal.
6 : #![cfg_attr(not(test), warn(clippy::expect_used))]
7 :
8 : mod shutdown;
9 : mod telemetry;
10 : mod wiring;
11 :
12 : use antares_api::AppState;
13 :
14 : // ANTARES_DATABASE_URL: accepted — the ETSI compose wires one DB per broker —
15 : // consumed by the postgres/timescale store modes.
16 : const KNOWN_KEYS: &[&str] = &[
17 : "ANTARES_HTTP_PORT",
18 : "ANTARES_HOST_ALIAS",
19 : // 5.8.1.4 distributed subscriptions: the public base URL other brokers
20 : // reach this one at (the reduced-copy notification endpoint); defaults
21 : // to http://{host_alias}.
22 : "ANTARES_PUBLIC_URL",
23 : "ANTARES_ROLES",
24 : // The example policy engine's rules document; accepted only in a build
25 : // that carries the engine, so a release binary rejects it like any
26 : // other key it has no code for.
27 : #[cfg(feature = "plugin-example")]
28 : antares_plugin_example::RULES_ENV,
29 : "ANTARES_DATABASE_URL",
30 : "ANTARES_STORE",
31 : // Temporal driver: a store mode, or `none` — history off (temporal
32 : // reads answer OperationNotSupported, Table 6.3.2-1). Defaults to the
33 : // current-state store, so one instance serves both seams.
34 : "ANTARES_TEMPORAL",
35 : // History gate: `all` (default) records every changed instance;
36 : // `observed` records only instances carrying observedAt; `none`
37 : // auto-records nothing (temporal API + reads stay on).
38 : "ANTARES_TEMPORAL_RECORD",
39 : "ANTARES_DATA_DIR",
40 : // Egress: private-range destinations are ALLOWED by default (ADR-0010 —
41 : // brokers federate inside private networks); a hardened deployment sets
42 : // this to false to arm the SSRF wall.
43 : "ANTARES_EGRESS_ALLOW_PRIVATE",
44 : // Refuse to start when the DB role bypasses RLS (production gate;
45 : // default off so the dev/ETSI superuser stack still boots).
46 : "ANTARES_REQUIRE_RLS",
47 : // Temporal retention horizon in days; absent = keep forever (a
48 : // maintenance job must never default to dropping data).
49 : "ANTARES_TEMPORAL_RETENTION_DAYS",
50 : // 4.22 GC interval (memory/file arm); default 900 s, the ETSI stack runs
51 : // at 2 s so the transient TPs (422_01) exercise the sweep itself.
52 : "ANTARES_SWEEP_SECS",
53 : // Batch entity-count cap; default 1000 — raised where a
54 : // trusted producer legitimately batches larger (the spec sets no ceiling).
55 : "ANTARES_MAX_BATCH_ITEMS",
56 : "ANTARES_MAX_BODY_BYTES",
57 : "ANTARES_CORS_ORIGINS",
58 : // The policy engine every operation is asked about, and how long it has
59 : // to answer before the seam denies (ADR-0020). The built-in allow-all
60 : // engine decides nothing and never waits; the header list is what an
61 : // engine is given to identify the caller by.
62 : "ANTARES_POLICY",
63 : "ANTARES_POLICY_SUBJECT_HEADERS",
64 : "ANTARES_POLICY_TIMEOUT_MS",
65 : // The HTTP surfaces mounted beside the NGSI-LD API root, comma-separated;
66 : // default `admin` (/q). An unknown name is fatal and names the shelf.
67 : "ANTARES_API_SURFACES",
68 : // Drain: the LB notice window, and the ceiling on waiting for
69 : // in-flight requests once the listener has closed.
70 : "ANTARES_DRAIN_DELAY_MS",
71 : "ANTARES_DRAIN_DEADLINE_SECS",
72 : // Optional PEM bundle of extra TLS trust anchors (private CAs,
73 : // incomplete-chain servers). Never disables verification.
74 : "ANTARES_EXTRA_CA_FILE",
75 : // The bus seam: local (default, single process, all roles) or
76 : // nats (the JetStream spine — requires a postgres/timescale store and
77 : // ANTARES_NATS_URL).
78 : "ANTARES_BUS",
79 : "ANTARES_NATS_URL",
80 : // Stream/KV replication factor on a clustered JetStream (3 for
81 : // the reference manifests' R3; default 1 for single-node).
82 : "ANTARES_NATS_REPLICAS",
83 : // OTLP/HTTP span export endpoint (e.g. http://collector:4318/v1/traces);
84 : // unset = no OTLP anywhere.
85 : "ANTARES_OTLP_ENDPOINT",
86 : "ANTARES_TELEMETRY",
87 : // Outbox drain on this pod, on (default) | off. `off` is the
88 : // crash-drill lever (rows commit but this pod never publishes them —
89 : // another pod's drain must) and the knob for a dedicated-drainer split.
90 : "ANTARES_OUTBOX_DRAIN",
91 : // Notification delivery policy: total attempts (default 1 = 5.8.6 as
92 : // written), first-retry backoff, and the age after which no retry
93 : // starts. An exhausted policy leaves a dead letter (/q/dead-letters).
94 : "ANTARES_NOTIFY_ATTEMPTS",
95 : "ANTARES_NOTIFY_BACKOFF_MS",
96 : "ANTARES_NOTIFY_MAX_AGE_SECS",
97 : // Postgres pool size (max connections); default 20.
98 : "ANTARES_PG_POOL",
99 : "ANTARES_PG_STATEMENT_TIMEOUT_MS",
100 : // bus=local over a shared postgres/timescale store is refused — every
101 : // replica would run its own matcher and fire its own copy of each
102 : // notification. This opt-in states the deployment runs exactly ONE
103 : // broker process against that database.
104 : "ANTARES_ALLOW_SHARED_LOCAL",
105 : // HTTP/1 header read timeout in ms (default 10000): a connection that
106 : // never finishes its request headers is closed instead of holding a
107 : // slot forever.
108 : "ANTARES_HEADER_READ_TIMEOUT_MS",
109 : // Ceiling on concurrently served connections (default 10000);
110 : // connections accepted above it are dropped immediately.
111 : "ANTARES_MAX_CONNECTIONS",
112 : // 5.8.6 delivery concurrency: notifications in flight at once for the
113 : // whole broker, and the share of that width one tenant may hold.
114 : "ANTARES_DELIVERY_WIDTH",
115 : "ANTARES_DELIVERY_WIDTH_PER_TENANT",
116 : // 5.7.2.4 fan-out ceiling: how many matching registrations one
117 : // distributed operation may contact.
118 : "ANTARES_FED_FANOUT",
119 : "ANTARES_FED_INFLIGHT",
120 : // Ceiling on the body this broker will read back from a forwarded
121 : // request.
122 : "ANTARES_MAX_FED_RESPONSE_BYTES",
123 : // 5.7.5/5.7.6 discovery scan ceiling (types/attributes listing).
124 : "ANTARES_DISCOVERY_SCAN_MAX",
125 : // Run the DDL on this process; off keeps replicas from racing the
126 : // migration on boot.
127 : "ANTARES_MIGRATE",
128 : ];
129 :
130 : /// Jemalloc with decay-based purging — RSS returns to ~live×1.2 when idle;
131 : /// tune via MALLOC_CONF (e.g. dirty_decay_ms). Deliberately not glibc
132 : /// malloc, whose arena fragmentation never gives memory back.
133 : #[global_allocator]
134 : static ALLOC: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
135 :
136 : /// Unknown-config-is-fatal, minus what the platform injects: a Service
137 : /// named `antares*` makes kubelet write ANTARES_PORT,
138 : /// ANTARES_PORT_9090_TCP*, ANTARES_SERVICE_* (and the antares-file /
139 : /// antares-api variants) into every pod, and treating those as typos put the
140 : /// shipped manifests into 100% CrashLoopBackOff. The manifests also set
141 : /// enableServiceLinks: false — this check is the belt for clusters that
142 : /// re-enable links or add their own Services.
143 8516 : fn unknown_config_key(key: &str) -> bool {
144 8516 : if !key.starts_with("ANTARES_") || key.starts_with("ANTARES_TEST_") {
145 8099 : return false;
146 417 : }
147 417 : if KNOWN_KEYS.contains(&key) {
148 385 : return false;
149 32 : }
150 : // kubelet service-link shapes for the Services OUR manifests ship
151 : // (antares, antares-file, antares-api, antares-worker): {NAME}_PORT,
152 : // {NAME}_PORT_<n>_<proto>*, {NAME}_SERVICE_*. Only those exact name
153 : // infixes are exempt — an arbitrary ANTARES_*-shaped var stays a fatal
154 : // typo, and foreign Services are covered by
155 : // enableServiceLinks: false in the manifests.
156 32 : let rest = &key["ANTARES_".len()..];
157 92 : let injected = ["", "FILE_", "API_", "WORKER_"].iter().any(|infix| {
158 92 : rest.strip_prefix(infix)
159 92 : .is_some_and(|t| t == "PORT" || t.starts_with("PORT_") || t.starts_with("SERVICE_"))
160 92 : });
161 32 : !injected
162 8516 : }
163 :
164 : /// ANTARES_SWEEP_SECS paces the 4.22 expiry sweep in every store mode. Absent
165 : /// is the 15 min default; anything that is not a positive integer is fatal,
166 : /// because a garbage cadence silently becoming the default one is exactly the
167 : /// misconfiguration the unknown-key policy exists to catch.
168 69 : fn parse_sweep_secs(raw: Option<&str>) -> Result<u64, String> {
169 69 : let Some(v) = raw else {
170 45 : return Ok(15 * 60);
171 : };
172 24 : match v.parse::<u64>() {
173 18 : Ok(0) | Err(_) => Err(format!(
174 18 : "ANTARES_SWEEP_SECS must be a positive integer number of seconds, got {v:?}"
175 18 : )),
176 6 : Ok(n) => Ok(n),
177 : }
178 69 : }
179 :
180 : /// An explicitly-off switch value, whatever the operator's spelling. Used by
181 : /// the knobs where the DEFAULT is off, so that only an off value keeps them
182 : /// off and a typo cannot silently disable a security control.
183 74 : fn is_off(v: &str) -> bool {
184 74 : let v = v.trim();
185 74 : v.is_empty()
186 66 : || v == "0"
187 58 : || v.eq_ignore_ascii_case("false")
188 48 : || v.eq_ignore_ascii_case("off")
189 40 : || v.eq_ignore_ascii_case("no")
190 74 : }
191 :
192 : /// One plain HTTP/1.0 GET of /q/health on the configured port; anything but
193 : /// a 200 status line is an error, so `HEALTHCHECK` sees exit 1.
194 6 : fn health_probe() -> Result<(), Box<dyn std::error::Error>> {
195 : use std::io::{Read, Write};
196 6 : let port = std::env::var("ANTARES_HTTP_PORT").unwrap_or_else(|_| "9090".into());
197 6 : let timeout = std::time::Duration::from_secs(3);
198 6 : let addr = format!("127.0.0.1:{port}").parse::<std::net::SocketAddr>()?;
199 6 : let mut s = std::net::TcpStream::connect_timeout(&addr, timeout)?;
200 2 : s.set_read_timeout(Some(timeout))?;
201 2 : s.write_all(b"GET /q/health HTTP/1.0\r\nHost: localhost\r\n\r\n")?;
202 2 : let mut head = [0u8; 16];
203 2 : s.read_exact(&mut head)?;
204 2 : if head.starts_with(b"HTTP/1.1 200") || head.starts_with(b"HTTP/1.0 200") {
205 2 : Ok(())
206 : } else {
207 0 : Err(format!("health: {}", String::from_utf8_lossy(&head)).into())
208 : }
209 6 : }
210 :
211 65 : fn main() -> Result<(), Box<dyn std::error::Error>> {
212 : // --version answers without starting anything (a bare `antares
213 : // --version` used to boot a server).
214 : // `args()`/`vars()` PANIC on non-UTF-8; the *_os variants do not, and a
215 : // stray byte in the environment or argv must not kill the process.
216 73 : if std::env::args_os().any(|a| a == "--version" || a == "-V") {
217 0 : println!(
218 : "antares {} ({})",
219 : env!("CARGO_PKG_VERSION"),
220 : antares_api::GIT_HASH
221 : );
222 0 : return Ok(());
223 65 : }
224 73 : if std::env::args_os().any(|a| a == "--help" || a == "-h") {
225 2 : println!(
226 : "antares {} — NGSI-LD context broker (ETSI GS CIM 009 V1.9.1)\n\n\
227 : Usage: antares [--version | --health | --help]\n\n\
228 : Configuration is environment only; every accepted key is listed\n\
229 : below and documented with its default in docs/src/configuration.md\n\
230 : (an unknown ANTARES_* key is fatal at startup).\n",
231 : env!("CARGO_PKG_VERSION")
232 : );
233 88 : for key in KNOWN_KEYS {
234 88 : println!(" {key}");
235 88 : }
236 2 : return Ok(());
237 63 : }
238 : // --health is the container health probe: the image has no shell or
239 : // curl, so the binary asks its own /q/health and exits 0 only on 200.
240 69 : if std::env::args_os().any(|a| a == "--health") {
241 6 : return health_probe();
242 57 : }
243 : // reqwest is built provider-less, so the FIRST client anything in this
244 : // process builds decides whether it panics — and the OTLP exporter
245 : // builds one inside `opentelemetry-http`, before any broker code runs.
246 : // Installed here, ahead of telemetry, rather than only in
247 : // `client_builder`, which a dependency does not call.
248 57 : antares_jsonld::install_crypto_provider();
249 :
250 : // Tracing (fmt + env-gated OTLP [+ console feature]) and, with the
251 : // `telemetry` feature, the Prometheus recorder rendering /q/metrics.
252 57 : let metrics_render = telemetry::init()?;
253 :
254 : // Unknown-config-is-fatal: catch typos before they become Scorpio's
255 : // $[quarkus.uuid} class of silent misconfiguration. ANTARES_TEST_* is the
256 : // reserved harness namespace (ANTARES_TEST_DATABASE_URL, ANTARES_TEST_MQTT_URL,
257 : // …) — CI exports those for the integration tests, and they land in the env of
258 : // any broker a test spawns. Reserving the prefix here beats making every
259 : // spawn site remember an env_remove allowlist.
260 8198 : for (key, _) in std::env::vars_os() {
261 8198 : let key = key.to_string_lossy();
262 8198 : if unknown_config_key(&key) {
263 2 : return Err(format!("unknown config key {key} (known: {KNOWN_KEYS:?})").into());
264 8196 : }
265 : }
266 :
267 55 : let port_raw = std::env::var("ANTARES_HTTP_PORT").unwrap_or_else(|_| "9090".into());
268 55 : let port: u16 = port_raw.parse().map_err(|e| {
269 4 : format!("ANTARES_HTTP_PORT must be a port number 0-65535, got {port_raw:?} ({e})")
270 4 : })?;
271 : // Every remaining config value is parsed HERE, before the runtime starts,
272 : // so a garbage window, cadence or switch fails the process instead of
273 : // silently running at its default.
274 51 : let sweep_secs = parse_sweep_secs(std::env::var("ANTARES_SWEEP_SECS").ok().as_deref())?;
275 : // Same: validated here, read again where the state is built.
276 47 : antares_api::DeliveryPolicy::from_env()?;
277 47 : let drain_delay = shutdown::drain_delay()?;
278 45 : let drain_deadline = shutdown::drain_deadline()?;
279 : // Validated here so a typo fails startup; the value itself is read again
280 : // where the drain task is wired, which is the only place it is used.
281 43 : wiring::outbox_drain_enabled()?;
282 41 : let host_alias = std::env::var("ANTARES_HOST_ALIAS").unwrap_or_else(|_| "antares".into());
283 : // 6.3.18 sends this as the Via pseudonym — an RFC 7230 token. `~` is
284 : // reserved as the tenant separator (federation::alias_for), so allowing
285 : // it in the configured alias would let `a~b` in the default tenant
286 : // collide with `a` in tenant `b` and cross-detect as a loop. Fatal at
287 : // startup, like every other bad config value.
288 41 : if host_alias.is_empty()
289 39 : || !host_alias
290 39 : .bytes()
291 263 : .all(|b| b.is_ascii_alphanumeric() || b"!#$%&'*+-.^_`|".contains(&b))
292 : {
293 4 : return Err(format!(
294 4 : "ANTARES_HOST_ALIAS {host_alias:?} is not a valid RFC 7230 token \
295 4 : (and may not contain '~', the tenant separator)"
296 4 : )
297 4 : .into());
298 37 : }
299 37 : let roles = std::env::var("ANTARES_ROLES").unwrap_or_else(|_| "all".into());
300 : // Unknown store backend is fatal BEFORE the runtime spins up, and the
301 : // message names the shelf this binary was built with — never a silent
302 : // fallback to memory.
303 37 : let store_name = std::env::var("ANTARES_STORE").unwrap_or_else(|_| "memory".into());
304 37 : if !store_shelf().contains(&store_name.as_str()) {
305 4 : return Err(format!(
306 4 : "unknown ANTARES_STORE {store_name:?}; built with {}",
307 4 : store_shelf().join("|")
308 4 : )
309 4 : .into());
310 33 : }
311 : // bus=local wires an in-process matcher into every process, so N
312 : // replicas over ONE shared database each fire their own copy of every
313 : // notification. Refused here — before any store connection is attempted
314 : // — unless the deployment states it runs exactly one broker process.
315 : // (Mirror of the nats arm's store check in `run`.)
316 33 : let bus_mode = std::env::var("ANTARES_BUS").unwrap_or_else(|_| "local".into());
317 33 : if bus_mode == "local"
318 31 : && shared_state(&store_name)
319 6 : && !std::env::var("ANTARES_ALLOW_SHARED_LOCAL")
320 6 : .is_ok_and(|v| matches!(v.as_str(), "1" | "true"))
321 : {
322 2 : return Err(format!(
323 2 : "ANTARES_BUS=local with ANTARES_STORE={store_name} double-fires notifications when \
324 2 : more than one broker process shares the database (each process runs its own \
325 2 : matcher). Use ANTARES_BUS=nats, or set ANTARES_ALLOW_SHARED_LOCAL=1 for a \
326 2 : strictly single-process deployment"
327 2 : )
328 2 : .into());
329 31 : }
330 :
331 31 : runtime()?.block_on(async {
332 31 : let drivers = build_drivers(&store_name).await?;
333 24 : run(
334 24 : port,
335 24 : host_alias,
336 24 : roles,
337 24 : drivers.store,
338 24 : drivers.temporal,
339 24 : store_name,
340 24 : drivers.temporal_name,
341 24 : drivers.maintenance,
342 24 : metrics_render,
343 24 : sweep_secs,
344 24 : drain_delay,
345 24 : drain_deadline,
346 24 : )
347 24 : .await
348 31 : })
349 65 : }
350 :
351 : /// ANTARES_PG_STATEMENT_TIMEOUT_MS → the per-session `statement_timeout`
352 : /// every pooled connection carries (a runaway query is cancelled, 5.5.2
353 : /// InternalError); absent = 30 000; not a positive integer = fatal.
354 11 : fn parse_pg_statement_timeout(raw: Option<&str>) -> Result<std::time::Duration, String> {
355 11 : match raw {
356 5 : None => Ok(std::time::Duration::from_secs(30)),
357 6 : Some(v) => match v.parse::<u64>() {
358 4 : Ok(n) if n > 0 => Ok(std::time::Duration::from_millis(n)),
359 4 : _ => Err(format!(
360 4 : "ANTARES_PG_STATEMENT_TIMEOUT_MS must be a positive integer (got {v:?})"
361 4 : )),
362 : },
363 : }
364 11 : }
365 :
366 : /// ANTARES_PG_POOL → pool size: absent defaults to 20; anything that is not
367 : /// a positive integer is fatal (a misread size must never silently run with
368 : /// a default, matching the unknown-key policy).
369 13 : fn parse_pg_pool(raw: Option<&str>) -> Result<u32, String> {
370 13 : match raw {
371 5 : None => Ok(20),
372 8 : Some(v) => match v.parse::<u32>() {
373 4 : Ok(n) if n > 0 => Ok(n),
374 6 : _ => Err(format!(
375 6 : "ANTARES_PG_POOL must be a positive integer (got {v:?})"
376 6 : )),
377 : },
378 : }
379 13 : }
380 :
381 : /// What ANTARES_TEMPORAL resolved to, before anything is built.
382 : #[derive(Debug, PartialEq, Eq)]
383 : enum TemporalChoice {
384 : /// The current-state store records and serves history too (default).
385 : SameAsStore,
386 : /// History off: `NoTemporal`.
387 : None,
388 : /// A second store instance of this backend, used only through its
389 : /// temporal half.
390 : Second(String),
391 : }
392 :
393 : /// The current-state backends this binary can build: the built-ins, plus
394 : /// any driver compiled in from outside this workspace. A list of NAMES and
395 : /// not an enum, because a plugin's driver is not one of `StoreMode`'s arms
396 : /// — adding a backend must never mean editing a core crate.
397 58 : fn store_shelf() -> Vec<&'static str> {
398 : // Backends from outside this workspace, in selection order. Each is
399 : // compiled in by its own feature; without one the list is empty and the
400 : // shelf is exactly `StoreMode`.
401 : const PLUGINS: &[&str] = &[
402 : #[cfg(feature = "plugin-example")]
403 : antares_plugin_example::NAME,
404 : ];
405 : antares_sql::StoreMode::ALL
406 58 : .into_iter()
407 232 : .map(|m| m.as_str())
408 58 : .chain(PLUGINS.iter().copied())
409 58 : .collect()
410 58 : }
411 :
412 : /// Does this backend keep its state where several broker processes can
413 : /// reach it? The precondition for `ANTARES_BUS=nats`, and the reason
414 : /// `bus=local` over one shared database double-fires notifications. Only
415 : /// the database backends qualify: a per-process store — memory, file, or a
416 : /// driver from outside the workspace — never does.
417 41 : fn shared_state(name: &str) -> bool {
418 41 : name.parse::<antares_sql::StoreMode>()
419 41 : .is_ok_and(|m| m.is_pg())
420 41 : }
421 :
422 : /// The shelf this binary was built with, rendered from the backend list
423 : /// rather than spelled out: a backend reaches every message that names the
424 : /// shelf without a second edit, whether it came from `StoreMode` or from a
425 : /// crate outside this workspace.
426 8 : fn built_with() -> String {
427 8 : format!("{} (temporal also: none)", store_shelf().join("|"))
428 8 : }
429 :
430 : /// ANTARES_TEMPORAL → driver choice. Absent or the store's own mode = one
431 : /// instance for both seams; `none` = no history; any other backend name =
432 : /// a second store. An unknown name is fatal and names the shelf.
433 37 : fn temporal_choice(store_name: &str, raw: Option<&str>) -> Result<TemporalChoice, String> {
434 7 : match raw {
435 26 : None => Ok(TemporalChoice::SameAsStore),
436 11 : Some(m) if m == store_name => Ok(TemporalChoice::SameAsStore),
437 9 : Some("none") => Ok(TemporalChoice::None),
438 7 : Some(other) if store_shelf().contains(&other) => {
439 3 : Ok(TemporalChoice::Second(other.to_owned()))
440 : }
441 4 : Some(other) => Err(format!(
442 4 : "ANTARES_TEMPORAL: unknown backend {other:?}; built with {}",
443 4 : built_with()
444 4 : )),
445 : }
446 37 : }
447 :
448 : /// One entry of the surface shelf: the name a deployment selects it with,
449 : /// and how to build it.
450 : type SurfaceCtor = fn() -> Box<dyn antares_api::ApiSurface>;
451 :
452 : /// The HTTP surfaces this binary was built with, outside the NGSI-LD API
453 : /// root. Every one is compiled in, so the shelf is static; a feature-gated
454 : /// surface would drop out here.
455 : const SURFACE_SHELF: &[(&str, SurfaceCtor)] = &[
456 12 : ("admin", || Box::new(antares_api::Admin)),
457 : #[cfg(feature = "plugin-example")]
458 : ("example", || {
459 : Box::new(antares_plugin_example::ExampleSurface)
460 : }),
461 : ];
462 :
463 : /// ANTARES_API_SURFACES → the surfaces mounted beside the NGSI-LD API,
464 : /// comma-separated; absent = `admin`. An unknown name is fatal and names
465 : /// the shelf, and so is a selection that ends up empty: health, readiness
466 : /// and metrics are the admin surface, and a pod that serves none of them
467 : /// can never report itself up.
468 28 : fn api_surfaces(raw: Option<&str>) -> Result<Vec<(&'static str, SurfaceCtor)>, String> {
469 28 : let mut chosen = Vec::new();
470 28 : for name in raw
471 28 : .unwrap_or("admin")
472 28 : .split(',')
473 28 : .map(str::trim)
474 36 : .filter(|n| !n.is_empty())
475 : {
476 22 : let entry = SURFACE_SHELF
477 22 : .iter()
478 22 : .find(|(n, _)| *n == name)
479 22 : .ok_or_else(|| {
480 2 : let shelf: Vec<&str> = SURFACE_SHELF.iter().map(|(n, _)| *n).collect();
481 2 : format!(
482 : "ANTARES_API_SURFACES: unknown surface {name:?}; built with {}",
483 2 : shelf.join("|")
484 : )
485 2 : })?;
486 20 : chosen.push(*entry);
487 : }
488 26 : if chosen.is_empty() {
489 8 : return Err(
490 8 : "ANTARES_API_SURFACES selects no surface; /q (health, readiness, metrics) \
491 8 : is the admin surface"
492 8 : .into(),
493 8 : );
494 18 : }
495 18 : Ok(chosen)
496 28 : }
497 :
498 : /// One entry of the policy shelf: the name a deployment selects the engine
499 : /// with, and how to build it.
500 : type PolicyCtor = fn() -> std::sync::Arc<dyn antares_api::policy::PolicyEngine>;
501 :
502 : /// The policy engines this binary was built with (ADR-0020). `allow-all` is
503 : /// the one the broker ships and the one conformance is asserted against; an
504 : /// engine from outside this workspace joins the list behind its own
505 : /// off-by-default feature, exactly as a store or a surface does, and is
506 : /// absent from a release build.
507 : const POLICY_SHELF: &[(&str, PolicyCtor)] = &[
508 0 : ("allow-all", || {
509 0 : std::sync::Arc::new(antares_api::policy::AllowAll)
510 0 : }),
511 : #[cfg(feature = "plugin-example")]
512 : (antares_plugin_example::POLICY_NAME, || {
513 : std::sync::Arc::new(antares_plugin_example::ExamplePolicy::from_env())
514 : }),
515 : ];
516 :
517 : /// ANTARES_POLICY → the engine every operation is asked about; absent =
518 : /// `allow-all`. An unknown name is fatal and names the shelf: a deployment
519 : /// that meant to run its engine and got a typo would otherwise serve every
520 : /// request wide open and never know.
521 20 : fn policy_engine(raw: Option<&str>) -> Result<(&'static str, PolicyCtor), String> {
522 20 : let name = raw
523 20 : .map(str::trim)
524 20 : .filter(|n| !n.is_empty())
525 20 : .unwrap_or("allow-all");
526 20 : POLICY_SHELF
527 20 : .iter()
528 20 : .find(|(n, _)| *n == name)
529 20 : .copied()
530 20 : .ok_or_else(|| {
531 2 : let shelf: Vec<&str> = POLICY_SHELF.iter().map(|(n, _)| *n).collect();
532 2 : format!(
533 : "ANTARES_POLICY: unknown policy engine {name:?}; built with {}",
534 2 : shelf.join("|")
535 : )
536 2 : })
537 20 : }
538 :
539 : /// The backend registry: the two driver seams from their configured names.
540 : /// Every backend is one arm of `build_store`; the temporal driver is by
541 : /// default the same instance (history recorded and served by the
542 : /// current-state store), `none` turns history off (temporal reads answer
543 : /// OperationNotSupported 422, Table 6.3.2-1; the recorder produces
544 : /// nothing), and a different backend name builds a second store used only
545 : /// through its temporal half.
546 31 : async fn build_drivers(store_name: &str) -> Result<Drivers, Box<dyn std::error::Error>> {
547 31 : let built = build_store(store_name, false).await?;
548 : // Every pg half gets the maintenance job: partitions, retention and the
549 : // 4.22 reap belong to whichever database holds the history.
550 25 : let mut maintenance = Vec::from_iter(built.maintenance);
551 25 : let raw = std::env::var("ANTARES_TEMPORAL").ok();
552 25 : let (temporal, temporal_name) = match temporal_choice(store_name, raw.as_deref())? {
553 : TemporalChoice::SameAsStore => {
554 24 : let name = built.temporal.supported().then(|| store_name.to_owned());
555 24 : (built.temporal.clone(), name)
556 : }
557 0 : TemporalChoice::None => (
558 0 : std::sync::Arc::new(antares_store::NoTemporal)
559 0 : as std::sync::Arc<dyn antares_store::TemporalDriver>,
560 0 : None,
561 0 : ),
562 1 : TemporalChoice::Second(name) => {
563 1 : let second = build_store(&name, true).await?;
564 0 : maintenance.extend(second.maintenance);
565 0 : (second.temporal, Some(name))
566 : }
567 : };
568 24 : Ok(Drivers {
569 24 : store: built.store,
570 24 : temporal,
571 24 : maintenance,
572 24 : temporal_name,
573 24 : })
574 31 : }
575 :
576 : /// The two storage seams a running broker holds, and what the maintenance
577 : /// job and `/q/health` need to know about them.
578 : struct Drivers {
579 : store: std::sync::Arc<dyn antares_store::CurrentStateDriver>,
580 : temporal: std::sync::Arc<dyn antares_store::TemporalDriver>,
581 : maintenance: Vec<(
582 : antares_sql::sqlx::PgPool,
583 : antares_sql::store::pg::maintenance::TemporalBackend,
584 : )>,
585 : /// What `/q/health` calls the history backend; `None` = history off.
586 : temporal_name: Option<String>,
587 : }
588 :
589 : /// One built backend: the two driver seams of a single store instance, and
590 : /// the Postgres handles its maintenance job needs.
591 : struct Built {
592 : store: std::sync::Arc<dyn antares_store::CurrentStateDriver>,
593 : temporal: std::sync::Arc<dyn antares_store::TemporalDriver>,
594 : maintenance: Option<(
595 : antares_sql::sqlx::PgPool,
596 : antares_sql::store::pg::maintenance::TemporalBackend,
597 : )>,
598 : }
599 :
600 : /// One backend by name. `temporal_only` builds an instance whose
601 : /// current-state half is never served — the second store of a split
602 : /// `ANTARES_TEMPORAL`. An unknown name is an error naming the shelf.
603 32 : async fn build_store(name: &str, temporal_only: bool) -> Result<Built, Box<dyn std::error::Error>> {
604 : #[cfg(feature = "plugin-example")]
605 : if name == antares_plugin_example::NAME {
606 : // A driver from outside `crates/` mounts through exactly the two
607 : // trait objects a built-in does; nothing downstream can tell them
608 : // apart, and its history half is its own — no `temporal_only` flag
609 : // to set, because the current-state half is simply never asked.
610 : let store = std::sync::Arc::new(antares_plugin_example::ExampleStore::new());
611 : return Ok(Built {
612 : store: store.clone(),
613 : temporal: store,
614 : maintenance: None,
615 : });
616 : }
617 32 : let mode: antares_sql::StoreMode = name.parse().map_err(|_| {
618 0 : format!(
619 : "unknown store backend {name:?}; built with {}",
620 0 : store_shelf().join("|")
621 : )
622 0 : })?;
623 32 : let (store, backend) = build_builtin(mode).await?;
624 25 : let store = if temporal_only {
625 0 : store.temporal_only()
626 : } else {
627 25 : store
628 : };
629 25 : let maintenance = match (&store, backend) {
630 2 : (antares_sql::store::any::AnyStore::Pg(p), Some(backend)) => {
631 2 : Some((p.docs.pool().clone(), backend))
632 : }
633 23 : _ => None,
634 : };
635 25 : let store = std::sync::Arc::new(store);
636 25 : Ok(Built {
637 25 : store: store.clone(),
638 25 : temporal: store,
639 25 : maintenance,
640 25 : })
641 32 : }
642 :
643 : /// ANTARES_STORE → store construction: `file` requires ANTARES_DATA_DIR
644 : /// (never a default inside the image); postgres and timescale require
645 : /// ANTARES_DATABASE_URL, connect ONE shared pool, run the embedded
646 : /// migrations at start and serve from the Pg backend.
647 32 : async fn build_builtin(
648 32 : mode: antares_sql::StoreMode,
649 32 : ) -> Result<
650 32 : (
651 32 : antares_sql::store::any::AnyStore,
652 32 : Option<antares_sql::store::pg::maintenance::TemporalBackend>,
653 32 : ),
654 32 : Box<dyn std::error::Error>,
655 32 : > {
656 : use antares_sql::store::any::{AnyStore, PgBackend};
657 : use antares_sql::store::pg::maintenance::TemporalBackend;
658 : use antares_sql::store::Store;
659 : use antares_sql::StoreMode;
660 32 : match mode {
661 21 : StoreMode::Memory => Ok((AnyStore::Mem(Store::default()), None)),
662 : StoreMode::File => {
663 6 : let dir = std::env::var("ANTARES_DATA_DIR").map_err(|_| {
664 2 : "ANTARES_STORE=file requires ANTARES_DATA_DIR (a mounted volume — data \
665 2 : must never live inside the image)"
666 2 : })?;
667 4 : let dir = std::path::PathBuf::from(dir);
668 4 : warn_if_not_mount_point(&dir);
669 4 : Ok((AnyStore::Mem(Store::open_file(&dir)?), None))
670 : }
671 : StoreMode::Postgres | StoreMode::Timescale => {
672 5 : let url = std::env::var("ANTARES_DATABASE_URL")
673 5 : .map_err(|_| format!("ANTARES_STORE={mode} requires ANTARES_DATABASE_URL"))?;
674 3 : let pool_size = parse_pg_pool(std::env::var("ANTARES_PG_POOL").ok().as_deref())?;
675 3 : let statement_timeout = parse_pg_statement_timeout(
676 3 : std::env::var("ANTARES_PG_STATEMENT_TIMEOUT_MS")
677 3 : .ok()
678 3 : .as_deref(),
679 0 : )?;
680 : // The DB container may still be booting — bounded retry, then die.
681 3 : let mut last = String::new();
682 3 : for _ in 0..30 {
683 3 : match antares_sql::store::pg::connect_with(&url, pool_size, statement_timeout).await
684 : {
685 3 : Ok(pool) => {
686 : // The temporal backend is what the migrations actually
687 : // BUILT, detected once from the catalog and pinned —
688 : // the maintenance branch can never disagree with the
689 : // DDL on disk, whatever happened to the extension since.
690 3 : let backend =
691 3 : antares_sql::store::pg::maintenance::detect_temporal_backend(&pool)
692 3 : .await
693 3 : .map_err(|e| format!("ANTARES_STORE={mode}: {e}"))?;
694 : // Never silently fall back — timescale mode whose
695 : // database is not hypertable-shaped is a config error,
696 : // not a downgrade (extension missing at first boot, or
697 : // installed only after the migrations ran).
698 3 : if mode == StoreMode::Timescale && backend != TemporalBackend::Hypertable {
699 1 : return Err(format!(
700 1 : "timescale requested (ANTARES_STORE or ANTARES_TEMPORAL) but \
701 1 : attr_instances is {backend:?} — the timescaledb extension was \
702 1 : not CREATEd when the migrations first ran. Install it in a fresh \
703 1 : database (CREATE EXTENSION timescaledb before first boot) or use \
704 1 : postgres"
705 1 : )
706 1 : .into());
707 2 : }
708 2 : if mode == StoreMode::Postgres && backend == TemporalBackend::Hypertable {
709 0 : tracing::info!(
710 : "attr_instances is a hypertable (migrations ran with \
711 : the timescaledb extension present); the plain-mode partition \
712 : job stands down, retention runs via drop_chunks"
713 : );
714 2 : }
715 : // RLS is a belt only when the role wears it —
716 : // superuser/BYPASSRLS makes every policy inert. Warn
717 : // always; in production set ANTARES_REQUIRE_RLS=1 to turn
718 : // the warning into a hard refusal so a superuser DSN can
719 : // never silently ship (dev/ETSI stacks leave it unset).
720 2 : if antares_sql::store::pg::role_bypasses_rls(&pool).await {
721 : // A gate that only understands two spellings
722 : // fails OPEN on `TRUE`/`yes`/`on`: the operator
723 : // believes RLS is enforced and the broker serves
724 : // with a BYPASSRLS role. Anything but an explicit
725 : // off value turns it on.
726 2 : let strict =
727 2 : std::env::var("ANTARES_REQUIRE_RLS").is_ok_and(|v| !is_off(&v));
728 2 : if strict {
729 0 : return Err(
730 0 : "ANTARES_REQUIRE_RLS=1 but the database role bypasses \
731 0 : row-level security (superuser or BYPASSRLS) — connect as a \
732 0 : non-superuser, non-BYPASSRLS role so the RLS tenant-isolation \
733 0 : backstop is enforced"
734 0 : .into(),
735 0 : );
736 2 : }
737 2 : tracing::warn!(
738 : "database role bypasses row-level security (superuser or \
739 : BYPASSRLS) — tenant isolation rests on the explicit \
740 : predicates only; use a non-superuser role in production \
741 : (set ANTARES_REQUIRE_RLS=1 to enforce)"
742 : );
743 0 : }
744 2 : tracing::info!(
745 : "ANTARES_STORE={mode}: pool up, migrations applied, serving \
746 : from postgres (temporal backend: {backend:?})"
747 : );
748 : // Read once, here: /q/health is polled, so what it
749 : // says about the server is captured at startup and
750 : // never queried on the request.
751 2 : let version = antares_sql::store::pg::version_info(&pool).await;
752 2 : return Ok((
753 2 : AnyStore::Pg(PgBackend::new(pool).with_version(version)),
754 2 : Some(backend),
755 2 : ));
756 : }
757 0 : Err(e) if antares_sql::store::pg::is_schema_mismatch(&e) => {
758 : // Waiting cannot make this database match: its
759 : // migration history was written by a different
760 : // release. Retried, the boot ends 30 s later blaming
761 : // the network for a schema that will never fit.
762 0 : return Err(format!(
763 0 : "ANTARES_STORE={mode}: {e} — the database was migrated by a \
764 0 : different release of the broker, so this binary cannot serve \
765 0 : it. Point it at a database migrated by this release, or start \
766 0 : from an empty one"
767 0 : )
768 0 : .into());
769 : }
770 0 : Err(e) => {
771 0 : last = e.to_string();
772 0 : tokio::time::sleep(std::time::Duration::from_secs(1)).await;
773 : }
774 : }
775 : }
776 0 : Err(format!("ANTARES_STORE={mode}: database not reachable after 30 s: {last}").into())
777 : }
778 : }
779 32 : }
780 :
781 : /// Warn when the data dir shares a device with its parent — i.e. it is
782 : /// not a mount point, so the redb file dies with the container.
783 : #[cfg(unix)]
784 4 : fn warn_if_not_mount_point(dir: &std::path::Path) {
785 : use std::os::unix::fs::MetadataExt;
786 4 : let _ = std::fs::create_dir_all(dir);
787 4 : if let (Ok(md), Some(Ok(parent_md))) =
788 4 : (std::fs::metadata(dir), dir.parent().map(std::fs::metadata))
789 : {
790 4 : if md.dev() == parent_md.dev() {
791 4 : eprintln!(
792 4 : "WARN: ANTARES_DATA_DIR {} is not a mount point — data will be lost when \
793 4 : the container is removed",
794 4 : dir.display()
795 4 : );
796 4 : }
797 0 : }
798 4 : }
799 :
800 : #[cfg(not(unix))]
801 : fn warn_if_not_mount_point(_dir: &std::path::Path) {}
802 :
803 : // Every parameter is one config value `main` parsed fatally before the
804 : // runtime started; a struct would only rename the same ten values.
805 : #[allow(clippy::too_many_arguments)]
806 24 : async fn run(
807 24 : port: u16,
808 24 : host_alias: String,
809 24 : roles: String,
810 24 : store: std::sync::Arc<dyn antares_store::CurrentStateDriver>,
811 24 : temporal: std::sync::Arc<dyn antares_store::TemporalDriver>,
812 24 : store_name: String,
813 24 : temporal_name: Option<String>,
814 24 : maintenance: Vec<(
815 24 : antares_sql::sqlx::PgPool,
816 24 : antares_sql::store::pg::maintenance::TemporalBackend,
817 24 : )>,
818 24 : metrics_render: Option<telemetry::MetricsRender>,
819 24 : sweep_secs: u64,
820 24 : drain_delay: std::time::Duration,
821 24 : drain_deadline: std::time::Duration,
822 24 : ) -> Result<(), Box<dyn std::error::Error>> {
823 24 : let roles = wiring::Roles::parse(&roles).map_err(|e| format!("ANTARES_ROLES: {e}"))?;
824 : // Bus seam: local (default) or nats. An unknown value is fatal.
825 22 : let bus_mode = std::env::var("ANTARES_BUS").unwrap_or_else(|_| "local".into());
826 22 : match bus_mode.as_str() {
827 22 : "local" => {
828 : // bus=local means ONE process running every role — a role
829 : // split without a shared bus would silently drop whole concerns.
830 20 : if !roles.all() {
831 8 : return Err(
832 8 : "ANTARES_BUS=local requires all roles in one process (ANTARES_ROLES=all); \
833 8 : role splits need ANTARES_BUS=nats"
834 8 : .into(),
835 8 : );
836 12 : }
837 : }
838 2 : "nats" => {
839 0 : if !shared_state(&store_name) {
840 0 : return Err(format!(
841 0 : "ANTARES_BUS=nats requires a shared store (ANTARES_STORE=postgres|timescale); \
842 0 : {store_name} state is per-process and cannot back multiple instances"
843 0 : )
844 0 : .into());
845 0 : }
846 : }
847 2 : other => return Err(format!("unknown ANTARES_BUS={other} (local|nats)").into()),
848 : }
849 12 : tracing::info!(port, store = %store_name, %bus_mode, ?roles, "starting antares");
850 :
851 : // Trailing-slash tolerance: Table 6.2-1 spells collection resources with a
852 : // trailing '/'; normalize before routing.
853 12 : let mut state = AppState::with_drivers(host_alias, store, temporal, &store_name);
854 12 : state.temporal_name = temporal_name;
855 12 : state.delivery = antares_api::DeliveryPolicy::from_env().unwrap_or_default();
856 12 : state.temporal_record = std::env::var("ANTARES_TEMPORAL_RECORD")
857 12 : .as_deref()
858 12 : .unwrap_or("all")
859 12 : .parse()?;
860 : // The surfaces mounted beside the NGSI-LD API root, from configuration
861 : // rather than from a hard-wired list: the selection replaces what the
862 : // default mounting put there, before the state is shared.
863 12 : let selected = api_surfaces(std::env::var("ANTARES_API_SURFACES").ok().as_deref())?;
864 12 : state = state.with_surfaces(selected.iter().map(|(_, build)| build()).collect())?;
865 : // The policy engine, from configuration and from the shelf this binary
866 : // was built with. Without one the broker asks `allow-all` and behaves
867 : // exactly as it did before the seam existed.
868 12 : let (policy_name, build_policy) =
869 12 : policy_engine(std::env::var("ANTARES_POLICY").ok().as_deref())?;
870 12 : tracing::info!("policy engine: {policy_name}");
871 : // The built-in engine is attached as no engine at all: it decides
872 : // nothing, and a gate with nothing to ask skips the whole apparatus
873 : // rather than boxing a future that always answers allow.
874 12 : if policy_name != antares_api::policy::BUILT_IN_NAME {
875 0 : state = state.with_policy(build_policy());
876 12 : }
877 : // A notification binding compiled in from outside the workspace mounts
878 : // exactly like the two shipped ones: one registration, no core-crate
879 : // edit. It is not in a release build — the feature is off by default,
880 : // so the shipped binary serves network schemes only.
881 : #[cfg(feature = "plugin-example")]
882 : let mut state = state.with_sink(Box::new(antares_plugin_example::MemorySink::new()));
883 : // /q/metrics renders through this closure (None without the
884 : // `telemetry` feature — the endpoint answers 404); the sampler feeds
885 : // the process-level gauges the whole run.
886 12 : state.metrics_render = metrics_render;
887 : // Heap stats on /q/health (allocated/resident bytes via jemalloc-ctl)
888 16 : state.mem_stats = Some(std::sync::Arc::new(|| {
889 : use tikv_jemalloc_ctl::{epoch, stats};
890 16 : let _ = epoch::advance();
891 16 : serde_json::json!({
892 16 : "allocatedBytes": stats::allocated::read().unwrap_or(0),
893 16 : "residentBytes": stats::resident::read().unwrap_or(0),
894 : })
895 16 : }));
896 : // 5.8.1.4 consumer half, whichever bus carries the delivery. A
897 : // distributed Subscription is served by an internal Context Source
898 : // Registration Subscription that notifies to `urn:antares:distsub:…`,
899 : // and the delivery path drops that notification unless this handler is
900 : // installed — the Subscription is accepted and no copy is ever
901 : // forwarded. `antares_api::wire` installs it with the in-process
902 : // matcher; the role-split fleet wires its matcher through `wire_nats`
903 : // and needs it installed here.
904 12 : antares_api::install_csource_notification(&mut state);
905 12 : if bus_mode == "nats" {
906 : // Outbox producer + drain, KV/registry mirrors, durable
907 : // consumers per role, topology asserted before traffic.
908 0 : let url = std::env::var("ANTARES_NATS_URL")
909 0 : .map_err(|_| "ANTARES_BUS=nats requires ANTARES_NATS_URL")?;
910 0 : wiring::wire_nats(&mut state, &url, roles).await?;
911 : } else {
912 12 : antares_api::wire(&mut state).await; // in-process matcher + notifier + interval firing
913 : }
914 12 : telemetry::spawn_sampler(state.clone());
915 :
916 : // Boot preload — Cached rows persisted by the AppState write-through
917 : // re-seed the parsed-context cache on start, so expansion doesn't refetch
918 : // what a previous life already downloaded. (The writer itself is wired in
919 : // AppState::with_store — rows are the 5.13 source of truth.)
920 : {
921 : // Metadata first, then ONE body at a time. Reading every row whole
922 : // put up to MAX_CONTEXT_BYTES per row in memory at once, capped only
923 : // for Cached rows and not at all for the rest — a boot that a client
924 : // could make impossible by storing large @contexts.
925 : // No Tenant: the store then hands back the rows that belong to none
926 : // (ADR-0021), which is exactly the `Cached` set this warm wants.
927 12 : for row in state
928 12 : .store
929 12 : .context_list_meta(None)
930 12 : .await
931 12 : .unwrap_or_default()
932 : {
933 0 : if row.get("kind").and_then(|v| v.as_str()) != Some("Cached") {
934 0 : continue;
935 0 : }
936 0 : let (Some(url), Some(id), Some(created)) = (
937 0 : row.get("url").and_then(|v| v.as_str()),
938 0 : row.get("localId").and_then(|v| v.as_str()),
939 0 : row.get("createdAt").and_then(|v| v.as_str()),
940 : ) else {
941 0 : continue;
942 : };
943 0 : let Some(full) = state.store.context_get(None, id).await.ok().flatten() else {
944 0 : continue;
945 : };
946 0 : if let Some(v) = full.pointer("/body/@context") {
947 0 : state.loader.seed_cached(url, id, created, v.clone()).await;
948 0 : }
949 : }
950 : }
951 :
952 : // 4.22 GC on the memory/file arm: reads already refuse expired entities;
953 : // this reaps them (spec-sanctioned lag). The Pg arm's sweep runs inside
954 : // the maintenance job below — one job per backend, mode-switched.
955 : // ANTARES_SWEEP_SECS paces 4.22 GC identically across ALL backends —
956 : // this loop and the Pg/Timescale maintenance job below both tick on it
957 : // (the ETSI stack runs at 2 s so transient TPs observe GC, not just the
958 : // read filter); parsed at startup, default 15 min. The loop runs for
959 : // every driver and asks it what it reaped: a backend whose GC lives
960 : // elsewhere — the Pg arm's is inside the maintenance job — answers 0.
961 12 : let sweeper = state.store.clone();
962 12 : let doc_sweeper = state.clone();
963 12 : tokio::spawn(async move {
964 12 : let mut tick = tokio::time::interval(std::time::Duration::from_secs(sweep_secs));
965 : loop {
966 788 : tick.tick().await;
967 776 : let n = sweeper.sweep_expired().await;
968 776 : if n > 0 {
969 4 : tracing::debug!("4.22 sweep reaped {n} expired entities");
970 772 : }
971 : // Registrations, Snapshots and EntityMaps carry their own
972 : // expiry, and every read already refuses one that has passed;
973 : // this is what frees the row behind it.
974 776 : let n = antares_api::sweep_expired_docs(&doc_sweeper).await;
975 776 : if n > 0 {
976 8 : tracing::debug!("expiry sweep reaped {n} expired documents");
977 768 : }
978 : }
979 : });
980 : // Temporal maintenance — plain-mode partition pre-creation and the
981 : // (opt-in) retention horizon, single-winner via SKIP LOCKED.
982 : // One job per pg half (current state and/or history), PINNED to the
983 : // backend detected at startup; memory and file halves get none.
984 12 : if !maintenance.is_empty() {
985 2 : let retention: Option<i64> = std::env::var("ANTARES_TEMPORAL_RETENTION_DAYS")
986 2 : .ok()
987 2 : .map(|v| {
988 0 : v.parse::<i64>()
989 0 : .map_err(|_| "ANTARES_TEMPORAL_RETENTION_DAYS must be an integer")
990 : // A zero/negative horizon inverts `now() - make_interval(days)`
991 : // and would reap all current + future history — a data-loss
992 : // footgun. Retention is opt-in; a bad value must not silently
993 : // delete. Cap at i32 too (bound as $1::int downstream).
994 0 : .and_then(|d| {
995 0 : (d > 0 && d <= i64::from(i32::MAX)).then_some(d).ok_or(
996 0 : "ANTARES_TEMPORAL_RETENTION_DAYS must be between 1 and 2147483647",
997 : )
998 0 : })
999 0 : })
1000 2 : .transpose()?;
1001 2 : for (pool, backend) in maintenance {
1002 2 : tokio::spawn(async move {
1003 : // Same ANTARES_SWEEP_SECS cadence as the Mem arm — the job's 4.22
1004 : // reap is the sweep here; the partition/retention steps riding on
1005 : // the same tick are idempotent and SKIP LOCKED single-winner.
1006 2 : let mut tick = tokio::time::interval(std::time::Duration::from_secs(sweep_secs));
1007 : loop {
1008 396 : tick.tick().await; // first tick is immediate: partitions at boot
1009 394 : match antares_sql::store::pg::maintenance::temporal_maintenance(
1010 394 : &pool, backend, retention,
1011 : )
1012 394 : .await
1013 : {
1014 394 : Ok(msg) => tracing::debug!("temporal maintenance: {msg}"),
1015 0 : Err(e) => tracing::warn!("temporal maintenance failed: {e}"),
1016 : }
1017 : }
1018 : });
1019 : }
1020 10 : }
1021 : // Handles the drain needs, taken before `state` is consumed by the
1022 : // router — the flag the health endpoint reads, and the store whose pools
1023 : // close last.
1024 12 : let draining = state.draining.clone();
1025 12 : let store_for_drain = state.store.clone();
1026 : // The temporal seam may be a second store with its own pool; the drain
1027 : // closes both.
1028 12 : let temporal_for_drain = state.temporal.clone();
1029 12 : let pending_for_drain = state.pending_changes.clone();
1030 : // Only the api role serves the NGSI-LD surface — a worker pod
1031 : // exposes health/ready/metrics and nothing else (a subscription created
1032 : // on a worker would bypass the roles.api KV sync and never notify).
1033 12 : let routed = if roles.api {
1034 12 : antares_api::router(state)
1035 : } else {
1036 0 : antares_api::ops_router(state)
1037 : };
1038 12 : let app = tower::Layer::layer(
1039 12 : &tower_http::normalize_path::NormalizePathLayer::trim_trailing_slash(),
1040 12 : routed,
1041 : );
1042 :
1043 : // A connection that never finishes its request headers must not hold a
1044 : // slot forever; hyper closes it after this timeout.
1045 12 : let header_read_timeout = std::env::var("ANTARES_HEADER_READ_TIMEOUT_MS")
1046 12 : .ok()
1047 12 : .map(|v| {
1048 2 : v.parse::<u64>().map_err(|_| {
1049 2 : format!("ANTARES_HEADER_READ_TIMEOUT_MS must be an integer (got {v:?})")
1050 2 : })
1051 2 : })
1052 12 : .transpose()?
1053 10 : .map(std::time::Duration::from_millis)
1054 10 : .unwrap_or(std::time::Duration::from_secs(10));
1055 : // Ceiling on concurrently served connections: accepted streams beyond
1056 : // it are dropped at once — refusing cheaply beats queueing work the box
1057 : // cannot serve (each served connection is a spawned task).
1058 10 : let max_connections = max_connections()?;
1059 8 : let conn_permits = std::sync::Arc::new(tokio::sync::Semaphore::new(max_connections));
1060 :
1061 8 : let listener = tokio::net::TcpListener::bind(("0.0.0.0", port)).await?;
1062 8 : tracing::info!("listening on http://0.0.0.0:{port}");
1063 : // Count open connections so the drain can wait for them. Incremented
1064 : // before the task is spawned — incrementing inside the task would race the
1065 : // drain's first check and let a just-accepted connection be missed.
1066 : // This counts CONNECTIONS, not requests, and stays that way on purpose:
1067 : // hyper's graceful_shutdown below draws the distinction already (an idle
1068 : // keep-alive closes at once, an active request finishes first), so a
1069 : // request-layer counter would add a middleware without changing what the
1070 : // drain waits for.
1071 8 : let inflight = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
1072 : // The drain signal each connection listens for. On drain,
1073 : // hyper's graceful_shutdown closes IDLE keep-alive connections immediately
1074 : // (the LB holds one per backend — counting them as in-flight made every
1075 : // api roll burn the full deadline) while an active request still finishes.
1076 8 : let (drain_tx, drain_rx) = tokio::sync::watch::channel(false);
1077 : // The signal future is created ONCE and polled by reference. Written
1078 : // inline in the select, it would be dropped and re-created on every
1079 : // accepted connection — and a SIGTERM landing in that drop-to-recreate
1080 : // window is lost for good (tokio signal streams do not replay events from
1081 : // before their creation). Under health-check polling that window is hit
1082 : // constantly, which is exactly how the drain test caught it.
1083 8 : let mut sigterm = std::pin::pin!(shutdown::signal());
1084 : // A pod whose drain is switched off publishes nothing, so waiting for the
1085 : // outbox to empty there would only burn the deadline.
1086 8 : let flush_outbox = wiring::outbox_drain_enabled()?;
1087 : // Manual serve loop: the ETSI suite reads response headers case-sensitively
1088 : // ("Location"), so HTTP/1 responses are written with title-case headers.
1089 : loop {
1090 12865 : let stream = tokio::select! {
1091 12865 : s = accept(&listener) => s,
1092 12865 : _ = &mut sigterm => {
1093 : // 1+2: unhealthy FIRST, then keep serving for the LB's notice
1094 : // window — still inside this select, so connections arriving
1095 : // during it are accepted normally.
1096 8 : shutdown::begin(&draining, drain_delay);
1097 8 : let until = tokio::time::Instant::now() + drain_delay;
1098 : loop {
1099 12 : tokio::select! {
1100 12 : stream = accept(&listener) => {
1101 4 : serve(stream, app.clone(), inflight.clone(), drain_rx.clone(),
1102 4 : conn_permits.clone(), header_read_timeout);
1103 4 : }
1104 12 : _ = tokio::time::sleep_until(until) => break,
1105 : }
1106 : }
1107 : // 3–6: listener dropped, idle conns told to close (active
1108 : // requests finish), in-flight drained, pools closed.
1109 8 : drop(listener);
1110 8 : let _ = drain_tx.send(true);
1111 8 : shutdown::drain(
1112 8 : &inflight,
1113 8 : &pending_for_drain,
1114 8 : &*store_for_drain,
1115 8 : &*temporal_for_drain,
1116 8 : drain_deadline,
1117 8 : flush_outbox,
1118 8 : )
1119 8 : .await;
1120 8 : tracing::info!("shutting down");
1121 8 : return Ok(());
1122 : }
1123 : };
1124 12857 : serve(
1125 12857 : stream,
1126 12857 : app.clone(),
1127 12857 : inflight.clone(),
1128 12857 : drain_rx.clone(),
1129 12857 : conn_permits.clone(),
1130 12857 : header_read_timeout,
1131 : );
1132 : }
1133 24 : }
1134 :
1135 : /// ANTARES_MAX_CONNECTIONS: the ceiling on concurrently served connections
1136 : /// (default 10 000).
1137 10 : fn max_connections() -> Result<usize, Box<dyn std::error::Error>> {
1138 10 : Ok(std::env::var("ANTARES_MAX_CONNECTIONS")
1139 10 : .ok()
1140 10 : .map(|v| {
1141 2 : v.parse::<usize>().ok().filter(|n| *n > 0).ok_or_else(|| {
1142 2 : format!("ANTARES_MAX_CONNECTIONS must be a positive integer (got {v:?})")
1143 2 : })
1144 2 : })
1145 10 : .transpose()?
1146 8 : .unwrap_or(10_000))
1147 10 : }
1148 :
1149 : /// The request runtime. Store calls are futures the workers poll, so a
1150 : /// waiting caller holds no thread and the blocking pool carries only what
1151 : /// is genuinely blocking — the `file` mode's per-commit fsync, which redb
1152 : /// serializes behind its single writer. Tokio's own bounds hold.
1153 31 : fn runtime() -> std::io::Result<tokio::runtime::Runtime> {
1154 31 : tokio::runtime::Builder::new_multi_thread()
1155 31 : .enable_all()
1156 31 : .build()
1157 31 : }
1158 :
1159 : /// Accept the next connection. There is no failure value: `accept()`
1160 : /// propagates every non-`WouldBlock` errno from the syscall, and an
1161 : /// `ECONNABORTED` (a client resetting between SYN and accept), `EMFILE`/
1162 : /// `ENFILE` (the fd ceiling — the connection cap defaults above many
1163 : /// containers' `nofile`) or `ENOBUFS` must never take the broker down.
1164 12879 : async fn accept(listener: &tokio::net::TcpListener) -> tokio::net::TcpStream {
1165 : loop {
1166 12879 : match listener.accept().await {
1167 12863 : Ok((stream, _)) => {
1168 : // Nagle held every multi-segment response for the peer's
1169 : // delayed ACK: a fixed ~40 ms on any body over one segment,
1170 : // 25 req/s per connection whatever the core count.
1171 12863 : let _ = stream.set_nodelay(true);
1172 12863 : return stream;
1173 : }
1174 0 : Err(e) => {
1175 0 : tracing::warn!("accept failed, retrying: {e}");
1176 0 : if accept_backoff(&e) {
1177 0 : tokio::time::sleep(std::time::Duration::from_millis(100)).await;
1178 0 : }
1179 : }
1180 : }
1181 : }
1182 12863 : }
1183 :
1184 : /// Whether an `accept()` error warrants a pause before the next attempt.
1185 : /// The per-connection failures retry at once (the next connection is
1186 : /// unaffected); a resource exhaustion would otherwise spin the loop at full
1187 : /// speed until the pressure clears. Neither is fatal.
1188 16 : fn accept_backoff(e: &std::io::Error) -> bool {
1189 10 : !matches!(
1190 16 : e.kind(),
1191 : std::io::ErrorKind::ConnectionAborted
1192 : | std::io::ErrorKind::ConnectionReset
1193 : | std::io::ErrorKind::Interrupted
1194 : )
1195 16 : }
1196 :
1197 : /// The served app: the router under trailing-slash normalization.
1198 : type App = tower_http::normalize_path::NormalizePath<axum::Router>;
1199 :
1200 : /// One accepted connection. Split out of the accept loop so the drain's
1201 : /// notice window serves connections with identical behaviour, and so the
1202 : /// in-flight counter is incremented in exactly one place.
1203 12861 : fn serve(
1204 12861 : stream: tokio::net::TcpStream,
1205 12861 : app: App,
1206 12861 : inflight: std::sync::Arc<std::sync::atomic::AtomicUsize>,
1207 12861 : mut drain_rx: tokio::sync::watch::Receiver<bool>,
1208 12861 : conn_permits: std::sync::Arc<tokio::sync::Semaphore>,
1209 12861 : header_read_timeout: std::time::Duration,
1210 12861 : ) {
1211 : // Over the connection cap: drop the accepted stream immediately. The
1212 : // permit rides in the connection task, so the slot frees exactly when
1213 : // the connection ends — and never enters the inflight drain accounting.
1214 12861 : let Ok(permit) = conn_permits.try_acquire_owned() else {
1215 0 : return;
1216 : };
1217 12861 : inflight.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1218 12861 : tokio::spawn(async move {
1219 12861 : let _permit = permit;
1220 12861 : let svc = hyper::service::service_fn(move |req: hyper::Request<hyper::body::Incoming>| {
1221 12861 : let mut app = app.clone();
1222 12861 : async move { tower::Service::call(&mut app, req.map(axum::body::Body::new)).await }
1223 12861 : });
1224 12861 : let mut builder =
1225 12861 : hyper_util::server::conn::auto::Builder::new(hyper_util::rt::TokioExecutor::new());
1226 12861 : builder
1227 12861 : .http1()
1228 12861 : .timer(hyper_util::rt::TokioTimer::new())
1229 12861 : .header_read_timeout(header_read_timeout)
1230 12861 : .title_case_headers(true);
1231 12861 : let conn = builder.serve_connection(hyper_util::rt::TokioIo::new(stream), svc);
1232 12861 : let mut conn = std::pin::pin!(conn);
1233 : // On drain, close an IDLE keep-alive connection immediately —
1234 : // hyper finishes any active request first, then closes. Without this
1235 : // the LB's idle keep-alives count as in-flight and every roll waits
1236 : // out the entire drain deadline.
1237 12861 : tokio::select! {
1238 12861 : r = conn.as_mut() => { let _ = r; }
1239 12861 : _ = wait_drain(&mut drain_rx) => {
1240 2 : conn.as_mut().graceful_shutdown();
1241 2 : let _ = conn.as_mut().await;
1242 : }
1243 : }
1244 12861 : inflight.fetch_sub(1, std::sync::atomic::Ordering::Relaxed);
1245 12861 : });
1246 12861 : }
1247 :
1248 : /// Resolves when the drain signal fires; pends forever once the sender is
1249 : /// gone (the connection future then completes on its own in the select).
1250 12861 : async fn wait_drain(rx: &mut tokio::sync::watch::Receiver<bool>) {
1251 : loop {
1252 12858 : if *rx.borrow() {
1253 2 : return;
1254 12856 : }
1255 12856 : if rx.changed().await.is_err() {
1256 0 : std::future::pending::<()>().await;
1257 2 : }
1258 : }
1259 2 : }
1260 :
1261 : #[cfg(test)]
1262 : mod config_key_tests {
1263 : use super::unknown_config_key;
1264 :
1265 : /// Kubelet-injected service links must never be fatal; real typos
1266 : /// must stay fatal.
1267 : #[test]
1268 2 : fn kubelet_service_links_are_not_typos() {
1269 14 : for k in [
1270 2 : "ANTARES_PORT",
1271 2 : "ANTARES_PORT_9090_TCP",
1272 2 : "ANTARES_PORT_9090_TCP_ADDR",
1273 2 : "ANTARES_SERVICE_HOST",
1274 2 : "ANTARES_SERVICE_PORT",
1275 2 : "ANTARES_API_SERVICE_HOST",
1276 2 : "ANTARES_FILE_PORT_9090_TCP_PROTO",
1277 2 : ] {
1278 14 : assert!(!unknown_config_key(k), "{k} is platform-injected");
1279 : }
1280 14 : for k in [
1281 2 : "ANTARES_HTTP_PORT",
1282 2 : "ANTARES_STORE",
1283 2 : "ANTARES_TEST_ANYTHING",
1284 2 : "ANTARES_PG_POOL",
1285 2 : "ANTARES_ALLOW_SHARED_LOCAL",
1286 2 : "ANTARES_HEADER_READ_TIMEOUT_MS",
1287 2 : "ANTARES_MAX_CONNECTIONS",
1288 2 : ] {
1289 14 : assert!(!unknown_config_key(k), "{k} is known/reserved");
1290 : }
1291 2 : assert!(
1292 2 : unknown_config_key("ANTARES_STROE"),
1293 : "a real typo stays fatal"
1294 : );
1295 2 : assert!(
1296 2 : unknown_config_key("ANTARES_HTPT_PORT"),
1297 : "a typo'd *_PORT var is NOT a service link"
1298 : );
1299 2 : assert!(unknown_config_key("ANTARES_BOGUS_FLAG"));
1300 2 : }
1301 :
1302 : /// The exemption is narrow on purpose: only the exact kubelet shapes for
1303 : /// the Services this repo ships. Near-misses stay fatal, and a non-ANTARES
1304 : /// variable is never our business.
1305 : #[test]
1306 2 : fn the_service_link_exemption_does_not_over_reach() {
1307 10 : for k in [
1308 2 : "ANTARES_",
1309 2 : "ANTARES_PORTAL",
1310 2 : "ANTARES_SERVICEHOST",
1311 2 : "ANTARES_DB_PORT",
1312 2 : "ANTARES_WORKER_PROT",
1313 2 : ] {
1314 10 : assert!(unknown_config_key(k), "{k} must stay a fatal typo");
1315 : }
1316 10 : for k in ["PATH", "HOME", "antares_store", "ANTARE_STORE", ""] {
1317 10 : assert!(!unknown_config_key(k), "{k} is not broker config");
1318 : }
1319 2 : }
1320 :
1321 : /// Unknown keys are FATAL, so every ANTARES_* variable the workspace
1322 : /// actually reads has to be accepted — otherwise setting a documented
1323 : /// deployment knob is a CrashLoopBackOff and the knob cannot be used at
1324 : /// all. Scans the sources rather than restating the list, so the check
1325 : /// keeps holding as crates add knobs.
1326 : #[test]
1327 2 : fn known_keys_cover_every_variable_the_workspace_reads() {
1328 2 : let crates = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
1329 2 : .parent()
1330 2 : .expect("crates dir")
1331 2 : .to_path_buf();
1332 2 : let needles = [
1333 2 : concat!("var(", "\"ANTARES_"),
1334 2 : concat!("var_os(", "\"ANTARES_"),
1335 2 : ];
1336 2 : let mut missing: Vec<String> = Vec::new();
1337 2 : let mut stack = vec![crates];
1338 80 : while let Some(dir) = stack.pop() {
1339 530 : for entry in std::fs::read_dir(&dir).expect("read_dir").flatten() {
1340 530 : let path = entry.path();
1341 530 : if path.is_dir() {
1342 76 : stack.push(path);
1343 76 : continue;
1344 454 : }
1345 454 : if path.extension().is_none_or(|e| e != "rs") {
1346 46 : continue;
1347 408 : }
1348 408 : let src = std::fs::read_to_string(&path).unwrap_or_default();
1349 816 : for needle in needles {
1350 816 : for (i, _) in src.match_indices(needle) {
1351 264 : let rest = &src[i + needle.len() - "\"ANTARES_".len() + 1..];
1352 264 : let Some(end) = rest.find('"') else { continue };
1353 264 : let key = &rest[..end];
1354 264 : if unknown_config_key(key) {
1355 0 : missing.push(format!("{key} (read in {})", path.display()));
1356 264 : }
1357 : }
1358 : }
1359 : }
1360 : }
1361 2 : missing.sort();
1362 2 : missing.dedup();
1363 2 : assert!(
1364 2 : missing.is_empty(),
1365 : "KNOWN_KEYS is missing variables the workspace reads: {missing:#?}"
1366 : );
1367 2 : }
1368 : }
1369 :
1370 : #[cfg(test)]
1371 : mod accept_loop_tests {
1372 : use super::{accept, accept_backoff};
1373 :
1374 : /// An accept() error must never end the serve loop: tokio propagates
1375 : /// every non-WouldBlock errno straight from the syscall, so ECONNABORTED
1376 : /// (a client resetting between SYN and accept), EMFILE/ENFILE (the fd
1377 : /// ceiling) and ENOBUFS all used to take the whole broker down. `accept`
1378 : /// has no failure value at all — the only decision left is whether to
1379 : /// pause before retrying.
1380 : #[test]
1381 2 : fn no_accept_error_is_fatal_and_resource_errors_back_off() {
1382 : use std::io::ErrorKind::*;
1383 6 : for kind in [ConnectionAborted, ConnectionReset, Interrupted] {
1384 6 : assert!(
1385 6 : !accept_backoff(&std::io::Error::new(kind, "x")),
1386 : "{kind:?} is per-connection — the next accept must run at once"
1387 : );
1388 : }
1389 10 : for kind in [
1390 2 : Other,
1391 2 : OutOfMemory,
1392 2 : PermissionDenied,
1393 2 : InvalidInput,
1394 2 : NotConnected,
1395 2 : ] {
1396 10 : assert!(
1397 10 : accept_backoff(&std::io::Error::new(kind, "x")),
1398 : "{kind:?} would spin the loop without a pause"
1399 : );
1400 : }
1401 2 : }
1402 :
1403 : /// …and the healthy path still hands the loop its connection.
1404 : #[tokio::test]
1405 2 : async fn accept_yields_the_next_connection() {
1406 2 : let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
1407 2 : .await
1408 2 : .expect("bind");
1409 2 : let addr = listener.local_addr().expect("addr");
1410 2 : let client = tokio::spawn(async move { tokio::net::TcpStream::connect(addr).await });
1411 2 : let stream = tokio::time::timeout(std::time::Duration::from_secs(5), accept(&listener))
1412 2 : .await
1413 2 : .expect("a connection is accepted");
1414 2 : assert!(stream.peer_addr().is_ok());
1415 2 : let _ = client.await;
1416 2 : }
1417 : }
1418 :
1419 : #[cfg(test)]
1420 : mod switch_tests {
1421 : use super::is_off;
1422 :
1423 : /// The knobs whose default is OFF (ANTARES_REQUIRE_RLS, ANTARES_TELEMETRY)
1424 : /// must fail SAFE on a spelling they do not know: recognizing only
1425 : /// `1|true` turned the RLS gate off on `TRUE`/`yes`/`on` while the
1426 : /// operator believed it was enforced.
1427 : #[test]
1428 2 : fn only_an_explicit_off_value_reads_as_off() {
1429 20 : for off in [
1430 2 : "0", "false", "FALSE", "False", "off", "OFF", "no", "", " ", " 0\t",
1431 2 : ] {
1432 20 : assert!(is_off(off), "{off:?} must read as off");
1433 : }
1434 20 : for on in [
1435 2 : "1", "true", "TRUE", "True", "on", "On", "yes", "YES", " 1 ", "enabled",
1436 2 : ] {
1437 20 : assert!(!is_off(on), "{on:?} must NOT read as off");
1438 : }
1439 2 : }
1440 : }
1441 :
1442 : #[cfg(test)]
1443 : mod sweep_secs_tests {
1444 : use super::parse_sweep_secs;
1445 :
1446 : /// ANTARES_SWEEP_SECS paces the 4.22 GC in every store mode: absent is
1447 : /// the 15 min default, and a value that is not a positive integer is
1448 : /// fatal — a garbage cadence must never silently become the default one.
1449 : #[test]
1450 2 : fn sweep_secs_defaults_and_rejects() {
1451 2 : assert_eq!(parse_sweep_secs(None).expect("default"), 900);
1452 2 : assert_eq!(parse_sweep_secs(Some("2")).expect("explicit"), 2);
1453 14 : for bad in ["0", "-1", "", "2s", "abc", "1.5", "99999999999999999999999"] {
1454 14 : let err =
1455 14 : parse_sweep_secs(Some(bad)).expect_err(&format!("SWEEP_SECS={bad:?} is fatal"));
1456 14 : assert!(err.contains("ANTARES_SWEEP_SECS"), "{err}");
1457 : }
1458 2 : }
1459 : }
1460 :
1461 : #[cfg(test)]
1462 : mod driver_registry_tests {
1463 : use super::{built_with, shared_state, store_shelf, temporal_choice, TemporalChoice};
1464 : use antares_sql::StoreMode;
1465 :
1466 : /// The shelf in the message is rendered from the backend list, so a
1467 : /// backend added to `StoreMode` — or compiled in from outside this
1468 : /// workspace — cannot go missing from what the broker claims to have
1469 : /// been built with.
1470 : #[test]
1471 2 : fn the_shelf_names_every_store_mode() {
1472 2 : let shelf = built_with();
1473 8 : for m in StoreMode::ALL {
1474 8 : assert!(shelf.contains(m.as_str()), "{shelf} omits {m}");
1475 : }
1476 2 : assert!(
1477 2 : shelf.contains("none"),
1478 : "temporal `none` is part of the shelf: {shelf}"
1479 : );
1480 8 : for name in store_shelf() {
1481 8 : assert!(shelf.contains(name), "{shelf} omits {name}");
1482 : }
1483 2 : }
1484 :
1485 : /// Absent, or the store's own name, means one instance serves both
1486 : /// seams — no second store is ever built for the default.
1487 : #[test]
1488 2 : fn absent_or_same_name_shares_the_store() {
1489 2 : assert_eq!(
1490 2 : temporal_choice("postgres", None).expect("ok"),
1491 : TemporalChoice::SameAsStore
1492 : );
1493 2 : assert_eq!(
1494 2 : temporal_choice("postgres", Some("postgres")).expect("ok"),
1495 : TemporalChoice::SameAsStore
1496 : );
1497 2 : }
1498 :
1499 : #[test]
1500 2 : fn none_turns_history_off_and_other_names_build_a_second_store() {
1501 2 : assert_eq!(
1502 2 : temporal_choice("memory", Some("none")).expect("ok"),
1503 : TemporalChoice::None
1504 : );
1505 2 : assert_eq!(
1506 2 : temporal_choice("memory", Some("timescale")).expect("ok"),
1507 2 : TemporalChoice::Second("timescale".to_owned())
1508 : );
1509 2 : }
1510 :
1511 : /// An unknown backend is fatal at startup and the message names the
1512 : /// shelf this binary was built with — never a silent default.
1513 : #[test]
1514 2 : fn unknown_backend_is_fatal_and_lists_the_shelf() {
1515 2 : let err = temporal_choice("memory", Some("mongo")).expect_err("must fail");
1516 2 : assert!(err.contains("mongo"), "{err}");
1517 2 : assert!(err.contains(&built_with()), "{err}");
1518 2 : assert!(
1519 2 : temporal_choice("memory", Some("")).is_err(),
1520 : "an empty name is not a default"
1521 : );
1522 2 : }
1523 :
1524 : /// Shared state is what `ANTARES_BUS=nats` needs and what `bus=local`
1525 : /// refuses: a per-process backend — the built-in memory and file arms,
1526 : /// and any driver from outside the workspace — is never shared.
1527 : #[test]
1528 2 : fn only_the_database_backends_hold_shared_state() {
1529 2 : assert!(shared_state("postgres"));
1530 2 : assert!(shared_state("timescale"));
1531 2 : assert!(!shared_state("memory"));
1532 2 : assert!(!shared_state("file"));
1533 2 : assert!(
1534 2 : !shared_state("mongo"),
1535 : "an unknown name is not shared state"
1536 : );
1537 2 : }
1538 :
1539 : /// A driver compiled in from outside `crates/` is selected by name
1540 : /// exactly like a built-in, and serves both storage seams.
1541 : #[cfg(feature = "plugin-example")]
1542 : #[tokio::test]
1543 : async fn a_plugin_backend_is_on_the_shelf_and_builds_both_seams() {
1544 : let name = antares_plugin_example::NAME;
1545 : assert!(
1546 : store_shelf().contains(&name),
1547 : "the shelf omits the compiled-in plugin: {:?}",
1548 : store_shelf()
1549 : );
1550 : assert_eq!(
1551 : temporal_choice(name, None).expect("ok"),
1552 : TemporalChoice::SameAsStore
1553 : );
1554 : let drivers = super::build_drivers(name)
1555 : .await
1556 : .expect("the plugin driver builds");
1557 : assert_eq!(drivers.temporal_name.as_deref(), Some(name));
1558 : assert!(
1559 : drivers.maintenance.is_empty(),
1560 : "a plugin backend brings no Postgres maintenance job"
1561 : );
1562 : let tenant = antares_model::TenantId::default();
1563 : drivers
1564 : .store
1565 : .create(
1566 : &tenant,
1567 : antares_store::Kind::Entity,
1568 : "urn:ngsi-ld:Probe:1",
1569 : serde_json::json!({"id": "urn:ngsi-ld:Probe:1"}),
1570 : )
1571 : .await
1572 : .expect("create through the plugin driver");
1573 : assert_eq!(
1574 : drivers
1575 : .store
1576 : .list(&tenant, antares_store::Kind::Entity)
1577 : .await
1578 : .expect("list")
1579 : .len(),
1580 : 1
1581 : );
1582 : }
1583 : }
1584 :
1585 : #[cfg(test)]
1586 : mod api_surface_tests {
1587 : use super::{api_surfaces, SURFACE_SHELF};
1588 :
1589 6 : fn names(raw: Option<&str>) -> Vec<&'static str> {
1590 6 : api_surfaces(raw)
1591 6 : .expect("selection")
1592 6 : .iter()
1593 6 : .map(|(n, _)| *n)
1594 6 : .collect()
1595 6 : }
1596 :
1597 : /// Absent means the operational surface, and nothing else: a pod that
1598 : /// was never configured still answers its probes.
1599 : #[test]
1600 2 : fn absent_selects_admin() {
1601 2 : assert_eq!(names(None), vec!["admin"]);
1602 2 : assert_eq!(names(Some("admin")), vec!["admin"]);
1603 2 : assert_eq!(
1604 2 : names(Some(" admin , ")),
1605 2 : vec!["admin"],
1606 : "spacing is not a name"
1607 : );
1608 2 : }
1609 :
1610 : /// An unknown name is fatal at startup and the message names the shelf
1611 : /// this binary was built with — never a silently ignored surface.
1612 : #[test]
1613 2 : fn an_unknown_surface_is_fatal_and_lists_the_shelf() {
1614 2 : let err = api_surfaces(Some("admin,dashboard")).expect_err("must fail");
1615 2 : assert!(err.contains("dashboard"), "{err}");
1616 2 : for (name, _) in SURFACE_SHELF {
1617 2 : assert!(err.contains(name), "the message lists {name}: {err}");
1618 : }
1619 2 : }
1620 :
1621 : /// A surface compiled in from outside `crates/` is selected by name
1622 : /// beside the operational one, and claims its own prefix under `/x`.
1623 : #[cfg(feature = "plugin-example")]
1624 : #[test]
1625 : fn a_plugin_surface_mounts_beside_admin() {
1626 : assert_eq!(names(Some("admin,example")), vec!["admin", "example"]);
1627 : let built: Vec<Box<dyn antares_api::ApiSurface>> = api_surfaces(Some("admin,example"))
1628 : .expect("selection")
1629 : .iter()
1630 : .map(|(_, build)| build())
1631 : .collect();
1632 : let example = built
1633 : .iter()
1634 : .find(|s| s.name() == "example")
1635 : .expect("the plugin surface is built");
1636 : assert!(
1637 : example.prefix().starts_with("/x/"),
1638 : "a plugin surface lives under /x, never under the NGSI-LD root: {}",
1639 : example.prefix()
1640 : );
1641 : }
1642 :
1643 : /// An empty selection is fatal too: readiness, health and metrics are
1644 : /// the admin surface, so a pod without one can never report itself up.
1645 : #[test]
1646 2 : fn an_empty_selection_is_fatal() {
1647 8 : for raw in ["", " ", ",", " , "] {
1648 8 : let err = api_surfaces(Some(raw)).expect_err(&format!("{raw:?} selects nothing"));
1649 8 : assert!(err.contains("ANTARES_API_SURFACES"), "{err}");
1650 : }
1651 2 : }
1652 : }
1653 :
1654 : #[cfg(test)]
1655 : mod policy_shelf_tests {
1656 : use super::*;
1657 :
1658 4 : fn shelf() -> Vec<&'static str> {
1659 4 : POLICY_SHELF.iter().map(|(n, _)| *n).collect()
1660 4 : }
1661 :
1662 : /// ADR-0020: the broker ships one engine, and conformance is asserted
1663 : /// against it. A release build carries no other — an engine reaches the
1664 : /// shelf only through an off-by-default feature, so this assertion is
1665 : /// what "no addon in the shipped build" means for the policy seam.
1666 : #[cfg(not(feature = "plugin-example"))]
1667 : #[test]
1668 2 : fn the_shipped_build_carries_only_the_built_in_engine() {
1669 2 : assert_eq!(shelf(), vec!["allow-all"]);
1670 2 : }
1671 :
1672 : /// No selection is the built-in engine, which is the only default that
1673 : /// keeps a broker without a policy behaving like a broker.
1674 : #[test]
1675 2 : fn no_selection_is_the_built_in_engine() {
1676 6 : for raw in [None, Some(""), Some(" ")] {
1677 6 : assert_eq!(policy_engine(raw).expect("default").0, "allow-all");
1678 : }
1679 2 : }
1680 :
1681 : /// A typo must not serve every request wide open, so it is fatal and
1682 : /// the message names what this binary was built with.
1683 : #[test]
1684 2 : fn an_unknown_engine_is_fatal_and_names_the_shelf() {
1685 2 : let err = policy_engine(Some("opa")).expect_err("unknown engine");
1686 2 : assert!(err.contains("ANTARES_POLICY"), "{err}");
1687 2 : for name in shelf() {
1688 2 : assert!(err.contains(name), "the message lists {name}: {err}");
1689 : }
1690 2 : }
1691 :
1692 : /// An engine compiled in from outside `crates/` is selected by name
1693 : /// exactly like the built-in one — and, given no rules to enforce,
1694 : /// refuses every operation rather than allowing it (ADR-0020: a broken
1695 : /// engine costs service, never access rules).
1696 : #[cfg(feature = "plugin-example")]
1697 : #[test]
1698 : fn a_plugin_engine_is_on_the_shelf_and_fails_closed() {
1699 : let name = antares_plugin_example::POLICY_NAME;
1700 : assert!(shelf().contains(&name), "the shelf omits it: {:?}", shelf());
1701 : let (selected, build) = policy_engine(Some(name)).expect("selection");
1702 : assert_eq!(selected, name);
1703 : assert_eq!(build().name(), name);
1704 : assert!(
1705 : std::env::var(antares_plugin_example::RULES_ENV).is_err(),
1706 : "this test reads the engine built with no rules document"
1707 : );
1708 : assert!(
1709 : antares_plugin_example::ExamplePolicy::from_env()
1710 : .broken()
1711 : .is_some(),
1712 : "an engine selected without rules would otherwise allow everything"
1713 : );
1714 : }
1715 : }
1716 :
1717 : #[cfg(test)]
1718 : mod pg_pool_tests {
1719 : use super::{parse_pg_pool, parse_pg_statement_timeout};
1720 :
1721 : /// ANTARES_PG_POOL: absent defaults to 20; a value that is not a
1722 : /// positive integer is fatal — misconfiguration must never silently
1723 : /// run with a default.
1724 : #[test]
1725 2 : fn pg_pool_parse_defaults_and_rejects() {
1726 2 : assert_eq!(
1727 2 : parse_pg_statement_timeout(None).expect("default"),
1728 2 : std::time::Duration::from_secs(30)
1729 : );
1730 2 : assert_eq!(
1731 2 : parse_pg_statement_timeout(Some("1500")).expect("explicit"),
1732 2 : std::time::Duration::from_millis(1500)
1733 : );
1734 2 : assert!(parse_pg_statement_timeout(Some("0")).is_err());
1735 2 : assert!(parse_pg_statement_timeout(Some("30s")).is_err());
1736 2 : assert_eq!(parse_pg_pool(None).expect("default"), 20);
1737 2 : assert_eq!(parse_pg_pool(Some("7")).expect("explicit"), 7);
1738 2 : assert!(
1739 2 : parse_pg_pool(Some("abc")).is_err(),
1740 : "non-numeric ANTARES_PG_POOL must be fatal"
1741 : );
1742 2 : assert!(
1743 2 : parse_pg_pool(Some("0")).is_err(),
1744 : "a zero-sized pool must be fatal"
1745 : );
1746 2 : assert!(parse_pg_pool(Some("-3")).is_err());
1747 2 : }
1748 : }
|
