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