Implement hot-reload safety checks for boot-time configuration fields; reject changes to listeners, admin, shutdown, telemetry metrics, and access log during reload

Author: fabian4

ARCHITECTURE.md

@@ -3,7 +3,7 @@
 ## 1. Axioms / Invariants
 - **Frozen Data Plane** — Every semantic decision (routing, retries, TLS, RBAC, health, observability) **MUST** be resolved before serialization. No runtime component may construct or reinterpret policy.
 - **No Runtime Interpretation** — The runtime **MUST NOT** parse text configs, infer defaults, evaluate scripts, or execute dynamic code. It only deserializes trusted `.pvs` artifacts produced by the compiler pipeline.
-- **Atomic Reload Only** — Configuration transitions **MUST** occur via an all-or-nothing swap of the entire artifact. Partial updates, incremental edits, and mutable in-place structures are forbidden.
+- **Atomic Runtime Swap Only** — Runtime-safe configuration transitions **MUST** occur via an all-or-nothing swap of the live `RuntimeState`. Fields that require listener, admin, metrics, access-log, or other boot-time service wiring are not hot-reloadable in the current architecture and **MUST** be rejected during reload.
 - **Fail-Closed Semantics** — Any validation error, environment violation, or artifact incompatibility **MUST** leave the runtime serving the last-known-good state. There is no graceful degradation, no fallback heuristics, and no best-effort mode.
 
 ## 2. Semantic Boundary
@@ -20,7 +20,7 @@
 - Corruption, mismatched architecture, or unsupported version **MUST** cause immediate rejection and a return to LKG. The relay is responsible for monotonic versioning; the runtime never mutates artifacts and never attempts repair.
 
 ## 4. Failure Semantics
-- Reload success is binary. Either the artifact validates and atomically replaces the live state, or it is rejected with no partial side effects.
+- Reload success is binary. Either the runtime-safe portion validates and atomically replaces the live state, or the artifact is rejected with no partial side effects.
 - Rollback is not a special path; refusal to load a new artifact simply keeps the current state. Manual rollback requires resealing or redelivering a prior artifact.
 - LKG is authoritative. The runtime **MUST** keep serving the last applied artifact until a new artifact is proven valid. There is no shadow config, preview mode, or best-effort merging.
 - There is no graceful degradation. If upstreams fail or artifacts are invalid, the runtime fails-closed and surfaces explicit errors instead of improvising behavior.
@@ -37,6 +37,7 @@
 - `build()` wires telemetry, runtime state, resolver/health monitors, listeners, agents, and admin endpoints **before** any service is started; `run()` is the only place that calls `Server::run_forever`.
 - `listener::tls::TlsRuntime` owns TLS materialization. Pingora never sees raw paths—TLS certificates, private keys, and client-auth CA bundles are loaded and validated upfront with explicit `Optional`/`Required` semantics.
 - All listener, telemetry, and agent wiring happens before `run()` is invoked, guaranteeing that a failed dependency (cert unreadable, metrics bind failure, etc.) aborts the boot rather than partially starting services.
+- Because those services are boot-wired, reload currently rejects changes to `listeners`, `admin`, `shutdown`, `telemetry.metrics`, and `telemetry.access_log` instead of trying to mutate the service graph in place.
 
 ## 7. Request Telemetry & Identity Handling
 - **RequestTelemetry** — Every inbound request now owns a `RequestTelemetry` struct that captures the immutable `RequestId` plus the active tracing span. `RouterContext` delegates all span/state mutations to this struct so Pingora phases cannot forget to emit span updates or accidentally swap request ids mid-flight.
@@ -51,7 +52,7 @@
 
 ## 9. Materialized Runtime Config
 - Reload now produces a `MaterializedRuntimeConfig` that owns the pre-built router and upstream manager so request threads never see partially-constructed state.
-- `RuntimeState` deref-coerces to the materialized struct and only exposes a single `ConfigVersion` newtype (non-zero) so metrics/admin surfaces no longer juggle `Option<u64>` defaults.
+- `RuntimeState` deref-coerces to the materialized struct, while boot-time resources remain owned by the bootstrap layer. Config versioning is still surfaced through `Option<ConfigVersion>` during pre-LKG/bootstrap phases and becomes concrete after a versioned artifact is applied.
 - Upstream clusters are segmented into `cluster/state.rs`, `cluster/health.rs`, `cluster/pool.rs`, and `cluster/tls.rs` modules, which keeps load balancing, health tracking, pooling, and TLS materialization isolated behind explicit APIs.
 
 ## 10. Health Monitoring & TLS Reuse
@@ -63,4 +64,3 @@
 - Metrics recording flows through `MetricsRegistry`, a thin wrapper around the Prometheus handle. All request/cluster/telemetry helpers share this registry so instrumentation stays centralized.
 - The Prometheus endpoint is implemented by `telemetry::metrics::PrometheusEndpoint`, which depends on a pluggable `MetricsTransport`. The default transport binds a `tokio::net::TcpListener`, but tests can now inject custom transports without touching network sockets.
 - Metrics exporting remains best-effort: if recorder installation fails or the listener cannot bind, the endpoint logs the error once and metrics stay disabled rather than partially-initialized.
-

crates/pavis-relay/README.md

@@ -4,11 +4,11 @@ The central configuration distribution hub for the Pavis ecosystem.
 
 ## Purpose
 
-`pavis-relay` implements the control plane for configuration management, responsible for ingesting validated `.pvs` artifacts, assigning monotonic version numbers, and distributing updates to runtime instances via HTTP long-polling.
+`pavis-relay` implements the control plane for configuration management, responsible for ingesting `.pvs` artifacts that have already passed compiler-side validation, assigning monotonic version numbers, and distributing updates to runtime instances via HTTP long-polling.
 
 ## Responsibilities
 
-- Accepting and validating published configuration artifacts
+- Accepting published artifacts and verifying binary integrity
 - Assigning strictly monotonic version numbers
 - Maintaining Last Known Good (LKG) configuration on disk
 - Serving configurations via long-polling HTTP API

crates/pavis-relay/src/app.rs

@@ -111,9 +111,7 @@ fn build_codec(config: &PipelineConfig) -> Result<BoxedCodec> {
         config::CodecKind::Serde => {
             #[cfg(feature = "codec-serde")]
             {
-                Ok(Box::new(pavis_codec_serde::SerdeCodec {
-                    format: pavis_codec_serde::SerdeFormat::Yaml,
-                }))
+                Ok(Box::new(crate::codec::AutoSerdeCodec))
             }
             #[cfg(not(feature = "codec-serde"))]
             {

crates/pavis-relay/src/codec.rs

@@ -1,3 +1,65 @@
-use pavis_codec_api::{Codec, CodecError};
+use pavis_codec_api::{CheckedArtifact, Codec, CodecError};
+use pavis_core::RuntimeConfig;
+use pavis_ingest_api::{Artifact, Format};
 
 pub type BoxedCodec = Box<dyn Codec<Error = CodecError> + Send + Sync>;
+
+#[cfg(feature = "codec-serde")]
+#[derive(Debug, Default)]
+pub struct AutoSerdeCodec;
+
+#[cfg(feature = "codec-serde")]
+impl AutoSerdeCodec {
+    fn codec_for_format(format: Format) -> Result<pavis_codec_serde::SerdeCodec, CodecError> {
+        let serde_format = match format {
+            Format::Yaml => pavis_codec_serde::SerdeFormat::Yaml,
+            Format::Json => pavis_codec_serde::SerdeFormat::Json,
+            other => {
+                return Err(CodecError::Check(anyhow::anyhow!(
+                    "Unsupported format: {:?}",
+                    other
+                )));
+            }
+        };
+        Ok(pavis_codec_serde::SerdeCodec {
+            format: serde_format,
+        })
+    }
+}
+
+#[cfg(feature = "codec-serde")]
+impl Codec for AutoSerdeCodec {
+    type Error = CodecError;
+
+    fn check(&self, artifact: Artifact) -> Result<CheckedArtifact, Self::Error> {
+        Self::codec_for_format(artifact.format)?.check(artifact)
+    }
+
+    fn compile(&self, checked: &CheckedArtifact) -> Result<RuntimeConfig, Self::Error> {
+        Self::codec_for_format(checked.artifact.format)?.compile(checked)
+    }
+}
+
+#[cfg(all(test, feature = "codec-serde"))]
+mod tests {
+    use super::AutoSerdeCodec;
+    use pavis_codec_api::{Codec, CompactionLevel};
+    use pavis_ingest_api::{Artifact, Format, SourceInfo};
+    use std::path::PathBuf;
+
+    fn fixture_path(name: &str) -> PathBuf {
+        PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+            .join("../pavis-codec-serde/tests/fixtures")
+            .join(name)
+    }
+
+    #[test]
+    fn auto_codec_accepts_json_artifacts() {
+        let codec = AutoSerdeCodec;
+        let bytes = std::fs::read(fixture_path("minimal.json")).expect("read fixture");
+        let artifact = Artifact::new(bytes.into(), Format::Json, SourceInfo::unknown());
+
+        let validated = codec.materialize(artifact, CompactionLevel::Off);
+        assert!(validated.is_ok());
+    }
+}

crates/pavis/src/agent/driver.rs

@@ -23,6 +23,7 @@ use tokio::sync::watch;
 
 use crate::agent::fsm::{Effect, Event, Fsm, Response, VerifiedUpdate};
 use crate::agent::lkg::{load_lkg_config, tmp_path_for, write_atomic};
+use crate::reload::ensure_reload_safe;
 use crate::state::{RuntimeState, RuntimeStateHandle};
 use crate::telemetry::metrics::MetricsRegistry;
 use crate::validate_env::{self, RuntimeEnvError};
@@ -674,6 +675,8 @@ impl ConfigAgent {
         };
         let validated = unsafe { pavis_core::ValidatedRuntimeConfig::from_trusted(config) };
         let current = self.state.load();
+        ensure_reload_safe(&current.config, &validated)
+            .map_err(|err| self.record_validation_failure(err))?;
         if let Err(err) = validate_env::validate_runtime_env(&validated, Some(&current.config)) {
             let _ = tokio::fs::remove_file(&tmp_path).await;
             return Err(self.record_validation_failure(err.into()));
@@ -838,6 +841,29 @@ mod tests {
         ServiceName, Telemetry, TracingPolicy,
     };
     use pavis_pvs::PvsError;
+
+    fn test_validated_config(service_name: &str) -> pavis_core::ValidatedRuntimeConfig {
+        let listener = ListenerBuilder::new()
+            .name(ListenerName("test".to_string()))
+            .address("127.0.0.1:0".parse().unwrap())
+            .build()
+            .unwrap();
+        let config = RuntimeConfigBuilder::new()
+            .telemetry(Telemetry {
+                level: LogLevel::Info,
+                pingora: LogLevel::Info,
+                service_name: ServiceName(service_name.to_string()),
+                metrics: Metrics::Disabled,
+                access_log: AccessLogPolicy::Disabled,
+                tracing: TracingPolicy::Disabled,
+            })
+            .add_listener(listener)
+            .build()
+            .unwrap();
+
+        pavis_core::validate_runtime(config).unwrap()
+    }
+
     #[test]
     fn test_classify_validation_error() {
         let err = anyhow::anyhow!(PvsError::VersionMismatch {
@@ -1031,7 +1057,10 @@ mod tests {
     async fn test_config_agent_verify_update_success() {
         let dir = tempfile::tempdir().unwrap();
         let lkg = dir.path().join("lkg.pvs");
-        let state = Arc::new(RuntimeStateHandle::new(RuntimeState::default()));
+        let current = test_validated_config("current");
+        let state = Arc::new(RuntimeStateHandle::new(
+            RuntimeState::from_config(&current).unwrap(),
+        ));
         let agent = ConfigAgent::new(
             "http://localhost".to_string(),
             lkg,
@@ -1040,25 +1069,9 @@ mod tests {
         )
         .unwrap();
 
-        let listener = ListenerBuilder::new()
-            .name(ListenerName("test".to_string()))
-            .address("127.0.0.1:0".parse().unwrap())
-            .build()
-            .unwrap();
-        let config = RuntimeConfigBuilder::new()
-            .telemetry(Telemetry {
-                level: LogLevel::Info,
-                pingora: LogLevel::Info,
-                service_name: ServiceName("test".to_string()),
-                metrics: Metrics::Disabled,
-                access_log: AccessLogPolicy::Disabled,
-                tracing: TracingPolicy::Disabled,
-            })
-            .add_listener(listener)
-            .build()
-            .unwrap();
+        let config = test_validated_config("next");
 
-        let bytes = pavis_pvs::encode(&config).unwrap();
+        let bytes = pavis_pvs::encode(config.as_ref()).unwrap();
         let etag = checksum_for_bytes(&bytes);
 
         let update = agent
@@ -1277,25 +1290,10 @@ mod tests {
 
     #[tokio::test]
     async fn test_poll_once_updated() {
-        let listener = ListenerBuilder::new()
-            .name(ListenerName("test".to_string()))
-            .address("127.0.0.1:0".parse().unwrap())
-            .build()
-            .unwrap();
-        let config = RuntimeConfigBuilder::new()
-            .telemetry(Telemetry {
-                level: LogLevel::Info,
-                pingora: LogLevel::Info,
-                service_name: ServiceName("test".to_string()),
-                metrics: Metrics::Disabled,
-                access_log: AccessLogPolicy::Disabled,
-                tracing: TracingPolicy::Disabled,
-            })
-            .add_listener(listener)
-            .build()
-            .unwrap();
+        let current = test_validated_config("current");
+        let config = test_validated_config("next");
 
-        let bytes = pavis_pvs::encode(&config).unwrap();
+        let bytes = pavis_pvs::encode(config.as_ref()).unwrap();
         let etag = checksum_for_bytes(&bytes);
 
         let stub = RelayStub {
@@ -1307,7 +1305,9 @@ mod tests {
 
         let dir = tempfile::tempdir().unwrap();
         let lkg = dir.path().join("lkg.pvs");
-        let state = Arc::new(RuntimeStateHandle::new(RuntimeState::default()));
+        let state = Arc::new(RuntimeStateHandle::new(
+            RuntimeState::from_config(&current).unwrap(),
+        ));
         let agent = ConfigAgent::new(url, lkg, state, Duration::from_secs(1)).unwrap();
 
         let outcome = agent.poll_once(100).await.unwrap();
@@ -1341,7 +1341,11 @@ mod tests {
     async fn test_worker_run_loop_verify_and_apply() {
         let dir = tempfile::tempdir().unwrap();
         let lkg = dir.path().join("lkg.pvs");
-        let state = Arc::new(RuntimeStateHandle::new(RuntimeState::default()));
+        let current = test_validated_config("current");
+        let next = test_validated_config("next");
+        let state = Arc::new(RuntimeStateHandle::new(
+            RuntimeState::from_config(&current).unwrap(),
+        ));
         let agent = Arc::new(
             ConfigAgent::new(
                 "http://localhost".to_string(),
@@ -1352,25 +1356,7 @@ mod tests {
             .unwrap(),
         );
 
-        let listener = ListenerBuilder::new()
-            .name(ListenerName("test".to_string()))
-            .address("127.0.0.1:0".parse().unwrap())
-            .build()
-            .unwrap();
-        let config = RuntimeConfigBuilder::new()
-            .telemetry(Telemetry {
-                level: LogLevel::Info,
-                pingora: LogLevel::Info,
-                service_name: ServiceName("test".to_string()),
-                metrics: Metrics::Disabled,
-                access_log: AccessLogPolicy::Disabled,
-                tracing: TracingPolicy::Disabled,
-            })
-            .add_listener(listener)
-            .build()
-            .unwrap();
-
-        let bytes = pavis_pvs::encode(&config).unwrap();
+        let bytes = pavis_pvs::encode(next.as_ref()).unwrap();
         let etag = checksum_for_bytes(&bytes);
 
         let (shutdown_tx, mut shutdown_rx) = watch::channel(false);

crates/pavis/src/lib.rs

@@ -5,6 +5,7 @@ pub mod listener;
 pub mod load;
 pub mod proxy;
 pub mod regex_validator;
+pub mod reload;
 pub mod retry;
 pub mod router;
 pub mod shutdown;