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