1use std::collections::{BTreeMap, BTreeSet};
2use std::path::{Path, PathBuf};
3
4use crate::ServerError;
5
6use super::types::{LoadedSchema, ServerConfig, ServiceProfile};
7
8pub 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 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
56fn 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
80fn resolve_schema_path(schema_ref: &Path, base_dir: Option<&Path>) -> PathBuf {
84 base_dir.map_or_else(|| schema_ref.to_path_buf(), |dir| dir.join(schema_ref))
87}
88
89fn 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
152 if !duplicates.is_empty() {
153 let names = duplicates.into_iter().collect::<Vec<_>>().join(", ");
154 errors.push(format!("channels.name: duplicate channel names: {names}"));
155 }
156}
157
158fn validate_routing_rules(config: &ServerConfig, errors: &mut Vec<String>) {
159 let channel_names = config
160 .channels
161 .iter()
162 .map(|channel| channel.name.as_str())
163 .collect::<BTreeSet<_>>();
164
165 for (index, rule) in config.routing_rules.iter().enumerate() {
166 let source = rule.source_channel.trim();
167 if source.is_empty() {
168 errors.push(format!(
169 "routing_rules[{index}].source_channel: source channel must not be empty"
170 ));
171 } else if !channel_names.contains(source) {
172 errors.push(format!(
173 "routing_rules[{index}].source_channel: unknown channel '{source}'"
174 ));
175 }
176
177 let target = rule.target_channel.trim();
178 if target.is_empty() {
179 errors.push(format!(
180 "routing_rules[{index}].target_channel: target channel must not be empty"
181 ));
182 } else if !channel_names.contains(target) {
183 errors.push(format!(
184 "routing_rules[{index}].target_channel: unknown channel '{target}'"
185 ));
186 }
187 }
188}
189
190fn validate_persistence_path(config: &ServerConfig, errors: &mut Vec<String>) {
191 let Some(path) = config.persistence_path.as_deref() else {
192 return;
193 };
194
195 match std::fs::metadata(path) {
196 Ok(metadata) => {
197 if !metadata.is_dir() {
198 errors.push(format!(
199 "persistence_path '{}': path must be an existing directory",
200 path.display()
201 ));
202 } else if metadata.permissions().readonly() {
203 errors.push(format!(
204 "persistence_path '{}': path is not writable",
205 path.display()
206 ));
207 }
208 }
209 Err(error) => {
210 errors.push(format!(
211 "persistence_path '{}': path is unreachable: {error}",
212 path.display()
213 ));
214 }
215 }
216}
217
218fn validate_cluster(config: &ServerConfig, errors: &mut Vec<String>) {
219 let Some(cluster) = config.cluster.as_ref() else {
220 return;
221 };
222
223 if cluster.node_name.trim().is_empty() {
224 errors.push("cluster.node_name: node name must not be empty".to_owned());
225 }
226
227 if cluster.cookie.is_empty() {
228 errors.push("cluster.cookie: distribution cookie must not be empty".to_owned());
229 }
230
231 if cluster.listen_address.port() == 0 {
232 errors.push("cluster.listen_address: distribution port must be non-zero".to_owned());
233 }
234
235 if cluster.listen_address == config.listen_address {
236 errors.push(
237 "cluster.listen_address: distribution port must differ from the client listen_address"
238 .to_owned(),
239 );
240 }
241
242 let mut seed_node_counts = BTreeMap::new();
243 for (index, seed_node) in cluster.seed_nodes.iter().enumerate() {
244 if seed_node.port() == 0 {
245 errors.push(format!(
246 "cluster.seed_nodes[{index}]: seed node port must be non-zero"
247 ));
248 }
249 seed_node_counts
250 .entry(seed_node.to_string())
251 .and_modify(|count| *count += 1)
252 .or_insert(1_usize);
253 }
254
255 let duplicates = seed_node_counts
256 .into_iter()
257 .filter_map(|(seed_node, count)| (count > 1).then_some(seed_node))
258 .collect::<Vec<_>>();
259
260 if !duplicates.is_empty() {
261 errors.push(format!(
262 "cluster.seed_nodes: duplicate seed nodes: {}",
263 duplicates.join(", ")
264 ));
265 }
266}
267
268fn validate_auth(config: &ServerConfig, errors: &mut Vec<String>) {
274 let Some(auth) = config.auth.as_ref() else {
275 return;
276 };
277
278 if auth.token.is_empty() {
279 errors.push("auth.token: authentication token must not be empty".to_owned());
280 }
281}
282
283fn validate_services(config: &ServerConfig, errors: &mut Vec<String>) {
289 let profile = match config.services.profile() {
290 Ok(profile) => profile,
291 Err(error) => {
292 match error {
296 crate::ServerError::ConfigValidation { message } => errors.push(message),
297 other => errors.push(other.to_string()),
298 }
299 return;
300 }
301 };
302
303 if profile == ServiceProfile::WorkerFrontDoor {
304 errors.extend(worker_front_door_field_errors(config));
305 }
306}
307
308fn validate_websocket(config: &ServerConfig, errors: &mut Vec<String>) {
319 let Some(websocket) = config.websocket.as_ref() else {
320 return;
321 };
322 validate_websocket_endpoint(config, websocket, errors);
323 validate_websocket_origins(websocket, errors);
324 validate_websocket_keepalive(websocket, errors);
325}
326
327fn validate_websocket_endpoint(
329 config: &ServerConfig,
330 websocket: &super::types::WebSocketConfig,
331 errors: &mut Vec<String>,
332) {
333 if websocket.listen_address.port() == 0 {
334 errors.push("websocket.listen_address: port must be non-zero".to_owned());
335 }
336 if websocket.listen_address == config.listen_address {
337 errors.push(
338 "websocket.listen_address: must differ from listen_address (the WebSocket \
339 acceptor is a sibling listener, not the main wire port)"
340 .to_owned(),
341 );
342 }
343 if websocket.listen_address == config.health_listen_address {
344 errors.push("websocket.listen_address: must differ from health_listen_address".to_owned());
345 }
346 if let Some(cluster) = config.cluster.as_ref() {
347 if websocket.listen_address == cluster.listen_address {
348 errors.push(
349 "websocket.listen_address: must differ from cluster.listen_address".to_owned(),
350 );
351 }
352 }
353
354 if websocket.path.is_empty() {
355 errors.push("websocket.path: upgrade path must not be empty".to_owned());
356 } else if !websocket.path.starts_with('/') {
357 errors.push("websocket.path: upgrade path must start with '/'".to_owned());
358 }
359 if websocket
360 .path
361 .chars()
362 .any(|character| character.is_ascii_whitespace() || character.is_ascii_control())
363 {
364 errors.push(
365 "websocket.path: upgrade path must not contain whitespace or control characters"
366 .to_owned(),
367 );
368 }
369 if websocket.path.contains(['?', '#']) {
370 errors.push(
371 "websocket.path: upgrade path is a single exact path and must not carry a query \
372 or fragment"
373 .to_owned(),
374 );
375 }
376}
377
378fn validate_websocket_origins(websocket: &super::types::WebSocketConfig, errors: &mut Vec<String>) {
381 let mut seen_origins = BTreeSet::new();
382 for origin in &websocket.allowed_origins {
383 if origin.is_empty() {
384 errors.push("websocket.allowed_origins: origin entries must not be empty".to_owned());
385 continue;
386 }
387 if origin
388 .chars()
389 .any(|character| character.is_ascii_whitespace() || character.is_ascii_control())
390 {
391 errors.push(format!(
392 "websocket.allowed_origins: origin '{origin}' must not contain whitespace or \
393 control characters"
394 ));
395 }
396 if origin != "null" {
401 match origin.split_once("://") {
402 None => errors.push(format!(
403 "websocket.allowed_origins: origin '{origin}' must be a serialized origin \
404 (scheme://host[:port]) or the literal 'null'"
405 )),
406 Some((scheme, rest)) => {
407 if scheme.is_empty() || rest.is_empty() {
408 errors.push(format!(
409 "websocket.allowed_origins: origin '{origin}' must name both a \
410 scheme and a host"
411 ));
412 }
413 if rest.contains('/') {
414 errors.push(format!(
415 "websocket.allowed_origins: origin '{origin}' must not contain a \
416 path or trailing slash (a serialized Origin header never does)"
417 ));
418 }
419 }
420 }
421 }
422 if !seen_origins.insert(origin.as_str()) {
423 errors.push(format!(
424 "websocket.allowed_origins: duplicate origin '{origin}'"
425 ));
426 }
427 }
428}
429
430fn validate_websocket_keepalive(
433 websocket: &super::types::WebSocketConfig,
434 errors: &mut Vec<String>,
435) {
436 match websocket.ping_interval_ms {
437 Some(0) => errors.push(
438 "websocket.ping_interval_ms: must be greater than zero when configured (omit the \
439 key to disable keepalive pings)"
440 .to_owned(),
441 ),
442 Some(interval_ms) => {
443 let interval = std::time::Duration::from_millis(interval_ms);
446 if std::time::Instant::now().checked_add(interval).is_none() {
447 errors.push(format!(
448 "websocket.ping_interval_ms: {interval_ms} overflows the monotonic clock"
449 ));
450 }
451 }
452 None => {}
453 }
454}
455
456fn validate_participant(config: &ServerConfig, errors: &mut Vec<String>) {
461 let Some(participant) = config.participant.as_ref() else {
462 return;
463 };
464 participant.collect_errors(errors);
465 if participant.wire_frame_limit != 0
466 && let Err(error) = crate::server::participant::normalize_configured_frame_limit(
467 participant.wire_frame_limit,
468 )
469 {
470 errors.push(format!(
471 "participant.wire_frame_limit: {} is below the protocol's minimum complete \
472 participant frame ({error:?})",
473 participant.wire_frame_limit
474 ));
475 }
476}
477
478pub(crate) fn worker_front_door_field_errors(config: &ServerConfig) -> Vec<String> {
491 let mut errors = Vec::new();
492 if !config.channels.is_empty() {
493 errors.push(
494 "services.profile: \"worker-front-door\" builds no channels; remove the \
495 [[channels]] entries or use profile = \"full\""
496 .to_owned(),
497 );
498 }
499 if !config.routing_rules.is_empty() {
500 errors.push(
501 "services.profile: \"worker-front-door\" builds no channels to route between; \
502 remove the [[routing_rules]] entries or use profile = \"full\""
503 .to_owned(),
504 );
505 }
506 if config.persistence_path.is_some() {
507 errors.push(
508 "services.profile: \"worker-front-door\" builds no durable store; remove \
509 persistence_path or use profile = \"full\""
510 .to_owned(),
511 );
512 }
513 if config.cluster.is_some() {
514 errors.push(
515 "services.profile: \"worker-front-door\" builds no channel cluster; remove the \
516 [cluster] section or use profile = \"full\""
517 .to_owned(),
518 );
519 }
520 if config.participant.is_some() {
521 errors.push(
522 "services.profile: \"worker-front-door\" installs no participant service; remove \
523 the [participant] section or use profile = \"full\""
524 .to_owned(),
525 );
526 }
527 errors
528}
529
530#[cfg(test)]
531mod tests {
532 use std::fs;
533 use std::net::SocketAddr;
534 use std::path::PathBuf;
535 use std::sync::atomic::{AtomicU64, Ordering};
536
537 use crate::ServerError;
538
539 use super::validate;
540 use crate::config::types::{
541 AuthConfig, ChannelDef, ClusterConfig, LimitsConfig, ParticipantConfig, RoutingRuleDef,
542 ServerConfig, ServicesConfig,
543 };
544
545 static NEXT_TEMP_DIR_ID: AtomicU64 = AtomicU64::new(0);
546
547 fn socket(address: &str) -> Result<SocketAddr, Box<dyn std::error::Error>> {
548 Ok(address.parse()?)
549 }
550
551 fn sample_config() -> Result<ServerConfig, Box<dyn std::error::Error>> {
552 Ok(ServerConfig {
553 listen_address: socket("127.0.0.1:8080")?,
554 health_listen_address: socket("127.0.0.1:8081")?,
555 drain_timeout_ms: 30_000,
556 channels: vec![ChannelDef {
557 name: "orders".to_owned(),
558 schema_ref: None,
559 durable: true,
560 loaded_schema: None,
561 }],
562 routing_rules: vec![RoutingRuleDef {
563 source_channel: "orders".to_owned(),
564 target_channel: "orders".to_owned(),
565 predicate: None,
566 }],
567 persistence_path: None,
568 cluster: Some(ClusterConfig {
569 node_name: "node-a".to_owned(),
570 listen_address: socket("127.0.0.1:9000")?,
571 seed_nodes: vec![socket("127.0.0.1:9001")?],
572 cookie: "test-cookie".to_owned(),
573 }),
574 auth: None,
575 services: ServicesConfig::default(),
576 limits: LimitsConfig::default(),
577 participant: None,
578 websocket: None,
579 })
580 }
581
582 fn worker_front_door_config() -> Result<ServerConfig, Box<dyn std::error::Error>> {
585 Ok(ServerConfig {
586 channels: Vec::new(),
587 routing_rules: Vec::new(),
588 persistence_path: None,
589 cluster: None,
590 services: ServicesConfig {
591 profile: "worker-front-door".to_owned(),
592 },
593 ..sample_config()?
594 })
595 }
596
597 fn unique_temp_dir(label: &str) -> PathBuf {
598 let id = NEXT_TEMP_DIR_ID.fetch_add(1, Ordering::Relaxed);
599 std::env::temp_dir().join(format!(
600 "liminal-server-validation-{label}-{}-{id}",
601 std::process::id()
602 ))
603 }
604
605 fn config_validation_message(result: Result<(), ServerError>) -> String {
606 let Err(ServerError::ConfigValidation { message }) = result else {
607 return String::new();
608 };
609 message
610 }
611
612 #[test]
613 fn valid_config_passes_validation() -> Result<(), Box<dyn std::error::Error>> {
614 let mut config = sample_config()?;
615
616 validate(&mut config, None)?;
617
618 Ok(())
619 }
620
621 #[test]
622 fn invalid_listen_address_reports_field_name() -> Result<(), Box<dyn std::error::Error>> {
623 let mut config = sample_config()?;
624 config.listen_address = socket("127.0.0.1:0")?;
625
626 let message = config_validation_message(validate(&mut config, None));
627
628 assert!(message.contains("listen_address"));
629 assert!(message.contains("port"));
630
631 Ok(())
632 }
633
634 #[test]
635 fn invalid_health_listen_address_reports_field_name() -> Result<(), Box<dyn std::error::Error>>
636 {
637 let mut config = sample_config()?;
638 config.health_listen_address = socket("127.0.0.1:0")?;
639
640 let message = config_validation_message(validate(&mut config, None));
641
642 assert!(message.contains("health_listen_address"));
643 assert!(message.contains("port"));
644
645 Ok(())
646 }
647
648 #[test]
649 fn matching_health_and_main_listen_addresses_are_rejected()
650 -> Result<(), Box<dyn std::error::Error>> {
651 let mut config = sample_config()?;
652 config.health_listen_address = config.listen_address;
653
654 let message = config_validation_message(validate(&mut config, None));
655
656 assert!(message.contains("health_listen_address"));
657 assert!(message.contains("listen_address"));
658
659 Ok(())
660 }
661
662 #[test]
663 fn matching_health_and_main_listen_ports_are_rejected() -> Result<(), Box<dyn std::error::Error>>
664 {
665 let mut config = sample_config()?;
666 config.health_listen_address = socket("0.0.0.0:8080")?;
667
668 let message = config_validation_message(validate(&mut config, None));
669
670 assert!(message.contains("health_listen_address"));
671 assert!(message.contains("port"));
672
673 Ok(())
674 }
675
676 #[test]
680 fn zero_drain_timeout_is_accepted_because_the_field_is_ignored()
681 -> Result<(), Box<dyn std::error::Error>> {
682 let mut config = sample_config()?;
683 config.drain_timeout_ms = 0;
684
685 validate(&mut config, None)?;
686
687 assert_eq!(config.drain_timeout_ms, 0);
688
689 Ok(())
690 }
691
692 #[test]
693 fn duplicate_channel_names_are_listed() -> Result<(), Box<dyn std::error::Error>> {
694 let mut config = sample_config()?;
695 config.channels.push(ChannelDef {
696 name: "orders".to_owned(),
697 schema_ref: None,
698 durable: false,
699 loaded_schema: None,
700 });
701
702 let message = config_validation_message(validate(&mut config, None));
703
704 assert!(message.contains("duplicate"));
705 assert!(message.contains("orders"));
706
707 Ok(())
708 }
709
710 #[test]
711 fn unreachable_persistence_path_reports_path() -> Result<(), Box<dyn std::error::Error>> {
712 let mut config = sample_config()?;
713 let path = unique_temp_dir("missing");
714 config.persistence_path = Some(path.clone());
715
716 let message = config_validation_message(validate(&mut config, None));
717
718 assert!(message.contains("persistence_path"));
719 assert!(message.contains(&path.display().to_string()));
720
721 Ok(())
722 }
723
724 #[test]
725 fn file_persistence_path_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
726 let mut config = sample_config()?;
727 let path = unique_temp_dir("file");
728 fs::write(&path, "not a directory")?;
729 config.persistence_path = Some(path.clone());
730
731 let message = config_validation_message(validate(&mut config, None));
732 fs::remove_file(&path)?;
733
734 assert!(message.contains("persistence_path"));
735 assert!(message.contains("directory"));
736
737 Ok(())
738 }
739
740 #[test]
741 fn multiple_validation_errors_are_reported_together() -> Result<(), Box<dyn std::error::Error>>
742 {
743 let mut config = sample_config()?;
744 let missing_path = unique_temp_dir("multi-missing");
745 config.listen_address = socket("127.0.0.1:0")?;
746 config.channels.push(ChannelDef {
747 name: "orders".to_owned(),
748 schema_ref: None,
749 durable: false,
750 loaded_schema: None,
751 });
752 config.persistence_path = Some(missing_path.clone());
753
754 let message = config_validation_message(validate(&mut config, None));
755
756 assert!(message.contains("listen_address"));
757 assert!(message.contains("duplicate channel names: orders"));
758 assert!(message.contains(&missing_path.display().to_string()));
759
760 Ok(())
761 }
762
763 #[test]
764 fn routing_rules_reference_configured_channels() -> Result<(), Box<dyn std::error::Error>> {
765 let mut config = sample_config()?;
766 config.routing_rules[0].target_channel = "unknown".to_owned();
767
768 let message = config_validation_message(validate(&mut config, None));
769
770 assert!(message.contains("routing_rules[0].target_channel"));
771 assert!(message.contains("unknown"));
772
773 Ok(())
774 }
775
776 fn write_temp_schema(
778 label: &str,
779 contents: &str,
780 ) -> Result<PathBuf, Box<dyn std::error::Error>> {
781 let path = unique_temp_dir(label).with_extension("json");
782 fs::write(&path, contents)?;
783 Ok(path)
784 }
785
786 #[test]
787 fn absolute_schema_ref_is_loaded_and_parsed() -> Result<(), Box<dyn std::error::Error>> {
788 let schema = r#"{"type":"object","properties":{"id":{"type":"integer"}}}"#;
789 let schema_path = write_temp_schema("load-ok", schema)?;
790 let mut config = sample_config()?;
791 config.channels[0].schema_ref = Some(schema_path.clone());
792
793 let result = validate(&mut config, None);
794 fs::remove_file(&schema_path)?;
795 result?;
796
797 let loaded = config.channels[0]
798 .loaded_schema
799 .as_ref()
800 .ok_or("schema should have been loaded onto the channel")?;
801 assert_eq!(loaded.bytes, schema.as_bytes());
802 assert_eq!(
803 loaded.document.get("type").and_then(|t| t.as_str()),
804 Some("object")
805 );
806
807 Ok(())
808 }
809
810 #[test]
811 fn relative_schema_ref_resolves_against_base_dir() -> Result<(), Box<dyn std::error::Error>> {
812 let dir = unique_temp_dir("relative-base");
813 fs::create_dir_all(&dir)?;
814 let schema = r#"{"type":"object"}"#;
815 fs::write(dir.join("orders.json"), schema)?;
816
817 let mut config = sample_config()?;
818 config.channels[0].schema_ref = Some(PathBuf::from("orders.json"));
819
820 let result = validate(&mut config, Some(&dir));
821 fs::remove_dir_all(&dir)?;
822 result?;
823
824 assert!(config.channels[0].loaded_schema.is_some());
825
826 Ok(())
827 }
828
829 #[test]
830 fn missing_schema_ref_file_reports_validation_error() -> Result<(), Box<dyn std::error::Error>>
831 {
832 let missing = unique_temp_dir("missing-schema").with_extension("json");
833 let mut config = sample_config()?;
834 config.channels[0].schema_ref = Some(missing.clone());
835
836 let message = config_validation_message(validate(&mut config, None));
837
838 assert!(message.contains("schema_ref"));
839 assert!(message.contains(&missing.display().to_string()));
840 assert!(message.contains("unreadable"));
841
842 Ok(())
843 }
844
845 #[test]
846 fn invalid_json_schema_ref_reports_validation_error() -> Result<(), Box<dyn std::error::Error>>
847 {
848 let schema_path = write_temp_schema("bad-json", "{ this is not json")?;
849 let mut config = sample_config()?;
850 config.channels[0].schema_ref = Some(schema_path.clone());
851
852 let message = config_validation_message(validate(&mut config, None));
853 fs::remove_file(&schema_path)?;
854
855 assert!(message.contains("schema_ref"));
856 assert!(message.contains("not valid JSON"));
857
858 Ok(())
859 }
860
861 #[test]
862 fn valid_json_invalid_schema_ref_reports_validation_error()
863 -> Result<(), Box<dyn std::error::Error>> {
864 let schema_path = write_temp_schema("bad-schema", "[]")?;
867 let mut config = sample_config()?;
868 config.channels[0].schema_ref = Some(schema_path.clone());
869
870 let message = config_validation_message(validate(&mut config, None));
871 fs::remove_file(&schema_path)?;
872
873 assert!(message.contains("schema_ref"));
874 assert!(message.contains("not a valid JSON Schema"));
875
876 Ok(())
877 }
878
879 #[test]
880 fn present_non_empty_auth_token_passes_validation() -> Result<(), Box<dyn std::error::Error>> {
881 let mut config = sample_config()?;
882 config.auth = Some(AuthConfig {
883 pass: None,
884 token: "s3cr3t".to_owned(),
885 });
886
887 validate(&mut config, None)?;
888
889 Ok(())
890 }
891
892 #[test]
893 fn empty_auth_token_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
894 let mut config = sample_config()?;
895 config.auth = Some(AuthConfig {
896 pass: None,
897 token: String::new(),
898 });
899
900 let message = config_validation_message(validate(&mut config, None));
901
902 assert!(message.contains("auth.token"));
903 assert!(message.contains("must not be empty"));
904
905 Ok(())
906 }
907
908 #[test]
909 fn absent_auth_section_passes_validation() -> Result<(), Box<dyn std::error::Error>> {
910 let mut config = sample_config()?;
911 config.auth = None;
912
913 validate(&mut config, None)?;
914
915 Ok(())
916 }
917
918 #[test]
919 fn default_profile_is_full_and_passes_validation() -> Result<(), Box<dyn std::error::Error>> {
920 let mut config = sample_config()?;
921
922 assert_eq!(
924 config.services.profile()?,
925 crate::config::types::ServiceProfile::Full
926 );
927 validate(&mut config, None)?;
928
929 Ok(())
930 }
931
932 #[test]
933 fn unknown_profile_is_a_validation_error() -> Result<(), Box<dyn std::error::Error>> {
934 let mut config = sample_config()?;
935 config.services = ServicesConfig {
936 profile: "banana".to_owned(),
937 };
938
939 let message = config_validation_message(validate(&mut config, None));
940
941 assert!(message.contains("services.profile"));
942 assert!(message.contains("banana"));
943 assert!(message.contains("worker-front-door"));
944
945 Ok(())
946 }
947
948 #[test]
949 fn worker_front_door_profile_with_empty_topology_passes()
950 -> Result<(), Box<dyn std::error::Error>> {
951 let mut config = worker_front_door_config()?;
952
953 assert_eq!(
954 config.services.profile()?,
955 crate::config::types::ServiceProfile::WorkerFrontDoor
956 );
957 validate(&mut config, None)?;
958
959 Ok(())
960 }
961
962 #[test]
963 fn worker_front_door_profile_rejects_channels_persistence_and_cluster()
964 -> Result<(), Box<dyn std::error::Error>> {
965 let mut config = sample_config()?;
968 config.services = ServicesConfig {
969 profile: "worker-front-door".to_owned(),
970 };
971 config.persistence_path = Some(PathBuf::from("/tmp"));
972
973 let message = config_validation_message(validate(&mut config, None));
974
975 assert!(message.contains("builds no channels"));
976 assert!(message.contains("builds no durable store"));
977 assert!(message.contains("builds no channel cluster"));
978
979 Ok(())
980 }
981
982 #[test]
983 fn default_limits_pass_validation_and_carry_signed_numbers()
984 -> Result<(), Box<dyn std::error::Error>> {
985 let mut config = sample_config()?;
986 assert_eq!(config.limits.max_connections, 256);
988 assert_eq!(config.limits.max_subscriptions_per_connection, 32);
989 assert_eq!(config.limits.max_conversations_per_connection, 32);
990 assert_eq!(config.limits.max_pending_pushes_per_connection, 32);
991 assert_eq!(
992 config
993 .limits
994 .max_pending_conversation_replies_per_connection,
995 32
996 );
997 assert_eq!(config.limits.max_pending_replies_per_conversation, 8);
998 assert_eq!(config.limits.max_connection_inbox_bytes, 4 * 1024 * 1024);
999 assert_eq!(config.limits.max_connection_outbound_bytes, 4 * 1024 * 1024);
1004 assert_eq!(
1005 config.limits.max_connection_outbound_bytes, config.limits.max_connection_inbox_bytes,
1006 "the two per-connection byte budgets are deliberately equal"
1007 );
1008 assert_eq!(config.limits.max_subscription_inbox_depth, 4096);
1016 assert_eq!(
1017 config.limits.max_connection_inbox_bytes / config.limits.max_subscription_inbox_depth,
1018 1024,
1019 "the byte-vs-count crossover must stay at 1 KiB per record"
1020 );
1021 assert_eq!(config.limits.delivery_slice_budget, 32);
1022 validate(&mut config, None)?;
1023 Ok(())
1024 }
1025
1026 #[test]
1029 fn zero_limits_are_typed_config_errors() -> Result<(), Box<dyn std::error::Error>> {
1030 type LimitMutator = (&'static str, fn(&mut ServerConfig));
1031 let mutators: [LimitMutator; 10] = [
1032 ("max_connections", |c| c.limits.max_connections = 0),
1033 ("max_subscriptions_per_connection", |c| {
1034 c.limits.max_subscriptions_per_connection = 0;
1035 }),
1036 ("max_conversations_per_connection", |c| {
1037 c.limits.max_conversations_per_connection = 0;
1038 }),
1039 ("max_pending_pushes_per_connection", |c| {
1040 c.limits.max_pending_pushes_per_connection = 0;
1041 }),
1042 ("max_pending_conversation_replies_per_connection", |c| {
1043 c.limits.max_pending_conversation_replies_per_connection = 0;
1044 }),
1045 ("max_pending_replies_per_conversation", |c| {
1046 c.limits.max_pending_replies_per_conversation = 0;
1047 }),
1048 ("max_connection_inbox_bytes", |c| {
1049 c.limits.max_connection_inbox_bytes = 0;
1050 }),
1051 ("max_connection_outbound_bytes", |c| {
1052 c.limits.max_connection_outbound_bytes = 0;
1053 }),
1054 ("max_subscription_inbox_depth", |c| {
1055 c.limits.max_subscription_inbox_depth = 0;
1056 }),
1057 ("delivery_slice_budget", |c| {
1058 c.limits.delivery_slice_budget = 0;
1059 }),
1060 ];
1061 for (field, mutate) in mutators {
1062 let mut config = sample_config()?;
1063 mutate(&mut config);
1064 let message = config_validation_message(validate(&mut config, None));
1065 assert!(
1066 message.contains(&format!("limits.{field}")),
1067 "zero {field} must report a typed limits.{field} error, got: {message}"
1068 );
1069 assert!(
1070 message.contains("greater than zero"),
1071 "the {field} refusal must say why: {message}"
1072 );
1073 }
1074 Ok(())
1075 }
1076
1077 const fn sample_participant() -> ParticipantConfig {
1079 ParticipantConfig {
1080 wire_frame_limit: 65_536,
1081 attach_receipt_ttl_ms: 60_000,
1082 receipt_provenance_ttl_ms: 600_000,
1083 live_receipt_server_report_threshold: 1_024,
1084 max_live_attach_receipts_per_participant: 8,
1085 receipt_provenance_server_report_threshold: 4_096,
1086 receipt_provenance_per_conversation_report_threshold: 256,
1087 max_receipt_provenance_per_participant: 64,
1088 max_retired_identity_slots_server: 1_024,
1089 identity_slots: 4,
1090 observer_recovery_max_entries: 64,
1091 max_semantic_conversations_per_connection: 32,
1092 max_ordinary_record_entries: 1,
1093 max_ordinary_record_bytes: 131_072,
1094 max_generated_marker_entries: 1,
1095 max_generated_marker_bytes: 4_096,
1096 mandatory_transaction_bound_entries: 4,
1097 mandatory_transaction_bound_bytes: 16_384,
1098 full_recovery_claim_entries: 4,
1099 full_recovery_claim_bytes: 16_384,
1100 retained_capacity_entries: 2_048,
1101 retained_capacity_bytes: 16_777_216,
1102 max_retained_record_rows: 1_024,
1103 closure_episode_churn_limit: 1_024,
1104 }
1105 }
1106
1107 #[test]
1108 fn valid_participant_section_passes_validation() -> Result<(), Box<dyn std::error::Error>> {
1109 let mut config = sample_config()?;
1110 config.participant = Some(sample_participant());
1111 let temp_dir = std::env::temp_dir().join(format!(
1112 "liminal-server-participant-validation-{}-{}",
1113 std::process::id(),
1114 NEXT_TEMP_DIR_ID.fetch_add(1, Ordering::Relaxed)
1115 ));
1116 fs::create_dir_all(&temp_dir)?;
1117 config.persistence_path = Some(temp_dir.clone());
1118 let result = validate(&mut config, None);
1119 fs::remove_dir_all(&temp_dir)?;
1120 assert!(result.is_ok(), "expected valid config, got {result:?}");
1121 Ok(())
1122 }
1123
1124 #[test]
1125 fn zero_participant_values_are_typed_config_errors() -> Result<(), Box<dyn std::error::Error>> {
1126 type ParticipantMutator = (&'static str, fn(&mut ParticipantConfig));
1127 let mutators: [ParticipantMutator; 12] = [
1128 ("wire_frame_limit", |p| p.wire_frame_limit = 0),
1129 ("attach_receipt_ttl_ms", |p| p.attach_receipt_ttl_ms = 0),
1130 ("receipt_provenance_ttl_ms", |p| {
1131 p.receipt_provenance_ttl_ms = 0;
1132 }),
1133 ("live_receipt_server_report_threshold", |p| {
1134 p.live_receipt_server_report_threshold = 0;
1135 }),
1136 ("max_live_attach_receipts_per_participant", |p| {
1137 p.max_live_attach_receipts_per_participant = 0;
1138 }),
1139 ("receipt_provenance_server_report_threshold", |p| {
1140 p.receipt_provenance_server_report_threshold = 0;
1141 }),
1142 (
1143 "receipt_provenance_per_conversation_report_threshold",
1144 |p| {
1145 p.receipt_provenance_per_conversation_report_threshold = 0;
1146 },
1147 ),
1148 ("max_receipt_provenance_per_participant", |p| {
1149 p.max_receipt_provenance_per_participant = 0;
1150 }),
1151 ("max_retired_identity_slots_server", |p| {
1152 p.max_retired_identity_slots_server = 0;
1153 }),
1154 ("identity_slots", |p| p.identity_slots = 0),
1155 ("observer_recovery_max_entries", |p| {
1156 p.observer_recovery_max_entries = 0;
1157 }),
1158 ("max_semantic_conversations_per_connection", |p| {
1159 p.max_semantic_conversations_per_connection = 0;
1160 }),
1161 ];
1162 for (field, mutate) in mutators {
1163 let mut config = sample_config()?;
1164 let mut participant = sample_participant();
1165 mutate(&mut participant);
1166 config.participant = Some(participant);
1167 let message = config_validation_message(validate(&mut config, None));
1168 assert!(
1169 message.contains(&format!("participant.{field}")),
1170 "expected typed error for participant.{field}, got: {message}"
1171 );
1172 }
1173 Ok(())
1174 }
1175
1176 #[test]
1177 fn provenance_ttl_shorter_than_receipt_is_a_typed_config_error()
1178 -> Result<(), Box<dyn std::error::Error>> {
1179 let mut config = sample_config()?;
1180 let mut participant = sample_participant();
1181 participant.receipt_provenance_ttl_ms = participant.attach_receipt_ttl_ms - 1;
1182 config.participant = Some(participant);
1183 let message = config_validation_message(validate(&mut config, None));
1184 assert!(
1185 message.contains("participant.receipt_provenance_ttl_ms"),
1186 "expected TTL-ordering error, got: {message}"
1187 );
1188 Ok(())
1189 }
1190
1191 #[test]
1192 fn undersized_wire_frame_limit_is_a_typed_config_error()
1193 -> Result<(), Box<dyn std::error::Error>> {
1194 let mut config = sample_config()?;
1195 let mut participant = sample_participant();
1196 participant.wire_frame_limit = 1;
1197 config.participant = Some(participant);
1198 let message = config_validation_message(validate(&mut config, None));
1199 assert!(
1200 message.contains("participant.wire_frame_limit"),
1201 "expected minimum-frame error, got: {message}"
1202 );
1203 Ok(())
1204 }
1205
1206 #[test]
1207 fn worker_front_door_rejects_participant_section() -> Result<(), Box<dyn std::error::Error>> {
1208 let mut config = worker_front_door_config()?;
1209 config.participant = Some(sample_participant());
1210 let message = config_validation_message(validate(&mut config, None));
1211 assert!(
1212 message.contains("installs no participant service"),
1213 "expected worker-front-door participant rejection, got: {message}"
1214 );
1215 Ok(())
1216 }
1217
1218 fn sample_websocket()
1221 -> Result<crate::config::types::WebSocketConfig, Box<dyn std::error::Error>> {
1222 Ok(crate::config::types::WebSocketConfig {
1223 listen_address: socket("127.0.0.1:8082")?,
1224 path: "/liminal".to_owned(),
1225 allowed_origins: vec!["https://app.example.com".to_owned()],
1226 ping_interval_ms: Some(30_000),
1227 })
1228 }
1229
1230 #[test]
1231 fn valid_websocket_section_passes_validation() -> Result<(), Box<dyn std::error::Error>> {
1232 let mut config = sample_config()?;
1233 config.websocket = Some(sample_websocket()?);
1234 validate(&mut config, None)?;
1235 Ok(())
1236 }
1237
1238 #[test]
1239 fn websocket_empty_origin_list_is_valid_fail_closed_configuration()
1240 -> Result<(), Box<dyn std::error::Error>> {
1241 let mut config = sample_config()?;
1242 let mut websocket = sample_websocket()?;
1243 websocket.allowed_origins = Vec::new();
1244 config.websocket = Some(websocket);
1245 validate(&mut config, None)?;
1246 Ok(())
1247 }
1248
1249 #[test]
1250 fn websocket_zero_port_is_a_typed_config_error() -> Result<(), Box<dyn std::error::Error>> {
1251 let mut config = sample_config()?;
1252 let mut websocket = sample_websocket()?;
1253 websocket.listen_address = socket("127.0.0.1:0")?;
1254 config.websocket = Some(websocket);
1255 let message = config_validation_message(validate(&mut config, None));
1256 assert!(
1257 message.contains("websocket.listen_address"),
1258 "expected websocket.listen_address error, got: {message}"
1259 );
1260 Ok(())
1261 }
1262
1263 #[test]
1264 fn websocket_address_conflicts_are_typed_config_errors()
1265 -> Result<(), Box<dyn std::error::Error>> {
1266 for conflict in ["127.0.0.1:8080", "127.0.0.1:8081", "127.0.0.1:9000"] {
1267 let mut config = sample_config()?;
1268 let mut websocket = sample_websocket()?;
1269 websocket.listen_address = socket(conflict)?;
1270 config.websocket = Some(websocket);
1271 let message = config_validation_message(validate(&mut config, None));
1272 assert!(
1273 message.contains("websocket.listen_address: must differ from"),
1274 "expected an address-conflict error for {conflict}, got: {message}"
1275 );
1276 }
1277 Ok(())
1278 }
1279
1280 #[test]
1281 fn websocket_path_rules_are_typed_config_errors() -> Result<(), Box<dyn std::error::Error>> {
1282 for (bad_path, expectation) in [
1283 ("", "must not be empty"),
1284 ("liminal", "must start with '/'"),
1285 ("/limi nal", "whitespace"),
1286 ("/liminal?x=1", "query"),
1287 ("/liminal#frag", "query"),
1288 ] {
1289 let mut config = sample_config()?;
1290 let mut websocket = sample_websocket()?;
1291 websocket.path = bad_path.to_owned();
1292 config.websocket = Some(websocket);
1293 let message = config_validation_message(validate(&mut config, None));
1294 assert!(
1295 message.contains("websocket.path") && message.contains(expectation),
1296 "expected websocket.path '{bad_path}' to report '{expectation}', got: {message}"
1297 );
1298 }
1299 Ok(())
1300 }
1301
1302 #[test]
1303 fn websocket_malformed_origin_entries_are_typed_config_errors()
1304 -> Result<(), Box<dyn std::error::Error>> {
1305 for (bad_origin, expectation) in [
1306 ("", "must not be empty"),
1307 ("app.example.com", "serialized origin"),
1308 ("https://app.example.com/", "path or trailing slash"),
1309 ("https://app.example.com/app", "path or trailing slash"),
1310 ("https:// app.example.com", "whitespace"),
1311 ] {
1312 let mut config = sample_config()?;
1313 let mut websocket = sample_websocket()?;
1314 websocket.allowed_origins = vec![bad_origin.to_owned()];
1315 config.websocket = Some(websocket);
1316 let message = config_validation_message(validate(&mut config, None));
1317 assert!(
1318 message.contains("websocket.allowed_origins") && message.contains(expectation),
1319 "expected origin '{bad_origin}' to report '{expectation}', got: {message}"
1320 );
1321 }
1322 Ok(())
1323 }
1324
1325 #[test]
1326 fn websocket_duplicate_origin_is_a_typed_config_error() -> Result<(), Box<dyn std::error::Error>>
1327 {
1328 let mut config = sample_config()?;
1329 let mut websocket = sample_websocket()?;
1330 websocket.allowed_origins = vec![
1331 "https://app.example.com".to_owned(),
1332 "https://app.example.com".to_owned(),
1333 ];
1334 config.websocket = Some(websocket);
1335 let message = config_validation_message(validate(&mut config, None));
1336 assert!(
1337 message.contains("duplicate origin"),
1338 "expected a duplicate-origin error, got: {message}"
1339 );
1340 Ok(())
1341 }
1342
1343 #[test]
1344 fn websocket_zero_ping_interval_is_a_typed_config_error()
1345 -> Result<(), Box<dyn std::error::Error>> {
1346 let mut config = sample_config()?;
1347 let mut websocket = sample_websocket()?;
1348 websocket.ping_interval_ms = Some(0);
1349 config.websocket = Some(websocket);
1350 let message = config_validation_message(validate(&mut config, None));
1351 assert!(
1352 message.contains("websocket.ping_interval_ms"),
1353 "expected a zero-interval error, got: {message}"
1354 );
1355 Ok(())
1356 }
1357
1358 #[test]
1359 fn websocket_absent_ping_interval_means_disabled_and_valid()
1360 -> Result<(), Box<dyn std::error::Error>> {
1361 let mut config = sample_config()?;
1362 let mut websocket = sample_websocket()?;
1363 websocket.ping_interval_ms = None;
1364 config.websocket = Some(websocket);
1365 validate(&mut config, None)?;
1366 Ok(())
1367 }
1368}