Skip to main content

dynomite/conf/
pool.rs

1//! Pool body schema, default application, and validation.
2//!
3//! [`ConfPool`] is the parsed body of one server pool. Every field
4//! whose configuration value may be left unset is wrapped in
5//! [`Option`]; [`ConfPool::apply_defaults`] later fills in the
6//! defaults for the fields the operator omitted.
7
8use 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
22/// Default configuration constants applied when an operator omits
23/// the corresponding key.
24pub mod defaults {
25    /// Default request timeout in milliseconds.
26    pub const TIMEOUT_MS: i64 = 5_000;
27    /// Default `listen()` backlog.
28    pub const LISTEN_BACKLOG: i64 = 512;
29    /// Default `client_connections:` value (0 = unlimited).
30    pub const CLIENT_CONNECTIONS: i64 = 0;
31    /// Default `data_store:` value (0 = redis).
32    pub const DATA_STORE: i64 = 0;
33    /// Default `preconnect:` value.
34    ///
35    /// `preconnect` defaults to false: clients can connect to
36    /// dynomited before the local datastore is reachable. The lazy
37    /// connect avoids a hard dependency on boot ordering.
38    pub const PRECONNECT: bool = false;
39    /// Default `auto_eject_hosts:` value.
40    pub const AUTO_EJECT_HOSTS: bool = true;
41    /// Default `server_retry_timeout:` (ms).
42    pub const SERVER_RETRY_TIMEOUT_MS: i64 = 10 * 1000;
43    /// Default `server_failure_limit:`.
44    pub const SERVER_FAILURE_LIMIT: i64 = 3;
45    /// Default `dyn_read_timeout:` (ms).
46    pub const DYN_READ_TIMEOUT_MS: i64 = 10_000;
47    /// Default `dyn_write_timeout:` (ms).
48    pub const DYN_WRITE_TIMEOUT_MS: i64 = 10_000;
49    /// Default `dyn_connections:`.
50    pub const DYN_CONNECTIONS: i64 = 100;
51    /// Default `gos_interval:` (ms).
52    pub const GOS_INTERVAL_MS: i64 = 30_000;
53    /// Default `enable_hinted_handoff:` value. The feature is
54    /// off by default until operators opt in.
55    pub const ENABLE_HINTED_HANDOFF: bool = false;
56    /// Default `hint_ttl_seconds:` (24h). Hints older than this
57    /// are dropped during expiry sweeps.
58    pub const HINT_TTL_SECONDS: u64 = 86_400;
59    /// Default `hint_store_max_bytes:` (64 MiB) cap on the
60    /// node-local in-memory hint store.
61    pub const HINT_STORE_MAX_BYTES: u64 = 64 * 1024 * 1024;
62    /// Default `hint_drain_interval_ms:` between hint drainer
63    /// sweeps.
64    pub const HINT_DRAIN_INTERVAL_MS: u64 = 30_000;
65    /// Default per-connection message rate.
66    pub const CONN_MSG_RATE: u32 = 50_000;
67    /// Default `stats_interval:` (ms).
68    pub const STATS_INTERVAL_MS: i64 = 30 * 1000;
69    /// Default stats listener address.
70    pub const STATS_PNAME: &str = "0.0.0.0:22222";
71    /// Default datastore-side connection count.
72    pub const DATASTORE_CONNECTIONS: u8 = 1;
73    /// Default local-peer connection count.
74    pub const LOCAL_PEER_CONNECTIONS: u8 = 1;
75    /// Default remote-peer connection count.
76    pub const REMOTE_PEER_CONNECTIONS: u8 = 1;
77    /// Default rack name.
78    pub const RACK: &str = "localrack";
79    /// Default datacenter name.
80    pub const DC: &str = "localdc";
81    /// Default `secure_server_option:` value.
82    pub const SECURE_SERVER_OPTION: &str = "none";
83    /// Default `read_consistency:` / `write_consistency:`.
84    pub const CONSISTENCY: &str = "DC_ONE";
85    /// Default `dyn_seed_provider:`.
86    pub const SEED_PROVIDER: &str = "simple_provider";
87    /// Default `env:` (cloud environment marker).
88    pub const ENV: &str = "aws";
89    /// Default PEM key file path.
90    pub const PEM_KEY_FILE: &str = "conf/dynomite.pem";
91    /// Default reconciliation key file path.
92    pub const RECON_KEY_FILE: &str = "conf/recon_key.pem";
93    /// Default reconciliation IV file path.
94    pub const RECON_IV_FILE: &str = "conf/recon_iv.pem";
95    /// Default cadence (in seconds) of the entropy reconciliation
96    /// run loop. Mirrors the brief's five-minute default; ignored
97    /// when the entropy task is not enabled.
98    pub const RECON_INTERVAL_SECONDS: u64 = 300;
99    /// Smallest valid `mbuf_size:`.
100    pub const MBUF_MIN_SIZE: i64 = 512;
101    /// Largest valid `mbuf_size:`.
102    pub const MBUF_MAX_SIZE: i64 = 512_000;
103    /// Smallest valid `max_msgs:`.
104    pub const ALLOC_MSGS_MIN: i64 = 100_000;
105    /// Largest valid `max_msgs:`.
106    pub const ALLOC_MSGS_MAX: i64 = 1_000_000;
107}
108
109/// Wrapper for the `servers:` field that enforces the invariant
110/// of "exactly one datastore" without losing the YAML list shape.
111///
112/// # Examples
113///
114/// ```
115/// use dynomite::conf::{ConfServer, Servers};
116/// let s = Servers::from_vec(vec![ConfServer::parse("127.0.0.1:6379:1").unwrap()]);
117/// assert_eq!(s.len(), 1);
118/// assert!(!s.is_empty());
119/// assert!(s.datastore().is_some());
120/// ```
121#[derive(Debug, Clone, Default, Eq, PartialEq, Serialize, Deserialize)]
122#[serde(transparent)]
123pub struct Servers(pub(crate) Vec<ConfServer>);
124
125impl Servers {
126    /// Construct from an explicit list. Validation enforces a length
127    /// of one when called via `Config::validate`.
128    ///
129    /// # Examples
130    ///
131    /// ```
132    /// use dynomite::conf::{ConfServer, Servers};
133    /// let s = Servers::from_vec(vec![ConfServer::parse("127.0.0.1:6379:1").unwrap()]);
134    /// assert_eq!(s.len(), 1);
135    /// ```
136    pub fn from_vec(v: Vec<ConfServer>) -> Self {
137        Self(v)
138    }
139}
140
141impl Servers {
142    /// Borrow the entries.
143    ///
144    /// # Examples
145    ///
146    /// ```
147    /// use dynomite::conf::{ConfServer, Servers};
148    /// let s = Servers::from_vec(vec![ConfServer::parse("127.0.0.1:6379:1").unwrap()]);
149    /// assert_eq!(s.entries().len(), 1);
150    /// ```
151    pub fn entries(&self) -> &[ConfServer] {
152        &self.0
153    }
154    /// Number of entries.
155    ///
156    /// # Examples
157    ///
158    /// ```
159    /// use dynomite::conf::Servers;
160    /// assert_eq!(Servers::default().len(), 0);
161    /// ```
162    pub fn len(&self) -> usize {
163        self.0.len()
164    }
165    /// Whether the list is empty.
166    ///
167    /// # Examples
168    ///
169    /// ```
170    /// use dynomite::conf::Servers;
171    /// assert!(Servers::default().is_empty());
172    /// ```
173    pub fn is_empty(&self) -> bool {
174        self.0.is_empty()
175    }
176    /// The single datastore (returns the first entry, if any).
177    ///
178    /// # Examples
179    ///
180    /// ```
181    /// use dynomite::conf::{ConfServer, Servers};
182    /// let s = Servers::from_vec(vec![ConfServer::parse("127.0.0.1:6379:1").unwrap()]);
183    /// assert!(s.datastore().is_some());
184    /// assert!(Servers::default().datastore().is_none());
185    /// ```
186    pub fn datastore(&self) -> Option<&ConfServer> {
187        self.0.first()
188    }
189}
190
191/// Pool configuration body. One per top-level YAML pool name.
192///
193/// # Examples
194///
195/// ```
196/// use dynomite::conf::{ConfPool, ConfListen};
197/// let mut p = ConfPool::default();
198/// assert!(p.listen.is_none());
199/// p.listen = Some(ConfListen::parse("listen", "127.0.0.1:8102").unwrap());
200/// p.apply_defaults();
201/// assert_eq!(p.timeout, Some(5_000));
202/// ```
203#[derive(Debug, Clone, Default, Serialize, Deserialize)]
204#[serde(deny_unknown_fields, default)]
205pub struct ConfPool {
206    /// `listen:` - client-facing listener address.
207    pub listen: Option<ConfListen>,
208    /// `dyn_listen:` - peer-facing listener address.
209    pub dyn_listen: Option<ConfListen>,
210    /// `stats_listen:` - HTTP stats endpoint.
211    pub stats_listen: Option<ConfListen>,
212
213    /// `hash:` - hash function name.
214    pub hash: Option<HashType>,
215    /// `hash_tag:` - two-character delimiter pair.
216    pub hash_tag: Option<String>,
217
218    /// `distribution:` - distribution algorithm. Defaults to
219    /// [`Distribution::Vnode`]. Setting one of the legacy
220    /// `ketama` / `modula` / `random` values is accepted but
221    /// emits a deprecation warning at config-load time and
222    /// collapses to `vnode` at runtime.
223    #[serde(default)]
224    pub distribution: Option<Distribution>,
225    /// `distribution_shadow:` - optional shadow distribution
226    /// computed alongside the live one. When set, the
227    /// dispatcher routes via [`Self::distribution`] but also
228    /// computes the shadow route for every key and bumps a
229    /// counter when the two disagree. Used to validate a
230    /// migration before flipping the live distribution.
231    #[serde(default)]
232    pub distribution_shadow: Option<Distribution>,
233    /// `server_connections:` - deprecated; recorded for warning but ignored.
234    #[serde(default)]
235    pub server_connections: Option<i64>,
236
237    /// `timeout:` - request timeout in milliseconds.
238    pub timeout: Option<i64>,
239    /// `backlog:` - listen backlog.
240    pub backlog: Option<i64>,
241    /// `client_connections:` - max client connections.
242    pub client_connections: Option<i64>,
243    /// `data_store:` - 0 = valkey, 1 = memcache, 2 = dyniak.
244    /// Operators may also write the textual form (`valkey`,
245    /// `memcache`, `dyniak`); the back-compat alias `redis` maps
246    /// to `valkey`. The deserializer normalises every shape to
247    /// the integer code that the rest of the engine consumes.
248    #[serde(default, deserialize_with = "deserialize_data_store")]
249    pub data_store: Option<i64>,
250    /// `noxu_path:` - filesystem directory the in-process Noxu
251    /// DB environment opens at. Required when
252    /// `data_store: dyniak` is selected; ignored otherwise. The
253    /// directory must be writable; an existing environment is
254    /// reused, otherwise one is created.
255    #[serde(default)]
256    pub noxu_path: Option<PathBuf>,
257    /// `search_index_dir:` - filesystem directory the RediSearch
258    /// FT.* index registry snapshots to. When set, the search
259    /// surface persists its index definitions, indexed
260    /// documents, text fields, and suggestion dictionaries to a
261    /// snapshot file under this directory and reloads them on
262    /// restart, so a process kill no longer drops indexes. When
263    /// unset (the default) the registry is purely in-memory:
264    /// indexes are lost on restart and the client must recreate
265    /// them. Only consulted when the binary is built with the
266    /// `search` feature.
267    #[serde(default)]
268    pub search_index_dir: Option<PathBuf>,
269    /// `preconnect:` - eagerly establish connections at startup.
270    pub preconnect: Option<bool>,
271    /// `redis_requirepass:` - optional password sent as `AUTH <pw>`
272    /// on every backend connection right after the TCP handshake.
273    /// Mirrors the Redis server option of the same name. Leave
274    /// unset to disable. Memcache backends are not authenticated
275    /// (`AUTH` is Redis-specific; memcache binary SASL is not
276    /// implemented).
277    #[serde(default)]
278    pub redis_requirepass: Option<String>,
279    /// `auto_eject_hosts:` - automatically eject failing peers.
280    pub auto_eject_hosts: Option<bool>,
281    /// `server_retry_timeout:` - retry interval for ejected servers (ms).
282    pub server_retry_timeout: Option<i64>,
283    /// `server_failure_limit:` - consecutive failures before eject.
284    pub server_failure_limit: Option<i64>,
285
286    /// `servers:` - the (single-element) datastore list.
287    pub servers: Option<Servers>,
288
289    /// `dyn_read_timeout:` - inter-node read timeout (ms).
290    pub dyn_read_timeout: Option<i64>,
291    /// `dyn_write_timeout:` - inter-node write timeout (ms).
292    pub dyn_write_timeout: Option<i64>,
293    /// `dyn_seed_provider:` - seeds backend.
294    pub dyn_seed_provider: Option<String>,
295    /// `dyn_seeds:` - peer dynomite nodes.
296    pub dyn_seeds: Option<Vec<ConfDynSeed>>,
297    /// `dyn_port:` - default peer port.
298    pub dyn_port: Option<i64>,
299    /// `dyn_connections:` - per-peer connection count.
300    pub dyn_connections: Option<i64>,
301    /// `rack:` - this node's rack.
302    pub rack: Option<String>,
303    /// `tokens:` - this node's tokens.
304    pub tokens: Option<TokenList>,
305    /// `gos_interval:` - gossip period (ms).
306    pub gos_interval: Option<i64>,
307    /// `secure_server_option:` - inter-node TLS mode.
308    pub secure_server_option: Option<String>,
309    /// `pem_key_file:` - path to the PEM private key.
310    pub pem_key_file: Option<String>,
311    /// `recon_key_file:` - reconciliation key path.
312    pub recon_key_file: Option<String>,
313    /// `recon_iv_file:` - reconciliation IV path.
314    pub recon_iv_file: Option<String>,
315    /// `recon_interval_seconds:` - period (in seconds) of the
316    /// background entropy reconciliation cycle.
317    ///
318    /// Ignored when [`Self::recon_key_file`] is unset or when the
319    /// configured key file cannot be opened at startup. When the
320    /// entropy task is enabled the default cadence is 300 seconds
321    /// (five minutes); operators can override it via this YAML
322    /// directive.
323    #[serde(default)]
324    pub recon_interval_seconds: Option<u64>,
325    /// `datacenter:` - this node's datacenter.
326    pub datacenter: Option<String>,
327    /// `env:` - cloud environment marker.
328    pub env: Option<String>,
329    /// `conn_msg_rate:` - per-connection message rate cap.
330    pub conn_msg_rate: Option<u32>,
331    /// `read_consistency:` - quorum policy for reads.
332    pub read_consistency: Option<String>,
333    /// `write_consistency:` - quorum policy for writes.
334    pub write_consistency: Option<String>,
335    /// `stats_interval:` - stats aggregation period (ms).
336    pub stats_interval: Option<i64>,
337    /// `enable_gossip:` - enable / disable gossip thread.
338    pub enable_gossip: Option<bool>,
339    /// `membership:` - selects the membership / failure-detection
340    /// backend. `gossip` (the default) is Dynamo-style gossip plus
341    /// the phi-accrual detector; `swim` selects the opt-in
342    /// SWIM + Lifeguard backend. When unset the engine selects
343    /// [`Membership::Gossip`].
344    #[serde(default)]
345    pub membership: Option<Membership>,
346    /// `peer_tls_cert:` - PEM certificate path for the dnode
347    /// listener and outbound dnode connections. When both this
348    /// field and [`Self::peer_tls_key`] are set the peer plane
349    /// runs over TLS; when both are absent the peer plane runs
350    /// in plaintext (the historical behaviour). Setting one
351    /// without the other is rejected at validation time.
352    #[serde(default)]
353    pub peer_tls_cert: Option<PathBuf>,
354    /// `peer_tls_key:` - PEM private-key path matching
355    /// [`Self::peer_tls_cert`].
356    #[serde(default)]
357    pub peer_tls_key: Option<PathBuf>,
358    /// `peer_tls_ca:` - optional PEM CA bundle. When set, the
359    /// dnode listener requires every inbound peer to present a
360    /// certificate signed by a CA from this bundle (mutual TLS).
361    /// When unset, the listener still terminates TLS but does
362    /// not request a client certificate. The outbound side uses
363    /// this bundle as its trust anchor; when unset, the bundled
364    /// `webpki_roots` Mozilla bundle is used.
365    #[serde(default)]
366    pub peer_tls_ca: Option<PathBuf>,
367    /// `peer_tls_profiles:` - per-DC TLS material lookup. Each
368    /// entry is keyed by the target peer's datacenter name; the
369    /// value is a [`ConfTlsProfile`] giving the cert / key / CA
370    /// triple to use when negotiating peer-plane TLS to or from a
371    /// peer in that DC. When the inbound listener accepts a
372    /// connection it picks the cert by SNI hostname
373    /// (`dc-<dc-name>.dynomite.local`); the outbound peer
374    /// supervisor dials with the same SNI hostname so the remote
375    /// listener can route the handshake.
376    ///
377    /// When this map is empty (the default), the legacy
378    /// `peer_tls_cert` / `peer_tls_key` / `peer_tls_ca` triple is
379    /// the only profile in use, applied to every peer regardless
380    /// of DC. When the map is non-empty, each entry takes
381    /// precedence over the legacy fields for matching DCs; the
382    /// legacy fields become the implicit "default" profile used
383    /// for any DC without an explicit entry. When neither the
384    /// map nor the legacy fields are set, the peer plane is
385    /// plaintext.
386    #[serde(default)]
387    pub peer_tls_profiles: BTreeMap<String, ConfTlsProfile>,
388    /// `mbuf_size:` - mbuf chunk size in bytes.
389    pub mbuf_size: Option<i64>,
390    /// `max_msgs:` - allocated message buffer size.
391    pub max_msgs: Option<i64>,
392    /// `datastore_connections:` - count of connections to the datastore.
393    pub datastore_connections: Option<u8>,
394    /// `local_peer_connections:` - count of connections to local-DC peers.
395    pub local_peer_connections: Option<u8>,
396    /// `remote_peer_connections:` - count of connections to remote peers.
397    pub remote_peer_connections: Option<u8>,
398    /// `read_repairs_enabled:` - enable read-repair on quorum mismatch.
399    pub read_repairs_enabled: Option<bool>,
400    /// `enable_hinted_handoff:` - when true, writes whose target
401    /// peer is in [`crate::cluster::peer::PeerState::Down`] (or
402    /// whose outbound channel is closed / full) are stored in a
403    /// node-local hint queue and counted toward the consistency
404    /// threshold; a background drainer ships the hints to the
405    /// peer once it returns to
406    /// [`crate::cluster::peer::PeerState::Normal`]. When false
407    /// (the default) the dispatcher behaviour is unchanged: a
408    /// Down or unreachable target is silently skipped and the
409    /// request fails with `DynomiteNoQuorumAchieved` if the
410    /// remaining targets cannot satisfy the consistency level.
411    #[serde(default)]
412    pub enable_hinted_handoff: Option<bool>,
413    /// `hint_ttl_seconds:` - per-hint expiry. Hints older than
414    /// this many seconds are dropped during periodic sweeps to
415    /// bound the in-memory store. Defaults to 86400 (24 hours).
416    /// Ignored when `enable_hinted_handoff` is false.
417    #[serde(default)]
418    pub hint_ttl_seconds: Option<u64>,
419    /// `hint_store_max_bytes:` - upper bound on the in-memory
420    /// hint store. Once the store reaches this many bytes,
421    /// further enqueues fail with
422    /// [`crate::cluster::hints::HintStoreError::OverCapacity`]
423    /// and the dispatcher falls back to its non-handoff error
424    /// path. Defaults to 64 MiB. Ignored when
425    /// `enable_hinted_handoff` is false.
426    #[serde(default)]
427    pub hint_store_max_bytes: Option<u64>,
428    /// `hint_drain_interval_ms:` - period of the background hint
429    /// drainer sweep. Defaults to 30000 ms (30 seconds). Ignored
430    /// when `enable_hinted_handoff` is false.
431    #[serde(default)]
432    pub hint_drain_interval_ms: Option<u64>,
433    /// `hint_dir:` - filesystem directory for the durable
434    /// hinted-handoff backend. When set (and
435    /// `enable_hinted_handoff` is true), the hint store keeps one
436    /// append-only segment file per peer under this directory and
437    /// replays them at startup, so hints queued for a temporarily
438    /// down peer survive a coordinator restart. When unset (the
439    /// default) the hint store is RAM-only and queued hints are
440    /// lost on restart. Ignored when `enable_hinted_handoff` is
441    /// false.
442    #[serde(default)]
443    pub hint_dir: Option<PathBuf>,
444    /// `log_format:` - selectable shape for tracing output.
445    ///
446    /// Accepted values are `default`, `rfc5424`, `rfc3164`, `json`,
447    /// and `ndjson` (alias of `json`). When unset, the historical
448    /// default text format is used. Parsing is performed at
449    /// log-installation time by [`crate::core::log::LogFormat::parse`];
450    /// invalid values fail the `dynomited --test-conf` gate.
451    pub log_format: Option<String>,
452
453    /// `observability:` - opt-in observability knobs (distributed
454    /// tracing OTLP exporter and an OTLP log-appender bridge).
455    /// Absent / null disables every observability surface,
456    /// preserving the silent default.
457    /// See [`ObservabilityConfig`].
458    #[serde(default)]
459    pub observability: Option<ObservabilityConfig>,
460
461    /// `bucket_types:` - per-bucket routing-property bundles.
462    ///
463    /// Each entry is a [`ConfBucketType`] keyed by name. The
464    /// dispatcher extracts the bucket name from the request key
465    /// (the prefix before the first `/`; see
466    /// [`crate::proto::redis::bucket_name`]) and, when a matching
467    /// entry exists, swaps in that type's `read_consistency`,
468    /// `write_consistency`, and `n_val` for the lifetime of that
469    /// request. Pool-level fields stay the fallback for keys
470    /// without a slash, for keys whose bucket prefix is unknown,
471    /// and for any field the bucket-type stanza leaves at its
472    /// default. Empty list disables the feature; entries must have
473    /// unique names.
474    #[serde(default)]
475    pub bucket_types: Vec<ConfBucketType>,
476
477    /// `default_bucket_type:` - name of the bucket type to apply
478    /// when the request key has no slash (or has an empty prefix).
479    /// Must reference an entry of `bucket_types` when set; unset
480    /// (the default) falls all the way back to pool-level
481    /// consistency.
482    #[serde(default)]
483    pub default_bucket_type: Option<String>,
484
485    /// `riak:` - optional Riak-mode listener / AAE configuration.
486    ///
487    /// Consumed only when `dynomited` is built with the
488    /// `--features riak` Cargo feature: the binary then
489    /// instantiates a Protocol Buffers Client (PBC) listener,
490    /// an HTTP gateway, and (optionally) the active anti-entropy
491    /// scheduler against the supplied addresses. When the field
492    /// is absent (or every inner option is unset), the binary
493    /// behaves identically to a Redis / Memcache deployment.
494    /// The block is parsed unconditionally so YAML files
495    /// authored against the Riak-enabled binary still validate
496    /// under the default build.
497    #[serde(default)]
498    pub riak: Option<ConfRiak>,
499
500    /// `transport:` - selects the network stack the proxy
501    /// listener binds. `tcp` is the historical default and is
502    /// the only option a build without the `quic` Cargo
503    /// feature satisfies. `quic` requires the engine's `quic`
504    /// feature and a server cert / key pair supplied via
505    /// [`Self::quic_cert_file`] and [`Self::quic_key_file`];
506    /// the listener binds a UDP socket and serves the
507    /// configured datastore protocol over a single QUIC
508    /// bidirectional stream per accepted connection.
509    ///
510    /// When unset (the default), the engine selects
511    /// [`Transport::Tcp`].
512    #[serde(default)]
513    pub transport: Option<Transport>,
514
515    /// `quic_cert_file:` - PEM certificate chain path used by
516    /// the QUIC listener when `transport: quic` is selected.
517    /// Required when [`Self::transport`] resolves to
518    /// [`Transport::Quic`]; ignored otherwise.
519    #[serde(default)]
520    pub quic_cert_file: Option<PathBuf>,
521
522    /// `quic_key_file:` - PEM private-key path matching
523    /// [`Self::quic_cert_file`]. Required when
524    /// [`Self::transport`] resolves to [`Transport::Quic`];
525    /// ignored otherwise.
526    #[serde(default)]
527    pub quic_key_file: Option<PathBuf>,
528}
529
530/// Optional Riak-mode listener / AAE knobs.
531///
532/// Every field is optional. The PBC and HTTP listeners are
533/// independent: setting one without the other is supported.
534/// When `aae_enabled` is `true` the active anti-entropy
535/// scheduler is spawned; the cadence knobs default to the
536/// values shipped by `dyniak::aae::config`.
537///
538/// # Examples
539///
540/// ```
541/// use dynomite::conf::ConfRiak;
542/// let r = ConfRiak {
543///     pbc_listen: Some("127.0.0.1:8087".into()),
544///     http_listen: Some("127.0.0.1:8098".into()),
545///     quic_listen: None,
546///     aae_enabled: Some(false),
547///     aae_full_sweep_interval_seconds: None,
548///     aae_segment_interval_seconds: None,
549///     tls_cert: None,
550///     tls_key: None,
551///     tls_ca: None,
552///     wasm_modules: None,
553/// };
554/// assert!(r.validate().is_ok());
555/// ```
556#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
557#[serde(deny_unknown_fields, default)]
558pub struct ConfRiak {
559    /// Address the Riak Protocol Buffers Client listener binds
560    /// to (`host:port`). When unset, the PBC listener is not
561    /// started.
562    pub pbc_listen: Option<String>,
563    /// Address the Riak HTTP gateway listener binds to
564    /// (`host:port`). When unset, the HTTP gateway is not
565    /// started.
566    pub http_listen: Option<String>,
567    /// Address the Riak Protocol Buffers Client listener binds
568    /// to over QUIC (`host:port`, a UDP socket). When unset,
569    /// the QUIC PBC listener is not started. The PBC framing is
570    /// identical to the TCP listener; only the byte transport
571    /// differs (a single QUIC bidirectional stream per accepted
572    /// connection).
573    ///
574    /// QUIC mandates TLS. By default a QUIC listener reuses the
575    /// [`Self::tls_cert`] / [`Self::tls_key`] pair; but that same
576    /// pair also drives the TCP PBC and HTTP listeners, so setting it
577    /// to satisfy QUIC would force TLS onto those plaintext listeners
578    /// too. To run a QUIC listener alongside plaintext TCP / HTTP,
579    /// set a QUIC-only pair via [`Self::quic_tls_cert`] /
580    /// [`Self::quic_tls_key`]; the QUIC listener then uses that pair
581    /// and the TCP / HTTP listeners stay plaintext (unless the shared
582    /// `tls_cert` / `tls_key` are also set). Setting `quic_listen`
583    /// without a QUIC-only pair AND without the shared pair is
584    /// rejected at validation time. Serving QUIC also requires
585    /// `dynomited` to be built with the `quic` Cargo feature; when
586    /// the knob is set but the feature is absent the binary fails
587    /// fast at startup with a clean configuration error rather than
588    /// silently ignoring the directive.
589    pub quic_listen: Option<String>,
590    /// When `true`, the Riak active anti-entropy scheduler is
591    /// spawned alongside the listeners. Default: `false`.
592    pub aae_enabled: Option<bool>,
593    /// Override for the AAE full-sweep cadence, in seconds.
594    /// When unset, `dyniak::aae::config::DEFAULT_FULL_SWEEP_SECONDS`
595    /// (24h) is used.
596    pub aae_full_sweep_interval_seconds: Option<u64>,
597    /// Override for the AAE per-segment exchange cadence, in
598    /// seconds. When unset, `dyniak::aae::config::DEFAULT_SEGMENT_SECONDS`
599    /// (60s) is used.
600    pub aae_segment_interval_seconds: Option<u64>,
601    /// `tls_cert:` - PEM certificate path for the Riak PBC and
602    /// HTTP listeners. When both `tls_cert` and `tls_key` are
603    /// set, both Riak listeners terminate TLS; when both are
604    /// absent, both listeners run in plaintext (the historical
605    /// behaviour). Setting one without the other is rejected at
606    /// validation time.
607    #[serde(default)]
608    pub tls_cert: Option<PathBuf>,
609    /// `tls_key:` - PEM private-key path matching
610    /// [`Self::tls_cert`].
611    #[serde(default)]
612    pub tls_key: Option<PathBuf>,
613    /// `tls_ca:` - optional PEM CA bundle for mutual TLS on the
614    /// Riak listeners. When set, every inbound client must
615    /// present a certificate signed by a CA from this bundle.
616    /// When unset, the listeners terminate TLS without
617    /// requesting a client certificate.
618    #[serde(default)]
619    pub tls_ca: Option<PathBuf>,
620    /// `quic_tls_cert:` - PEM certificate path used ONLY by the QUIC
621    /// PBC listener. When set (with [`Self::quic_tls_key`]), the QUIC
622    /// listener uses this pair and the TCP PBC / HTTP listeners are
623    /// unaffected -- so QUIC can be TLS-secured while TCP / HTTP stay
624    /// plaintext. When unset, the QUIC listener falls back to the
625    /// shared [`Self::tls_cert`] / [`Self::tls_key`] pair. Setting one
626    /// of the QUIC-only pair without the other is rejected at
627    /// validation time.
628    #[serde(default)]
629    pub quic_tls_cert: Option<PathBuf>,
630    /// `quic_tls_key:` - PEM private-key path matching
631    /// [`Self::quic_tls_cert`].
632    #[serde(default)]
633    pub quic_tls_key: Option<PathBuf>,
634    /// `wasm_modules:` - optional list of Wasm modules to
635    /// register with the MapReduce executor at startup. Each
636    /// entry pairs a logical `id` with the on-disk `path` of a
637    /// Wasm binary (`.wasm`) or WAT text (`.wat`) file. When
638    /// `dynomited` is built with the `wasm` Cargo feature it
639    /// loads every entry through the dyniak MapReduce Wasm
640    /// loader (`dyniak::mapreduce::wasm::load_modules_from_config`)
641    /// and exposes the resulting store on the executor; without
642    /// the feature the field is parsed and validated but the
643    /// loader is never called (the runtime returns the typed
644    /// `WasmNotImplemented` error if a `Phase::WasmModule` is
645    /// submitted).
646    ///
647    /// Validation: every `id` must be unique and every `path`
648    /// must point at an existing file at validation time.
649    #[serde(default)]
650    pub wasm_modules: Option<Vec<ConfRiakWasmModule>>,
651}
652
653/// One Wasm module entry inside a [`ConfRiak::wasm_modules`]
654/// list.
655///
656/// # Examples
657///
658/// ```
659/// use std::path::PathBuf;
660/// use dynomite::conf::ConfRiakWasmModule;
661/// let m = ConfRiakWasmModule {
662///     id: "identity".into(),
663///     path: PathBuf::from("/etc/dynomited/wasm/identity.wasm"),
664/// };
665/// assert_eq!(m.id, "identity");
666/// ```
667#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
668#[serde(deny_unknown_fields)]
669pub struct ConfRiakWasmModule {
670    /// Logical module identifier referenced from a
671    /// `Phase::WasmModule { module_id }` MapReduce phase.
672    pub id: String,
673    /// Filesystem path to the Wasm binary (`.wasm`) or WAT
674    /// text (`.wat`) file. Read once at startup.
675    pub path: PathBuf,
676}
677
678/// One per-DC TLS profile inside [`ConfPool::peer_tls_profiles`].
679///
680/// Each profile names a PEM cert / private-key pair and an
681/// optional CA bundle. The map's key is the datacenter name
682/// (matching the value an operator sets in `dyn_seeds:` for
683/// peers in that DC); when a peer's DC has no entry, the
684/// connection falls back to the legacy `peer_tls_*` fields
685/// (treated as the implicit "default" profile). When neither a
686/// per-DC entry nor the default fields are set, the connection
687/// is plaintext.
688///
689/// # Examples
690///
691/// ```
692/// use std::path::PathBuf;
693/// use dynomite::conf::ConfTlsProfile;
694/// let p = ConfTlsProfile {
695///     cert: Some(PathBuf::from("/etc/dynomite/dc1.pem")),
696///     key: Some(PathBuf::from("/etc/dynomite/dc1.key")),
697///     ca: Some(PathBuf::from("/etc/dynomite/dc1-ca.pem")),
698/// };
699/// assert!(p.validate("dc1").is_ok());
700/// ```
701#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
702#[serde(deny_unknown_fields, default)]
703pub struct ConfTlsProfile {
704    /// PEM certificate path. Must be set together with [`Self::key`].
705    pub cert: Option<PathBuf>,
706    /// PEM private-key path matching [`Self::cert`].
707    pub key: Option<PathBuf>,
708    /// Optional PEM CA bundle. When set, peer-plane connections
709    /// using this profile pin the bundle as their trust anchor
710    /// (and the listener requires inbound peers to present a
711    /// certificate signed by a CA in the bundle for mTLS). When
712    /// unset, the listener does not request a client certificate
713    /// and the outbound side falls back to the bundled
714    /// `webpki_roots` Mozilla anchors.
715    pub ca: Option<PathBuf>,
716}
717
718impl ConfTlsProfile {
719    /// Validate that the profile is internally consistent.
720    ///
721    /// `cert` and `key` must both be set or both be unset; a
722    /// `ca` requires the cert / key pair. The `dc` argument is
723    /// the map key the profile lives under and is included in
724    /// any error message so an operator can identify the
725    /// offending entry.
726    ///
727    /// # Errors
728    /// Returns [`ConfError::BadServer`] when the cert / key
729    /// pair is mismatched, or when `ca` is set without the cert
730    /// / key pair.
731    ///
732    /// # Examples
733    ///
734    /// ```
735    /// use std::path::PathBuf;
736    /// use dynomite::conf::ConfTlsProfile;
737    /// let p = ConfTlsProfile {
738    ///     cert: Some(PathBuf::from("/etc/x.pem")),
739    ///     key: None,
740    ///     ca: None,
741    /// };
742    /// assert!(p.validate("dc1").is_err());
743    /// ```
744    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    /// Validate the cross-field invariants of the Riak block.
784    ///
785    /// # Errors
786    /// Returns a [`ConfError::BadServer`] when an address fails
787    /// to parse as a `host:port` socket address, when an AAE
788    /// cadence is zero, or when `aae_segment_interval_seconds`
789    /// exceeds `aae_full_sweep_interval_seconds`.
790    ///
791    /// # Examples
792    ///
793    /// ```
794    /// use dynomite::conf::ConfRiak;
795    /// let r = ConfRiak {
796    ///     pbc_listen: Some("not-a-socket-addr".into()),
797    ///     ..ConfRiak::default()
798    /// };
799    /// assert!(r.validate().is_err());
800    /// ```
801    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            // QUIC mandates TLS. It accepts either a QUIC-only pair
811            // (quic_tls_cert / quic_tls_key), which leaves the TCP /
812            // HTTP listeners plaintext, or the shared tls_cert /
813            // tls_key pair. At least one complete pair must be set.
814            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
909/// Cross-check a `(cert, key)` TLS pair: both must be `Some` or
910/// both must be `None`. The `cert_field` and `key_field` static
911/// strings name the YAML keys for the error message.
912fn 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    // Accept anything that resolves; mirrors how the rest of
947    // the engine validates `listen:` strings (parse first,
948    // resolve as a fallback).
949    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/// Routing-property bundle attached to a key bucket.
973///
974/// Bucket types let operators give different key classes
975/// different SLAs without running multiple pools. Cache-style
976/// keys can pin to `DC_ONE` while transactional keys sit on
977/// `DC_EACH_SAFE_QUORUM`; the same dynomited binary serves both.
978///
979/// `n_val` caps the replica fan-out for the lifetime of one
980/// request: when the topology offers more replicas than `n_val`,
981/// the dispatcher takes the first `n_val` (the existing rack /
982/// DC ordering puts preferred replicas first). A value of `0`
983/// means "no cap" and is treated identically to omitting the
984/// field.
985///
986/// # Examples
987///
988/// ```
989/// use dynomite::conf::{ConfBucketType, ConsistencyLevel};
990/// let bt = ConfBucketType {
991///     name: "sessions".into(),
992///     read_consistency: "DC_QUORUM".into(),
993///     write_consistency: "DC_EACH_SAFE_QUORUM".into(),
994///     n_val: 3,
995/// };
996/// assert_eq!(bt.name, "sessions");
997/// assert_eq!(
998///     ConsistencyLevel::parse("read_consistency", &bt.read_consistency).unwrap(),
999///     ConsistencyLevel::DcQuorum,
1000/// );
1001/// assert_eq!(bt.n_val, 3);
1002/// ```
1003#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
1004#[serde(deny_unknown_fields)]
1005pub struct ConfBucketType {
1006    /// Bucket name. Compared verbatim against the bytes returned
1007    /// by [`crate::proto::redis::bucket_name`]; no normalisation
1008    /// is performed. Names must be unique within a pool and must
1009    /// not be empty.
1010    pub name: String,
1011    /// Read-side consistency level for keys in this bucket.
1012    /// Stored as a string so the YAML round-trip is
1013    /// human-readable; parsed via
1014    /// [`ConsistencyLevel::parse`](crate::conf::ConsistencyLevel::parse)
1015    /// during validation.
1016    pub read_consistency: String,
1017    /// Write-side consistency level for keys in this bucket.
1018    /// See [`ConfBucketType::read_consistency`].
1019    pub write_consistency: String,
1020    /// Replication-factor cap. The dispatcher trims its replica
1021    /// fan-out to at most this many targets. `0` means "no cap";
1022    /// any positive value caps the fan-out to the leading
1023    /// `n_val` peers (rack-local first).
1024    #[serde(default)]
1025    pub n_val: u8,
1026}
1027
1028impl ConfBucketType {
1029    /// Parse [`Self::read_consistency`] into the typed enum.
1030    ///
1031    /// # Examples
1032    ///
1033    /// ```
1034    /// use dynomite::conf::{ConfBucketType, ConsistencyLevel};
1035    /// let bt = ConfBucketType {
1036    ///     name: "s".into(),
1037    ///     read_consistency: "DC_QUORUM".into(),
1038    ///     write_consistency: "DC_ONE".into(),
1039    ///     n_val: 0,
1040    /// };
1041    /// assert_eq!(bt.read_level().unwrap(), ConsistencyLevel::DcQuorum);
1042    /// ```
1043    pub fn read_level(&self) -> Result<ConsistencyLevel, ConfError> {
1044        ConsistencyLevel::parse("read_consistency", &self.read_consistency)
1045    }
1046
1047    /// Parse [`Self::write_consistency`] into the typed enum.
1048    ///
1049    /// # Examples
1050    ///
1051    /// ```
1052    /// use dynomite::conf::{ConfBucketType, ConsistencyLevel};
1053    /// let bt = ConfBucketType {
1054    ///     name: "s".into(),
1055    ///     read_consistency: "DC_ONE".into(),
1056    ///     write_consistency: "DC_SAFE_QUORUM".into(),
1057    ///     n_val: 0,
1058    /// };
1059    /// assert_eq!(bt.write_level().unwrap(), ConsistencyLevel::DcSafeQuorum);
1060    /// ```
1061    pub fn write_level(&self) -> Result<ConsistencyLevel, ConfError> {
1062        ConsistencyLevel::parse("write_consistency", &self.write_consistency)
1063    }
1064}
1065
1066/// Opt-in observability configuration.
1067///
1068/// Absent or null disables every observability surface. Covers both
1069/// distributed tracing and an OTLP log-appender bridge that share
1070/// the same configuration fields.
1071///
1072/// # Examples
1073///
1074/// ```
1075/// use dynomite::conf::ObservabilityConfig;
1076/// let cfg = ObservabilityConfig {
1077///     otlp_traces_endpoint: Some("http://collector:4317".into()),
1078///     otlp_logs_endpoint: None,
1079///     service_name: Some("dynomited".into()),
1080///     traces_sampling: Some(0.1),
1081/// };
1082/// assert!(cfg.otlp_traces_endpoint.is_some());
1083/// ```
1084#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)]
1085#[serde(deny_unknown_fields, default)]
1086pub struct ObservabilityConfig {
1087    /// OTLP gRPC endpoint for distributed traces (e.g.
1088    /// `http://localhost:4317`). When `None` the binary skips
1089    /// the OTel SDK install entirely; tracing keeps using the
1090    /// configured `tracing-subscriber` log layer only.
1091    pub otlp_traces_endpoint: Option<String>,
1092    /// OTLP gRPC endpoint for log records. When set the binary
1093    /// installs an `opentelemetry-appender-tracing` bridge
1094    /// alongside the local log writer, forwarding log records to
1095    /// the collector. When `None` the binary keeps the local log
1096    /// writer only.
1097    pub otlp_logs_endpoint: Option<String>,
1098    /// Service name attached to every emitted span / log record.
1099    /// Defaults to `"dynomited"` when unset.
1100    pub service_name: Option<String>,
1101    /// Trace sampling ratio in `[0.0, 1.0]`. `1.0` records every
1102    /// trace, `0.0` records none. Defaults to `1.0` when unset.
1103    pub traces_sampling: Option<f64>,
1104}
1105
1106impl ConfPool {
1107    /// Resolve the configured [`Distribution`] for the engine.
1108    ///
1109    /// Folds the legacy `ketama` / `modula` / `random` aliases
1110    /// down to [`Distribution::Vnode`] (the only algorithm those
1111    /// names resolve to). Emits a
1112    /// `tracing::warn!` for the legacy aliases the first time
1113    /// the field is read so the operator notices the
1114    /// deprecation.
1115    ///
1116    /// # Examples
1117    ///
1118    /// ```
1119    /// use dynomite::conf::{ConfPool, Distribution};
1120    /// let p = ConfPool {
1121    ///     distribution: Some(Distribution::RandomSlicing),
1122    ///     ..ConfPool::default()
1123    /// };
1124    /// assert_eq!(p.resolved_distribution(), Distribution::RandomSlicing);
1125    /// ```
1126    #[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    /// Apply defaults to any field still left `None` after parsing.
1147    ///
1148    /// # Examples
1149    ///
1150    /// ```
1151    /// use dynomite::conf::ConfPool;
1152    /// let mut p = ConfPool::default();
1153    /// p.apply_defaults();
1154    /// assert_eq!(p.timeout, Some(5_000));
1155    /// assert_eq!(p.rack.as_deref(), Some("localrack"));
1156    /// ```
1157    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        // client_connections is unconditionally reset to its
1171        // default here regardless of any configured value.
1172        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            // Safe: the constant is a hard-coded valid pname.
1223            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    /// Fill in the hinted-handoff knobs and the transport
1262    /// selector. Factored out of [`Self::apply_defaults`] to
1263    /// keep the parent method under the project's per-function
1264    /// line budget while still covering every recently-added
1265    /// key with a default.
1266    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    /// Run the full validation pass against the (presumably finalized)
1288    /// pool body.
1289    ///
1290    /// # Examples
1291    ///
1292    /// ```
1293    /// use dynomite::conf::{ConfListen, ConfPool, ConfServer, Servers, TokenList};
1294    /// let mut p = ConfPool {
1295    ///     listen: Some(ConfListen::parse("listen", "127.0.0.1:8102").unwrap()),
1296    ///     servers: Some(Servers::from_vec(vec![ConfServer::parse("127.0.0.1:6379:1").unwrap()])),
1297    ///     tokens: Some(TokenList::parse("0").unwrap()),
1298    ///     ..ConfPool::default()
1299    /// };
1300    /// p.apply_defaults();
1301    /// assert!(p.validate("dyn_o_mite").is_ok());
1302    /// ```
1303    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    /// Validate the cross-field invariants of the dyniak
1484    /// datastore selection.
1485    ///
1486    /// Selecting `data_store: dyniak` is permitted only when the
1487    /// binary was built with `--features riak`; without it,
1488    /// `dynomited` cannot construct a `NoxuDatastore` because
1489    /// the `dyniak` crate (which owns the type) is not
1490    /// linked. The check is gated on a `cfg!(feature = ...)`
1491    /// expression that the parent crate threads through via
1492    /// the [`crate::conf::set_dyniak_supported`] toggle: the
1493    /// engine ships with the toggle off, the `dynomited` binary
1494    /// turns it on under `--features riak`. The toggle is
1495    /// global because `data_store: dyniak` is a build-time
1496    /// configuration constraint, not a per-pool one.
1497    ///
1498    /// `noxu_path:` must be set and non-empty.
1499    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    /// Cross-check the peer-plane TLS knobs.
1551    ///
1552    /// `peer_tls_cert` and `peer_tls_key` must both be set or
1553    /// both be unset. `peer_tls_ca` is independent (it controls
1554    /// optional mutual TLS) but only meaningful when the cert /
1555    /// key pair is set. Each per-DC profile in
1556    /// `peer_tls_profiles` is validated by
1557    /// [`ConfTlsProfile::validate`]; the per-DC profile names
1558    /// must be non-empty.
1559    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    /// Cross-check the `transport:` selection against the
1590    /// QUIC cert / key knobs.
1591    ///
1592    /// `transport: quic` requires both [`Self::quic_cert_file`]
1593    /// and [`Self::quic_key_file`] to be set; the QUIC listener
1594    /// in `dynomite::net::quic::QuicConfig` cannot bind without
1595    /// a server cert chain and matching private key. The fields
1596    /// are tolerated (but ignored) when `transport: tcp` so
1597    /// operators can switch transports by toggling a single
1598    /// directive without rewriting the whole pool block.
1599    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        // We render the pool body by re-serializing through serde_yaml
1626        // so the round-trip is well defined; this is used by `test_conf`
1627        // and rustdoc examples.
1628        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
1661/// Custom deserializer for `data_store:` that accepts either the
1662/// historical integer form (`0`, `1`, `2`) or the textual form
1663/// (`valkey`, `memcache`, `dyniak`, plus the back-compat alias
1664/// `redis`). Both shapes normalise to the integer code that the
1665/// rest of the engine consumes.
1666fn 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        // apply_defaults restores pem_key_file because it's `Some("")`,
1825        // which is non-None; so we expect MissingRequired("pem_key_file").
1826        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    /// Lock serialising tests that mutate the process-wide
1841    /// `DYNIAK_SUPPORTED` flag. cargo test runs tests on multiple
1842    /// threads; without serialisation a parallel test can flip
1843    /// the flag back before the assertion runs.
1844    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        // Default state: dyniak support flag is off; selecting
1852        // dyniak must be rejected with the documented message.
1853        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        // String form `data_store: dyniak` and integer form
1893        // `data_store: 2` both normalise to integer 2 on parse.
1894        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        // Re-emit and re-parse: round-trip is stable.
1909        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        // The historical `redis` name must keep loading and map
1918        // to the Valkey variant (integer 0).
1919        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        // Re-emit and re-parse: round-trip preserves the data.
2056        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        // apply_defaults() does NOT overwrite Some(0) with the
2183        // default; the validator should reject it.
2184        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        // With handoff off, the validator must NOT reject
2213        // out-of-range values: operators may legitimately leave
2214        // them at zero with handoff off.
2215        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        // A QUIC-only cert pair satisfies quic_listen while leaving the
2429        // TCP PBC listener plaintext (no shared tls_cert / tls_key).
2430        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        // The quic_listen "needs a complete TLS pair" check fires first
2457        // (a lone quic_tls_cert is not a complete pair), so the reported
2458        // field is quic_listen; the point is that it is rejected.
2459        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        // Round-trip back to YAML and re-parse.
2508        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        // Setting cert / key under TCP is tolerated; the
2790        // listener is plain TCP so the QUIC knobs are simply
2791        // unused.
2792        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}