1use std::collections::{BTreeMap, BTreeSet};
2use std::path::{Path, PathBuf};
3
4use crate::ServerError;
5
6use super::types::{LoadedSchema, SelfStreamsConfig, 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 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
198pub(crate) const MISSING_PERSISTENCE_PATH: &str = "persistence_path: full services require an existing directory; set persistence_path or LIMINAL_PERSISTENCE_PATH; no store is opened";
199
200pub(crate) fn required_persistence_path(config: &ServerConfig) -> Result<&Path, ServerError> {
201 config
202 .persistence_path
203 .as_deref()
204 .ok_or_else(|| ServerError::ConfigValidation {
205 message: MISSING_PERSISTENCE_PATH.to_owned(),
206 })
207}
208
209fn validate_persistence_path(config: &ServerConfig, errors: &mut Vec<String>) {
210 let Some(path) = config.persistence_path.as_deref() else {
211 if matches!(config.services.profile(), Ok(ServiceProfile::Full)) {
212 errors.push(MISSING_PERSISTENCE_PATH.to_owned());
213 }
214 return;
215 };
216
217 match std::fs::metadata(path) {
218 Ok(metadata) => {
219 if !metadata.is_dir() {
220 errors.push(format!(
221 "persistence_path '{}': path must be an existing directory",
222 path.display()
223 ));
224 } else if metadata.permissions().readonly() {
225 errors.push(format!(
226 "persistence_path '{}': path is not writable",
227 path.display()
228 ));
229 }
230 }
231 Err(error) => {
232 errors.push(format!(
233 "persistence_path '{}': path is unreachable: {error}",
234 path.display()
235 ));
236 }
237 }
238}
239
240fn validate_cluster(config: &ServerConfig, errors: &mut Vec<String>) {
241 let Some(cluster) = config.cluster.as_ref() else {
242 return;
243 };
244
245 if cluster.node_name.trim().is_empty() {
246 errors.push("cluster.node_name: node name must not be empty".to_owned());
247 }
248
249 if cluster.cookie.is_empty() {
250 errors.push("cluster.cookie: distribution cookie must not be empty".to_owned());
251 }
252
253 if cluster.listen_address.port() == 0 {
254 errors.push("cluster.listen_address: distribution port must be non-zero".to_owned());
255 }
256
257 if cluster.listen_address == config.listen_address {
258 errors.push(
259 "cluster.listen_address: distribution port must differ from the client listen_address"
260 .to_owned(),
261 );
262 }
263
264 let mut seed_node_counts = BTreeMap::new();
265 for (index, seed_node) in cluster.seed_nodes.iter().enumerate() {
266 if seed_node.port() == 0 {
267 errors.push(format!(
268 "cluster.seed_nodes[{index}]: seed node port must be non-zero"
269 ));
270 }
271 seed_node_counts
272 .entry(seed_node.to_string())
273 .and_modify(|count| *count += 1)
274 .or_insert(1_usize);
275 }
276
277 let duplicates = seed_node_counts
278 .into_iter()
279 .filter_map(|(seed_node, count)| (count > 1).then_some(seed_node))
280 .collect::<Vec<_>>();
281
282 if !duplicates.is_empty() {
283 errors.push(format!(
284 "cluster.seed_nodes: duplicate seed nodes: {}",
285 duplicates.join(", ")
286 ));
287 }
288}
289
290fn validate_auth(config: &ServerConfig, errors: &mut Vec<String>) {
296 let Some(auth) = config.auth.as_ref() else {
297 return;
298 };
299
300 if auth.token.is_empty() {
301 errors.push("auth.token: authentication token must not be empty".to_owned());
302 }
303 let max = crate::server::connection::pre_admission::MAX_CONNECT_AUTH_TOKEN_BYTES;
306 if auth.token.len() > max {
307 errors.push(format!(
308 "auth.token: authentication token is {} bytes; a Connect can carry at most {max}",
309 auth.token.len()
310 ));
311 }
312 if auth.require_verified_worker && auth.pass.is_none() {
315 errors.push(
316 "auth.require_verified_worker: requires [auth.pass], or no worker could register"
317 .to_owned(),
318 );
319 }
320}
321
322impl SelfStreamsConfig {
328 pub fn validate(&self) -> Result<(), ServerError> {
333 let mut errors = Vec::new();
334 collect_self_stream_bound_errors(self, &mut errors);
335 if errors.is_empty() {
336 Ok(())
337 } else {
338 Err(ServerError::ConfigValidation {
339 message: errors.join("; "),
340 })
341 }
342 }
343}
344
345fn collect_self_stream_bound_errors(config: &SelfStreamsConfig, errors: &mut Vec<String>) {
346 errors.extend(
347 [
348 ("cost_rows", config.cost_rows),
349 ("events_rows", config.events_rows),
350 ]
351 .into_iter()
352 .filter(|(_, rows)| *rows == 0)
353 .map(|(name, _)| format!("services.self_streams.{name}: must be non-zero")),
354 );
355}
356
357fn validate_services(config: &ServerConfig, errors: &mut Vec<String>) {
358 collect_self_stream_bound_errors(&config.services.self_streams, errors);
359 let profile = match config.services.profile() {
360 Ok(profile) => profile,
361 Err(error) => {
362 match error {
366 crate::ServerError::ConfigValidation { message } => errors.push(message),
367 other => errors.push(other.to_string()),
368 }
369 return;
370 }
371 };
372
373 if profile == ServiceProfile::WorkerFrontDoor {
374 errors.extend(worker_front_door_field_errors(config));
375 }
376}
377
378fn validate_websocket(config: &ServerConfig, errors: &mut Vec<String>) {
389 let Some(websocket) = config.websocket.as_ref() else {
390 return;
391 };
392 validate_websocket_endpoint(config, websocket, errors);
393 validate_websocket_origins(websocket, errors);
394 validate_websocket_keepalive(websocket, errors);
395}
396
397fn validate_websocket_endpoint(
399 config: &ServerConfig,
400 websocket: &super::types::WebSocketConfig,
401 errors: &mut Vec<String>,
402) {
403 if websocket.listen_address.port() == 0 {
404 errors.push("websocket.listen_address: port must be non-zero".to_owned());
405 }
406 if websocket.listen_address == config.listen_address {
407 errors.push(
408 "websocket.listen_address: must differ from listen_address (the WebSocket \
409 acceptor is a sibling listener, not the main wire port)"
410 .to_owned(),
411 );
412 }
413 if websocket.listen_address == config.health_listen_address {
414 errors.push("websocket.listen_address: must differ from health_listen_address".to_owned());
415 }
416 if let Some(cluster) = config.cluster.as_ref() {
417 if websocket.listen_address == cluster.listen_address {
418 errors.push(
419 "websocket.listen_address: must differ from cluster.listen_address".to_owned(),
420 );
421 }
422 }
423
424 if websocket.path.is_empty() {
425 errors.push("websocket.path: upgrade path must not be empty".to_owned());
426 } else if !websocket.path.starts_with('/') {
427 errors.push("websocket.path: upgrade path must start with '/'".to_owned());
428 }
429 if websocket
430 .path
431 .chars()
432 .any(|character| character.is_ascii_whitespace() || character.is_ascii_control())
433 {
434 errors.push(
435 "websocket.path: upgrade path must not contain whitespace or control characters"
436 .to_owned(),
437 );
438 }
439 if websocket.path.contains(['?', '#']) {
440 errors.push(
441 "websocket.path: upgrade path is a single exact path and must not carry a query \
442 or fragment"
443 .to_owned(),
444 );
445 }
446}
447
448fn validate_websocket_origins(websocket: &super::types::WebSocketConfig, errors: &mut Vec<String>) {
451 let mut seen_origins = BTreeSet::new();
452 for origin in &websocket.allowed_origins {
453 if origin.is_empty() {
454 errors.push("websocket.allowed_origins: origin entries must not be empty".to_owned());
455 continue;
456 }
457 if origin
458 .chars()
459 .any(|character| character.is_ascii_whitespace() || character.is_ascii_control())
460 {
461 errors.push(format!(
462 "websocket.allowed_origins: origin '{origin}' must not contain whitespace or \
463 control characters"
464 ));
465 }
466 if origin != "null" {
471 match origin.split_once("://") {
472 None => errors.push(format!(
473 "websocket.allowed_origins: origin '{origin}' must be a serialized origin \
474 (scheme://host[:port]) or the literal 'null'"
475 )),
476 Some((scheme, rest)) => {
477 if scheme.is_empty() || rest.is_empty() {
478 errors.push(format!(
479 "websocket.allowed_origins: origin '{origin}' must name both a \
480 scheme and a host"
481 ));
482 }
483 if rest.contains('/') {
484 errors.push(format!(
485 "websocket.allowed_origins: origin '{origin}' must not contain a \
486 path or trailing slash (a serialized Origin header never does)"
487 ));
488 }
489 }
490 }
491 }
492 if !seen_origins.insert(origin.as_str()) {
493 errors.push(format!(
494 "websocket.allowed_origins: duplicate origin '{origin}'"
495 ));
496 }
497 }
498}
499
500fn validate_websocket_keepalive(
503 websocket: &super::types::WebSocketConfig,
504 errors: &mut Vec<String>,
505) {
506 match websocket.ping_interval_ms {
507 Some(0) => errors.push(
508 "websocket.ping_interval_ms: must be greater than zero when configured (omit the \
509 key to disable keepalive pings)"
510 .to_owned(),
511 ),
512 Some(interval_ms) => {
513 let interval = std::time::Duration::from_millis(interval_ms);
516 if std::time::Instant::now().checked_add(interval).is_none() {
517 errors.push(format!(
518 "websocket.ping_interval_ms: {interval_ms} overflows the monotonic clock"
519 ));
520 }
521 }
522 None => {}
523 }
524}
525
526fn validate_participant(config: &ServerConfig, errors: &mut Vec<String>) {
531 let Some(participant) = config.participant.as_ref() else {
532 return;
533 };
534 participant.collect_errors(errors);
535 if participant.wire_frame_limit != 0
536 && let Err(error) = crate::server::participant::normalize_configured_frame_limit(
537 participant.wire_frame_limit,
538 )
539 {
540 errors.push(format!(
541 "participant.wire_frame_limit: {} is below the protocol's minimum complete \
542 participant frame ({error:?})",
543 participant.wire_frame_limit
544 ));
545 }
546}
547
548pub(crate) fn worker_front_door_field_errors(config: &ServerConfig) -> Vec<String> {
561 let mut errors = Vec::new();
562 if !config.channels.is_empty() {
563 errors.push(
564 "services.profile: \"worker-front-door\" builds no channels; remove the \
565 [[channels]] entries or use profile = \"full\""
566 .to_owned(),
567 );
568 }
569 if !config.routing_rules.is_empty() {
570 errors.push(
571 "services.profile: \"worker-front-door\" builds no channels to route between; \
572 remove the [[routing_rules]] entries or use profile = \"full\""
573 .to_owned(),
574 );
575 }
576 if config.persistence_path.is_some() {
577 errors.push(
578 "services.profile: \"worker-front-door\" builds no durable store; remove \
579 persistence_path or use profile = \"full\""
580 .to_owned(),
581 );
582 }
583 if config.cluster.is_some() {
584 errors.push(
585 "services.profile: \"worker-front-door\" builds no channel cluster; remove the \
586 [cluster] section or use profile = \"full\""
587 .to_owned(),
588 );
589 }
590 if config.participant.is_some() {
591 errors.push(
592 "services.profile: \"worker-front-door\" installs no participant service; remove \
593 the [participant] section or use profile = \"full\""
594 .to_owned(),
595 );
596 }
597 errors
598}
599
600#[cfg(test)]
601mod tests {
602 use std::fs;
603 use std::net::SocketAddr;
604 use std::path::PathBuf;
605 use std::sync::atomic::{AtomicU64, Ordering};
606
607 use crate::ServerError;
608
609 use super::validate;
610 use crate::config::types::{
611 AuthConfig, ChannelDef, ClusterConfig, LimitsConfig, ParticipantConfig, RoutingRuleDef,
612 ServerConfig, ServicesConfig,
613 };
614
615 static NEXT_TEMP_DIR_ID: AtomicU64 = AtomicU64::new(0);
616
617 fn socket(address: &str) -> Result<SocketAddr, Box<dyn std::error::Error>> {
618 Ok(address.parse()?)
619 }
620
621 fn sample_config() -> Result<ServerConfig, Box<dyn std::error::Error>> {
622 Ok(ServerConfig {
623 listen_address: socket("127.0.0.1:8080")?,
624 health_listen_address: socket("127.0.0.1:8081")?,
625 drain_timeout_ms: 30_000,
626 channels: vec![ChannelDef {
627 name: "orders".to_owned(),
628 schema_ref: None,
629 durable: true,
630 loaded_schema: None,
631 }],
632 routing_rules: vec![RoutingRuleDef {
633 source_channel: "orders".to_owned(),
634 target_channel: "orders".to_owned(),
635 predicate: None,
636 }],
637 persistence_path: Some(PathBuf::from(env!("CARGO_MANIFEST_DIR"))),
638 cluster: Some(ClusterConfig {
639 node_name: "node-a".to_owned(),
640 listen_address: socket("127.0.0.1:9000")?,
641 seed_nodes: vec![socket("127.0.0.1:9001")?],
642 cookie: "test-cookie".to_owned(),
643 }),
644 auth: None,
645 services: ServicesConfig::default(),
646 limits: LimitsConfig::default(),
647 participant: None,
648 websocket: None,
649 })
650 }
651
652 fn worker_front_door_config() -> Result<ServerConfig, Box<dyn std::error::Error>> {
655 Ok(ServerConfig {
656 channels: Vec::new(),
657 routing_rules: Vec::new(),
658 persistence_path: None,
659 cluster: None,
660 services: ServicesConfig {
661 profile: "worker-front-door".to_owned(),
662 ..ServicesConfig::default()
663 },
664 ..sample_config()?
665 })
666 }
667
668 fn unique_temp_dir(label: &str) -> PathBuf {
669 let id = NEXT_TEMP_DIR_ID.fetch_add(1, Ordering::Relaxed);
670 std::env::temp_dir().join(format!(
671 "liminal-server-validation-{label}-{}-{id}",
672 std::process::id()
673 ))
674 }
675
676 fn config_validation_message(result: Result<(), ServerError>) -> String {
677 let Err(ServerError::ConfigValidation { message }) = result else {
678 return String::new();
679 };
680 message
681 }
682
683 #[test]
686 fn a_durable_console_channel_is_refused_in_config() -> Result<(), Box<dyn std::error::Error>> {
687 let console = |durable| ChannelDef {
688 name: "console.topology".to_owned(),
689 schema_ref: None,
690 durable,
691 loaded_schema: None,
692 };
693 let mut config = sample_config()?;
694 config.channels.push(console(true));
695 let message = config_validation_message(validate(&mut config, None));
696 assert!(message.contains("console channel 'console.topology' must not be durable"));
697
698 let mut config = sample_config()?;
699 config.channels.push(console(false));
700 validate(&mut config, None)?;
701 Ok(())
702 }
703
704 #[test]
705 fn valid_config_passes_validation() -> Result<(), Box<dyn std::error::Error>> {
706 let mut config = sample_config()?;
707
708 validate(&mut config, None)?;
709
710 Ok(())
711 }
712
713 #[test]
714 fn invalid_listen_address_reports_field_name() -> Result<(), Box<dyn std::error::Error>> {
715 let mut config = sample_config()?;
716 config.listen_address = socket("127.0.0.1:0")?;
717
718 let message = config_validation_message(validate(&mut config, None));
719
720 assert!(message.contains("listen_address"));
721 assert!(message.contains("port"));
722
723 Ok(())
724 }
725
726 #[test]
727 fn invalid_health_listen_address_reports_field_name() -> Result<(), Box<dyn std::error::Error>>
728 {
729 let mut config = sample_config()?;
730 config.health_listen_address = socket("127.0.0.1:0")?;
731
732 let message = config_validation_message(validate(&mut config, None));
733
734 assert!(message.contains("health_listen_address"));
735 assert!(message.contains("port"));
736
737 Ok(())
738 }
739
740 #[test]
741 fn matching_health_and_main_listen_addresses_are_rejected()
742 -> Result<(), Box<dyn std::error::Error>> {
743 let mut config = sample_config()?;
744 config.health_listen_address = config.listen_address;
745
746 let message = config_validation_message(validate(&mut config, None));
747
748 assert!(message.contains("health_listen_address"));
749 assert!(message.contains("listen_address"));
750
751 Ok(())
752 }
753
754 #[test]
755 fn matching_health_and_main_listen_ports_are_rejected() -> Result<(), Box<dyn std::error::Error>>
756 {
757 let mut config = sample_config()?;
758 config.health_listen_address = socket("0.0.0.0:8080")?;
759
760 let message = config_validation_message(validate(&mut config, None));
761
762 assert!(message.contains("health_listen_address"));
763 assert!(message.contains("port"));
764
765 Ok(())
766 }
767
768 #[test]
772 fn zero_drain_timeout_is_accepted_because_the_field_is_ignored()
773 -> Result<(), Box<dyn std::error::Error>> {
774 let mut config = sample_config()?;
775 config.drain_timeout_ms = 0;
776
777 validate(&mut config, None)?;
778
779 assert_eq!(config.drain_timeout_ms, 0);
780
781 Ok(())
782 }
783
784 #[test]
785 fn duplicate_channel_names_are_listed() -> Result<(), Box<dyn std::error::Error>> {
786 let mut config = sample_config()?;
787 config.channels.push(ChannelDef {
788 name: "orders".to_owned(),
789 schema_ref: None,
790 durable: false,
791 loaded_schema: None,
792 });
793
794 let message = config_validation_message(validate(&mut config, None));
795
796 assert!(message.contains("duplicate"));
797 assert!(message.contains("orders"));
798
799 Ok(())
800 }
801
802 #[test]
803 fn unreachable_persistence_path_reports_path() -> Result<(), Box<dyn std::error::Error>> {
804 let mut config = sample_config()?;
805 let path = unique_temp_dir("missing");
806 config.persistence_path = Some(path.clone());
807
808 let message = config_validation_message(validate(&mut config, None));
809
810 assert!(message.contains("persistence_path"));
811 assert!(message.contains(&path.display().to_string()));
812
813 Ok(())
814 }
815
816 #[test]
817 fn file_persistence_path_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
818 let mut config = sample_config()?;
819 let path = unique_temp_dir("file");
820 fs::write(&path, "not a directory")?;
821 config.persistence_path = Some(path.clone());
822
823 let message = config_validation_message(validate(&mut config, None));
824 fs::remove_file(&path)?;
825
826 assert!(message.contains("persistence_path"));
827 assert!(message.contains("directory"));
828
829 Ok(())
830 }
831
832 #[test]
833 fn multiple_validation_errors_are_reported_together() -> Result<(), Box<dyn std::error::Error>>
834 {
835 let mut config = sample_config()?;
836 let missing_path = unique_temp_dir("multi-missing");
837 config.listen_address = socket("127.0.0.1:0")?;
838 config.channels.push(ChannelDef {
839 name: "orders".to_owned(),
840 schema_ref: None,
841 durable: false,
842 loaded_schema: None,
843 });
844 config.persistence_path = Some(missing_path.clone());
845
846 let message = config_validation_message(validate(&mut config, None));
847
848 assert!(message.contains("listen_address"));
849 assert!(message.contains("duplicate channel names: orders"));
850 assert!(message.contains(&missing_path.display().to_string()));
851
852 Ok(())
853 }
854
855 #[test]
856 fn routing_rules_reference_configured_channels() -> Result<(), Box<dyn std::error::Error>> {
857 let mut config = sample_config()?;
858 config.routing_rules[0].target_channel = "unknown".to_owned();
859
860 let message = config_validation_message(validate(&mut config, None));
861
862 assert!(message.contains("routing_rules[0].target_channel"));
863 assert!(message.contains("unknown"));
864
865 Ok(())
866 }
867
868 fn write_temp_schema(
870 label: &str,
871 contents: &str,
872 ) -> Result<PathBuf, Box<dyn std::error::Error>> {
873 let path = unique_temp_dir(label).with_extension("json");
874 fs::write(&path, contents)?;
875 Ok(path)
876 }
877
878 #[test]
879 fn absolute_schema_ref_is_loaded_and_parsed() -> Result<(), Box<dyn std::error::Error>> {
880 let schema = r#"{"type":"object","properties":{"id":{"type":"integer"}}}"#;
881 let schema_path = write_temp_schema("load-ok", schema)?;
882 let mut config = sample_config()?;
883 config.channels[0].schema_ref = Some(schema_path.clone());
884
885 let result = validate(&mut config, None);
886 fs::remove_file(&schema_path)?;
887 result?;
888
889 let loaded = config.channels[0]
890 .loaded_schema
891 .as_ref()
892 .ok_or("schema should have been loaded onto the channel")?;
893 assert_eq!(loaded.bytes, schema.as_bytes());
894 assert_eq!(
895 loaded.document.get("type").and_then(|t| t.as_str()),
896 Some("object")
897 );
898
899 Ok(())
900 }
901
902 #[test]
903 fn relative_schema_ref_resolves_against_base_dir() -> Result<(), Box<dyn std::error::Error>> {
904 let dir = unique_temp_dir("relative-base");
905 fs::create_dir_all(&dir)?;
906 let schema = r#"{"type":"object"}"#;
907 fs::write(dir.join("orders.json"), schema)?;
908
909 let mut config = sample_config()?;
910 config.channels[0].schema_ref = Some(PathBuf::from("orders.json"));
911
912 let result = validate(&mut config, Some(&dir));
913 fs::remove_dir_all(&dir)?;
914 result?;
915
916 assert!(config.channels[0].loaded_schema.is_some());
917
918 Ok(())
919 }
920
921 #[test]
922 fn missing_schema_ref_file_reports_validation_error() -> Result<(), Box<dyn std::error::Error>>
923 {
924 let missing = unique_temp_dir("missing-schema").with_extension("json");
925 let mut config = sample_config()?;
926 config.channels[0].schema_ref = Some(missing.clone());
927
928 let message = config_validation_message(validate(&mut config, None));
929
930 assert!(message.contains("schema_ref"));
931 assert!(message.contains(&missing.display().to_string()));
932 assert!(message.contains("unreadable"));
933
934 Ok(())
935 }
936
937 #[test]
938 fn invalid_json_schema_ref_reports_validation_error() -> Result<(), Box<dyn std::error::Error>>
939 {
940 let schema_path = write_temp_schema("bad-json", "{ this is not json")?;
941 let mut config = sample_config()?;
942 config.channels[0].schema_ref = Some(schema_path.clone());
943
944 let message = config_validation_message(validate(&mut config, None));
945 fs::remove_file(&schema_path)?;
946
947 assert!(message.contains("schema_ref"));
948 assert!(message.contains("not valid JSON"));
949
950 Ok(())
951 }
952
953 #[test]
954 fn valid_json_invalid_schema_ref_reports_validation_error()
955 -> Result<(), Box<dyn std::error::Error>> {
956 let schema_path = write_temp_schema("bad-schema", "[]")?;
959 let mut config = sample_config()?;
960 config.channels[0].schema_ref = Some(schema_path.clone());
961
962 let message = config_validation_message(validate(&mut config, None));
963 fs::remove_file(&schema_path)?;
964
965 assert!(message.contains("schema_ref"));
966 assert!(message.contains("not a valid JSON Schema"));
967
968 Ok(())
969 }
970
971 #[test]
972 fn present_non_empty_auth_token_passes_validation() -> Result<(), Box<dyn std::error::Error>> {
973 let mut config = sample_config()?;
974 config.auth = Some(AuthConfig {
975 pass: None,
976 token: "s3cr3t".to_owned(),
977 require_verified_worker: false,
978 });
979
980 validate(&mut config, None)?;
981
982 Ok(())
983 }
984
985 #[test]
986 fn empty_auth_token_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
987 let mut config = sample_config()?;
988 config.auth = Some(AuthConfig {
989 pass: None,
990 token: String::new(),
991 require_verified_worker: false,
992 });
993
994 let message = config_validation_message(validate(&mut config, None));
995
996 assert!(message.contains("auth.token"));
997 assert!(message.contains("must not be empty"));
998
999 Ok(())
1000 }
1001
1002 #[test]
1005 fn requiring_a_verified_worker_without_a_pass_verifier_is_rejected()
1006 -> Result<(), Box<dyn std::error::Error>> {
1007 let mut config = sample_config()?;
1008 config.auth = Some(AuthConfig {
1009 pass: None,
1010 token: "s3cr3t".to_owned(),
1011 require_verified_worker: true,
1012 });
1013
1014 let message = config_validation_message(validate(&mut config, None));
1015
1016 assert!(
1017 message.contains("auth.require_verified_worker") && message.contains("auth.pass"),
1018 "the refusal must name the flag and the missing section: {message}"
1019 );
1020 Ok(())
1021 }
1022
1023 #[test]
1024 fn auth_token_longer_than_a_connect_can_carry_is_rejected()
1025 -> Result<(), Box<dyn std::error::Error>> {
1026 let max = crate::server::connection::pre_admission::MAX_CONNECT_AUTH_TOKEN_BYTES;
1027 let mut config = sample_config()?;
1028 config.auth = Some(AuthConfig {
1029 pass: None,
1030 token: "t".repeat(max),
1031 require_verified_worker: false,
1032 });
1033 validate(&mut config, None)?;
1034
1035 config.auth = Some(AuthConfig {
1036 pass: None,
1037 token: "t".repeat(max + 1),
1038 require_verified_worker: false,
1039 });
1040 let message = config_validation_message(validate(&mut config, None));
1041
1042 assert!(message.contains("auth.token"));
1043 assert!(message.contains(&format!("at most {max}")));
1044 Ok(())
1045 }
1046
1047 #[test]
1048 fn absent_auth_section_passes_validation() -> Result<(), Box<dyn std::error::Error>> {
1049 let mut config = sample_config()?;
1050 config.auth = None;
1051
1052 validate(&mut config, None)?;
1053
1054 Ok(())
1055 }
1056
1057 #[test]
1058 fn default_profile_is_full_and_passes_validation() -> Result<(), Box<dyn std::error::Error>> {
1059 let mut config = sample_config()?;
1060
1061 assert_eq!(
1063 config.services.profile()?,
1064 crate::config::types::ServiceProfile::Full
1065 );
1066 validate(&mut config, None)?;
1067
1068 Ok(())
1069 }
1070
1071 #[test]
1072 fn unknown_profile_is_a_validation_error() -> Result<(), Box<dyn std::error::Error>> {
1073 let mut config = sample_config()?;
1074 config.services = ServicesConfig {
1075 profile: "banana".to_owned(),
1076 ..ServicesConfig::default()
1077 };
1078
1079 let message = config_validation_message(validate(&mut config, None));
1080
1081 assert!(message.contains("services.profile"));
1082 assert!(message.contains("banana"));
1083 assert!(message.contains("worker-front-door"));
1084
1085 Ok(())
1086 }
1087
1088 #[test]
1089 fn worker_front_door_profile_with_empty_topology_passes()
1090 -> Result<(), Box<dyn std::error::Error>> {
1091 let mut config = worker_front_door_config()?;
1092
1093 assert_eq!(
1094 config.services.profile()?,
1095 crate::config::types::ServiceProfile::WorkerFrontDoor
1096 );
1097 validate(&mut config, None)?;
1098
1099 Ok(())
1100 }
1101
1102 #[test]
1103 fn worker_front_door_profile_rejects_channels_persistence_and_cluster()
1104 -> Result<(), Box<dyn std::error::Error>> {
1105 let mut config = sample_config()?;
1108 config.services = ServicesConfig {
1109 profile: "worker-front-door".to_owned(),
1110 ..ServicesConfig::default()
1111 };
1112 config.persistence_path = Some(PathBuf::from("/tmp"));
1113
1114 let message = config_validation_message(validate(&mut config, None));
1115
1116 assert!(message.contains("builds no channels"));
1117 assert!(message.contains("builds no durable store"));
1118 assert!(message.contains("builds no channel cluster"));
1119
1120 Ok(())
1121 }
1122
1123 #[test]
1124 fn default_limits_pass_validation_and_carry_signed_numbers()
1125 -> Result<(), Box<dyn std::error::Error>> {
1126 let mut config = sample_config()?;
1127 assert_eq!(config.limits.max_connections, 256);
1129 assert_eq!(config.limits.max_subscriptions_per_connection, 32);
1130 assert_eq!(config.limits.max_conversations_per_connection, 32);
1131 assert_eq!(config.limits.max_pending_pushes_per_connection, 32);
1132 assert_eq!(
1133 config
1134 .limits
1135 .max_pending_conversation_replies_per_connection,
1136 32
1137 );
1138 assert_eq!(config.limits.max_pending_replies_per_conversation, 8);
1139 assert_eq!(config.limits.max_connection_inbox_bytes, 4 * 1024 * 1024);
1140 assert_eq!(config.limits.max_connection_outbound_bytes, 4 * 1024 * 1024);
1145 assert_eq!(
1146 config.limits.max_connection_outbound_bytes, config.limits.max_connection_inbox_bytes,
1147 "the two per-connection byte budgets are deliberately equal"
1148 );
1149 assert_eq!(config.limits.max_subscription_inbox_depth, 4096);
1157 assert_eq!(
1158 config.limits.max_connection_inbox_bytes / config.limits.max_subscription_inbox_depth,
1159 1024,
1160 "the byte-vs-count crossover must stay at 1 KiB per record"
1161 );
1162 assert_eq!(config.limits.delivery_slice_budget, 32);
1163 validate(&mut config, None)?;
1164 Ok(())
1165 }
1166
1167 #[test]
1170 fn zero_limits_are_typed_config_errors() -> Result<(), Box<dyn std::error::Error>> {
1171 type LimitMutator = (&'static str, fn(&mut ServerConfig));
1172 let mutators: [LimitMutator; 10] = [
1173 ("max_connections", |c| c.limits.max_connections = 0),
1174 ("max_subscriptions_per_connection", |c| {
1175 c.limits.max_subscriptions_per_connection = 0;
1176 }),
1177 ("max_conversations_per_connection", |c| {
1178 c.limits.max_conversations_per_connection = 0;
1179 }),
1180 ("max_pending_pushes_per_connection", |c| {
1181 c.limits.max_pending_pushes_per_connection = 0;
1182 }),
1183 ("max_pending_conversation_replies_per_connection", |c| {
1184 c.limits.max_pending_conversation_replies_per_connection = 0;
1185 }),
1186 ("max_pending_replies_per_conversation", |c| {
1187 c.limits.max_pending_replies_per_conversation = 0;
1188 }),
1189 ("max_connection_inbox_bytes", |c| {
1190 c.limits.max_connection_inbox_bytes = 0;
1191 }),
1192 ("max_connection_outbound_bytes", |c| {
1193 c.limits.max_connection_outbound_bytes = 0;
1194 }),
1195 ("max_subscription_inbox_depth", |c| {
1196 c.limits.max_subscription_inbox_depth = 0;
1197 }),
1198 ("delivery_slice_budget", |c| {
1199 c.limits.delivery_slice_budget = 0;
1200 }),
1201 ];
1202 for (field, mutate) in mutators {
1203 let mut config = sample_config()?;
1204 mutate(&mut config);
1205 let message = config_validation_message(validate(&mut config, None));
1206 assert!(
1207 message.contains(&format!("limits.{field}")),
1208 "zero {field} must report a typed limits.{field} error, got: {message}"
1209 );
1210 assert!(
1211 message.contains("greater than zero"),
1212 "the {field} refusal must say why: {message}"
1213 );
1214 }
1215 Ok(())
1216 }
1217
1218 const fn sample_participant() -> ParticipantConfig {
1220 ParticipantConfig {
1221 wire_frame_limit: 65_536,
1222 attach_receipt_ttl_ms: 60_000,
1223 receipt_provenance_ttl_ms: 600_000,
1224 live_receipt_server_report_threshold: 1_024,
1225 max_live_attach_receipts_per_participant: 8,
1226 receipt_provenance_server_report_threshold: 4_096,
1227 receipt_provenance_per_conversation_report_threshold: 256,
1228 max_receipt_provenance_per_participant: 64,
1229 max_retired_identity_slots_server: 1_024,
1230 identity_slots: 4,
1231 observer_recovery_max_entries: 64,
1232 max_semantic_conversations_per_connection: 32,
1233 max_ordinary_record_entries: 1,
1234 max_ordinary_record_bytes: 131_072,
1235 max_generated_marker_entries: 1,
1236 max_generated_marker_bytes: 4_096,
1237 mandatory_transaction_bound_entries: 4,
1238 mandatory_transaction_bound_bytes: 16_384,
1239 full_recovery_claim_entries: 4,
1240 full_recovery_claim_bytes: 16_384,
1241 retained_capacity_entries: 2_048,
1242 retained_capacity_bytes: 16_777_216,
1243 max_retained_record_rows: 1_024,
1244 closure_episode_churn_limit: 1_024,
1245 }
1246 }
1247
1248 #[test]
1249 fn valid_participant_section_passes_validation() -> Result<(), Box<dyn std::error::Error>> {
1250 let mut config = sample_config()?;
1251 config.participant = Some(sample_participant());
1252 let temp_dir = std::env::temp_dir().join(format!(
1253 "liminal-server-participant-validation-{}-{}",
1254 std::process::id(),
1255 NEXT_TEMP_DIR_ID.fetch_add(1, Ordering::Relaxed)
1256 ));
1257 fs::create_dir_all(&temp_dir)?;
1258 config.persistence_path = Some(temp_dir.clone());
1259 let result = validate(&mut config, None);
1260 fs::remove_dir_all(&temp_dir)?;
1261 assert!(result.is_ok(), "expected valid config, got {result:?}");
1262 Ok(())
1263 }
1264
1265 #[test]
1266 fn zero_participant_values_are_typed_config_errors() -> Result<(), Box<dyn std::error::Error>> {
1267 type ParticipantMutator = (&'static str, fn(&mut ParticipantConfig));
1268 let mutators: [ParticipantMutator; 12] = [
1269 ("wire_frame_limit", |p| p.wire_frame_limit = 0),
1270 ("attach_receipt_ttl_ms", |p| p.attach_receipt_ttl_ms = 0),
1271 ("receipt_provenance_ttl_ms", |p| {
1272 p.receipt_provenance_ttl_ms = 0;
1273 }),
1274 ("live_receipt_server_report_threshold", |p| {
1275 p.live_receipt_server_report_threshold = 0;
1276 }),
1277 ("max_live_attach_receipts_per_participant", |p| {
1278 p.max_live_attach_receipts_per_participant = 0;
1279 }),
1280 ("receipt_provenance_server_report_threshold", |p| {
1281 p.receipt_provenance_server_report_threshold = 0;
1282 }),
1283 (
1284 "receipt_provenance_per_conversation_report_threshold",
1285 |p| {
1286 p.receipt_provenance_per_conversation_report_threshold = 0;
1287 },
1288 ),
1289 ("max_receipt_provenance_per_participant", |p| {
1290 p.max_receipt_provenance_per_participant = 0;
1291 }),
1292 ("max_retired_identity_slots_server", |p| {
1293 p.max_retired_identity_slots_server = 0;
1294 }),
1295 ("identity_slots", |p| p.identity_slots = 0),
1296 ("observer_recovery_max_entries", |p| {
1297 p.observer_recovery_max_entries = 0;
1298 }),
1299 ("max_semantic_conversations_per_connection", |p| {
1300 p.max_semantic_conversations_per_connection = 0;
1301 }),
1302 ];
1303 for (field, mutate) in mutators {
1304 let mut config = sample_config()?;
1305 let mut participant = sample_participant();
1306 mutate(&mut participant);
1307 config.participant = Some(participant);
1308 let message = config_validation_message(validate(&mut config, None));
1309 assert!(
1310 message.contains(&format!("participant.{field}")),
1311 "expected typed error for participant.{field}, got: {message}"
1312 );
1313 }
1314 Ok(())
1315 }
1316
1317 #[test]
1318 fn provenance_ttl_shorter_than_receipt_is_a_typed_config_error()
1319 -> Result<(), Box<dyn std::error::Error>> {
1320 let mut config = sample_config()?;
1321 let mut participant = sample_participant();
1322 participant.receipt_provenance_ttl_ms = participant.attach_receipt_ttl_ms - 1;
1323 config.participant = Some(participant);
1324 let message = config_validation_message(validate(&mut config, None));
1325 assert!(
1326 message.contains("participant.receipt_provenance_ttl_ms"),
1327 "expected TTL-ordering error, got: {message}"
1328 );
1329 Ok(())
1330 }
1331
1332 #[test]
1333 fn undersized_wire_frame_limit_is_a_typed_config_error()
1334 -> Result<(), Box<dyn std::error::Error>> {
1335 let mut config = sample_config()?;
1336 let mut participant = sample_participant();
1337 participant.wire_frame_limit = 1;
1338 config.participant = Some(participant);
1339 let message = config_validation_message(validate(&mut config, None));
1340 assert!(
1341 message.contains("participant.wire_frame_limit"),
1342 "expected minimum-frame error, got: {message}"
1343 );
1344 Ok(())
1345 }
1346
1347 #[test]
1348 fn worker_front_door_rejects_participant_section() -> Result<(), Box<dyn std::error::Error>> {
1349 let mut config = worker_front_door_config()?;
1350 config.participant = Some(sample_participant());
1351 let message = config_validation_message(validate(&mut config, None));
1352 assert!(
1353 message.contains("installs no participant service"),
1354 "expected worker-front-door participant rejection, got: {message}"
1355 );
1356 Ok(())
1357 }
1358
1359 fn sample_websocket()
1362 -> Result<crate::config::types::WebSocketConfig, Box<dyn std::error::Error>> {
1363 Ok(crate::config::types::WebSocketConfig {
1364 listen_address: socket("127.0.0.1:8082")?,
1365 path: "/liminal".to_owned(),
1366 allowed_origins: vec!["https://app.example.com".to_owned()],
1367 ping_interval_ms: Some(30_000),
1368 })
1369 }
1370
1371 #[test]
1372 fn valid_websocket_section_passes_validation() -> Result<(), Box<dyn std::error::Error>> {
1373 let mut config = sample_config()?;
1374 config.websocket = Some(sample_websocket()?);
1375 validate(&mut config, None)?;
1376 Ok(())
1377 }
1378
1379 #[test]
1380 fn websocket_empty_origin_list_is_valid_fail_closed_configuration()
1381 -> Result<(), Box<dyn std::error::Error>> {
1382 let mut config = sample_config()?;
1383 let mut websocket = sample_websocket()?;
1384 websocket.allowed_origins = Vec::new();
1385 config.websocket = Some(websocket);
1386 validate(&mut config, None)?;
1387 Ok(())
1388 }
1389
1390 #[test]
1391 fn websocket_zero_port_is_a_typed_config_error() -> Result<(), Box<dyn std::error::Error>> {
1392 let mut config = sample_config()?;
1393 let mut websocket = sample_websocket()?;
1394 websocket.listen_address = socket("127.0.0.1:0")?;
1395 config.websocket = Some(websocket);
1396 let message = config_validation_message(validate(&mut config, None));
1397 assert!(
1398 message.contains("websocket.listen_address"),
1399 "expected websocket.listen_address error, got: {message}"
1400 );
1401 Ok(())
1402 }
1403
1404 #[test]
1405 fn websocket_address_conflicts_are_typed_config_errors()
1406 -> Result<(), Box<dyn std::error::Error>> {
1407 for conflict in ["127.0.0.1:8080", "127.0.0.1:8081", "127.0.0.1:9000"] {
1408 let mut config = sample_config()?;
1409 let mut websocket = sample_websocket()?;
1410 websocket.listen_address = socket(conflict)?;
1411 config.websocket = Some(websocket);
1412 let message = config_validation_message(validate(&mut config, None));
1413 assert!(
1414 message.contains("websocket.listen_address: must differ from"),
1415 "expected an address-conflict error for {conflict}, got: {message}"
1416 );
1417 }
1418 Ok(())
1419 }
1420
1421 #[test]
1422 fn websocket_path_rules_are_typed_config_errors() -> Result<(), Box<dyn std::error::Error>> {
1423 for (bad_path, expectation) in [
1424 ("", "must not be empty"),
1425 ("liminal", "must start with '/'"),
1426 ("/limi nal", "whitespace"),
1427 ("/liminal?x=1", "query"),
1428 ("/liminal#frag", "query"),
1429 ] {
1430 let mut config = sample_config()?;
1431 let mut websocket = sample_websocket()?;
1432 websocket.path = bad_path.to_owned();
1433 config.websocket = Some(websocket);
1434 let message = config_validation_message(validate(&mut config, None));
1435 assert!(
1436 message.contains("websocket.path") && message.contains(expectation),
1437 "expected websocket.path '{bad_path}' to report '{expectation}', got: {message}"
1438 );
1439 }
1440 Ok(())
1441 }
1442
1443 #[test]
1444 fn websocket_malformed_origin_entries_are_typed_config_errors()
1445 -> Result<(), Box<dyn std::error::Error>> {
1446 for (bad_origin, expectation) in [
1447 ("", "must not be empty"),
1448 ("app.example.com", "serialized origin"),
1449 ("https://app.example.com/", "path or trailing slash"),
1450 ("https://app.example.com/app", "path or trailing slash"),
1451 ("https:// app.example.com", "whitespace"),
1452 ] {
1453 let mut config = sample_config()?;
1454 let mut websocket = sample_websocket()?;
1455 websocket.allowed_origins = vec![bad_origin.to_owned()];
1456 config.websocket = Some(websocket);
1457 let message = config_validation_message(validate(&mut config, None));
1458 assert!(
1459 message.contains("websocket.allowed_origins") && message.contains(expectation),
1460 "expected origin '{bad_origin}' to report '{expectation}', got: {message}"
1461 );
1462 }
1463 Ok(())
1464 }
1465
1466 #[test]
1467 fn websocket_duplicate_origin_is_a_typed_config_error() -> Result<(), Box<dyn std::error::Error>>
1468 {
1469 let mut config = sample_config()?;
1470 let mut websocket = sample_websocket()?;
1471 websocket.allowed_origins = vec![
1472 "https://app.example.com".to_owned(),
1473 "https://app.example.com".to_owned(),
1474 ];
1475 config.websocket = Some(websocket);
1476 let message = config_validation_message(validate(&mut config, None));
1477 assert!(
1478 message.contains("duplicate origin"),
1479 "expected a duplicate-origin error, got: {message}"
1480 );
1481 Ok(())
1482 }
1483
1484 #[test]
1485 fn websocket_zero_ping_interval_is_a_typed_config_error()
1486 -> Result<(), Box<dyn std::error::Error>> {
1487 let mut config = sample_config()?;
1488 let mut websocket = sample_websocket()?;
1489 websocket.ping_interval_ms = Some(0);
1490 config.websocket = Some(websocket);
1491 let message = config_validation_message(validate(&mut config, None));
1492 assert!(
1493 message.contains("websocket.ping_interval_ms"),
1494 "expected a zero-interval error, got: {message}"
1495 );
1496 Ok(())
1497 }
1498
1499 #[test]
1500 fn websocket_absent_ping_interval_means_disabled_and_valid()
1501 -> Result<(), Box<dyn std::error::Error>> {
1502 let mut config = sample_config()?;
1503 let mut websocket = sample_websocket()?;
1504 websocket.ping_interval_ms = None;
1505 config.websocket = Some(websocket);
1506 validate(&mut config, None)?;
1507 Ok(())
1508 }
1509}
1510
1511#[cfg(test)]
1512mod self_stream_tests {
1513 use super::super::types::ServicesConfig;
1514
1515 #[test]
1516 fn self_stream_bounds_default_and_override() -> Result<(), Box<dyn std::error::Error>> {
1517 let defaults: ServicesConfig = toml::from_str("")?;
1518 assert_eq!(defaults.self_streams.cost_rows, 10_000);
1519 assert_eq!(defaults.self_streams.events_rows, 100_000);
1520 let configured: ServicesConfig =
1521 toml::from_str("[self_streams]\ncost_rows = 3\nevents_rows = 5\n")?;
1522 assert_eq!(configured.self_streams.cost_rows, 3);
1523 assert_eq!(configured.self_streams.events_rows, 5);
1524 configured.profile()?;
1525 Ok(())
1526 }
1527
1528 #[test]
1529 fn self_stream_bounds_validate_without_participant_and_independently_of_profile()
1530 -> Result<(), Box<dyn std::error::Error>> {
1531 let mut config: super::ServerConfig = toml::from_str(
1532 "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",
1533 )?;
1534 assert!(config.participant.is_none());
1535 assert_eq!(config.services.profile()?, super::ServiceProfile::Full);
1536 let error = super::validate(&mut config, None)
1537 .err()
1538 .ok_or("disabled self streams accepted invalid bounds")?;
1539 let message = error.to_string();
1540 assert!(message.contains("services.self_streams.cost_rows"));
1541 assert!(message.contains("services.self_streams.events_rows"));
1542 config.services.profile = "invalid".to_owned();
1543 let error = super::validate(&mut config, None)
1544 .err()
1545 .ok_or("invalid profile and bounds were accepted")?;
1546 let message = error.to_string();
1547 assert!(message.contains("services.profile"));
1548 assert!(message.contains("services.self_streams.cost_rows"));
1549 assert!(message.contains("services.self_streams.events_rows"));
1550 Ok(())
1551 }
1552
1553 #[test]
1554 fn zero_self_stream_bounds_are_named() -> Result<(), Box<dyn std::error::Error>> {
1555 let configured: ServicesConfig =
1556 toml::from_str("[self_streams]\ncost_rows = 0\nevents_rows = 0\n")?;
1557 let Err(error) = configured.self_streams.validate() else {
1558 return Err("zero row bounds must be refused".into());
1559 };
1560 let message = error.to_string();
1561 assert!(message.contains("services.self_streams.cost_rows"));
1562 assert!(message.contains("services.self_streams.events_rows"));
1563 Ok(())
1564 }
1565}