Skip to main content

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}