Skip to main content

liminal_server/config/
validation.rs

1use std::collections::{BTreeMap, BTreeSet};
2use std::path::{Path, PathBuf};
3
4use crate::ServerError;
5
6use super::types::{LoadedSchema, SelfStreamsConfig, ServerConfig, ServiceProfile};
7
8/// Validates a fully loaded server configuration before startup.
9///
10/// Validation is intentionally limited to deterministic semantic checks and
11/// filesystem inspection. It does not bind sockets, connect to peers, or perform
12/// any other network I/O. Beyond the semantic checks it also resolves and loads
13/// each channel's `schema_ref` from disk (relative to `base_dir`, or verbatim for
14/// absolute paths), parses the JSON Schema document, and stores it on the channel
15/// so the channel can later be built with a real schema. A missing, unreadable, or
16/// non-JSON schema file is an accumulated validation error like any other.
17///
18/// `base_dir` is the directory the config file was loaded from; relative
19/// `schema_ref` paths resolve against it. When it is `None` (e.g. a config
20/// assembled in memory), relative paths resolve against the process working
21/// directory, so callers that construct a config directly should use absolute
22/// `schema_ref` paths.
23///
24/// # Errors
25///
26/// Returns [`ServerError::ConfigValidation`] containing all discovered validation
27/// errors when the configuration is not safe to use for startup.
28pub fn validate(config: &mut ServerConfig, base_dir: Option<&Path>) -> Result<(), ServerError> {
29    let mut errors = Vec::new();
30
31    validate_listen_address(config, &mut errors);
32    validate_health_listen_address(config, &mut errors);
33    // NO `validate_drain_timeout`. `drain_timeout_ms` has been ignored since
34    // 0.14.3 (see `ServerConfig::drain_timeout_ms`), and a refusal is an effect:
35    // a field that changes nothing must not be able to stop a startup.
36    validate_channels(config, &mut errors);
37    validate_routing_rules(config, &mut errors);
38    validate_persistence_path(config, &mut errors);
39    validate_cluster(config, &mut errors);
40    validate_auth(config, &mut errors);
41    validate_services(config, &mut errors);
42    validate_websocket(config, &mut errors);
43    config.limits.collect_errors(&mut errors);
44    validate_participant(config, &mut errors);
45    load_channel_schemas(config, base_dir, &mut errors);
46
47    if errors.is_empty() {
48        Ok(())
49    } else {
50        Err(ServerError::ConfigValidation {
51            message: errors.join("; "),
52        })
53    }
54}
55
56/// Resolves, reads, and parses each channel's `schema_ref`, storing the loaded
57/// document on the channel. Follows the same deterministic-local-FS discipline as
58/// [`validate_persistence_path`]: every failure is accumulated rather than
59/// short-circuiting, so an operator sees all schema problems at once.
60fn load_channel_schemas(
61    config: &mut ServerConfig,
62    base_dir: Option<&Path>,
63    errors: &mut Vec<String>,
64) {
65    for channel in &mut config.channels {
66        let Some(schema_ref) = channel.schema_ref.as_ref() else {
67            continue;
68        };
69        let path = resolve_schema_path(schema_ref, base_dir);
70        match load_schema_document(&path) {
71            Ok(loaded) => channel.loaded_schema = Some(loaded),
72            Err(reason) => errors.push(format!(
73                "channels.schema_ref '{}': {reason}",
74                schema_ref.display()
75            )),
76        }
77    }
78}
79
80/// Resolves a `schema_ref` to a concrete path: absolute refs are used verbatim,
81/// relative refs are joined onto `base_dir` (or the working directory when there
82/// is no base directory).
83fn resolve_schema_path(schema_ref: &Path, base_dir: Option<&Path>) -> PathBuf {
84    // `Path::join` returns `schema_ref` unchanged when it is absolute, so the
85    // base-dir arm covers both the absolute and relative cases.
86    base_dir.map_or_else(|| schema_ref.to_path_buf(), |dir| dir.join(schema_ref))
87}
88
89/// Reads, JSON-parses, and schema-compiles a schema file, returning the loaded
90/// document or a human-readable reason on failure (missing/unreadable file,
91/// invalid JSON, or valid JSON that is not a compilable JSON Schema). The
92/// compile check runs here so every schema problem surfaces in the accumulated
93/// validation pass instead of deferring to a different error class at channel
94/// construction.
95fn load_schema_document(path: &Path) -> Result<LoadedSchema, String> {
96    let bytes = std::fs::read(path)
97        .map_err(|error| format!("schema file '{}' is unreadable: {error}", path.display()))?;
98    let document: serde_json::Value = serde_json::from_slice(&bytes).map_err(|error| {
99        format!(
100            "schema file '{}' is not valid JSON: {error}",
101            path.display()
102        )
103    })?;
104    liminal::channel::Schema::new(document.clone()).map_err(|error| {
105        format!(
106            "schema file '{}' is not a valid JSON Schema: {error}",
107            path.display()
108        )
109    })?;
110    Ok(LoadedSchema { bytes, document })
111}
112
113fn validate_listen_address(config: &ServerConfig, errors: &mut Vec<String>) {
114    if config.listen_address.port() == 0 {
115        errors.push("listen_address: port must be non-zero".to_owned());
116    }
117}
118
119fn validate_health_listen_address(config: &ServerConfig, errors: &mut Vec<String>) {
120    if config.health_listen_address.port() == 0 {
121        errors.push("health_listen_address: port must be non-zero".to_owned());
122    }
123
124    if config.health_listen_address == config.listen_address {
125        errors.push(
126            "health_listen_address: must differ from listen_address for probe isolation".to_owned(),
127        );
128    } else if config.health_listen_address.port() == config.listen_address.port() {
129        errors.push(
130            "health_listen_address: port must differ from listen_address port for probe isolation"
131                .to_owned(),
132        );
133    }
134}
135
136fn validate_channels(config: &ServerConfig, errors: &mut Vec<String>) {
137    let mut seen = BTreeSet::new();
138    let mut duplicates = BTreeSet::new();
139
140    for channel in &config.channels {
141        let name = channel.name.trim();
142        if name.is_empty() {
143            errors.push("channels.name: channel name must not be empty".to_owned());
144            continue;
145        }
146
147        if !seen.insert(name.to_owned()) {
148            duplicates.insert(name.to_owned());
149        }
150
151        // SRV-012 R4: the console's channels are ephemeral and server-owned;
152        // configured durable they would write the store.
153        if channel.durable && crate::server::connection::is_console_subject(name) {
154            errors.push(format!(
155                "channels.durable: console channel '{name}' must not be durable"
156            ));
157        }
158    }
159
160    if !duplicates.is_empty() {
161        let names = duplicates.into_iter().collect::<Vec<_>>().join(", ");
162        errors.push(format!("channels.name: duplicate channel names: {names}"));
163    }
164}
165
166fn validate_routing_rules(config: &ServerConfig, errors: &mut Vec<String>) {
167    let channel_names = config
168        .channels
169        .iter()
170        .map(|channel| channel.name.as_str())
171        .collect::<BTreeSet<_>>();
172
173    for (index, rule) in config.routing_rules.iter().enumerate() {
174        let source = rule.source_channel.trim();
175        if source.is_empty() {
176            errors.push(format!(
177                "routing_rules[{index}].source_channel: source channel must not be empty"
178            ));
179        } else if !channel_names.contains(source) {
180            errors.push(format!(
181                "routing_rules[{index}].source_channel: unknown channel '{source}'"
182            ));
183        }
184
185        let target = rule.target_channel.trim();
186        if target.is_empty() {
187            errors.push(format!(
188                "routing_rules[{index}].target_channel: target channel must not be empty"
189            ));
190        } else if !channel_names.contains(target) {
191            errors.push(format!(
192                "routing_rules[{index}].target_channel: unknown channel '{target}'"
193            ));
194        }
195    }
196}
197
198fn validate_persistence_path(config: &ServerConfig, errors: &mut Vec<String>) {
199    let Some(path) = config.persistence_path.as_deref() else {
200        return;
201    };
202
203    match std::fs::metadata(path) {
204        Ok(metadata) => {
205            if !metadata.is_dir() {
206                errors.push(format!(
207                    "persistence_path '{}': path must be an existing directory",
208                    path.display()
209                ));
210            } else if metadata.permissions().readonly() {
211                errors.push(format!(
212                    "persistence_path '{}': path is not writable",
213                    path.display()
214                ));
215            }
216        }
217        Err(error) => {
218            errors.push(format!(
219                "persistence_path '{}': path is unreachable: {error}",
220                path.display()
221            ));
222        }
223    }
224}
225
226fn validate_cluster(config: &ServerConfig, errors: &mut Vec<String>) {
227    let Some(cluster) = config.cluster.as_ref() else {
228        return;
229    };
230
231    if cluster.node_name.trim().is_empty() {
232        errors.push("cluster.node_name: node name must not be empty".to_owned());
233    }
234
235    if cluster.cookie.is_empty() {
236        errors.push("cluster.cookie: distribution cookie must not be empty".to_owned());
237    }
238
239    if cluster.listen_address.port() == 0 {
240        errors.push("cluster.listen_address: distribution port must be non-zero".to_owned());
241    }
242
243    if cluster.listen_address == config.listen_address {
244        errors.push(
245            "cluster.listen_address: distribution port must differ from the client listen_address"
246                .to_owned(),
247        );
248    }
249
250    let mut seed_node_counts = BTreeMap::new();
251    for (index, seed_node) in cluster.seed_nodes.iter().enumerate() {
252        if seed_node.port() == 0 {
253            errors.push(format!(
254                "cluster.seed_nodes[{index}]: seed node port must be non-zero"
255            ));
256        }
257        seed_node_counts
258            .entry(seed_node.to_string())
259            .and_modify(|count| *count += 1)
260            .or_insert(1_usize);
261    }
262
263    let duplicates = seed_node_counts
264        .into_iter()
265        .filter_map(|(seed_node, count)| (count > 1).then_some(seed_node))
266        .collect::<Vec<_>>();
267
268    if !duplicates.is_empty() {
269        errors.push(format!(
270            "cluster.seed_nodes: duplicate seed nodes: {}",
271            duplicates.join(", ")
272        ));
273    }
274}
275
276/// Validates the optional `[auth]` section. When present its token must be
277/// non-empty: an empty token would gate nothing (every client's empty `auth_token`
278/// would match), so it is rejected rather than silently leaving the server open.
279/// The token is not trimmed — a shared secret may legitimately contain leading or
280/// trailing whitespace.
281fn validate_auth(config: &ServerConfig, errors: &mut Vec<String>) {
282    let Some(auth) = config.auth.as_ref() else {
283        return;
284    };
285
286    if auth.token.is_empty() {
287        errors.push("auth.token: authentication token must not be empty".to_owned());
288    }
289    // A token longer than a pre-admission `Connect` can carry could never be
290    // presented: every client would be refused before its token was compared.
291    let max = crate::server::connection::pre_admission::MAX_CONNECT_AUTH_TOKEN_BYTES;
292    if auth.token.len() > max {
293        errors.push(format!(
294            "auth.token: authentication token is {} bytes; a Connect can carry at most {max}",
295            auth.token.len()
296        ));
297    }
298    // ACCESS-004 R3: only a pass proves a holder, so requiring one without a
299    // pass verifier would refuse every worker.
300    if auth.require_verified_worker && auth.pass.is_none() {
301        errors.push(
302            "auth.require_verified_worker: requires [auth.pass], or no worker could register"
303                .to_owned(),
304        );
305    }
306}
307
308/// Validates the `[services]` profile selection.
309///
310/// An unrecognised `profile` value is a typed config validation error. When the
311/// profile is `worker-front-door`, config that asks for machinery the profile does
312/// not build is rejected via [`worker_front_door_field_errors`].
313impl SelfStreamsConfig {
314    /// Validates both stream bounds before service construction.
315    ///
316    /// # Errors
317    /// Returns a named configuration error for each zero bound.
318    pub fn validate(&self) -> Result<(), ServerError> {
319        let mut errors = Vec::new();
320        collect_self_stream_bound_errors(self, &mut errors);
321        if errors.is_empty() {
322            Ok(())
323        } else {
324            Err(ServerError::ConfigValidation {
325                message: errors.join("; "),
326            })
327        }
328    }
329}
330
331fn collect_self_stream_bound_errors(config: &SelfStreamsConfig, errors: &mut Vec<String>) {
332    errors.extend(
333        [
334            ("cost_rows", config.cost_rows),
335            ("events_rows", config.events_rows),
336        ]
337        .into_iter()
338        .filter(|(_, rows)| *rows == 0)
339        .map(|(name, _)| format!("services.self_streams.{name}: must be non-zero")),
340    );
341}
342
343fn validate_services(config: &ServerConfig, errors: &mut Vec<String>) {
344    collect_self_stream_bound_errors(&config.services.self_streams, errors);
345    let profile = match config.services.profile() {
346        Ok(profile) => profile,
347        Err(error) => {
348            // `profile()` only ever yields a `ConfigValidation` carrying the bare
349            // field message; surface it directly so it reads like every other
350            // accumulated error rather than the wrapped `Display` prefix.
351            match error {
352                crate::ServerError::ConfigValidation { message } => errors.push(message),
353                other => errors.push(other.to_string()),
354            }
355            return;
356        }
357    };
358
359    if profile == ServiceProfile::WorkerFrontDoor {
360        errors.extend(worker_front_door_field_errors(config));
361    }
362}
363
364/// Semantic checks for the optional `[websocket]` section (LP-WS-TRANSPORT R1.1).
365///
366/// The acceptor is explicit opt-in; when the section is present its listen
367/// address must be a usable, isolated port and its upgrade path must be a single
368/// exact absolute path. The origin allow-list FAILS CLOSED by design, so an
369/// absent or empty list is VALID configuration (it refuses every Origin-bearing
370/// upgrade); individual entries are still checked for obvious malformation so a
371/// typo'd entry cannot silently never match. A configured-but-zero keepalive
372/// interval is rejected: zero is not "disabled" (absence is) and would otherwise
373/// be an unbounded ping rate.
374fn validate_websocket(config: &ServerConfig, errors: &mut Vec<String>) {
375    let Some(websocket) = config.websocket.as_ref() else {
376        return;
377    };
378    validate_websocket_endpoint(config, websocket, errors);
379    validate_websocket_origins(websocket, errors);
380    validate_websocket_keepalive(websocket, errors);
381}
382
383/// Address and upgrade-path rules for the `[websocket]` section.
384fn validate_websocket_endpoint(
385    config: &ServerConfig,
386    websocket: &super::types::WebSocketConfig,
387    errors: &mut Vec<String>,
388) {
389    if websocket.listen_address.port() == 0 {
390        errors.push("websocket.listen_address: port must be non-zero".to_owned());
391    }
392    if websocket.listen_address == config.listen_address {
393        errors.push(
394            "websocket.listen_address: must differ from listen_address (the WebSocket \
395             acceptor is a sibling listener, not the main wire port)"
396                .to_owned(),
397        );
398    }
399    if websocket.listen_address == config.health_listen_address {
400        errors.push("websocket.listen_address: must differ from health_listen_address".to_owned());
401    }
402    if let Some(cluster) = config.cluster.as_ref() {
403        if websocket.listen_address == cluster.listen_address {
404            errors.push(
405                "websocket.listen_address: must differ from cluster.listen_address".to_owned(),
406            );
407        }
408    }
409
410    if websocket.path.is_empty() {
411        errors.push("websocket.path: upgrade path must not be empty".to_owned());
412    } else if !websocket.path.starts_with('/') {
413        errors.push("websocket.path: upgrade path must start with '/'".to_owned());
414    }
415    if websocket
416        .path
417        .chars()
418        .any(|character| character.is_ascii_whitespace() || character.is_ascii_control())
419    {
420        errors.push(
421            "websocket.path: upgrade path must not contain whitespace or control characters"
422                .to_owned(),
423        );
424    }
425    if websocket.path.contains(['?', '#']) {
426        errors.push(
427            "websocket.path: upgrade path is a single exact path and must not carry a query \
428             or fragment"
429                .to_owned(),
430        );
431    }
432}
433
434/// F6 allow-list entry hygiene: the list itself may be empty (fail closed),
435/// but a present entry must be a serialized origin that COULD ever match.
436fn validate_websocket_origins(websocket: &super::types::WebSocketConfig, errors: &mut Vec<String>) {
437    let mut seen_origins = BTreeSet::new();
438    for origin in &websocket.allowed_origins {
439        if origin.is_empty() {
440            errors.push("websocket.allowed_origins: origin entries must not be empty".to_owned());
441            continue;
442        }
443        if origin
444            .chars()
445            .any(|character| character.is_ascii_whitespace() || character.is_ascii_control())
446        {
447            errors.push(format!(
448                "websocket.allowed_origins: origin '{origin}' must not contain whitespace or \
449                 control characters"
450            ));
451        }
452        // `null` is the serialized opaque origin (sandboxed documents); every
453        // other legal entry is a scheme://host[:port] serialization with no
454        // trailing slash or path (RFC 6454 — the browser sends exactly that
455        // serialization, so anything else could never match).
456        if origin != "null" {
457            match origin.split_once("://") {
458                None => errors.push(format!(
459                    "websocket.allowed_origins: origin '{origin}' must be a serialized origin \
460                     (scheme://host[:port]) or the literal 'null'"
461                )),
462                Some((scheme, rest)) => {
463                    if scheme.is_empty() || rest.is_empty() {
464                        errors.push(format!(
465                            "websocket.allowed_origins: origin '{origin}' must name both a \
466                             scheme and a host"
467                        ));
468                    }
469                    if rest.contains('/') {
470                        errors.push(format!(
471                            "websocket.allowed_origins: origin '{origin}' must not contain a \
472                             path or trailing slash (a serialized Origin header never does)"
473                        ));
474                    }
475                }
476            }
477        }
478        if !seen_origins.insert(origin.as_str()) {
479            errors.push(format!(
480                "websocket.allowed_origins: duplicate origin '{origin}'"
481            ));
482        }
483    }
484}
485
486/// Q-A keepalive rules: zero is not "disabled" (absence is), and an extreme
487/// interval must refuse at validation rather than panic on clock arithmetic.
488fn validate_websocket_keepalive(
489    websocket: &super::types::WebSocketConfig,
490    errors: &mut Vec<String>,
491) {
492    match websocket.ping_interval_ms {
493        Some(0) => errors.push(
494            "websocket.ping_interval_ms: must be greater than zero when configured (omit the \
495             key to disable keepalive pings)"
496                .to_owned(),
497        ),
498        Some(interval_ms) => {
499            // S5 precedent: an extreme duration must be a typed refusal at
500            // validation, never a later monotonic-clock addition panic.
501            let interval = std::time::Duration::from_millis(interval_ms);
502            if std::time::Instant::now().checked_add(interval).is_none() {
503                errors.push(format!(
504                    "websocket.ping_interval_ms: {interval_ms} overflows the monotonic clock"
505                ));
506            }
507        }
508        None => {}
509    }
510}
511
512/// Semantic checks for the optional `[participant]` section: the shared
513/// nonzero/ordering rules plus the protocol codec's own minimum-frame check on
514/// `wire_frame_limit`, so an impossible limit fails at validation rather than
515/// at service construction.
516fn validate_participant(config: &ServerConfig, errors: &mut Vec<String>) {
517    let Some(participant) = config.participant.as_ref() else {
518        return;
519    };
520    participant.collect_errors(errors);
521    if participant.wire_frame_limit != 0
522        && let Err(error) = crate::server::participant::normalize_configured_frame_limit(
523            participant.wire_frame_limit,
524        )
525    {
526        errors.push(format!(
527            "participant.wire_frame_limit: {} is below the protocol's minimum complete \
528             participant frame ({error:?})",
529            participant.wire_frame_limit
530        ));
531    }
532}
533
534/// Cross-field checks for the worker-front-door profile: config that asks for
535/// machinery the profile does not build — channels, routing rules, a persistence
536/// path, or a cluster — is rejected rather than silently ignored. The front door
537/// constructs no channel, conversation, haematite, or distribution services, so
538/// honouring any of those keys is impossible and accepting them quietly would be a
539/// silent tradeoff.
540///
541/// Called from BOTH the file-loading validation pass ([`validate_services`]) and
542/// the runtime construction path
543/// ([`build_connection_services`](crate::server::connection::build_connection_services)),
544/// so a directly-constructed `ServerConfig` that skips file validation still cannot
545/// combine the worker profile with full-only machinery.
546pub(crate) fn worker_front_door_field_errors(config: &ServerConfig) -> Vec<String> {
547    let mut errors = Vec::new();
548    if !config.channels.is_empty() {
549        errors.push(
550            "services.profile: \"worker-front-door\" builds no channels; remove the \
551             [[channels]] entries or use profile = \"full\""
552                .to_owned(),
553        );
554    }
555    if !config.routing_rules.is_empty() {
556        errors.push(
557            "services.profile: \"worker-front-door\" builds no channels to route between; \
558             remove the [[routing_rules]] entries or use profile = \"full\""
559                .to_owned(),
560        );
561    }
562    if config.persistence_path.is_some() {
563        errors.push(
564            "services.profile: \"worker-front-door\" builds no durable store; remove \
565             persistence_path or use profile = \"full\""
566                .to_owned(),
567        );
568    }
569    if config.cluster.is_some() {
570        errors.push(
571            "services.profile: \"worker-front-door\" builds no channel cluster; remove the \
572             [cluster] section or use profile = \"full\""
573                .to_owned(),
574        );
575    }
576    if config.participant.is_some() {
577        errors.push(
578            "services.profile: \"worker-front-door\" installs no participant service; remove \
579             the [participant] section or use profile = \"full\""
580                .to_owned(),
581        );
582    }
583    errors
584}
585
586#[cfg(test)]
587mod tests {
588    use std::fs;
589    use std::net::SocketAddr;
590    use std::path::PathBuf;
591    use std::sync::atomic::{AtomicU64, Ordering};
592
593    use crate::ServerError;
594
595    use super::validate;
596    use crate::config::types::{
597        AuthConfig, ChannelDef, ClusterConfig, LimitsConfig, ParticipantConfig, RoutingRuleDef,
598        ServerConfig, ServicesConfig,
599    };
600
601    static NEXT_TEMP_DIR_ID: AtomicU64 = AtomicU64::new(0);
602
603    fn socket(address: &str) -> Result<SocketAddr, Box<dyn std::error::Error>> {
604        Ok(address.parse()?)
605    }
606
607    fn sample_config() -> Result<ServerConfig, Box<dyn std::error::Error>> {
608        Ok(ServerConfig {
609            listen_address: socket("127.0.0.1:8080")?,
610            health_listen_address: socket("127.0.0.1:8081")?,
611            drain_timeout_ms: 30_000,
612            channels: vec![ChannelDef {
613                name: "orders".to_owned(),
614                schema_ref: None,
615                durable: true,
616                loaded_schema: None,
617            }],
618            routing_rules: vec![RoutingRuleDef {
619                source_channel: "orders".to_owned(),
620                target_channel: "orders".to_owned(),
621                predicate: None,
622            }],
623            persistence_path: None,
624            cluster: Some(ClusterConfig {
625                node_name: "node-a".to_owned(),
626                listen_address: socket("127.0.0.1:9000")?,
627                seed_nodes: vec![socket("127.0.0.1:9001")?],
628                cookie: "test-cookie".to_owned(),
629            }),
630            auth: None,
631            services: ServicesConfig::default(),
632            limits: LimitsConfig::default(),
633            participant: None,
634            websocket: None,
635        })
636    }
637
638    /// A worker-front-door config: no channels, routing, persistence, or cluster —
639    /// the shape the front-door profile requires.
640    fn worker_front_door_config() -> Result<ServerConfig, Box<dyn std::error::Error>> {
641        Ok(ServerConfig {
642            channels: Vec::new(),
643            routing_rules: Vec::new(),
644            persistence_path: None,
645            cluster: None,
646            services: ServicesConfig {
647                profile: "worker-front-door".to_owned(),
648                ..ServicesConfig::default()
649            },
650            ..sample_config()?
651        })
652    }
653
654    fn unique_temp_dir(label: &str) -> PathBuf {
655        let id = NEXT_TEMP_DIR_ID.fetch_add(1, Ordering::Relaxed);
656        std::env::temp_dir().join(format!(
657            "liminal-server-validation-{label}-{}-{id}",
658            std::process::id()
659        ))
660    }
661
662    fn config_validation_message(result: Result<(), ServerError>) -> String {
663        let Err(ServerError::ConfigValidation { message }) = result else {
664            return String::new();
665        };
666        message
667    }
668
669    /// SRV-012 R4: a console channel configured durable is refused by name;
670    /// the same channel ephemeral, and an ordinary durable one, pass.
671    #[test]
672    fn a_durable_console_channel_is_refused_in_config() -> Result<(), Box<dyn std::error::Error>> {
673        let console = |durable| ChannelDef {
674            name: "console.topology".to_owned(),
675            schema_ref: None,
676            durable,
677            loaded_schema: None,
678        };
679        let mut config = sample_config()?;
680        config.channels.push(console(true));
681        let message = config_validation_message(validate(&mut config, None));
682        assert!(message.contains("console channel 'console.topology' must not be durable"));
683
684        let mut config = sample_config()?;
685        config.channels.push(console(false));
686        validate(&mut config, None)?;
687        Ok(())
688    }
689
690    #[test]
691    fn valid_config_passes_validation() -> Result<(), Box<dyn std::error::Error>> {
692        let mut config = sample_config()?;
693
694        validate(&mut config, None)?;
695
696        Ok(())
697    }
698
699    #[test]
700    fn invalid_listen_address_reports_field_name() -> Result<(), Box<dyn std::error::Error>> {
701        let mut config = sample_config()?;
702        config.listen_address = socket("127.0.0.1:0")?;
703
704        let message = config_validation_message(validate(&mut config, None));
705
706        assert!(message.contains("listen_address"));
707        assert!(message.contains("port"));
708
709        Ok(())
710    }
711
712    #[test]
713    fn invalid_health_listen_address_reports_field_name() -> Result<(), Box<dyn std::error::Error>>
714    {
715        let mut config = sample_config()?;
716        config.health_listen_address = socket("127.0.0.1:0")?;
717
718        let message = config_validation_message(validate(&mut config, None));
719
720        assert!(message.contains("health_listen_address"));
721        assert!(message.contains("port"));
722
723        Ok(())
724    }
725
726    #[test]
727    fn matching_health_and_main_listen_addresses_are_rejected()
728    -> Result<(), Box<dyn std::error::Error>> {
729        let mut config = sample_config()?;
730        config.health_listen_address = config.listen_address;
731
732        let message = config_validation_message(validate(&mut config, None));
733
734        assert!(message.contains("health_listen_address"));
735        assert!(message.contains("listen_address"));
736
737        Ok(())
738    }
739
740    #[test]
741    fn matching_health_and_main_listen_ports_are_rejected() -> Result<(), Box<dyn std::error::Error>>
742    {
743        let mut config = sample_config()?;
744        config.health_listen_address = socket("0.0.0.0:8080")?;
745
746        let message = config_validation_message(validate(&mut config, None));
747
748        assert!(message.contains("health_listen_address"));
749        assert!(message.contains("port"));
750
751        Ok(())
752    }
753
754    /// `drain_timeout_ms` has been ignored since 0.14.3, so it can no longer
755    /// refuse a startup — a field that changes nothing must not be able to stop
756    /// a server coming up. Zero loads exactly like any other value.
757    #[test]
758    fn zero_drain_timeout_is_accepted_because_the_field_is_ignored()
759    -> Result<(), Box<dyn std::error::Error>> {
760        let mut config = sample_config()?;
761        config.drain_timeout_ms = 0;
762
763        validate(&mut config, None)?;
764
765        assert_eq!(config.drain_timeout_ms, 0);
766
767        Ok(())
768    }
769
770    #[test]
771    fn duplicate_channel_names_are_listed() -> Result<(), Box<dyn std::error::Error>> {
772        let mut config = sample_config()?;
773        config.channels.push(ChannelDef {
774            name: "orders".to_owned(),
775            schema_ref: None,
776            durable: false,
777            loaded_schema: None,
778        });
779
780        let message = config_validation_message(validate(&mut config, None));
781
782        assert!(message.contains("duplicate"));
783        assert!(message.contains("orders"));
784
785        Ok(())
786    }
787
788    #[test]
789    fn unreachable_persistence_path_reports_path() -> Result<(), Box<dyn std::error::Error>> {
790        let mut config = sample_config()?;
791        let path = unique_temp_dir("missing");
792        config.persistence_path = Some(path.clone());
793
794        let message = config_validation_message(validate(&mut config, None));
795
796        assert!(message.contains("persistence_path"));
797        assert!(message.contains(&path.display().to_string()));
798
799        Ok(())
800    }
801
802    #[test]
803    fn file_persistence_path_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
804        let mut config = sample_config()?;
805        let path = unique_temp_dir("file");
806        fs::write(&path, "not a directory")?;
807        config.persistence_path = Some(path.clone());
808
809        let message = config_validation_message(validate(&mut config, None));
810        fs::remove_file(&path)?;
811
812        assert!(message.contains("persistence_path"));
813        assert!(message.contains("directory"));
814
815        Ok(())
816    }
817
818    #[test]
819    fn multiple_validation_errors_are_reported_together() -> Result<(), Box<dyn std::error::Error>>
820    {
821        let mut config = sample_config()?;
822        let missing_path = unique_temp_dir("multi-missing");
823        config.listen_address = socket("127.0.0.1:0")?;
824        config.channels.push(ChannelDef {
825            name: "orders".to_owned(),
826            schema_ref: None,
827            durable: false,
828            loaded_schema: None,
829        });
830        config.persistence_path = Some(missing_path.clone());
831
832        let message = config_validation_message(validate(&mut config, None));
833
834        assert!(message.contains("listen_address"));
835        assert!(message.contains("duplicate channel names: orders"));
836        assert!(message.contains(&missing_path.display().to_string()));
837
838        Ok(())
839    }
840
841    #[test]
842    fn routing_rules_reference_configured_channels() -> Result<(), Box<dyn std::error::Error>> {
843        let mut config = sample_config()?;
844        config.routing_rules[0].target_channel = "unknown".to_owned();
845
846        let message = config_validation_message(validate(&mut config, None));
847
848        assert!(message.contains("routing_rules[0].target_channel"));
849        assert!(message.contains("unknown"));
850
851        Ok(())
852    }
853
854    /// Writes `contents` to a fresh uniquely-named temp file and returns its path.
855    fn write_temp_schema(
856        label: &str,
857        contents: &str,
858    ) -> Result<PathBuf, Box<dyn std::error::Error>> {
859        let path = unique_temp_dir(label).with_extension("json");
860        fs::write(&path, contents)?;
861        Ok(path)
862    }
863
864    #[test]
865    fn absolute_schema_ref_is_loaded_and_parsed() -> Result<(), Box<dyn std::error::Error>> {
866        let schema = r#"{"type":"object","properties":{"id":{"type":"integer"}}}"#;
867        let schema_path = write_temp_schema("load-ok", schema)?;
868        let mut config = sample_config()?;
869        config.channels[0].schema_ref = Some(schema_path.clone());
870
871        let result = validate(&mut config, None);
872        fs::remove_file(&schema_path)?;
873        result?;
874
875        let loaded = config.channels[0]
876            .loaded_schema
877            .as_ref()
878            .ok_or("schema should have been loaded onto the channel")?;
879        assert_eq!(loaded.bytes, schema.as_bytes());
880        assert_eq!(
881            loaded.document.get("type").and_then(|t| t.as_str()),
882            Some("object")
883        );
884
885        Ok(())
886    }
887
888    #[test]
889    fn relative_schema_ref_resolves_against_base_dir() -> Result<(), Box<dyn std::error::Error>> {
890        let dir = unique_temp_dir("relative-base");
891        fs::create_dir_all(&dir)?;
892        let schema = r#"{"type":"object"}"#;
893        fs::write(dir.join("orders.json"), schema)?;
894
895        let mut config = sample_config()?;
896        config.channels[0].schema_ref = Some(PathBuf::from("orders.json"));
897
898        let result = validate(&mut config, Some(&dir));
899        fs::remove_dir_all(&dir)?;
900        result?;
901
902        assert!(config.channels[0].loaded_schema.is_some());
903
904        Ok(())
905    }
906
907    #[test]
908    fn missing_schema_ref_file_reports_validation_error() -> Result<(), Box<dyn std::error::Error>>
909    {
910        let missing = unique_temp_dir("missing-schema").with_extension("json");
911        let mut config = sample_config()?;
912        config.channels[0].schema_ref = Some(missing.clone());
913
914        let message = config_validation_message(validate(&mut config, None));
915
916        assert!(message.contains("schema_ref"));
917        assert!(message.contains(&missing.display().to_string()));
918        assert!(message.contains("unreadable"));
919
920        Ok(())
921    }
922
923    #[test]
924    fn invalid_json_schema_ref_reports_validation_error() -> Result<(), Box<dyn std::error::Error>>
925    {
926        let schema_path = write_temp_schema("bad-json", "{ this is not json")?;
927        let mut config = sample_config()?;
928        config.channels[0].schema_ref = Some(schema_path.clone());
929
930        let message = config_validation_message(validate(&mut config, None));
931        fs::remove_file(&schema_path)?;
932
933        assert!(message.contains("schema_ref"));
934        assert!(message.contains("not valid JSON"));
935
936        Ok(())
937    }
938
939    #[test]
940    fn valid_json_invalid_schema_ref_reports_validation_error()
941    -> Result<(), Box<dyn std::error::Error>> {
942        // Valid JSON that is not a compilable JSON Schema: a schema document
943        // must be an object, so a bare array parses but fails compilation.
944        let schema_path = write_temp_schema("bad-schema", "[]")?;
945        let mut config = sample_config()?;
946        config.channels[0].schema_ref = Some(schema_path.clone());
947
948        let message = config_validation_message(validate(&mut config, None));
949        fs::remove_file(&schema_path)?;
950
951        assert!(message.contains("schema_ref"));
952        assert!(message.contains("not a valid JSON Schema"));
953
954        Ok(())
955    }
956
957    #[test]
958    fn present_non_empty_auth_token_passes_validation() -> Result<(), Box<dyn std::error::Error>> {
959        let mut config = sample_config()?;
960        config.auth = Some(AuthConfig {
961            pass: None,
962            token: "s3cr3t".to_owned(),
963            require_verified_worker: false,
964        });
965
966        validate(&mut config, None)?;
967
968        Ok(())
969    }
970
971    #[test]
972    fn empty_auth_token_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
973        let mut config = sample_config()?;
974        config.auth = Some(AuthConfig {
975            pass: None,
976            token: String::new(),
977            require_verified_worker: false,
978        });
979
980        let message = config_validation_message(validate(&mut config, None));
981
982        assert!(message.contains("auth.token"));
983        assert!(message.contains("must not be empty"));
984
985        Ok(())
986    }
987
988    /// ACCESS-004 R3, red at main: requiring a verified holder with no
989    /// `[auth.pass]` would refuse every worker, so it is refused by name.
990    #[test]
991    fn requiring_a_verified_worker_without_a_pass_verifier_is_rejected()
992    -> Result<(), Box<dyn std::error::Error>> {
993        let mut config = sample_config()?;
994        config.auth = Some(AuthConfig {
995            pass: None,
996            token: "s3cr3t".to_owned(),
997            require_verified_worker: true,
998        });
999
1000        let message = config_validation_message(validate(&mut config, None));
1001
1002        assert!(
1003            message.contains("auth.require_verified_worker") && message.contains("auth.pass"),
1004            "the refusal must name the flag and the missing section: {message}"
1005        );
1006        Ok(())
1007    }
1008
1009    #[test]
1010    fn auth_token_longer_than_a_connect_can_carry_is_rejected()
1011    -> Result<(), Box<dyn std::error::Error>> {
1012        let max = crate::server::connection::pre_admission::MAX_CONNECT_AUTH_TOKEN_BYTES;
1013        let mut config = sample_config()?;
1014        config.auth = Some(AuthConfig {
1015            pass: None,
1016            token: "t".repeat(max),
1017            require_verified_worker: false,
1018        });
1019        validate(&mut config, None)?;
1020
1021        config.auth = Some(AuthConfig {
1022            pass: None,
1023            token: "t".repeat(max + 1),
1024            require_verified_worker: false,
1025        });
1026        let message = config_validation_message(validate(&mut config, None));
1027
1028        assert!(message.contains("auth.token"));
1029        assert!(message.contains(&format!("at most {max}")));
1030        Ok(())
1031    }
1032
1033    #[test]
1034    fn absent_auth_section_passes_validation() -> Result<(), Box<dyn std::error::Error>> {
1035        let mut config = sample_config()?;
1036        config.auth = None;
1037
1038        validate(&mut config, None)?;
1039
1040        Ok(())
1041    }
1042
1043    #[test]
1044    fn default_profile_is_full_and_passes_validation() -> Result<(), Box<dyn std::error::Error>> {
1045        let mut config = sample_config()?;
1046
1047        // The default services config resolves to the full profile.
1048        assert_eq!(
1049            config.services.profile()?,
1050            crate::config::types::ServiceProfile::Full
1051        );
1052        validate(&mut config, None)?;
1053
1054        Ok(())
1055    }
1056
1057    #[test]
1058    fn unknown_profile_is_a_validation_error() -> Result<(), Box<dyn std::error::Error>> {
1059        let mut config = sample_config()?;
1060        config.services = ServicesConfig {
1061            profile: "banana".to_owned(),
1062            ..ServicesConfig::default()
1063        };
1064
1065        let message = config_validation_message(validate(&mut config, None));
1066
1067        assert!(message.contains("services.profile"));
1068        assert!(message.contains("banana"));
1069        assert!(message.contains("worker-front-door"));
1070
1071        Ok(())
1072    }
1073
1074    #[test]
1075    fn worker_front_door_profile_with_empty_topology_passes()
1076    -> Result<(), Box<dyn std::error::Error>> {
1077        let mut config = worker_front_door_config()?;
1078
1079        assert_eq!(
1080            config.services.profile()?,
1081            crate::config::types::ServiceProfile::WorkerFrontDoor
1082        );
1083        validate(&mut config, None)?;
1084
1085        Ok(())
1086    }
1087
1088    #[test]
1089    fn worker_front_door_profile_rejects_channels_persistence_and_cluster()
1090    -> Result<(), Box<dyn std::error::Error>> {
1091        // Start from the full sample (channels + cluster present) but flip only the
1092        // profile: every full-mode-only knob must be rejected, not silently ignored.
1093        let mut config = sample_config()?;
1094        config.services = ServicesConfig {
1095            profile: "worker-front-door".to_owned(),
1096            ..ServicesConfig::default()
1097        };
1098        config.persistence_path = Some(PathBuf::from("/tmp"));
1099
1100        let message = config_validation_message(validate(&mut config, None));
1101
1102        assert!(message.contains("builds no channels"));
1103        assert!(message.contains("builds no durable store"));
1104        assert!(message.contains("builds no channel cluster"));
1105
1106        Ok(())
1107    }
1108
1109    #[test]
1110    fn default_limits_pass_validation_and_carry_signed_numbers()
1111    -> Result<(), Box<dyn std::error::Error>> {
1112        let mut config = sample_config()?;
1113        // The signed §5 defaults resolve from an absent `[limits]` section.
1114        assert_eq!(config.limits.max_connections, 256);
1115        assert_eq!(config.limits.max_subscriptions_per_connection, 32);
1116        assert_eq!(config.limits.max_conversations_per_connection, 32);
1117        assert_eq!(config.limits.max_pending_pushes_per_connection, 32);
1118        assert_eq!(
1119            config
1120                .limits
1121                .max_pending_conversation_replies_per_connection,
1122            32
1123        );
1124        assert_eq!(config.limits.max_pending_replies_per_conversation, 8);
1125        assert_eq!(config.limits.max_connection_inbox_bytes, 4 * 1024 * 1024);
1126        // The outbound budget is the inbound one's twin and carries the same
1127        // number: a connection's two directions are bounded alike, and the
1128        // inbound field's own documentation has always said so. Asserted beside
1129        // it so moving one without the other fails here.
1130        assert_eq!(config.limits.max_connection_outbound_bytes, 4 * 1024 * 1024);
1131        assert_eq!(
1132            config.limits.max_connection_outbound_bytes, config.limits.max_connection_inbox_bytes,
1133            "the two per-connection byte budgets are deliberately equal"
1134        );
1135        // P0 #55 part 2: 4096, not the §5-era 256. The two caps that bound a
1136        // subscription inbox are asserted TOGETHER here because the ruling is
1137        // about their RATIO, not either number: the byte budget must be the one
1138        // that binds for realistic records, which requires
1139        // `max_connection_inbox_bytes / max_subscription_inbox_depth` to sit at or
1140        // below the record sizes real traffic carries. Changing either constant
1141        // without the other moves that crossover, so this pin fails on both.
1142        assert_eq!(config.limits.max_subscription_inbox_depth, 4096);
1143        assert_eq!(
1144            config.limits.max_connection_inbox_bytes / config.limits.max_subscription_inbox_depth,
1145            1024,
1146            "the byte-vs-count crossover must stay at 1 KiB per record"
1147        );
1148        assert_eq!(config.limits.delivery_slice_budget, 32);
1149        validate(&mut config, None)?;
1150        Ok(())
1151    }
1152
1153    /// §5 cap-refusal (config half): every zero cap is a typed config validation
1154    /// error — the unlimited-by-silence state §5 outlaws — reported by field name.
1155    #[test]
1156    fn zero_limits_are_typed_config_errors() -> Result<(), Box<dyn std::error::Error>> {
1157        type LimitMutator = (&'static str, fn(&mut ServerConfig));
1158        let mutators: [LimitMutator; 10] = [
1159            ("max_connections", |c| c.limits.max_connections = 0),
1160            ("max_subscriptions_per_connection", |c| {
1161                c.limits.max_subscriptions_per_connection = 0;
1162            }),
1163            ("max_conversations_per_connection", |c| {
1164                c.limits.max_conversations_per_connection = 0;
1165            }),
1166            ("max_pending_pushes_per_connection", |c| {
1167                c.limits.max_pending_pushes_per_connection = 0;
1168            }),
1169            ("max_pending_conversation_replies_per_connection", |c| {
1170                c.limits.max_pending_conversation_replies_per_connection = 0;
1171            }),
1172            ("max_pending_replies_per_conversation", |c| {
1173                c.limits.max_pending_replies_per_conversation = 0;
1174            }),
1175            ("max_connection_inbox_bytes", |c| {
1176                c.limits.max_connection_inbox_bytes = 0;
1177            }),
1178            ("max_connection_outbound_bytes", |c| {
1179                c.limits.max_connection_outbound_bytes = 0;
1180            }),
1181            ("max_subscription_inbox_depth", |c| {
1182                c.limits.max_subscription_inbox_depth = 0;
1183            }),
1184            ("delivery_slice_budget", |c| {
1185                c.limits.delivery_slice_budget = 0;
1186            }),
1187        ];
1188        for (field, mutate) in mutators {
1189            let mut config = sample_config()?;
1190            mutate(&mut config);
1191            let message = config_validation_message(validate(&mut config, None));
1192            assert!(
1193                message.contains(&format!("limits.{field}")),
1194                "zero {field} must report a typed limits.{field} error, got: {message}"
1195            );
1196            assert!(
1197                message.contains("greater than zero"),
1198                "the {field} refusal must say why: {message}"
1199            );
1200        }
1201        Ok(())
1202    }
1203
1204    /// A complete participant section with deployment-plausible nonzero values.
1205    const fn sample_participant() -> ParticipantConfig {
1206        ParticipantConfig {
1207            wire_frame_limit: 65_536,
1208            attach_receipt_ttl_ms: 60_000,
1209            receipt_provenance_ttl_ms: 600_000,
1210            live_receipt_server_report_threshold: 1_024,
1211            max_live_attach_receipts_per_participant: 8,
1212            receipt_provenance_server_report_threshold: 4_096,
1213            receipt_provenance_per_conversation_report_threshold: 256,
1214            max_receipt_provenance_per_participant: 64,
1215            max_retired_identity_slots_server: 1_024,
1216            identity_slots: 4,
1217            observer_recovery_max_entries: 64,
1218            max_semantic_conversations_per_connection: 32,
1219            max_ordinary_record_entries: 1,
1220            max_ordinary_record_bytes: 131_072,
1221            max_generated_marker_entries: 1,
1222            max_generated_marker_bytes: 4_096,
1223            mandatory_transaction_bound_entries: 4,
1224            mandatory_transaction_bound_bytes: 16_384,
1225            full_recovery_claim_entries: 4,
1226            full_recovery_claim_bytes: 16_384,
1227            retained_capacity_entries: 2_048,
1228            retained_capacity_bytes: 16_777_216,
1229            max_retained_record_rows: 1_024,
1230            closure_episode_churn_limit: 1_024,
1231        }
1232    }
1233
1234    #[test]
1235    fn valid_participant_section_passes_validation() -> Result<(), Box<dyn std::error::Error>> {
1236        let mut config = sample_config()?;
1237        config.participant = Some(sample_participant());
1238        let temp_dir = std::env::temp_dir().join(format!(
1239            "liminal-server-participant-validation-{}-{}",
1240            std::process::id(),
1241            NEXT_TEMP_DIR_ID.fetch_add(1, Ordering::Relaxed)
1242        ));
1243        fs::create_dir_all(&temp_dir)?;
1244        config.persistence_path = Some(temp_dir.clone());
1245        let result = validate(&mut config, None);
1246        fs::remove_dir_all(&temp_dir)?;
1247        assert!(result.is_ok(), "expected valid config, got {result:?}");
1248        Ok(())
1249    }
1250
1251    #[test]
1252    fn zero_participant_values_are_typed_config_errors() -> Result<(), Box<dyn std::error::Error>> {
1253        type ParticipantMutator = (&'static str, fn(&mut ParticipantConfig));
1254        let mutators: [ParticipantMutator; 12] = [
1255            ("wire_frame_limit", |p| p.wire_frame_limit = 0),
1256            ("attach_receipt_ttl_ms", |p| p.attach_receipt_ttl_ms = 0),
1257            ("receipt_provenance_ttl_ms", |p| {
1258                p.receipt_provenance_ttl_ms = 0;
1259            }),
1260            ("live_receipt_server_report_threshold", |p| {
1261                p.live_receipt_server_report_threshold = 0;
1262            }),
1263            ("max_live_attach_receipts_per_participant", |p| {
1264                p.max_live_attach_receipts_per_participant = 0;
1265            }),
1266            ("receipt_provenance_server_report_threshold", |p| {
1267                p.receipt_provenance_server_report_threshold = 0;
1268            }),
1269            (
1270                "receipt_provenance_per_conversation_report_threshold",
1271                |p| {
1272                    p.receipt_provenance_per_conversation_report_threshold = 0;
1273                },
1274            ),
1275            ("max_receipt_provenance_per_participant", |p| {
1276                p.max_receipt_provenance_per_participant = 0;
1277            }),
1278            ("max_retired_identity_slots_server", |p| {
1279                p.max_retired_identity_slots_server = 0;
1280            }),
1281            ("identity_slots", |p| p.identity_slots = 0),
1282            ("observer_recovery_max_entries", |p| {
1283                p.observer_recovery_max_entries = 0;
1284            }),
1285            ("max_semantic_conversations_per_connection", |p| {
1286                p.max_semantic_conversations_per_connection = 0;
1287            }),
1288        ];
1289        for (field, mutate) in mutators {
1290            let mut config = sample_config()?;
1291            let mut participant = sample_participant();
1292            mutate(&mut participant);
1293            config.participant = Some(participant);
1294            let message = config_validation_message(validate(&mut config, None));
1295            assert!(
1296                message.contains(&format!("participant.{field}")),
1297                "expected typed error for participant.{field}, got: {message}"
1298            );
1299        }
1300        Ok(())
1301    }
1302
1303    #[test]
1304    fn provenance_ttl_shorter_than_receipt_is_a_typed_config_error()
1305    -> Result<(), Box<dyn std::error::Error>> {
1306        let mut config = sample_config()?;
1307        let mut participant = sample_participant();
1308        participant.receipt_provenance_ttl_ms = participant.attach_receipt_ttl_ms - 1;
1309        config.participant = Some(participant);
1310        let message = config_validation_message(validate(&mut config, None));
1311        assert!(
1312            message.contains("participant.receipt_provenance_ttl_ms"),
1313            "expected TTL-ordering error, got: {message}"
1314        );
1315        Ok(())
1316    }
1317
1318    #[test]
1319    fn undersized_wire_frame_limit_is_a_typed_config_error()
1320    -> Result<(), Box<dyn std::error::Error>> {
1321        let mut config = sample_config()?;
1322        let mut participant = sample_participant();
1323        participant.wire_frame_limit = 1;
1324        config.participant = Some(participant);
1325        let message = config_validation_message(validate(&mut config, None));
1326        assert!(
1327            message.contains("participant.wire_frame_limit"),
1328            "expected minimum-frame error, got: {message}"
1329        );
1330        Ok(())
1331    }
1332
1333    #[test]
1334    fn worker_front_door_rejects_participant_section() -> Result<(), Box<dyn std::error::Error>> {
1335        let mut config = worker_front_door_config()?;
1336        config.participant = Some(sample_participant());
1337        let message = config_validation_message(validate(&mut config, None));
1338        assert!(
1339            message.contains("installs no participant service"),
1340            "expected worker-front-door participant rejection, got: {message}"
1341        );
1342        Ok(())
1343    }
1344
1345    // ---- LP-WS-TRANSPORT R1.1: [websocket] section validation ----
1346
1347    fn sample_websocket()
1348    -> Result<crate::config::types::WebSocketConfig, Box<dyn std::error::Error>> {
1349        Ok(crate::config::types::WebSocketConfig {
1350            listen_address: socket("127.0.0.1:8082")?,
1351            path: "/liminal".to_owned(),
1352            allowed_origins: vec!["https://app.example.com".to_owned()],
1353            ping_interval_ms: Some(30_000),
1354        })
1355    }
1356
1357    #[test]
1358    fn valid_websocket_section_passes_validation() -> Result<(), Box<dyn std::error::Error>> {
1359        let mut config = sample_config()?;
1360        config.websocket = Some(sample_websocket()?);
1361        validate(&mut config, None)?;
1362        Ok(())
1363    }
1364
1365    #[test]
1366    fn websocket_empty_origin_list_is_valid_fail_closed_configuration()
1367    -> Result<(), Box<dyn std::error::Error>> {
1368        let mut config = sample_config()?;
1369        let mut websocket = sample_websocket()?;
1370        websocket.allowed_origins = Vec::new();
1371        config.websocket = Some(websocket);
1372        validate(&mut config, None)?;
1373        Ok(())
1374    }
1375
1376    #[test]
1377    fn websocket_zero_port_is_a_typed_config_error() -> Result<(), Box<dyn std::error::Error>> {
1378        let mut config = sample_config()?;
1379        let mut websocket = sample_websocket()?;
1380        websocket.listen_address = socket("127.0.0.1:0")?;
1381        config.websocket = Some(websocket);
1382        let message = config_validation_message(validate(&mut config, None));
1383        assert!(
1384            message.contains("websocket.listen_address"),
1385            "expected websocket.listen_address error, got: {message}"
1386        );
1387        Ok(())
1388    }
1389
1390    #[test]
1391    fn websocket_address_conflicts_are_typed_config_errors()
1392    -> Result<(), Box<dyn std::error::Error>> {
1393        for conflict in ["127.0.0.1:8080", "127.0.0.1:8081", "127.0.0.1:9000"] {
1394            let mut config = sample_config()?;
1395            let mut websocket = sample_websocket()?;
1396            websocket.listen_address = socket(conflict)?;
1397            config.websocket = Some(websocket);
1398            let message = config_validation_message(validate(&mut config, None));
1399            assert!(
1400                message.contains("websocket.listen_address: must differ from"),
1401                "expected an address-conflict error for {conflict}, got: {message}"
1402            );
1403        }
1404        Ok(())
1405    }
1406
1407    #[test]
1408    fn websocket_path_rules_are_typed_config_errors() -> Result<(), Box<dyn std::error::Error>> {
1409        for (bad_path, expectation) in [
1410            ("", "must not be empty"),
1411            ("liminal", "must start with '/'"),
1412            ("/limi nal", "whitespace"),
1413            ("/liminal?x=1", "query"),
1414            ("/liminal#frag", "query"),
1415        ] {
1416            let mut config = sample_config()?;
1417            let mut websocket = sample_websocket()?;
1418            websocket.path = bad_path.to_owned();
1419            config.websocket = Some(websocket);
1420            let message = config_validation_message(validate(&mut config, None));
1421            assert!(
1422                message.contains("websocket.path") && message.contains(expectation),
1423                "expected websocket.path '{bad_path}' to report '{expectation}', got: {message}"
1424            );
1425        }
1426        Ok(())
1427    }
1428
1429    #[test]
1430    fn websocket_malformed_origin_entries_are_typed_config_errors()
1431    -> Result<(), Box<dyn std::error::Error>> {
1432        for (bad_origin, expectation) in [
1433            ("", "must not be empty"),
1434            ("app.example.com", "serialized origin"),
1435            ("https://app.example.com/", "path or trailing slash"),
1436            ("https://app.example.com/app", "path or trailing slash"),
1437            ("https:// app.example.com", "whitespace"),
1438        ] {
1439            let mut config = sample_config()?;
1440            let mut websocket = sample_websocket()?;
1441            websocket.allowed_origins = vec![bad_origin.to_owned()];
1442            config.websocket = Some(websocket);
1443            let message = config_validation_message(validate(&mut config, None));
1444            assert!(
1445                message.contains("websocket.allowed_origins") && message.contains(expectation),
1446                "expected origin '{bad_origin}' to report '{expectation}', got: {message}"
1447            );
1448        }
1449        Ok(())
1450    }
1451
1452    #[test]
1453    fn websocket_duplicate_origin_is_a_typed_config_error() -> Result<(), Box<dyn std::error::Error>>
1454    {
1455        let mut config = sample_config()?;
1456        let mut websocket = sample_websocket()?;
1457        websocket.allowed_origins = vec![
1458            "https://app.example.com".to_owned(),
1459            "https://app.example.com".to_owned(),
1460        ];
1461        config.websocket = Some(websocket);
1462        let message = config_validation_message(validate(&mut config, None));
1463        assert!(
1464            message.contains("duplicate origin"),
1465            "expected a duplicate-origin error, got: {message}"
1466        );
1467        Ok(())
1468    }
1469
1470    #[test]
1471    fn websocket_zero_ping_interval_is_a_typed_config_error()
1472    -> Result<(), Box<dyn std::error::Error>> {
1473        let mut config = sample_config()?;
1474        let mut websocket = sample_websocket()?;
1475        websocket.ping_interval_ms = Some(0);
1476        config.websocket = Some(websocket);
1477        let message = config_validation_message(validate(&mut config, None));
1478        assert!(
1479            message.contains("websocket.ping_interval_ms"),
1480            "expected a zero-interval error, got: {message}"
1481        );
1482        Ok(())
1483    }
1484
1485    #[test]
1486    fn websocket_absent_ping_interval_means_disabled_and_valid()
1487    -> Result<(), Box<dyn std::error::Error>> {
1488        let mut config = sample_config()?;
1489        let mut websocket = sample_websocket()?;
1490        websocket.ping_interval_ms = None;
1491        config.websocket = Some(websocket);
1492        validate(&mut config, None)?;
1493        Ok(())
1494    }
1495}
1496
1497#[cfg(test)]
1498mod self_stream_tests {
1499    use super::super::types::ServicesConfig;
1500
1501    #[test]
1502    fn self_stream_bounds_default_and_override() -> Result<(), Box<dyn std::error::Error>> {
1503        let defaults: ServicesConfig = toml::from_str("")?;
1504        assert_eq!(defaults.self_streams.cost_rows, 10_000);
1505        assert_eq!(defaults.self_streams.events_rows, 100_000);
1506        let configured: ServicesConfig =
1507            toml::from_str("[self_streams]\ncost_rows = 3\nevents_rows = 5\n")?;
1508        assert_eq!(configured.self_streams.cost_rows, 3);
1509        assert_eq!(configured.self_streams.events_rows, 5);
1510        configured.profile()?;
1511        Ok(())
1512    }
1513
1514    #[test]
1515    fn self_stream_bounds_validate_without_participant_and_independently_of_profile()
1516    -> Result<(), Box<dyn std::error::Error>> {
1517        let mut config: super::ServerConfig = toml::from_str(
1518            "listen_address = '127.0.0.1:8080'\nhealth_listen_address = '127.0.0.1:8081'\ndrain_timeout_ms = 0\nchannels = []\nrouting_rules = []\n[services.self_streams]\ncost_rows = 0\nevents_rows = 0\n",
1519        )?;
1520        assert!(config.participant.is_none());
1521        assert_eq!(config.services.profile()?, super::ServiceProfile::Full);
1522        let error = super::validate(&mut config, None)
1523            .err()
1524            .ok_or("disabled self streams accepted invalid bounds")?;
1525        let message = error.to_string();
1526        assert!(message.contains("services.self_streams.cost_rows"));
1527        assert!(message.contains("services.self_streams.events_rows"));
1528        config.services.profile = "invalid".to_owned();
1529        let error = super::validate(&mut config, None)
1530            .err()
1531            .ok_or("invalid profile and bounds were accepted")?;
1532        let message = error.to_string();
1533        assert!(message.contains("services.profile"));
1534        assert!(message.contains("services.self_streams.cost_rows"));
1535        assert!(message.contains("services.self_streams.events_rows"));
1536        Ok(())
1537    }
1538
1539    #[test]
1540    fn zero_self_stream_bounds_are_named() -> Result<(), Box<dyn std::error::Error>> {
1541        let configured: ServicesConfig =
1542            toml::from_str("[self_streams]\ncost_rows = 0\nevents_rows = 0\n")?;
1543        let Err(error) = configured.self_streams.validate() else {
1544            return Err("zero row bounds must be refused".into());
1545        };
1546        let message = error.to_string();
1547        assert!(message.contains("services.self_streams.cost_rows"));
1548        assert!(message.contains("services.self_streams.events_rows"));
1549        Ok(())
1550    }
1551}