plecto_control/lib.rs
1//! plecto-control — the control plane (ADR 000007 / 000008).
2//!
3//! A single declarative TOML manifest pins filters by OCI digest and declares one chain. An
4//! `ArtifactStore` resolves each filter from a local, offline OCI image-layout (remote
5//! registry fetch via `wkg` is an out-of-band operator step, kept out of the runtime), the
6//! `Host` loads it through the ADR 000006 provenance gate, and a chain dispatcher drives a
7//! request through the loaded filters. `reload` rebuilds the set and swaps it **atomically**
8//! (ArcSwap): new requests see the new set, in-flight holders keep the old one until it
9//! drops. Single-node-first (ADR 000008).
10//!
11//! The trust policy lives on the `Host` and is fixed at construction; changing trust roots
12//! requires a new `Control`. `reload` swaps only the filter set + chain (same `Host`, same
13//! epoch ticker), so a runaway filter stays bounded across reloads.
14
15// Hot-path discipline (bp-rust): no unwrap/expect/panic/indexing on the data plane. Exempted
16// under `cfg(test)` — this crate's own `#[cfg(test)] mod` blocks legitimately use them;
17// `tests/*.rs` integration tests are separate crates and are never subject to this attribute.
18// plecto-control is config/build-time (not per-request), but its Maglev/weighted-split hashing
19// and route matching are still touched by the fast path indirectly, so the same discipline
20// applies (added at Stage 3, after the `hash.rs`/`maglev.rs`/`weighted.rs` fixes it surfaces).
21//
22// `clippy::pedantic`/`clippy::nursery` are NOT enabled here (nor in plecto-host/plecto-server):
23// a dry run measures 400+ pre-existing hits crate-wide, almost entirely pre-existing stylistic
24// noise unrelated to this refactor's scope; scoped-allow-ing all of them would be disproportionate
25// busy-work. Left as a known, explicit gap rather than silently skipped.
26// `rust_2018_idioms` (warn) and `unsafe_op_in_unsafe_fn` (deny) come from `[workspace.lints]`.
27#![cfg_attr(
28 not(test),
29 warn(
30 clippy::unwrap_used,
31 clippy::expect_used,
32 clippy::panic,
33 clippy::indexing_slicing
34 )
35)]
36
37mod artifact;
38mod chain;
39mod control_observability;
40mod control_reload;
41mod diagnostic;
42mod error;
43mod hash;
44mod maglev;
45mod manifest;
46pub mod oci;
47mod ratelimit;
48mod reload;
49mod rng;
50mod route;
51mod snapshot;
52mod stek;
53mod tls;
54mod upstream;
55mod weighted;
56
57use std::collections::{HashMap, HashSet};
58use std::path::{Path, PathBuf};
59use std::sync::Arc;
60
61use arc_swap::ArcSwap;
62use plecto_host::{LoadedFilter, SignedArtifact};
63
64pub use artifact::{ArtifactStore, MemoryStore, ResolvedArtifact};
65pub use chain::{ChainOutcome, RequestBodyOutcome, ResponseOutcome};
66pub use diagnostic::{
67 DEV_KEY_IN_TRUST, Diagnostic, PATH_NORMALIZATION_REJECTED, QUOTA_EXCEEDED,
68 SIGNATURE_VERIFICATION_FAILED, diagnose, diagnosed_message,
69};
70pub use error::ControlError;
71pub use manifest::{
72 Chain, CircuitBreaker, CompressionAlgorithm, FilterEntry, HealthConfig, IsolationKind,
73 Manifest, Observability, OutlierDetection, ProxyProtocolTrust, RateLimitKeyKind, Route,
74 RouteCompression, RouteRateLimit, State, StateBackendKind, TlsCert, Trust, Upstream,
75};
76pub use ratelimit::RateLimitDecision;
77#[cfg(unix)]
78pub use reload::SignalReloadSource;
79pub use reload::{ReloadOutcome, ReloadSource, serve_reloads};
80pub use route::{CompressionConfig, RouteInfo, UpgradeConfig, normalize_path};
81/// The rustls TLS client config the fast path re-encrypts upstream forward legs with
82/// (ADR 000042), re-exported for the same reason as [`TlsServerConfig`].
83pub use rustls::ClientConfig as TlsClientConfig;
84/// The rustls TLS server config the fast path terminates with (ADR 000014), re-exported so
85/// `plecto-server` names the same `rustls` type the control plane built.
86pub use rustls::ServerConfig as TlsServerConfig;
87pub use snapshot::ConfigSnapshot;
88pub use upstream::{
89 Endpoints, HashInput, HashKeySource, Pick, UpstreamGroup, UpstreamInstance, UpstreamRegistry,
90};
91
92// Re-export the host surface a caller drives the control plane with, so they need not depend
93// on `plecto-host` directly for the common path — including the ADR 000009 observability
94// types (build a `Host` with a sink, then drive snapshots that carry the trace context).
95pub use plecto_host::{
96 FanOutSink, FilterSpan, Header, Host, HttpRequest, HttpResponse, InMemorySink, MetricsSink,
97 MetricsSnapshot, NoopSink, RequestTrace, SpanOutcome, TelemetrySink, TrustPolicy,
98};
99// Filter Dev Kit (ADR 000065): `plecto conformance` / `plecto dev` / `plecto new-filter` need
100// the generic conformance battery and the persistent dev-signing key. Re-exported the same way
101// as the rest of the host surface above — `plecto-server` never takes a direct `plecto-host`
102// production dependency; a plain (non-`test-support`) `plecto-host` build needs no wasm32
103// toolchain, so this widens no dependency edge, just this crate's existing re-export list.
104pub use plecto_host::{
105 ConformanceCheck, ConformanceReport, DEV_KEY_MARKER, DevKeyError, DevSigner, FILTER_WIT,
106 bound_sbom, public_key_path_for, run_conformance,
107};
108// The OTLP export surface (ADR 000040): the fast-path server drives the span buffer + the
109// hand-written wire encoding through the control plane, without depending on `plecto-host`.
110pub use plecto_host::otlp;
111
112/// The atomically-swappable active configuration: the loaded filters, the chain order, and
113/// the `content_hash` of the manifest that produced them. Held behind an `ArcSwap`; never
114/// mutated in place — `reload` replaces it wholesale. The hash rides with the config it
115/// describes so `reload_from_disk` can compare the running `config version` without a
116/// separate lock.
117pub(crate) struct ActiveConfig {
118 pub(crate) filters: HashMap<String, Arc<LoadedFilter>>,
119 /// The manifest's default `[chain]`, resolved to the loaded filter in order — built once per
120 /// reload so the default-chain convenience (`ConfigSnapshot::on_request` / `on_response`)
121 /// never re-hashes a filter id against `filters` on every request (mirrors
122 /// `CompiledRoute::resolved_chain`).
123 pub(crate) resolved_chain: Vec<Arc<LoadedFilter>>,
124 /// Compiled routing table (ADR 000013): empty unless the manifest declares `[[route]]`.
125 /// The fast-path server matches against these; the chain-only `on_request` ignores them.
126 pub(crate) routes: Vec<route::CompiledRoute>,
127 /// TLS server config built from `[[tls]]` (ADR 000014), or `None` for plain HTTP/1.1. Rides
128 /// the `ArcSwap` with the rest, so a reload swaps certs atomically (new conns get new certs).
129 pub(crate) tls: Option<Arc<rustls::ServerConfig>>,
130 /// QUIC TLS server config for HTTP/3 (ADR 000016): ALPN `h3`, TLS 1.3, same SNI cert resolver
131 /// as `tls`. `None` whenever `tls` is `None` (h3 requires TLS). Rides the same `ArcSwap`.
132 pub(crate) quic_tls: Option<Arc<rustls::ServerConfig>>,
133 pub(crate) hash: String,
134}
135
136/// The control plane: owns the `Host` (and thus the trust policy + epoch ticker) and the
137/// artifact store, and holds the active filter set behind an `ArcSwap` for lock-free reads
138/// and atomic reload. `manifest_path` is `Some` only when the plane was built from an on-disk
139/// manifest — that is what `reload_from_disk` (and the SIGHUP loop) re-reads.
140pub struct Control {
141 host: Host,
142 store: Box<dyn ArtifactStore>,
143 active: ArcSwap<ActiveConfig>,
144 /// Serializes reloads: `build_active` reconciles the shared `upstreams` registry in place and
145 /// then stores `active`, so two interleaved reloads could leave routes holding groups the
146 /// registry no longer probes (permanently pessimistic → 503). The shipped SIGHUP loop is
147 /// single-threaded; this guard closes the hole for any other embedder of the public API.
148 reload_gate: parking_lot::Mutex<()>,
149 /// The upstream instances + their health state (ADR 000017). Lives OUTSIDE `active` so a
150 /// reload's `build_active` reconciles it in place — health state survives the swap. The
151 /// fast-path server reads it both via routing (`RouteInfo.upstream`, resolved at build time)
152 /// and via `upstream_groups` (the health-check supervisor).
153 upstreams: Arc<UpstreamRegistry>,
154 manifest_path: Option<PathBuf>,
155 /// The `[trust]` section the `Host` was built from, captured at construction. A reload that
156 /// would change it is rejected (`TrustChangeRequiresRestart`) rather than silently dropped
157 /// (f000004 #1): trust roots are fixed for the life of the `Host` / epoch ticker.
158 trust: Trust,
159 /// The `[state]` section the `Host`'s `KvBackend` was built from (ADR 000041), captured at
160 /// construction. Same contract as `trust`: the backend lives for the life of the `Host`,
161 /// so a reload that would change it is rejected (`StateChangeRequiresRestart`).
162 state: manifest::State,
163 /// Base directory the manifest's relative paths (filter `source`, TLS `cert_path`/`key_path`)
164 /// resolve against (ADR 000014). Captured at construction so a reload re-reads certs from the
165 /// same root. `"."` for the in-memory `load` core (tests use absolute cert paths).
166 base_dir: PathBuf,
167 /// Host-aggregated filter-execution metrics (ADR 000009): the `MetricsSink` wired into the
168 /// `Host` at construction, snapshotted by the fast path's admin `/metrics` endpoint.
169 filter_metrics: Arc<MetricsSink>,
170 /// Operational observability config (`[observability]`, ADR 000009), captured at construction:
171 /// the admin endpoint bind address and the access-log toggle. Not part of the config version.
172 observability: Observability,
173 /// The data-plane listener config (`[listen]`), captured at construction like
174 /// `observability`: the listener binds once at startup, so a reload does not re-bind.
175 listen: manifest::Listen,
176 /// The parsed `[listen.proxy_protocol]` trust (ADR 000057), captured at construction like
177 /// `listen` itself: the TCP listener consults it once at startup, so a reload does not
178 /// change it. `None` = PROXY v2 reception off (the default).
179 proxy_protocol: Option<manifest::ProxyProtocolTrust>,
180 /// The OTLP span buffer (ADR 000040), present iff `[observability] otlp_endpoint` is set:
181 /// fanned in beside the sinks above at `Host` construction, drained by the fast path's
182 /// export pump. Like the admin listener, it binds once at startup — a reload swaps only the
183 /// filter set, so the buffer (and the endpoint) live for the process.
184 otlp: Option<Arc<plecto_host::otlp::OtlpBuffer>>,
185}
186
187impl Control {
188 /// The ONE place a `Control` is put together: every public constructor reduces to "obtain a
189 /// `Host` + store + observability handles, then assemble" — a new field means one edit here,
190 /// not four (the four constructors previously hand-built the 15-field struct and had already
191 /// drifted subtly on `filter_metrics`).
192 fn assemble(
193 host: Host,
194 store: Box<dyn ArtifactStore>,
195 manifest: &Manifest,
196 base_dir: &Path,
197 manifest_path: Option<&Path>,
198 filter_metrics: Arc<MetricsSink>,
199 otlp: Option<Arc<plecto_host::otlp::OtlpBuffer>>,
200 ) -> Result<Self, ControlError> {
201 let upstreams = Arc::new(UpstreamRegistry::new());
202 let active = build_active(&host, manifest, store.as_ref(), base_dir, &upstreams)?;
203 Ok(Self {
204 host,
205 store,
206 active: ArcSwap::from_pointee(active),
207 reload_gate: parking_lot::Mutex::new(()),
208 upstreams,
209 manifest_path: manifest_path.map(Path::to_path_buf),
210 trust: manifest.trust.clone(),
211 state: manifest.state.clone(),
212 base_dir: base_dir.to_path_buf(),
213 filter_metrics,
214 observability: manifest.observability.clone(),
215 listen: manifest.listen.clone(),
216 proxy_protocol: manifest.listen.proxy_protocol_trust()?,
217 otlp,
218 })
219 }
220
221 /// Build a control plane entirely from a manifest and a base directory — the ops
222 /// entrypoint. Reads the trusted-key PEMs (ADR 000006), constructs the `Host`, and
223 /// resolves filters from offline OCI image-layouts under `base_dir` (ADR 000007). Every
224 /// path in the manifest (`trust.keys`, each filter `source`) is resolved relative to
225 /// `base_dir`. Remote fetch (`wkg`) is an out-of-band step that populates those layouts.
226 pub fn from_manifest(manifest: &Manifest, base_dir: &Path) -> Result<Self, ControlError> {
227 let (host, store, filter_metrics, otlp) = build_host_and_store(manifest, base_dir)?;
228 Self::assemble(
229 host,
230 Box::new(store),
231 manifest,
232 base_dir,
233 None,
234 filter_metrics,
235 otlp,
236 )
237 }
238
239 /// Build from a pre-constructed `Host` (carrying its `TrustPolicy`) and an artifact store
240 /// — the testable core. Each manifest filter is resolved through `store` (digest pin),
241 /// loaded through the host's ADR 000006 gate (signature + SBOM), and the chain order is
242 /// validated against the loaded set. Any failure aborts the build (nothing is loaded
243 /// half-way into a live set).
244 pub fn load(
245 host: Host,
246 manifest: &Manifest,
247 store: Box<dyn ArtifactStore>,
248 ) -> Result<Self, ControlError> {
249 // The in-memory core has no manifest directory; relative paths resolve against the cwd.
250 // Tests that exercise `[[tls]]` use absolute cert paths, so this base does not bite them.
251 // OTLP export (ADR 000040): fan the span buffer in BESIDE the caller's sink (never
252 // replacing it), before `build_active` loads filters (the sink is cloned into each).
253 // The caller supplied the `Host`, so its sink is the caller's (or `NoopSink`); this
254 // testable core keeps its own empty `filter_metrics` tally rather than reaching into it.
255 let (host, otlp) = add_otlp_buffer(host, manifest);
256 Self::assemble(
257 host,
258 store,
259 manifest,
260 Path::new("."),
261 None,
262 Arc::new(MetricsSink::new()),
263 otlp,
264 )
265 }
266
267 /// Build the whole control plane from a single on-disk manifest file — the
268 /// disk-reloadable ops entrypoint (ADR 000007 / 000008). Like `from_manifest`, but reads
269 /// and *remembers* the manifest path so SIGHUP / `reload_from_disk` can pick up an
270 /// operator's edits. Trusted-key PEMs and filter layouts resolve relative to the
271 /// manifest's own directory.
272 pub fn from_manifest_path(manifest_path: &Path) -> Result<Self, ControlError> {
273 let base_dir = manifest_path.parent().unwrap_or_else(|| Path::new("."));
274 let manifest = read_manifest(manifest_path)?;
275 let (host, store, filter_metrics, otlp) = build_host_and_store(&manifest, base_dir)?;
276 Self::assemble(
277 host,
278 Box::new(store),
279 &manifest,
280 base_dir,
281 Some(manifest_path),
282 filter_metrics,
283 otlp,
284 )
285 }
286
287 /// Like `load`, but the manifest lives on disk at `manifest_path`: the path is remembered
288 /// so `reload_from_disk` can re-read it, while artifacts still resolve through the injected
289 /// `store` (so a test can pair an on-disk manifest with an in-memory artifact store). The
290 /// trust policy stays fixed on `host`.
291 pub fn load_at(
292 host: Host,
293 manifest_path: &Path,
294 store: Box<dyn ArtifactStore>,
295 ) -> Result<Self, ControlError> {
296 let base_dir = manifest_path.parent().unwrap_or_else(|| Path::new("."));
297 let manifest = read_manifest(manifest_path)?;
298 let (host, otlp) = add_otlp_buffer(host, &manifest);
299 Self::assemble(
300 host,
301 store,
302 &manifest,
303 base_dir,
304 Some(manifest_path),
305 Arc::new(MetricsSink::new()),
306 otlp,
307 )
308 }
309
310 /// The ids currently loaded (for diagnostics / tests). Order is unspecified.
311 pub fn loaded_ids(&self) -> Vec<String> {
312 self.active.load().filters.keys().cloned().collect()
313 }
314}
315
316/// Read + parse a manifest from disk (shared by the on-disk constructors and `reload_from_disk`).
317fn read_manifest(path: &Path) -> Result<Manifest, ControlError> {
318 let toml = std::fs::read_to_string(path).map_err(|e| ControlError::IoAt {
319 path: path.to_path_buf(),
320 source: e,
321 })?;
322 Manifest::from_toml(&toml)
323}
324
325/// Read a manifest-referenced file with the path attached to any failure (DECREE §3: an error a
326/// human acts on must say WHICH file — trust keys and manifests are read from several places).
327fn read_file(path: &Path) -> Result<Vec<u8>, ControlError> {
328 std::fs::read(path).map_err(|e| ControlError::IoAt {
329 path: path.to_path_buf(),
330 source: e,
331 })
332}
333
334/// [`validate_manifest`] / [`validate_manifest_path`]'s success value: the manifest's config
335/// version plus any non-fatal [`Diagnostic`] warnings (ADR 000065 decision 5) — currently just
336/// [`DEV_KEY_IN_TRUST`], raised when a `[trust]` key file carries `plecto_host::DEV_KEY_MARKER`.
337/// A warning never fails validation: a `plecto dev`-generated manifest is SUPPOSED to trip it.
338#[derive(Debug)]
339pub struct ValidateOutcome {
340 pub config_version: String,
341 pub warnings: Vec<Diagnostic>,
342}
343
344/// Statically validate `manifest` — the `plecto validate` core (the `nginx -t` shape): every
345/// check the server would fail closed on at startup that needs no artifact and mutates nothing.
346/// Covers the strict parse (the caller already ran it), `[trust]` key files, `[state]` coherence,
347/// per-filter metering/rate-limit ranges, duplicate ids, chain and route references, the weighted
348/// split, `[[tls]]` cert/key loads, and `[[upstream]]` (LB config + `[upstream.tls]` CA loads).
349/// Returns the manifest's config version (semantic content hash, ADR 000008) on success.
350///
351/// Deliberately NOT covered, so a CI run needs only the manifest + its referenced config files:
352/// OCI artifact resolution and the signature/SBOM load gate (the deploy dir may not exist where
353/// validation runs — startup still enforces them, ADR 000006/000007), and the `[state]` backend
354/// open (validation must never create a redb file).
355pub fn validate_manifest(
356 manifest: &Manifest,
357 base_dir: &Path,
358) -> Result<ValidateOutcome, ControlError> {
359 let mut pems: Vec<Vec<u8>> = Vec::with_capacity(manifest.trust.keys.len());
360 let mut warnings = Vec::new();
361 for key_path in &manifest.trust.keys {
362 let pem = read_file(&base_dir.join(key_path))?;
363 if pem.starts_with(plecto_host::DEV_KEY_MARKER.as_bytes()) {
364 warnings.push(DEV_KEY_IN_TRUST);
365 }
366 pems.push(pem);
367 }
368 TrustPolicy::from_pem_keys(&pems).map_err(|e| ControlError::TrustKey(e.to_string()))?;
369 manifest.state.validate()?;
370 manifest.listen.validate()?;
371 let filter_ids = validate_filters_and_chain(manifest)?;
372 let upstream_names: HashSet<&str> =
373 manifest.upstreams.iter().map(|u| u.name.as_str()).collect();
374 route::validate_routes(&manifest.routes, &filter_ids, &upstream_names)?;
375 // ONE read of the client-auth CA, shared between the verifier build and the config version
376 // (same rule as `build_active`): the reported version always describes the validated bytes.
377 let client_auth_ca = manifest.read_client_auth_ca(base_dir)?;
378 tls::build_server_configs(
379 &manifest.tls,
380 manifest.resumption.as_ref(),
381 manifest
382 .listen
383 .client_auth
384 .as_ref()
385 .zip(client_auth_ca.as_deref()),
386 base_dir,
387 )?;
388 // A throwaway registry runs the full upstream validation (names, LB, `[upstream.tls]` CA
389 // loads) without touching any live state.
390 UpstreamRegistry::new().reconcile(&manifest.upstreams, base_dir)?;
391 let config_version = manifest.content_hash_with_ca(client_auth_ca.as_deref())?;
392 Ok(ValidateOutcome {
393 config_version,
394 warnings,
395 })
396}
397
398/// [`validate_manifest`] for an on-disk manifest: reads + strictly parses `path`, resolving
399/// relative paths against the manifest's own directory (the same rule the server applies).
400pub fn validate_manifest_path(path: &Path) -> Result<ValidateOutcome, ControlError> {
401 let base_dir = path.parent().unwrap_or_else(|| Path::new("."));
402 let manifest = read_manifest(path)?;
403 validate_manifest(&manifest, base_dir)
404}
405
406/// The manifest's JSON Schema (ADR 000049), derived from the very serde model `from_toml` parses
407/// with — the schema cannot drift from the structs, and `deny_unknown_fields` surfaces as
408/// `additionalProperties: false`, so editor validation rejects exactly what `validate` rejects.
409/// draft-07 output: the level taplo / Even Better TOML consume (schemars' 2020-12 default is
410/// outside taplo's documented support). Serialising a just-generated schema cannot fail, but if
411/// it ever did, the error surfaces to the CLI caller — a silent `"{}"` fallback would be a
412/// schema that validates EVERYTHING, fail-open for the editor validation this exists to provide.
413pub fn manifest_json_schema() -> Result<String, ControlError> {
414 let generator = schemars::generate::SchemaSettings::draft07().into_generator();
415 let schema = generator.into_root_schema_for::<Manifest>();
416 Ok(serde_json::to_string_pretty(&schema)?)
417}
418
419/// What `build_host_and_store` assembles for the manifest-driven constructors: the `Host` (sinks
420/// wired), the offline OCI store, and the observability handles `Control` retains.
421type BuiltHost = (
422 Host,
423 oci::OciLayoutStore,
424 Arc<MetricsSink>,
425 Option<Arc<plecto_host::otlp::OtlpBuffer>>,
426);
427
428/// Construct the `Host` (trust roots from the manifest's PEMs, ADR 000006; state backend from
429/// `[state]`, ADR 000041) and the offline OCI artifact store, both rooted at `base_dir`.
430/// Shared by `from_manifest` and `from_manifest_path`.
431fn build_host_and_store(manifest: &Manifest, base_dir: &Path) -> Result<BuiltHost, ControlError> {
432 let mut pems: Vec<Vec<u8>> = Vec::with_capacity(manifest.trust.keys.len());
433 for key_path in &manifest.trust.keys {
434 pems.push(read_file(&base_dir.join(key_path))?);
435 }
436 let trust =
437 TrustPolicy::from_pem_keys(&pems).map_err(|e| ControlError::TrustKey(e.to_string()))?;
438 let kv = build_state_backend(&manifest.state, base_dir)?;
439 // Wire the host-aggregated filter metrics (ADR 000009): a `MetricsSink` tallies every filter
440 // execution. Set BEFORE filters load (the sink is cloned into each at `load`), and retained on
441 // `Control` so the fast path's admin endpoint can snapshot it. The default was `NoopSink`
442 // (observability off) — this is the wiring that makes the M5 span/metrics stage observable.
443 let filter_metrics = Arc::new(MetricsSink::new());
444 let host = Host::with_backend(trust, kv)
445 .map_err(|e| ControlError::HostInit(e.to_string()))?
446 .with_telemetry_sink(filter_metrics.clone());
447 // OTLP export (ADR 000040): the span buffer fans in beside the metrics tally.
448 let (host, otlp) = add_otlp_buffer(host, manifest);
449 let store = oci::OciLayoutStore::new(base_dir);
450 Ok((host, store, filter_metrics, otlp))
451}
452
453/// Build the `KvBackend` the manifest's `[state]` selects (ADR 000041): the one store the
454/// `host-kv` / `host-counter` / `host-ratelimit` capabilities share. `memory` keeps today's
455/// process-lifetime behaviour; `redb` opens (or creates) the database at the manifest-relative
456/// `path`. The parent directory must already exist — a typo'd path errors here instead of
457/// silently growing a new tree (directory preparation is the operator's responsibility).
458fn build_state_backend(
459 state: &manifest::State,
460 base_dir: &Path,
461) -> Result<Arc<dyn plecto_host::KvBackend>, ControlError> {
462 state.validate()?;
463 match state.backend {
464 StateBackendKind::Memory => Ok(Arc::new(plecto_host::MemoryBackend::default())),
465 StateBackendKind::Redb => {
466 let path = base_dir.join(state.path.as_deref().unwrap_or_default());
467 if !path.parent().is_some_and(Path::is_dir) {
468 return Err(ControlError::StateBackendInit(format!(
469 "parent directory of {} does not exist",
470 path.display()
471 )));
472 }
473 let backend = plecto_host::RedbBackend::open(&path)
474 .map_err(|e| ControlError::StateBackendInit(e.to_string()))?;
475 Ok(Arc::new(backend))
476 }
477 }
478}
479
480/// When `[observability] otlp_endpoint` is set, fan the OTLP span buffer (ADR 000040) in beside
481/// the host's current sink. Must run before filters load (the sink is cloned into each).
482fn add_otlp_buffer(
483 host: Host,
484 manifest: &Manifest,
485) -> (Host, Option<Arc<plecto_host::otlp::OtlpBuffer>>) {
486 if manifest.observability.otlp_endpoint.is_none() {
487 return (host, None);
488 }
489 let buffer = Arc::new(plecto_host::otlp::OtlpBuffer::default());
490 (host.with_added_telemetry_sink(buffer.clone()), Some(buffer))
491}
492
493/// The pure filter/chain-semantics checks shared by [`validate_manifest`] (the `nginx -t` core)
494/// and [`build_active`] (the load path): duplicate filter ids, per-entry metering / rate-limit
495/// ranges, and default-chain references. ONE function so a check added for one caller cannot be
496/// silently missed by the other (the two previously re-implemented this sequence in parallel).
497fn validate_filters_and_chain(manifest: &Manifest) -> Result<HashSet<&str>, ControlError> {
498 let mut filter_ids: HashSet<&str> = HashSet::with_capacity(manifest.filters.len());
499 for entry in &manifest.filters {
500 if !filter_ids.insert(entry.id.as_str()) {
501 return Err(ControlError::DuplicateFilterId(entry.id.clone()));
502 }
503 // Reject out-of-range metering / rate-limit values before they reach the host.
504 entry.validate()?;
505 }
506 for id in &manifest.chain.filters {
507 if !filter_ids.contains(id.as_str()) {
508 return Err(ControlError::UnknownChainFilter(id.clone()));
509 }
510 }
511 Ok(filter_ids)
512}
513
514/// Resolve + verify + load every manifest filter into a fresh `ActiveConfig`. Pure w.r.t. the
515/// live set: it touches nothing until it fully succeeds, so a failed `reload` leaves the
516/// running set untouched.
517fn build_active(
518 host: &Host,
519 manifest: &Manifest,
520 store: &dyn ArtifactStore,
521 base_dir: &Path,
522 registry: &UpstreamRegistry,
523) -> Result<ActiveConfig, ControlError> {
524 // The pure semantic checks run FIRST (shared with `validate_manifest`), so the load loop
525 // below never sees a duplicate id or an unreferenced chain filter.
526 let filter_ids = validate_filters_and_chain(manifest)?;
527 let mut filters: HashMap<String, Arc<LoadedFilter>> = HashMap::new();
528 for entry in &manifest.filters {
529 let artifact = store.resolve(&entry.source, &entry.digest)?;
530 let signed = SignedArtifact {
531 component_bytes: &artifact.component,
532 component_signature: &artifact.component_signature,
533 sbom: &artifact.sbom,
534 sbom_signature: &artifact.sbom_signature,
535 };
536 let loaded = host
537 .load(&entry.id, &signed, entry.load_options())
538 .map_err(|err| ControlError::Load {
539 id: entry.id.clone(),
540 err,
541 })?;
542 filters.insert(entry.id.clone(), Arc::new(loaded));
543 }
544
545 // Routing table (ADR 000013 / 000017). Validate every route reference (upstream name, filter
546 // ids), the weighted split, and the native rate limit PURELY first — before the persistent
547 // upstream registry is mutated — so a manifest we'd reject never reconciles the registry
548 // (reload stays all-or-nothing; the running upstream health state is untouched on a failed
549 // reload). `validated_routes` carries each route's already-resolved forwarding targets, reused
550 // below instead of calling `targets()` again.
551 let upstream_names: HashSet<&str> =
552 manifest.upstreams.iter().map(|u| u.name.as_str()).collect();
553 let validated_routes = route::validate_routes(&manifest.routes, &filter_ids, &upstream_names)?;
554
555 // TLS termination config (ADR 000014 TCP / ADR 000016 QUIC): build the rustls ServerConfigs
556 // from `[[tls]]`, sharing one SNI cert resolver. A bad cert is fail-closed here, so a failed
557 // reload never swaps in a TLS config that cannot serve. Built before the registry is touched.
558 // The client-auth CA is read ONCE and shared with the content hash below, so the recorded
559 // config version always describes the trust roots the verifier was actually built from.
560 let client_auth_ca = manifest.read_client_auth_ca(base_dir)?;
561 let (tls, quic_tls) = match tls::build_server_configs(
562 &manifest.tls,
563 manifest.resumption.as_ref(),
564 manifest
565 .listen
566 .client_auth
567 .as_ref()
568 .zip(client_auth_ca.as_deref()),
569 base_dir,
570 )? {
571 Some(configs) => (Some(configs.tcp), Some(configs.quic)),
572 None => (None, None),
573 };
574
575 // Compute the content hash BEFORE the registry reconcile (review f000005 P3#8). `reconcile`
576 // is the step that MUTATES persistent state (the health registry, which survives reloads), so
577 // every other fallible step — including this hash — must run before it for the "after reconcile
578 // the build is infallible" / all-or-nothing invariant to hold literally, not just in practice.
579 let hash = manifest.content_hash_with_ca(client_auth_ca.as_deref())?;
580
581 // Reconcile the upstream registry LAST among the fallible steps (ADR 000017): this validates
582 // duplicate names / empty address lists and preserves health for unchanged `(name, address)`
583 // instances across the reload. After it returns Ok the build is infallible, so a rejected
584 // reload never leaves the registry reconciled to a manifest whose `active` was not swapped in.
585 registry.reconcile(&manifest.upstreams, base_dir)?;
586 let mut routes = Vec::with_capacity(validated_routes.len());
587 for route::ValidatedRoute { route: r, targets } in validated_routes {
588 // Resolve the route's forwarding targets (already validated above) to their upstream
589 // groups, then compile the weighted split (ADR 000034). A single `upstream` becomes a
590 // one-element set.
591 let mut resolved = Vec::with_capacity(targets.len());
592 for (name, weight) in targets {
593 // present: the name was validated above and reconcile built a group for each manifest
594 // upstream. Fall back to the error (unreachable) rather than panic (data-plane no-panic).
595 let Some(group) = registry.group(name) else {
596 return Err(ControlError::UnknownRouteUpstream {
597 path_prefix: r.matcher.path_prefix.clone(),
598 upstream: name.to_string(),
599 });
600 };
601 resolved.push((group, weight));
602 }
603 let backends = weighted::WeightedBackends::new(resolved).map_err(|reason| {
604 ControlError::InvalidRoute {
605 path_prefix: r.matcher.path_prefix.clone(),
606 reason,
607 }
608 })?;
609 // The compilation itself (pre-normalised match dimensions, resolved chain, limiter /
610 // upgrade / compression facilities) lives beside `CompiledRoute` in route.rs.
611 routes.push(route::CompiledRoute::compile(r, backends, &filters));
612 }
613
614 let resolved_chain = manifest
615 .chain
616 .filters
617 .iter()
618 .filter_map(|id| filters.get(id).cloned())
619 .collect();
620
621 Ok(ActiveConfig {
622 filters,
623 resolved_chain,
624 routes,
625 tls,
626 quic_tls,
627 hash,
628 })
629}