1use std::collections::BTreeMap;
9use std::fmt;
10use std::path::PathBuf;
11
12use serde::{Deserialize, Serialize};
13
14use super::endpoint::ConfListen;
15use super::enums::{
16 ConsistencyLevel, DataStore, Distribution, HashType, Membership, SecureServerOption, Transport,
17};
18use super::error::ConfError;
19use super::server::{ConfDynSeed, ConfServer};
20use super::tokens::TokenList;
21
22pub mod defaults {
25 pub const TIMEOUT_MS: i64 = 5_000;
27 pub const LISTEN_BACKLOG: i64 = 512;
29 pub const CLIENT_CONNECTIONS: i64 = 0;
31 pub const DATA_STORE: i64 = 0;
33 pub const PRECONNECT: bool = false;
39 pub const AUTO_EJECT_HOSTS: bool = true;
41 pub const SERVER_RETRY_TIMEOUT_MS: i64 = 10 * 1000;
43 pub const SERVER_FAILURE_LIMIT: i64 = 3;
45 pub const DYN_READ_TIMEOUT_MS: i64 = 10_000;
47 pub const DYN_WRITE_TIMEOUT_MS: i64 = 10_000;
49 pub const DYN_CONNECTIONS: i64 = 100;
51 pub const GOS_INTERVAL_MS: i64 = 30_000;
53 pub const ENABLE_HINTED_HANDOFF: bool = false;
56 pub const HINT_TTL_SECONDS: u64 = 86_400;
59 pub const HINT_STORE_MAX_BYTES: u64 = 64 * 1024 * 1024;
62 pub const HINT_DRAIN_INTERVAL_MS: u64 = 30_000;
65 pub const CONN_MSG_RATE: u32 = 50_000;
67 pub const STATS_INTERVAL_MS: i64 = 30 * 1000;
69 pub const STATS_PNAME: &str = "0.0.0.0:22222";
71 pub const DATASTORE_CONNECTIONS: u8 = 1;
73 pub const LOCAL_PEER_CONNECTIONS: u8 = 1;
75 pub const REMOTE_PEER_CONNECTIONS: u8 = 1;
77 pub const RACK: &str = "localrack";
79 pub const DC: &str = "localdc";
81 pub const SECURE_SERVER_OPTION: &str = "none";
83 pub const CONSISTENCY: &str = "DC_ONE";
85 pub const SEED_PROVIDER: &str = "simple_provider";
87 pub const ENV: &str = "aws";
89 pub const PEM_KEY_FILE: &str = "conf/dynomite.pem";
91 pub const RECON_KEY_FILE: &str = "conf/recon_key.pem";
93 pub const RECON_IV_FILE: &str = "conf/recon_iv.pem";
95 pub const RECON_INTERVAL_SECONDS: u64 = 300;
99 pub const MBUF_MIN_SIZE: i64 = 512;
101 pub const MBUF_MAX_SIZE: i64 = 512_000;
103 pub const ALLOC_MSGS_MIN: i64 = 100_000;
105 pub const ALLOC_MSGS_MAX: i64 = 1_000_000;
107}
108
109#[derive(Debug, Clone, Default, Eq, PartialEq, Serialize, Deserialize)]
122#[serde(transparent)]
123pub struct Servers(pub(crate) Vec<ConfServer>);
124
125impl Servers {
126 pub fn from_vec(v: Vec<ConfServer>) -> Self {
137 Self(v)
138 }
139}
140
141impl Servers {
142 pub fn entries(&self) -> &[ConfServer] {
152 &self.0
153 }
154 pub fn len(&self) -> usize {
163 self.0.len()
164 }
165 pub fn is_empty(&self) -> bool {
174 self.0.is_empty()
175 }
176 pub fn datastore(&self) -> Option<&ConfServer> {
187 self.0.first()
188 }
189}
190
191#[derive(Debug, Clone, Default, Serialize, Deserialize)]
204#[serde(deny_unknown_fields, default)]
205pub struct ConfPool {
206 pub listen: Option<ConfListen>,
208 pub dyn_listen: Option<ConfListen>,
210 pub stats_listen: Option<ConfListen>,
212
213 pub hash: Option<HashType>,
215 pub hash_tag: Option<String>,
217
218 #[serde(default)]
224 pub distribution: Option<Distribution>,
225 #[serde(default)]
232 pub distribution_shadow: Option<Distribution>,
233 #[serde(default)]
235 pub server_connections: Option<i64>,
236
237 pub timeout: Option<i64>,
239 pub backlog: Option<i64>,
241 pub client_connections: Option<i64>,
243 #[serde(default, deserialize_with = "deserialize_data_store")]
249 pub data_store: Option<i64>,
250 #[serde(default)]
256 pub noxu_path: Option<PathBuf>,
257 #[serde(default)]
268 pub search_index_dir: Option<PathBuf>,
269 pub preconnect: Option<bool>,
271 #[serde(default)]
278 pub redis_requirepass: Option<String>,
279 pub auto_eject_hosts: Option<bool>,
281 pub server_retry_timeout: Option<i64>,
283 pub server_failure_limit: Option<i64>,
285
286 pub servers: Option<Servers>,
288
289 pub dyn_read_timeout: Option<i64>,
291 pub dyn_write_timeout: Option<i64>,
293 pub dyn_seed_provider: Option<String>,
295 pub dyn_seeds: Option<Vec<ConfDynSeed>>,
297 pub dyn_port: Option<i64>,
299 pub dyn_connections: Option<i64>,
301 pub rack: Option<String>,
303 pub tokens: Option<TokenList>,
305 pub gos_interval: Option<i64>,
307 pub secure_server_option: Option<String>,
309 pub pem_key_file: Option<String>,
311 pub recon_key_file: Option<String>,
313 pub recon_iv_file: Option<String>,
315 #[serde(default)]
324 pub recon_interval_seconds: Option<u64>,
325 pub datacenter: Option<String>,
327 pub env: Option<String>,
329 pub conn_msg_rate: Option<u32>,
331 pub read_consistency: Option<String>,
333 pub write_consistency: Option<String>,
335 pub stats_interval: Option<i64>,
337 pub enable_gossip: Option<bool>,
339 #[serde(default)]
345 pub membership: Option<Membership>,
346 #[serde(default)]
353 pub peer_tls_cert: Option<PathBuf>,
354 #[serde(default)]
357 pub peer_tls_key: Option<PathBuf>,
358 #[serde(default)]
366 pub peer_tls_ca: Option<PathBuf>,
367 #[serde(default)]
387 pub peer_tls_profiles: BTreeMap<String, ConfTlsProfile>,
388 pub mbuf_size: Option<i64>,
390 pub max_msgs: Option<i64>,
392 pub datastore_connections: Option<u8>,
394 pub local_peer_connections: Option<u8>,
396 pub remote_peer_connections: Option<u8>,
398 pub read_repairs_enabled: Option<bool>,
400 #[serde(default)]
412 pub enable_hinted_handoff: Option<bool>,
413 #[serde(default)]
418 pub hint_ttl_seconds: Option<u64>,
419 #[serde(default)]
427 pub hint_store_max_bytes: Option<u64>,
428 #[serde(default)]
432 pub hint_drain_interval_ms: Option<u64>,
433 #[serde(default)]
443 pub hint_dir: Option<PathBuf>,
444 pub log_format: Option<String>,
452
453 #[serde(default)]
459 pub observability: Option<ObservabilityConfig>,
460
461 #[serde(default)]
475 pub bucket_types: Vec<ConfBucketType>,
476
477 #[serde(default)]
483 pub default_bucket_type: Option<String>,
484
485 #[serde(default)]
498 pub riak: Option<ConfRiak>,
499
500 #[serde(default)]
513 pub transport: Option<Transport>,
514
515 #[serde(default)]
520 pub quic_cert_file: Option<PathBuf>,
521
522 #[serde(default)]
527 pub quic_key_file: Option<PathBuf>,
528}
529
530#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
557#[serde(deny_unknown_fields, default)]
558pub struct ConfRiak {
559 pub pbc_listen: Option<String>,
563 pub http_listen: Option<String>,
567 pub quic_listen: Option<String>,
590 pub aae_enabled: Option<bool>,
593 pub aae_full_sweep_interval_seconds: Option<u64>,
597 pub aae_segment_interval_seconds: Option<u64>,
601 #[serde(default)]
608 pub tls_cert: Option<PathBuf>,
609 #[serde(default)]
612 pub tls_key: Option<PathBuf>,
613 #[serde(default)]
619 pub tls_ca: Option<PathBuf>,
620 #[serde(default)]
629 pub quic_tls_cert: Option<PathBuf>,
630 #[serde(default)]
633 pub quic_tls_key: Option<PathBuf>,
634 #[serde(default)]
650 pub wasm_modules: Option<Vec<ConfRiakWasmModule>>,
651}
652
653#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
668#[serde(deny_unknown_fields)]
669pub struct ConfRiakWasmModule {
670 pub id: String,
673 pub path: PathBuf,
676}
677
678#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
702#[serde(deny_unknown_fields, default)]
703pub struct ConfTlsProfile {
704 pub cert: Option<PathBuf>,
706 pub key: Option<PathBuf>,
708 pub ca: Option<PathBuf>,
716}
717
718impl ConfTlsProfile {
719 pub fn validate(&self, dc: &str) -> Result<(), ConfError> {
745 match (self.cert.as_deref(), self.key.as_deref()) {
746 (Some(_), Some(_)) | (None, None) => {}
747 (Some(c), None) => {
748 return Err(ConfError::BadServer {
749 field: "peer_tls_profiles.cert",
750 value: c.display().to_string(),
751 reason: format!(
752 "peer_tls_profiles[{dc}].cert is set but .key is not; both must be set together"
753 ),
754 });
755 }
756 (None, Some(k)) => {
757 return Err(ConfError::BadServer {
758 field: "peer_tls_profiles.key",
759 value: k.display().to_string(),
760 reason: format!(
761 "peer_tls_profiles[{dc}].key is set but .cert is not; both must be set together"
762 ),
763 });
764 }
765 }
766 if self.ca.is_some() && self.cert.is_none() {
767 return Err(ConfError::BadServer {
768 field: "peer_tls_profiles.ca",
769 value: self
770 .ca
771 .as_ref()
772 .map_or_else(String::new, |p| p.display().to_string()),
773 reason: format!(
774 "peer_tls_profiles[{dc}].ca requires .cert and .key to also be set"
775 ),
776 });
777 }
778 Ok(())
779 }
780}
781
782impl ConfRiak {
783 pub fn validate(&self) -> Result<(), ConfError> {
802 if let Some(addr) = self.pbc_listen.as_deref() {
803 validate_riak_addr("pbc_listen", addr)?;
804 }
805 if let Some(addr) = self.http_listen.as_deref() {
806 validate_riak_addr("http_listen", addr)?;
807 }
808 if let Some(addr) = self.quic_listen.as_deref() {
809 validate_riak_addr("quic_listen", addr)?;
810 let has_quic_pair = self.quic_tls_cert.is_some() && self.quic_tls_key.is_some();
815 let has_shared_pair = self.tls_cert.is_some() && self.tls_key.is_some();
816 if !has_quic_pair && !has_shared_pair {
817 return Err(ConfError::BadServer {
818 field: "quic_listen",
819 value: addr.to_string(),
820 reason: "quic_listen requires a TLS cert/key pair: set quic_tls_cert + \
821 quic_tls_key (QUIC-only, TCP/HTTP stay plaintext) or tls_cert + \
822 tls_key (shared)"
823 .into(),
824 });
825 }
826 }
827 if let Some(n) = self.aae_full_sweep_interval_seconds {
828 if n == 0 {
829 return Err(ConfError::BadServer {
830 field: "aae_full_sweep_interval_seconds",
831 value: n.to_string(),
832 reason: "must be > 0".into(),
833 });
834 }
835 }
836 if let Some(n) = self.aae_segment_interval_seconds {
837 if n == 0 {
838 return Err(ConfError::BadServer {
839 field: "aae_segment_interval_seconds",
840 value: n.to_string(),
841 reason: "must be > 0".into(),
842 });
843 }
844 }
845 if let (Some(seg), Some(full)) = (
846 self.aae_segment_interval_seconds,
847 self.aae_full_sweep_interval_seconds,
848 ) {
849 if seg > full {
850 return Err(ConfError::BadServer {
851 field: "aae_segment_interval_seconds",
852 value: seg.to_string(),
853 reason: format!("must be <= aae_full_sweep_interval_seconds ({full})"),
854 });
855 }
856 }
857 validate_tls_pair(
858 "tls_cert",
859 "tls_key",
860 self.tls_cert.as_deref(),
861 self.tls_key.as_deref(),
862 )?;
863 validate_tls_pair(
864 "quic_tls_cert",
865 "quic_tls_key",
866 self.quic_tls_cert.as_deref(),
867 self.quic_tls_key.as_deref(),
868 )?;
869 if self.tls_ca.is_some() && self.tls_cert.is_none() {
870 return Err(ConfError::BadServer {
871 field: "tls_ca",
872 value: self
873 .tls_ca
874 .as_ref()
875 .map_or_else(String::new, |p| p.display().to_string()),
876 reason: "requires tls_cert and tls_key to also be set".into(),
877 });
878 }
879 if let Some(modules) = self.wasm_modules.as_deref() {
880 let mut seen: std::collections::BTreeSet<&str> = std::collections::BTreeSet::new();
881 for m in modules {
882 if m.id.is_empty() {
883 return Err(ConfError::BadServer {
884 field: "wasm_modules.id",
885 value: String::new(),
886 reason: "wasm module id must not be empty".into(),
887 });
888 }
889 if !seen.insert(m.id.as_str()) {
890 return Err(ConfError::BadServer {
891 field: "wasm_modules.id",
892 value: m.id.clone(),
893 reason: "wasm module ids must be unique".into(),
894 });
895 }
896 if !m.path.is_file() {
897 return Err(ConfError::BadServer {
898 field: "wasm_modules.path",
899 value: m.path.display().to_string(),
900 reason: format!("wasm module file not found for id '{}'", m.id),
901 });
902 }
903 }
904 }
905 Ok(())
906 }
907}
908
909fn validate_tls_pair(
913 cert_field: &'static str,
914 key_field: &'static str,
915 cert: Option<&std::path::Path>,
916 key: Option<&std::path::Path>,
917) -> Result<(), ConfError> {
918 match (cert, key) {
919 (Some(_), Some(_)) | (None, None) => Ok(()),
920 (Some(c), None) => Err(ConfError::BadServer {
921 field: cert_field,
922 value: c.display().to_string(),
923 reason: format!(
924 "{cert_field} is set but {key_field} is not; both must be set together"
925 ),
926 }),
927 (None, Some(k)) => Err(ConfError::BadServer {
928 field: key_field,
929 value: k.display().to_string(),
930 reason: format!(
931 "{key_field} is set but {cert_field} is not; both must be set together"
932 ),
933 }),
934 }
935}
936
937fn validate_riak_addr(field: &'static str, value: &str) -> Result<(), ConfError> {
938 use std::net::ToSocketAddrs;
939 if value.is_empty() {
940 return Err(ConfError::BadServer {
941 field,
942 value: value.to_string(),
943 reason: "riak listen address must not be empty".into(),
944 });
945 }
946 if value.parse::<std::net::SocketAddr>().is_ok() {
950 return Ok(());
951 }
952 match value.to_socket_addrs() {
953 Ok(mut iter) => {
954 if iter.next().is_some() {
955 Ok(())
956 } else {
957 Err(ConfError::BadServer {
958 field,
959 value: value.to_string(),
960 reason: "resolved to no addresses".into(),
961 })
962 }
963 }
964 Err(e) => Err(ConfError::BadServer {
965 field,
966 value: value.to_string(),
967 reason: format!("could not resolve: {e}"),
968 }),
969 }
970}
971
972#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
1004#[serde(deny_unknown_fields)]
1005pub struct ConfBucketType {
1006 pub name: String,
1011 pub read_consistency: String,
1017 pub write_consistency: String,
1020 #[serde(default)]
1025 pub n_val: u8,
1026}
1027
1028impl ConfBucketType {
1029 pub fn read_level(&self) -> Result<ConsistencyLevel, ConfError> {
1044 ConsistencyLevel::parse("read_consistency", &self.read_consistency)
1045 }
1046
1047 pub fn write_level(&self) -> Result<ConsistencyLevel, ConfError> {
1062 ConsistencyLevel::parse("write_consistency", &self.write_consistency)
1063 }
1064}
1065
1066#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)]
1085#[serde(deny_unknown_fields, default)]
1086pub struct ObservabilityConfig {
1087 pub otlp_traces_endpoint: Option<String>,
1092 pub otlp_logs_endpoint: Option<String>,
1098 pub service_name: Option<String>,
1101 pub traces_sampling: Option<f64>,
1104}
1105
1106impl ConfPool {
1107 #[must_use]
1127 pub fn resolved_distribution(&self) -> Distribution {
1128 match self.distribution {
1129 None | Some(Distribution::Vnode) => Distribution::Vnode,
1130 Some(Distribution::RandomSlicing) => Distribution::RandomSlicing,
1131 Some(other) => {
1132 tracing::warn!(
1133 target: "dynomite::conf",
1134 distribution = other.as_str(),
1135 "distribution mode '{}' is a legacy alias and resolves to 'vnode'; \
1136 update the YAML to either 'vnode' or 'random_slicing'",
1137 other
1138 );
1139 Distribution::Vnode
1140 }
1141 }
1142 }
1143}
1144
1145impl ConfPool {
1146 pub fn apply_defaults(&mut self) {
1158 if self.dyn_seed_provider.is_none() {
1159 self.dyn_seed_provider = Some(defaults::SEED_PROVIDER.to_string());
1160 }
1161 if self.hash.is_none() {
1162 self.hash = Some(HashType::Murmur);
1163 }
1164 if self.timeout.is_none() {
1165 self.timeout = Some(defaults::TIMEOUT_MS);
1166 }
1167 if self.backlog.is_none() {
1168 self.backlog = Some(defaults::LISTEN_BACKLOG);
1169 }
1170 self.client_connections = Some(defaults::CLIENT_CONNECTIONS);
1173 if self.data_store.is_none() {
1174 self.data_store = Some(defaults::DATA_STORE);
1175 }
1176 if self.preconnect.is_none() {
1177 self.preconnect = Some(defaults::PRECONNECT);
1178 }
1179 if self.auto_eject_hosts.is_none() {
1180 self.auto_eject_hosts = Some(defaults::AUTO_EJECT_HOSTS);
1181 }
1182 if self.server_retry_timeout.is_none() {
1183 self.server_retry_timeout = Some(defaults::SERVER_RETRY_TIMEOUT_MS);
1184 }
1185 if self.server_failure_limit.is_none() {
1186 self.server_failure_limit = Some(defaults::SERVER_FAILURE_LIMIT);
1187 }
1188 if self.dyn_read_timeout.is_none() {
1189 self.dyn_read_timeout = Some(defaults::DYN_READ_TIMEOUT_MS);
1190 }
1191 if self.dyn_write_timeout.is_none() {
1192 self.dyn_write_timeout = Some(defaults::DYN_WRITE_TIMEOUT_MS);
1193 }
1194 if self.dyn_connections.is_none() {
1195 self.dyn_connections = Some(defaults::DYN_CONNECTIONS);
1196 }
1197 if self.gos_interval.is_none() {
1198 self.gos_interval = Some(defaults::GOS_INTERVAL_MS);
1199 }
1200 if self.conn_msg_rate.is_none() {
1201 self.conn_msg_rate = Some(defaults::CONN_MSG_RATE);
1202 }
1203 if self.rack.is_none() {
1204 self.rack = Some(defaults::RACK.to_string());
1205 }
1206 if self.datacenter.is_none() {
1207 self.datacenter = Some(defaults::DC.to_string());
1208 }
1209 if self.secure_server_option.is_none() {
1210 self.secure_server_option = Some(defaults::SECURE_SERVER_OPTION.to_string());
1211 }
1212 if self.read_consistency.is_none() {
1213 self.read_consistency = Some(defaults::CONSISTENCY.to_string());
1214 }
1215 if self.write_consistency.is_none() {
1216 self.write_consistency = Some(defaults::CONSISTENCY.to_string());
1217 }
1218 if self.stats_interval.is_none() {
1219 self.stats_interval = Some(defaults::STATS_INTERVAL_MS);
1220 }
1221 if self.stats_listen.is_none() {
1222 self.stats_listen = Some(
1224 ConfListen::parse("stats_listen", defaults::STATS_PNAME)
1225 .expect("invariant: STATS_PNAME constant is valid"),
1226 );
1227 }
1228 if self.env.is_none() {
1229 self.env = Some(defaults::ENV.to_string());
1230 }
1231 if self.pem_key_file.is_none() {
1232 self.pem_key_file = Some(defaults::PEM_KEY_FILE.to_string());
1233 }
1234 if self.recon_key_file.is_none() {
1235 self.recon_key_file = Some(defaults::RECON_KEY_FILE.to_string());
1236 }
1237 if self.recon_iv_file.is_none() {
1238 self.recon_iv_file = Some(defaults::RECON_IV_FILE.to_string());
1239 }
1240 if self.recon_interval_seconds.is_none() {
1241 self.recon_interval_seconds = Some(defaults::RECON_INTERVAL_SECONDS);
1242 }
1243 if self.datastore_connections.is_none() {
1244 self.datastore_connections = Some(defaults::DATASTORE_CONNECTIONS);
1245 }
1246 if self.local_peer_connections.is_none() {
1247 self.local_peer_connections = Some(defaults::LOCAL_PEER_CONNECTIONS);
1248 }
1249 if self.remote_peer_connections.is_none() {
1250 self.remote_peer_connections = Some(defaults::REMOTE_PEER_CONNECTIONS);
1251 }
1252 if self.read_repairs_enabled.is_none() {
1253 self.read_repairs_enabled = Some(false);
1254 }
1255 if self.enable_gossip.is_none() {
1256 self.enable_gossip = Some(false);
1257 }
1258 self.apply_hinted_handoff_defaults();
1259 }
1260
1261 fn apply_hinted_handoff_defaults(&mut self) {
1267 if self.enable_hinted_handoff.is_none() {
1268 self.enable_hinted_handoff = Some(defaults::ENABLE_HINTED_HANDOFF);
1269 }
1270 if self.hint_ttl_seconds.is_none() {
1271 self.hint_ttl_seconds = Some(defaults::HINT_TTL_SECONDS);
1272 }
1273 if self.hint_store_max_bytes.is_none() {
1274 self.hint_store_max_bytes = Some(defaults::HINT_STORE_MAX_BYTES);
1275 }
1276 if self.hint_drain_interval_ms.is_none() {
1277 self.hint_drain_interval_ms = Some(defaults::HINT_DRAIN_INTERVAL_MS);
1278 }
1279 if self.transport.is_none() {
1280 self.transport = Some(Transport::default());
1281 }
1282 if self.membership.is_none() {
1283 self.membership = Some(Membership::default());
1284 }
1285 }
1286
1287 pub fn validate(&self, pool_name: &str) -> Result<(), ConfError> {
1304 if pool_name.is_empty() {
1305 return Err(ConfError::EmptyPoolName);
1306 }
1307
1308 if self.listen.is_none() {
1309 return Err(ConfError::MissingRequired("listen"));
1310 }
1311
1312 self.validate_numeric_ranges()?;
1313 self.validate_mbuf_size()?;
1314 self.validate_max_msgs()?;
1315
1316 if let Some(n) = self.data_store {
1317 let ds = DataStore::from_int(n)?;
1318 if ds == DataStore::Dyniak {
1319 self.validate_dyniak()?;
1320 }
1321 }
1322 if let Some(tag) = &self.hash_tag {
1323 if tag.chars().count() != 2 {
1324 return Err(ConfError::BadHashTag(tag.clone()));
1325 }
1326 }
1327
1328 let secure = if let Some(s) = &self.secure_server_option {
1329 SecureServerOption::parse(s)?
1330 } else {
1331 SecureServerOption::None
1332 };
1333 if let Some(s) = &self.read_consistency {
1334 ConsistencyLevel::parse("read_consistency", s)?;
1335 }
1336 if let Some(s) = &self.write_consistency {
1337 ConsistencyLevel::parse("write_consistency", s)?;
1338 }
1339 if secure != SecureServerOption::None {
1340 match &self.pem_key_file {
1341 Some(s) if !s.is_empty() => {}
1342 _ => return Err(ConfError::MissingRequired("pem_key_file")),
1343 }
1344 }
1345
1346 if let Some(s) = &self.log_format {
1347 crate::core::log::LogFormat::parse(s).map_err(|e| ConfError::BadServer {
1348 field: "log_format",
1349 value: s.clone(),
1350 reason: e.to_string(),
1351 })?;
1352 }
1353
1354 self.validate_bucket_types()?;
1355 self.validate_hinted_handoff()?;
1356 self.validate_peer_tls()?;
1357 self.validate_transport()?;
1358 if let Some(r) = &self.riak {
1359 r.validate()?;
1360 }
1361
1362 match &self.servers {
1363 None => return Err(ConfError::MissingRequired("servers")),
1364 Some(s) if s.is_empty() => return Err(ConfError::MissingRequired("servers")),
1365 Some(s) if s.len() > 1 => {
1366 return Err(ConfError::BadServer {
1367 field: "servers",
1368 value: s.len().to_string(),
1369 reason: "expected exactly one datastore entry".to_string(),
1370 });
1371 }
1372 Some(_) => {}
1373 }
1374
1375 Ok(())
1376 }
1377
1378 fn validate_numeric_ranges(&self) -> Result<(), ConfError> {
1379 check_positive("timeout", self.timeout)?;
1380 check_positive("backlog", self.backlog)?;
1381 check_non_negative("client_connections", self.client_connections)?;
1382 check_positive("server_retry_timeout", self.server_retry_timeout)?;
1383 check_positive("server_failure_limit", self.server_failure_limit)?;
1384 check_positive("dyn_read_timeout", self.dyn_read_timeout)?;
1385 check_positive("dyn_write_timeout", self.dyn_write_timeout)?;
1386 check_positive("gos_interval", self.gos_interval)?;
1387 check_positive("stats_interval", self.stats_interval)?;
1388
1389 if let Some(n) = self.dyn_connections {
1390 if n <= 0 {
1391 return Err(ConfError::OutOfRange {
1392 field: "dyn_connections",
1393 value: n,
1394 reason: "must be a positive non-zero number",
1395 });
1396 }
1397 }
1398 Ok(())
1399 }
1400
1401 fn validate_mbuf_size(&self) -> Result<(), ConfError> {
1402 let Some(n) = self.mbuf_size else {
1403 return Ok(());
1404 };
1405 if n <= 0 {
1406 return Err(ConfError::OutOfRange {
1407 field: "mbuf_size",
1408 value: n,
1409 reason: "must be a positive number",
1410 });
1411 }
1412 if !(defaults::MBUF_MIN_SIZE..=defaults::MBUF_MAX_SIZE).contains(&n) {
1413 return Err(ConfError::OutOfRange {
1414 field: "mbuf_size",
1415 value: n,
1416 reason: "must be between 512 and 512000 bytes",
1417 });
1418 }
1419 if n % 16 != 0 {
1420 return Err(ConfError::OutOfRange {
1421 field: "mbuf_size",
1422 value: n,
1423 reason: "must be a multiple of 16",
1424 });
1425 }
1426 Ok(())
1427 }
1428
1429 fn validate_max_msgs(&self) -> Result<(), ConfError> {
1430 let Some(n) = self.max_msgs else {
1431 return Ok(());
1432 };
1433 if n <= 0 {
1434 return Err(ConfError::OutOfRange {
1435 field: "max_msgs",
1436 value: n,
1437 reason: "requires a non-zero number",
1438 });
1439 }
1440 if !(defaults::ALLOC_MSGS_MIN..=defaults::ALLOC_MSGS_MAX).contains(&n) {
1441 return Err(ConfError::OutOfRange {
1442 field: "max_msgs",
1443 value: n,
1444 reason: "must be between 100000 and 1000000 messages",
1445 });
1446 }
1447 Ok(())
1448 }
1449
1450 fn validate_bucket_types(&self) -> Result<(), ConfError> {
1451 use std::collections::BTreeSet;
1452 let mut seen: BTreeSet<&str> = BTreeSet::new();
1453 for bt in &self.bucket_types {
1454 if bt.name.is_empty() {
1455 return Err(ConfError::BadServer {
1456 field: "bucket_types",
1457 value: String::new(),
1458 reason: "bucket-type name must not be empty".to_string(),
1459 });
1460 }
1461 if !seen.insert(bt.name.as_str()) {
1462 return Err(ConfError::BadServer {
1463 field: "bucket_types",
1464 value: bt.name.clone(),
1465 reason: "duplicate bucket-type name".to_string(),
1466 });
1467 }
1468 ConsistencyLevel::parse("read_consistency", &bt.read_consistency)?;
1469 ConsistencyLevel::parse("write_consistency", &bt.write_consistency)?;
1470 }
1471 if let Some(name) = &self.default_bucket_type {
1472 if !self.bucket_types.iter().any(|bt| &bt.name == name) {
1473 return Err(ConfError::BadServer {
1474 field: "default_bucket_type",
1475 value: name.clone(),
1476 reason: "references an undefined bucket-type name".to_string(),
1477 });
1478 }
1479 }
1480 Ok(())
1481 }
1482
1483 fn validate_dyniak(&self) -> Result<(), ConfError> {
1500 if !crate::conf::is_dyniak_supported() {
1501 return Err(ConfError::BadDyniakConfig(
1502 "dyniak data_store requires dynomited built with --features riak",
1503 ));
1504 }
1505 match self.noxu_path.as_deref() {
1506 Some(p) if !p.as_os_str().is_empty() => Ok(()),
1507 _ => Err(ConfError::BadDyniakConfig(
1508 "data_store: dyniak requires a non-empty 'noxu_path:' directive",
1509 )),
1510 }
1511 }
1512
1513 fn validate_hinted_handoff(&self) -> Result<(), ConfError> {
1514 if self.enable_hinted_handoff != Some(true) {
1515 return Ok(());
1516 }
1517 if let Some(ttl) = self.hint_ttl_seconds {
1518 if ttl == 0 {
1519 return Err(ConfError::BadServer {
1520 field: "hint_ttl_seconds",
1521 value: ttl.to_string(),
1522 reason: "must be a positive number when enable_hinted_handoff is true"
1523 .to_string(),
1524 });
1525 }
1526 }
1527 if let Some(cap) = self.hint_store_max_bytes {
1528 if cap == 0 {
1529 return Err(ConfError::BadServer {
1530 field: "hint_store_max_bytes",
1531 value: cap.to_string(),
1532 reason: "must be a positive number when enable_hinted_handoff is true"
1533 .to_string(),
1534 });
1535 }
1536 }
1537 if let Some(period) = self.hint_drain_interval_ms {
1538 if period == 0 {
1539 return Err(ConfError::BadServer {
1540 field: "hint_drain_interval_ms",
1541 value: period.to_string(),
1542 reason: "must be a positive number when enable_hinted_handoff is true"
1543 .to_string(),
1544 });
1545 }
1546 }
1547 Ok(())
1548 }
1549
1550 fn validate_peer_tls(&self) -> Result<(), ConfError> {
1560 validate_tls_pair(
1561 "peer_tls_cert",
1562 "peer_tls_key",
1563 self.peer_tls_cert.as_deref(),
1564 self.peer_tls_key.as_deref(),
1565 )?;
1566 if self.peer_tls_ca.is_some() && self.peer_tls_cert.is_none() {
1567 return Err(ConfError::BadServer {
1568 field: "peer_tls_ca",
1569 value: self
1570 .peer_tls_ca
1571 .as_ref()
1572 .map_or_else(String::new, |p| p.display().to_string()),
1573 reason: "requires peer_tls_cert and peer_tls_key to also be set".into(),
1574 });
1575 }
1576 for (dc, profile) in &self.peer_tls_profiles {
1577 if dc.is_empty() {
1578 return Err(ConfError::BadServer {
1579 field: "peer_tls_profiles",
1580 value: String::new(),
1581 reason: "per-DC TLS profile name must not be empty".into(),
1582 });
1583 }
1584 profile.validate(dc)?;
1585 }
1586 Ok(())
1587 }
1588
1589 fn validate_transport(&self) -> Result<(), ConfError> {
1600 let resolved = self.transport.unwrap_or_default();
1601 if resolved != Transport::Quic {
1602 return Ok(());
1603 }
1604 match (
1605 self.quic_cert_file.as_deref(),
1606 self.quic_key_file.as_deref(),
1607 ) {
1608 (Some(_), Some(_)) => Ok(()),
1609 (None, _) => Err(ConfError::BadServer {
1610 field: "quic_cert_file",
1611 value: String::new(),
1612 reason: "transport: quic requires quic_cert_file to be set".into(),
1613 }),
1614 (Some(_), None) => Err(ConfError::BadServer {
1615 field: "quic_key_file",
1616 value: String::new(),
1617 reason: "transport: quic requires quic_key_file to be set".into(),
1618 }),
1619 }
1620 }
1621}
1622
1623impl fmt::Display for ConfPool {
1624 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1625 match serde_yaml::to_string(self) {
1629 Ok(s) => f.write_str(&s),
1630 Err(_) => Err(fmt::Error),
1631 }
1632 }
1633}
1634
1635fn check_positive(field: &'static str, v: Option<i64>) -> Result<(), ConfError> {
1636 if let Some(n) = v {
1637 if n <= 0 {
1638 return Err(ConfError::OutOfRange {
1639 field,
1640 value: n,
1641 reason: "must be a positive number",
1642 });
1643 }
1644 }
1645 Ok(())
1646}
1647
1648fn check_non_negative(field: &'static str, v: Option<i64>) -> Result<(), ConfError> {
1649 if let Some(n) = v {
1650 if n < 0 {
1651 return Err(ConfError::OutOfRange {
1652 field,
1653 value: n,
1654 reason: "must be a non-negative number",
1655 });
1656 }
1657 }
1658 Ok(())
1659}
1660
1661fn deserialize_data_store<'de, D>(de: D) -> Result<Option<i64>, D::Error>
1667where
1668 D: serde::Deserializer<'de>,
1669{
1670 use serde::de::{self, Visitor};
1671 use std::fmt;
1672
1673 struct V;
1674 impl<'de> Visitor<'de> for V {
1675 type Value = Option<i64>;
1676
1677 fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1678 f.write_str(
1679 "a data_store value: integer (0, 1, 2) or string (valkey, memcache, dyniak)",
1680 )
1681 }
1682
1683 fn visit_none<E: de::Error>(self) -> Result<Self::Value, E> {
1684 Ok(None)
1685 }
1686
1687 fn visit_unit<E: de::Error>(self) -> Result<Self::Value, E> {
1688 Ok(None)
1689 }
1690
1691 fn visit_some<D2: serde::Deserializer<'de>>(
1692 self,
1693 de: D2,
1694 ) -> Result<Self::Value, D2::Error> {
1695 de.deserialize_any(V)
1696 }
1697
1698 fn visit_i64<E: de::Error>(self, v: i64) -> Result<Self::Value, E> {
1699 DataStore::from_int(v)
1700 .map(|d| Some(d.as_int()))
1701 .map_err(|e| E::custom(e.to_string()))
1702 }
1703
1704 fn visit_u64<E: de::Error>(self, v: u64) -> Result<Self::Value, E> {
1705 let n = i64::try_from(v).map_err(|_| E::custom("data_store integer overflow"))?;
1706 self.visit_i64(n)
1707 }
1708
1709 fn visit_str<E: de::Error>(self, v: &str) -> Result<Self::Value, E> {
1710 DataStore::from_name(v)
1711 .map(|d| Some(d.as_int()))
1712 .map_err(|_| {
1713 E::custom(format!(
1714 "data_store: unknown name '{v}'; expected one of: valkey, memcache, dyniak"
1715 ))
1716 })
1717 }
1718
1719 fn visit_string<E: de::Error>(self, v: String) -> Result<Self::Value, E> {
1720 self.visit_str(&v)
1721 }
1722 }
1723
1724 de.deserialize_any(V)
1725}
1726
1727#[cfg(test)]
1728mod tests {
1729 use super::*;
1730
1731 fn pool() -> ConfPool {
1732 ConfPool {
1733 listen: Some(ConfListen::parse("listen", "127.0.0.1:8102").unwrap()),
1734 servers: Some(Servers::from_vec(vec![ConfServer::parse(
1735 "127.0.0.1:6379:1",
1736 )
1737 .unwrap()])),
1738 tokens: Some(TokenList::parse("0").unwrap()),
1739 ..ConfPool::default()
1740 }
1741 }
1742
1743 #[test]
1744 fn validate_minimal_post_finalize() {
1745 let mut p = pool();
1746 p.apply_defaults();
1747 p.validate("dyn_o_mite").unwrap();
1748 }
1749
1750 #[test]
1751 fn missing_listen_rejected() {
1752 let mut p = pool();
1753 p.listen = None;
1754 p.apply_defaults();
1755 assert!(matches!(
1756 p.validate("p"),
1757 Err(ConfError::MissingRequired("listen"))
1758 ));
1759 }
1760
1761 #[test]
1762 fn out_of_range_mbuf_rejected() {
1763 let mut p = pool();
1764 p.mbuf_size = Some(127);
1765 p.apply_defaults();
1766 assert!(matches!(p.validate("p"), Err(ConfError::OutOfRange { .. })));
1767 }
1768
1769 #[test]
1770 fn distribution_field_round_trips_through_yaml() {
1771 let yaml = r"
1772p:
1773 listen: 127.0.0.1:8102
1774 dyn_listen: 127.0.0.1:8101
1775 tokens: '0'
1776 servers:
1777 - 127.0.0.1:6379:1
1778 data_store: 0
1779 distribution: random_slicing
1780 distribution_shadow: vnode
1781 hash: murmur3_x64_64
1782";
1783 let parsed: std::collections::BTreeMap<String, ConfPool> =
1784 serde_yaml::from_str(yaml).unwrap();
1785 let pool = parsed.get("p").unwrap();
1786 assert_eq!(pool.distribution, Some(Distribution::RandomSlicing));
1787 assert_eq!(pool.distribution_shadow, Some(Distribution::Vnode));
1788 assert_eq!(pool.hash, Some(HashType::Murmur3X64_64));
1789 assert_eq!(pool.resolved_distribution(), Distribution::RandomSlicing);
1790 }
1791
1792 #[test]
1793 fn distribution_legacy_alias_resolves_to_vnode() {
1794 let mut p = pool();
1795 p.distribution = Some(Distribution::Ketama);
1796 assert_eq!(p.resolved_distribution(), Distribution::Vnode);
1797 p.distribution = Some(Distribution::Modula);
1798 assert_eq!(p.resolved_distribution(), Distribution::Vnode);
1799 p.distribution = Some(Distribution::Random);
1800 assert_eq!(p.resolved_distribution(), Distribution::Vnode);
1801 }
1802
1803 #[test]
1804 fn distribution_default_unset_is_vnode() {
1805 let p = pool();
1806 assert!(p.distribution.is_none());
1807 assert_eq!(p.resolved_distribution(), Distribution::Vnode);
1808 }
1809
1810 #[test]
1811 fn mbuf_size_not_multiple_of_16_rejected() {
1812 let mut p = pool();
1813 p.mbuf_size = Some(513);
1814 p.apply_defaults();
1815 assert!(matches!(p.validate("p"), Err(ConfError::OutOfRange { .. })));
1816 }
1817
1818 #[test]
1819 fn pem_required_when_secure() {
1820 let mut p = pool();
1821 p.secure_server_option = Some("datacenter".to_string());
1822 p.pem_key_file = Some(String::new());
1823 p.apply_defaults();
1824 assert!(matches!(
1827 p.validate("p"),
1828 Err(ConfError::MissingRequired("pem_key_file"))
1829 ));
1830 }
1831
1832 #[test]
1833 fn data_store_out_of_range_rejected() {
1834 let mut p = pool();
1835 p.data_store = Some(7);
1836 p.apply_defaults();
1837 assert!(matches!(p.validate("p"), Err(ConfError::BadDataStore(7))));
1838 }
1839
1840 static DYNIAK_FLAG_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
1845
1846 #[test]
1847 fn data_store_dyniak_requires_riak_feature() {
1848 let _g = DYNIAK_FLAG_LOCK
1849 .lock()
1850 .unwrap_or_else(std::sync::PoisonError::into_inner);
1851 let prev = crate::conf::is_dyniak_supported();
1854 crate::conf::set_dyniak_supported(false);
1855 let mut p = pool();
1856 p.data_store = Some(2);
1857 p.noxu_path = Some("/scratch/test".into());
1858 p.apply_defaults();
1859 let err = p.validate("p");
1860 crate::conf::set_dyniak_supported(prev);
1861 match err {
1862 Err(ConfError::BadDyniakConfig(msg)) => {
1863 assert!(msg.contains("--features riak"), "unexpected message: {msg}");
1864 }
1865 other => panic!("expected BadDyniakConfig, got {other:?}"),
1866 }
1867 }
1868
1869 #[test]
1870 fn data_store_dyniak_requires_path() {
1871 let _g = DYNIAK_FLAG_LOCK
1872 .lock()
1873 .unwrap_or_else(std::sync::PoisonError::into_inner);
1874 let prev = crate::conf::is_dyniak_supported();
1875 crate::conf::set_dyniak_supported(true);
1876 let mut p = pool();
1877 p.data_store = Some(2);
1878 p.noxu_path = None;
1879 p.apply_defaults();
1880 let err = p.validate("p");
1881 crate::conf::set_dyniak_supported(prev);
1882 match err {
1883 Err(ConfError::BadDyniakConfig(msg)) => {
1884 assert!(msg.contains("noxu_path"), "unexpected message: {msg}");
1885 }
1886 other => panic!("expected BadDyniakConfig, got {other:?}"),
1887 }
1888 }
1889
1890 #[test]
1891 fn data_store_dyniak_yaml_round_trip_string_form() {
1892 let yaml = r"
1895listen: 127.0.0.1:8102
1896servers:
1897- 127.0.0.1:6379:1
1898tokens: '0'
1899data_store: dyniak
1900noxu_path: /scratch/test
1901";
1902 let p: ConfPool = serde_yaml::from_str(yaml).unwrap();
1903 assert_eq!(p.data_store, Some(2));
1904 assert_eq!(
1905 p.noxu_path.as_deref(),
1906 Some(std::path::Path::new("/scratch/test"))
1907 );
1908 let dumped = serde_yaml::to_string(&p).unwrap();
1910 let p2: ConfPool = serde_yaml::from_str(&dumped).unwrap();
1911 assert_eq!(p2.data_store, p.data_store);
1912 assert_eq!(p2.noxu_path, p.noxu_path);
1913 }
1914
1915 #[test]
1916 fn data_store_redis_alias_maps_to_valkey() {
1917 let yaml = r"
1920listen: 127.0.0.1:8102
1921servers:
1922- 127.0.0.1:6379:1
1923tokens: '0'
1924data_store: redis
1925";
1926 let p: ConfPool = serde_yaml::from_str(yaml).unwrap();
1927 assert_eq!(p.data_store, Some(0));
1928 }
1929
1930 #[test]
1931 fn data_store_yaml_int_form_still_works() {
1932 let yaml = r"
1933listen: 127.0.0.1:8102
1934servers:
1935- 127.0.0.1:6379:1
1936tokens: '0'
1937data_store: 2
1938noxu_path: /scratch/test
1939";
1940 let p: ConfPool = serde_yaml::from_str(yaml).unwrap();
1941 assert_eq!(p.data_store, Some(2));
1942 }
1943
1944 #[test]
1945 fn data_store_string_form_unknown_rejected() {
1946 let yaml = r"
1947listen: 127.0.0.1:8102
1948servers:
1949- 127.0.0.1:6379:1
1950tokens: '0'
1951data_store: postgres
1952";
1953 let err = serde_yaml::from_str::<ConfPool>(yaml).unwrap_err();
1954 let msg = err.to_string();
1955 assert!(
1956 msg.contains("unknown name") || msg.contains("data_store"),
1957 "unexpected message: {msg}"
1958 );
1959 }
1960
1961 #[test]
1962 fn hash_tag_must_be_two_chars() {
1963 let mut p = pool();
1964 p.hash_tag = Some("abc".to_string());
1965 p.apply_defaults();
1966 assert!(matches!(p.validate("p"), Err(ConfError::BadHashTag(_))));
1967 }
1968
1969 #[test]
1970 fn empty_servers_rejected() {
1971 let mut p = pool();
1972 p.servers = Some(Servers::from_vec(vec![]));
1973 p.apply_defaults();
1974 assert!(matches!(
1975 p.validate("p"),
1976 Err(ConfError::MissingRequired("servers"))
1977 ));
1978 }
1979
1980 #[test]
1981 fn log_format_known_values_accepted() {
1982 for value in ["default", "rfc5424", "rfc3164", "json", "ndjson", "DEFAULT"] {
1983 let mut p = pool();
1984 p.log_format = Some(value.to_string());
1985 p.apply_defaults();
1986 assert!(p.validate("p").is_ok(), "value {value:?} should validate");
1987 }
1988 }
1989
1990 #[test]
1991 fn log_format_unknown_rejected() {
1992 let mut p = pool();
1993 p.log_format = Some("yaml".to_string());
1994 p.apply_defaults();
1995 let err = p.validate("p").unwrap_err();
1996 assert!(
1997 matches!(
1998 err,
1999 ConfError::BadServer {
2000 field: "log_format",
2001 ..
2002 }
2003 ),
2004 "unexpected error: {err:?}"
2005 );
2006 }
2007
2008 #[test]
2009 fn observability_block_round_trips() {
2010 let yaml = r"
2011observability:
2012 otlp_logs_endpoint: http://collector:4317
2013 service_name: dynomited
2014listen: 127.0.0.1:8102
2015servers:
2016- 127.0.0.1:6379:1
2017tokens: '0'
2018";
2019 let p: ConfPool = serde_yaml::from_str(yaml).unwrap();
2020 let obs = p.observability.as_ref().expect("observability set");
2021 assert_eq!(
2022 obs.otlp_logs_endpoint.as_deref(),
2023 Some("http://collector:4317")
2024 );
2025 assert_eq!(obs.service_name.as_deref(), Some("dynomited"));
2026 }
2027
2028 #[test]
2029 fn bucket_types_round_trip() {
2030 let yaml = r"
2031listen: 127.0.0.1:8102
2032servers:
2033- 127.0.0.1:6379:1
2034tokens: '0'
2035bucket_types:
2036- name: hot
2037 read_consistency: DC_QUORUM
2038 write_consistency: DC_EACH_SAFE_QUORUM
2039 n_val: 3
2040- name: cold
2041 read_consistency: DC_ONE
2042 write_consistency: DC_ONE
2043 n_val: 1
2044default_bucket_type: cold
2045";
2046 let p: ConfPool = serde_yaml::from_str(yaml).unwrap();
2047 assert_eq!(p.bucket_types.len(), 2);
2048 assert_eq!(p.bucket_types[0].name, "hot");
2049 assert_eq!(p.bucket_types[0].n_val, 3);
2050 assert_eq!(
2051 p.bucket_types[0].read_level().unwrap(),
2052 crate::conf::ConsistencyLevel::DcQuorum,
2053 );
2054 assert_eq!(p.default_bucket_type.as_deref(), Some("cold"));
2055 let dumped = serde_yaml::to_string(&p).unwrap();
2057 let p2: ConfPool = serde_yaml::from_str(&dumped).unwrap();
2058 assert_eq!(p2.bucket_types, p.bucket_types);
2059 assert_eq!(p2.default_bucket_type, p.default_bucket_type);
2060 }
2061
2062 #[test]
2063 fn bucket_types_default_is_empty() {
2064 let mut p = pool();
2065 p.apply_defaults();
2066 assert!(p.bucket_types.is_empty());
2067 assert!(p.default_bucket_type.is_none());
2068 assert!(p.validate("p").is_ok());
2069 }
2070
2071 #[test]
2072 fn duplicate_bucket_type_name_rejected() {
2073 let mut p = pool();
2074 p.bucket_types = vec![
2075 ConfBucketType {
2076 name: "a".into(),
2077 read_consistency: "DC_ONE".into(),
2078 write_consistency: "DC_ONE".into(),
2079 n_val: 0,
2080 },
2081 ConfBucketType {
2082 name: "a".into(),
2083 read_consistency: "DC_ONE".into(),
2084 write_consistency: "DC_ONE".into(),
2085 n_val: 0,
2086 },
2087 ];
2088 p.apply_defaults();
2089 let err = p.validate("p").unwrap_err();
2090 assert!(
2091 matches!(
2092 err,
2093 ConfError::BadServer {
2094 field: "bucket_types",
2095 ..
2096 }
2097 ),
2098 "unexpected error: {err:?}",
2099 );
2100 }
2101
2102 #[test]
2103 fn bucket_type_unknown_consistency_rejected() {
2104 let mut p = pool();
2105 p.bucket_types = vec![ConfBucketType {
2106 name: "a".into(),
2107 read_consistency: "DC_PURPLE".into(),
2108 write_consistency: "DC_ONE".into(),
2109 n_val: 0,
2110 }];
2111 p.apply_defaults();
2112 let err = p.validate("p").unwrap_err();
2113 assert!(matches!(err, ConfError::BadConsistency { .. }));
2114 }
2115
2116 #[test]
2117 fn unknown_default_bucket_type_rejected() {
2118 let mut p = pool();
2119 p.default_bucket_type = Some("missing".into());
2120 p.apply_defaults();
2121 let err = p.validate("p").unwrap_err();
2122 assert!(matches!(
2123 err,
2124 ConfError::BadServer {
2125 field: "default_bucket_type",
2126 ..
2127 }
2128 ));
2129 }
2130
2131 #[test]
2132 fn hinted_handoff_default_off_with_canonical_constants() {
2133 let mut p = pool();
2134 p.apply_defaults();
2135 assert_eq!(p.enable_hinted_handoff, Some(false));
2136 assert_eq!(p.hint_ttl_seconds, Some(defaults::HINT_TTL_SECONDS));
2137 assert_eq!(p.hint_store_max_bytes, Some(defaults::HINT_STORE_MAX_BYTES));
2138 assert_eq!(
2139 p.hint_drain_interval_ms,
2140 Some(defaults::HINT_DRAIN_INTERVAL_MS)
2141 );
2142 assert!(p.validate("p").is_ok());
2143 }
2144
2145 #[test]
2146 fn hinted_handoff_yaml_round_trip() {
2147 let yaml = r"
2148listen: 127.0.0.1:8102
2149servers:
2150- 127.0.0.1:6379:1
2151tokens: '0'
2152enable_hinted_handoff: true
2153hint_ttl_seconds: 7200
2154hint_store_max_bytes: 8388608
2155hint_drain_interval_ms: 5000
2156hint_dir: /scratch/dynomite-hints
2157";
2158 let p: ConfPool = serde_yaml::from_str(yaml).unwrap();
2159 assert_eq!(p.enable_hinted_handoff, Some(true));
2160 assert_eq!(p.hint_ttl_seconds, Some(7200));
2161 assert_eq!(p.hint_store_max_bytes, Some(8_388_608));
2162 assert_eq!(p.hint_drain_interval_ms, Some(5_000));
2163 assert_eq!(
2164 p.hint_dir.as_deref(),
2165 Some(std::path::Path::new("/scratch/dynomite-hints"))
2166 );
2167 let dumped = serde_yaml::to_string(&p).unwrap();
2168 let p2: ConfPool = serde_yaml::from_str(&dumped).unwrap();
2169 assert_eq!(p2.enable_hinted_handoff, p.enable_hinted_handoff);
2170 assert_eq!(p2.hint_ttl_seconds, p.hint_ttl_seconds);
2171 assert_eq!(p2.hint_store_max_bytes, p.hint_store_max_bytes);
2172 assert_eq!(p2.hint_drain_interval_ms, p.hint_drain_interval_ms);
2173 assert_eq!(p2.hint_dir, p.hint_dir);
2174 }
2175
2176 #[test]
2177 fn hinted_handoff_zero_ttl_rejected_when_enabled() {
2178 let mut p = pool();
2179 p.enable_hinted_handoff = Some(true);
2180 p.hint_ttl_seconds = Some(0);
2181 p.apply_defaults();
2182 let err = p.validate("p").unwrap_err();
2185 assert!(matches!(
2186 err,
2187 ConfError::BadServer {
2188 field: "hint_ttl_seconds",
2189 ..
2190 }
2191 ));
2192 }
2193
2194 #[test]
2195 fn hinted_handoff_zero_max_bytes_rejected_when_enabled() {
2196 let mut p = pool();
2197 p.enable_hinted_handoff = Some(true);
2198 p.hint_store_max_bytes = Some(0);
2199 p.apply_defaults();
2200 let err = p.validate("p").unwrap_err();
2201 assert!(matches!(
2202 err,
2203 ConfError::BadServer {
2204 field: "hint_store_max_bytes",
2205 ..
2206 }
2207 ));
2208 }
2209
2210 #[test]
2211 fn hinted_handoff_zero_values_ignored_when_disabled() {
2212 let mut p = pool();
2216 p.enable_hinted_handoff = Some(false);
2217 p.hint_ttl_seconds = Some(0);
2218 p.hint_store_max_bytes = Some(0);
2219 p.hint_drain_interval_ms = Some(0);
2220 p.apply_defaults();
2221 assert!(p.validate("p").is_ok());
2222 }
2223
2224 #[test]
2225 fn riak_block_validates_when_unset() {
2226 let mut p = pool();
2227 p.riak = Some(ConfRiak::default());
2228 p.apply_defaults();
2229 assert!(p.validate("p").is_ok());
2230 }
2231
2232 #[test]
2233 fn riak_block_validates_with_addresses() {
2234 let mut p = pool();
2235 p.riak = Some(ConfRiak {
2236 pbc_listen: Some("127.0.0.1:8087".into()),
2237 http_listen: Some("127.0.0.1:8098".into()),
2238 ..ConfRiak::default()
2239 });
2240 p.apply_defaults();
2241 assert!(p.validate("p").is_ok());
2242 }
2243
2244 #[test]
2245 fn riak_block_rejects_bad_pbc_addr() {
2246 let mut p = pool();
2247 p.riak = Some(ConfRiak {
2248 pbc_listen: Some(String::new()),
2249 ..ConfRiak::default()
2250 });
2251 p.apply_defaults();
2252 assert!(matches!(p.validate("p"), Err(ConfError::BadServer { .. })));
2253 }
2254
2255 #[test]
2256 fn riak_block_rejects_segment_above_full_sweep() {
2257 let mut p = pool();
2258 p.riak = Some(ConfRiak {
2259 aae_segment_interval_seconds: Some(120),
2260 aae_full_sweep_interval_seconds: Some(60),
2261 ..ConfRiak::default()
2262 });
2263 p.apply_defaults();
2264 assert!(matches!(p.validate("p"), Err(ConfError::BadServer { .. })));
2265 }
2266
2267 #[test]
2268 fn riak_block_round_trips_through_yaml() {
2269 let yaml = r"
2270p:
2271 listen: 127.0.0.1:1
2272 dyn_listen: 127.0.0.1:2
2273 tokens: '0'
2274 servers:
2275 - 127.0.0.1:3:1
2276 data_store: 0
2277 riak:
2278 pbc_listen: 127.0.0.1:8087
2279 http_listen: 127.0.0.1:8098
2280 aae_enabled: true
2281 aae_full_sweep_interval_seconds: 3600
2282 aae_segment_interval_seconds: 30
2283";
2284 let cfg: std::collections::BTreeMap<String, ConfPool> = serde_yaml::from_str(yaml).unwrap();
2285 let p = cfg.get("p").unwrap();
2286 let r = p.riak.as_ref().unwrap();
2287 assert_eq!(r.pbc_listen.as_deref(), Some("127.0.0.1:8087"));
2288 assert_eq!(r.http_listen.as_deref(), Some("127.0.0.1:8098"));
2289 assert_eq!(r.aae_enabled, Some(true));
2290 assert_eq!(r.aae_full_sweep_interval_seconds, Some(3600));
2291 assert_eq!(r.aae_segment_interval_seconds, Some(30));
2292 }
2293
2294 #[test]
2295 fn peer_tls_pair_unset_is_ok() {
2296 let mut p = pool();
2297 p.apply_defaults();
2298 assert!(p.validate("p").is_ok(), "plaintext default must validate");
2299 }
2300
2301 #[test]
2302 fn peer_tls_pair_both_set_is_ok() {
2303 let mut p = pool();
2304 p.peer_tls_cert = Some(std::path::PathBuf::from("/etc/dynomite/peer.crt"));
2305 p.peer_tls_key = Some(std::path::PathBuf::from("/etc/dynomite/peer.key"));
2306 p.apply_defaults();
2307 assert!(p.validate("p").is_ok());
2308 }
2309
2310 #[test]
2311 fn peer_tls_cert_without_key_rejected() {
2312 let mut p = pool();
2313 p.peer_tls_cert = Some(std::path::PathBuf::from("/x.crt"));
2314 p.apply_defaults();
2315 let err = p.validate("p").unwrap_err();
2316 assert!(
2317 matches!(
2318 err,
2319 ConfError::BadServer {
2320 field: "peer_tls_cert",
2321 ..
2322 }
2323 ),
2324 "got {err:?}"
2325 );
2326 }
2327
2328 #[test]
2329 fn peer_tls_key_without_cert_rejected() {
2330 let mut p = pool();
2331 p.peer_tls_key = Some(std::path::PathBuf::from("/x.key"));
2332 p.apply_defaults();
2333 let err = p.validate("p").unwrap_err();
2334 assert!(
2335 matches!(
2336 err,
2337 ConfError::BadServer {
2338 field: "peer_tls_key",
2339 ..
2340 }
2341 ),
2342 "got {err:?}"
2343 );
2344 }
2345
2346 #[test]
2347 fn peer_tls_ca_without_cert_rejected() {
2348 let mut p = pool();
2349 p.peer_tls_ca = Some(std::path::PathBuf::from("/x.ca"));
2350 p.apply_defaults();
2351 let err = p.validate("p").unwrap_err();
2352 assert!(
2353 matches!(
2354 err,
2355 ConfError::BadServer {
2356 field: "peer_tls_ca",
2357 ..
2358 }
2359 ),
2360 "got {err:?}"
2361 );
2362 }
2363
2364 #[test]
2365 fn riak_tls_cert_without_key_rejected() {
2366 let mut p = pool();
2367 p.riak = Some(ConfRiak {
2368 pbc_listen: Some("127.0.0.1:8087".into()),
2369 tls_cert: Some(std::path::PathBuf::from("/x.crt")),
2370 ..ConfRiak::default()
2371 });
2372 p.apply_defaults();
2373 let err = p.validate("p").unwrap_err();
2374 assert!(
2375 matches!(
2376 err,
2377 ConfError::BadServer {
2378 field: "tls_cert",
2379 ..
2380 }
2381 ),
2382 "got {err:?}"
2383 );
2384 }
2385
2386 #[test]
2387 fn riak_tls_pair_both_set_is_ok() {
2388 let mut p = pool();
2389 p.riak = Some(ConfRiak {
2390 pbc_listen: Some("127.0.0.1:8087".into()),
2391 tls_cert: Some(std::path::PathBuf::from("/x.crt")),
2392 tls_key: Some(std::path::PathBuf::from("/x.key")),
2393 ..ConfRiak::default()
2394 });
2395 p.apply_defaults();
2396 assert!(p.validate("p").is_ok());
2397 }
2398
2399 #[test]
2400 fn riak_quic_listen_requires_tls_pair() {
2401 let mut p = pool();
2402 p.riak = Some(ConfRiak {
2403 quic_listen: Some("127.0.0.1:8089".into()),
2404 ..ConfRiak::default()
2405 });
2406 p.apply_defaults();
2407 let Err(ConfError::BadServer { field, .. }) = p.validate("p") else {
2408 panic!("quic_listen without tls_cert/tls_key must be rejected");
2409 };
2410 assert_eq!(field, "quic_listen");
2411 }
2412
2413 #[test]
2414 fn riak_quic_listen_with_tls_pair_is_ok() {
2415 let mut p = pool();
2416 p.riak = Some(ConfRiak {
2417 quic_listen: Some("127.0.0.1:8089".into()),
2418 tls_cert: Some(std::path::PathBuf::from("/x.crt")),
2419 tls_key: Some(std::path::PathBuf::from("/x.key")),
2420 ..ConfRiak::default()
2421 });
2422 p.apply_defaults();
2423 assert!(p.validate("p").is_ok());
2424 }
2425
2426 #[test]
2427 fn riak_quic_listen_with_quic_only_pair_is_ok_and_tcp_stays_plaintext() {
2428 let mut p = pool();
2431 p.riak = Some(ConfRiak {
2432 pbc_listen: Some("127.0.0.1:8087".into()),
2433 quic_listen: Some("127.0.0.1:8089".into()),
2434 quic_tls_cert: Some(std::path::PathBuf::from("/q.crt")),
2435 quic_tls_key: Some(std::path::PathBuf::from("/q.key")),
2436 ..ConfRiak::default()
2437 });
2438 p.apply_defaults();
2439 assert!(p.validate("p").is_ok());
2440 let r = p.riak.as_ref().unwrap();
2441 assert!(r.tls_cert.is_none(), "TCP listener must stay plaintext");
2442 }
2443
2444 #[test]
2445 fn riak_quic_only_pair_half_set_is_rejected() {
2446 let mut p = pool();
2447 p.riak = Some(ConfRiak {
2448 quic_listen: Some("127.0.0.1:8089".into()),
2449 quic_tls_cert: Some(std::path::PathBuf::from("/q.crt")),
2450 ..ConfRiak::default()
2451 });
2452 p.apply_defaults();
2453 let Err(ConfError::BadServer { field, .. }) = p.validate("p") else {
2454 panic!("a half-set quic_tls pair must be rejected");
2455 };
2456 assert_eq!(field, "quic_listen");
2460 }
2461
2462 #[test]
2463 fn riak_quic_listen_rejects_bad_addr() {
2464 let mut p = pool();
2465 p.riak = Some(ConfRiak {
2466 quic_listen: Some("not-an-addr".into()),
2467 tls_cert: Some(std::path::PathBuf::from("/x.crt")),
2468 tls_key: Some(std::path::PathBuf::from("/x.key")),
2469 ..ConfRiak::default()
2470 });
2471 p.apply_defaults();
2472 assert!(p.validate("p").is_err());
2473 }
2474
2475 #[test]
2476 fn riak_wasm_modules_yaml_round_trip() {
2477 let dir = tempfile::tempdir().unwrap();
2478 let m1 = dir.path().join("identity.wasm");
2479 let m2 = dir.path().join("sum.wasm");
2480 std::fs::write(&m1, b"\0asm\x01\0\0\0").unwrap();
2481 std::fs::write(&m2, b"\0asm\x01\0\0\0").unwrap();
2482 let yaml = format!(
2483 r"
2484listen: 127.0.0.1:8102
2485servers:
2486- 127.0.0.1:6379:1
2487tokens: '0'
2488riak:
2489 pbc_listen: 127.0.0.1:8087
2490 wasm_modules:
2491 - id: identity
2492 path: {m1}
2493 - id: sum
2494 path: {m2}
2495",
2496 m1 = m1.display(),
2497 m2 = m2.display(),
2498 );
2499 let p: ConfPool = serde_yaml::from_str(&yaml).unwrap();
2500 let r = p.riak.as_ref().unwrap();
2501 let mods = r.wasm_modules.as_ref().unwrap();
2502 assert_eq!(mods.len(), 2);
2503 assert_eq!(mods[0].id, "identity");
2504 assert_eq!(mods[0].path, m1);
2505 assert_eq!(mods[1].id, "sum");
2506 assert_eq!(mods[1].path, m2);
2507 let dumped = serde_yaml::to_string(&p).unwrap();
2509 let p2: ConfPool = serde_yaml::from_str(&dumped).unwrap();
2510 assert_eq!(p2.riak.unwrap().wasm_modules, r.wasm_modules);
2511 }
2512
2513 #[test]
2514 fn riak_wasm_modules_unique_ids_required() {
2515 let dir = tempfile::tempdir().unwrap();
2516 let path = dir.path().join("m.wasm");
2517 std::fs::write(&path, b"\0").unwrap();
2518 let r = ConfRiak {
2519 wasm_modules: Some(vec![
2520 ConfRiakWasmModule {
2521 id: "m".into(),
2522 path: path.clone(),
2523 },
2524 ConfRiakWasmModule {
2525 id: "m".into(),
2526 path: path.clone(),
2527 },
2528 ]),
2529 ..ConfRiak::default()
2530 };
2531 let err = r.validate().unwrap_err();
2532 assert!(matches!(
2533 err,
2534 ConfError::BadServer {
2535 field: "wasm_modules.id",
2536 ..
2537 }
2538 ));
2539 }
2540
2541 #[test]
2542 fn riak_wasm_modules_path_must_exist() {
2543 let r = ConfRiak {
2544 wasm_modules: Some(vec![ConfRiakWasmModule {
2545 id: "missing".into(),
2546 path: std::path::PathBuf::from("/no/such/path/at/all.wasm"),
2547 }]),
2548 ..ConfRiak::default()
2549 };
2550 let err = r.validate().unwrap_err();
2551 assert!(matches!(
2552 err,
2553 ConfError::BadServer {
2554 field: "wasm_modules.path",
2555 ..
2556 }
2557 ));
2558 }
2559
2560 #[test]
2561 fn riak_wasm_modules_empty_id_rejected() {
2562 let dir = tempfile::tempdir().unwrap();
2563 let path = dir.path().join("m.wasm");
2564 std::fs::write(&path, b"\0").unwrap();
2565 let r = ConfRiak {
2566 wasm_modules: Some(vec![ConfRiakWasmModule {
2567 id: String::new(),
2568 path,
2569 }]),
2570 ..ConfRiak::default()
2571 };
2572 let err = r.validate().unwrap_err();
2573 assert!(matches!(
2574 err,
2575 ConfError::BadServer {
2576 field: "wasm_modules.id",
2577 ..
2578 }
2579 ));
2580 }
2581
2582 #[test]
2583 fn peer_tls_profile_pair_unset_is_ok() {
2584 let p = ConfTlsProfile::default();
2585 assert!(p.validate("dc1").is_ok());
2586 }
2587
2588 #[test]
2589 fn peer_tls_profile_cert_without_key_rejected() {
2590 let p = ConfTlsProfile {
2591 cert: Some(std::path::PathBuf::from("/x.crt")),
2592 ..ConfTlsProfile::default()
2593 };
2594 let err = p.validate("dc1").unwrap_err();
2595 assert!(matches!(
2596 err,
2597 ConfError::BadServer {
2598 field: "peer_tls_profiles.cert",
2599 ..
2600 }
2601 ));
2602 }
2603
2604 #[test]
2605 fn peer_tls_profile_key_without_cert_rejected() {
2606 let p = ConfTlsProfile {
2607 key: Some(std::path::PathBuf::from("/x.key")),
2608 ..ConfTlsProfile::default()
2609 };
2610 let err = p.validate("dc1").unwrap_err();
2611 assert!(matches!(
2612 err,
2613 ConfError::BadServer {
2614 field: "peer_tls_profiles.key",
2615 ..
2616 }
2617 ));
2618 }
2619
2620 #[test]
2621 fn peer_tls_profile_ca_without_cert_rejected() {
2622 let p = ConfTlsProfile {
2623 ca: Some(std::path::PathBuf::from("/x.ca")),
2624 ..ConfTlsProfile::default()
2625 };
2626 let err = p.validate("dc1").unwrap_err();
2627 assert!(matches!(
2628 err,
2629 ConfError::BadServer {
2630 field: "peer_tls_profiles.ca",
2631 ..
2632 }
2633 ));
2634 }
2635
2636 #[test]
2637 fn peer_tls_profiles_empty_dc_name_rejected() {
2638 let mut p = pool();
2639 p.peer_tls_profiles.insert(
2640 String::new(),
2641 ConfTlsProfile {
2642 cert: Some(std::path::PathBuf::from("/x.crt")),
2643 key: Some(std::path::PathBuf::from("/x.key")),
2644 ca: None,
2645 },
2646 );
2647 p.apply_defaults();
2648 let err = p.validate("p").unwrap_err();
2649 assert!(matches!(
2650 err,
2651 ConfError::BadServer {
2652 field: "peer_tls_profiles",
2653 ..
2654 }
2655 ));
2656 }
2657
2658 #[test]
2659 fn peer_tls_profiles_per_dc_pair_validates() {
2660 let mut p = pool();
2661 p.peer_tls_profiles.insert(
2662 "dc1".into(),
2663 ConfTlsProfile {
2664 cert: Some(std::path::PathBuf::from("/dc1.crt")),
2665 key: Some(std::path::PathBuf::from("/dc1.key")),
2666 ca: None,
2667 },
2668 );
2669 p.apply_defaults();
2670 assert!(p.validate("p").is_ok());
2671 }
2672
2673 #[test]
2674 fn peer_tls_profiles_per_dc_cert_without_key_rejected() {
2675 let mut p = pool();
2676 p.peer_tls_profiles.insert(
2677 "dc1".into(),
2678 ConfTlsProfile {
2679 cert: Some(std::path::PathBuf::from("/dc1.crt")),
2680 key: None,
2681 ca: None,
2682 },
2683 );
2684 p.apply_defaults();
2685 let err = p.validate("p").unwrap_err();
2686 assert!(matches!(
2687 err,
2688 ConfError::BadServer {
2689 field: "peer_tls_profiles.cert",
2690 ..
2691 }
2692 ));
2693 }
2694
2695 #[test]
2696 fn peer_tls_profiles_yaml_round_trip() {
2697 let yaml = r"
2698listen: 127.0.0.1:8102
2699servers:
2700- 127.0.0.1:6379:1
2701tokens: '0'
2702peer_tls_profiles:
2703 dc1:
2704 cert: /etc/dynomite/dc1.pem
2705 key: /etc/dynomite/dc1.key
2706 ca: /etc/dynomite/dc1-ca.pem
2707 dc2:
2708 cert: /etc/dynomite/dc2.pem
2709 key: /etc/dynomite/dc2.key
2710";
2711 let p: ConfPool = serde_yaml::from_str(yaml).unwrap();
2712 assert_eq!(p.peer_tls_profiles.len(), 2);
2713 assert_eq!(
2714 p.peer_tls_profiles["dc1"].cert.as_deref(),
2715 Some(std::path::Path::new("/etc/dynomite/dc1.pem"))
2716 );
2717 assert!(p.peer_tls_profiles["dc2"].ca.is_none());
2718 let dumped = serde_yaml::to_string(&p).unwrap();
2719 let p2: ConfPool = serde_yaml::from_str(&dumped).unwrap();
2720 assert_eq!(p2.peer_tls_profiles, p.peer_tls_profiles);
2721 }
2722
2723 #[test]
2724 fn transport_default_is_tcp_after_finalize() {
2725 let mut p = pool();
2726 p.apply_defaults();
2727 assert_eq!(p.transport, Some(Transport::Tcp));
2728 assert!(p.validate("p").is_ok());
2729 }
2730
2731 #[test]
2732 fn transport_quic_yaml_round_trip() {
2733 let yaml = r"
2734listen: 127.0.0.1:8102
2735servers:
2736- 127.0.0.1:6379:1
2737tokens: '0'
2738transport: quic
2739quic_cert_file: /tmp/test.crt
2740quic_key_file: /tmp/test.key
2741";
2742 let p: ConfPool = serde_yaml::from_str(yaml).unwrap();
2743 assert_eq!(p.transport, Some(Transport::Quic));
2744 assert_eq!(
2745 p.quic_cert_file.as_deref(),
2746 Some(std::path::Path::new("/tmp/test.crt"))
2747 );
2748 assert_eq!(
2749 p.quic_key_file.as_deref(),
2750 Some(std::path::Path::new("/tmp/test.key"))
2751 );
2752 let dumped = serde_yaml::to_string(&p).unwrap();
2753 let p2: ConfPool = serde_yaml::from_str(&dumped).unwrap();
2754 assert_eq!(p2.transport, p.transport);
2755 assert_eq!(p2.quic_cert_file, p.quic_cert_file);
2756 assert_eq!(p2.quic_key_file, p.quic_key_file);
2757 }
2758
2759 #[test]
2760 fn transport_quic_requires_cert_and_key() {
2761 let mut p = pool();
2762 p.transport = Some(Transport::Quic);
2763 p.apply_defaults();
2764 let err = p.validate("p").unwrap_err();
2765 assert!(matches!(
2766 err,
2767 ConfError::BadServer {
2768 field: "quic_cert_file",
2769 ..
2770 }
2771 ));
2772 p.quic_cert_file = Some(std::path::PathBuf::from("/tmp/c.pem"));
2773 let err = p.validate("p").unwrap_err();
2774 assert!(matches!(
2775 err,
2776 ConfError::BadServer {
2777 field: "quic_key_file",
2778 ..
2779 }
2780 ));
2781 p.quic_key_file = Some(std::path::PathBuf::from("/tmp/k.pem"));
2782 assert!(p.validate("p").is_ok());
2783 }
2784
2785 #[test]
2786 fn transport_tcp_ignores_quic_files() {
2787 let mut p = pool();
2788 p.transport = Some(Transport::Tcp);
2789 p.quic_cert_file = Some(std::path::PathBuf::from("/tmp/c.pem"));
2793 p.apply_defaults();
2794 assert!(p.validate("p").is_ok());
2795 }
2796}