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