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